o biE@sfddlZddlmZmZmZmZmZmZmZm Z ddl Z eegdfZ ee gdfZ GdddZdS)N)AnyCallableDictIterableOptionalSetTupleUnionc@seZdZdZddefddZ  ddejdee d ee fd d Z  dd e ejdee d ee fd dZ dejfddZdeejfddZedefddZdejfddZ  ddeeeefdeefddZdS) RayEventManageraOEvent manager for Ray futures. The event manager can be used to track futures and invoke callbacks when they resolve. Futures are tracked with :meth:`track_future`. Future can then be awaited with :meth:`wait`. When futures successfully resolve, they trigger an optional ``on_result`` callback that can be passed to :meth:`track_future`. If they fail, they trigger an optional ``on_error`` callback. Args: shuffle_futures: If True, futures will be shuffled before awaited. This will avoid implicit prioritization of futures within Ray. Tshuffle_futurescCs||_i|_dSN)_shuffle_futures_tracked_futures)selfr r]/home/ubuntu/.local/lib/python3.10/site-packages/ray/air/execution/_internal/event_manager.py__init__szRayEventManager.__init__Nfuture on_resulton_errorcCs||f|j|<dS)aTrack a single future and invoke callbacks on resolution. Control has to be yielded to the event manager for the callbacks to be invoked, either via :meth:`wait` or via :meth:`resolve_future`. Args: future: Ray future to await. on_result: Callback to invoke when the future resolves successfully. on_error: Callback to invoke when the future fails. N)r)rrrrrrr track_future"szRayEventManager.track_futurefuturescCs|D] }|j|||dqdS)aTrack multiple futures and invoke callbacks on resolution. Control has to be yielded to the event manager for the callbacks to be invoked, either via :meth:`wait` or via :meth:`resolve_future`. Args: futures: Ray futures to await. on_result: Callback to invoke when the future resolves successfully. on_error: Callback to invoke when the future fails. )rrN)r)rrrrrrrr track_futures5szRayEventManager.track_futurescCs|j|ddS)zRemove future from tracking. The future will not be awaited anymore, and it will not trigger any callbacks. Args: future: Ray futures to discard. N)rpop)rrrrrdiscard_futureIszRayEventManager.discard_futurereturncC t|jS)z)Get futures tracked by the event manager.)setrrrrr get_futuresS zRayEventManager.get_futurescCrr )lenrrrrr num_futuresWr zRayEventManager.num_futuresc Csz |j|\}}Wnty}z td|d|d}~wwzt|}WntyC}z|r6||n|WYd}~dSd}~ww|rL||dSdS)aResolve a single future. This method will block until the future is available. It will then trigger the callback associated to the future and the event (success or error), if specified. Args: future: Ray future to resolve. zFuture z' is not tracked by this RayEventManagerN)rrKeyError ValueErrorrayget Exception)rrrreresultrrrresolve_future[s,    zRayEventManager.resolve_futuretimeout num_resultscCsXt|}|jrt||pt|}tjt|||d\}}|D]}||q"dS)a8Wait up to ``timeout`` seconds for ``num_results`` futures to resolve. If ``timeout=None``, this method will block until all `num_results`` futures resolve. If ``num_results=None``, this method will await all tracked futures. For every future that resolves, the respective associated callbacks will be invoked. Args: timeout: Timeout in second to wait for futures to resolve. num_results: Number of futures to await. If ``None``, will wait for all tracked futures to resolve. )r, num_returnsN) listrr randomshuffler!r%waitr*)rr,r-rready_rrrrr2xs    zRayEventManager.wait)T)NN)Nr+)__name__ __module__ __qualname____doc__boolrr% ObjectRefr_ResultCallback_ErrorCallbackrrrrrrpropertyintr"r*r floatr2rrrrr sB    r )r0typingrrrrrrrr r%r;r'r<r rrrrs (