マルチプロセッシングパイプsend（）ブロック

Question

Pythonのドキュメントによれば、recv()ブロックだけがありますが、send()ではありません。私はGUIのスドクゲームを作ろうとしている次のコードを書いた。 tkinterがmainloopを実行していてもゲームボードを更新できるようにしました。しかし、テストの実行中に、ゲームが更新されている間にウィンドウを閉じると、pipe.send()がブロックされ始めます（CPythonプロファイラを使用していることがわかりました）。この問題？

問題を解決するには、スクリプトの更新中にポップアップウィンドウを閉じます。つまり、いくつかの数字がコンソールに表示されている間にウィンドウを閉じます。

私のシステム：MacOSのシエラ10.12.5

import multiprocessing as mp 
import threading 
import random 
import time 
try: 
    import tkinter as tk # Python3 
except ImportError: 
    import Tkinter as tk # Python2 

class VisualizedBoard: 
    def __init__(self,input_string,pipe): 
     '''input_string: a string has a length of at least 81 that represent the board from top-left to bottom right. 
     empty cell is 0''' 

     self.update_scheduled=False 
     self.pipe=pipe 

     # create board 
     self.root = tk.Tk() 
     self.canvas = tk.Canvas(self.root, width=500, height=500) 
     self.canvas.create_rectangle(50, 50, 500, 500, width=2) 
     for i in range(1, 10): 
      self.canvas.create_text(25 + 50 * i, 30, text=str(i)) 
      self.canvas.create_text(30, 25 + 50 * i, text=str(i)) 
      self.canvas.create_line(50 + 50 * i, 50, 50 + 50 * i, 500, width=2 if i % 3 == 0 else 1) 
      self.canvas.create_line(50, 50 + 50 * i, 500, 50 + 50 * i, width=2 if i % 3 == 0 else 1) 

     for i in range(81): 
      if input_string[i] != '0': 
       self.canvas.create_text(75 + 50 * (i // 9), 75 + 50 * (i % 9), tags=str((i//9+1,i%9+1)).replace(' ',''),text=input_string[i], fill='black') 

     self.canvas.pack() 
     self.root.attributes('-topmost', True) 

     self.root.geometry('550x550+%d+%d' % ((self.root.winfo_screenwidth() - 550) // 2, (self.root.winfo_screenheight() - 550) // 2)) 
     self.root.wm_protocol('WM_DELETE_WINDOW',lambda :(self.root.destroy())) 
     threading.Thread(target=self.listen, args=()).start() 
     self.root.mainloop() 


    def update(self,coordinate,value,color='magenta'): 
     """ 
       :parameter coordinate: a tuple (x,y) 
       :parameter value: single digit 
       :returns: None 
     """ 
     try: 
      assert isinstance(coordinate,tuple) 
     except AssertionError: 
      print('Update Failed. Coordinate should be a tuple (x,y)') 

     coordinate_tag=str(coordinate).replace(' ','') 
     self.canvas.delete(coordinate_tag) 
     if value != 0 and value != '0': 
      self.canvas.create_text(25+50*coordinate[0],25+50*coordinate[1],tags=coordinate_tag,text=str(value),fill=color) 

     self.postponed_update() 

     #self.canvas.update() 

    def postponed_update(self): 
     if not self.update_scheduled: 
      self.canvas.after(50,self.scheduled_update) 
      self.update_scheduled=True 

    def scheduled_update(self): 
     self.canvas.update() 
     self.update_scheduled=False 

    def new_board(self,input_string): 
     self.root.destroy() 
     return VisualizedBoard(input_string,self.pipe) 

    def listen(self): 
     try: 
      while True: 
       msg=self.pipe.recv() 
       self.update(*msg) 
     except EOFError: 
      self.pipe.close() 
      tk.Label(self.root,text='Connection to the main script has been closed.\nIt is safe to close this window now.').pack() 
     except Exception as m: 
      self.pipe.close() 
      print('Error during listing:',m) 



class BoardConnection: 
    def __init__(self,input_string): 
     ctx = mp.get_context('spawn') 
     self.receive,self.pipe=ctx.Pipe(False) 
     self.process=ctx.Process(target=VisualizedBoard,args=(input_string,self.receive)) 
     self.process.start() 

    def update(self,coordinate,value,color='magenta'): 
     """ 
     :parameter coordinate: a tuple (x,y) 
     :parameter value: single digit 
     :returns: None 
     """ 
     self.pipe.send((coordinate,value,color)) 

    def close(self): 
     self.pipe.close() 
     self.process.terminate() 





if __name__ == "__main__": 
    b=BoardConnection('000000000302540000050301070000000004409006005023054790000000050700810000080060009') 
    start = time.time() 
    for i in range(5000): #test updating using random numbers 
     b.update((random.randint(1, 9), random.randint(1, 9)), random.randrange(10)) 
     print(i) 
    print(time.time() - start) 

Answer 1

のPython PipeはOS無名のパイプの上に抽象化です。

OSパイプは、一般に、カーネル内の特定のサイズのメモリバッファとして実装されます。デフォルトでは、バッファがいっぱいになるとsend/writeへの次の呼び出しがブロックされます。

消費者がいなくてもデータを公開したい場合は、multiprocessing.Queueまたはasyncioのいずれかの機能を使用する必要があります。

multiprocessing.Queueは、「無制限」バッファとスレッドを使用してデータをOSパイプにプッシュします。パイプがいっぱいになると、公開されたデータがQueueバッファに積み重ねられて、呼び出し元は実行を継続します。

IIRC、asyncioはパイプO_NONBLOCKフラグをセットし、パイプが消費されるのを待ちます。追加のメッセージは、multiprocessing.Queueの場合のように「無制限」バッファに格納されます。