,hXddlmZddlZddlmZmZddlmZmZddl m Z m Z ddl m Z ddlmZddlmZmZddlZdd lmZmZddlZd d lmZmZmZmZmZd d lmZm Z m!Z!m"Z"d d l#m$Z$m%Z%m&Z&m'Z'm(Z(ejRdkrddl*m+Z+erddl,m-Z-ddl.m/Z/m0Z0e/dZ1ndejdvr ddl.m/Z/e/dZ1 d(dZ4erGdde5dZ6ne'e4Z6ejnGddZ8ejrGddeeZ:e&ejrdddGd d!eee%"Z;e&ejrdddGd#d$eee%"Z<Gd%d&eeZ= d)d'Z>y#e3$rdZ1YwxYw)*) annotationsN) OrderedDictdeque)AsyncGeneratorCallable)AbstractAsyncContextManagerasynccontextmanager)wraps)inf) TYPE_CHECKINGGeneric)ErrorValue)ReceiveChannel ReceiveType SendChannelSendTypeT)Abort RaiseCancelTTaskenable_ki_protection)MultipleExceptionErrorNoPublicConstructorfinalgeneric_function!raise_single_exception_from_group) )BaseExceptionGroup) TracebackType) ParamSpecSelfPsphinx)r#.c|tk7rt|ts td|dkr t dt |}t tj|ttj|fS)u Open a channel for passing objects between tasks within a process. Memory channels are lightweight, cheap to allocate, and entirely in-memory. They don't involve any operating-system resources, or any kind of serialization. They just pass Python objects directly between tasks (with a possible stop in an internal buffer along the way). Channel objects can be closed by calling `~trio.abc.AsyncResource.aclose` or using ``async with``. They are *not* automatically closed when garbage collected. Closing memory channels isn't mandatory, but it is generally a good idea, because it helps avoid situations where tasks get stuck waiting on a channel when there's no-one on the other side. See :ref:`channel-shutdown` for details. Memory channel operations are all atomic with respect to cancellation, either `~trio.abc.ReceiveChannel.receive` will successfully return an object, or it will raise :exc:`Cancelled` while leaving the channel unchanged. Args: max_buffer_size (int or math.inf): The maximum number of items that can be buffered in the channel before :meth:`~trio.abc.SendChannel.send` blocks. Choosing a sensible value here is important to ensure that backpressure is communicated promptly and avoid unnecessary latency; see :ref:`channel-buffering` for more details. If in doubt, use 0. Returns: A pair ``(send_channel, receive_channel)``. If you have trouble remembering which order these go in, remember: data flows from left → right. In addition to the standard channel methods, all memory channel objects provide a ``statistics()`` method, which returns an object with the following fields: * ``current_buffer_used``: The number of items currently stored in the channel buffer. * ``max_buffer_size``: The maximum number of items allowed in the buffer, as passed to :func:`open_memory_channel`. * ``open_send_channels``: The number of open :class:`MemorySendChannel` endpoints pointing to this channel. Initially 1, but can be increased by :meth:`MemorySendChannel.clone`. * ``open_receive_channels``: Likewise, but for open :class:`MemoryReceiveChannel` endpoints. * ``tasks_waiting_send``: The number of tasks blocked in ``send`` on this channel (summing over all clones). * ``tasks_waiting_receive``: The number of tasks blocked in ``receive`` on this channel (summing over all clones). z.max_buffer_size must be an integer or math.infrzmax_buffer_size must be >= 0) r isinstanceint TypeError ValueErrorMemoryChannelStateMemorySendChannelr_createMemoryReceiveChannel)max_buffer_sizestates O/var/www/html/Resume-Scraper/venv/lib/python3.12/site-packages/trio/_channel.py_open_memory_channelr30spl#j#&FHII788#5o#FE!$$U+Q''. c$eZdZ ddZddZy)open_memory_channelct|SN)r3)clsr0s r2__new__zopen_memory_channel.__new__ws(8 8r4cyr8)selfr0s r2__init__zopen_memory_channel.__init__}s r4Nr0 int | floatreturnz4tuple[MemorySendChannel[T], MemoryReceiveChannel[T]])r0r@rANone)__name__ __module__ __qualname__r:r>r<r4r2r6r6vs 9( 9B 9  r4r6)zMemorySendChannel[T]MemoryReceiveChannel[T]cJeZdZUded<ded<ded<ded<ded<ded<y ) MemoryChannelStatisticsr)current_buffer_usedr@r0open_send_channelsopen_receive_channelstasks_waiting_sendtasks_waiting_receiveN)rCrDrE__annotations__r<r4r2rHrHs&  r4rHceZdZUded<ej eZded<dZded<dZ ded<ej e Z d ed <ej e Z d ed <dd Z y)r,r@r0zdeque[T]datarr)rJrKzOrderedDict[Task, T] send_taskszOrderedDict[Task, None] receive_tasksc tt|j|j|j|j t|j t|jS)N)rIr0rJrKrLrM)rHlenrPr0rJrKrQrRr=s r2 statisticszMemoryChannelState.statisticssO& #DII 00#66"&"<"<"4??3"%d&8&8"9   r4NrArH)rCrDrErNattrsFactoryrrPrJrKrrQrRrVr<r4r2r,r,sd  "U]]5)D()!"3"'4u}}['AJ$A-:U]];-GM*G r4r,F)eqreprslotsceZdZUded<dZded<ej eZded<ddZ dd Z dd Z e dd Z e dd Ze dd ZddZ ddZe ddZe ddZy)r-zMemoryChannelState[SendType]_stateFbool_closedz set[Task]_taskscB|jxjdz c_yNr)r^rJrUs r2__attrs_post_init__z%MemorySendChannel.__attrs_post_init__s &&!+&r4cPdt|ddt|jddS)Nzidr^rUs r2__repr__zMemorySendChannel.__repr__s+"2d8B-/A"T[[/RTAUUVWWr4c6|jjSz[Returns a `MemoryChannelStatistics` for the memory channel this is associated with.r^rVrUs r2rVzMemorySendChannel.statisticss{{%%''r4c|jrtj|jjdk(rtj |jj r|jjrJ|jj jd\}}|jjj|tjj|t|yt|jj|jj kr&|jjj#|ytj$)zLike `~trio.abc.SendChannel.send`, but if the channel's buffer is full, raises `WouldBlock` instead of blocking. rFlastN)r`trioClosedResourceErrorr^rKBrokenResourceErrorrRrPpopitemcustom_sleep_datararemovelowlevel reschedulerrTr0append WouldBlock)r=valuetask_s r2 send_nowaitzMemorySendChannel.send_nowaits <<** * ;; , , 1** * ;; $ ${{'' 'kk//77U7CGD!  " " ) ) 0 0 6 MM $ $T5< 8 !! "T[[%@%@ @ KK   # #E *// !r4cKtjjd{ j|tjj d{y7=7#tj $rYnwxYwtjj jj|jj<_ dfd }tjj|d{7yw)zSee `SendChannel.send `. Memory channels allow multiple tasks to call `send` at the same time. Ncjjjj=tj j jSr8)rarwr^rQrrrxr SUCCEEDEDr~r=r}s r2abort_fnz(MemorySendChannel.send..abort_fns= KK  t $ &&t,==&&00 0r4r~rrAr) rrrxcheckpoint_if_cancelledrcancel_shielded_checkpointr{ current_taskraaddr^rQrvwait_task_rescheduledr=r|rr}s` @r2sendzMemorySendChannel.sendsmm33555    U #--::< < <  6 =    }}))+ ', t$!% 1 mm11(;;;sP"D A$D A(!D A&D &D (A>;D =A>>BD DD cx|jrtjtj |j S)aClone this send channel object. This returns a new `MemorySendChannel` object, which acts as a duplicate of the original: sending on the new object does exactly the same thing as sending on the old object. (If you're familiar with `os.dup`, then this is a similar idea.) However, closing one of the objects does not close the other, and receivers don't get `EndOfChannel` until *all* clones have been closed. This is useful for communication patterns that involve multiple producers all sending objects to the same destination. If you give each producer its own clone of the `MemorySendChannel`, and then make sure to close each `MemorySendChannel` when it's finished, receivers will automatically get notified when all producers are finished. See :ref:`channel-mpmc` for examples. Raises: trio.ClosedResourceError: if you already closed this `MemorySendChannel` object. )r`rrrsr-r.r^rUs r2clonezMemorySendChannel.clones,2 <<** * ((55r4c|Sr8r<rUs r2 __enter__zMemorySendChannel.__enter__  r4c$|jyr8closer=exc_type exc_value tracebacks r2__exit__zMemorySendChannel.__exit__ r4c|jryd|_|jD]T}tjj |t tj |jj|=V|jj|jxjdzc_ |jjdk(r|jjrJ|jjD]b}|jjj|tjj |t tjd|jjjyy)aClose this send channel object synchronously. All channel objects have an asynchronous `~.AsyncResource.aclose` method. Memory channels can also be closed synchronously. This has the same effect on the channel and other tasks using it, but `close` is not a trio checkpoint. This simplifies cleaning up in cancelled tasks. Using ``with send_channel:`` will close the channel object on leaving the with block. NTrr)r`rarrrxryrrsr^rQclearrJrRrvrw EndOfChannelr=r}s r2rzMemorySendChannel.closes <<  KK -D MM $ $T51I1I1K+L M &&t, -  &&!+& ;; ) )Q .{{-- - 11 K&&--44T: ((uT5F5F5H/IJ K KK % % + + - /r4c|K|jtjjd{y7w)zVClose this send channel object asynchronously. See `MemorySendChannel.close`.Nrrrrx checkpointrUs r2aclosezMemorySendChannel.aclose2' mm&&((( 2<:<NrArBrAstrrW)r|rrArB)rAzMemorySendChannel[SendType]rAr$rztype[BaseException] | NonerzBaseException | NonerzTracebackType | NonerArB)rCrDrErNr`rXrYsetrardrkrVrrrrrrrrr<r4r2r-r-s )(GT& c*FI*,X( ""&<<8668,((   ..6))r4r-) metaclassceZdZUded<dZded<ej eZded<ddZ dd Z dd Z e dd Z e dd Ze dd ZddZ ddZe ddZe ddZy)r/zMemoryChannelState[ReceiveType]r^Fr_r`zset[trio._core._run.Task]racB|jxjdz c_yrc)r^rKrUs r2rdz(MemoryReceiveChannel.__attrs_post_init__Bs ))Q.)r4c6|jjSrmrnrUs r2rVzMemoryReceiveChannel.statisticsEs{{%%''r4cPdt|ddt|jddS)Nz`. Memory channels allow multiple tasks to call `receive` at the same time. The first task will get the first item sent, the second task will get the second item sent, and so on. Ncjjjj=tj j jSr8)rarwr^rRrrrxrrrs r2rz.MemoryReceiveChannel.receive..abort_fnzs= KK  t $ ))$/==&&00 0r4r) rrrxrrrr{rrarr^rRrvrrs` @r2receivezMemoryReceiveChannel.receivecsmm33555 '')E--::< < <L 6 =    }}))+ *. !!$'!% 1]]88BBBBsP"D A$D A(!D A&D &D (A>;D =A>>BD DD cx|jrtjtj |j S)aClone this receive channel object. This returns a new `MemoryReceiveChannel` object, which acts as a duplicate of the original: receiving on the new object does exactly the same thing as receiving on the old object. However, closing one of the objects does not close the other, and the underlying channel is not closed until all clones are closed. (If you're familiar with `os.dup`, then this is a similar idea.) This is useful for communication patterns that involve multiple consumers all receiving objects from the same underlying channel. See :ref:`channel-mpmc` for examples. .. warning:: The clones all share the same underlying channel. Whenever a clone :meth:`receive`\s a value, it is removed from the channel and the other clones do *not* receive that value. If you want to send multiple copies of the same stream of values to multiple destinations, like :func:`itertools.tee`, then you need to find some other solution; this method does *not* do that. Raises: trio.ClosedResourceError: if you already closed this `MemoryReceiveChannel` object. )r`rrrsr/r.r^rUs r2rzMemoryReceiveChannel.clones,8 <<** *#++DKK88r4c|Sr8r<rUs r2rzMemoryReceiveChannel.__enter__rr4c$|jyr8rrs r2rzMemoryReceiveChannel.__exit__rr4cN|jryd|_|jD]T}tjj |t tj |jj|=V|jj|jxjdzc_ |jjdk(r|jjrJ|jjD]b}|jjj|tjj |t tjd|jjj|jjjyy)aClose this receive channel object synchronously. All channel objects have an asynchronous `~.AsyncResource.aclose` method. Memory channels can also be closed synchronously. This has the same effect on the channel and other tasks using it, but `close` is not a trio checkpoint. This simplifies cleaning up in cancelled tasks. Using ``with receive_channel:`` will close the channel object on leaving the with block. NTrr)r`rarrrxryrrsr^rRrrKrQrvrwrtrPrs r2rzMemoryReceiveChannel.closes/ <<  KK 0D MM $ $T51I1I1K+L M ))$/ 0  ))Q.) ;; , , 1{{00 0 .. R&&--44T: ((uT5M5M5O/PQ R KK " " ( ( * KK   " " $ 2r4c|K|jtjjd{y7w)z\Close this receive channel object asynchronously. See `MemoryReceiveChannel.close`.NrrUs r2rzMemoryReceiveChannel.acloserrNrrWr)rAr)rAz!MemoryReceiveChannel[ReceiveType]rr)rCrDrErNr`rXrYrrardrVrkrrrrrrrrr<r4r2r/r/;s ,+GT(5 c(:F %:/(  &CC>99>,((   %%8))r4r/cPeZdZ ddZddZd dZd dZ d dZy) RecvChanWrapperc ||_||_yr8) _recv_chan_send_semaphore)r= recv_chansend_semaphores r2r>zRecvChanWrapper.__init__s$-r4cK|jj|jjd{S7wr8)rreleaserrrUs r2rzRecvChanWrapper.receives1 $$&__,,....s8A?AcTK|jjd{y7wr8)rrrUs r2rzRecvChanWrapper.aclosesoo$$&&&s (&(c|Sr8r<rUs r2rzRecvChanWrapper.__enter__rr4c8|jjyr8)rrrs r2rzRecvChanWrapper.__exit__s r4N)rrFrtrio.SemaphorerArB)rArrrr)rCrDrEr>rrrrr<r4r2rrsY.0.BP. . /' , ( (    r4rcntt dfd } dd|S)aDecorate an async generator function to make it cancellation-safe. The ``yield`` keyword offers a very convenient way to write iterators... which makes it really unfortunate that async generators are so difficult to call correctly. Yielding from the inside of a cancel scope or a nursery to the outside `violates structured concurrency `_ with consequences explained in :pep:`789`. Even then, resource cleanup errors remain common (:pep:`533`) unless you wrap every call in :func:`~contextlib.aclosing`. This decorator gives you the best of both worlds: with careful exception handling and a background task we preserve structured concurrency by offering only the safe interface, and you can still write your iterables with the convenience of ``yield``. For example:: @as_safe_channel async def my_async_iterable(arg, *, kwarg=True): while ...: item = await ... yield item async with my_async_iterable(...) as recv_chan: async for item in recv_chan: ... While the combined async-with-async-for can be inconvenient at first, the context manager is indispensable for both correctness and for prompt cleanup of resources. cPKtjtd\}} tjd4d{} |i|}tjd}|j |||d{t ||5}|ddd|jjdddd{y77K#1swY8xYw7#1d{7swYyxYw#t$r2} t|n#t$rtd|gdwxYwYd}~yd}~wwxYww)NrT)strict_exception_groupsz}Encountered exception during cleanup of generator object, as well as exception in the contextmanager body - unable to unwrap.) rrr6r open_nursery Semaphorestartr cancel_scopecancelr!rr) argskwargs send_chanrnurseryagenrwrapped_recv_chaneg_move_elems_to_channelfns r2context_managerz(as_safe_channel..context_managers: $77:1= 9 ((F . .'4*6*!%!2mm*D)^%Y?,CT++,$$++- . . . ,, . . . ."  1"5) )TD 6 s D&C(CC(5C7C8CC"C0 C(;C<C(D&C(CC CC(C%C C%!C($D&%C(( D#2 C>=D>DDD&D##D&cK|5 |j |jd{ |jd{}|j |d{K777 #t$r$Y|j d{7dddywxYw79#|j d{7wxYw#1swYyxYwwr8)startedacquire __anext__StopAsyncIterationrr)rrr task_statusr|s r2rz/as_safe_channel.._move_elems_to_channel2s $ $##%(00222&*nn&6 6$..///2 6- kkm## $ $0kkm## $ $sCB7$BA#BA'A%A' BBB%A'' B0B1B7BB7 CBBB4-B0 .B44B77C<C)rzP.argsrzP.kwargsrAz6AsyncGenerator[trio._channel.RecvChanWrapper[T], None]) rzAsyncGenerator[T, None]rztrio.MemorySendChannel[T]rrrztrio.TaskStatusrArB)r r )rrrs` @r2as_safe_channelrswF 2Y!) ?>$%$,$'$% $  $0 r4r?)rz$Callable[P, AsyncGenerator[T, None]]rAz;Callable[P, AbstractAsyncContextManager[ReceiveChannel[T]]])? __future__rsys collectionsrrcollections.abcrr contextlibrr functoolsr mathr typingr r rXoutcomerrrr_abcrrrrr_corerrrr_utilrrrrr version_infoexceptiongroupr!typesr"typing_extensionsr#r$r%modules ImportErrorr3tupler6frozenrHdefiner,r-r/rrr<r4r2rs" *4G  GGBBg1#1#A / cN> >9>He$UV++?@   , U%0O) H-9LO)1O)d U%0T)>+6BUT)1T)n nQ' 4\,\@\I  s,FF  F