File: //proc/self/root/usr/local/lib/python3.9/site-packages/celery/apps/__pycache__/multi.cpython-39.pyc
a
X>h�? � @ s* d Z ddlZddlZddlZddlZddlZddlmZmZm Z ddl
m Z ddl m
Z
ddlmZ ddlmZ ddlmZ dd lmZmZmZ dd
lmZmZmZm Z dd l m Z d Z d
Z dd� Z!dd� Z"dd� Z#dd� Z$dd� Z%G dd� d�Z&G dd� d�Z'd d � Z(G d d � d �Z)G d d!� d!e�Z*dS )"zStart/stop/manage workers.� N)� OrderedDict�UserList� defaultdict)�partial)�Popen)�sleep)� from_utf8)�cached_property)�
IS_WINDOWS�Pidfile� signal_name)� gethostname� host_format� node_format� nodesplit)�saferepr)�Cluster�Node�celeryc G s d� tf| �S )N� )�join�
CELERY_EXE)�args� r �;/usr/local/lib/python3.9/site-packages/celery/apps/multi.py�
celery_exe s r c C sN |}d| v r&t | �}t|�\}}|} n |� | � �}t |� d|� ��}| ||fS )N�@)r r )�name�prefix�suffix�hostname�nodename� shortnamer r r �build_nodename s �r# c C s t t| ||| ddd�S )Nz%iz%I)r �N�d�h�i�I)r r )r! r" r r r r �build_expander* s �r) c C s. |s| S | � d�r | � d|� �S | � d|� �S )N�--�=r )�
startswith)�opt�valuer r r �
format_opt6 s
r/ c C s dd� | � � D �S )Nc S s< i | ]4\}}t |�d kr*d�|�dd��nd|� �|� �qS )� z--{}�_�-)�len�format�replace)�.0�k�vr r r �
<dictcomp>? s �
�z+_kwargs_to_command_line.<locals>.<dictcomp>)�items)�kwargsr r r �_kwargs_to_command_line> s �r< c @ sD e Zd Zdd� Zdd� Zddd�Zddd �Zdd
d �Zdd
d�ZdS )�NamespacedOptionParserc C s, || _ t� | _g | _d| _tdd� �| _d S )N� c S s t � S �N)r r r r r �<lambda>M � z1NamespacedOptionParser.__init__.<locals>.<lambda>)r r �options�values� passthroughr �
namespaces)�selfr r r r �__init__H s
z NamespacedOptionParser.__init__c C s� dd� | j D �}d}|t|�k r�|| }|dkrHd�||d � �| _q�n�|d dkr�|d dkrt| �|dd � � q�d }t|�|d kr�||d d dkr�||d }|d7 }| �|dd � |� n | j�|� |d7 }qd S ) Nc S s g | ] }|r|�qS r r )r6 �argr r r �
<listcomp>P rA z0NamespacedOptionParser.parse.<locals>.<listcomp>r r* r r2 r0 � )r r3 r rD �process_long_opt�process_short_optrC �append)rF �rargs�posrH r. r r r �parseO s" $ z NamespacedOptionParser.parseNc C s, d|v r|� dd�\}}| j||dd� d S )Nr+ r0 F��short)�split�
add_option�rF rH r. r r r rK d s z'NamespacedOptionParser.process_long_optc C s | j ||dd� d S )NTrQ )rT rU r r r rL i s z(NamespacedOptionParser.process_short_optc C s$ |d u r| j }t|fi | j| ��S r? )rB r rE )rF �ns�defaultsr r r �optmergel s z NamespacedOptionParser.optmergeFc C sB |rdp
d}| j }d|v r2|�d�\}}| j| }|||| < d S )Nr2 r* �:)rB rS rE )rF r r. rR rV r �destr r r rT q s
z!NamespacedOptionParser.add_option)N)N)N)FN) �__name__�
__module__� __qualname__rG rP rK rL rX rT r r r r r= F s
r= c @ s� e Zd ZdZd*dd�Zdd� Zdd� Zd d
� Zd d � Zd
d� Z d+dd�Z
d,dd�Z e j
ddddfdd�Zd-dd�Zdd� Zdd� Zdd � Zed d � �Zed d � �Zed!d"� �Zejd#d"� �Zed$d%� �Z
ed&d'� �Zed(d)� �ZdS ).r z Represents a node in a cluster.Nc C s\ || _ |pdtdd�� �| _|| _|p(d| _| �|p8t� �| _| �� | _ | �
� | _ d | _ d S )Nz-m �workerz--detachr> )
r r �cmdrM �
extra_args�_annotate_with_default_optsr rB �_prepare_expander�expander�
_prepare_argv�argv�_pid)rF r r_ rM rB r` r r r rG } s
�
z
Node.__init__c C sD | j |d< | �|ddgd� | �|ddgd� | �|dgtj� |S ) N�-n� --pidfile�-pz/var/run/celery/%n.pid� --logfile�-fz/var/log/celery/%n%I.log� --executable)r �_setdefaultopt�sys�
executable)rF rB r r r ra � s
z Node._annotate_with_default_optsc C sv |dd � D ]&}z|| W S t y0 Y q 0 q |�|d tj�|��}tj�|�}|rrtj�|�srt�|� |S )Nr0 r )�KeyError�
setdefault�os�path�normpath�dirname�exists�makedirs)rF r% �altr. r- Zdir_pathr r r rm � s
zNode._setdefaultoptc C s | j �dd�\}}t| j ||�S )Nr r0 )r rS r) )rF r" r r r r rb � s �zNode._prepare_expanderc s� � � � j��d�}|�d�d }� j�� }� j�� D ]2\}}|dv r4|�|t|� � |��� |� |� q4d�
|�g}t |� fdd�|�� D � � j g �}� j
r�|� � � j
�f7 }|S )Nr r r0 ) z-Az--appz-bz--brokerz--result-backendz--loaderz--configz --workdirz-Cz
--no-colorz-qz--quietc s g | ]\}}t |� �|���qS r )r/ rc )r6 r- r. �rF r r rI � s �z&Node._prepare_argv.<locals>.<listcomp>)rc r_ rS �indexrB �copyr: �insertr/ �popr �tupler` rM )rF r_ r' rB r- r. re r ry r rd � s(
����zNode._prepare_argvc C s
| � d�S �Nr )�sendry r r r �alive� s z
Node.alivec
C sl | j }|r^zt�||� W n> tyX } z&|jtjkr8� t|| � W Y d }~dS d }~0 0 dS t|| � d S )NFT)�pidrr �kill�OSError�errnoZESRCH�
maybe_call)rF �sigZon_errorr� �excr r r r� � s
z Node.sendc K s | j | jf| j|d�|��S )N)rs �env)� _waitexecre ro )rF r� r; r r r �start� s ���z
Node.startc C sB | � ||�}t|| d�|�|d� t||d�}| j|�� ||d�S )Nr )�argstrr� )r� )� on_signalled�
on_failure)� prepare_argvr� r r �handle_process_exit�wait) rF re rs r� �on_spawnr� r� r� �piper r r r� � s �zNode._waitexecc C s4 |dk r t || | � | S |dkr0t || |� |S r )r� )rF �retcoder� r� r r r r� � s zNode.handle_process_exitc C s( d� |gt|� �}tjt|�t d�S )Nr )�posix)r �list�shlexrS r r
)rF re rs r r r r r� � s zNode.prepare_argvc G s> |D ](}z| j | W S ty* Y q0 qt|d ��d S r )rB rp )rF rx r- r r r �getopt� s z Node.getoptc C s dt | �j� d| j� d�S )N�<z: �>)�typer[ r ry r r r �__repr__� s z
Node.__repr__c C s | � | �dd��S )Nrh ri �rc r� ry r r r �pidfile� s z Node.pidfilec C s | � | �dd��S )Nrj rk r� ry r r r �logfile� s z Node.logfilec C s8 | j d ur| j S zt| j��� W S ty2 Y n0 d S r? )rf r r� Zread_pid�
ValueErrorry r r r r� � s
zNode.pidc C s
|| _ d S r? )rf )rF r. r r r r� s c C s
| j d S )Nrl �rB ry r r r ro s zNode.executablec C s | j f| j S r? )ro re ry r r r �argv_with_executable s zNode.argv_with_executablec K s | |t |�d�S )Nr� )r< )�clsr r; r r r � from_kwargs s zNode.from_kwargs)NNNN)N)N)NN) r[ r\ r] �__doc__rG ra rm rb rd r� r� r� rn ro r� r� r� r� r� r r� r� �propertyr� �setterr� � classmethodr� r r r r r z s> �
�
r c O s | d ur| |i |�� d S r? r )Zfunr r; r r r r� s r� c @ sL e Zd ZeZddd�Zdd� Zdd � Zd
d � Zd d
� Zdd� Z ddd�Z
dS )� MultiParser�
celery workerr> r c C s" || _ || _|| _|| _|| _d S r? )r_ rM r r � range_prefix)rF r_ rM r r r� r r r rG s
zMultiParser.__init__c s� �j }t�j��t|�dk}�j���d�j����d�j�� ��d��dt� ��}��d��pbd���d�j �pt|��d v r�dn����d
d�p��j
}|r�z�� |�| }�W n t y� Y n0 ��
�|� ���|� � ������fd d �|D �S )
Nr0 z--cmdz--appendz
--hostnamerg z--prefixr> z--suffix)z""z''z--range-prefixc
3 s$ | ] }�� �|���� ��V qd S r? )�_node_from_options)r6 r �rM r_ rB �pr rF r r r � <genexpr>= s ��z$MultiParser.parse.<locals>.<genexpr>)rC �dictrB r3 r r} r_ rM r
r r� � _get_rangesr� �_update_ns_opts�_update_ns_ranges)rF r� �names�rangesr r� r r� r rP ) s*
�zMultiParser.parsec C s> t |||�\}} }
| |jv r | n|}t| |||�||�|j�S r? )r# rE r rX rD ) rF r� r r r r_ rM rB � namespacer! r1 r r r r� C s
�z MultiParser._node_from_optionsc C s$ t |d �}dd� td|d �D �S )Nr c S s g | ] }t |��qS r ��str�r6 �nr r r rI L rA z+MultiParser._get_ranges.<locals>.<listcomp>r0 )�int�range)rF r� Z noderanger r r r� J s zMultiParser._get_rangesc C s� t |j�� �D ]l\}}|�� rt|�d }|dk r@td|����z|j|| �|� W q tyx td|����Y q0 qd S )Nr0 r zIndexes start at 1 got: zNo node at index )r� rE r: �isdigitr� rp �update�
IndexError)rF r� r� �ns_name�ns_optsZns_indexr r r r� N s zMultiParser._update_ns_optsc C s^ t |j�� �D ]J\}}d|v s*|rd|v r| �||�D ]}|j| �|� q6|j�|� qd S )N�,r2 )r� rE r: �_parse_ns_ranger� r} )rF r� r� r� r� Zsubnsr r r r� [ s
z MultiParser._update_ns_rangesFc C sr g }d|v r|� d�p|gD ]P}|rbd|v rb|� d�\}}|�dd� tt|�t|�d �D �� q |�|� q |S )Nr� r2 c s s | ]}t |�V qd S r? r� r� r r r r� g s z.MultiParser._parse_ns_range.<locals>.<genexpr>r0 )rS �extendr� r� rM )rF rV r� �ret�spacer� �stopr r r r� b s
�
zMultiParser._parse_ns_rangeN)r� r> r> r> r )F) r[ r\ r] r rG rP r� r� r� r� r� r r r r r� s �
r� c @ s� e Zd ZdZd"dd�Zdd� Zdd� Zd d
� Zd d � Zd
d� Z e
j fdd�Z dde
j fdd�Z
dde
j fdd�Zdde
j fdd�Ze
j dfdd�Zdd� Zd#d d �Zd d � Zed d!� �ZdS )$r z Represent a cluster of workers.Nc C sx || _ |ptd�| _|| _|| _|| _|| _|| _|| _| | _ |
| _
| | _ | | _ |
| _
|| _|| _|| _|| _|| _d S )Nr^ )�nodesr r_ r� �on_stopping_preamble�on_send_signal�on_still_waiting_for�on_still_waiting_progress�on_still_waiting_end�
on_node_start�on_node_restart�on_node_shutdown_ok�on_node_status�on_node_signal�on_node_signal_dead� on_node_down�on_child_spawn�on_child_signalled�on_child_failure)rF r� r_ r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r r r rG r s$ zCluster.__init__c s � fdd�� D �S )Nc s g | ]}� � |��qS r )�
start_node)r6 �nodery r r rI � rA z!Cluster.start.<locals>.<listcomp>r ry r ry r r� � s z
Cluster.startc C s( t | j|� | �|�}t | j||� |S r? )r� r� � _start_noder� )rF r� r� r r r r� � s
zCluster.start_nodec C s |j | j| j| j| jd�S )N)r� r� r� )r� r� r� r� r� )rF r� r r r r� � s �zCluster._start_nodec C s8 | j | jd�D ]$}t| j|t|�� |�|| j� qd S )N��on_down)�getpidsr� r� r� r r� r� )rF r� r� r r r �send_all� s zCluster.send_allc C s | � tj�S r? )r� �signal�SIGKILLry r r r r� � s z Cluster.killc s& g � � �fdd�}�j d||d� � S )Nc s2 t �j| � ��| �}t �j| |� � �|� d S r? )r� r� r� r� rM )r� �retval�ZretvalsrF r r �restart_on_down� s
z(Cluster.restart.<locals>.restart_on_downrJ ��retryr� r� �� _stop_nodes)rF r� r� r r� r �restart� s zCluster.restartc C s | j |||d�S �Nr� r� �rF r� �callbackr� r r r r� � s z Cluster.stoprJ c C s | j |||d�S r� r� r� r r r �stopwait� s zCluster.stopwaitc C sJ |d ur |n| j }t| j|d��}|rF| j|||d�D ]}t||� q6d S )Nr� )r� r� )r� r� r� �shutdown_nodesr� )rF r� r� r� r� r� r r r r� � s
zCluster._stop_nodesc c s t |�}t| j|� t � }|D ]4}t| j|t|�� |�|| j�s |�|� |V q ||8 }|r�t| j|� d}|r�t � }|D ]L}|d7 }t| j |� |�
� s~t| j |� |�|� |V t| j|� q�q~||8 }|rp|t |� spt
t|�� qpt| j� d S )Nr r0 )�setr� r� r� r r� r� �addr� r� r� r� r3 r �floatr� )rF r� r� r� �P� to_remover� Zitsr r r r� � s6
zCluster.shutdown_nodesc C s( | D ]}|j |kr| S qt|��d S r? )r rp )rF r r� r r r �find� s
z Cluster.findc c s& | D ] }|j r|V qt||� qd S r? )r� r� )rF r� r� r r r r� � s zCluster.getpidsc C s( dj t| �tdd� | D ��t| �jd�S )Nz<{name}({0}): {1}>c S s g | ]
}|j �qS r �r r� r r r rI � rA z$Cluster.__repr__.<locals>.<listcomp>r� )r4 r3 r r� r[ ry r r r r� � s �zCluster.__repr__c C s | j S r? )r� ry r r r �data� s z Cluster.data)NNNNNNNNNNNNNNNNN)N)r[ r\ r] r� rG r� r� r� r� r� r� �SIGTERMr� r� r� r� r� r� r� r� r� r� r r r r r o sB �
$
r )+r� r� rr r� r� rn � collectionsr r r � functoolsr �
subprocessr �timer Zkombu.utils.encodingr Zkombu.utils.objectsr Zcelery.platformsr
r r Zcelery.utils.nodenamesr
r r r Zcelery.utils.safereprr �__all__r r r# r) r/ r< r= r r� r� r r r r r �<module> s6 4 R