o
    wvXj'  ã                   @   sØ   d Z ddlmZmZ ddlmZ ddlmZ ddlm	Z	 ddl
mZ ddlmZ ddlmZ dd	lmZmZ dd
lmZ ddlmZ ddlmZ ddlmZ ddlmZ dZeeƒZej ej!Z Z!G dd„ dej"ƒZ#dS )z)Worker <-> Worker communication Bootstep.é    )Úabsolute_importÚunicode_literals)Údefaultdict)Úpartial)Úheappush)Ú
itemgetter)ÚConsumer)Ú	DummyLock)ÚContentDisallowedÚDecodeError)Ú	bootsteps)Úvalues)Ú
get_logger)ÚBunché   )ÚMingle)ÚGossipc                       sÂ   e Zd ZdZd ZefZedddddddƒZd	d
hZ			d-‡ fdd„	Z
dd„ Zd.dd„Zdd„ Zdd„ Z‡ fdd„Zdd„ Zdd„ Zdd „ Zd!d"„ Zd#d$„ Zd%d&„ Zd'd(„ Zd)d*„ Zd+d,„ Z‡  ZS )/r   zfBootstep consuming events from other workers.

    This keeps the logical clock value up to date.
    ÚidÚclockÚhostnameÚpidÚtopicÚactionÚcverÚamqpÚredisFç      @ç       @c                    s  | o|   |j¡| _|j| _| |_|jjj| _|j| _d | jt|j	ƒg¡| _
ttƒ tƒ tƒ d�| _|j| _| jrR|jjj| j| jdd�| _|jrMtƒ |_| jj| _|| _|| _d | _ttƒ| _i | _| j| j dœ| _!|jj"| _"d| j#i| _$t%t&| ƒj'|fi |¤Ž d S )NÚ.)Ú	node_joinÚ
node_leaveÚ	node_lostr   )Úon_node_joinÚon_node_leaveÚmax_tasks_in_memory)zworker.electzworker.elect.ackÚtask)(Úcompatible_transportÚappÚenabledÚgossipÚeventsÚReceiverr   ÚjoinÚstrr   Úfull_hostnamer   ÚsetÚonÚtimerÚStater"   r#   ÚstateÚhubr	   Ú_mutexÚeventÚupdate_stateÚintervalÚheartbeat_intervalÚ_trefr   ÚlistÚconsensus_requestsÚconsensus_repliesÚon_electÚon_elect_ackÚevent_handlersr   Ú	call_taskÚelection_handlersÚsuperr   Ú__init__)ÚselfÚcÚwithout_gossipr8   r9   Úkwargs©Ú	__class__© úZ/var/www/html/myproject/venv/lib/python3.10/site-packages/celery/worker/consumer/gossip.pyrD   '   sB   ýý

þ
ÿzGossip.__init__c                 C   s:   |  ¡ �}|jj| jv W  d   ƒ S 1 sw   Y  d S ©N)Úconnection_for_readÚ	transportÚdriver_typeÚcompatible_transports)rE   r'   ÚconnrK   rK   rL   r&   P   s   
$ÿzGossip.compatible_transportNc                 C   s$   g | j |< | jjd|||dd� d S )Nzworker-electr   )r   r   r   r   )r=   Ú
dispatcherÚsend)rE   r   r   r   rK   rK   rL   ÚelectionT   s
   

þzGossip.electionc              
   C   sJ   z| j  |¡ ¡  W d S  ty$ } zt d|¡ W Y d }~d S d }~ww )NzCould not call task: %r)r'   Ú	signatureÚapply_asyncÚ	ExceptionÚloggerÚ	exception)rE   r%   ÚexcrK   rK   rL   rA   [   s   €ÿzGossip.call_taskc           
   
   C   s‚   z|   |¡\}}}}}}}W n ty& }	 zt d|	¡W  Y d }	~	S d }	~	ww t| j| |d||f ||fƒ | jjd|d� d S )Nz!election request missing field %sz%s.%szworker-elect-ack)r   )Ú_cons_stamp_fieldsÚKeyErrorrY   rZ   r   r<   rS   rT   )
