o -wi@sdZddlmZddlZddlZddlZddlZddlZddlm Z m Z m Z m Z m Z mZedZdd d ZGd d d eZGd ddZGdddZejdddZejdddZdS)z)Functions for compatibility with asyncio.) annotationsN)Any AsyncIteratorCallable CoroutineIteratorTypeVar_Tfn%Callable[[], Coroutine[Any, Any, _T]]returnc Csdtjjdd!}t}||j|}z|W|WdS|w1s+wYdS)aRun `fn` in an asyncio loop in a new thread. This must always be used instead of `asyncio.run` which fails if there is an active `asyncio` event loop in the current thread. Since `wandb` was not originally designed with `asyncio` in mind, using `asyncio.run` would break users who were calling `wandb` methods from an `asyncio` loop. Note that due to starting a new thread, this is slightly slow. ) max_workersN) concurrentfuturesThreadPoolExecutor_Runnersubmitrunresultcancel)r executorrunnerfuturerY/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/wandb/sdk/lib/asyncio_compat.pyrs  rc@seZdZdZdS)_RunnerCancelledErrorz-The `_Runner.run()` invocation was cancelled.N)__name__ __module__ __qualname____doc__rrrrr$src@s8eZdZdZdddZdd d Zdd d Zdd dZdS)raRuns an asyncio event loop allowing cancellation. This is like `asyncio.run()`, except it provides a `cancel()` method meant to be called in a `finally` block. Without this, it is impossible to make `asyncio.run()` stop if it runs in a non-main thread. In particular, a KeyboardInterrupt causes the ThreadPoolExecutor above to block until the asyncio thread completes, but there is no way to tell the asyncio thread to cancel its work. A second KeyboardInterrupt makes ThreadPoolExecutor give up while the asyncio thread still runs in the background, with terrible effects if it prints to the user's terminal. r NonecCs,t|_d|_d|_d|_d|_d|_dS)NF) threadingLock_lock _is_cancelled_started_done_loop _cancel_eventselfrrr__init__7s  z_Runner.__init__r r r cCst||S)zRun a coroutine in asyncio, cancelling it on `cancel()`. Returns: The result of the coroutine returned by `fn`. Raises: _RunnerCancelledError: If `cancel()` is called. )asyncior_run_or_cancel)r+r rrrrAs z _Runner.runcs|j|jr tt|_t|_d|_Wdn1s"wYt |j }t |}z7tj ||gtj dIdH| rj| W|||j d|_WdS1sdwYSt|||j d|_Wdw1swYw)NT return_when)r$r%rr-get_running_loopr(Eventr)r& create_taskwaitFIRST_COMPLETEDdonerrr')r+r cancellation_taskfn_taskrrrr.Ls8    z_Runner._run_or_cancelcCs|j9|jr WddSd|_|js|js" WddS|js'J|js,J|j|jjWddS1s?wYdS)z+Cancel all asyncio work started by `run()`.NT)r$r%r'r&r(r)call_soon_threadsafesetr*rrrrms  "z_Runner.cancelNr r!r r r r )rrrr r,rr.rrrrrr(s   !rc@s8eZdZdZdddZddd Zdd d Zdd d ZdS) TaskGroupz'Object that `open_task_group()` yields.r r!cCs g|_dS)N)_tasksr*rrrr,s zTaskGroup.__init__coroCoroutine[Any, Any, Any]cCs|jt|dS)zSchedule a task in the group. Args: coro: The return value of the `async` function defining the task. N)r>appendr-r3)r+r?rrr start_soonszTaskGroup.start_soonc sVtj|jtjjdIdH\}}|D]}z |}r|Wqtjy(YqwdS)zBlock until all tasks complete. Raises: Exception: If one or more tasks raises an exception, one of these is raised arbitrarily. r/N)r-r4r>rrFIRST_EXCEPTION exceptionCancelledError)r+r6_taskexcrrr _wait_alls zTaskGroup._wait_allcCs|jD]}|qdS)zCancel all tasks.N)r>r)r+rGrrr _cancel_alls  zTaskGroup._cancel_allNr;)r?r@r r!)rrrr r,rBrIrJrrrrr=s    r=AsyncIterator[TaskGroup]cCs6t}z|V|IdHW|dS|w)aeCreate a task group. `asyncio` gained task groups in Python 3.11. This is an async context manager, meant to be used with `async with`. On exit, it blocks until all subtasks complete. If any subtask fails, or if the current task is cancelled, it cancels all subtasks in the group and raises the subtask's exception. If multiple subtasks fail simultaneously, one of their exceptions is chosen arbitrarily. NOTE: Subtask exceptions do not propagate until the context manager exits. This means that the task group cannot cancel code running inside the `async with` block . N)r=rIrJ) task_grouprrropen_task_groups rMr?r@Iterator[None]ccs\t|}zdVW|r|}r||dS|r)|}r)||w)zSchedule a task, cancelling it when exiting the context manager. If the given coroutine raises an exception, that exception is raised when exiting the context manager. N)r-r3r6rDr)r?rGrDrrrcancel_on_exits   rOr<)r rK)r?r@r rN)r __future__rr-rconcurrent.futures contextlibr"typingrrrrrrr r Exceptionrrr=asynccontextmanagerrMcontextmanagerrOrrrrs"   X*