File: //usr/local/lib/python3.9/site-packages/celery/worker/consumer/__pycache__/mingle.cpython-39.pyc
a
X>h� � @ s` d Z ddlmZ ddlmZ ddlmZ dZee�Z e j
e j e j Z
Z Z G dd� dej
�Zd S )
z-Worker <-> Worker Sync at startup (Bootstep).� )� bootsteps)�
get_logger� )�Events)�Minglec sv e Zd ZdZd ZefZh d�Zd� fdd� Zdd� Z dd � Z
d
d � Z d d
� Z dd� Z
ddd�Zdd� Zdd� Z� ZS )r z�Bootstep syncing state with neighbor workers.
At startup, or upon consumer restart, this will:
- Sync logical clocks.
- Sync revoked tasks.
> ZgcpubsubZamqpZredisFc s0 | o| � |j�| _t� j|fd|i|�� d S )N�without_mingle)�compatible_transport�app�enabled�super�__init__)�self�cr �kwargs�� __class__� �G/usr/local/lib/python3.9/site-packages/celery/worker/consumer/mingle.pyr s ���zMingle.__init__c C s: |� � � }|jj| jv W d � S 1 s,0 Y d S �N)Zconnection_for_read� transportZ driver_type�compatible_transports)r
r �connr r r r s
zMingle.compatible_transportc C s | � |� d S r )�sync)r
r r r r �start$ s z Mingle.startc s` t d� ��� �}|rTt dtdd� |�� D ��� � �fdd�|�� D � t d� nt d� d S )Nz mingle: searching for neighborszmingle: sync with %s nodesc S s g | ]\}}|r|�qS r r )�.0�reply�valuer r r �
<listcomp>, � z Mingle.sync.<locals>.<listcomp>c s" g | ]\}}|r�� � ||��qS r )�
on_node_reply)r �nodenamer �r r
r r r - s
�zmingle: sync completezmingle: all alone)�info�
send_hello�len�items)r
r �repliesr r! r r ' s
� �
z Mingle.syncc C sD |j jjd|jd�}|jjj}|�|j|j �p0i }|�
|jd � |S )Ng �?)�timeout�
connection) r �control�inspectr( �
controller�state�revokedZhello�hostname�_data�pop)r
r r* Z our_revokedr& r r r r# 3 s
zMingle.send_helloc
C sh t d|� z| j|fi |�� W nB ty4 � Y n0 tyb } ztd||� W Y d }~n
d }~0 0 d S )Nz mingle: processing reply from %sz mingle: sync with %s failed: %r)�debug�sync_with_node� MemoryError� Exception� exception)r
r r r �excr r r r : s
zMingle.on_node_replyNc K s | � ||� | �||� d S r )�on_clock_event�on_revoked_received)r
r �clockr- r r r r r2 C s zMingle.sync_with_nodec C s" |r|j j�|�n
|j j�� d S r )r r9 �adjustZforward)r
r r9 r r r r7 G s zMingle.on_clock_eventc C s |r|j jj�|� d S r )r+ r, r- �update)r
r r- r r r r8 J s zMingle.on_revoked_received)F)NN)�__name__�
__module__� __qualname__�__doc__�labelr �requiresr r r r r r# r r2 r7 r8 �
__classcell__r r r r r
s
r N)r? Zceleryr Zcelery.utils.logr �eventsr �__all__r< �loggerr1 r" r5 Z
StartStopStepr r r r r �<module> s