5^dZddlmZddlmZddlmZddlZddlZddlZddlZddl Z ddl m Z ddl Z ddl mZddl mZdd l mZdZd ZGd d eZGd deZe j,ej.GddeZe j,ej.GddeZGddeZGddeZGddeZGddeZGddeZGddeZ Gdd eZ!Gd!d"eZ"Gd#d$ejFZ$Gd%d&eZ%d'Z&y)(a.Parallel execution pools based on multithreading. This module provides 2 types of pools: - NullPool: executes work synchronously, in the current process - ThreadPool: executes work across multiple threads It also contains a convenience method GetPool to get the appropriate pool for the given number of threads. The general usage is as follows: >>> def identity(value): return value >>> with parallel.GetPool(num_threads) as pool: ... future = pool.ApplyAsync(identity, (42,)) ... assert future.Get() == 42 ... assert pool.Apply(f, (42,)) == 42 ... map_future = pool.MapAsync(identity, [1, 2, 3]) ... assert map_future.Get() == [1, 2, 3] ... assert pool.Map(identity, [1, 2, 3]) == [1, 2, 3] Errors are raised at the time of the Get() call on the future (which is implicit for Apply() and Map()). )absolute_import)division)unicode_literalsN) exceptions)map)queue)rangeg{Gz?ceZdZdZy)UnsupportedPlatformExceptionzHException indicating that a pool was created on an unsupported platform.N)__name__ __module__ __qualname____doc__(lib/googlecloudsdk/core/util/parallel.pyr r AsPrr c"eZdZdZfdZxZS)InvalidStateExceptionzGException to indicate that a parallel pool was put in an invalid state.c,tt| |yN)superr__init__)selfmsg __class__s rrzInvalidStateException.__init__Hs /4r)r r rrr __classcell__rs@rrrEsO55rrceZdZdZej dZej dZdZdZ dZ dZ ej dZ d Z d Zy ) BasePoolzBase class for parallel pools. Provides a limited subset of the multiprocessing.Pool API. Can be used as a context manager: >>> with pool: ... assert pool.Map(str, [1, 2, 3]) == ['1', '2', '3'] ct)zFInitialize non-trivial infrastructure (e.g. processes/threads/queues).NotImplementedErrorrs rStartzBasePool.StartX  rct)z%Clean up anything started in Start().r!r#s rJoinz BasePool.Join]r%rcB|j||jS)zFApplies func to each element in iterable and return a list of results.)MapAsyncGetrfunciterables rMapz BasePool.Mapbs ==x ( , , ..rc `t|Dcgc]}|j||fc}Scc}w)z=Applies func to each element in iterable and return a future.) _MultiFuture ApplyAsync)rr,r-args rr)zBasePool.MapAsyncfs, (K(3v6(K LLKs+cB|j||jS)aApplies func to each element in iterable and return a generator. The generator yields the result immediately after the task is done. So result for faster task will be yielded earlier than for slower task. Args: func: a function object iterable: an iterable object and each element is the arguments to func Returns: A generator to produce the results. )r)GetResultsEagerFetchr+s r MapEagerFetchzBasePool.MapEagerFetchjs ==x ( = = ??rcB|j||jS)z,Applies func to args and returns the result.)r1r*rr,argss rApplyzBasePool.Applyys ??4 & * * ,,rct)z'Apply func to args and return a future.r!r7s rr1zBasePool.ApplyAsync}r%rc&|j|Sr)r$r#s r __enter__zBasePool.__enter__sJJL Krc$|jyr)r')rexc_typeexc_valexc_tbs r__exit__zBasePool.__exit__s IIKrN)r r rrabcabstractmethodr$r'r.r)r5r9r1r<rArrrrrLsu/M @-rrc^eZdZdZdZej dZej dZy) BaseFuturezAA future object containing a value that may not be available yet.c>|jjSr GetResult GetOrRaiser#s rr*zBaseFuture.Gets >>  & & ((rctrr!r#s rrHzBaseFuture.GetResult rctrr!r#s rDonezBaseFuture.DonerKrN) r r rrr*rBrCrHrMrrrrErEs>I)rrEc*eZdZdZddZdZdZdZy)_ResultaValue holder for a result (a value, if successful, or an error). Only one of {value, error, exc_info} can be set. Both error and exc_info exist due to issues with pickling. exc_info is better, because we can re-raise it and preserve the original stacktrace, but it can't be pickled. error gets re-raised from GetOrRaise(). Attributes: result: one-tuple of any object (optional), the result of the function. It's a one-tuple to distinguish a result of None from no result. error: Exception (optional), an exception that was thrown by the function exc_info: exc_info (optional) for the exception that occurred Ncttt|||gdkDr td|s|s |s td||_||_||_y)Nz:_Result may only have one of [value, error, exc_info] set.z6_Result must have one of [value, error, exc_info] set.)sumrbool ValueErrorvalueerrorexc_info)rrUrVrWs rrz_Result.__init__sW 3teUH- ./!3   Uh O PPDJDJDMrc|jr|jdS|jr |jtj|jd|jdy)NrrQ)tb)rUrVrreraiserWr#s rrIz_Result.GetOrRaisesJ zz ZZ]  JJq)dmmA.>?rc |jrt|jd}n|} tj||S#tj$r}t|cYd}~Sd}~wt $rL}ttjdj |tj|cYd}~Sd}~wwxYw)zReturn a pickleable version of this _Result. Traceback objects can't be pickled, so we just pass through the exc_value. Also, some values and exceptions can't be pickled. Returns: _Result: a pickleable version of this result. rQrVNz!Couldn't pickle result [{0}]: {1}) rWrOpickledumps PicklingError Exceptionformatsix text_type)rpickleable_resulterrs rToPickleableResultz_Result.ToPickleableResults }}! a(897 ll$%     3  7 6// - 4 4s!3 56777s0AB= A%B=% B=1AB82B=8B=cddj|j|j|jS)Nz+_Result(value={0}, error={1}, exc_info={2}))rbrUrVrWr#s r__str__z_Result.__str__s( 8 ? ? DJJ  //r)NNN)r r rrrrIrgrirrrrOrOs @0/rrOceZdZfdZxZS) MultiErrorc x||_d}tt|ddj t ||zy)Ncrdjt|jtj|S)Nz{}: {})rbtyper rcrd)es rz%MultiError.__init__..s"8??47#3#3S]]15EFrzOne or more errors occurred: z )errorsrrkrjoinr)rrqfnrs rrzMultiError.__init__s8DK FB *d$( CFO$ %&r)r r rrrrs@rrkrks &&rrkc(eZdZdZdZdZdZdZy)r0zFuture object that combines other Future objects. Returns the results of each future when they are all ready. Attributes: futures: list of BaseFuture. c||_yr)futures)rrvs rrz_MultiFuture.__init__s DLrcg}g}|jD]"} |j|j$|rt t |St |fS#t$r}|j|Yd}~hd}~wwxYw)Nr])rU)rvappendr*rarOrk)rresultsrqfuturerfs rrHz_MultiFuture.GetResultstG F,,vzz|$  :f- .. ' $$  csA A>#A99A>cnt|jDcgc]}|jc}Scc}wr)allrvrM)rrzs rrMz_MultiFuture.Dones) T\\:\6\: ;;:s2c#K|j}|r]g}|D]7}|jr |j'|j |9|}t j t|r\yy#t$r}|Yd}~ld}~wwxYww)zCollect the results of futures. Results are yielded immediately after the task is done. So result for faster task will be yielded earlier than for slower task. Yields: result which is done. N)rvrMr*rarxtimesleep_POLL_INTERVAL)runcollected_futurenext_uncollected_futurerzrfs rr4z!_MultiFuture.GetResultsEagerFetchs~  "&& ;;= **,  " ( ( 0'3 jj    I s3&B A/1B -B / B8B<B BB N)r r rrrrHrMr4rrrr0r0s %<!rr0c(eZdZdZdZdZdZdZy)_TaskzAn individual work unit to be performed in parallel. Attributes: func: callable, a function to be called with the given arguments. Must be serializable. args: tuple, the arguments to pass to func. Must be serializable. c ||_||_yr)r,r8r7s rrz_Task.__init__sDIDIrcXt|jj|jfSr)hashr,r r8r#s r__hash__z_Task.__hash__#s ##TYY/ 00rc|jj|jjk(xr|j|jk(Sr)r,r r8rothers r__eq__z _Task.__eq__&s3 99  !4!4 4 Pejj9PPrc&|j| Sr)rrs r__ne__z _Task.__ne__)s{{5! !!rN)r r rrrrrrrrrrrs1Q"rrceZdZdZdZdZy) _NullFuturec||_yrresult)rrs rrz_NullFuture.__init__4s DKrc|jSrrr#s rrHz_NullFuture.GetResult7s ;;rcy)NTrr#s rrMz_NullFuture.Done:s rN)r r rrrHrMrrrrr2srrc(eZdZdZdZdZdZdZy)NullPoolz)Serial analog of parallel execution Pool.cd|_y)NF)_startedr#s rrzNullPool.__init__As DMrc|js td t||f}t |S#ttj}Yt |SxYw)N&NullPool must be Start()ed before use.rW)rrrOsysrWr)rr,r8rs rr1zNullPool.ApplyAsyncDs[ == ""J KK0d ~&f v 0 /f v s 3 Ac@|jr tdd|_y)NzCan only start NullPool once.Trrr#s rr$zNullPool.StartPs }} !"A BBDMrc2|js tdy)Nrrr#s rr'z NullPool.JoinUs == ""J KK rN)r r rrrr1r$r'rrrrr>s1  Lrrc$eZdZdZdZdZdZy) _ThreadFuturec ||_||_yr_task _results_map)rtask results_maps rrz_ThreadFuture.__init__esDJ#Drc>|jjS)z6Return the value of the future, or raise an exception.rGr#s rr*z_ThreadFuture.Getis >>  & & ((rc |j|jvr|j|jStjtK)zGet the _Result of the future.)rrr~rrr#s rrHz_ThreadFuture.GetResultms=  t(( (  ,, jj  rc2|j|jvS)z8Return True if the task finished with or without errors.rr#s rrMz_ThreadFuture.Donets ::** **rN)r r rrr*rHrMrrrrrcs$)!+rrceZdZdZy) _ThreadTaskc||_yr)r)rrs rrz_ThreadTask.__init__{s DIrN)r r rrrrrrrysrrc$eZdZfdZdZxZS) _WorkerThreadcFtt| ||_||_yr)rrr work_queuer)rrrrs rrz_WorkerThread.__init__s -') DO"Drc( |jj}|tury|j} t |j |j f}||j|j<m#t tj}Yrs;0''  *  Q9Q5I53;;:v: :z3;;    </f</~&&0!:0!f"F"8 * LxLJ+J+,& 2I$$2(  P#r