U Ã@ÛfDfã@sXddlmZddlmZddlZeƒZGdd„deƒZGdd„deƒZ Gdd „d ƒZ dS) é)ÚEvent)Ú greenthreadNc@seZdZdZdS)Ú Collisiona DAGPool raises Collision when you try to launch two greenthreads with the same key, or post() a result for a key corresponding to a greenthread, or post() twice for the same key. As with KeyError, str(collision) names the key in question. N)Ú__name__Ú __module__Ú __qualname__Ú__doc__©r r úJ/var/www/html/chatgpt/venv/lib/python3.8/site-packages/eventlet/dagpool.pyrsrcs(eZdZdZ‡fdd„Zdd„Z‡ZS)ÚPropagateErrora8 When a DAGPool greenthread terminates with an exception instead of returning a result, attempting to retrieve its value raises PropagateError. Attributes: key the key of the greenthread which raised the exception exc the exception object raised by the greenthread cs>d ||jj|¡}tƒ |¡||_||f|_||_||_dS)NzPropagateError({}): {}: {}) ÚformatÚ __class__rÚsuperÚ__init__ÚmsgÚargsÚkeyÚexc)Úselfrrr©r r r r'sÿ  zPropagateError.__init__cCs|jS©N)r©rr r r Ú__str__3szPropagateError.__str__)rrrrrrÚ __classcell__r r rr r s r c@sôeZdZdZe dd¡Zifdd„Zdd„Ze fdd „Z e fd d „Z e fd d „Z e fdd„Z dd„Zdd„Zedd„ƒZdd„Zdd„Zdd„Zdd„Zdd„Zd4d!d"„Zd#d$„Zd5d&d'„Zd(d)„Zd*d+„Zd,d-„Zd.d/„Zd0d1„Ze fd2d3„Zd%S)6ÚDAGPoola¬ A DAGPool is a pool that constrains greenthreads, not by max concurrency, but by data dependencies. This is a way to implement general DAG dependencies. A simple dependency tree (flowing in either direction) can straightforwardly be implemented using recursion and (e.g.) :meth:`GreenThread.imap() `. What gets complicated is when a given node depends on several other nodes as well as contributing to several other nodes. With DAGPool, you concurrently launch all applicable greenthreads; each will proceed as soon as it has all required inputs. The DAG is implicit in which items are required by each greenthread. Each greenthread is launched in a DAGPool with a key: any value that can serve as a Python dict key. The caller also specifies an iterable of other keys on which this greenthread depends. This iterable may be empty. The greenthread callable must accept (key, results), where: key is its own key results is an iterable of (key, value) pairs. A newly-launched DAGPool greenthread is entered immediately, and can perform any necessary setup work. At some point it will iterate over the (key, value) pairs from the passed 'results' iterable. Doing so blocks the greenthread until a value is available for each of the keys specified in its initial dependencies iterable. These (key, value) pairs are delivered in chronological order, *not* the order in which they are initially specified: each value will be delivered as soon as it becomes available. The value returned by a DAGPool greenthread becomes the value for its key, which unblocks any other greenthreads waiting on that key. If a DAGPool greenthread terminates with an exception instead of returning a value, attempting to retrieve the value raises :class:`PropagateError`, which binds the key of the original greenthread and the original exception. Unless the greenthread attempting to retrieve the value handles PropagateError, that exception will in turn be wrapped in a PropagateError of its own, and so forth. The code that ultimately handles PropagateError can follow the chain of PropagateError.exc attributes to discover the flow of that exception through the DAG of greenthreads. External greenthreads may also interact with a DAGPool. See :meth:`wait_each`, :meth:`waitall`, :meth:`post`. It is not recommended to constrain external DAGPool producer greenthreads in a :class:`GreenPool `: it may be hard to provably avoid deadlock. .. automethod:: __init__ .. automethod:: __getitem__ Ú_Coro)rÚpendingcCsBz | ¡}Wntk r$|}YnXt|ƒ|_i|_tƒ|_dS)zé DAGPool can be prepopulated with an initial dict or iterable of (key, value) pairs. These (key, value) pairs are of course immediately available for any greenthread that depends on any of those keys. N)ÚitemsÚAttributeErrorÚdictÚvaluesÚcorosrÚevent)rZpreloadÚ iteritemsr r r rts   zDAGPool.__init__cCs| ¡S)aP waitall() blocks the calling greenthread until there is a value for every DAGPool greenthread launched by :meth:`spawn`. It returns a dict containing all :class:`preload data `, all data from :meth:`post` and all values returned by spawned greenthreads. See also :meth:`wait`. )Úwaitrr r r Úwaitall‹s zDAGPool.waitallcCst| |¡ƒS)a{ *keys* is an optional iterable of keys. If you omit the argument, it waits for all the keys from :class:`preload data `, from :meth:`post` calls and from :meth:`spawn` calls: in other words, all the keys of which this DAGPool is aware. wait() blocks the calling greenthread until all of the relevant keys have values. wait() returns a dict whose keys are the relevant keys, and whose values come from the *preload* data, from values returned by DAGPool greenthreads or from :meth:`post` calls. If a DAGPool greenthread terminates with an exception, wait() will raise :class:`PropagateError` wrapping that exception. If more than one greenthread terminates with an exception, it is indeterminate which one wait() will raise. If an external greenthread posts a :class:`PropagateError` instance, wait() will raise that PropagateError. If more than one greenthread posts PropagateError, it is indeterminate which one wait() will raise. See also :meth:`wait_each_success`, :meth:`wait_each_exception`. )rÚ wait_each©rÚkeysr r r r$—sz DAGPool.waitcCs| | |¡¡S)a+ *keys* is an optional iterable of keys. If you omit the argument, it waits for all the keys from :class:`preload data `, from :meth:`post` calls and from :meth:`spawn` calls: in other words, all the keys of which this DAGPool is aware. wait_each() is a generator producing (key, value) pairs as a value becomes available for each requested key. wait_each() blocks the calling greenthread until the next value becomes available. If the DAGPool was prepopulated with values for any of the relevant keys, of course those can be delivered immediately without waiting. Delivery order is intentionally decoupled from the initial sequence of keys: each value is delivered as soon as it becomes available. If multiple keys are available at the same time, wait_each() delivers each of the ready ones in arbitrary order before blocking again. The DAGPool does not distinguish between a value returned by one of its own greenthreads and one provided by a :meth:`post` call or *preload* data. The wait_each() generator terminates (raises StopIteration) when all specified keys have been delivered. Thus, typical usage might be: :: for key, value in dagpool.wait_each(keys): # process this ready key and value # continue processing now that we've gotten values for all keys By implication, if you pass wait_each() an empty iterable of keys, it returns immediately without yielding anything. If the value to be delivered is a :class:`PropagateError` exception object, the generator raises that PropagateError instead of yielding it. See also :meth:`wait_each_success`, :meth:`wait_each_exception`. )Ú _wait_eachÚ_get_keyset_for_wait_eachr'r r r r&±s'zDAGPool.wait_eachccs2| | |¡¡D]\}}t|tƒs||fVqdS)aÈ wait_each_success() filters results so that only success values are yielded. In other words, unlike :meth:`wait_each`, wait_each_success() will not raise :class:`PropagateError`. Not every provided (or defaulted) key will necessarily be represented, though naturally the generator will not finish until all have completed. In all other respects, wait_each_success() behaves like :meth:`wait_each`. N©Ú_wait_each_rawr*Ú isinstancer ©rr(rÚvaluer r r Úwait_each_successÚs  zDAGPool.wait_each_successccs2| | |¡¡D]\}}t|tƒr||fVqdS)aó wait_each_exception() filters results so that only exceptions are yielded. Not every provided (or defaulted) key will necessarily be represented, though naturally the generator will not finish until all have completed. Unlike other DAGPool methods, wait_each_exception() simply yields :class:`PropagateError` instances as values rather than raising them. In all other respects, wait_each_exception() behaves like :meth:`wait_each`. Nr+r.r r r Úwait_each_exceptionès  zDAGPool.wait_each_exceptioncCs0|tk rt|ƒSt|j ¡ƒt|j ¡ƒBSdS)ax wait_each(), wait_each_success() and wait_each_exception() promise that if you pass an iterable of keys, the method will wait for results from those keys -- but if you omit the keys argument, the method will wait for results from all known keys. This helper implements that distinction, returning a set() of the relevant keys. N)Ú_MISSINGÚsetr!r(r r'r r r r*øsz!DAGPool._get_keyset_for_wait_eachccs(| |¡D]\}}|| |¡fVq dS)z£ When _wait_each() encounters a value of PropagateError, it raises it. In all other respects, _wait_each() behaves like _wait_each_raw(). N)r,Ú_value_or_raise©rrrr/r r r r)szDAGPool._wait_eachcCst|tƒr|‚|Sr)r-r )r/r r r r4s zDAGPool._value_or_raiseccsN| ¡D].}|j |t¡}|tk r| |¡||fVq|s>qJ|j ¡qdS)a° pending is a set() of keys for which we intend to wait. THIS SET WILL BE DESTRUCTIVELY MODIFIED: as each key acquires a value, that key will be removed from the passed 'pending' set. _wait_each_raw() does not treat a PropagateError instance specially: it will be yielded to the caller like any other value. In all other respects, _wait_each_raw() behaves like wait_each(). N)Úcopyr Úgetr2Úremover"r$r5r r r r,s   zDAGPool._wait_each_rawcOsZ||jks||jkrt|ƒ‚t|ƒ}tj|j||| |¡f|ž|Ž}| ||¡|j|<dS)aÉ Launch the passed *function(key, results, ...)* as a greenthread, passing it: - the specified *key* - an iterable of (key, value) pairs - whatever other positional args or keywords you specify. Iterating over the *results* iterable behaves like calling :meth:`wait_each(depends) `. Returning from *function()* behaves like :meth:`post(key, return_value) `. If *function()* terminates with an exception, that exception is wrapped in :class:`PropagateError` with the greenthread's *key* and (effectively) posted as the value for that key. Attempting to retrieve that value will raise that PropagateError. Thus, if the greenthread with key 'a' terminates with an exception, and greenthread 'b' depends on 'a', when greenthread 'b' attempts to iterate through its *results* argument, it will encounter PropagateError. So by default, an uncaught exception will propagate through all the downstream dependencies. If you pass :meth:`spawn` a key already passed to spawn() or :meth:`post`, spawn() raises :class:`Collision`. N) r!r rr3rÚspawnÚ_wrapperr)r)rrÚdependsÚfunctionrÚkwdsrZnewcoror r r r94s ÿþþz DAGPool.spawnc Os~zHz|||f|ž|Ž}Wn,tk rD}zt||ƒ}W5d}~XYnXW5|j|=Xz| ||¡Wntk rxYnX|S)z™ This wrapper runs the top-level function in a DAGPool greenthread, posting its return value (or PropagateError) to the DAGPool. N)r!Ú Exceptionr Úpostr)rr<rÚresultsrr=ÚresultÚerrr r r r:`s  zDAGPool._wrappercOs,| ¡D]\}}|j|||f|ž|ŽqdS)a~ spawn_many() accepts a single *function* whose parameters are the same as for :meth:`spawn`. The difference is that spawn_many() accepts a dependency dict *depends*. A new greenthread is spawned for each key in the dict. That dict key's value should be an iterable of other keys on which this greenthread depends. If the *depends* dict contains any key already passed to :meth:`spawn` or :meth:`post`, spawn_many() raises :class:`Collision`. It is indeterminate how many of the other keys in *depends* will have successfully spawned greenthreads. N)rr9)rr;r<rr=rÚdepsr r r Ú spawn_many{szDAGPool.spawn_manycCs|j|j ¡|j|=dS)zŽ Kill the greenthread that was spawned with the specified *key*. If no such greenthread was spawned, raise KeyError. N)r!rÚkill)rrr r r rEsz DAGPool.killFcCsb|j |t¡}|tk r,|jt ¡k r,t|ƒ‚||jkrB|sBt|ƒ‚||j|<|j ¡t ƒ|_dS)aV post(key, value) stores the passed *value* for the passed *key*. It then causes each greenthread blocked on its results iterable, or on :meth:`wait_each(keys) `, to check for new values. A waiting greenthread might not literally resume on every single post() of a relevant key, but the first post() of a relevant key ensures that it will resume eventually, and when it does it will catch up with all relevant post() calls. Calling post(key, value) when there is a running greenthread with that same *key* raises :class:`Collision`. If you must post(key, value) instead of letting the greenthread run to completion, you must first call :meth:`kill(key) `. The DAGPool implicitly post()s the return value from each of its greenthreads. But a greenthread may explicitly post() a value for its own key, which will cause its return value to be discarded. Calling post(key, value, replace=False) (the default *replace*) when a value for that key has already been posted, by any means, raises :class:`Collision`. Calling post(key, value, replace=True) when a value for that key has already been posted, by any means, replaces the previously-stored value. However, that may make it complicated to reason about the behavior of greenthreads waiting on that key. After a post(key, value1) followed by post(key, value2, replace=True), it is unspecified which pending :meth:`wait_each([key...]) ` calls (or greenthreads iterating over *results* involving that key) will observe *value1* versus *value2*. It is guaranteed that subsequent wait_each([key...]) calls (or greenthreads spawned after that point) will observe *value2*. A successful call to post(key, :class:`PropagateError(key, ExceptionSubclass) `) ensures that any subsequent attempt to retrieve that key's value will raise that PropagateError instance. N) r!r7r2rZ getcurrentrr r"Úsendr)rrr/ÚreplaceÚcoror r r r?šs,  z DAGPool.postcCs | |f¡D]\}}|SdS)zw __getitem__(key) (aka dagpool[key]) blocks until *key* has a value, then delivers that value. N)r&)rrÚ_r/r r r Ú __getitem__àszDAGPool.__getitem__NcCs| |j ||¡¡S)zy get() returns the value for *key*. If *key* does not yet have a value, get() returns *default*. )r4r r7)rrÚdefaultr r r r7ësz DAGPool.getcCst|j ¡ƒS)zU Return a snapshot tuple of keys for which we currently have values. )Útupler r(rr r r r(òsz DAGPool.keyscst‡fdd„ˆj ¡DƒƒS)zT Return a snapshot tuple of currently-available (key, value) pairs. c3s |]\}}|ˆ |¡fVqdSr)r4)Ú.0rr/rr r Ú sÿz DAGPool.items..)rLr rrr rr rús ÿz DAGPool.itemscCs t|jƒS)zÞ Return number of running DAGPool greenthreads. This includes greenthreads blocked while iterating through their *results* iterable, that is, greenthreads waiting on values from other keys. )Úlenr!rr r r ÚrunningszDAGPool.runningcCst|j ¡ƒS)zÝ Return keys for running DAGPool greenthreads. This includes greenthreads blocked while iterating through their *results* iterable, that is, greenthreads waiting on values from other keys. )rLr!r(rr r r Ú running_keys szDAGPool.running_keyscCs t| ¡ƒS)a  Return number of waiting DAGPool greenthreads, that is, greenthreads still waiting on values from other keys. This explicitly does *not* include external greenthreads waiting on :meth:`wait`, :meth:`waitall`, :meth:`wait_each`. )rOÚ waiting_forrr r r Úwaitings zDAGPool.waitingcsht|j ¡ƒ‰|tk rF|j |t¡}|tkr<|j|tƒS|jˆSdd„‡fdd„|j ¡DƒDƒS)aö waiting_for(key) returns a set() of the keys for which the DAGPool greenthread spawned with that *key* is still waiting. If you pass a *key* for which no greenthread was spawned, waiting_for() raises KeyError. waiting_for() without argument returns a dict. Its keys are the keys of DAGPool greenthreads still waiting on one or more values. In the returned dict, the value of each such key is the set of other keys for which that greenthread is still waiting. This method allows diagnosing a "hung" DAGPool. If certain greenthreads are making no progress, it's possible that they are waiting on keys for which there is no greenthread and no :meth:`post` data. cSsi|]\}}|r||“qSr r )rMrrr r r Ú Vsýz'DAGPool.waiting_for..c3s |]\}}||jˆfVqdSr)r)rMrrH©Ú availabler r rNWsÿz&DAGPool.waiting_for..)r3r r(r2r!r7rr)rrrHr rUr rR$s   ÿÿzDAGPool.waiting_for)F)N) rrrrÚ collectionsÚ namedtuplerrr%r2r$r&r0r1r*r)Ú staticmethodr4r,r9r:rDrEr?rJr7r(rrPrQrSrRr r r r r7s4:    )    , F   r) Zeventlet.eventrZeventletrrWÚobjectr2r>rr rr r r r Ús