File: //opt/saltstack/salt/lib/python3.10/multiprocessing/__pycache__/queues.cpython-310.pyc
o
�N�g�. � @ s� g d �Z ddlZddlZddlZddlZddlZddlZddlZddlZddl m
Z
m Z ddl Z ddl
mZ ddl
mZ ejjZddlmZmZmZmZmZ G dd � d e�Ze� ZG d
d � d e�Z G d d
� d
e�Z dS ))�Queue� SimpleQueue�
JoinableQueue� N)�Empty�Full� )�
connection)�context)�debug�info�Finalize�register_after_fork�
is_exitingc @ s� e Zd Zd-dd�Zdd� Zdd� Zdd � Zd.d d �Zd/dd�Zd/dd�Z dd� Z
dd� Z dd� Z dd� Z
dd � Zd d � Zd d � Zd!d"� Zd#d$� Zed%d&� �Zed'd(� �Zed)d*� �Zed+d,� �ZdS )0r r c C s� |dkr
ddl m} || _tjdd�\| _| _|�� | _t �
� | _ t j
dkr*d | _n|�� | _|�|�| _d| _| �� t j
dkrIt| tj� d S d S )Nr r )�
SEM_VALUE_MAXF�Zduplex�win32)Z synchronizer �_maxsizer �Pipe�_reader�_writer�Lock�_rlock�os�getpid�_opid�sys�platform�_wlock�BoundedSemaphore�_sem�
_ignore_epipe�_resetr
r � _after_fork��self�maxsize�ctx� r' �</opt/saltstack/salt/lib/python3.10/multiprocessing/queues.py�__init__% s
�zQueue.__init__c C s. t �| � | j| j| j| j| j| j| j| j fS �N)
r �assert_spawningr r r r r r r r �r$ r' r' r( � __getstate__9 s
�zQueue.__getstate__c C s0 |\| _ | _| _| _| _| _| _| _| �� d S r* ) r r r r r r r r r! �r$ �stater' r' r( � __setstate__> s � zQueue.__setstate__c C s t d� | jdd� d S )NzQueue._after_fork()T)�
after_fork)r
r! r, r' r' r( r" C s zQueue._after_forkFc C sj |r| j �� nt�t�� �| _ t�� | _d | _d | _ d| _
d| _ d | _ | j
j| _| jj| _| jj| _d S �NF)� _notempty�_at_fork_reinit� threading� Conditionr � collections�deque�_buffer�_thread� _jointhread�_joincancelled�_closed�_closer �
send_bytes� _send_bytesr �
recv_bytes� _recv_bytes�poll�_poll)r$ r1 r' r' r( r! G s
z Queue._resetTNc C s| | j r td| �d���| j�||�st�| j� | jd u r!| �� | j� |� | j�
� W d � d S 1 s7w Y d S �NzQueue z
is closed) r= �
ValueErrorr �acquirer r3 r: �
_start_threadr9 �append�notify�r$ �obj�block�timeoutr' r' r( �putV s
"�z Queue.putc C s� | j r td| �d���|r.|d u r.| j� | �� }W d � n1 s#w Y | j�� n>|r6t�� | }| j�||�s?t �z&|rP|t�� }| �
|�sOt �n| �
� sVt �| �� }| j�� W | j�� n| j�� w t � |�S rE )
r= rF r rB r �release�time� monotonicrG r rD �_ForkingPickler�loads)r$ rM rN �resZdeadliner' r' r( �getb s.
�
�
z Queue.getc C s | j | jj�� S r* )r r �_semlock�
_get_valuer, r' r' r( �qsize| s z Queue.qsizec C �
| � � S r* �rD r, r' r' r( �empty� �
z Queue.emptyc C s | j j�� S r* )r rW �_is_zeror, r' r' r( �full� � z
Queue.fullc C s
| � d�S r2 )rV r, r' r' r( �
get_nowait� r] zQueue.get_nowaitc C s | � |d�S r2 )rO �r$ rL r' r' r( �
put_nowait� r` zQueue.put_nowaitc C s$ d| _ | j}|rd | _|� d S d S )NT)r= r> )r$ �closer' r' r( rd � s
�z Queue.closec C s2 t d� | jsJ d�| ���| jr| �� d S d S )NzQueue.join_thread()zQueue {0!r} not closed)r
r= �formatr; r, r' r' r( � join_thread� s
�zQueue.join_threadc C s4 t d� d| _z| j�� W d S ty Y d S w )NzQueue.cancel_join_thread()T)r
r< r; �cancel�AttributeErrorr, r' r' r( �cancel_join_thread� s �zQueue.cancel_join_threadc C s� t d� | j�� tjtj| j| j| j| j | j
j | j j | j
| j| jf dd�| _d| j_t d� | j�� t d� | jsKt| jtjt�| j�gdd�| _t| tj| j| jgd d�| _d S )
NzQueue._start_thread()ZQueueFeederThread)�target�args�nameTzdoing self._thread.start()z ... done self._thread.start()���)Z exitpriority�
)r
r9 �clearr5 �Threadr �_feedr3 r@ r r rd r r �_on_queue_feeder_errorr r: �daemon�startr< r �_finalize_join�weakref�refr; �_finalize_closer> r, r' r' r( rH � s4
��
�
�zQueue._start_threadc C s6 t d� | � }|d ur|�� t d� d S t d� d S )Nzjoining queue threadz... queue thread joinedz ... queue thread already dead)r
�join)Ztwr�threadr' r' r( ru � s zQueue._finalize_joinc C sD t d� |� | �t� |�� W d � d S 1 sw Y d S )Nz telling queue thread to quit)r
rI � _sentinelrJ )�buffer�notemptyr' r' r( rx � s
"�zQueue._finalize_closec
C sf t d� |j} |j}
|j} | j} t}
tjdkr |j}|j}nd } zT| � z
| s+| � W |
� n|
� w z6 | � }||
u rKt d� |� |� W W d S t� |�}|d u rY||� n|� z ||� W |� n|� w q6 t
ys Y nw W n< t y� } z0|r�t |dd�t
jkr�W Y d }~d S t� r�td|� W Y d }~d S |�� |||� W Y d }~nd }~ww q!)Nz$starting thread to feed data to piper r z%feeder thread got sentinel -- exiting�errnor zerror in queue thread: %s)r
rG rP �wait�popleftr{ r r rS �dumps�
IndexError� Exception�getattrr~ ZEPIPEr r )r| r} r? Z writelockZ reader_closeZ writer_closeZ ignore_epipe�onerrorZ queue_semZnacquireZnreleaseZnwaitZbpopleft�sentinelZwacquireZwreleaserL �er' r' r( rq � sb
�
� �
���z Queue._feedc C s ddl }|�� dS )z�
Private API hook called when feeding data in the background thread
raises an exception. For overriding by concurrent.futures.
r N)� traceback� print_exc)r� rL r� r' r' r( rr s z Queue._on_queue_feeder_error�r )F�TN)�__name__�
__module__� __qualname__r) r- r0 r" r! rO rV rY r\ r_ ra rc rd rf ri rH � staticmethodru rx rq rr r' r' r' r( r # s2
!
>r c @ s@ e Zd Zddd�Zdd� Zdd� Zdd
d �Zd d
� Zdd� Zd S )r r c C s* t j| ||d� |�d�| _|�� | _d S )N)r& r )r r) � Semaphore�_unfinished_tasksr6 �_condr# r' r' r( r) ( s zJoinableQueue.__init__c C s t �| �| j| jf S r* )r r- r� r� r, r' r' r( r- - s zJoinableQueue.__getstate__c C s, t �| |d d� � |dd � \| _| _d S )N���)r r0 r� r� r. r' r' r( r0 0 s zJoinableQueue.__setstate__TNc C s� | j r td| �d���| j�||�st�| j�= | j�! | jd u r%| �� | j �
|� | j � � | j�
� W d � n1 s?w Y W d � d S W d � d S 1 sWw Y d S rE )r= rF r rG r r3 r� r: rH r9 rI r� rP rJ rK r' r' r( rO 4 s
P�zJoinableQueue.putc C sf | j �& | j�d�std��| jj�� r!| j �� W d � d S W d � d S 1 s,w Y d S )NFz!task_done() called too many times)r� r� rG rF rW r^ �
notify_allr, r' r' r( � task_doneA s �"�zJoinableQueue.task_donec C sR | j � | jj�� s| j �� W d � d S W d � d S 1 s"w Y d S r* )r� r� rW r^ r r, r' r' r( ry H s �"�zJoinableQueue.joinr� r� ) r� r� r� r) r- r0 rO r� ry r' r' r' r( r & s
r c @ sN e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z e
e j �Z
dS )r c C sJ t jdd�\| _| _|�� | _| jj| _tj dkr d | _
d S |�� | _
d S )NFr r ) r r r r r r rC rD r r r )r$ r& r' r' r( r) S s
zSimpleQueue.__init__c C s | j �� | j�� d S r* )r rd r r, r' r' r( rd \ s
zSimpleQueue.closec C rZ r* r[ r, r' r' r( r\ ` r] zSimpleQueue.emptyc C s t �| � | j| j| j| jfS r* )r r+ r r r r r, r' r' r( r- c s
zSimpleQueue.__getstate__c C s" |\| _ | _| _| _| j j| _d S r* )r r r r rC rD r. r' r' r( r0 g s zSimpleQueue.__setstate__c C s: | j �
| j�� }W d � n1 sw Y t�|�S r* )r r rA rS rT )r$ rU r' r' r( rV k s �
zSimpleQueue.getc C s\ t �|�}| jd u r| j�|� d S | j� | j�|� W d � d S 1 s'w Y d S r* )rS r� r r r? rb r' r' r( rO q s
"�zSimpleQueue.putN)r� r� r� r) rd r\ r- r0 rV rO � classmethod�types� GenericAlias�__class_getitem__r' r' r' r( r Q s
r ) �__all__r r r5 r7 rQ r� rv r~ �queuer r Z_multiprocessing� r r Z reductionZForkingPicklerrS �utilr
r r r
r �objectr r{ r r r' r' r' r( �<module> s( z
+