我正在使用通過 uvicorn 提供的 FastAPI 構建一個 API。該 API 具有使用 python 多處理庫的端點。
端點為 CPU 系結任務生成多個行程以并行執行它們。這是高級代碼邏輯概述：
import multiprocessing as mp
class Compute:
def single_compute(self, single_comp_data):
# Computational Task CPU BOUND
global queue
queue.put(self.compute(single_comp_data))
def multi_compute(self, task_ids):
# Prepare for Compuation
output = {}
processes = []
global queue
queue = mp.Queue()
# Start Test Objs Computation
for tid in task_ids:
# Load task data here, to make use of object in memory cache
single_comp_data = self.load_data_from_cache(tid)
p = mp.Process(target=self.single_compute, args=single_comp_data)
p.start()
processes.append(p)
# Collect Parallel Computation
for p in processes:
result = queue.get()
output[result["tid"]]= result
p.join()
return output
這是簡單的 API 代碼：
from fastapi import FastAPI, Response
import json
app = FastAPI()
#comp holds an in memory cache, thats why its created in global scope
comp = Compute()
@app.get("/compute")
def compute(task_ids):
result = comp.multi_compute(task_ids)
return Response(content=json.dumps(result, default=str), media_type="application/json")
當像這樣與多個工人一起運行時：
uvicorn compute_api:app --host 0.0.0.0 --port 7000 --workers 2
我收到這個 python 錯誤
TypeError: can't pickle _thread.lock objects
只有 1 個作業行程就可以了。該程式在 UNIX/LINUX 作業系統上運行。
有人可以向我解釋為什么這里有多個 uvicorn 行程無法分叉一個新行程，以及為什么我會遇到這個胎面鎖？
最后應該實作的目標很簡單：
uvicorn 行程使用該 uvicorn 行程的記憶體副本生成多個其他行程（通過 fork 的子行程）。執行 cpu 系結任務。
uj5u.com熱心網友回復：
TypeError：無法腌制 _thread.lock 物件
源于您傳遞到子流程中的任何資料
p = mp.Process(target=self.single_compute, args=single_comp_data)
包含不可腌制的物件。
所有發送到multiprocessing子行程的 args/kwargs（無論是通過 Process，還是 中的更高級別的方法Pool）都必須是可腌制的，同樣，函式 run 的回傳值必須是可腌制的，以便可以將其發送回父行程。
如果您在 UNIX 上并使用forkstart 方法進行多處理（這是 Linux 上的默認設定，但在 macOS 上不是），您還可以利用寫時復制記憶體語意來避免“向下”復制到子行程通過使資料可用，例如通過實體狀態，全域變數，...，在產生子行程之前進行處理，并讓它通過參考獲取它，而不是將資料本身作為引數傳遞下來。
此示例imap_unordered用于性能（假設不需要按順序處理 id），并將回傳一個 dict 將輸入 ID 映射到它創建的結果。
class Compute:
_cache = {} # could be an instance variable too but whatever
def get_data(self, id):
if id not in self._cache:
self._cache[id] = get_data_from_somewhere(id)
return self._cache[id]
def compute_item(self, id):
data = self.get_data(id)
result = 42 # ... do heavy computation here ...
return (id, result)
def compute_result(self, ids) -> dict:
for id in ids:
self.get_data(id) # populate in parent process
with multiprocessing.Pool() as p:
return dict(p.imap_unordered(self.compute_item, ids))
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/504224.html
上一篇：在多執行緒環境中訪問映射
下一篇：在x86上獲取發布