<s> __all__ = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> import sys <EOL> import os <EOL> import threading <EOL> import collections <EOL> import time <EOL> import atexit <EOL> import weakref <EOL> from Queue import Empty , Full <EOL> import _multiprocessing <EOL> from multiprocessing import Pipe <EOL> from multiprocessing . synchronize import Lock , BoundedSemaphore , Semaphore , Condition <EOL> from multiprocessing . util import debug , info , Finalize , register_after_fork <EOL> from multiprocessing . forking import assert_spawning <EOL> class Queue ( object ) : <EOL> def __init__ ( self , maxsize = <NUM_LIT:0> ) : <EOL> if maxsize <= <NUM_LIT:0> : <EOL> maxsize = _multiprocessing . SemLock . SEM_VALUE_MAX <EOL> self . _maxsize = maxsize <EOL> self . _reader , self . _writer = Pipe ( duplex = False ) <EOL> self . _rlock = Lock ( ) <EOL> self . _opid = os . getpid ( ) <EOL> if sys . platform == '<STR_LIT:win32>' : <EOL> self . _wlock = None <EOL> else : <EOL> self . _wlock = Lock ( ) <EOL> self . _sem = BoundedSemaphore ( maxsize ) <EOL> self . _after_fork ( ) <EOL> if sys . platform != '<STR_LIT:win32>' : <EOL> register_after_fork ( self , Queue . _after_fork ) <EOL> def __getstate__ ( self ) : <EOL> assert_spawning ( self ) <EOL> return ( self . _maxsize , self . _reader , self . _writer , <EOL> self . _rlock , self . _wlock , self . _sem , self . _opid ) <EOL> def __setstate__ ( self , state ) : <EOL> ( self . _maxsize , self . _reader , self . _writer , <EOL> self . _rlock , self . _wlock , self . _sem , self . _opid ) = state <EOL> self . _after_fork ( ) <EOL> def _after_fork ( self ) : <EOL> debug ( '<STR_LIT>' ) <EOL> self . _notempty = threading . Condition ( threading . Lock ( ) ) <EOL> self . _buffer = collections . deque ( ) <EOL> self . _thread = None <EOL> self . _jointhread = None <EOL> self . _joincancelled = False <EOL> self . _closed = False <EOL> self . _close = None <EOL> self . _send = self . _writer . send <EOL> self . _recv = self . _reader . recv <EOL> self . _poll = self . _reader . poll <EOL> def put ( self , obj , block = True , timeout = None ) : <EOL> assert not self . _closed <EOL> if not self . _sem . acquire ( block , timeout ) : <EOL> raise Full <EOL> self . _notempty . acquire ( ) <EOL> try : <EOL> if self . _thread is None : <EOL> self . _start_thread ( ) <EOL> self . _buffer . append ( obj ) <EOL> self . _notempty . notify ( ) <EOL> finally : <EOL> self . _notempty . release ( ) <EOL> def get ( self , block = True , timeout = None ) : <EOL> if block and timeout is None : <EOL> self . _rlock . acquire ( ) <EOL> try : <EOL> res = self . _recv ( ) <EOL> self . _sem . release ( ) <EOL> return res <EOL> finally : <EOL> self . _rlock . release ( ) <EOL> else : <EOL> if block : <EOL> deadline = time . time ( ) + timeout <EOL> if not self . _rlock . acquire ( block , timeout ) : <EOL> raise Empty <EOL> try : <EOL> if not self . _poll ( block and ( deadline - time . time ( ) ) or <NUM_LIT:0.0> ) : <EOL> raise Empty <EOL> res = self . _recv ( ) <EOL> self . _sem . release ( ) <EOL> return res <EOL> finally : <EOL> self . _rlock . release ( ) <EOL> def qsize ( self ) : <EOL> return self . _maxsize - self . _sem . _semlock . _get_value ( ) <EOL> def empty ( self ) : <EOL> return not self . _poll ( ) <EOL> def full ( self ) : <EOL> return self . _sem . _semlock . _is_zero ( ) <EOL> def get_nowait ( self ) : <EOL> return self . get ( False ) <EOL> def put_nowait ( self , obj ) : <EOL> return self . put ( obj , False ) <EOL> def close ( self ) : <EOL> self . _closed = True <EOL> self . _reader . close ( ) <EOL> if self . _close : <EOL> self . _close ( ) <EOL> def join_thread ( self ) : <EOL> debug ( '<STR_LIT>' ) <EOL> assert self . _closed <EOL> if self . _jointhread : <EOL> self . _jointhread ( ) <EOL> def cancel_join_thread ( self ) : <EOL> debug ( '<STR_LIT>' ) <EOL> self . _joincancelled = True <EOL> try : <EOL> self . _jointhread . cancel ( ) <EOL> except AttributeError : <EOL> pass <EOL> def _start_thread ( self ) : <EOL> debug ( '<STR_LIT>' ) <EOL> self . _buffer . clear ( ) <EOL> self . _thread = threading . Thread ( <EOL> target = Queue . _feed , <EOL> args = ( self . _buffer , self . _notempty , self . _send , <EOL> self . _wlock , self . _writer . close ) , <EOL> name = '<STR_LIT>' <EOL> ) <EOL> self . _thread . daemon = True <EOL> debug ( '<STR_LIT>' ) <EOL> self . _thread . start ( ) <EOL> debug ( '<STR_LIT>' ) <EOL> created_by_this_process = ( self . _opid == os . getpid ( ) ) <EOL> if not self . _joincancelled and not created_by_this_process : <EOL> self . _jointhread = Finalize ( <EOL> self . _thread , Queue . _finalize_join , <EOL> [ weakref . ref ( self . _thread ) ] , <EOL> exitpriority = - <NUM_LIT:5> <EOL> ) <EOL> self . _close = Finalize ( <EOL> self , Queue . _finalize_close , <EOL> [ self . _buffer , self . _notempty ] , <EOL> exitpriority = <NUM_LIT:10> <EOL> ) <EOL> @ staticmethod <EOL> def _finalize_join ( twr ) : <EOL> debug ( '<STR_LIT>' ) <EOL> thread = twr ( ) <EOL> if thread is not None : <EOL> thread . join ( ) <EOL> debug ( '<STR_LIT>' ) <EOL> else : <EOL> debug ( '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def _finalize_close ( buffer , notempty ) : <EOL> debug ( '<STR_LIT>' ) <EOL> notempty . acquire ( ) <EOL> try : <EOL> buffer . append ( _sentinel ) <EOL> notempty . notify ( ) <EOL> finally : <EOL> notempty . release ( ) <EOL> @ staticmethod <EOL> def _feed ( buffer , notempty , send , writelock , close ) : <EOL> debug ( '<STR_LIT>' ) <EOL> from . util import is_exiting <EOL> nacquire = notempty . acquire <EOL> nrelease = notempty . release <EOL> nwait = notempty . wait <EOL> bpopleft = buffer . popleft <EOL> sentinel = _sentinel <EOL> if sys . platform != '<STR_LIT:win32>' : <EOL> wacquire = writelock . acquire <EOL> wrelease = writelock . release <EOL> else : <EOL> wacquire = None <EOL> try : <EOL> while <NUM_LIT:1> : <EOL> nacquire ( ) <EOL> try : <EOL> if not buffer : <EOL> nwait ( ) <EOL> finally : <EOL> nrelease ( ) <EOL> try : <EOL> while <NUM_LIT:1> : <EOL> obj = bpopleft ( ) <EOL> if obj is sentinel : <EOL> debug ( '<STR_LIT>' ) <EOL> close ( ) <EOL> return <EOL> if wacquire is None : <EOL> send ( obj ) <EOL> else : <EOL> wacquire ( ) <EOL> try : <EOL> send ( obj ) <EOL> finally : <EOL> wrelease ( ) <EOL> except IndexError : <EOL> pass <EOL> except Exception , e : <EOL> try : <EOL> if is_exiting ( ) : <EOL> info ( '<STR_LIT>' , e ) <EOL> else : <EOL> import traceback <EOL> traceback . print_exc ( ) <EOL> except Exception : <EOL> pass <EOL> _sentinel = object ( ) <EOL> class JoinableQueue ( Queue ) : <EOL> def __init__ ( self , maxsize = <NUM_LIT:0> ) : <EOL> Queue . __init__ ( self , maxsize ) <EOL> self . _unfinished_tasks = Semaphore ( <NUM_LIT:0> ) <EOL> self . _cond = Condition ( ) <EOL> def __getstate__ ( self ) : <EOL> return Queue . __getstate__ ( self ) + ( self . _cond , self . _unfinished_tasks ) <EOL> def __setstate__ ( self , state ) : <EOL> Queue . __setstate__ ( self , state [ : - <NUM_LIT:2> ] ) <EOL> self . _cond , self . _unfinished_tasks = state [ - <NUM_LIT:2> : ] <EOL> def put ( self , item , block = True , timeout = None ) : <EOL> Queue . put ( self , item , block , timeout ) <EOL> self . _unfinished_tasks . release ( ) <EOL> def task_done ( self ) : <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> if not self . _unfinished_tasks . acquire ( False ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if self . _unfinished_tasks . _semlock . _is_zero ( ) : <EOL> self . _cond . notify_all ( ) <EOL> finally : <EOL> self . _cond . release ( ) <EOL> def join ( self ) : <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> if not self . _unfinished_tasks . _semlock . _is_zero ( ) : <EOL> self . _cond . wait ( ) <EOL> finally : <EOL> self . _cond . release ( ) <EOL> class SimpleQueue ( object ) : <EOL> def __init__ ( self ) : <EOL> self . _reader , self . _writer = Pipe ( duplex = False ) <EOL> self . _rlock = Lock ( ) <EOL> if sys . platform == '<STR_LIT:win32>' : <EOL> self . _wlock = None <EOL> else : <EOL> self . _wlock = Lock ( ) <EOL> self . _make_methods ( ) <EOL> def empty ( self ) : <EOL> return not self . _reader . poll ( ) <EOL> def __getstate__ ( self ) : <EOL> assert_spawning ( self ) <EOL> return ( self . _reader , self . _writer , self . _rlock , self . _wlock ) <EOL> def __setstate__ ( self , state ) : <EOL> ( self . _reader , self . _writer , self . _rlock , self . _wlock ) = state <EOL> self . _make_methods ( ) <EOL> def _make_methods ( self ) : <EOL> recv = self . _reader . recv <EOL> racquire , rrelease = self . _rlock . acquire , self . _rlock . release <EOL> def get ( ) : <EOL> racquire ( ) <EOL> try : <EOL> return recv ( ) <EOL> finally : <EOL> rrelease ( ) <EOL> self . get = get <EOL> if self . _wlock is None : <EOL> self . put = self . _writer . send <EOL> else : <EOL> send = self . _writer . send <EOL> wacquire , wrelease = self . _wlock . acquire , self . _wlock . release <EOL> def put ( obj ) : <EOL> wacquire ( ) <EOL> try : <EOL> return send ( obj ) <EOL> finally : <EOL> wrelease ( ) <EOL> self . put = put </s>
