Effective python note chapter 7

*第七章: Concurrency and Parallelism*

# Item 52: 使用 subprocess 來管理 child processes

• 使用 subprocess module 來執行 child processes, 並管理 input output streams
• Child process 平行的執行, 最大化 CPU 多核心的資源使用率
• 用 popen class 做更進階的使用, 例如 UNIX-tyle pipelines
• 使用 communicate method 的 timeout 參數, 防止 deadlocks

# Item 53: 使用 thread 做 blocking I/O, 避免 parallelism

• 由於 global interpreter lock (GIL), 在多核心中 python threads 沒辦法平行的處理
• 儘管如此, threads 還是很有用的, 提供簡單的方式, 讓程式看起來在同一時間做許多事情
• 使用 python threads 可以平行的執行 system call, 這讓你可以在計算的同時可以執行 blocking I/O (blocking I/O 包含了像是讀寫檔案、與網路互動、與裝置通訊等等的工作。執行緒能夠協助我們處理阻斷式I/O，讓我們的程式不必等候OS回應我的請求時間。)

# Item 54: 在 threads 中, 使用 lock 來防止 data race

• 即便有 GIL 限制, 還是需要注意 threads 間的data race
• 如果沒有使用 mutual-exclusion lock (mutexes), 讓多個 threads 去修改同個 objects, 資料就會出錯, 這時可以使用 lock 來防止

from threading import Thread, Lock

class LockingCounter:
    def __init__(self):
        self.lock = Lock()
        self.count = 0

    def increment(self, offset):
        with self.lock:
            self.count += offset

def worker(sensor_index, how_many, counter):
    for _ in range(how_many):
        counter.increment(1)

how_many = 10 ** 5
threads = []
counter = LockingCounter()

for i in range(5):
    thread = Thread(target=worker, args=(i, how_many, counter))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

expected = how_many * 5
found = counter.count
print(f'Counter should be {expected}, got {found}')
>>>
Counter should be 500000, got 500000

# Item 55: 使用 queue 來調來調節 threads 間的工作

• 用 python threads 時, pipeline 對於組織有順序的工作是不錯的, 特別是對於 I/O-bound 程式
• 特別留意 Concurrent pipeline 會遇到的問題, 例如 busy waiting, 何時告訴 worker 要停止 及 潛在記憶體巨變
• Queue class 提供了所需要的工具讓你構建穩固的 pipelines, blocking operations, buffer sizes, and joining

from queue import Queue
from threading import Thread

class ClosableQueue(Queue):
    SENTINEL = object()

    def close(self):
        self.put(self.SENTINEL)

    def __iter__(self):
        while True:
            item = self.get()
            try:
                if item is self.SENTINEL:
                    return  # Cause the thread to exit
                yield item
            finally:
                self.task_done()

class StoppableWorker(Thread):
    def __init__(self, func, in_queue, out_queue):
        super().__init__()
        self.func = func
        self.in_queue = in_queue
        self.out_queue = out_queue

    def run(self):
        for item in self.in_queue:
            result = self.func(item)
            self.out_queue.put(result)

def download(item):
    return item

def resize(item):
    return item

def upload(item):
    return item

download_queue = ClosableQueue()
resize_queue = ClosableQueue()
upload_queue = ClosableQueue()
done_queue = ClosableQueue()
threads = [
    StoppableWorker(download, download_queue, resize_queue),
    StoppableWorker(resize, resize_queue, upload_queue),
    StoppableWorker(upload, upload_queue, done_queue),
]

for thread in threads:
    thread.start()
for _ in range(1000):
    download_queue.put(object())

download_queue.close()
download_queue.join()
resize_queue.close()
resize_queue.join()
upload_queue.close()
upload_queue.join()
print(done_queue.qsize(), 'items finished')

# Item 56: 可以辨別在什麼情況下, 需要使用 concurrency

• 當程式越來越複雑時, 會需要 multiple concurrent
• 常見的 concurrency coordination 種類為 fan-out (產生新的 concurrency units) 及 fan-in(等待現有的 concurrency units 完成)

# Item 57: 避免當有請求時, 就一昧建立新的 thread

• Threads 有許多不利的地方, 若需要使用許多 threads, 需耗費較多資源及記憶體並且要依靠特殊的工具來協調, 比如 Lock
• Threads 內建沒有提供 raise exceptions 處理機制, 造成其造成出錯或是等待其他 threads 時, 難以 debug

# Item 58: 當需要使用 concurrency 重構時, 了解如何使用 queue

• 使用 queue 搭配固定的固定的 worker threads 數量, 改善 fan-out 及 fan-in 的可擴展性
• 用 queue 來重構現有的程式碼, 尤其是需要多個階段的pipeline 時
• 與其他內建的 python 功能模塊相比, 使用 queue 從根本上限制了 I/O 的並行總數

# Item 59: 當 threads 需要 concurrency 時, 考慮 ThreadPoolExecutor

• ThreadPoolExecutor 通過有限制的重構, 實現簡單的 I/O 並行, 避免每次 fanout concurrency 時, thread 的啟動成本
• 儘管 ThreadPoolExecutor 消除了直接使用 threads 可能造成的記憶體記憶體問題, 但也須預先指定 max_workers 來限制 I/O 並行性

# Item 60: 使用 coroutines 達到高 conurrent I/O

• Function 中使用關鍵字 async, 稱為 coroutines, 使用關鍵字 await, caller 可以從獨立的 coroutine 收到結果
• Coroutines 提供非常有效率的方式,看起來像同一時間執行數千個 functions
• Coroutines 可以使用 fan-out 及 fan-in 達到平行的 I/O, 同時克服了使用 threads 時產生的問題

# Item 61: 知道如何從 thread I/O 轉成 asyncio

• Python 提供了非同步的 for loops, with statements, generators, comprehensions, and library helper functions, 可以用做 coroutine 的替代品
• 內建的 asyncio module, 使原本使用 threading 及 blocking I/O 可以直覺的轉換成 coroutine, asynchronous I/O

# Item 62: 混合 threads 及 coroutines 減少轉換到 asyncio 的過渡期

• Asyncio event loop 裡的 awaitable run_in_executor method 使 coroutines 可以在 threadpoolexecutor pools 執行同步的 functions. 這有助於從上到下搬遷到使用 asyncio
• Asyncio event loop 裡的 run_until_complete method 使同步的程式可以執行 coroutine 直到完成. asyncio.run_coroutine_threadsafe 函數提供跨線程邊界的相同 functions. 有助於從下到上的搬遷

# Item 63: 避免 blocking asyncio event loop 來達到最大的響應

• 在 coroutine 中進行系統調用(包含 blocking I/O 及啟動線程), 可以降低程式的響應速度
• 傳遞 debug = True 的參數給 asyncio.run, 以便檢測何時有 coroutines preventing the event loop, 並快速做出反應

# Item 64: 為了真正的平行, 考慮 concurrent.futures

• 將 CPU 瓶頸轉移到使用 C-extension modules, 可以最大程度的提升 python 程式效能, 然而成本很高且容易產生 bugs
• Multiprocessing module 提供了強大的功能, 可以平行的執行某些類型的計算
• 最好通過內建的 concurrent.future module 及其 ProcessPoolExecutor class 來使用 multiprocessing
• 在用盡其他方法前, 最後在使用 multiprocessing module

有問題或是有錯誤的地方, 歡迎留言與我討論, 謝謝您。

[embed]Previous article: Effective python note chapter 6 第六章: Metaclasses and Attributessmedium.com

[embed]Next article: Effective python note chapter 8 第八章: Robustness and Performancemedium.com