<s> import collections <EOL> import multiprocessing <EOL> import threading <EOL> import six <EOL> from six . moves import _thread <EOL> from taskflow . utils import misc <EOL> def is_alive ( thread ) : <EOL> """<STR_LIT>""" <EOL> if not thread : <EOL> return False <EOL> return thread . is_alive ( ) <EOL> def get_ident ( ) : <EOL> """<STR_LIT>""" <EOL> return _thread . get_ident ( ) <EOL> def get_optimal_thread_count ( default = <NUM_LIT:2> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return multiprocessing . cpu_count ( ) + <NUM_LIT:1> <EOL> except NotImplementedError : <EOL> return default <EOL> def daemon_thread ( target , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> thread = threading . Thread ( target = target , args = args , kwargs = kwargs ) <EOL> thread . daemon = True <EOL> return thread <EOL> _ThreadBuilder = collections . namedtuple ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> _ThreadBuilder . fields = tuple ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) <EOL> def no_op ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> class ThreadBundle ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _threads = [ ] <EOL> self . _lock = threading . Lock ( ) <EOL> def bind ( self , thread_factory , <EOL> before_start = None , after_start = None , <EOL> before_join = None , after_join = None ) : <EOL> """<STR_LIT>""" <EOL> if before_start is None : <EOL> before_start = no_op <EOL> if after_start is None : <EOL> after_start = no_op <EOL> if before_join is None : <EOL> before_join = no_op <EOL> if after_join is None : <EOL> after_join = no_op <EOL> builder = _ThreadBuilder ( thread_factory , <EOL> before_start , after_start , <EOL> before_join , after_join ) <EOL> for attr_name in builder . fields : <EOL> cb = getattr ( builder , attr_name ) <EOL> if not six . callable ( cb ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" % attr_name ) <EOL> with self . _lock : <EOL> self . _threads . append ( [ <EOL> builder , <EOL> None , <EOL> False , <EOL> ] ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> count = <NUM_LIT:0> <EOL> with self . _lock : <EOL> it = enumerate ( self . _threads ) <EOL> for i , ( builder , thread , started ) in it : <EOL> if thread and started : <EOL> continue <EOL> if not thread : <EOL> self . _threads [ i ] [ <NUM_LIT:1> ] = thread = builder . thread_factory ( ) <EOL> builder . before_start ( thread ) <EOL> thread . start ( ) <EOL> count += <NUM_LIT:1> <EOL> try : <EOL> builder . after_start ( thread ) <EOL> finally : <EOL> self . _threads [ i ] [ <NUM_LIT:2> ] = started = True <EOL> return count <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> count = <NUM_LIT:0> <EOL> with self . _lock : <EOL> it = misc . reverse_enumerate ( self . _threads ) <EOL> for i , ( builder , thread , started ) in it : <EOL> if not thread or not started : <EOL> continue <EOL> builder . before_join ( thread ) <EOL> thread . join ( ) <EOL> count += <NUM_LIT:1> <EOL> try : <EOL> builder . after_join ( thread ) <EOL> finally : <EOL> self . _threads [ i ] [ <NUM_LIT:1> ] = thread = None <EOL> self . _threads [ i ] [ <NUM_LIT:2> ] = started = False <EOL> return count <EOL> def __len__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . _threads ) </s>
