テクノロジー · 実用ガイド
Python multiprocessing.Queue が join で停止？feeder と drain の順序を可視化する
process、feeder thread、unfinished task の三つの待機を区別し、silent data loss を避けて終了順序を直します。
子 process が target function を終えても、Queue feeder が誰も読まない pipe へ buffer を flush していれば終了できません。見えている join は待機鎖の末端です。
どの join かを特定し、writer、feeder、pipe、reader の所有関係を描いてから drain と close の順序を修正します。
1. 停止している join を正確に特定する
何かを terminate する前に thread/process stack を保存します。Process.join は child exit、Queue.join_thread は close 後の feeder exit、JoinableQueue.join は全 task_done を待ちます。
start method、Python version、OS、queue type、creator、producer、consumer、最後の put/get/task_done を記録し、qsize を証拠にしません。
2. writer–feeder–pipe–reader の待機図を作る
Queue.put 後、background feeder が object を serialize して pipe へ送ります。reader が止まり pipe が満杯になると feeder の write が停止し、producer の終了も flush 待ちになります。
全 read/write end、close owner、sentinel sender、待機中も読む actor を列挙します。pipe capacity は有限で正しさの前提にはできません。
3. 所有契約に沿って順序を修正する
result queue は worker が生きている間に drain し、全 producer を先に join しません。新規 work を止め、必要数の sentinel、必要 result の drain、writer close、feeder/process join の順を待機図で決めます。
JoinableQueue は成功した get ごとに例外時も一度だけ task_done を呼び、unfinished count を減らす consumer を join で塞ぎません。
4. cancel_join_thread と terminate を損失判断として扱う
cancel_join_thread は feeder flush を待たずに exit できるため buffer data を失う可能性があります。明示的な loss policy と test がある場合だけ使います。
terminate/kill も queue state や in-flight bytes を壊し得るため、replay、deduplicate、abandon の境界を先に定義します。
5. 正常終了と完全配送を同時に証明する
通常の pipe capacity を超える payload で produced/consumed ID を照合し、deployment と同じ start method で繰り返します。exit だけでなく全 record、feeder 消滅、exit code、破損なしを確認します。
reader を停止する negative test では bounded backpressure または controlled timeout を観測し、silent drop を許しません。
