`fpRdZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZddlmZmZdd lmZdd lmZdd lmZdd lmZmZdd lmZmZddlmZddl m!Z!m"Z"m#Z#m$Z$m%Z%ddl&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1m2Z2ddl3m4Z4ddl5m6Z6m7Z7m8Z8m9Z9m:Z:dZ;ej<Ze)e?Z@e@jAe@jBe@jCe@jDe@jEf\ZAZBZFZDZGdZHdZIdZJdZKdZLdZMdZNd ZOd!ZPd"ZQd#ZRd$ZSGd%d&ZTGd'd(ejUZVdS))zWorker Consumer Blueprint. This module contains the components responsible for consuming messages from the broker, processing the messages and keeping the broker connections up and running. N) defaultdict)sleep) restart_state)RestartFreqExceeded) DummyLock)ContentDisallowed DecodeError)_detect_environment) safe_repr) TokenBucket)ppartialpromise) bootstepssignals) build_tracer)CPendingDeprecationWarningInvalidTaskError NotRegisteredWorkerShutdownWorkerTerminate)noop) get_logger) gethostname)Bunch)truncate)humanize_secondsrate)loops)active_requestsmaybe_shutdownrequestsreserved_requests task_reserved)ConsumerEvloop dump_bodyzMconsumer: Connection to broker lost. Trying to re-establish the connection...z0Trying again {when}... ({retries}/{max_retries})z'consumer: Cannot connect to %s: %s. %s zWill retry using next failover.zkReceived and deleted unknown message. Wrong destination?!? The full contents of the message body was: %s aReceived unregistered task of type %s. The message has been ignored and discarded. Did you remember to import the module containing this task? Or maybe you're using relative imports? Please see https://docs.celeryq.dev/en/latest/internals/protocol.html for more information. The full contents of the message body was: %s The full contents of the message headers: %s The delivery info for this task is: %s a Received invalid task message: %s The message has been ignored and discarded. Please ensure your message conforms to the task message protocol as described here: https://docs.celeryq.dev/en/latest/internals/protocol.html The full contents of the message body was: %s zICan't decode message body: %r [type:%r encoding:%r headers:%s] body: %s zTbody: {0} {{content_type:{1} content_encoding:{2} delivery_info:{3} headers={4}}} z}Task %s cannot be acknowledged after a connection loss since late acknowledgement is enabled for it. Terminating it instead. a In Celery 5.1 we introduced an optional breaking change which on connection loss cancels all currently executed tasks with late acknowledgement enabled. These tasks cannot be acknowledged as the connection is gone, and the tasks are automatically redelivered back to the queue. You can enable this behavior using the worker_cancel_long_running_tasks_on_connection_loss setting. In Celery 5.1 it is set to False by default. The setting will be set to True by default in Celery 6.0. c||jn|}dtt|dt |jS)z+Format message body for debugging purposes.Nz{} ({}b)i)bodyformatrr len)mr(s Q/var/www/html/env/lib/python3.11/site-packages/celery/worker/consumer/consumer.pyr&r&sH\166tD   Xioot<< [[ * **c eZdZdZeZdZdZdZdZ dZ Gdde j Z e ddddddddddd f d Zd Zd Zd ZdZd4dZdZdZdZdZdZdZdZdZdZdZdZdZdZ dZ!dZ"d Z#d!Z$d5d"Z%d5d#Z&d$Z'd%Z(d&Z) d6d'Z*d(Z+d)Z,d*Z-d+Z.d,Z/d-Z0d.Z1e2fd/Z3d0Z4e5d1Z6e5d2Z7d3Z8dS)7r$Consumer blueprint.NTc$eZdZdZdZgdZdZdS)Consumer.Blueprintr/r$) z,celery.worker.consumer.connection:Connectionz$celery.worker.consumer.mingle:Minglez$celery.worker.consumer.events:Eventsz$celery.worker.consumer.gossip:Gossipz"celery.worker.consumer.heart:Heartz&celery.worker.consumer.control:Controlz"celery.worker.consumer.tasks:Tasksz&celery.worker.consumer.consumer:Evloopz"celery.worker.consumer.agent:Agentc2||ddS)Nshutdown)send_all)selfparents r,r4zConsumer.Blueprint.shutdowns MM&* - - - - -r-N)__name__ __module__ __qualname____doc__name default_stepsr4r-r, Blueprintr2s?!!     . . . . .r-r?Fc ||_||_||_|p t|_t j|_||_||_ | |_ |j |_ |j j|_|j j|_t!dd|_t$t(j|_d|_||_t3|_|jjj|_| |_| |_| |_ d|_!tEd|_#|$||_%|j%stM|jddr%| |_'|j'|jjj(|_'nd|_'tS|d s|r tTj+n tTj,|_-t]d krd|jj_/g|_0g|_1|2|jj1d |j3 |_4|j4j5|fitm| pifi|dS) NrA)maxRmaxTrTcdSNr>r>r-r,z#Consumer.__init__..sr-is_greenFloopgeventconsumer)stepson_close)7app controller init_callbackrhostnameosgetpidpidpooltimer Strategies strategiesconnection_for_readconninfoconnection_errorschannel_errorsr_restart_statelogger isEnabledForloggingINFO _does_info _limit_orderon_task_requestseton_task_messageconfbroker_heartbeat_checkrateamqheartbeat_ratedisable_rate_limitsinitial_prefetch_countprefetch_multiplier_maximum_prefetch_restoredr task_bucketsreset_rate_limitshubgetattr amqheartbeatbroker_heartbeathasattrrasynloopsynlooprJr broker_connection_timeout_pending_operationsrMr?rN blueprintapplydict)r6rerQrRrVrOrWrPrqrsworker_optionsrkrlrmkwargss r,__init__zConsumer.__init__s+ $* 1KMM 9;;  //++4466 !%!@"m:+;;; --gl;;."uu!%!I#6 &<##6 *.'( 55     8 "wty*e<< " ,D  ($(HM$B! !D tV$$ A*-@5=DI  H , ,7;DHM 3#%  (.,](   TJJT.*>B%I%I&%I%IJJJJJr-ct|g|Ri|}|jr|j|S|j||SrG)r rq call_soonryappend)r6pargsr~s r,rzConsumer.call_soons[ Q ( ( ( ( ( ( 8 )8%%a(( (  ''***r-c|jse|jr` |jn2#t$r%}td|Yd}~nd}~wwxYw|j\dSdSdS)NzPending callback raised: %r)rqrypop Exceptionr_ exceptionr6excs r,perform_pending_operationsz#Consumer.perform_pending_operationssx I* II2D,00224444 III$$%BCHHHHHHHHI* I I I I I I Is!2 A!AA!chtt|dd}|rt|dndS)N rate_limitrA)capacity)rrrr )r6typelimits r,bucket_for_taskzConsumer.bucket_for_tasks8WT<667716@{51----D@r-cjfdjjDdS)Nc3LK|]\}}||fVdSrG)r).0ntr6s r, z-Consumer.reset_rate_limits..sK! ! -1QQ$$Q'' (! ! ! ! ! ! r-)roupdaterOtasksitemsr6s`r,rpzConsumer.reset_rate_limitss^   ! ! ! ! 59X^5I5I5K5K! ! !      r-rc|jj}|jr|sdS|jj|jz|_||S)aUpdate prefetch count after pool/shrink grow operations. Index must be the change in number of processes as a positive (increasing) or negative (decreasing) number. Note: Currently pool grow operations will end up with an offset of +1 if the initial size of the pool was 0 (e.g. :option:`--autoscale=1,0 `). N)rV num_processesrlrm_update_qos_eventually)r6indexrs r,_update_prefetch_countzConsumer._update_prefetch_countsS / * -  F I #d&> > #**5111r-cz|dkr |jjn |jjt||jzS)Nr)qosdecrement_eventuallyincrement_eventuallyabsrm)r6rs r,rzConsumer._update_qos_eventuallysA3--X2 JJ1 133 3r-cNt|||dSrG)r#re)r6requests r,_limit_move_to_poolzConsumer._limit_move_to_pools*g W%%%%%r-c |\}}n#t$rYdSwxYw||r||V|j||f|jdzdzx}|_||}|j ||j |f|dS)NTrA )priority) r IndexError can_consumercontents appendleftrd expected_timerW call_after_schedule_bucket_request)r6bucketrtokenspriholds r,rz!Consumer._schedule_bucket_request#s  "(**,,    !!&)) ((111**GV+<===+/+Y''U?3344''DHD['''       s 2 ??cD|j|d|fddS)Nregister_with_event_loopz Hub.register)r description)rzr5)r6rqs r,rz!Consumer.register_with_event_loops8  ,C6&      r-c:|j|dSrG)rzr4rs r,r4zConsumer.shutdowns %%%%%r-c:|j|dSrG)rzstoprs r,rz Consumer.stops D!!!!!r-cB|jdc}|_|r ||dSdSrG)rQ)r6callbacks r,on_readyzConsumer.on_readys7'+'94$$$   HTNNNNN  r-c r||j|j|j|j|j|j|jj|jf SrG) r task_consumerrzrqrrsrOclockrjrs r, loop_argszConsumer.loop_argss6dot'9$(D4E 68 8r-c tt||j|jt |jt ||jd|dS)a.Callback called if an error occurs while decoding a message. Simply logs the error and acknowledges the message so it doesn't enter a loop. Arguments: message (kombu.Message): The message received. exc (Exception): The exception being handled. rArN) rMESSAGE_DECODE_ERROR content_typecontent_encodingr headersr&r(ack)r6messagers r,on_decode_errorzConsumer.on_decode_errors\ ! '&(@ w ' '7GL)I)I      r-c|jr*|jjr|jj|jr|j|jD]}|r|tD]}|tvrt|=tj|j r'|j j r|j dSdSdSrG) rP semaphoreclearrWrovalues clear_pendingr"r!rVflush)r6r request_ids r,rNzConsumer.on_closes ? .t8 . O % + + - - - :  J     '..00 ' 'F '$$&&&+ ) )JX%%Z(!!! 9   IOO         r-c||j}|jr%|j|j|j|S)zEstablish the broker connection used for consuming tasks. Retries establishing the connection if the :setting:`broker_connection_retry` setting is enabled  heartbeat)rZrsrq transportrr)r6conns r,connectzConsumer.connectsI ''$2C'DD 8 O N 3 3DOTX N N N r-c^||j|SNr )ensure_connectedrOrZr6r s r,rZzConsumer.connection_for_reads0$$ H ( (9 ( = =?? ?r-c^||j|Sr)rrOconnection_for_writers r,rzConsumer.connection_for_writes3$$ H ) )I ) > >@@ @r-ctffd }d}jjjGjjj }t jtdjjjdn,jrjjj }njjj }|r d_S |jjj td_S)Nc"tddr |dkrt}|t|ddt |dz jjj}tt ||dS)Naltrin r@)whenretries max_retries) rrCONNECTION_FAILOVERr)rintrOrhbroker_connection_max_retriesrrr)rinterval next_stepr r6s r,_error_handlerz1Consumer.ensure_connected.._error_handlerstUD)) 0h!mm/ !((%hc::HqL)) HMG)III "DKKMM3 B B B B Br-Fa$The broker_connection_retry configuration setting will no longer determine whether broker connection retries are made during startup in Celery 6.0 and above. If you wish to retain the existing behavior for retrying connections on startup, you should set broker_connection_retry_on_startup to .)r) CONNECTION_RETRY_STEPrOrhrrrrrrrensure_connectionrr )r6r r"retry_disableds`` r,rzConsumer.ensure_connecteds/5J C C C C C C C 8= ; C"&!FFN M*uMQHMLquuuvv    , K%)X]%U!U%)X]%J!J   LLNNN,1D )K%% DHMG#&  ).% r-cJ|jr|jdSdSrG)event_dispatcherrrs r, _flush_eventszConsumer._flush_eventss2   *  ! ' ' ) ) ) ) ) * *r-c`|jr&|jj|jdSdSrG)rq_readyrr)rs r,on_send_event_bufferedzConsumer.on_send_event_buffereds7 8 4 HO   2 3 3 3 3 3 4 4r-c ,|j}|jjj}||vr ||}n||n|}|dn|}|j|f|||d|}||s;|||td|dSdS)Ndirect)exchange exchange_type routing_keyzStarted consuming from %s) rrOamqpqueues select_addconsuming_from add_queueconsumer) r6queuer/r0r1optionscsetr3qs r,add_task_queuezConsumer.add_task_queues!% F??u AA ( 0uuhH)6)>XX"/ !!%F+30=.9FF>EFFA""5)) 5 NN1    LLNNN ,e 4 4 4 4 4 5 5r-ctd||jjj||j|dS)NzCanceling queue %s)rrOr2r3deselectrcancel_by_queue)r6r8s r,cancel_task_queuezConsumer.cancel_task_queue3sK !5)))  %%e,,, **511111r-ct||||jdS)zAMethod called by the timer to apply a task with an ETA/countdown.N)r#rerr)r6rs r,apply_eta_taskzConsumer.apply_eta_task8s>d T""" %%'''''r-c tt||t|jt|jt|jt|jSrG)MESSAGE_REPORTr)r&r rr delivery_inforr6r(rs r,_message_reportzConsumer._message_report>s`$$Yw%=%=%.w/C%D%D%.w/G%H%H%.w/D%E%E%.w%?%? AA Ar-ctt||||t|jt j||ddS)Nsenderrr) rUNKNOWN_FORMATrGreject_log_errorr_r\r task_rejectedsendrFs r,on_unknown_messagezConsumer.on_unknown_messageEs_ ^T11$@@AAA  )?@@@""$T"JJJJJr-c tt|t|||j|jd |jd|jd}}|jd}n)#t $r|j}|d|d}}d}YnwxYwt|d||j d|j dd}| t|j |j j|t!|| |jr!|jd |d |d  t&j|||||dS)NTridrroot_idcorrelation_idreply_to)r<chordrRrSrTerrbacks)rz task-failedzNotRegistered())uuidr)rJrrr<rQ)rUNKNOWN_TASK_ERRORr&rrEgetKeyErrorpayloadr propertiesrLr_r\rObackendmark_as_failurerr(rNr task_unknown) r6r(rrid_r<rRr\rs r,on_unknown_taskzConsumer.on_unknown_taskJs  &&o#       -wv/FCo)))44GG   oG wvCGGG T7"-112BCC'++J77       )?@@@ (( t$$g )        ! & &C44444 '    !!c "     s4A))#BBctt|t||d|t|jt j|||dS)NTrrI) rINVALID_TASK_ERRORr&rLr_r\rrMrN)r6r(rrs r,on_invalid_taskzConsumer.on_invalid_taskksj  #y$'?'?      )?@@@""$S"IIIIIr-c|jj}|jjD]K\}}||j||j|<t ||||j|j|_LdS)N)rO) rOloaderrrstart_strategyrYrrR __trace__)r6rgr<rs r,update_strategieszConsumer.update_strategiesqs}(...00 8 8JD$$($7$7$$G$GDOD !)$fdm.2h888DNN 8 8r-cjjjjjjfd}|S)Ncd} |jd}n#t$r d|cYSt$ry |}n/#t$r"} ||cYd}~cYSd}~wwxYw |d|}}n%#ttf$r ||cYcYSwxYwYnwxYw |}  |jf j} |jf j} j s] j dkrR j j krB| j j| j j|||||dS#ttf$r} |||cYd}~Sd}~wt $r } ||cYd}~Sd}~wwxYw#t$r} d||cYd}~Sd}~wwxYw)Nr)on_errorr)r TypeErrorr[decoderr ack_log_error0_restore_prefetch_count_after_connection_restartrLrnr_new_prefetch_countrthenrrr )rr\type_rstrategyack_log_error_promisereject_log_error_promiser callbacksrerOrbrr6rYs r,on_task_receivedz6Consumer.create_task_handler..on_task_receiveds G @/ 9 9 9))$88888 @ @ @>%nn..GG >>>//==========>@%,V_g7EE!8,@@@--gw???????@E  @$ >%e,>,3G! .0!%!V---) 07w! 13!%!V000,!;v .22 48OOO-2243h<@>>//========>C ; ; ;&tWc:::::::: ;s B- B-A  B- A6A1)A6*B-1A66B-: BB-B'"B-&B''B-,B-1F4:B%E!!F12 F>F1 F1F,&F1,F14 G> G GG)rYrOrbrergr) r6rryrrxrerOrbrYs `` @@@@@@r,create_task_handlerzConsumer.create_task_handlerxs~_ !4..( N 5 >5 >5 >5 >5 >5 >5 >5 >5 >5 >5 >5 >n r-c|jj5|jr ddddSt|j|j}|x|j_|_|j|jj|j}||jk|_|dur+|jdur"t d|jddddS#1swxYwYdS)NFTzcResuming normal operations following a restart. Prefetch count has been restored to the maximum of ) r_mutexrnminrrrvaluerlrfr_r)r6rrnew_prefetch_countalready_restoreds r,rqz9Consumer._restore_prefetch_count_after_connection_restartsG X_  .         "%T%V!W!W ;M MDHNT8 HLL ( ( (#> .@DD[.[D +5((T-LPT-T-T dJNJadd                  s CBCC C c*|jj|jzSrG)rVrrmrs r,rzConsumer.max_prefetch_countsy&)AAAr-c*|jj|jzSrG)rr~rmrs r,rrzConsumer._new_prefetch_countsx~ 888r-c^d||jS)z``repr(self)``.z%)r6r)r)rz human_staters r,__repr__zConsumer.__repr__s26==T^7799>   r-)rrG)NNN)9r8r9r:r;r|rXrQrVrWrrrr?rrrrrrprrrrrrrrrrrr4rrrrrNrrZrrr)r,r<r@rBrGrOrbrerjrrzrqpropertyrrrrr>r-r,r$r$sXJM D EM $.....I'...( $d$T $%()q :K:K:K:KxIIIAAA   2222&333 &&&0555555 ),),),V000 (((<   &&&""" 888  $   ????@@@@111f***444BF#'5555,222 ((( AAAKKK    BJJJ 888+2? ? ? ? B$BBXB99X9     r-r$c&eZdZdZdZdZdZdZdS)r%zHEvent loop service. Note: This is always started last. z event loopTch|||j|dSrG) patch_allrJrr6cs r,rz Evloop.starts1 q r-c6t|j_dSrG)rrr|rs r,rzEvloop.patch_alls {{ r-N)r8r9r:r;labellastrrr>r-r,r%r%sH E D#####r-r%)Wr;rrarSr collectionsrtimerbilliard.commonrbilliard.exceptionsrkombu.asynchronous.semaphorerkombu.exceptionsrr kombu.utils.compatr kombu.utils.encodingr kombu.utils.limitsr viner rceleryrrcelery.app.tracercelery.exceptionsrrrrrcelery.utils.functionalrcelery.utils.logrcelery.utils.nodenamesrcelery.utils.objectsrcelery.utils.textrcelery.utils.timerr celery.workerrcelery.worker.staterr r!r"r#__all__CLOSE TERMINATErr8r_debugrwarningrcriticalrrrr$rrrKrYrdrrDrrr&r$ StartStopStepr%r>r-r,rs6   ######))))))333333222222;;;;;;;;222222************""""""""%%%%%%%%))))))00000000000000((((((''''''......&&&&&&&&&&&&44444444kkkkkkkkkkkkkk -   )$ H  "(, V^"(,"AtT5$  ,   73 ***M  M  M  M  M  M  M  M  `#####Y $#####r-