マルチプロセス.Queue：壊れたパイプエラー

Question

注1： 私はマルチプロセスで使用したいです.Queueを複数のプロセスで、私はこの問題がシングルプロセスの状況で発生しました。したがって、以下のコードは、単一のプロセスを使用して質問を単純化しています。

Broken pipe error with multiprocessing.Queueと同様の質問があります。

この記事の答えは、キュースレッドがジョブを終了する前にメインスレッドが終了するためです。彼は固定の方法は、それは彼のコードにsleep(0.1)を追加することであった。

import multiprocessing 
import time 
def main(): 
    q = multiprocessing.Queue() 
    for i in range(10): 
     print i 
     q.put(i) 
    time.sleep(0.1) # Just enough to let the Queue finish 

if __name__ == "__main__": 
    main() 

しかし、私は睡眠は生産コードのための安定した方法ではないと思うので、私はそれを行うためにjoinを使用しようとしました。下のコードを見ることはできますが、残念ながら動作しません。睡眠なしでこれを行う方法を知っている人はいますか？

import multiprocessing 
import time 


def main(): 
    q = multiprocessing.Queue() 
    for i in range(10): 
     q.put(i) 
    # time.sleep(4) 
    q.close() 
    q.join_thread() 

if __name__ == "__main__": 
    main() 

Answer 1

まず、multiprocessing.Queueの詳細を説明しましょう。オブジェクトがキューに置かれたとき

、オブジェクトが漬けと背景スレッド後の基礎となるパイプに漬けたデータをフラッシュしています。

パイプはreader, writer = socket.socketpair()で作成されます。

queue.close()は、マルチプロセスのために設計されており、それは二つのものに近いreader

1. ん（重要！）を
2. かの出会い、そのような値が
バックグラウンドスレッド が終了します、 queue.bufferに sentinel値を送信

シングルプロセスの場合、queue.close()が機能しない理由は、ステップ1の場合、bufferに何らかのデータが残っていると、バックグラウンドスレッドはすでに閉じたソケットにデータを書き込んだり、Broken pipeエラーにつながります。

マルチプロセスの場合

import socket 

reader, writer = socket.socketpair() 
writer.send("1") 

# queue.close() will internally call reader.close() 
reader.close() 

# got a Broken pipe error 
writer.send("2") 

エラーをデモするための単純な例では、メイン処理におけるreader近いのみ、下層のソケット（メインと子プロセスを共有ソケット）の参照カウントをしないデクリメントソケットを実際に閉じる（またはシャットダウンする）。