<s> from __future__ import absolute_import <EOL> from __future__ import unicode_literals <EOL> import logging <EOL> import operator <EOL> import sys <EOL> from threading import Thread <EOL> from docker . errors import APIError <EOL> from six . moves import _thread as thread <EOL> from six . moves . queue import Empty <EOL> from six . moves . queue import Queue <EOL> from compose . cli . signals import ShutdownException <EOL> from compose . utils import get_output_stream <EOL> log = logging . getLogger ( __name__ ) <EOL> STOP = object ( ) <EOL> def parallel_execute ( objects , func , get_name , msg , get_deps = None ) : <EOL> """<STR_LIT>""" <EOL> objects = list ( objects ) <EOL> stream = get_output_stream ( sys . stderr ) <EOL> writer = ParallelStreamWriter ( stream , msg ) <EOL> for obj in objects : <EOL> writer . initialize ( get_name ( obj ) ) <EOL> events = parallel_execute_iter ( objects , func , get_deps ) <EOL> errors = { } <EOL> results = [ ] <EOL> error_to_reraise = None <EOL> for obj , result , exception in events : <EOL> if exception is None : <EOL> writer . write ( get_name ( obj ) , '<STR_LIT>' ) <EOL> results . append ( result ) <EOL> elif isinstance ( exception , APIError ) : <EOL> errors [ get_name ( obj ) ] = exception . explanation <EOL> writer . write ( get_name ( obj ) , '<STR_LIT:error>' ) <EOL> elif isinstance ( exception , UpstreamError ) : <EOL> writer . write ( get_name ( obj ) , '<STR_LIT:error>' ) <EOL> else : <EOL> errors [ get_name ( obj ) ] = exception <EOL> error_to_reraise = exception <EOL> for obj_name , error in errors . items ( ) : <EOL> stream . write ( "<STR_LIT>" . format ( obj_name , error ) ) <EOL> if error_to_reraise : <EOL> raise error_to_reraise <EOL> return results <EOL> def _no_deps ( x ) : <EOL> return [ ] <EOL> class State ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , objects ) : <EOL> self . objects = objects <EOL> self . started = set ( ) <EOL> self . finished = set ( ) <EOL> self . failed = set ( ) <EOL> def is_done ( self ) : <EOL> return len ( self . finished ) + len ( self . failed ) >= len ( self . objects ) <EOL> def pending ( self ) : <EOL> return set ( self . objects ) - self . started - self . finished - self . failed <EOL> def parallel_execute_iter ( objects , func , get_deps ) : <EOL> """<STR_LIT>""" <EOL> if get_deps is None : <EOL> get_deps = _no_deps <EOL> results = Queue ( ) <EOL> state = State ( objects ) <EOL> while True : <EOL> feed_queue ( objects , func , get_deps , results , state ) <EOL> try : <EOL> event = results . get ( timeout = <NUM_LIT:0.1> ) <EOL> except Empty : <EOL> continue <EOL> except thread . error : <EOL> raise ShutdownException ( ) <EOL> if event is STOP : <EOL> break <EOL> obj , _ , exception = event <EOL> if exception is None : <EOL> log . debug ( '<STR_LIT>' . format ( obj ) ) <EOL> state . finished . add ( obj ) <EOL> else : <EOL> log . debug ( '<STR_LIT>' . format ( obj ) ) <EOL> state . failed . add ( obj ) <EOL> yield event <EOL> def producer ( obj , func , results ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> result = func ( obj ) <EOL> results . put ( ( obj , result , None ) ) <EOL> except Exception as e : <EOL> results . put ( ( obj , None , e ) ) <EOL> def feed_queue ( objects , func , get_deps , results , state ) : <EOL> """<STR_LIT>""" <EOL> pending = state . pending ( ) <EOL> log . debug ( '<STR_LIT>' . format ( pending ) ) <EOL> for obj in pending : <EOL> deps = get_deps ( obj ) <EOL> if any ( dep in state . failed for dep in deps ) : <EOL> log . debug ( '<STR_LIT>' . format ( obj ) ) <EOL> results . put ( ( obj , None , UpstreamError ( ) ) ) <EOL> state . failed . add ( obj ) <EOL> elif all ( <EOL> dep not in objects or dep in state . finished <EOL> for dep in deps <EOL> ) : <EOL> log . debug ( '<STR_LIT>' . format ( obj ) ) <EOL> t = Thread ( target = producer , args = ( obj , func , results ) ) <EOL> t . daemon = True <EOL> t . start ( ) <EOL> state . started . add ( obj ) <EOL> if state . is_done ( ) : <EOL> results . put ( STOP ) <EOL> class UpstreamError ( Exception ) : <EOL> pass <EOL> class ParallelStreamWriter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , stream , msg ) : <EOL> self . stream = stream <EOL> self . msg = msg <EOL> self . lines = [ ] <EOL> def initialize ( self , obj_index ) : <EOL> if self . msg is None : <EOL> return <EOL> self . lines . append ( obj_index ) <EOL> self . stream . write ( "<STR_LIT>" . format ( self . msg , obj_index ) ) <EOL> self . stream . flush ( ) <EOL> def write ( self , obj_index , status ) : <EOL> if self . msg is None : <EOL> return <EOL> position = self . lines . index ( obj_index ) <EOL> diff = len ( self . lines ) - position <EOL> self . stream . write ( "<STR_LIT>" % ( <NUM_LIT> , diff ) ) <EOL> self . stream . write ( "<STR_LIT>" % <NUM_LIT> ) <EOL> self . stream . write ( "<STR_LIT>" . format ( self . msg , obj_index , status ) ) <EOL> self . stream . write ( "<STR_LIT>" % ( <NUM_LIT> , diff ) ) <EOL> self . stream . flush ( ) <EOL> def parallel_operation ( containers , operation , options , message ) : <EOL> parallel_execute ( <EOL> containers , <EOL> operator . methodcaller ( operation , ** options ) , <EOL> operator . attrgetter ( '<STR_LIT:name>' ) , <EOL> message ) <EOL> def parallel_remove ( containers , options ) : <EOL> stopped_containers = [ c for c in containers if not c . is_running ] <EOL> parallel_operation ( stopped_containers , '<STR_LIT>' , options , '<STR_LIT>' ) <EOL> def parallel_start ( containers , options ) : <EOL> parallel_operation ( containers , '<STR_LIT:start>' , options , '<STR_LIT>' ) <EOL> def parallel_pause ( containers , options ) : <EOL> parallel_operation ( containers , '<STR_LIT>' , options , '<STR_LIT>' ) <EOL> def parallel_unpause ( containers , options ) : <EOL> parallel_operation ( containers , '<STR_LIT>' , options , '<STR_LIT>' ) <EOL> def parallel_kill ( containers , options ) : <EOL> parallel_operation ( containers , '<STR_LIT>' , options , '<STR_LIT>' ) <EOL> def parallel_restart ( containers , options ) : <EOL> parallel_operation ( containers , '<STR_LIT>' , options , '<STR_LIT>' ) </s>
