Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/celery/concurrency/__pycache__/asynpool.cpython-39.pyc
Ðазад
a t�h � � @ s� d Z ddlZddlZddlZddlZddlZddlZddlmZm Z m Z ddl m Z ddl mZ ddlmZ ddlmZmZmZ ddlmZ dd lmZmZ dd lmZ dd l m Z m Z dd l m Z m!Z!m"Z"m#Z#m$Z$ dd l%m&Z& ddl'm(Z(m)Z) ddl*mZ+ ddl,m-Z- ddl.m/Z/ ddl0m1Z1 ddl2m3Z3 ddl4m5Z5 ddl6m7Z7 ddl8m9Z: zddl;m<Z= dZ>W n2 e?�y� ej<fdd�Z=dZ>efd d �ZY n0 d Z@e7eA�ZBeBjCeBjD ZCZDeEejFejGh�ZHd ZId ZJd!ZKd"ZLeLeLeKeKeLd#�ZMd$d%� eM�N� D �ZOe d&d'�ZPd(d)� ZQd*d+� ZReSed,��r>ddddejTejUejVejWfd-d.�ZXn d:d/d.�ZXddddeXfd0d1�ZYd2d3� ZZG d4d5� d5ej[�Z[G d6d7� d7ej\�Z\G d8d9� d9ej]�Z^dS );a� Version of multiprocessing.Pool using Async I/O. .. note:: This module will be moved soon, so don't use it directly. This is a non-blocking version of :class:`multiprocessing.Pool`. This code deals with three major challenges: #. Starting up child processes and keeping them running. #. Sending jobs to the processes and receiving results back. #. Safely shutting down this system. � N)�Counter�deque� namedtuple)�BytesIO)�Integral)�HIGHEST_PROTOCOL)�pack�unpack� unpack_from)�sleep)�WeakValueDictionary�ref)�pool)� isblocking� setblocking)�ACK�NACK�RUN� TERMINATE� WorkersJoined)� _SimpleQueue)�ERR�WRITE)�pickle)� SELECT_BAD_FD)�fxrange)�promise)� worker_before_create_process)�noop)� get_logger)�state)�readTc C s( || |�}t |�}|dkr$|�|� |S �Nr )�len�write)�fd�buf�sizer! �chunk�n� r* ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/celery/concurrency/asynpool.py�__read__5 s r, Fc C s || |� � �S �N)�getvalue)�fmtZiobufr r* r* r+ r = s r )�AsynPool� g @� � )N�default�fastZfcfsZfairc C s i | ]\}}||�qS r* r* )�.0�k�vr* r* r+ � <dictcomp>W � r9 �Ack)�idr% �payloadc C s t �| �dkS )z(Return true if generator is not started.� GEN_CREATED)�inspect�getgeneratorstate)�genr* r* r+ �gen_not_started\ s rB c C s( z | j }W n ty Y n0 |� S d S r- )�_writer�AttributeError)�job�writerr* r* r+ �_get_job_writera s rG �pollc s� |� }|j �| r$��fdd�| D � |r<��fdd�|D � |rT� �fdd�|D � t� t� } } |rr|dk rrdn t|d �}|�|�} | D ]P\} } t| t�s�| �� } | �@ r�| �| � | �@ r�| �| � | � @ r�| �| � q�| | dfS )Nc s g | ]}�|� ��qS r* r* �r6 r% )�POLLIN�registerr* r+ � <listcomp>r r: z _select_imp.<locals>.<listcomp>c s g | ]}�|� ��qS r* r* rI )�POLLOUTrK r* r+ rL t r: c s g | ]}�|� ��qS r* r* rI )�POLLERRrK r* r+ rL v r: r g @�@)rK �set�roundrH � isinstancer �fileno�add)�readers�writers�err�timeoutrH rJ rM rN Zpoller�R�W�eventsr% �eventr* )rN rJ rM rK r+ � _select_impk s* r\ c C s8 t � | |||�\}}}|r.tt|�t|�B �}||dfS r" )�select�listrO )rT rU rV rW �r�w�er* r* r+ r\ � s c C s@ | du rt � n| } |du r t � n|}|du r2t � n|}z|| |||�W S t�y: } z�|j}|tjkr�t � t � dfW Y d}~S |tv �r$| |B |B D ]p}zt�|gg g d� W q� t�y } z:|j}|tvrԂ | �|� |�|� |�|� W Y d}~q�d}~0 0 q�t � t � dfW Y d}~S � W Y d}~n d}~0 0 dS )a< Simple wrapper to :class:`~select.select`, using :`~select.poll`. Arguments: readers (Set[Fd]): Set of reader fds to test if readable. writers (Set[Fd]): Set of writer fds to test if writable. err (Set[Fd]): Set of fds to test for error condition. All fd sets passed must be mutable as this function will remove non-working fds from them, this also means the caller must make sure there are still fds in the sets before calling us again. Returns: Tuple[Set, Set, Set]: of ``(readable, writable, again)``, where ``readable`` is a set of fds that have data available for read, ``writable`` is a set of fds that's ready to be written to and ``again`` is a flag that if set means the caller must throw away the result and call us again. Nr2 r )rO �OSError�errno�EINTRr r] �discard)rT rU rV rW rH �exc�_errnor% r* r* r+ �_select� s, "rh c s� � �fdd�}g }| D ]\�|� | }}z|�g|�R i |�� W q t tfyp tjd�dd� |��� Y q0 q|r�|D ]L�z&t|d�r�|��� n |��d� W q| ty� t�d�|� Y q|0 q|dS ) a� Apply hub method to fds in iter, remove from list if failure. Some file descriptors may become stale through OS reasons or possibly other reasons, so safely manage our lists of FDs. :param fds_iter: the file descriptors to iterate and apply hub_method :param source_data: data source to remove FD if it renders OSError :param hub_method: the method to call with each fd and kwargs :*args to pass through to the hub_method; with a special syntax string '*fd*' represents a substitution for the current fd object in the iteration (for some callers). :**kwargs to pass through to the hub method (no substitutions needed) c s" � } d| v r �fdd�� D �} | S )N�*fd*c s g | ]}|d kr� n|�qS )ri r* )r6 �arg)r% r* r+ rL � r: zTiterate_file_descriptors_safely.<locals>._meta_fd_argument_maker.<locals>.<listcomp>r* )Z call_args��argsr% r* r+ �_meta_fd_argument_maker� s z@iterate_file_descriptors_safely.<locals>._meta_fd_argument_makerz)Encountered OSError when accessing fd %s T��exc_info�removeNz*ValueError trying to invalidate %s from %s) rb �FileNotFoundError�logger�warning�append�hasattrrp �pop� ValueError) Zfds_iterZ source_dataZ hub_methodrl �kwargsrm Z stale_fdsZhub_argsZ hub_kwargsr* rk r+ � iterate_file_descriptors_safely� s, � �ry c @ s e Zd ZdZdd� ZdS )�WorkerzPool worker process.c C s | j �t|ff� d S r- )�outq�put� WORKER_UP)�self�pidr* r* r+ � on_loop_start� s zWorker.on_loop_startN)�__name__� __module__� __qualname__�__doc__r� r* r* r* r+ rz � s rz c s^ e Zd ZdZ� fdd�Zeeeee j fdd�Z dd� Z dd � Z d d � Zd d � Zdd� Z� ZS )� ResultHandlerz)Handles messages from the pool processes.c s: |� d�| _|� d�| _t� j|i |�� | j| jt< d S )N�fileno_to_outq�on_process_alive)rv r� r� �super�__init__Zstate_handlersr} )r~ rl rx �� __class__r* r+ r� � s zResultHandler.__init__c c s� d } } |r t d�} t| �} n |� } } | dk r�z$|||rF| | d � n| d| �} W n6 ty� } z |jtvrp� d V W Y d }~q(d }~0 0 | dkr�| r�td�nt� �| | 7 } q(|d| �\}|r�t |�} t| �} n |� } } | |k �rnz$|||r�| | d � n| || �} W n: t�yD } z |jtv�r*� d V W Y d }~q�d }~0 0 | dk�rd| �r^td�nt� �| | 7 } q�||| j|� |�r�||| ��}n| �d� || �}|�r�||� d S )Nr r3 zEnd of file during messagez>i)� bytearray� memoryviewrb rc �UNAVAIL�EOFError� handle_event�seek)r~ � add_readerr% �callbackr, � readcanbufr r �loadZHrZBrr& Zbufvr) rf � body_size�messager* r* r+ � _recv_message� sZ � � � � zResultHandler._recv_messagec s6 | j �| j�|j� |j�| j�� ����fdd�}|S )z3Coroutine reading messages from the pool processes.c s~ z �| W n t y& �| � Y S 0 �� | ��}z t|� W n. tyR Y n( ttfyn �| � Y n 0 � | |� d S r- )�KeyError�next� StopIterationrb r� )rR �it�r� r� �on_state_changeZ recv_message� remove_readerr* r+ �on_result_readable? s z>ResultHandler._make_process_result.<locals>.on_result_readable)r� r� r� r� r� )r~ �hubr� r* r� r+ �_make_process_result7 s z"ResultHandler._make_process_resultc C s | � |�| _d S r- )r� r� )r~ r� r* r* r+ �register_with_event_loopO s z&ResultHandler.register_with_event_loopc G s t d��d S )Nz Not registered with event loop)� RuntimeError)r~ rl r* r* r+ r� R s zResultHandler.handle_eventc C s� | j }| j}| j}| j}| j}t|�}|r�|r�| jtkr�|d urF|� t� }|D ]L}t|g| j| j |j ||� z|dd� W qP t y� t d� Y d S 0 qP|� |� q&d S )NT)�shutdownz&result handler: all workers terminated)�cache�check_timeoutsr� r� �join_exited_workersrO �_stater ry �_flush_outqueuerS r �debug�difference_update) r~ r� r� r� r� r� Z outqueuesZpending_remove_fdr% r* r* r+ �on_stop_not_startedW s* � z!ResultHandler.on_stop_not_startedc C sJ z || }W n t y& ||� Y S 0 |jj}zt|d� W n tyX ||� Y S 0 z�z$|�d�rr|�� }n d }td� W nN ttfy� ||� Y W zt|d� W S ty� ||� Y S 0 S 0 |r�||� W zt|d� W n t�y ||� Y S 0 n:zt|d� W n( t�yB ||� Y Y S 0 0 d S )Nr2 r � �?) r� r{ �_readerr rb rH �recvr r� )r~ r% rp Z process_indexr� �proc�reader�taskr* r* r+ r� s s> � �z ResultHandler._flush_outqueue)r� r� r� r� r� r, r� r r �_pickler� r� r� r� r� r� r� � __classcell__r* r* r� r+ r� � s � 9 r� c s~ e Zd ZdZeZeZdZ� fdd�ZdN� fdd� Z� fdd �Z d d � Z d d � Z dd� Z dd� Z dd� Zdd� Zdd� Zdd� Zdd� Zeejefd d �Zd d � Zd d!� Zd"d#� Zd$d%� Zd&d'� Z d(d)� Z d*d+� Z d,d-� Z d.d/� Z d0d1� Z!d2d3� Z"e#d4d5� �Z$� fd6d7�Z%d8d9� Z&d:d;� Z'd<d=� Z(d>d?� Z)d@dA� Z*dBdC� Z+ejeefdDdE�Z,e-dFdG� �Z.dHdI� Z/e-dJdK� �Z0e1dLdM� �Z2� Z3S )Or0 zAsyncIO Pool (no threads).Fc s t � �|�}d|_|S )NF)r� � WorkerProcess�dead)r~ Zworkerr� r* r+ r� � s zAsynPool.WorkerProcessNc s t �||�� _|d u r � �� n|}|� _� fdd�t|�D �� _i � _i � _i � _ |d u r^t n|� _ t � � _ t � � _t � � _t � � _t � � _� jj� _t� � _t� � _t� j|g|�R i |�� � jD ] }|� j|j< |� j |j < q�t � j dt �� _ t � j dt �� _!d S )Nc s i | ]}� � � d �qS r- ��create_process_queues�r6 �_�r~ r* r+ r9 � s z%AsynPool.__init__.<locals>.<dictcomp>�on_soft_timeout�on_hard_timeout)"�SCHED_STRATEGIES�get�sched_strategy� cpu_count�synack�range�_queues�_fileno_to_inq�_fileno_to_outq�_fileno_to_synq�PROC_ALIVE_TIMEOUT�_proc_alive_timeoutrO �_waiting_to_start� _all_inqueues�_active_writes�_active_writers� _busy_workersre �_mark_worker_as_availabler �outbound_bufferr � write_statsr� r� �_pool�outqR_fd�synqW_fd�getattrZ_timeout_handlerr r� r� )r~ Z processesr� r� Zproc_alive_timeoutrl rx r� r� r� r+ r� � s@ � � � ��zAsynPool.__init__c s t j| d� t�� t� �|�S )N)Zsender)r �send�gcZcollectr� �_create_worker_process)r~ �ir� r* r+ r� � s z AsynPool._create_worker_processc C s | � ||� | �� d S r- )�_untrack_child_process� maintain_pool)r~ r� r� r* r* r+ �_event_process_exit� s z AsynPool._event_process_exitc C sN z |j }W n& ty0 t�|jj� }|_ Y n0 t|gd|j| j||� dS )z4Helper method determines appropriate fd for process.N) �_sentinel_pollrD �os�dupZ_popen�sentinelry r� r� �r~ r� r� r% r* r* r+ �_track_child_process� s �z AsynPool._track_child_processc C s0 |j d ur,|j d }|_ |�|� t�|� d S r- )r� rp r� �closer� r* r* r+ r� � s z AsynPool._untrack_child_processc s� �j �� � �j j�_��� � ��� � ��� � � �fdd��jD � t�j �j � j �jd� �j � � D ]\}}� � ||� ql�js�� j��j� d�_dS )z4Register the async pool with the current event loop.c s g | ]}�� |� ��qS r* )r� �r6 r` �r� r~ r* r+ rL r: z5AsynPool.register_with_event_loop.<locals>.<listcomp>ri TN)�_result_handlerr� r� �handle_result_event�_create_timelimit_handlers�_create_process_handlers�_create_write_handlersr� ry r� r� �timers�itemsZcall_repeatedly�_registered_with_event_loopZon_tickrS � on_poll_start)r~ r� �handler�intervalr* r� r+ r� � s �z!AsynPool.register_with_event_loopc sR �j �t� ��_����fdd�}|�_�fdd�� � �_� fdd�}|�_dS )z.Create handlers used to implement time limits.c s@ |r"� |�j | j||���| j< n|r<� |�j| j��| j< d S r- )�_on_soft_timeout�_job�_on_hard_timeout)rX �soft�hard)� call_laterr� r~ �trefsr* r+ �on_timeout_set s � �z;AsynPool._create_timelimit_handlers.<locals>.on_timeout_setc s4 z� � | �}|�� ~W n ttfy. Y n0 d S r- )rv �cancelr� rD )rE Ztref)r� r* r+ � _discard_tref) s z:AsynPool._create_timelimit_handlers.<locals>._discard_trefc s � | j � d S r- )r� )rX )r� r* r+ �on_timeout_cancel2 s z>AsynPool._create_timelimit_handlers.<locals>.on_timeout_cancelN)r� r � _tref_for_idr� r� r� )r~ r� r� r� r* )r� r� r� r~ r� r+ r� s z#AsynPool._create_timelimit_handlersc C sr |r |� || | j|�| j|< z>z| j| }W n ty@ Y n 0 | �|� W |sn| �|� n|sl| �|� 0 d S r- )r� r� r� �_cacher� r� r� )r~ rE r� r� r� �resultr* r* r+ r� 6 s � �zAsynPool._on_soft_timeoutc C sL z:z| j | }W n ty" Y n 0 | �|� W | �|� n | �|� 0 d S r- )r� r� r� r� )r~ rE r� r* r* r+ r� G s zAsynPool._on_hard_timeoutc C s | � |� d S r- )r� )r~ rE r� �obj�inqW_fdr* r* r+ � on_job_readyS s zAsynPool.on_job_readyc s� � j � j� j �� � � j�� j�� j�� j�� j�� j�� j �� j � � j ��� �fdd������� � ��fdd�}|� _ d dd�� � ������ � � � � �f dd �}|� _ dS ) z/Create handlers called on process up/down, etc.c sj | � } | d urf| � � rf| �v rf| j� v s,J �� | j | u s>J �| j�jv sNJ �td| � t�| jd� d S )Nz(Timed out waiting for UP message from %r� )� _is_aliver� rT �errorr� �killr �r� )r� r� �waiting_to_startr* r+ �verify_process_alivee s � z?AsynPool._create_process_handlers.<locals>.verify_process_alivec s� | j }��� D ]4}|jr*|jj |kr*| |_|jr|jj |kr| |_q| �| j< ��| �� t| jj�rjJ �� | j�| j� �� | � �� �j �t | �� dS )z"Called when a process has started.N) r �values� _write_to�_scheduled_forr� r� r r{ r� rS r� r� r )r� �infdrE )r� r� r� r� r� r~ r r r* r+ � on_process_upo s �z8AsynPool._create_process_handlers.<locals>.on_process_upNc S sn z | � � }W n ty Y d S 0 z || |u r<|�|d � W n tyP Y n0 ||� |d urj||� |S r- )rR rb rv r� )r r� �indexZ remove_funr� r% r* r* r+ �_remove_from_index� s z=AsynPool._create_process_handlers.<locals>._remove_from_indexc s� t | dd�rdS �| � � | jj| ��� | jrB� | jj| �� � � | jj| �� �jd�}|rh��|� � �| �� � �| � � j�| j � � | jj� �| jj� | j r��| jj� | j rֈ j�| j � �| jj� dS )z#Called when a worker process exits.r� N�r� ) r� r{ r� �synqrC �inqre r� r� r ZsynqR_fdr� )r� r ) r � all_inqueues� busy_workers� fileno_to_inqr� �fileno_to_synqr� �process_flush_queuesr� � remove_writerr~ r r* r+ �on_process_down� s4 � � � z:AsynPool._create_process_handlers.<locals>.on_process_down)N)r� r� r r� r� r� r� r� r� r� r r� r r )r~ r� r r r* )r r� r r r� r r� r r� r� r r� r r~ r r r+ r� V s" � " z!AsynPool._create_process_handlersc s� �j ��j� �j��j��j��j��j��j}�j��j �� j �� j � j � ��j �|j ��j �|j � �jj� �j� �jtk�tj�tj� t��td�t��td�i�tjf� ��fdd� }|�_����� ���fdd�}|�_ ����fdd�}|�_ ��_ d��������������fd d � } | � _ �� ���fd d �} | �_ � �fd d������ � fdd��� ����� fdd�} | �_!d�� fdd� � dS )z6Create handlers used to write data to child processes.)r c sR | j d us| j�v r<| js.| �d |� � � d � | �| j � n| �vrN��| � d S r- )Z _terminatedZcorrelation_id� _acceptedZ_ackZ_set_terminated� appendleft)rE �_time)�getpid�outbound� revoked_tasksr* r+ � _put_back� s �z2AsynPool._create_write_handlers.<locals>._put_backc sT �� �} �r"�o t ��t ��k }n�}|rDt| ��d ttB dd� n t| ��� d S )NT)Z consolidate)r# ry r r )ZinactiveZadd_cond)� active_writesr r �diff�hub_add� hub_remove�is_fair_strategyr r* r+ r� � s ��z6AsynPool._create_write_handlers.<locals>.on_poll_startc sR �� | � z0�| |u r8��| d � � � | � �� | � W n tyL Y n0 d S r- )re rv r� )r% r� )r" r r r r* r+ �on_inqueue_close s z9AsynPool._create_write_handlers.<locals>.on_inqueue_closeNc sr |s dg}t | �}t|�D �]P}| |d | }|d d7 < |�v rJq�rX|�v rXq|�vrj�|� qz � � }W n0 ty� ���D ] }�|� q�Y �qnY q0 |jsz�| }|_W n ty� � |� Y qY n0 � |||�}t|�|_� |� � |� � |� z t|� W nH t �y, Y q t �y` } z | j t j k�rL� W Y d } ~ qd } ~ 0 0 �||� qd S )Nr r2 ) r# r� � IndexErrorr r r� r rC r� r� rb rc �EBADF) Z ready_fdsZtotal_write_countZ num_readyr� Zready_fdrE Zinqfdr� �corrf )� _write_jobr" � add_writerr r r# r r% r& �mark_worker_as_busy�mark_write_fd_as_active�mark_write_gen_as_active� pop_message� put_messager* r+ �schedule_writes sL z8AsynPool._create_write_handlers.<locals>.schedule_writesc sN � | �d�}t |�}�d|�}�| d d �}t|�t|�|f|_�|� d S )N��protocol�>Ir2 r )r# r� �_payload)�tup�bodyr� �headerrE )�dumps�get_jobr r4 r1 r* r+ �send_joba s z1AsynPool._create_write_handlers.<locals>.send_jobc s: t �d| | j|� | �� r"| �� � �|� ��|� d S )Nz"Process inqueue damaged: %r %r: %r)rr � exception�exitcoder � terminaterp r! )r� r% rE rf r� r* r+ �on_not_recoveringm s � z:AsynPool._create_write_handlers.<locals>.on_not_recoveringc 3 s� |j \}}}d}�zR| |_| j}d }} |dk r�z||||�7 }W n` ty� } zHt| dd �tvrd� |d7 }|dkr��| ||| � t� �d V W Y d } ~ q(d } ~ 0 0 d}q(| |k �r2z| ||| �7 } W nd t�y* } zJt| dd �tvr� |d7 }|dk�r�| ||| � t� �d V W Y d } ~ q�d } ~ 0 0 d}q�W �|� �| j d7 < � �|� �|� � � n2�|� �| j d7 < � �|� �|� � � 0 d S )Nr r3 rc r2 �d ) r6 r Zsend_job_offset� Exceptionr� r� r� r re rC ) r� r% rE r9 r8 r� �errorsr� �Hw�Bwrf )r" r% r@ �write_generator_doner� r* r+ r+ u sL � z3AsynPool._create_write_handlers.<locals>._write_jobc sL t ||�| �}t��}� |||d�}�|� �|� |f|_�||� d S )Nr )r; r rl )�responser rE r% �msgr� r* )� _write_ackr, r. r/ �precalcrF r* r+ �send_ack� s z1AsynPool._create_write_handlers.<locals>.send_ackc 3 s0 |d \}}}�zz �| }W n t y6 t� �Y n0 |j}d }} |dk r�z||||�7 }W qF ty� } z$t| dd �tvr�� d V W Y d } ~ qFd } ~ 0 0 qF| |k r�z| ||| �7 } W q� ty� } z$t| dd �tvr܂ d V W Y d } ~ q�d } ~ 0 0 q�W |�r|� � �| � n|�r |� � �| � 0 d S )N� r r3 rc )r� r� Zsend_syn_offsetrB r� r� re ) r% Zackr� r9 r8 r� r� r� rD rE rf )r" r r* r+ rI � s8 �z3AsynPool._create_write_handlers.<locals>._write_ack)N)N)"r� r� r� �popleftrt r� r� r� r� � differencer, rS rp re r� � __getitem__r� r� �SCHED_STRATEGY_FAIR� worker_stateZrevokedr� r r �_create_payloadr �timer! r� r' r% Zconsolidate_callback� _quick_putrK ) r~ r� r r: r4 Zactive_writersr! r� r' r2 r< rK r* ) rI r+ r" r, r r r# r: r r r; r r� r$ r% r& r- r. r/ r@ r r r0 rJ r4 r1 r r~ rF r� r+ r� � sP � (G 3 z AsynPool._create_write_handlersc C s� | j tkrd S | jr2| j�� D ]}|js |�� q | jrB| j�� | � � �zJ| j t k�rlt ddddd�}i }| j�� D ] }t |�}|d urx|||< qx| j s�| j�� n�| j �rXt| j �}|D ]�}|jdk�rt|��rz || }W n ty� Y n 0 |�� | j �|� q�z || }W n t�y. Y q�0 |j}|�� �rL| �||� |�� q�q�| � � tt|�� W | j�� | j �� | j�� | j�� n*| j�� | j �� | j�� | j�� 0 d S )N�{�G�z�?g�������?T)Z repeatlastr+ )r� r r� r� r r Z_cancelr� �clearr� r r rG r� r^ r� rB r� re r r � _flush_writerr r� r� r� )r~ rE Z intervalsZowned_byrF rU rA Zjob_procr* r* r+ �flush� sb � � zAsynPool.flushc C s� |j jh}zj|rf|�� sqft||dd�\}}}|s |s:|r z t|� W q tttfyb Y qfY q 0 q W | j� |� n| j� |� 0 d S )Nr� )rU rV rW ) r rC r rh r� r� rb r� r� re )r~ r� rF �fds�readable�writable�againr* r* r+ rW # s � zAsynPool._flush_writerc C s t dd� | j�� D ��S )z�Get queues for a new process. Here we'll find an unused slot, as there should always be one available when we start a new process. c s s | ]\}}|d u r|V qd S r- r* �r6 �q�ownerr* r* r+ � <genexpr>: s �z.AsynPool.get_process_queues.<locals>.<genexpr>)r� r� r� r� r* r* r+ �get_process_queues4 s zAsynPool.get_process_queuesc s< t � jt� j� d�}|r8� j�� fdd�t|�D �� dS )z!Grow the pool by ``n`` processes.r c s i | ]}� � � d �qS r- r� r� r� r* r+ r9 A s z$AsynPool.on_grow.<locals>.<dictcomp>N)�maxZ _processesr# r� �updater� )r~ r) r# r* r� r+ �on_grow= s �zAsynPool.on_growc C s dS )z#Shrink the pool by ``n`` processes.Nr* )r~ r) r* r* r+ � on_shrinkE s zAsynPool.on_shrinkc C s� t dd�}t dd�}d}t|j�s&J �t|j�r4J �t|j�rBJ �t|j�sPJ �| jr|t dd�}t|j�snJ �t|j�r|J �|||fS )z5Create new in, out, etc. queues, returned as a tuple.T)Z wnonblock)Z rnonblockN)r r r� rC r� )r~ r r{ r r* r* r+ r� H s z AsynPool.create_process_queuesc s� z t � fdd�| jD ��}W n ty: t�d� � Y S 0 |j| jvsLJ �|j| jvs\J �| j� |� || j|j< || j |j < | j� |j� dS )zsCalled when receiving the :const:`WORKER_UP` message. Marks the process as ready to receive work. c 3 s | ]}|j � kr|V qd S r- �r r� rf r* r+ r` ` r: z,AsynPool.on_process_alive.<locals>.<genexpr>z"process with pid=%s already exitedN) r� r� r� rr rs r r� r� r� re r� r� rS )r~ r r� r* rf r+ r� Z s zAsynPool.on_process_alivec C s> |j r |j �� s | �||j � n|jr:|j�� s:| �|� dS )z:Called for each job when the process assigned to it exits.N)r r �on_partial_readr r! )r~ rE Zpid_goner* r* r+ �on_job_process_downj s z AsynPool.on_job_process_downc C s | � ||� dS )z�Called when the process executing job' exits. This happens when the process job' was assigned to exited by mysterious means (error exitcodes and signals). N)Zmark_as_worker_lost)r~ rE r r> r* r* r+ �on_job_process_lostt s z AsynPool.on_job_process_lostc s� | j d u rdS t| j �� �}t|��dd� � �� �rB�t| j � nd��d�� �fdd�|D ��d�tt|��t� | j | j �t| j �t| j �d�d �S ) NzN/Ac S s | rt | �| ndd�S )Nr z.2f)�float)r8 �totalr* r* r+ �per� s z'AsynPool.human_write_stats.<locals>.perr z, c 3 s | ]}� |��V qd S r- r* )r6 r8 �rl rk r* r+ r` � r: z-AsynPool.human_write_stats.<locals>.<genexpr>)rk Zactive)rk Zavg�all�rawZstrategyZinqueues) r� r^ r �sumr# �join�map�str�SCHED_STRATEGY_TO_NAMEr� r� r� r� )r~ �valsr* rm r+ �human_write_stats} s ���zAsynPool.human_write_statsc C s6 |j s2zd| j| �|�< W n ttfy0 Y n0 dS )z-Called to clean up queues after process exit.N)r� r� �_find_worker_queuesr� rw �r~ r� r* r* r+ �_process_cleanup_queues� s z AsynPool._process_cleanup_queuesc C sz | j D ]n}zt|jjd� W n ty. Y q0 z|j�d� W q tyr } z|jtjkr^� W Y d}~qd}~0 0 qdS )z>Called at shutdown to tell processes that we're shutting down.r2 N)r r r rC rb r| rc r) )Z task_handlerr� rf r* r* r+ �_stop_task_handler� s zAsynPool._stop_task_handlerc s t � j| j| jd�S )N)r� r� )r� �create_result_handlerr� r� r� r� r* r+ r{ � s �z AsynPool.create_result_handlerc C s8 || j v sJ �t| j �}|| j |< |t| j �ks4J �dS )z;Mark new ownership for ``queues`` to update fileno indices.N)r� r# )r~ r� �queues�br* r* r+ �_process_register_queues� s z!AsynPool._process_register_queuesc s> z t � fdd�| j�� D ��W S ty8 t� ��Y n0 dS )z"Find the queues owned by ``proc``.c 3 s | ]\}}|� kr|V qd S r- r* r] r r* r+ r` � s �z/AsynPool._find_worker_queues.<locals>.<genexpr>N)r� r� r� r� rw rx r* r r+ rw � s z AsynPool._find_worker_queuesc C s" d | _ d | _ | _ | _| _d S r- )rT Z_inqueue� _outqueueZ _quick_getZ _poll_resultr� r* r* r+ � _setup_queues� s ��zAsynPool._setup_queuesc C s |j j}| jj}|h}|r�|js�| jtkr�t|d|dd�\}}}|r�z |�� }W n� t t fy� } zlt |dd�} | t j kr�W Y d}~qn0| t jkr�W Y d}~q�n| tvr�td||dd� W Y d}~q�W Y d}~q�d}~0 0 |du r�td|� q�q�||� qq�qdS ) a Flush all queues. Including the outbound buffer, so that all tasks that haven't been started will be discarded. In Celery this is called whenever the transport connection is lost (consumer restart), and when a process is terminated. NrU �rW rc z got %r while flushing process %rr2 rn z&got sentinel while flushing process %r)r{ r� r� r� �closedr� r rh r� rb r� r� rc rd �EAGAINr� r� ) r~ r� Zresqr� rY rZ r� r� rf rg r* r* r+ r � s0 �" z AsynPool.process_flush_queuesc C s� |j s| �|� t|�}|r*| j�|� ~|js�d|_t| j�}z(| �|�}| � ||�rfd| j| � � < W n t yz Y n0 t| j�|ks�J �dS )z8Called when a job was partially written to exited child.TN) r r! rG r� re r� r# r� rw �destroy_queuesr� rw )r~ rE r� rF �beforer| r* r* r+ rg � s zAsynPool.on_partial_readc C s� |� � r J �| j�|� d}z| j�|� W n tyB d}Y n0 z| �|d j�� |� W n t yp Y n0 |D ]J}|rv|j |jfD ]4}|j s�| � |� z |� � W q� t y� Y q�0 q�qv|S )zqDestroy queues that can no longer be used. This way they can be replaced by new usable sockets. r2 r )r r� re r� rv r� r' rC rR rb r� r� r% r� )r~ r| r� Zremoved�queue�sockr* r* r+ r� s* zAsynPool.destroy_queuesc C s, |||f|d�}t |�}|d|�}|||fS )Nr3 r5 )r# ) r~ �type_rl r: r r4 r8 r' r9 r* r* r+ rR # s zAsynPool._create_payloadc C s d S r- r* )�clsr r� r* r* r+ �_set_result_sentinel+ s z AsynPool._set_result_sentinelc C s | j fS r- )r� r� r* r* r+ �_help_stuff_finish_args0 s z AsynPool._help_stuff_finish_argsc C s� t d� i }t� }|D ]:}z"|jj�� }|�|� |||< W q tyN Y q0 q|r�t|dd�\}}}|rnqR|stq�|D ]}|| jj�� qxt d� qRd S )Nz7removing tasks from inqueue until task handler finishedr� r� r ) r� rO r r� rR rS rb rh r� r ) r� r Zfileno_to_procZinqRr` r% rZ r� r\ r* r* r+ �_help_stuff_finish5 s* � zAsynPool._help_stuff_finishc C s | j diS )Ng @)r� r� r* r* r+ r� N s zAsynPool.timers)NFNN)4r� r� r� r� r� rz r� r� r� r� r� r� r� r� r� r� r� r r� r r� r: r r� rX rW ra rd re r� r� rh ri rv ry � staticmethodrz r{ r~ rw r� r rg r� rR � classmethodr� r� r� �propertyr� r� r* r* r� r+ r0 � sj �> k� H $ � r0 )NNNr )_r� rc r� r? r� r] rS � collectionsr r r �ior �numbersr r r �structr r r r �weakrefr r Zbilliardr r� Zbilliard.compatr r Z billiard.poolr r r r r Zbilliard.queuesr Zkombu.asynchronousr r Zkombu.serializationr� Zkombu.utils.eventior Zkombu.utils.functionalr Zviner Zcelery.signalsr Zcelery.utils.functionalr Zcelery.utils.logr Z celery.workerr rQ Z _billiardr! r, r� � ImportError�__all__r� rr r r� � frozensetr� rd r� r} r� ZSCHED_STRATEGY_FCFSrP r� r� rt r; rB rG ru rH rJ rM rN r\ rh ry rz r� ZPoolr0 r* r* r* r+ �<module> s� � � � 0-
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка