[筆記] python3 多執行緒與多核心平行計算

Multi-Thread 與 Multi-Process

*本篇資料來源為莫煩 python:
https://morvanzhou.github.io/

  • python threading 使用

    • 添加thread, 檢視thread, 執行thread
    

    





















    

    
#coding=utf-8
import threading

def thread_job():
    

    # 把目前的 thread 顯示出來看看
    print("This is an added Thread, number is {}\n".format(threading.current_thread()))

def main():
    # 添加一個 thread 
    added_thread = threading.Thread(target=thread_job)
    # 執行 thread
    added_thread.start() # This is an added Thread, number is <Thread(Thread-1, started 123145466363904)>
    # 看目前有幾個 thread
    print(threading.active_count()) # 2
    # 把所有的 thread 顯示出來看看
    print(threading.enumerate()) # [<_MainThread(MainThread, started 140736627270592)>, <Thread(Thread-1, started 123145466363904)>]
    # 把目前的 thread 顯示出來看看
    print(threading.current_thread()) #<_MainThread(MainThread, started 140736627270592)>

if __name__ == '__main__':
    main()
    • join 的功用與用法
      例1
    

    

















    

    
# without join
import threading
import time

def thread_job():
    print('T1 start\n')
    for i in range(10):
        time.sleep(0.1)
    print('T1 finish')

def main():
    thread1 = threading.Thread(target=thread_job, name='T1')
    thread1.start()
    print('all done')

if __name__=='__main__':
    main()
    output:
    因為 thread_job 裡面做的事情比較慢，所以T1 finish 在 all done 後面，也就是主線程(main thread)先執行完。
    
T1 start
all done

T1 finish
    如果主線程要等 T1 做完的話，就要在 print(‘all done’) 前面使用 join()
    例1 使用 join():
    

    

















    

    
# use join
import threading
import time

def thread_job():
    print('T1 start\n')
    for i in range(10):
        time.sleep(0.1)
    print('T1 finish')

def main():
    thread1 = threading.Thread(target=thread_job, name='T1')
    thread1.start()
    print('all done')

if __name__=='__main__':
    main()
    output:
    T1 start

T1 finish
all done
    例2:
    我們新增另一個 thread T2 他要做 T2_job
    

    

























    

    
# without join
import threading
import time

def thread_job():
    print('T1 start\n')
    for i in range(10):
        time.sleep(0.1)
    print('T1 finish')

def T2_job():
    print('T2 start')
    time.sleep(0.1)
    print('T2 finish')

def main():
    thread1 = threading.Thread(target=thread_job, name='T1')
    thread1.start()

    thread2 = threading.Thread(target=T2_job, name='T2')
    thread2.start()
    print('all done')

if __name__=='__main__':
    main()
    output:
    T1 start
T2 start

all done
T2 finish
T1 finish
    例2 使用 join:
    

    





























    

    
# without join
import threading
import time

def thread_job():
    print('T1 start\n')
    for i in range(10):
        time.sleep(0.1)
    print('T1 finish')

def T2_job():
    print('T2 start')
    time.sleep(0.1)
    print('T2 finish')

def main():
    thread1 = threading.Thread(target=thread_job, name='T1')
    thread1.start()

    thread2 = threading.Thread(target=T2_job, name='T2')
    thread2.start()
    
    thread2.join()
    thread1.join()
    
    print('all done')

if __name__=='__main__':
    main()
    output:
    T1 start

T2 start
T2 finish
T1 finish
all done
    • 使用 queue 來接收 thread 執行後回傳的資料
      目前我們的 thread 所做的事情是沒有回傳值的
      如果要讓 thread 運算執行後回傳資料的話，則需要使用 queue 來接。
      threading03_queue.py
    

    





































    

    
#coding=utf-8
import threading
import time
from queue import Queue

def thread_job(arr, q):
    for i in range(len(arr)):
        arr[i] = arr[i]**2
    q.put(arr) # 將結果放進 queue
    # return arr # 上面這一步驟取代了 return

def multithreading():
    q = Queue()   # 宣告 Queue 物件
    threads = []  # 用來放 thread 的 array
    data = [[1,2,3],[4,5,6],[7,7,7],[5,5,5]]
    for i in range(4):
        t = threading.Thread(target=thread_job, args=(data[i], q)) # 將 data 與 queue 傳入 thread 裡面
        t.start()
        threads.append(t)

    for thread in threads:
        thread.join() # 每個 thread 都要做 join

    results = [] # 用來接收與顯示結果的 array

    for _ in range(4):
        results.append(q.get()) # 取出 queue 裡面的資料
    print(results) # 顯示執行後的結果

