<s> from __future__ import print_function <EOL> import sys , time , os <EOL> import ginga . util . six as six <EOL> if six . PY2 : <EOL> import thread <EOL> import Queue <EOL> else : <EOL> import _thread as thread <EOL> import queue as Queue <EOL> _swival = <NUM_LIT> <EOL> sys . setswitchinterval ( _swival ) <EOL> import threading <EOL> import traceback <EOL> from ginga . util . six . moves import map , zip <EOL> class TaskError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class TaskTimeout ( TaskError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class UserTaskException ( Exception ) : <EOL> pass <EOL> class Task ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ev_done = threading . Event ( ) <EOL> self . callbacks = Queue . Queue ( ) <EOL> self . tag = None <EOL> self . logger = None <EOL> self . threadPool = None <EOL> self . tlock = threading . RLock ( ) <EOL> self . shares = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> super ( Task , self ) . __init__ ( ) <EOL> def initialize ( self , taskParent , override = None ) : <EOL> """<STR_LIT>""" <EOL> if taskParent and hasattr ( taskParent , '<STR_LIT>' ) : <EOL> for var in taskParent . shares : <EOL> if override and var in override : <EOL> self . __dict__ [ var ] = override [ var ] <EOL> else : <EOL> self . __dict__ [ var ] = taskParent . __dict__ [ var ] <EOL> else : <EOL> pass <EOL> if not self . tag : <EOL> try : <EOL> self . tag = str ( taskParent ) + '<STR_LIT:.>' + self . tagger . get_tag ( self ) <EOL> except : <EOL> self . tag = get_tag ( taskParent ) <EOL> self . ev_done . clear ( ) <EOL> self . starttime = time . time ( ) <EOL> self . endtime = <NUM_LIT:0> <EOL> self . totaltime = <NUM_LIT:0> <EOL> self . result = None <EOL> return self . tag <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . threadPool : <EOL> self . threadPool . addTask ( self ) <EOL> time . sleep ( <NUM_LIT:0> ) <EOL> else : <EOL> raise TaskError ( "<STR_LIT>" % self ) <EOL> def init_and_start ( self , taskParent , override = { } ) : <EOL> """<STR_LIT>""" <EOL> tag = self . initialize ( taskParent , override = override ) <EOL> self . start ( ) <EOL> return tag <EOL> def check_state ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def extend_shares ( self , varlist ) : <EOL> shares = set ( self . shares ) <EOL> for var in varlist : <EOL> if hasattr ( self , var ) : <EOL> shares . add ( var ) <EOL> self . shares = shares <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> raise TaskError ( "<STR_LIT>" % ( <EOL> self ) ) <EOL> def pause ( self ) : <EOL> """<STR_LIT>""" <EOL> raise TaskError ( "<STR_LIT>" % ( <EOL> self ) ) <EOL> def resume ( self ) : <EOL> """<STR_LIT>""" <EOL> raise TaskError ( "<STR_LIT>" % ( <EOL> self ) ) <EOL> def wait ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> self . ev_done . wait ( timeout = timeout ) <EOL> if not self . ev_done . isSet ( ) : <EOL> raise TaskTimeout ( "<STR_LIT>" % self ) <EOL> if isinstance ( self . result , Exception ) : <EOL> raise self . result <EOL> return self . result <EOL> def step ( self ) : <EOL> """<STR_LIT>""" <EOL> raise TaskError ( "<STR_LIT>" % self ) <EOL> def execute ( self ) : <EOL> """<STR_LIT>""" <EOL> raise TaskError ( "<STR_LIT>" % self ) <EOL> def done ( self , result , noraise = False ) : <EOL> """<STR_LIT>""" <EOL> if self . ev_done . isSet ( ) : <EOL> if isinstance ( self . result , Exception ) and ( not noraise ) : <EOL> raise self . result <EOL> return self . result <EOL> self . endtime = time . time ( ) <EOL> try : <EOL> self . totaltime = self . endtime - self . starttime <EOL> except AttributeError : <EOL> self . totaltime = <NUM_LIT:0.0> <EOL> self . result = result <EOL> self . ev_done . set ( ) <EOL> self . do_callbacks ( ) <EOL> if isinstance ( result , Exception ) and ( not noraise ) : <EOL> raise result <EOL> return result <EOL> def register_callback ( self , fn , args = None ) : <EOL> """<STR_LIT>""" <EOL> if args is None : <EOL> args = [ ] <EOL> if callable ( fn ) : <EOL> self . callbacks . put ( ( fn , args ) ) <EOL> else : <EOL> raise TaskError ( "<STR_LIT>" % str ( fn ) ) <EOL> def do_callbacks ( self ) : <EOL> """<STR_LIT>""" <EOL> while not self . callbacks . empty ( ) : <EOL> ( fn , rest ) = self . callbacks . get ( ) <EOL> args = [ self . result ] <EOL> args . extend ( rest ) <EOL> fn ( * args ) <EOL> def get_tag ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . tag <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return str ( self . tag ) <EOL> def __lt__ ( self , other ) : <EOL> return False <EOL> def getExecutionTime ( self ) : <EOL> return self . totaltime <EOL> def runTask ( self , task , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> task . initialize ( self ) <EOL> task . start ( ) <EOL> time . sleep ( <NUM_LIT:0> ) <EOL> res = task . wait ( timeout = timeout ) <EOL> return res <EOL> def run ( self , task , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> return self . runTask ( task , timeout = timeout ) <EOL> class printTask ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , msg ) : <EOL> self . msg = msg <EOL> super ( printTask , self ) . __init__ ( ) <EOL> def execute ( self ) : <EOL> print ( self . msg ) <EOL> class sleepTask ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , delay ) : <EOL> self . delay = delay <EOL> super ( sleepTask , self ) . __init__ ( ) <EOL> def execute ( self ) : <EOL> self . ev_done . wait ( timeout = self . delay ) <EOL> class FuncTask ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , func , args , kwdargs , logger = None ) : <EOL> self . func = func <EOL> self . args = args <EOL> self . kwdargs = kwdargs <EOL> self . logger = logger <EOL> super ( FuncTask , self ) . __init__ ( ) <EOL> def execute ( self ) : <EOL> if self . logger : <EOL> s_args = str ( self . args ) <EOL> if len ( s_args ) > <NUM_LIT> : <EOL> s_args = s_args [ : <NUM_LIT> ] <EOL> s_kwdargs = str ( self . kwdargs ) <EOL> if len ( s_kwdargs ) > <NUM_LIT> : <EOL> s_kwdargs = s_kwdargs [ : <NUM_LIT> ] <EOL> self . logger . debug ( "<STR_LIT>" % ( <EOL> self . func . __name__ , s_args , s_kwdargs ) ) <EOL> s_args = None <EOL> s_kwdargs = None <EOL> try : <EOL> res = self . func ( * self . args , ** self . kwdargs ) <EOL> self . done ( res ) <EOL> if self . logger : <EOL> self . logger . debug ( "<STR_LIT>" % ( <EOL> str ( res ) ) ) <EOL> except Exception as e : <EOL> if self . logger : <EOL> self . logger . error ( "<STR_LIT>" % ( str ( self ) , str ( e ) ) ) <EOL> try : <EOL> ( type , value , tb ) = sys . exc_info ( ) <EOL> self . logger . error ( "<STR_LIT>" % "<STR_LIT>" . join ( traceback . format_tb ( tb ) ) ) <EOL> tb = None <EOL> except Exception : <EOL> self . logger . error ( "<STR_LIT>" ) <EOL> self . done ( e ) <EOL> class FuncTask2 ( FuncTask ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , func , * args , ** kwdargs ) : <EOL> super ( FuncTask2 , self ) . __init__ ( func , args , kwdargs ) <EOL> def set_logger ( self , logger ) : <EOL> self . logger = logger <EOL> def make_tasker ( func ) : <EOL> """<STR_LIT>""" <EOL> def anonFunc ( * args , ** kwdargs ) : <EOL> class anonTask ( Task ) : <EOL> def execute ( self ) : <EOL> self . logger . debug ( "<STR_LIT>" % func ) <EOL> try : <EOL> val = func ( * args , ** kwdargs ) <EOL> self . logger . debug ( "<STR_LIT>" % func ) <EOL> return val <EOL> except Exception as e : <EOL> self . logger . error ( "<STR_LIT>" % ( <EOL> func , str ( e ) ) ) <EOL> raise e <EOL> return anonTask ( ) <EOL> return anonFunc <EOL> class SequentialTaskset ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , taskseq ) : <EOL> super ( SequentialTaskset , self ) . __init__ ( ) <EOL> self . tasklist = list ( taskseq ) <EOL> def initialize ( self , taskParent , ** kwdargs ) : <EOL> self . index = <NUM_LIT:0> <EOL> super ( SequentialTaskset , self ) . initialize ( taskParent , ** kwdargs ) <EOL> def step ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . index >= len ( self . tasklist ) : <EOL> raise TaskError ( "<STR_LIT>" % self ) <EOL> self . check_state ( ) <EOL> self . task = self . tasklist [ self . index ] <EOL> self . index += <NUM_LIT:1> <EOL> return self . runTask ( self . task ) <EOL> def execute ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . index < len ( self . tasklist ) : <EOL> res = self . step ( ) <EOL> self . logger . debug ( '<STR_LIT>' % ( self . index , res ) ) <EOL> return res <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . task . stop ( ) <EOL> except TaskError as e : <EOL> self . logger . error ( "<STR_LIT>" % ( str ( e ) ) ) <EOL> def addTask ( self , task ) : <EOL> """<STR_LIT>""" <EOL> self . tasklist . append ( task ) <EOL> class oldConcurrentAndTaskset ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , taskseq ) : <EOL> super ( oldConcurrentAndTaskset , self ) . __init__ ( ) <EOL> self . taskseq = taskseq <EOL> self . ev_intr = threading . Event ( ) <EOL> self . regcond = threading . Condition ( ) <EOL> def execute ( self ) : <EOL> """<STR_LIT>""" <EOL> self . count = <NUM_LIT:0> <EOL> self . taskset = [ ] <EOL> self . results = { } <EOL> self . totaltime = time . time ( ) <EOL> for task in self . taskseq : <EOL> self . taskset . append ( task ) <EOL> task . register_callback ( self . child_done , args = [ self . count , task ] ) <EOL> self . count += <NUM_LIT:1> <EOL> self . numtasks = self . count <EOL> with self . regcond : <EOL> for task in self . taskset : <EOL> task . initialize ( self ) <EOL> task . start ( ) <EOL> self . totaltime = time . time ( ) - self . totaltime <EOL> while self . count > <NUM_LIT:0> : <EOL> self . regcond . wait ( ) <EOL> for key in self . results . keys ( ) : <EOL> value = self . results [ key ] <EOL> if isinstance ( value , Exception ) : <EOL> ( count , task ) = key <EOL> self . logger . error ( "<STR_LIT>" % ( <EOL> task . tag , str ( value ) ) ) <EOL> raise value <EOL> return <NUM_LIT:0> <EOL> def child_done ( self , result , count , task ) : <EOL> """<STR_LIT>""" <EOL> with self . regcond : <EOL> self . logger . debug ( '<STR_LIT>' % ( <EOL> self . count , self . numtasks ) ) <EOL> self . count -= <NUM_LIT:1> <EOL> self . taskset . remove ( task ) <EOL> self . totaltime += task . getExecutionTime ( ) <EOL> self . results [ ( count , task ) ] = result <EOL> if self . count <= <NUM_LIT:0> : <EOL> self . regcond . notifyAll ( ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> for task in self . taskset : <EOL> try : <EOL> task . stop ( ) <EOL> except TaskError as e : <EOL> pass <EOL> def addTask ( self , task ) : <EOL> """<STR_LIT>""" <EOL> with self . regcond : <EOL> self . taskset . append ( task ) <EOL> task . register_callback ( self . child_done , args = [ self . numtasks , task ] ) <EOL> self . numtasks += <NUM_LIT:1> <EOL> self . count += <NUM_LIT:1> <EOL> task . initialize ( self ) <EOL> task . start ( ) <EOL> class newConcurrentAndTaskset ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , taskseq ) : <EOL> super ( newConcurrentAndTaskset , self ) . __init__ ( ) <EOL> self . taskseq = taskseq <EOL> self . idletime = <NUM_LIT> <EOL> self . _lock_c = threading . RLock ( ) <EOL> def execute ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock_c : <EOL> self . count = <NUM_LIT:0> <EOL> self . numtasks = <NUM_LIT:0> <EOL> self . taskset = [ ] <EOL> self . results = { } <EOL> self . totaltime = time . time ( ) <EOL> for task in self . taskseq : <EOL> self . taskset . append ( task ) <EOL> self . numtasks += <NUM_LIT:1> <EOL> task . init_and_start ( self ) <EOL> num_tasks = self . getNumTasks ( ) <EOL> while num_tasks > <NUM_LIT:0> : <EOL> self . check_state ( ) <EOL> for i in range ( num_tasks ) : <EOL> try : <EOL> try : <EOL> task = self . getTask ( i ) <EOL> except IndexError : <EOL> break <EOL> res = task . wait ( timeout = self . idletime ) <EOL> self . child_done ( res , task ) <EOL> except TaskTimeout : <EOL> continue <EOL> except Exception as e : <EOL> self . child_done ( e , task ) <EOL> continue <EOL> num_tasks = self . getNumTasks ( ) <EOL> for key in self . results . keys ( ) : <EOL> value = self . results [ key ] <EOL> if isinstance ( value , Exception ) : <EOL> ( count , task ) = key <EOL> self . logger . error ( "<STR_LIT>" % ( <EOL> task . tag , str ( value ) ) ) <EOL> raise value <EOL> return value <EOL> def child_done ( self , result , task ) : <EOL> with self . _lock_c : <EOL> self . count += <NUM_LIT:1> <EOL> self . logger . debug ( '<STR_LIT>' % ( <EOL> self . count , self . numtasks ) ) <EOL> self . taskset . remove ( task ) <EOL> self . totaltime += task . getExecutionTime ( ) <EOL> self . results [ ( self . count , task ) ] = result <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock_c : <EOL> for task in self . taskset : <EOL> try : <EOL> task . stop ( ) <EOL> except TaskError as e : <EOL> pass <EOL> def addTask ( self , task ) : <EOL> """<STR_LIT>""" <EOL> task . initialize ( self ) <EOL> task . start ( ) <EOL> with self . _lock_c : <EOL> self . numtasks += <NUM_LIT:1> <EOL> self . taskset . append ( task ) <EOL> def getTask ( self , i ) : <EOL> with self . _lock_c : <EOL> return self . taskset [ i ] <EOL> def getNumTasks ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock_c : <EOL> return len ( self . taskset ) <EOL> class ConcurrentAndTaskset ( newConcurrentAndTaskset ) : <EOL> pass <EOL> class QueueTaskset ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , queue , waitflag = True , timeout = <NUM_LIT:0.1> , ev_quit = None ) : <EOL> super ( QueueTaskset , self ) . __init__ ( ) <EOL> self . queue = queue <EOL> self . waitflag = waitflag <EOL> self . lock = threading . RLock ( ) <EOL> self . timeout = timeout <EOL> self . task = None <EOL> self . ev_cancel = threading . Event ( ) <EOL> self . ev_pause = threading . Event ( ) <EOL> if ev_quit == None : <EOL> ev_quit = threading . Event ( ) <EOL> self . ev_quit = ev_quit <EOL> def flush ( self ) : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> while True : <EOL> try : <EOL> self . queue . get ( block = False ) <EOL> except Queue . Empty : <EOL> break <EOL> def stop ( self ) : <EOL> self . flush ( ) <EOL> try : <EOL> if self . task : <EOL> self . task . stop ( ) <EOL> except TaskError as e : <EOL> pass <EOL> self . queue . put ( None ) <EOL> def stop_child ( self ) : <EOL> self . flush ( ) <EOL> try : <EOL> if self . task : <EOL> self . task . stop ( ) <EOL> except TaskError as e : <EOL> pass <EOL> def execute ( self ) : <EOL> self . count = <NUM_LIT:0> <EOL> self . totaltime = <NUM_LIT:0> <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> while not self . ev_quit . isSet ( ) : <EOL> try : <EOL> self . check_state ( ) <EOL> task = self . queue . get ( block = True , timeout = self . timeout ) <EOL> if task == None : <EOL> break <EOL> self . task = task <EOL> task . register_callback ( self . child_done , args = [ task ] ) <EOL> with self . lock : <EOL> self . count += <NUM_LIT:1> <EOL> self . ev_cancel . clear ( ) <EOL> try : <EOL> task . initialize ( self ) <EOL> self . logger . debug ( "<STR_LIT>" % str ( task ) ) <EOL> task . start ( ) <EOL> if self . waitflag : <EOL> res = task . wait ( ) <EOL> self . logger . debug ( "<STR_LIT>" % ( <EOL> ( str ( task ) , str ( res ) ) ) ) <EOL> except Exception as e : <EOL> self . logger . error ( "<STR_LIT>" % ( str ( task ) , str ( e ) ) ) <EOL> try : <EOL> ( type , value , tb ) = sys . exc_info ( ) <EOL> self . logger . debug ( "<STR_LIT>" % "<STR_LIT>" . join ( traceback . format_tb ( tb ) ) ) <EOL> tb = None <EOL> except Exception as e : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> task . done ( e , noraise = True ) <EOL> except Queue . Empty : <EOL> continue <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> return self . result <EOL> def child_done ( self , result , task ) : <EOL> with self . lock : <EOL> self . count -= <NUM_LIT:1> <EOL> self . totaltime += task . getExecutionTime ( ) <EOL> self . result = result <EOL> def cancel ( self ) : <EOL> self . flush ( ) <EOL> super ( QueueTaskset , self ) . cancel ( ) <EOL> def addTask ( self , task ) : <EOL> self . queue . put ( task ) <EOL> class PriorityQueue ( Queue . PriorityQueue ) : <EOL> pass <EOL> class _WorkerReset ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class WorkerThread ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , queue , logger = None , ev_quit = None , <EOL> timeout = <NUM_LIT> , tpool = None ) : <EOL> self . queue = queue <EOL> self . logger = logger <EOL> self . timeout = timeout <EOL> if ev_quit : <EOL> self . ev_quit = ev_quit <EOL> else : <EOL> self . ev_quit = threading . Event ( ) <EOL> self . tpool = tpool <EOL> self . lock = threading . RLock ( ) <EOL> self . status = '<STR_LIT>' <EOL> self . time_start = <NUM_LIT:0.0> <EOL> def setstatus ( self , status ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . status = status <EOL> def getstatus ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> return ( self . status , self . time_start ) <EOL> def execute ( self , task ) : <EOL> """<STR_LIT>""" <EOL> taskid = str ( task ) <EOL> res = None <EOL> try : <EOL> self . time_start = time . time ( ) <EOL> self . setstatus ( '<STR_LIT>' % taskid ) <EOL> self . logger . debug ( "<STR_LIT>" % taskid ) <EOL> try : <EOL> res = task . execute ( ) <EOL> except UserTaskException as e : <EOL> res = e <EOL> except Exception as e : <EOL> self . logger . error ( "<STR_LIT>" % ( str ( task ) , str ( e ) ) ) <EOL> res = e <EOL> try : <EOL> ( type , value , tb ) = sys . exc_info ( ) <EOL> self . logger . debug ( "<STR_LIT>" % "<STR_LIT>" . join ( traceback . format_tb ( tb ) ) ) <EOL> tb = None <EOL> except Exception as e : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> finally : <EOL> self . logger . debug ( "<STR_LIT>" % str ( task ) ) <EOL> self . setstatus ( '<STR_LIT>' % taskid ) <EOL> task . done ( res , noraise = True ) <EOL> self . time_start = <NUM_LIT:0.0> <EOL> self . setstatus ( '<STR_LIT>' ) <EOL> def taskloop ( self ) : <EOL> self . setstatus ( '<STR_LIT>' ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> if self . tpool : <EOL> self . tpool . register_up ( ) <EOL> try : <EOL> self . setstatus ( '<STR_LIT>' ) <EOL> while not self . ev_quit . isSet ( ) : <EOL> try : <EOL> ( priority , task ) = self . queue . get ( block = True , <EOL> timeout = self . timeout ) <EOL> if task == None : <EOL> self . queue . put ( ( priority , task ) ) <EOL> break <EOL> self . execute ( task ) <EOL> except _WorkerReset : <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> except Queue . Empty as e : <EOL> pass <EOL> finally : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> if self . tpool : <EOL> self . tpool . register_dn ( ) <EOL> self . setstatus ( '<STR_LIT>' ) <EOL> def start ( self ) : <EOL> self . thread = threading . Thread ( target = self . taskloop , args = [ ] ) <EOL> self . thread . start ( ) <EOL> def stop ( self ) : <EOL> self . queue . put ( ( priority , task ) ) <EOL> self . ev_quit . set ( ) <EOL> class ThreadPool ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , numthreads = <NUM_LIT:1> , logger = None , ev_quit = None , <EOL> workerClass = WorkerThread ) : <EOL> self . numthreads = numthreads <EOL> self . logger = logger <EOL> if ev_quit : <EOL> self . ev_quit = ev_quit <EOL> else : <EOL> self . ev_quit = threading . Event ( ) <EOL> self . lock = threading . RLock ( ) <EOL> self . workerClass = workerClass <EOL> self . queue = PriorityQueue ( ) <EOL> self . workers = [ ] <EOL> self . tids = [ ] <EOL> self . regcond = threading . Condition ( ) <EOL> self . runningcount = <NUM_LIT:0> <EOL> self . status = '<STR_LIT>' <EOL> def startall ( self , wait = False , ** kwdargs ) : <EOL> """<STR_LIT>""" <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> with self . regcond : <EOL> while self . status != '<STR_LIT>' : <EOL> if self . status in ( '<STR_LIT:start>' , '<STR_LIT>' ) or self . ev_quit . isSet ( ) : <EOL> self . logger . error ( "<STR_LIT>" ) <EOL> return <EOL> self . logger . debug ( "<STR_LIT>" % self . runningcount ) <EOL> self . regcond . wait ( ) <EOL> if self . ev_quit . isSet ( ) : <EOL> return <EOL> self . runningcount = <NUM_LIT:0> <EOL> self . status = '<STR_LIT:start>' <EOL> self . workers = [ ] <EOL> if wait : <EOL> tpool = self <EOL> else : <EOL> tpool = None <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> for i in range ( self . numthreads ) : <EOL> t = self . workerClass ( self . queue , logger = self . logger , <EOL> ev_quit = self . ev_quit , tpool = tpool , <EOL> ** kwdargs ) <EOL> self . workers . append ( t ) <EOL> t . start ( ) <EOL> if wait : <EOL> while self . status != '<STR_LIT>' and not self . ev_quit . isSet ( ) : <EOL> self . logger . debug ( "<STR_LIT>" % self . runningcount ) <EOL> self . regcond . wait ( ) <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> def addThreads ( self , numthreads , ** kwdargs ) : <EOL> with self . regcond : <EOL> self . logger . debug ( "<STR_LIT>" % ( <EOL> numthreads ) ) <EOL> for i in range ( numthreads ) : <EOL> t = self . workerClass ( self . queue , logger = self . logger , <EOL> ev_quit = self . ev_quit , tpool = self . tpool , <EOL> ** kwdargs ) <EOL> self . workers . append ( t ) <EOL> t . start ( ) <EOL> self . numthreads += numthreads <EOL> def stopall ( self , wait = False ) : <EOL> """<STR_LIT>""" <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> with self . regcond : <EOL> while self . status != '<STR_LIT>' : <EOL> if self . status in ( '<STR_LIT>' , '<STR_LIT>' ) or self . ev_quit . isSet ( ) : <EOL> self . logger . warning ( "<STR_LIT>" ) <EOL> return <EOL> self . logger . debug ( "<STR_LIT>" % self . runningcount ) <EOL> self . regcond . wait ( ) <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> self . status = '<STR_LIT>' <EOL> self . ev_quit . set ( ) <EOL> if wait : <EOL> while self . status != '<STR_LIT>' : <EOL> self . logger . debug ( "<STR_LIT>" % self . runningcount ) <EOL> self . regcond . wait ( ) <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> def workerStatus ( self ) : <EOL> return list ( map ( lambda t : t . getstatus ( ) , self . workers ) ) <EOL> def addTask ( self , task , priority = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> self . queue . put ( ( priority , task ) ) <EOL> def delTask ( self , taskid ) : <EOL> self . logger . error ( "<STR_LIT>" ) <EOL> def purgeTasks ( self ) : <EOL> self . logger . error ( "<STR_LIT>" ) <EOL> def register_up ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . regcond : <EOL> self . runningcount += <NUM_LIT:1> <EOL> tid = thread . get_ident ( ) <EOL> self . tids . append ( tid ) <EOL> self . logger . debug ( "<STR_LIT>" % ( tid , self . runningcount ) ) <EOL> if self . runningcount == self . numthreads : <EOL> self . status = '<STR_LIT>' <EOL> self . regcond . notify ( ) <EOL> def register_dn ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . regcond : <EOL> self . runningcount -= <NUM_LIT:1> <EOL> tid = thread . get_ident ( ) <EOL> self . tids . remove ( tid ) <EOL> self . logger . debug ( "<STR_LIT>" % self . runningcount ) <EOL> self . logger . debug ( "<STR_LIT>" % str ( self . tids ) ) <EOL> if self . runningcount == <NUM_LIT:0> : <EOL> self . status = '<STR_LIT>' <EOL> self . regcond . notify ( ) <EOL> _lock_seqnum = threading . Lock ( ) <EOL> _count_seqnum = <NUM_LIT:0> <EOL> def get_tag ( taskParent ) : <EOL> global _count_seqnum <EOL> with _lock_seqnum : <EOL> generic_id = '<STR_LIT>' % ( _count_seqnum ) <EOL> _count_seqnum += <NUM_LIT:1> <EOL> if taskParent : <EOL> tag = str ( taskParent ) + '<STR_LIT:.>' + generic_id <EOL> else : <EOL> tag = generic_id <EOL> return tag </s>
