o b²“i‡ ã@s.ddlmZmZmZmZmZGdd„dƒZdS)é)ÚAnyÚCallableÚListÚOptionalÚTuplec@sŠeZdZdZddœdedeedgdffdd„Zdd „Zd d „Z e d e fd d„ƒZ e d efdd„ƒZ d eeefdd„Zddd„ZdS)ÚBarrieražBarrier to collect results and process them in bulk. A barrier can be used to collect multiple results and process them in bulk once a certain count or a timeout is reached. For instance, if ``max_results=N``, the ``on_completion`` callback will be invoked once :meth:`arrive` has been called ``N`` times. The completion callback will only be invoked once, even if more results arrive after completion. The collected results can be resetted with :meth:`reset`, after which the callback may be invoked again. The completion callback should expect one argument, which is the barrier object that completed. Args: max_results: Maximum number of results to collect before a call to :meth:`wait` resolves or the :meth:`on_completion` callback is invoked. on_completion: Callback to invoke when ``max_results`` results arrived at the barrier. N)Ú on_completionÚ max_resultsrcCs||_d|_||_g|_dS)NF)Ú _max_resultsÚ _completedÚ_on_completionÚ_results)Úselfr r©rúW/home/ubuntu/.local/lib/python3.10/site-packages/ray/air/execution/_internal/barrier.pyÚ__init__s zBarrier.__init__cGs,t|ƒdkr |d}|j |¡| ¡dS)a,Notify barrier that a result successfully arrived. This will count against the ``max_results`` limit. The received result will be included in a call to :meth:`get_results`. Args: *data: Result data to be cached. Can be obtained via :meth:`get_results`. érN)Úlenr ÚappendÚ_check_completion)rÚdatarrrÚarrive+s   zBarrier.arrivecCs8|jrdS|j|jkrd|_|jr| |¡dSdSdS)NT)r Ú num_resultsr r ©rrrrr;s ûzBarrier._check_completionÚreturncCó|jS)z)Returns True if the barrier is completed.)r rrrrÚ completedGszBarrier.completedcCs t|jƒS)z(Number of received (successful) results.)rr rrrrrLs zBarrier.num_resultscCr)z Return list of received results.)r rrrrÚ get_resultsQszBarrier.get_resultscCsd|_g|_dS)a Reset barrier, removing all received results. Resetting the barrier will reset the completion status. When ``max_results`` is set and enough new events arrive after resetting, the :meth:`on_completion` callback will be invoked again. FN)r r rrrrÚresetUs z Barrier.reset)rN)Ú__name__Ú __module__Ú __qualname__Ú__doc__ÚintrrrrrÚpropertyÚboolrrrrrrrrrrrrs üþ ü rN)ÚtypingrrrrrrrrrrÚs