if __name__=='__main__':
    multithreading()

'''
執行：python3 threading03_queue.py
output:
[[1, 4, 9], [16, 25, 36], [49, 49, 49], [25, 25, 25]]
'''
    • lock的用法
      當我們執行多線程時，如果想要等一個線程做完再做其他線程
      就必須要使用 lock 先鎖住 thread 然後做完事情後再 release
    

    



























    

    
import threading

def job1():
    global A, lock
    lock.acquire()
    for i in range(10):
        A += 1
        print('job1',A)
    lock.release()

def job2():
    global A, lock
    lock.acquire()
    for i in range(10):
        A += 10
        print('job2',A)
    lock.release()

if __name__=='__main__':
    lock = threading.Lock()
    A = 0
    t1 = threading.Thread(target=job1)
    t2 = threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    output:
    job1 1
job1 2
job1 3
job1 4
job1 5
job1 6
job1 7
job1 8
job1 9
job1 10
job2 20
job2 30
job2 40
job2 50
job2 60
job2 70
job2 80
job2 90
job2 100
job2 110

  • python multiprocess

    • 基本用法跟Threading很像
    

    



















    

    
#coding=utf-8
import multiprocessing as mp
import threading as td 
# 使用 multiprocessing 跟使用 threading 的方法非常相似

def job(a,b):
    print('doing job')

# 宣告
t1 = td.Thread(target=job, args=(1,2))
p1 = mp.Process(target=job, args=(1,2))

# 開始
t1.start()
p1.start()

# join 
t1.join()
p1.join()
    須在 main 裡面呼叫 multiprocess
    

    











    

    
#coding=utf-8
import multiprocessing as mp

def job(a,b):
    print('doing job')

# 使用 multiprocessing 要在 main 裡面呼叫
if __name__=='__main__':
    p1 = mp.Process(target=job, args=(1,2))
    p1.start()
    p1.join()
    • 使用Queue接function的回傳值
    

    

























    

    
#coding=utf-8
import multiprocessing as mp

def job(q):
    result = 0
    for i in range(10):
        result += i
    q.put(result)

# 使用 multiprocessing 須在 main 裡面用
if __name__=='__main__':
    q = mp.Queue()   # 使用 queue 接收 function 的回傳值
    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(res1)      # 45
    print(res2)      # 45
    print(res1+res2) # 90
    process 跟 process 之間彼此是獨立工作的，並不會共用同一份資料，所以 process 之間可以確保可平行化處理，跟 Thread 不同，Thread 有 shared data (可以直接共用 global 變數)，且會受 GIL(Global Interpreter Lock) 的保護影響，執行起來不一定能平行化
    GIL簡介：
    http://blog.ephrain.net/python-python-gil-的問題/
    • 一般情形，multi-Thread 與 multi-Process 效能比較
    

    







































































    

    
#coding=utf-8
import multiprocessing as mp
import threading as td 
import time

def job(q):
    result = 0
    for i in range(100000):
        result += i+i**2+i**3
    q.put(result)

# 使用 multi-process 
def multicore():  
    q = mp.Queue()   # 使用 queue 接收 function 的回傳值
    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)

# 使用 multi-thread
def multithread():
    q = mp.Queue()   # multiprocess 的 queue 可以用在 thread 
    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)

def normal():
    result = 0
    # 因為上面都是 2 個 thread 或 2 個 process (core) 
    # 所以要對比一般狀況的效能，要做兩次才公平
    for _ in range(2):
        for i in range(100000):
            result += i+i**2+i**3
    print('normal:',result)


if __name__=='__main__':
    time0 = time.time()
    normal()
    time1 = time.time()
    print('normal time:', time1 - time0)
    multithread()
    time2 = time.time()
    print('multithread time:', time2 - time1)
    multicore()
    time3 = time.time()
    print('multicore time:', time3 - time2)

''' 結果
('normal:', 49999666671666600000L)
('normal time:', 0.03933405876159668)
('multithread:', 49999666671666600000L)
('multithread time:', 0.06222200393676758)
('multicore:', 49999666671666600000L)
('multicore time:', 0.025996923446655273)
'''
    根據以上實驗結果，我們發現 multi-process (multicore) 最有效率，所需時間最短。
    multi-thread 因為會受 GIL 影響，在某些 case 之下效率未必能比普通狀況還要好，所以要使用 multi-thread 來提升程式運算效能，要看任務的性質找適當的時機使用。
    • 使用Process Pool自動分配多核心，接收function回傳值
    

    




























    

    
#coding=utf-8
import multiprocessing as mp

