<s> from __future__ import print_function <EOL> import argparse <EOL> import logging <EOL> import random <EOL> import time <EOL> from flowy import finish_order , first , LocalWorkflow , parallel_reduce , wait <EOL> from flowy . utils import setup_default_logger <EOL> def activity ( x = None , y = None , identity = None , sleep = None , err = None ) : <EOL> """<STR_LIT>""" <EOL> if sleep is None : <EOL> sleep = random . random ( ) <EOL> print ( '<STR_LIT>' % ( sleep , identity ) ) <EOL> time . sleep ( sleep ) <EOL> print ( '<STR_LIT>' % ( sleep , identity ) ) <EOL> if err is not None : <EOL> raise RuntimeError ( err ) <EOL> if y is None : <EOL> return x * x <EOL> return x + y <EOL> class Sequential ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> r = <NUM_LIT:2> <EOL> t = map ( lambda x : x * time_scale , [ <NUM_LIT:1.0> , <NUM_LIT:0.5> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> for identity , sleep in enumerate ( t , <NUM_LIT:1> ) : <EOL> r = self . a ( r , sleep = sleep , identity = identity ) <EOL> return r <EOL> class ParallelWait ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> results = [ ] <EOL> t = map ( lambda x : x * time_scale , [ <NUM_LIT:1.0> , <NUM_LIT:0.5> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> for i , sleep in enumerate ( t , <NUM_LIT:1> ) : <EOL> results . append ( self . a ( i , identity = i , sleep = sleep ) ) <EOL> for result in results : <EOL> wait ( result ) <EOL> return results [ - <NUM_LIT:2> ] <EOL> class ParallelSum ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> results = [ ] <EOL> t = map ( lambda x : x * time_scale , [ <NUM_LIT:1.0> , <NUM_LIT:0.5> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> for i , sleep in enumerate ( t , <NUM_LIT:1> ) : <EOL> results . append ( self . a ( i , identity = i , sleep = sleep ) ) <EOL> return sum ( results ) <EOL> class WaitFirst ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> x1 = self . a ( <NUM_LIT:1> , sleep = <NUM_LIT:1.0> * time_scale , identity = <NUM_LIT:1> ) <EOL> x2 = self . a ( <NUM_LIT:2> , sleep = <NUM_LIT:0.5> * time_scale , identity = <NUM_LIT:2> ) <EOL> x3 = self . a ( <NUM_LIT:3> , sleep = <NUM_LIT> * time_scale , identity = <NUM_LIT:3> ) <EOL> x4 = self . a ( <NUM_LIT:4> , sleep = <NUM_LIT> * time_scale , identity = <NUM_LIT:4> ) <EOL> return first ( x1 , x2 , x3 , x4 ) <EOL> class WaitFirstTwo ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> x1 = self . a ( <NUM_LIT:1> , sleep = <NUM_LIT:1.0> * time_scale , identity = <NUM_LIT:1> ) <EOL> x2 = self . a ( <NUM_LIT:2> , sleep = <NUM_LIT:0.5> * time_scale , identity = <NUM_LIT:2> ) <EOL> x3 = self . a ( <NUM_LIT:3> , sleep = <NUM_LIT> * time_scale , identity = <NUM_LIT:3> ) <EOL> x4 = self . a ( <NUM_LIT:4> , sleep = <NUM_LIT> * time_scale , identity = <NUM_LIT:4> ) <EOL> f_o = list ( finish_order ( x1 , x2 , x3 , x4 ) ) <EOL> return f_o [ <NUM_LIT:0> ] + f_o [ <NUM_LIT:1> ] <EOL> class Conditional ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> x = self . a ( <NUM_LIT:10> , identity = <NUM_LIT:1> , sleep = <NUM_LIT:1.0> * time_scale ) <EOL> if x >= <NUM_LIT:100> : <EOL> return self . a ( <NUM_LIT:20> , identity = '<STR_LIT>' , sleep = <NUM_LIT:1.0> * time_scale ) <EOL> return <NUM_LIT:1> <EOL> class NaiveMapReduce ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> reduce_f = lambda x , y : self . a ( x , y , <EOL> sleep = <NUM_LIT> * time_scale , <EOL> identity = '<STR_LIT>' % ( x , y ) ) <EOL> t = map ( lambda x : x * time_scale , [ <NUM_LIT:0.5> , <NUM_LIT> , <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> map_f = lambda x , sleep : self . a ( x , sleep = sleep , identity = '<STR_LIT>' % x ) <EOL> results = list ( map ( map_f , range ( <NUM_LIT:1> , <NUM_LIT:7> ) , t ) ) <EOL> v = results [ <NUM_LIT:0> ] <EOL> for r in results [ <NUM_LIT:1> : ] : <EOL> v = reduce_f ( v , r ) <EOL> return v <EOL> class FinishOrderMapReduce ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> reduce_f = lambda x , y : self . a ( x , y , <EOL> sleep = <NUM_LIT> * time_scale , <EOL> identity = '<STR_LIT>' % ( x , y ) ) <EOL> t = map ( lambda x : x * time_scale , [ <NUM_LIT:0.5> , <NUM_LIT> , <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> map_f = lambda x , sleep : self . a ( x , sleep = sleep , identity = '<STR_LIT>' % x ) <EOL> results = finish_order ( map ( map_f , range ( <NUM_LIT:1> , <NUM_LIT:7> ) , t ) ) <EOL> v = next ( results ) <EOL> for r in results : <EOL> v = reduce_f ( v , r ) <EOL> return v <EOL> class ParallelMapReduce ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , a ) : <EOL> self . a = a <EOL> def __call__ ( self , time_scale = <NUM_LIT:1> ) : <EOL> reduce_f = lambda x , y : self . a ( x , y , <EOL> sleep = <NUM_LIT> * time_scale , <EOL> identity = '<STR_LIT>' % ( x , y ) ) <EOL> t = map ( lambda x : x * time_scale , [ <NUM_LIT:0.5> , <NUM_LIT> , <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> map_f = lambda x , sleep : self . a ( x , sleep = sleep , identity = '<STR_LIT>' % x ) <EOL> results = map ( map_f , range ( <NUM_LIT:1> , <NUM_LIT:7> ) , t ) <EOL> return parallel_reduce ( reduce_f , results ) <EOL> def main ( ) : <EOL> logging . basicConfig ( ) <EOL> setup_default_logger ( ) <EOL> def workflow ( workflow_class ) : <EOL> try : <EOL> return globals ( ) [ workflow_class ] <EOL> except KeyError : <EOL> raise ValueError ( '<STR_LIT>' % workflow_class ) <EOL> parser = argparse . ArgumentParser ( description = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , action = '<STR_LIT:store>' , type = workflow ) <EOL> parser . add_argument ( '<STR_LIT>' , action = '<STR_LIT:store_true>' , default = False ) <EOL> parser . add_argument ( '<STR_LIT>' , <EOL> action = '<STR_LIT:store>' , <EOL> type = int , <EOL> default = <NUM_LIT:2> ) <EOL> parser . add_argument ( '<STR_LIT>' , <EOL> action = '<STR_LIT:store>' , <EOL> type = int , <EOL> default = <NUM_LIT:8> ) <EOL> parser . add_argument ( '<STR_LIT>' , action = '<STR_LIT:store_true>' , default = False ) <EOL> parser . add_argument ( '<STR_LIT>' , action = '<STR_LIT:store_true>' , default = False ) <EOL> parser . add_argument ( '<STR_LIT>' , action = '<STR_LIT:store_true>' , default = False ) <EOL> args = parser . parse_args ( ) <EOL> start = time . time ( ) <EOL> if args . pure : <EOL> wf = args . workflow ( activity ) <EOL> result = wf ( ) <EOL> else : <EOL> lw = LocalWorkflow ( args . workflow , <EOL> activity_workers = args . activity_workers , <EOL> workflow_workers = args . workflow_workers ) <EOL> lw . conf_activity ( '<STR_LIT:a>' , activity ) <EOL> result = lw . run ( _wait = args . wait_children , _trace = args . trace ) <EOL> if args . timeit : <EOL> print ( '<STR_LIT>' , time . time ( ) - start ) <EOL> print ( '<STR_LIT>' , result ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
