#NEWLINE# @ vimap . util . instancemethod_runonce ( ) #NEWLINE# def close( self ) : #NEWLINE# #INDENT# '\n        Closes any queues from the main method side. For input and output\n        queues, we will close the queue, join the queue thread, and delete\n        the corresponding attribute so any future attempted accesses will\n        fail.\n        ' #NEWLINE# finalize_methods = [ ] #NEWLINE# def _wait_close( queue_name , pipe_name , pipe ) : #NEWLINE# #INDENT# "Works around bugs (or misuse?) in multiprocessing.queue by waiting for\n            a queue's internal pipes to actually be closed.\n\n            See https://github.com/gatoatigrado/vimap/issues/14 for more information.\n            " #NEWLINE# if ( not pipe . closed ) : #NEWLINE# #INDENT# if self . debug : #NEWLINE# #INDENT# print ( 'Force-closing {0} pipe for queue {1}' . format ( pipe_name , queue_name ) ) #NEWLINE# #UNINDENT# pipe . close ( ) #NEWLINE# #UNINDENT# #UNINDENT# def _close_queue( name , queue ) : #NEWLINE# #INDENT# if self . debug : #NEWLINE# #INDENT# print ( 'Main thread queue manager: Closing and joining {0} queue' . format ( name ) ) #NEWLINE# #UNINDENT# queue . close ( ) #NEWLINE# queue . join_thread ( ) #NEWLINE# if ( hasattr ( queue , '_reader' ) and hasattr ( queue , '_writer' ) ) : #NEWLINE# #INDENT# ( reader_pipe , writer_pipe ) = ( queue . _reader , queue . _writer ) #NEWLINE# finalize_methods . append ( ( lambda : _wait_close ( name , 'reader' , reader_pipe ) ) ) #NEWLINE# finalize_methods . append ( ( lambda : _wait_close ( name , 'writer' , writer_pipe ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# _close_queue ( 'input' , self . input_queue ) #NEWLINE# del self . input_queue #NEWLINE# assert self . output_queue . empty ( ) , "You should *not* close the output queue before it's all consumed, else any workers putting items into the queuewill hang!" #NEWLINE# _close_queue ( 'output' , self . output_queue ) #NEWLINE# del self . output_queue #NEWLINE# for finalize_method in finalize_methods : #NEWLINE# #INDENT# finalize_method ( )