o h@sdZddlmZddlmZddlmZddlmZddl m Z ddl m Z ddl mZmZdd lmZdd lmZdd lmZd d lmZdZeeZejejZZGdddejZdS)z)Worker <-> Worker communication Bootstep.) defaultdict)partial)heappush) itemgetter)Consumer) DummyLock)ContentDisallowed DecodeError) bootsteps) get_logger)Bunch)Mingle)GossipcseZdZdZdZefZedddddddZd d hZ d-fdd Z ddZ d.ddZ ddZ ddZfddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+d,ZZS)/rzfBootstep consuming events from other workers. This keeps the logical clock value up to date. idclockhostnamepidtopicactioncveramqpredisF@@c s| o||j|_|j|_||_|jjj|_|j|_d|jt|j g|_ t t t t d|_ |j|_|jrR|jjj|j|jdd|_|jrMt|_|jj|_||_||_d|_tt|_i|_|j|j d|_!|jj"|_"d|j#i|_$t%j&|fi|dS)N.) node_join node_leave node_lostr ) on_node_join on_node_leavemax_tasks_in_memory)z worker.electzworker.elect.acktask)'compatible_transportappenabledgossipeventsReceiverrjoinstrr full_hostnamer setontimerStaterr statehubr_mutexevent update_stateintervalheartbeat_interval_trefrlistconsensus_requestsconsensus_replieson_elect on_elect_ackevent_handlersr call_taskelection_handlerssuper__init__)selfcwithout_gossipr5r6kwargs __class__O/var/www/html/env/lib/python3.10/site-packages/celery/worker/consumer/gossip.pyrA$sB    zGossip.__init__cCs:|}|jj|jvWdS1swYdSN)connection_for_read transport driver_typecompatible_transports)rBr$connrHrHrIr#Ms  $zGossip.compatible_transportNcCs$g|j|<|jjd|||dddS)Nz worker-electr )rrrr)r: dispatchersend)rBrrrrHrHrIelectionQs  zGossip.electionc CsJz |j|WdSty$}z td|WYd}~dSd}~ww)NzCould not call task: %r)r$ signature apply_async Exceptionlogger exception)rBr"excrHrHrIr>Xs zGossip.call_taskc Csz||\}}}}}}}Wnty&} z td| WYd} ~ Sd} ~ wwt|j|||d|||f|jjd|ddS)Nz!election request missing field %srzworker-elect-ack)r)_cons_stamp_fieldsKeyErrorrVrWrr9rPrQ) rBr3id_rrrrr_rXrHrHrIr;^s zGossip.on_electcst||j|_dSrJ)r@startevent_dispatcherrP)rBrCrFrHrIr]js  z Gossip.startc Cs|d}z|j|}Wn tyYdSwt|j}||dt|t|kru|j|j |\}}}}||j kr_t d|z|j |} WntyYt d|Yn w| |nt d|||j |d|j|ddSdS)NrrzI won the election %rzUnknown election topic %rznode %s elected for %r)r:rZr,r0 alive_workersappendlenr sort_heapr9r+infor?rVrWpop) rBr3rrepliesr_r\leaderrrhandlerrHrHrIr<ns0      zGossip.on_elect_ackcC td|j||jj|dS)Nz%s joined the party)debugr_call_handlersr-rrBworkerrHrHrIr zGossip.on_node_joincCrh)Nz%s left)rirrjr-rrkrHrHrIr rmzGossip.on_node_leavecCrh)Nzmissed heartbeat from %s)rcrrjr-rrkrHrHrI on_node_lostrmzGossip.on_node_lostc OsR|D]$}z ||i|Wqty&}z td||WYd}~qd}~wwdS)Nz!Ignored error from handler %r: %r)rUrVrW)rBhandlersargsrErgrXrHrHrIrjszGossip._call_handlerscCs,|jdur |j|j|j|j|_dSrJ)r7cancelr.call_repeatedlyr5periodic)rBrHrHrIregister_timers  zGossip.register_timercCsR|jj}t}|D]}|js||||q |D] }||jdqdSrJ) r0workersr,valuesaliveaddrnrdr)rBrudirtyrlrHrHrIrss   zGossip.periodiccCs>||j|d|jd}t||jgt|j|j|jddgS)Nzworker.#) routing_key queue_ttlT)queues on_messageacceptno_ack) rtr(r6rqueuerr}event_from_messager~)rBchannelevrHrHrI get_consumerss zGossip.get_consumersc Cs|jd}|ddddkrdSz|j|}Wn ty!Ynw||jS|jdp1|jd}||jkrbz||j\}}||WdSt t t fya}z t |WYd}~dSd}~ww|jdS)Nrzrr rr"r) delivery_infosplitr=rZpayloadheadersgetrr4r r TypeErrorrVerrorrforward) rBpreparemessage_typergrr\r3rXrHrHrIr}s*     zGossip.on_message)FrrrJ)__name__ __module__ __qualname____doc__labelrrequiresrrYrNrAr#rRr>r;r]r<rr rnrjrtrsrr} __classcell__rHrHrFrIrs2)    rN)r collectionsr functoolsrheapqroperatorrkomburkombu.asynchronous.semaphorerkombu.exceptionsrr celeryr celery.utils.logr celery.utils.objectsr mingler__all__rrVrirc ConsumerSteprrHrHrHrIs