S `,-ddlZddlZddlZddlZddlmZddlmZddlm Z ddlm Z ddlm Z ddl m Z ddl mZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZej8eZeZ GddZ!dZ"d'dZ#dZ$GddZ%GddZ&dZ'dZ(dZ)Gdde*Z+Gd d!Z,d"Z-d#Z.d$Z/d%Z0d&Z1y)(N)Empty)Queue)Lock) Semaphore)Thread)APIError) ImageNotFound)AnsiMode)green)red)ShutdownException)PARALLEL_LIMIT)CompletedUnsuccessfully)HealthCheckFailed)NoHealthCheckConfigured)OperationFailedErrorc0eZdZdZeeZedZy) GlobalLimitzSimple class to hold a global semaphore limiter for a project. This class should be treated as a singleton that is instantiated when the project is. c4|t}t||_yN)rrglobal_limiter)clsvalues 2/usr/lib/python3/dist-packages/compose/parallel.pyset_global_limitzGlobalLimit.set_global_limit%s ="E&u-N) __name__ __module__ __qualname____doc__rrr classmethodrrrrrs%~.N..rrcd}|D]~\}} } | Y|'||r|j|||dtn|j|||dt|j| ct | t r6| j |||<|j|||dt| }t | tr4| j |||<|j|||dtt | ttttfr5| j|||<|j|||dtBt | tr |j|||dtr| |||<| }|S)z Watch events from a parallel execution, update status and fill errors and results. Returns exception to re-raise. Nfaileddoneerror)writer r append isinstancer explanationrrrrrmsg UpstreamError) eventswritererrorsresultsr+get_name fail_checkerror_to_reraiseobjresult exceptions rparallel_execute_watchr7,sY"()VY  %*S/ S(3-3? S(3-? NN6 "  = 1%.$9$9F8C= ! LLhsmWc :(   8 ,$-$9$9F8C= ! LLhsmWc :  $8:KMd$;$=>$-MMF8C= ! LLhsmWc :  = 1 LLhsmWc :$-F8C= !( 1)2 rc t|}tj}tj t|}|D]} |j ||| |D]} |j ||| t||||} i} g} t| || | |||} | jD]&\}}|jdj||(| r| | | fS)aRuns func on objects in parallel while ensuring that func is ran on object only after it is ran on all its dependencies. get_deps called on object must return a collection with its dependencies. get_name called on object must return its name. fail_check is an additional failure check for cases that should display as a failure in the CLI logs, but don't raise an exception (such as attempting to start 0 containers) z ERROR: for {} {} ) listsysstderrParallelStreamWriterget_or_assign_instance add_object write_initialparallel_execute_iterr7itemsr'format)objectsfuncr1r+get_depslimitr2streamr.r4r-r/r0r3obj_namer&s rparallel_executerIMs7mG ZZF ! 8 89Mf9U VF.#x}-.1S(3-01#7D(E BF FG-h "<<>F% ,33HeDEF F?rcgSrr")xs r_no_depsrLqs Irc"eZdZdZdZdZdZy)Statez Holds the state of a partially-complete parallel operation. state.started: objects being processed state.finished: objects which have been processed state.failed: objects which either failed or whose dependencies failed cl||_t|_t|_t|_yr)rCsetstartedfinishedr$)selfrCs r__init__zState.__init__}s$ u  e rct|jt|jzt|jk\Sr)lenrRr$rCrSs ris_donez State.is_dones-4==!C $44DLL8IIIrczt|j|jz |jz |jz Sr)rPrCrQrRr$rWs rpendingz State.pendings,4<< 4<</$--?$++MMrN)rrrr rTrXrZr"rrrNrNusJNrrNceZdZdZdZy)NoLimitcyrr"rWs r __enter__zNoLimit.__enter__ rcyrr")rSexs r__exit__zNoLimit.__exit__r_rN)rrrr^rbr"rrr\r\s   rr\c#<K|t}| t}n t|}t}t |} t |||||| |j d}|tury|\}} } | @tjdj||jj!|n?tjdj||j"j!||#t$rYtj$r twxYww)a Runs func on objects in parallel while ensuring that func is ran on object only after it is ran on all its dependencies. Returns an iterator of tuples which look like: # if func returned normally when run on object (object, result, None) # if func raised an exception when run on object (object, None, exception) # if func raised an exception when run on one of object's dependencies (object, None, UpstreamError()) Ng?)timeoutzFinished processing: {}z Failed: {})rLr\rrrN feed_queuegetrthreadr&r STOPlogdebugrBrRaddr$) rCrDrErFlimiterr0stateeventr4_r6s rr@r@s  })E"gG 'NE 7D(GUGD &KKK,E D= !Q   II/66s; < NN  s # IIl))#. / LL  S ! -    || &#% % &s+AD C1BD1 D:D<DDc |5tj5 ||}|j||dfddddddy#t$r}|j|d|fYd}~3d}~wwxYw#1swYzfeed_queue..s:c3q6U\\):sz'{} has upstream errors - not processingNc3jK|]*\}}|vxs|jvxr | xs||,ywr)rR)rwrx ready_checkrCrms rryzfeed_queue..sK'c;7"5>>)S;/R+cBRs03zStarting producer thread for {})targetargsTzAHealthcheck for service(s) upstream of {} failed - not processingzIService(s) upstream of {} did not completed successfully - not processing)rZrirjrBanyrqr,r$rkallrrtdaemonstartrQrrrrXrh) rCrDrEr0rmrlrZr4depstrss ` ` rreresmmoGIIm""7+,(} (:T:: CJJ3OP S$ 89   %+/  ;BB3GH(#tWg1NO  !!#&!(< }} D"#:; ( II!!'  KKdA ' '& ( II!!'  KKdA ' '  (s%C0EG:-8F++ G:78G55G:c eZdZy)r,N)rrrr"rrr,r,srr,ceZdZdZej ZeZdZ eZ e dZ e dZ e dZd dZdZdZd Zd Zd Zy) r<zWrite out messages for operations happening in parallel. Each operation has its own line, and ANSI code characters are used to jump to the correct line, and write over the line. Nc|jSr)instance)rs r get_instancez!ParallelStreamWriter.get_instances ||rc|jj |j||_|j|jjS#|jjwxYwr) instance_lockacquirerrelease)rr.s rr=z+ParallelStreamWriter.get_or_assign_instancesV !!# (||#% <<    % % 'C   % % 's AA1c||_yr)default_ansi_mode)r ansi_modes rset_default_ansi_modez*ParallelStreamWriter.set_default_ansi_mode s )rcv| |j}||_|j||_g|_d|_y)Nr)rrGuse_ansi_codeslineswidth)rSrGrs rrTzParallelStreamWriter.__init__$s;  ..I '66v>  rc|y|jj||zt|jt |dz|z|_y)N )rr(maxrrVrSr+ obj_indexs rr>zParallelStreamWriter.add_object,s@ ;  # /*SsY)>%?@ rc.|y|j||dS)N) _write_noansirs rr?z"ParallelStreamWriter.write_initial2s ; !!#y"55rcB|jj|jj||z}t |j|z }|j j dd|fz|j j ddz|j j dj|dz|z||j|j j dd|fz|j j|jjy)Nz%c[%dAz%c[2K z{:<{width}} ... {} rrz%c[%dB) write_lockrrindexrVrGr'rBrflushr)rSr+rstatuspositiondiffs r _write_ansiz ParallelStreamWriter._write_ansi7s !::##C)O44::) (b$Z/0 )b.) 077c I8M  84 5 (b$Z/0  !rc|jjdj|dz|z||j|jj y)Nz{:<{width}} ... {} rr)rGr'rBrr)rSr+rrs rrz"ParallelStreamWriter._write_noansiFsM  $ + +c I%vTZZ ,  rc||y|jr|j||||y|j|||yr)rrr)rSr+rr color_funcs rr'zParallelStreamWriter.writeNs< ;       S)Z-? @   sIv 6rr)rrrr r AUTOrrrrrr!rr=rrTr>r?rrr'r"rrr<r<s ! JHFM((**A 6 "7rr<cnt|tj|fi|tjd|y)Nname)rIoperator methodcaller attrgetter) containers operationoptionsmessages rparallel_operationrWs1i373F# rcb|Dcgc]}|jr|}}t|d|dycc}w)NremoveRemoving) is_runningr)rrcstopped_containerss rparallel_remover`s1%/Dq||!DD)8WjIEs,,c t|d|dy)NpausePausingrrrs rparallel_pauseresz7GY?rc t|d|dy)Nunpause Unpausingrrs rparallel_unpauserisz9g{Crc t|d|dy)NkillKillingrrs r parallel_killrmsz67I>r)NNN)2_threadrgloggingrr:queuerr threadingrrr docker.errorsrr compose.cli.colorsr r r compose.cli.signalsr compose.constrcompose.errorsrrrr getLoggerrriobjectrhrr7rIrLrNr\r@rtrerrr,r<rrrrrr"rrrs "''$"1(2,2/g! x . .B!HNN,  1h (*Z I O7O7dJ @D?r