multiprocess test

mp_test.py

import multiprocessing as mp
import threading as td

def job(a, d):
    print('aaaaa')
    
    
if __name__ == '__main__':
    t1 = td.Thread(target=job, args=(1,2))
    p1 = mp.Process(target=job, args=(1,2))


    t1.start()
    p1.start()

    t1.join()
    p1.join()
    

multicore_vs_multithread.py

import multiprocessing as mp
import threading as td
import time


def job(q):
    res = 0
    for i in range(1000000):
        res += i + i**2 + i**3
    q.put(res) # queue
    
    
def normal():
    res = 0
    for _ in range(2):
        for i in range(10*10000):
            res += i + i**2 + i**3
    print('normal: ', res)
    
    
def multicore():
    q = mp.Queue()
    p1 = mp.Process(target=job, args=(q,))
    p2 = mp.Process(target=job, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    res1 = q.get()
    res2 = q.get()
    print('multicore:',res1 + res2)
    

    
def multithread():
    q = mp.Queue() # thread可放入process同样的queue中
    t1 = td.Thread(target=job, args=(q,))
    t2 = td.Thread(target=job, args=(q,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    res1 = q.get()
    res2 = q.get()
    print('multithread:', res1 + res2)
    
    
# compare time

if __name__ == '__main__':
    st = time.time()
    normal()
    st1 = time.time()
    print('normal time: ', st1 - st)
    multithread()
    st2 = time.time()
    print('multithread time: ', st2 - st1)
    multicore()
    print('multicore time: ', time.time() - st2)

queue_test

def job1(q):
    res = 0
    for i in range(1000):
        res += 1 + i**2 + i**3
    q.put(res)
    
def job2(q):
    res = 0
    for i in range(1000):
        res += 1 + i**2 
    q.put(res)
    
if __name__ == '__main__':
    q = mp.Queue()
    p1 = mp.Process(target=job1, args=(q, ))
    p2 = mp.Process(target=job2, args=(q, ))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    res1 = q.get()
    res2 = q.get()
    
    print(res1)
    print(res2)