o 5@sdZddlmZddlmZddlmZddlZddlZddlZddlZddl Z ddl m Z ddl Z ddl mZddl mZdd l mZdZd ZGd d d eZGd ddeZe ejGdddeZe ejGdddeZGdddeZGdddeZGdddeZGdddeZGdddeZGdddeZ Gdd d eZ!Gd!d"d"eZ"Gd#d$d$ej#Z$Gd%d&d&eZ%d'd(Z&dS))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@seZdZdZdS)UnsupportedPlatformExceptionzHException indicating that a pool was created on an unsupported platform.N)__name__ __module__ __qualname____doc__rr>/tmp/google-cloud-sdk/lib/googlecloudsdk/core/util/parallel.pyr Asr cs eZdZdZfddZZS)InvalidStateExceptionzGException to indicate that a parallel pool was put in an invalid state.cstt||dSN)superr__init__)selfmsg __class__rrrHzInvalidStateException.__init__)r r r r r __classcell__rrrrrEsrc@sjeZdZdZejddZejddZddZdd Z d d Z d d Z ejddZ ddZ ddZdS)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'] cCt)zFInitialize non-trivial infrastructure (e.g. processes/threads/queues).NotImplementedErrorrrrrStartXzBasePool.StartcCr)z%Clean up anything started in Start().rrrrrJoin]r z BasePool.JoincC|||S)zFApplies func to each element in iterable and return a list of results.)MapAsyncGetrfunciterablerrrMapbz BasePool.Mapcstfdd|DS)z=Applies func to each element in iterable and return a future.csg|] }|fqSr) ApplyAsync).0argr&rrr hsz%BasePool.MapAsync..) _MultiFuturer%rr-rr#fszBasePool.MapAsynccCr")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%rrr MapEagerFetchjs zBasePool.MapEagerFetchcCr")z,Applies func to args and returns the result.)r*r$rr&argsrrrApplyyr)zBasePool.ApplycCr)z'Apply func to args and return a future.rr2rrrr*}r zBasePool.ApplyAsynccCs ||Sr)rrrrr __enter__szBasePool.__enter__cCs |dSr)r!)rexc_typeexc_valexc_tbrrr__exit__ zBasePool.__exit__N)r r r r abcabstractmethodrr!r(r#r1r4r*r5r9rrrrrLs    rc@s4eZdZdZddZejddZejddZdS) BaseFuturezAA future object containing a value that may not be available yet.cC |Sr GetResult GetOrRaiserrrrr$r:zBaseFuture.GetcCrrrrrrrr@zBaseFuture.GetResultcCrrrrrrrDonerBzBaseFuture.DoneN) r r r r r$r;r<r@rCrrrrr=s r=c@s2eZdZdZd ddZddZddZd d ZdS) _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 NcCsJttt|||gdkrtd|s|s|std||_||_||_dS)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)rrIrJrKrrrrs  z_Result.__init__cCs:|jr|jdS|jr|jtj|jd|jdddS)NrrE)tb)rIrJrreraiserKrrrrrAs  z_Result.GetOrRaisec Cs|jr t|jdd}n|}zt|W|Stjy.}z t|dWYd}~Sd}~wtyN}zttd|t|dWYd}~Sd}~ww)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}) rKrDpickledumps PicklingError Exceptionformatsix text_type)rpickleable_resulterrrrrToPickleableResults"   z_Result.ToPickleableResultcCsd|j|j|jS)Nz+_Result(value={0}, error={1}, exc_info={2}))rTrIrJrKrrrr__str__s z_Result.__str__)NNN)r r r r rrArYrZrrrrrDs    rDcseZdZfddZZS) MultiErrorcs2||_dd}tt|ddt||dS)NcSsdt|jt|S)Nz{}: {})rTtyper rUrV)errrsz%MultiError.__init__..zOne or more errors occurred: z )errorsrr[rjoinr)rr_fnrrrrs zMultiError.__init__)r r r rrrrrrr[sr[c@0eZdZdZddZddZddZdd Zd S) r/zFuture object that combines other Future objects. Returns the results of each future when they are all ready. Attributes: futures: list of BaseFuture. cC ||_dSr)futures)rrdrrrr z_MultiFuture.__init__c Csrg}g}|jD]"}z ||Wqty)}z ||WYd}~qd}~ww|r3tt|dSt|fdS)NrO)rI)rdappendr$rSrDr[)rresultsr_futurerXrrrr@s  z_MultiFuture.GetResultcCstdd|jDS)NcSsg|]}|qSr)rC)r+rhrrrr.sz%_MultiFuture.Done..)allrdrrrrrCrz_MultiFuture.Donec cs|j}|r>g}|D](}|r-z|VWq ty,}z |VWYd}~q d}~ww||q |}tt|sdSdS)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)rdrCr$rSrftimesleep_POLL_INTERVAL)runcollected_futurenext_uncollected_futurerhrXrrrr0s     z!_MultiFuture.GetResultsEagerFetchN)r r r r rr@rCr0rrrrr/s  r/c@rb) _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. cC||_||_dSr)r&r3r2rrrr z_Task.__init__cCst|jj|jfSr)hashr&r r3rrrr__hash__#sz_Task.__hash__cCs|jj|jjko |j|jkSr)r&r r3rotherrrr__eq__&sz _Task.__eq__cCs || Sr)rvrtrrr__ne__)r:z _Task.__ne__N)r r r r rrsrvrwrrrrros  roc@s$eZdZddZddZddZdS) _NullFuturecCrcrresult)rrzrrrr4rez_NullFuture.__init__cCs|jSrryrrrrr@7sz_NullFuture.GetResultcCsdS)NTrrrrrrC:sz_NullFuture.DoneN)r r r rr@rCrrrrrx2s rxc@rb) NullPoolz)Serial analog of parallel execution Pool.cCs d|_dS)NF)_startedrrrrrArezNullPool.__init__cCsF|jstdz t||f}Wt|Sttd}Yt|S)N&NullPool must be Start()ed before use.rK)r|rrDsysrKrx)rr&r3rzrrrr*DszNullPool.ApplyAsynccCs|jrtdd|_dS)NzCan only start NullPool once.Tr|rrrrrrPs zNullPool.StartcCs|jstddS)Nr}rrrrrr!Usz NullPool.JoinN)r r r r rr*rr!rrrrr{>s  r{c@s,eZdZddZddZddZddZd S) _ThreadFuturecCrpr_task _results_map)rtask results_maprrrrerqz_ThreadFuture.__init__cCr>)z6Return the value of the future, or raise an exception.r?rrrrr$i z_ThreadFuture.GetcCs& |j|jvr |j|jSttq)zGet the _Result of the future.)rrrjrkrlrrrrr@ms    z_ThreadFuture.GetResultcCs |j|jvS)z8Return True if the task finished with or without errors.rrrrrrCtrz_ThreadFuture.DoneN)r r r rr$r@rCrrrrrcs  rc@seZdZddZdS) _ThreadTaskcCrcr)r)rrrrrr{rez_ThreadTask.__init__N)r r r rrrrrrys rcs$eZdZfddZddZZS) _WorkerThreadcstt|||_||_dSr)rrr work_queuer)rrrrrrrs z_WorkerThread.__init__cCsZ |j}|tur dS|j}z t|j|jf}Wn ttd}Y||j |j<q)NTr~) rget _STOP_WORKINGrrDr&r3rrKr)r thread_taskrrzrrrruns  z_WorkerThread.run)r r r rrrrrrrrs rc@rb) ThreadPoolz%Thread-based parallel execution Pool.cCs ||_t|_g|_i|_dSr) num_threadsrQueue _task_queueworker_threadsr)rrrrrrs  zThreadPool.__init__cCsD|jrtdt|jD]}t|j|j}|j||q dS)Nz(ThreadPool must be started at most once.) rrrrrrrrfstartr_threadrrrrs  zThreadPool.StartcCs8|jstdt||}t||j}|jt||SNz(ThreadPool must be Start()ed before use.)rrrorrrputr)rr&r3rrzrrrr*s   zThreadPool.ApplyAsynccCs>|jstd|jD]}|jtq |jD]}|qdSr)rrrrrr`rrrrr!s   zThreadPool.JoinN)r r r r rrr*r!rrrrrs  rcCs|dkrtSt|S)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)r{r)rrrrGetPools r)'r __future__rrrr;rPr threadingrjgooglecloudsdk.corerrU six.movesrrrrrlrSr r add_metaclassABCMetaobjectrr=rDr[r/rorxr{rrThreadrrrrrrrsB         =? 3 % (