U
|
Z ±d8 ã @ sö d dl mZmZ d dlmZ d dlZd dlmZmZ ddl m
|
Z
|
m Z m Z ddl mZmZ d dlZddlmZ ed d
|
Zejd d d G d d dZejd dG dd dZejdddG dd de
|
edZejdddG dd de edZdS )é )ÚdequeÚ OrderedDict)ÚinfN)ÚErrorÚValueé )Ú SendChannelÚReceiveChannelÚChannel)Úgeneric_functionÚNoPublicConstructor)Úenable_ki_protectionc C sF | t krt| tstd| dk r*tdt| }t |¡t |¡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.infr z max_buffer_size must be >= 0) r Ú
|
isinstanceÚintÚ TypeErrorÚ
|
ValueErrorÚMemoryChannelStateÚMemorySendChannelÚ_createÚMemoryReceiveChannel)Úmax_buffer_sizeÚstate© r úDd:\z\workplace\vscode\pyvenv\venv\Lib\site-packages\trio/_channel.pyÚopen_memory_channel s 5þr T)ÚfrozenÚslotsc @ s< e Zd Ze ¡ Ze ¡ Ze ¡ Ze ¡ Ze ¡ Z e ¡ Z
|
dS )ÚMemoryChannelStatsN) Ú__name__Ú
|
__module__Ú __qualname__ÚattrÚibÚcurrent_buffer_usedr Úopen_send_channelsÚopen_receive_channelsÚtasks_waiting_sendÚtasks_waiting_receiver r r r r N s r )r c @ sX e Zd Ze ¡ ZejedZejddZejddZ eje
|
dZ eje
|
dZ dd Z dS )r ©Úfactoryr ©Údefaultc C s, t t| j| j| j| jt| jt| jdS )N)r# r r$ r% r& r' )r ÚlenÚdatar r$ r% Ú
|
send_tasksÚ receive_tasks©Úselfr r r Ú
|
statisticsd s úz MemoryChannelState.statisticsN)r r r r! r" r r r- r$ r% r r. r/ r2 r r r r r X s r F)ÚeqÚreprc @ s e Zd Ze ¡ ZejddZejedZdd Z dd Z
|
dd Z e d
|
d Z e d d Ze dd Zdd Zdd Ze dd Ze dd ZdS )r Fr* r( c C s | j jd7 _d S ©Nr )Ú_stater$ r0 r r r Ú__attrs_post_init__x s z%MemorySendChannel.__attrs_post_init__c C s d t| t| j¡S )Nz.<send channel at {:#x}, using buffer at {:#x}>©ÚformatÚidr6 r0 r r r Ú__repr__{ s ÿzMemorySendChannel.__repr__c C s
|
| j ¡ S ©N©r6 r2 r0 r r r r2 s z MemorySendChannel.statisticsc C s | j r tj| jjdkr tj| jjrh| jjr2t| jjj dd\}}|j
|
j |¡ tj |t|¡ n*t| jj| jjk r| jj |¡ ntjdS )zLike `~trio.abc.SendChannel.send`, but if the channel's buffer is
|
full, raises `WouldBlock` instead of blocking.
|
|
r F©ÚlastN)Ú_closedÚtrioÚClosedResourceErrorr6 r% ÚBrokenResourceErrorr/ r- ÚAssertionErrorÚpopitemÚcustom_sleep_dataÚ_tasksÚremoveÚlowlevelÚ
|
rescheduler r, r ÚappendÚ
|
WouldBlock)r1 ÚvalueÚtaskÚ_r r r Ú send_nowait s z MemorySendChannel.send_nowaitc s t j ¡ I dH z |¡ W n t jk
|
r4 Y nX t j ¡ I dH dS t j ¡ j ¡ | j j
|
< _ fdd}t j |¡I dH dS )zSee `SendChannel.send <trio.abc.SendChannel.send>`.
|
|
Memory channels allow multiple tasks to call `send` at the same time.
|
|
Nc s j ¡ jj= tjjjS r< )rG rH r6 r. rA rI ÚAbortÚ SUCCEEDED©rO ©r1 rN r r Úabort_fn s
|
z(MemorySendChannel.send.<locals>.abort_fn) rA rI Úcheckpoint_if_cancelledrP rL Úcancel_shielded_checkpointÚ current_taskrG Úaddr6 r. rF Úwait_task_rescheduled©r1 rM rU r rT r Úsend s
|
zMemorySendChannel.sendc C s | j r tjt | j¡S )aü Clone 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@ rA rB r r r6 r0 r r r Úclone´ s zMemorySendChannel.clonec C s | S r< r r0 r r r Ú __enter__Ñ s zMemorySendChannel.__enter__c C s | ¡ d S r< ©Úclose©r1 Úexc_typeÚexc_valÚexc_tbr r r Ú__exit__Ô s zMemorySendChannel.__exit__c C s° | j r
|
dS d| _ | jD ]$}tj |tt ¡ ¡ | jj|= q| j ¡ | j j
|
d8 _
|
| jj
|
dkr¬| jjrnt | jj D ](}|j j |¡ tj |tt ¡ ¡ qv| jj ¡ dS )aÓ Close 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.
|
|
NTr r )r@ rG rA rI rJ r rB r6 r. Úclearr$ rD r/ rF rH Ú EndOfChannelrT r r r r` × s
|
|
zMemorySendChannel.closec à s | ¡ tj ¡ I d H d S r< ©r` rA rI Ú
|
checkpointr0 r r r Úacloseó s zMemorySendChannel.acloseN)r r r r! r" r6 r@ ÚsetrG r7 r; r2 r rP r\ r] r^ re r` rj r r r r r o s$
|
|
|
|
r )Ú metaclassc @ s e Zd Ze ¡ ZejddZejedZdd Z dd Z
|
dd Z e d
|
d Z e d d Ze dd Zdd Zdd Ze dd Ze dd ZdS )r Fr* r( c C s | j jd7 _d S r5 )r6 r% r0 r r r r7 ÿ s z(MemoryReceiveChannel.__attrs_post_init__c C s
|
| j ¡ S r< r= r0 r r r r2 s z MemoryReceiveChannel.statisticsc C s d t| t| j¡S )Nz1<receive channel at {:#x}, using buffer at {:#x}>r8 r0 r r r r; s ÿz MemoryReceiveChannel.__repr__c C s| | j r tj| jjrP| jjjdd\}}|jj |¡ tj
|
|¡ | jj |¡ | jj rd| jj ¡ S | jjsrtjtjdS )zLike `~trio.abc.ReceiveChannel.receive`, but if there's nothing
|
ready to receive, raises `WouldBlock` instead of blocking.
|
|
Fr> N)r@ rA rB r6 r. rE rF rG rH rI rJ r- rK Úpopleftr$ rg rL )r1 rN rM r r r Úreceive_nowait
|
s z#MemoryReceiveChannel.receive_nowaitc s t j ¡ I dH z ¡ }W n t jk
|
r2 Y nX t j ¡ I dH |S t j ¡ j ¡ d j j
|
< _ fdd}t j |¡I dH S )a See `ReceiveChannel.receive <trio.abc.ReceiveChannel.receive>`.
|
|
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.
|
|
Nc s j ¡ jj= tjjjS r< )rG rH r6 r/ rA rI rQ rR rS rT r r rU 5 s
|
z.MemoryReceiveChannel.receive.<locals>.abort_fn) rA rI rV rn rL rW rX rG rY r6 r/ rF rZ r[ r rT r Úreceive s
|
z MemoryReceiveChannel.receivec C s | j r tjt | j¡S )aÔ Clone 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@ rA rB r r r6 r0 r r r r] < s zMemoryReceiveChannel.clonec C s | S r< r r0 r r r r^ \ s z MemoryReceiveChannel.__enter__c C s | ¡ d S r< r_ ra r r r re _ s z MemoryReceiveChannel.__exit__c C s¼ | j r
|
dS d| _ | jD ]$}tj |tt ¡ ¡ | jj|= q| j ¡ | j j
|
d8 _
|
| jj
|
dkr¸| jjrnt | jj D ](}|j j |¡ tj |tt ¡ ¡ qv| jj ¡ | jj ¡ dS )aÙ Close 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.
|
|
NTr r )r@ rG rA rI rJ r rB r6 r/ rf r% rD r. rF rH rC r- rT r r r r` b s
|
|
zMemoryReceiveChannel.closec à s | ¡ tj ¡ I d H d S r< rh r0 r r r rj s zMemoryReceiveChannel.acloseN)r r r r! r" r6 r@ rk rG r7 r2 r; r rn ro r] r^ re r` rj r r r r r ù s$
|
|
|
|
r )Ú collectionsr r Úmathr r! Zoutcomer r Úabcr r r
|
Z_utilr r rA Z_corer r Úsr r r r r r r r Ú<module> s$
|
?
|
|
|
