File: //home/arjun/projects/env/lib/python3.10/site-packages/billiard/__pycache__/queues.cpython-310.pyc
o
!we�1 � @ s� d dl Z d dlZd dlZd dlZd dlZd dlZddlmZ ddlmZ ddl m
Z
d dl m Z d dl
mZmZ ddlmZmZmZmZmZmZ dd lmZ g d
�ZG d d � d �Ze� Z G d
d� de�Z G dd� d�Z G dd� de �Z dS )� N� )�
connection)�context)� get_errno)� monotonic)�Empty�Full)�debug�error�info�Finalize�register_after_fork�
is_exiting)�ForkingPickler)�Queue� SimpleQueue�
JoinableQueuec @ s� e Zd ZdZd)dd�Zdd� Zdd� Zd d
� Zd*d
d�Zd*dd�Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd � Zd d � Zd d � Zd!d"� Zed#d$� �Zed%d&� �Zed'd(� �Zd S )+r z4
Queue type using a pipe, buffer and thread
r c O s� z|d }W n t y td��w |dkr ddlm} || _tjdd�\| _| _|� � | _
t � � | _
tjdkr<d | _n|� � | _|�|�| _d| _| �� tjdkr[t| tj� d S d S ) N�ctxz-missing 1 required keyword-only argument: ctxr r )�
SEM_VALUE_MAXF��duplex�win32)�KeyError� TypeError� synchronizer �_maxsizer �Pipe�_reader�_writer�Lock�_rlock�os�getpid�_opid�sys�platform�_wlock�BoundedSemaphore�_sem�
_ignore_epipe� _after_forkr
r ��self�maxsize�args�kwargsr � r0 �H/home/arjun/projects/env/lib/python3.10/site-packages/billiard/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, r0 r0 r1 � __getstate__< s
�zQueue.__getstate__c C s0 |\| _ | _| _| _| _| _| _| _| �� d S r3 ) r) r r r r r& r( r# r* �r, �stater0 r0 r1 � __setstate__A s � zQueue.__setstate__c C sv t d� t�t�� �| _t�� | _d | _d | _ d| _
d| _ d | _ | j
j| _| jj| _| j
j| _| jj| _| jj| _d S )NzQueue._after_fork()F)r � threading� Conditionr � _notempty� collections�deque�_buffer�_thread� _jointhread�_joincancelled�_closed�_closer �send� _send_bytesr �recv�_recv�
send_bytes�
recv_bytes� _recv_bytes�poll�_pollr5 r0 r0 r1 r* F s
zQueue._after_forkTNc C sp | j rJ �| j�||�st�| j� | jd u r| �� | j�|� | j� � W d � d S 1 s1w Y d S r3 )
rC r( �acquirer r< r@ �
_start_threadr? �append�notify�r, �obj�block�timeoutr0 r0 r1 �putU s
"�z Queue.putc C s� |r#|d u r#| j � | �� }W d � n1 sw Y | j�� n@|r*t� | }| j �||�s3t�z)|rG|t� }|dk sD| �|�sFt�n| �� sMt�| �� }| j�� W | j �� n| j �� w t� |�S )Nr )
r rK r( �releaser rN r rM r �loads)r, rT rU �res�deadliner0 r0 r1 �get` s*
�
�
z Queue.getc C s | j | jj�� S r3 )r r( �_semlock�
_get_valuer5 r0 r0 r1 �qsizey s z Queue.qsizec C �
| � � S r3 �rM r5 r0 r0 r1 �empty~ �
z Queue.emptyc C s | j j�� S r3 )r( r\ �_is_zeror5 r0 r0 r1 �full� � z
Queue.fullc C s
| � d�S �NF)r[ r5 r0 r0 r1 �
get_nowait� rb zQueue.get_nowaitc C s | � |d�S rf )rV �r, rS r0 r0 r1 �
put_nowait� re zQueue.put_nowaitc C sL d| _ z| j�� W | j}|rd | _|� d S d S | j}|r%d | _|� w w )NT)rC r �closerD )r, rj r0 r0 r1 rj � s
���z Queue.closec C s( t d� | js J �| jr| �� d S d S )NzQueue.join_thread())r rC rA r5 r0 r0 r1 � 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 rB rA �cancel�AttributeErrorr5 r0 r0 r1 �cancel_join_thread� s �zQueue.cancel_join_threadc C s� t d� | j�� tjtj| j| j| j| j | j
j | j fdd�| _
d| j
_t d� | j
�� t d� | jt�� k}| jsM|sMt| j
tjt�| j
�gdd�| _t| tj| j| jgd d�| _d S )
NzQueue._start_thread()�QueueFeederThread)�targetr. �nameTzdoing self._thread.start()z ... done self._thread.start()���)� exitpriority�
)r r? �clearr: �Threadr �_feedr< rF r& r rj r) r@ �daemon�startr# r! r" rB r �_finalize_join�weakref�refrA �_finalize_closerD )r, �created_by_this_processr0 r0 r1 rO � s2
��
�
�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)�twr�threadr0 r0 r1 rz � 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 rP � _sentinelrQ )�buffer�notemptyr0 r0 r1 r} � s
"�zQueue._finalize_closec C s� t d� |j}|j}|j}| j} t}
tjdkr |j} |j} nd } zQ |� z
| s+|� W |� n|� w z3 | � }
|
|
u rHt d� |� W W d S t� |
�}
| d u rV||
� n| � z ||
� W | � n| � w q6 t
yp Y nw q" t y� } zD|r�t |�t
jkr�W Y d }~d S z t� r�td|dd� ntd|dd�s�dd l}|�� W n t y� Y nw W Y d }~d S W Y d }~d S d }~ww ) Nz$starting thread to feed data to piper r z%feeder thread got sentinel -- exitingzerror in queue thread: %rT)�exc_infor )r rN rW �wait�popleftr� r$ r% r �dumps�
IndexError� Exceptionr �errno�EPIPEr r r
� traceback� print_exc)r� r� rI � writelockrj � ignore_epipe�nacquire�nrelease�nwait�bpopleft�sentinel�wacquire�wreleaserS �excr� r0 r0 r1 rw � sn
�
� ���� ���z Queue._feed�r �TN)�__name__�
__module__� __qualname__�__doc__r2 r6 r9 r* rV r[ r^ ra rd rg ri rj rk rn rO � staticmethodrz r} rw r0 r0 r0 r1 r s.
&
r c @ sD e Zd ZdZddd�Zdd� Zdd� Zdd d �Zd
d� Zdd� Z d
S )r z�
A queue type which also supports join() and task_done() methods
Note that if you do not call task_done() for each finished task then
eventually the counter's semaphore may overflow causing Bad Things
to happen.
r c O sN z|d }W n t y td��w tj| ||d� |�d�| _|�� | _d S )Nr z(missing 1 required keyword argument: ctx)r r )r r r r2 � Semaphore�_unfinished_tasksr; �_condr+ r0 r0 r1 r2 " s � zJoinableQueue.__init__c C s t �| �| j| jf S r3 )r r6 r� r� r5 r0 r0 r1 r6 + s zJoinableQueue.__getstate__c C s, t �| |d d� � |dd � \| _| _d S )N���)r r9 r� r� r7 r0 r0 r1 r9 . s zJoinableQueue.__setstate__TNc C s� | j rJ �| j�||�st�| j�= | j�! | jd u r | �� | j� |� | j
� � | j� � W d � n1 s9w Y W d � d S W d � d S 1 sQw Y d S r3 )
rC r( rN r r< r� r@ rO r? rP r� rW rQ rR r0 r0 r1 rV 2 s
��"�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� rN �
ValueErrorr\ rc �
notify_allr5 r0 r0 r1 � task_done? s �"�zJoinableQueue.task_donec C sR | j � | jj�� s| j �� W d � d S W d � d S 1 s"w Y d S r3 )r� r� r\ rc r� r5 r0 r0 r1 r F s �"�zJoinableQueue.joinr� r� )
r� r� r� r� r2 r6 r9 rV r� r r0 r0 r0 r1 r s
r c @ sZ e Zd ZdZddd�Zdd� Zdd � Zd
d � Zd d
� Zdd� Z dd� Z
dd� Z dd� Z dS )� _SimpleQueuez<
Simplified Queue type -- really just a locked pipe
FNc C s2 t jd||d�\| _| _| jj| _d | _| _d S )NF)r � rnonblock� wnonblock)r r r r rL rM r r& )r, r� r� r r0 r0 r1 r2 Q s
�
z_SimpleQueue.__init__c C r_ r3 r` r5 r0 r0 r1 ra X rb z_SimpleQueue.emptyc C s t �| � | j| j| j| jfS r3 )r r4 r r r r& r5 r0 r0 r1 r6 [ s
z_SimpleQueue.__getstate__c C s |\| _ | _| _| _d S r3 )r r r r& r7 r0 r0 r1 r9 _ s z_SimpleQueue.__setstate__c C s
| j �� S r3 )r rJ r5 r0 r0 r1 � get_payloadb rb z_SimpleQueue.get_payloadc C s | j �|� d S r3 )r rI �r, �valuer0 r0 r1 � send_payloade s z_SimpleQueue.send_payloadc C s t �| �� �S r3 )r rX r� r5 r0 r0 r1 r[ h s z_SimpleQueue.getc C s | � t�|�� d S r3 )r� r r� rh r0 r0 r1 rV l s z_SimpleQueue.putc C sV | j d urz
| j �� W d | _ nd | _ w | jd ur)z | j�� W d | _d S d | _w d S r3 )r rj r r5 r0 r0 r1 rj p s
�z_SimpleQueue.close)FFN)
r� r� r� r� r2 ra r6 r9 r� r� r[ rV rj r0 r0 r0 r1 r� L s
r� c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )r c O sd z|d }W n t y td��w tjdd�\| _| _|�� | _tj dkr-|�� | _
d S d | _
d S )Nr z&missing required keyword argument: ctxFr r ) r r r r r r r r r$ r% r& )r, r. r/ r r0 r0 r1 r2 � s �
"zSimpleQueue.__init__c C s4 | j �
| j�� W d � S 1 sw Y d S r3 )r r rJ r5 r0 r0 r1 r� � s $�zSimpleQueue.get_payloadc C sR | j d u r
| j�|� d S | j � | j�|� W d � d S 1 s"w Y d S r3 )r& r rI r� r0 r0 r1 r� � s
"�zSimpleQueue.send_payloadN)r� r� r� r2 r� r� r0 r0 r0 r1 r ~ s r ) r$ r! r: r= r{ r� � r r �compatr �timer �queuer r �utilr r
r r r
r � reductionr �__all__r �objectr� r r� r r0 r0 r0 r1 �<module> s( x32