File: //usr/local/lib/python3.9/site-packages/celery/worker/consumer/__pycache__/gossip.cpython-39.pyc
a
X>h� � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl m
Z
ddl m Z ddl
mZmZ dd lmZ dd
lmZ dd lmZ d d
lmZ dZee�Zejej ZZ G dd� dej �Z dS )z)Worker <-> Worker communication Bootstep.� )� defaultdict)�partial)�heappush)�
itemgetter)�Consumer)� DummyLock)�ContentDisallowed� DecodeError)� bootsteps)�
get_logger)�Bunch� )�Mingle)�Gossipc s� e Zd ZdZd ZefZeddddddd�Zd d
hZ d-� fdd� Z
dd� Z d.dd�Z dd� Z
dd� Z� fdd�Zdd � Zd d � Zd d � Zd!d"� Zd#d$� Zd%d&� Zd'd(� Zd)d*� Zd+d,� Z� ZS )/r zfBootstep consuming events from other workers.
This keeps the logical clock value up to date.
�id�clock�hostname�pid�topic�action�cverZamqpZredisF� @� @c s | o| � |j�| _|j| _| |_|jjj| _|j| _d�| jt|j �g�| _
t t � t � t � d�| _
|j| _| jr�|jjj| j| jdd�| _|jr�t� |_| jj| _|| _|| _d | _tt �| _ i | _ | j | j d�| _!|jj"| _"d| j#i| _$t%� j&|fi |�� d S )N�.)� node_join�
node_leave� node_lostr
)� on_node_join�
on_node_leaveZmax_tasks_in_memory)z worker.electzworker.elect.ack�task)'�compatible_transport�app�enabledZgossip�events�Receiverr �join�strr �
full_hostnamer �set�on�timer�Stater r �stateZhubr Z_mutex�event� update_state�interval�heartbeat_interval�_trefr �list�consensus_requests�consensus_replies�on_elect� on_elect_ack�event_handlersr � call_task�election_handlers�super�__init__)�self�cZwithout_gossipr/ r0 �kwargs�� __class__� �G/usr/local/lib/python3.9/site-packages/celery/worker/consumer/gossip.pyr; $ sB ��
�
�zGossip.__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! �connrA rA rB r M s
zGossip.compatible_transportNc C s$ g | j |< | jjd|||dd� d S )Nz worker-electr
)r r r r )r4 �
dispatcher�send)r< r r r rA rA rB �electionQ s
�zGossip.electionc
C sJ z| j �|��� W n0 tyD } zt�d|� W Y d }~n
d }~0 0 d S )NzCould not call task: %r)r! � signatureZ apply_async� Exception�logger� exception)r< r �excrA rA rB r8 X s zGossip.call_taskc
C s� z | � |�\}}}}}}}W n0 tyL } zt�d| �W Y d } ~ S d } ~ 0 0 t| j| ||� d|� �||f� | jjd|d� d S )Nz!election request missing field %sr zworker-elect-ack)r )�_cons_stamp_fields�KeyErrorrL rM r r3 rG rH )
r< r- Zid_r r r r r �_rN rA rA rB r5 ^ s �
�"�zGossip.on_electc s t � �|� |j| _d S rC )r: �startZevent_dispatcherrG )r< r= r? rA rB rR j s z Gossip.startc
C s� |d }z| j | }W n ty* Y d S 0 t| j�� �}|�|d � t|�t|�kr�| j�| j | �\}}}}|| j
kr�t d|� z| j | } W n ty� t
�d|� Y q�0 | |� n t d||� | j �|d � | j �|d � d S )Nr r zI won the election %rzUnknown election topic %rznode %s elected for %r)r4 rP r( r, �
alive_workers�append�lenr Z sort_heapr3 r' �infor9 rL rM �pop)
r< r- r ZrepliesrS rQ Zleaderr r �handlerrA rA rB r6 n s* �
zGossip.on_elect_ackc C s t d|j� | �| jj|� d S )Nz%s joined the party)�debugr �_call_handlersr) r �r< �workerrA rA rB r � s zGossip.on_node_joinc C s t d|j� | �| jj|� d S )Nz%s left)rY r rZ r) r r[ rA rA rB r � s zGossip.on_node_leavec C s t d|j� | �| jj|� d S )Nzmissed heartbeat from %s)rV r rZ r) r r[ rA rA rB � on_node_lost� s zGossip.on_node_lostc O sT |D ]J}z||i |�� W q t yL } zt�d||� W Y d }~qd }~0 0 qd S )Nz!Ignored error from handler %r: %r)rK rL rM )r< �handlers�argsr> rX rN rA rA rB rZ � s �zGossip._call_handlersc C s, | j d ur| j �� | j�| j| j�| _ d S rC )r1 �cancelr* Zcall_repeatedlyr/ �periodic)r< rA rA rB �register_timer� s
zGossip.register_timerc C sR | j j}t� }|�� D ] }|js|�|� | �|� q|D ]}|�|jd � q:d S rC ) r, �workersr( �values�alive�addr] rW r )r< rc Zdirtyr\ rA rA rB ra � s
zGossip.periodicc C s> | � � | j|d| jd�}t||jgt| j|j�|jdd�gS )Nzworker.#)� routing_keyZ queue_ttlT)�queues�
on_message�acceptZno_ack) rb r$ r0 r �queuer ri Zevent_from_messagerj )r< ZchannelZevrA rA rB �
get_consumers� s � �zGossip.get_consumersc
C s� |j d }|�dd�d dkr"d S z| j| }W n tyB Y n 0 ||j�S |j�d�pb|jd }|| jkr�z ||j�\}}| �|� W q� t t
t fy� } zt �
|� W Y d }~q�d }~0 0 n
| j�� d S )Nrg r r
r r r )Z
delivery_info�splitr7 rP �payload�headers�getr r. r r � TypeErrorrL �errorr Zforward) r< �prepare�message�_typerX r rQ r- rN rA rA rB ri � s$
�
"zGossip.on_message)Fr r )N)�__name__�
__module__� __qualname__�__doc__�labelr �requiresr rO rE r; r rI r8 r5 rR r6 r r r] rZ rb ra rl ri �
__classcell__rA rA r? rB r s0 � �)
r N) ry � collectionsr � functoolsr �heapqr �operatorr Zkombur Z kombu.asynchronous.semaphorer Zkombu.exceptionsr r Zceleryr
Zcelery.utils.logr Zcelery.utils.objectsr Zmingler �__all__rv rL rY rV Z ConsumerStepr rA rA rA rB �<module> s