def job(x):
    return x*x

def multicore():
    # 使用 Pool 自動分配給 CPU 的每個一核心 (core)
    pool = mp.Pool() 
    result = pool.map(job, range(10)) # 使用 pool 還可以接到 function 的回傳值 
    print(result) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

     # 使用 Pool 指定分配給 CPU 的 3 核心 (core)
    pool = mp.Pool(processes=3) 
    result = pool.map(job, range(10)) # 使用 pool 還可以接到 function 的回傳值 
    print(result) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

    # 除了 map 功能以外，還有另一個功能： apply_async
    # apply_async 一次只能在一個核心中算一個東西
    res = pool.apply_async(job,(2,)) # 這個只能把一個值放在一個核心運算一次
    print(res.get()) # 4 

    # 如果要輸入一串的話，可以用迭代的方法搭配 apply_async，達到多核心計算的效能
    multi_res = [pool.apply_async(job, (i,)) for i in range(10)]
    print([res.get() for res in multi_res]) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

if __name__=='__main__':
    multicore()
    • Shared Memory共享內存
    

    








    

    
import multiprocessing as mp

# 宣告共用變數
value1 = mp.Value('i', 0) 
value2 = mp.Value('d', 3.14)

# 宣告共用陣列
array = mp.Array('i', [1, 2, 3, 4])
    multiprocess 的共享 Array 只能是一維
    • Process Lock
      以下是一個兩個 process 使用同一個 shared memory 的例子，先觀察一下如果沒有 lock 機制的情形:
    

    





















    

    
#coding=utf-8
import multiprocessing as mp
import time

def job(v, num):
    for _ in range(10):
        time.sleep(0.1)
        v.value += num # 使用共享資料取值要用 value
        print(v.value) 

def multicode():
    v = mp.Value("i",0) # 宣告一個 process 之間共享的變數
    p1 = mp.Process(target=job, args=(v,1))  # 把 v 傳值進去
    p2 = mp.Process(target=job, args=(v,3))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == '__main__':
    multicode()
    目前的 output
    1
4
5
8
9
12
13
16
19
19
20
23
24
24
25
25
26
26
27
27
    可以看出兩個 process 是平行處理的，所以會隨機搶奪 shared 的資料去執行 function
    加上 Lock
    用法跟之前的 Thread Lock 很像
    

    
























    

    
#coding=utf-8
import multiprocessing as mp
import time

def job(v, num, loc):
    loc.acquire()
    for _ in range(10):
        time.sleep(0.1)
        v.value += num # 使用共享資料取值要用 value
        print(v.value) 
    loc.release()

def multicode():
    loc = mp.Lock() # 宣告一個 Lock
    v = mp.Value("i",0) # 宣告一個 process 之間共享的變數
    p1 = mp.Process(target=job, args=(v,1, loc))  # 把 v 傳值進去
    p2 = mp.Process(target=job, args=(v,3, loc))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == '__main__':
    multicode()
    output:
    1
2
3
4
5
6
7
8
9
10
13
16
19
22
25
28
31
34
37
40
    有了 Lock 的保護機制，這次的 output 就很整齊，依序先做完 p1 做的事 (逐步加一) 再做 p2 做的事 (逐步加三)

留言

  1. 紹棟chou2019年7月9日 下午6:35

    使用 queue 來接收 thread 執行後回傳的資料時，為什麼要對threads array 每個 thread做 join?
    在執行for 迴圈的 thread.start的時候不是都跑完了嗎?

    回覆刪除
    回覆
    1. tao2019年7月18日 晚上11:23

      不好意思，最近太少來 blogger 來晚了 QQ
      的確，上面的例子當中做 join() 沒有什麼必要，加了 join() 也不會因此確保 array output 的順序，感謝大大指正 :D

      刪除
  2. 石方2021年8月14日 晚上11:06

    感謝分享 非常詳盡

    回覆刪除

張貼留言

這個網誌中的熱門文章

[筆記] CRLF跟LF之區別 --- 隱形的 bug

