<s> import random <EOL> import threading <EOL> from oslo_utils import reflection <EOL> from oslo_utils import timeutils <EOL> import six <EOL> from taskflow . engines . worker_based import protocol as pr <EOL> from taskflow import logging <EOL> from taskflow . utils import kombu_utils as ku <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class TopicWorker ( object ) : <EOL> """<STR_LIT>""" <EOL> _NO_IDENTITY = object ( ) <EOL> def __init__ ( self , topic , tasks , identity = _NO_IDENTITY ) : <EOL> self . tasks = [ ] <EOL> for task in tasks : <EOL> if not isinstance ( task , six . string_types ) : <EOL> task = reflection . get_class_name ( task ) <EOL> self . tasks . append ( task ) <EOL> self . topic = topic <EOL> self . identity = identity <EOL> self . last_seen = None <EOL> def performs ( self , task ) : <EOL> if not isinstance ( task , six . string_types ) : <EOL> task = reflection . get_class_name ( task ) <EOL> return task in self . tasks <EOL> def __eq__ ( self , other ) : <EOL> if not isinstance ( other , TopicWorker ) : <EOL> return NotImplemented <EOL> if len ( other . tasks ) != len ( self . tasks ) : <EOL> return False <EOL> if other . topic != self . topic : <EOL> return False <EOL> for task in other . tasks : <EOL> if not self . performs ( task ) : <EOL> return False <EOL> if self . _NO_IDENTITY in ( self . identity , other . identity ) : <EOL> return True <EOL> else : <EOL> return other . identity == self . identity <EOL> def __repr__ ( self ) : <EOL> r = reflection . get_class_name ( self , fully_qualified = False ) <EOL> if self . identity is not self . _NO_IDENTITY : <EOL> r += "<STR_LIT>" % ( self . identity , <EOL> self . tasks , self . topic ) <EOL> else : <EOL> r += "<STR_LIT>" % ( self . tasks , self . topic ) <EOL> return r <EOL> class ProxyWorkerFinder ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , uuid , proxy , topics , <EOL> beat_periodicity = pr . NOTIFY_PERIOD , <EOL> worker_expiry = pr . EXPIRES_AFTER ) : <EOL> self . _cond = threading . Condition ( ) <EOL> self . _proxy = proxy <EOL> self . _topics = topics <EOL> self . _workers = { } <EOL> self . _uuid = uuid <EOL> self . _seen_workers = <NUM_LIT:0> <EOL> self . _messages_processed = <NUM_LIT:0> <EOL> self . _messages_published = <NUM_LIT:0> <EOL> self . _worker_expiry = worker_expiry <EOL> self . _watch = timeutils . StopWatch ( duration = beat_periodicity ) <EOL> @ property <EOL> def total_workers ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . _workers ) <EOL> def wait_for_workers ( self , workers = <NUM_LIT:1> , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if workers <= <NUM_LIT:0> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> watch = timeutils . StopWatch ( duration = timeout ) <EOL> watch . start ( ) <EOL> with self . _cond : <EOL> while self . total_workers < workers : <EOL> if watch . expired ( ) : <EOL> return max ( <NUM_LIT:0> , workers - self . total_workers ) <EOL> self . _cond . wait ( watch . leftover ( return_none = True ) ) <EOL> return <NUM_LIT:0> <EOL> @ staticmethod <EOL> def _match_worker ( task , available_workers ) : <EOL> """<STR_LIT>""" <EOL> if len ( available_workers ) == <NUM_LIT:1> : <EOL> return available_workers [ <NUM_LIT:0> ] <EOL> else : <EOL> return random . choice ( available_workers ) <EOL> @ property <EOL> def messages_processed ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _messages_processed <EOL> def _next_worker ( self , topic , tasks , temporary = False ) : <EOL> if not temporary : <EOL> w = TopicWorker ( topic , tasks , identity = self . _seen_workers ) <EOL> self . _seen_workers += <NUM_LIT:1> <EOL> return w <EOL> else : <EOL> return TopicWorker ( topic , tasks ) <EOL> def maybe_publish ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _messages_published == <NUM_LIT:0> : <EOL> self . _proxy . publish ( pr . Notify ( ) , <EOL> self . _topics , reply_to = self . _uuid ) <EOL> self . _messages_published += <NUM_LIT:1> <EOL> self . _watch . restart ( ) <EOL> else : <EOL> if self . _watch . expired ( ) : <EOL> self . _proxy . publish ( pr . Notify ( ) , <EOL> self . _topics , reply_to = self . _uuid ) <EOL> self . _messages_published += <NUM_LIT:1> <EOL> self . _watch . restart ( ) <EOL> def _add ( self , topic , tasks ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> worker = self . _workers [ topic ] <EOL> if worker == self . _next_worker ( topic , tasks , temporary = True ) : <EOL> return ( worker , False ) <EOL> except KeyError : <EOL> pass <EOL> worker = self . _next_worker ( topic , tasks ) <EOL> self . _workers [ topic ] = worker <EOL> return ( worker , True ) <EOL> def process_response ( self , data , message ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( "<STR_LIT>" , <EOL> ku . DelayedPretty ( message ) ) <EOL> response = pr . Notify ( ** data ) <EOL> LOG . debug ( "<STR_LIT>" , response ) <EOL> with self . _cond : <EOL> worker , new_or_updated = self . _add ( response . topic , <EOL> response . tasks ) <EOL> if new_or_updated : <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , worker , self . total_workers ) <EOL> self . _cond . notify_all ( ) <EOL> worker . last_seen = timeutils . now ( ) <EOL> self . _messages_processed += <NUM_LIT:1> <EOL> def clean ( self ) : <EOL> """<STR_LIT>""" <EOL> if ( not self . _workers or <EOL> ( self . _worker_expiry is None or self . _worker_expiry <= <NUM_LIT:0> ) ) : <EOL> return <NUM_LIT:0> <EOL> dead_workers = { } <EOL> with self . _cond : <EOL> now = timeutils . now ( ) <EOL> for topic , worker in six . iteritems ( self . _workers ) : <EOL> if worker . last_seen is None : <EOL> continue <EOL> secs_since_last_seen = max ( <NUM_LIT:0> , now - worker . last_seen ) <EOL> if secs_since_last_seen >= self . _worker_expiry : <EOL> dead_workers [ topic ] = ( worker , secs_since_last_seen ) <EOL> for topic in six . iterkeys ( dead_workers ) : <EOL> self . _workers . pop ( topic ) <EOL> if dead_workers : <EOL> self . _cond . notify_all ( ) <EOL> if dead_workers and LOG . isEnabledFor ( logging . INFO ) : <EOL> for worker , secs_since_last_seen in six . itervalues ( dead_workers ) : <EOL> LOG . info ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> worker , secs_since_last_seen ) <EOL> return len ( dead_workers ) <EOL> def reset ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _cond : <EOL> self . _workers . clear ( ) <EOL> self . _messages_processed = <NUM_LIT:0> <EOL> self . _messages_published = <NUM_LIT:0> <EOL> self . _seen_workers = <NUM_LIT:0> <EOL> self . _cond . notify_all ( ) <EOL> def get_worker_for_task ( self , task ) : <EOL> """<STR_LIT>""" <EOL> available_workers = [ ] <EOL> with self . _cond : <EOL> for worker in six . itervalues ( self . _workers ) : <EOL> if worker . performs ( task ) : <EOL> available_workers . append ( worker ) <EOL> if available_workers : <EOL> return self . _match_worker ( task , available_workers ) <EOL> else : <EOL> return None </s>
