<s> '''<STR_LIT>''' <EOL> import threading <EOL> from cola . core . rpc import client_call <EOL> from cola . core . utils import get_rpc_prefix <EOL> class Stage ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , workers , func , prefix = None , app_name = None , logger = None ) : <EOL> self . workers = workers <EOL> self . func = func <EOL> self . prefix = get_rpc_prefix ( app_name , prefix ) <EOL> self . remote_func = self . prefix + func <EOL> self . logger = logger <EOL> def barrier ( self , parallel , * args ) : <EOL> results = [ False ] * len ( self . workers ) <EOL> def _call ( i , worker ) : <EOL> try : <EOL> result = client_call ( worker , self . remote_func , * args ) <EOL> except Exception , e : <EOL> if self . logger : <EOL> self . logger . error ( e ) <EOL> result = False <EOL> if isinstance ( result , bool ) : <EOL> results [ i ] = result <EOL> else : <EOL> results [ i ] = True <EOL> if not parallel : <EOL> for i , worker in enumerate ( self . workers ) : <EOL> _call ( i , worker ) <EOL> else : <EOL> threads = [ ] <EOL> for i , worker in enumerate ( self . workers ) : <EOL> t = threading . Thread ( target = _call , args = ( i , worker , ) ) <EOL> t . setDaemon ( True ) <EOL> threads . append ( t ) <EOL> for t in threads : <EOL> t . start ( ) <EOL> for t in threads : <EOL> t . join ( ) <EOL> return all ( results ) </s>