圖片
最近在當助教時，偶然間遇到一個隱形的Bug 為什麼稱之為隱形呢？ 請見下面兩張圖 說明：answer.txt 與 answer_test.txt 這兩個檔案的內容用文字編輯器打開時，完全一模一樣，但是我在使用 diff 指令的時候居然出現判定為每一行都不一樣的情形 指令 diff answer.txt answer_test.txt 由最上面第一張圖中可以注意到，answer.txt 的檔案大小是 958 byte answer_test.txt 的檔案大小則為 924 byte 此時不禁讓我疑惑到，兩個檔案內容相同，空白數量也完全相同的情況下，造成檔案大小不同的原因是什麼呢？ 下圖說明了一切 原來在 answer.txt 檔案當中，他的換行是採用 CRLF 而 answer_test.txt 的換行則是一般UNIX的情形 LF # 以下感謝梁浩穎同學解釋 CRLF 就是 \r\n 是只有在 windows 系統在用的形式 所有的 UNIX 系統都是用 \n   (的確，上圖中的 answer.txt 是用windows 7系統的記事本所編輯的，answer_test.txt 則是用 mac 系統透過 sublime text 所編輯的) CR是carriage return的意思， 也就是\r  LF是line feed的意思，也就是\n 這就是為什麼有些檔案從unix系統拿去windows上面看 會變成全部都在同一行 因為只有LF 他不知道是換行 windows只吃CRLF 除了一些比較聰明的文字編輯器會 分的出來 不過其實，CRLF才是真正的打字機運作方式，你可以想像：打字機打到最右邊，要先return到最左邊之後，再往下一行， 所以是CRLF 如果想要清空 answer.txt 裡面所有的 \r 的話，可以使用以下指令 tr -d '\r’ answer.txt # 遇到這個隱形的bug真是難得的好玩經驗，分享給大家 :)
Read more »

[ML筆記] Batch Normalization

圖片
Batch Normalization 本篇為台大電機系李宏毅老師 Machine Learning and having it Deep and Structured (2017)  課程筆記 上課影片： https://www.youtube.com/watch?v=BZh1ltr5Rkg 先從 Feature scaling 或是稱作 Feature Normalization 說起 假設 x1 跟 x2 的數值差距很大 x1 值的範圍 1,2,3,4, ... x2 值的範圍 100, 200, 300, … x1 的 weight 是 w1，x2 的 weight 是 w2 w1 前面乘的值比較小，所以他對於結果的影響比較小 w2 前面乘的值比較大，所以他對於結果的影響比較大
Read more »

[筆記] 統計實習(1) SAS 基礎用法 (匯入資料並另存SAS新檔，SUBSTR，計算總和與平均，BMI)

圖片
使用版本 SAS 9.3 繁體中文 OS: windows xp 匯入資料並另存成SAS新檔 step 1: 匯入資料 step 2: 選擇匯入檔案類型 step 3: 去資料夾 Browse 找尋要匯入的資料檔案 (這裡我命名為Man) step 4: 選好檔案後按OK step 5: 然後按Next step 6: 成功匯入excel資料  (我命名為 Man)，欲找尋這個檔案在哪 -> 按左側的資料館 step 7: 再選Work資料夾 step 8: 就會看到剛剛被我命名的 Man 檔案資料 step 9: 做些簡單的資料分析，在Editor視窗中寫code             Data test;             set man;             run;             以上code做的事會把這個 man (大小寫沒差)資料表複製一份，另命名成 test 存起來 step 10: 要複製另存成SAS檔，使用libname語法 後面放 目錄名稱'檔案路徑'               這裡的目錄名稱為 J 路徑為 '   ' 之中。                      step 11: 如此的code寫完，按下小人圖示提交執行程式 step 12: 會發現資料館裡面，多了一個名稱為 J 的資料夾目錄 step 13: 在 J 資料夾目錄裡面會成功看到被複製出來的名稱為 test 的資料 step 14: 實際跑到剛剛在 '   ' 之中的路徑看，也會看到test資料 SAS SUBSTR的用法 substr (sub string)顧名思義就是將一串文字當中切割出來一條子字串 語法為 新Table欄位名稱 = SUBSTR( Table欄位名稱 , 起始bit位置 , 要往後多少位元 ) 常用於將日期年月日分別切割出來變成數字來運算 step 1: 在Editor當中輸入語法 以第一橫排 16/10/1986為例 要取得日期DAY要從time欄位的第1個bit開始往後切長度
Read more »

[ML筆記] Ensemble - Bagging, Boosting & Stacking

圖片
ML Lecture 27: Ensemble 本篇為台大電機系李宏毅老師 Machine Learning (2016) 課程筆記 上課影片： https://www.youtube.com/watch?v=tH9FH1DH5n0 課程網 (投影片出處)： http://speech.ee.ntu.edu.tw/~tlkagk/courses_ML16.html Ensemble 方法其實就是 “團隊合作” 好幾個模型一起上的方法 假如我們手上有一堆 Classifier ，這些 Classifier 每個都有不同的屬性， 要把他們集合起來一起合作，集合再一起需要用比較好的方法 以下來介紹一些 Ensemble 的方法 Bagging
Read more »