5^SrSSKJr SSKJr SSKJr SSKrSSKrSSKrSSKrSSK r SSK J r SSK r SSK Jr SSK Jr SS K Jr SrS r"S S \5r"S S\5r\ R,"\R.5"SS\55r\ R,"\R.5"SS\55r"SS\5r"SS\5r"SS\5r"SS\5r"SS\5r"SS\5r "SS \5r!"S!S"\5r""S#S$\RF5r$"S%S&\5r%S'r&g)(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?c\rSrSrSrSrg)UnsupportedPlatformExceptionAzHException indicating that a pool was created on an unsupported platform.N)__name__ __module__ __qualname____firstlineno____doc____static_attributes__r (lib/googlecloudsdk/core/util/parallel.pyr r AsPrr c,^\rSrSrSrU4SjrSrU=r$)InvalidStateExceptionEzGException to indicate that a parallel pool was put in an invalid state.c,>[[U] U5 gN)superr__init__)selfmsg __class__s rrInvalidStateException.__init__Hs /4rr )rrrrrrr __classcell__rs@rrrEsO55rrc\rSrSrSr\R S5r\R S5rSr Sr Sr Sr \R S 5r S rS rS rg )BasePoolLzBase 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'] c[e)zFInitialize non-trivial infrastructure (e.g. processes/threads/queues).NotImplementedErrorrs rStartBasePool.StartX  rc[e)z%Clean up anything started in Start().r'r)s rJoin BasePool.Join]r,rc@URX5R5$)zFApplies func to each element in iterable and return a list of results.)MapAsyncGetrfunciterables rMap BasePool.Mapbs == ( , , ..rc b[UVs/sHo0RX45PM sn5$s snf)z=Applies func to each element in iterable and return a future.) _MultiFuture ApplyAsync)rr4r5args rr1BasePool.MapAsyncfs( (K(3v6(K LLKs,c@URX5R5$)ahApplies 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. )r1GetResultsEagerFetchr3s r MapEagerFetchBasePool.MapEagerFetchjs == ( = = ??rc@URX5R5$)z,Applies func to args and returns the result.)r:r2rr4argss rApplyBasePool.Applyys ??4 & * * ,,rc[e)z'Apply func to args and return a future.r'rBs rr:BasePool.ApplyAsync}r,rc&UR5 U$r)r*r)s r __enter__BasePool.__enter__sJJL Krc$UR5 gr)r.)rexc_typeexc_valexc_tbs r__exit__BasePool.__exit__s IIKrr N)rrrrrabcabstractmethodr*r.r6r1r?rDr:rIrOrr rrr$r$Lsu/M @-rr$cf\rSrSrSrSr\RS5r\RS5r Sr g) BaseFuturezAA future object containing a value that may not be available yet.c>UR5R5$r GetResult GetOrRaiser)s rr2BaseFuture.Gets >>  & & ((rc[err'r)s rrXBaseFuture.GetResult rc[err'r)s rDoneBaseFuture.Doner]rr N) rrrrrr2rQrRrXr_rr rrrTrTs>I)rrTc4\rSrSrSrS SjrSrSrSrSr g) _ResultamValue 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 Nc[[[XU/55S:a [S5eU(dU(dU(d [S5eXlX lX0lg)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)rrirjrks rr_Result.__init__sO 3teH- ./!3   Uh O PPJJMrcUR(aURS$UR(a URe[R"URSURSS9 g)Nrre)tb)rirjrreraiserkr)s rrY_Result.GetOrRaisesJ zz ZZ]  JJq)dmmA.>?rc UR(a[URSS9nOUn[R"U5 U$![Ran[US9sSnA$SnAf[ aKn[[R"SR U[R"U555S9sSnA$SnAff=f)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. rerjNz!Couldn't pickle result [{0}]: {1}) rkrbpickledumps PicklingError Exceptionformatsix text_type)rpickleable_resulterrs rToPickleableResult_Result.ToPickleableResults }}! a(897 ll$%     3  7 6// - 4 4s!3 56777s0AB>A& B>& B>3AB93B>9B>cdSRURURUR5$)Nz+_Result(value={0}, error={1}, exc_info={2}))rxrirjrkr)s r__str___Result.__str__s( 8 ? ? DJJ  //r)rjrkri)NNN) rrrrrrrYr}rrr rrrbrbs @0/rrbc(^\rSrSrU4SjrSrU=r$) MultiErrorc t>XlSn[[U]SSR [ X!55-5 g)NctSR[U5R[R"U55$)Nz{}: {})rxtyperryrz)es r%MultiError.__init__..s"8??47#3#3S]]15EFrzOne or more errors occurred: z )errorsrrrjoinr)rrfnrs rrMultiError.__init__s4K FB *d$( CO$ %&r)r)rrrrrrr!r"s@rrrs &&rrc0\rSrSrSrSrSrSrSrSr g) r9zFuture object that combines other Future objects. Returns the results of each future when they are all ready. Attributes: futures: list of BaseFuture. cXlgrfutures)rrs rr_MultiFuture.__init__sLrc/n/nURH#nURUR55 M% U(a[ [ U5S9$[ U4S9$![anURU5 SnAMkSnAff=f)Nrs)ri)rappendr2rwrbr)rresultsrfuturer|s rrX_MultiFuture.GetResultstG F,,vzz|$  :f- .. ' $$  csA B%A<<Bcr[URVs/sHoR5PM sn5$s snfr)allrr_)rrs rr__MultiFuture.Dones' T\\:\6\: ;;:s4c#B# URnU(al/nUH>nUR5(aUR5v M-UR U5 M@ Un[ R "[5 U(aMkgg![anUv SnAM}SnAff=f7f)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)rr_r2rwrtimesleep_POLL_INTERVAL)runcollected_futurenext_uncollected_futurerr|s rr>!_MultiFuture.GetResultsEagerFetchs  "&& ;;== **,  " ( ( 0'3 jj     II s40BB:BB B BBBBrN) rrrrrrrXr_r>rr rrr9r9s %<!rr9c0\rSrSrSrSrSrSrSrSr g) _TaskizAn 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. cXlX lgr)r4rCrBs rr_Task.__init__s IIrcX[URRUR45$r)hashr4rrCr)s r__hash___Task.__hash__#s ##TYY/ 00rcURRURR:H=(a URUR:H$r)r4rrCrothers r__eq__ _Task.__eq__&s3 99  !4!4 4 Pejj9PPrc.URU5(+$r)rrs r__ne__ _Task.__ne__)s{{5! !!r)rCr4N) rrrrrrrrrrr rrrrs1Q"rrc&\rSrSrSrSrSrSrg) _NullFuturei2cXlgrresult)rrs rr_NullFuture.__init__4sKrcUR$rrr)s rrX_NullFuture.GetResult7s ;;rcg)NTr r)s rr__NullFuture.Done:s rrN)rrrrrrXr_rr rrrr2srrc0\rSrSrSrSrSrSrSrSr g) NullPooli>z)Serial analog of parallel execution Pool.cSUlg)NF_startedr)s rrNullPool.__init__As DMrcUR(d [S5e[U"U645n[ U5$! [[R"5S9nN+=f)N&NullPool must be Start()ed before use.rk)rrrbsysrkr)rr4rCrs rr:NullPool.ApplyAsyncDsP == ""J KK0d ~&f v 0 /fs 8AcJUR(a [S5eSUlg)NzCan only start NullPool once.Trrr)s rr*NullPool.StartPs }} !"A BBDMrc<UR(d [S5eg)Nrrr)s rr. NullPool.JoinUs == ""J KK rrN) rrrrrrr:r*r.rr rrrr>s1  Lrrc,\rSrSrSrSrSrSrSrg) _ThreadFutureiccXlX lgr_task _results_map)rtask results_maps rr_ThreadFuture.__init__es J#rc>UR5R5$)z6Return the value of the future, or raise an exception.rWr)s rr2_ThreadFuture.Getis >>  & & ((rcURUR;aURUR$[R"[5 MO)zGet the _Result of the future.)rrrrrr)s rrX_ThreadFuture.GetResultms=  t(( (  ,, jj  rc4URUR;$)z8Return True if the task finished with or without errors.rr)s rr__ThreadFuture.Donets ::** **r)rrN) rrrrrr2rXr_rr rrrrcs$)!+rrc\rSrSrSrSrg) _ThreadTaskiycXlgrr)rrs rr_ThreadTask.__init__{sIrrN)rrrrrrr rrrrysrrc.^\rSrSrU4SjrSrSrU=r$) _WorkerThreadicB>[[U] 5 XlX lgr)rrr work_queuer)rrrrs rr_WorkerThread.__init__s -') O"rc&URR5nU[LagURn[ UR "UR 645nX0RUR'Mn! [ [R"5S9nN:=f)Nr) rget _STOP_WORKINGrrbr4rCrrkr)r thread_taskrrs rrun_WorkerThread.runs~ OO'')k  %   d2$))TYY/12,2{''( 2#,,.1s #A00B)rr)rrrrrrrr!r"s@rrrs# 2 2rrc0\rSrSrSrSrSrSrSrSr g) ThreadPooliz%Thread-based parallel execution Pool.c`Xl[R"5Ul/Ul0Ulgr) num_threadsrQueue _task_queueworker_threadsr)rrs rrThreadPool.__init__s&"{{}DDDrc UR(a [S5e[UR5HNn[ UR UR 5nURRU5 UR5 MP g)Nz(ThreadPool must be started at most once.) rrr rrrrrstartr_threads rr*ThreadPool.Startsa  !"L MM 4## $T--t/@/@Af   ( lln%rcUR(d [S5e[X5n[X0R5nUR R [U55 U$Nz(ThreadPool must be Start()ed before use.)rrrrrrputr)rr4rCrrs rr:ThreadPool.ApplyAsyncsO    !"L MM  D 4!2!2 3FT*+ MrcUR(d [S5eURH"nURR[5 M$ URHnUR 5 M gr)rrrrrrrs rr.ThreadPool.JoinsV    !"L MM   =)!%% kkm&r)rrrrN) rrrrrrr*r:r.rr rrrrs- rrc8US:Xa [5$[U5$)a-Returns a parallel execution pool for the given number of threads. Can return either: - NullPool: if num_threads is 1. - ThreadPool: if num_threads is greater than 1 Args: num_threads: int, the number of threads to use. Returns: BasePool instance appropriate for the given type of parallelism. re)rr)rs rGetPoolrsA : k ""r)'r __future__rrrrQrtr threadingrgooglecloudsdk.corerry six.movesrrr rrrwr r add_metaclassABCMetaobjectr$rTrbrr9rrrrrThreadrrrr rrr s70''  *  Q9Q5I53;;:v: :z3;;    </f</~&&0!:0!f"F"8 * LxLJ+J+,& 2I$$2(  P#r