rE   r6   Úid_r   r   r   r   r   Ú_r[   rK   rK   rL   r>   a   s   
ÿ
€ÿþzGossip.on_electc                    s   t t| ƒ |¡ |j| _d S rM   )rC   r   ÚstartÚevent_dispatcherrS   )rE   rF   rI   rK   rL   r`   m   s   zGossip.startc           
      C   sî   |d }z| j | }W n
 ty   Y d S w t| j ¡ ƒ}| |d ¡ t|ƒt|ƒkru| j | j	| ¡\}}}}|| j
kr_td|ƒ z| j| }	W n tyY   t d|¡ Y nw |	|ƒ ntd||ƒ | j	 |d ¡ | j  |d ¡ d S d S )Nr   r   zI won the election %rzUnknown election topic %rznode %s elected for %r)r=   r]   r/   r3   Úalive_workersÚappendÚlenr   Ú	sort_heapr<   r.   ÚinforB   rY   rZ   Úpop)
rE   r6   r   Úrepliesrb   r_   Úleaderr   r   ÚhandlerrK   rK   rL   r?   q   s0   ÿÿ

ÿ
ñzGossip.on_elect_ackc                 C   ó    t d|jƒ |  | jj|¡ d S )Nz%s joined the party)Údebugr   Ú_call_handlersr0   r   ©rE   ÚworkerrK   rK   rL   r"   ‹   ó   zGossip.on_node_joinc                 C   rk   )Nz%s left)rl   r   rm   r0   r    rn   rK   rK   rL   r#   �   rp   zGossip.on_node_leavec                 C   rk   )Nzmissed heartbeat from %s)rf   r   rm   r0   r!   rn   rK   rK   rL   Úon_node_lost“   rp   zGossip.on_node_lostc                 O   sR   |D ]$}z	||i |¤Ž W q t y& } zt d||¡ W Y d }~qd }~ww d S )Nz!Ignored error from handler %r: %r)rX   rY   rZ   )rE   ÚhandlersÚargsrH   rj   r[   rK   rK   rL   rm   —   s   ÿ€ÿýzGossip._call_handlersc                 C   s,   | j d ur
| j  ¡  | j | j| j¡| _ d S rM   )r:   Úcancelr1   Úcall_repeatedlyr8   Úperiodic)rE   rK   rK   rL   Úregister_timerŸ   s   

zGossip.register_timerc                 C   sR   | j j}tƒ }t|ƒD ]}|js| |¡ |  |¡ q|D ]	}| |jd ¡ qd S rM   )	r3   Úworkersr/   r   ÚaliveÚaddrq   rg   r   )rE   rx   Údirtyro   rK   rK   rL   rv   ¤   s   

€ÿzGossip.periodicc                 C   s:   |   ¡  | j|d| jd�}t||jgt| j|jƒdd�gS )Nzworker.#)Úrouting_keyÚ	queue_ttlT)ÚqueuesÚ
on_messageÚno_ack)rw   r+   r9   r   Úqueuer   r   Úevent_from_message)rE   ÚchannelÚevrK   rK   rL   Úget_consumers®   s   ÿüzGossip.get_consumersc           	   
   C   sÒ   |j d }| dd¡d dkrd S z| j| }W n	 ty!   Y nw ||jƒS |j d¡p1|jd }|| jkrbz||jƒ\}}|  |¡ W d S  t	t
tfya } zt |¡ W Y d }~d S d }~ww | j ¡  d S )Nr|   r   r   r   r%   r   )Údelivery_infoÚsplitr@   r]   ÚpayloadÚheadersÚgetr   r7   r   r
   Ú	TypeErrorrY   Úerrorr   Úforward)	rE   ÚprepareÚmessageÚ_typerj   r   r_   r6   r[   rK   rK   rL   r   ¹   s*   
ÿ
ÿ
€ÿzGossip.on_message)Fr   r   rM   )Ú__name__Ú
__module__Ú__qualname__Ú__doc__Úlabelr   Úrequiresr   r\   rQ   rD   r&   rU   rA   r>   r`   r?   r"   r#   rq   rm   rw   rv   r…   r   Ú__classcell__rK   rK   rI   rL   r      s2    ÿÿ)

r   N)$r”   Ú
__future__r   r   Úcollectionsr   Ú	functoolsr   Úheapqr   Úoperatorr   Úkombur   Úkombu.asynchronous.semaphorer	   Úkombu.exceptionsr
   r   Úceleryr   Úcelery.fiver   Úcelery.utils.logr   Úcelery.utils.objectsr   Úmingler   Ú__all__r‘   rY   rl   rf   ÚConsumerStepr   rK   rK   rK   rL   Ú<module>   s$    