o hM@s dZddlZddlZddlmZmZmZddlmZddl m Z ddl m Z ddl mZddlmZdd lmZdd lmZmZdd lmZd d lmZd dlmZdZdZee Z!edgdZ"ddZ#ddZ$GdddeZ%ddZ&ddZ'e'ddZ(e'dd d!efgd"dd$d%Z)d&d'Z*e'd(d)d*d+d,Z+ej,j-fd-d.Z.ej/j0ej1j0fd/d0Z2e&d1d)d*dd2d3Z3e&d4d5d*dd6d7Z4dd8d9Z5e&d1d:e6fgd;d<d=d>Z7e&d?e6fd@e6fgdAdBdCd@Z8e&d?e6fdDe9fdEe9fgdFdBddGdHZ:e'dIdJZ;e&ddKdLZe&dQdRZ?e'd#dSddTdUZ@e'dVdWdXdYZAe'dZd[ZBe'd\d]d^d_ZCd`daZDe'dbd]dcddZEe'ded]ddfdgZFe'dhd]didjZGe'dkdldmdnddodpZHe'dqdre6fdseIfdteIfgdudvddzd{ZJe'd|d}ZKe'd~eIfgddBdddZLe&deIfgddBdddZMe&deIfgddBdddZNe&dddZOe&deIfdeIfgddBdddZPe&dddZQe&de6fde6fde6fde6fgddB  dddZRe&de6fgddBddZSe'ddZTdS)z.Worker remote control command implementations.N)UserDict defaultdict namedtuple) TERM_SIGNAME) safe_repr)WorkerShutdown)signals) maybe_list) get_logger)jsonify strtobool)ratestate)Request)Panel)exchange routing_key rate_limitcontroller_info_t)aliastypevisibledefault_timeouthelp signatureargsvariadiccCd|iS)Nokvaluer!r!G/var/www/html/env/lib/python3.10/site-packages/celery/worker/control.pyr r cCr)Nerrorr!r"r!r!r$nok"r%r'c@s8eZdZdZiZiZeddZe   d dd ZdS) rz+Global registry of remote control commands.cOs(|r |jdi||S|jdi|S)Nr!) _register)clsrkwargsr!r!r$register,szPanel.registerNcontrolT?c s" f dd} | S)Nc s^p|j}p|jp ddd}|j|<t |j|<r-|j<|S)N r)__name____doc__stripsplitdatarmeta)fun control_name_help rrr)rrnamerrrrr!r$_inner7s    zPanel._register.._innerr!) r)r:rrrrrrrrr;r!r9r$r(2s zPanel._register) NNr,Tr-NNNN) r0 __module__ __qualname__r1r4r5 classmethodr+r(r!r!r!r$r&s rcKtjdddi|S)Nrr,r!rr+r*r!r!r$control_commandDrBcKr?)Nrinspectr!r@rAr!r!r$inspect_commandHrCrEcCst|jS)z6Information about Celery installation for bug reports.)r app bugreportrr!r!r$reportNrH dump_confz[include_defaults=False] with_defaults)rrrFcKst|jjj|dttdS)zList configuration.)rK) keyfilterunknown_type_filter)r rFconftable_wanted_config_keyr)rrKr*r!r!r$rNTsrNcCst|to |d S)N__) isinstancestr startswith)keyr!r!r$rP`srPidsz[id1 [id2 [... [idN]]]])rrcKsddtt|DS)z!Query for task information by id.cSs i|] }|jt||fqSr!)id_state_of_taskinfo).0reqr!r!r$ lszquery_task..)_find_requests_by_idr )rrVr*r!r!r$ query_taskfs r^c cs0|D]}z||VWqtyYqwdSN)KeyError)rV get_requesttask_idr!r!r$r]rs r]cCs||rdS||r dSdS)Nactivereservedreadyr!)request is_active is_reservedr!r!r$rX{s rXrbcKsRtt|pgd}}t||||fi|}t|tr!d|vr!|Std|dS)zRevoke task by task id (or list of ids). Keyword Arguments: terminate (bool): Also terminate the process if the task is active. signal (str): Name of signal to use for terminate (e.g., ``KILL``). Nr ztasks z flagged as revoked)setr _revokerRdictr )rrb terminatesignalr*task_idsr!r!r$revokes roheadersz/[key1=value1 [key2=value2 [... [keyN=valueN]]]]cKs,t|pt}t|trdd|D}|D]\}}ttj |p#gtt|}|tj|<q|s;t d|dSttj } t t } | D]=} t| dr| jr|D].\} } | | jvrt| } t| j| }t |t | @}|r| | || j|jj|dqTqF| st d|dSt d| dS) aRevoke task by header (or list of headers). Keyword Arguments: headers(dictionary): Dictionary that contains stamping scheme name as keys and stamps as values. If headers is a list, it will be converted to a dictionary. terminate (bool): Also terminate the process if the task is active. signal (str): Name of signal to use for terminate (e.g., ``KILL``). Sample headers input: {'mtask_id': [id1, id2, id3]} cSs&i|]}|dd|ddqS)=rr)r3)rZhr!r!r$r\s&z-revoke_by_stamped_headers..zheaders z' flagged as revoked, but not terminatedstampsrmz were not terminatedz revoked)_signalssignumrrRlistitemsr worker_staterevoked_stampsgetr active_requestsrrihasattrrsupdaterlconsumerpool)rrprlrmr*rvheaderrsupdated_stampsr|#terminated_scheme_to_stamps_mappingr[expected_header_keyexpected_header_value actual_headermatching_stamps_for_requestr!r!r$revoke_by_stamped_headerss0     rc Kst|}t}tj||rQt|pt}t|D]&}|j |vr@| |j t d|j ||j |jj|dt||kr@nq|sGtdStdd|Sd|} t d| |S)NzTerminating %s (%s)rtzterminate: tasks unknownz terminate: {}z, zTasks flagged as revoked: %s)lenriryrevokedr~rurvrr]rWaddloggerrYrlrrr formatjoin) rrnrlrmr*size terminatedrvrfidstrr!r!r$rjs&       rjrmz [id1 [id2 [... [idN]]]])rrrcKst||d|dS)z+Terminate task by task id (or list of ids).T)rlrm)ro)rrmrbr*r!r!r$rlsrl task_namerz0 )rrc Kszt|Wnty}z td|WYd}~Sd}~wwz ||jj|_Wnty>tjd|ddtdYSw|j |sPt d|t dSt d ||t d S) zTell worker(s) to modify the rate limit for a task by type. See Also: :attr:`celery.app.task.Task.rate_limit`. Arguments: task_name (str): Type of task to set rate limit for. rate_limit (int, str): New rate limit. zInvalid rate limit string: Nz&Rate limit attempt for unknown task %sTexc_info unknown taskz)Rate limits disabled for tasks of type %sz rate limit disabled successfullyz(New rate limit for tasks of type %s: %s.znew rate limit set successfully) r ValueErrorr'rFtasksrr`rr&rreset_rate_limitsrYr )rrrr*excr!r!r$rs,     softhardz# [hard_secs]cKs`z|jj|}Wntytjd|ddtdYSw||_||_td|||t dS)zTell worker(s) to modify the time limit for task by type. Arguments: task_name (str): Name of task to change. hard (float): Hard time limit. soft (float): Soft time limit. z-Change time limit attempt for unknown task %sTrrz5New time limits for tasks of type %s: soft=%s hard=%sztime limits set successfully) rFrr`rr&r'soft_time_limit time_limitrYr )rrrrr*taskr!r!r$rs   rcKsd|jjjiS)z Get current logical clock value.clock)rFrr#rr*r!r!r$r=rIrcKs"|jjr|jj|||dSdS)zHold election. Arguments: id (str): Unique election id. topic (str): Election topic. action (str): Action to take for elected actor. N)rgossipelection)rrWtopicactionr*r!r!r$rCs rcCs>|jj}|jrd|jvr|jdtdtdStdS)z+Tell worker(s) to send task-related events.rz)Events of group {task} enabled by remote.ztask events enabledztask events already enabled)revent_dispatchergroupsrrrYr r dispatcherr!r!r$ enable_eventsPs   rcCs8|jj}d|jvr|jdtdtdStdS)z3Tell worker(s) to stop sending task-related events.rz*Events of group {task} disabled by remote.ztask events disabledztask events already disabled)rrrdiscardrrYr rr!r!r$disable_events[s    rcCs,td|jj}|jddditjdS)z3Tell worker(s) to send event heartbeat immediately.zHeartbeat requested by remote.worker-heartbeatfreqN)r)rdebugrrsendry SOFTWARE_INFOrr!r!r$ heartbeatfs r)rcKsJ||jkr#td||rtj|tjtjj|jj dSdS)zRequest mingle sync-data.z sync with %s)rrN) hostnamerrYryrr~purge_datarFrforward)r from_noderr*r!r!r$hellops     rg?)rcKstdS)zPing worker(s).pong)r rr!r!r$pingsrcKs |jjS)z&Request worker statistics/information.)r controllerstatsrr!r!r$rs r dump_schedule)rcKstt|jjS)z0List of currently scheduled ETA/countdown tasks.)rw_iter_schedule_requestsrtimerrr!r!r$ scheduledsrc csj|jjD]-}z|jjd}Wn ttfyYqwt|tr2|jr(|j nd|j | dVqdS)Nr)etapriorityrf) schedulequeueentryr IndexError TypeErrorrRrr isoformatrrY)rwaitingarg0r!r!r$rs  r dump_reservedcKs.|tj|tj}|sgSdd|DS)zAList of currently reserved tasks, not including scheduled/active.cSsg|]}|qSr!rYrZrfr!r!r$ szreserved..)tsetryreserved_requestsr|)rr*reserved_tasksr!r!r$rds  rd dump_activec sfdd|tjDS)z'List of tasks currently being executed.csg|]}|jdqS)saferrrr!r$rszactive..)rryr|)rrr*r!rr$rcs  rc dump_revokedcKs ttjS)zList of revoked task-ids.)rwryrrr!r!r$rs r dump_tasks taskinfoitemsz[attr1 [attr2 [... [attrN]]]])rrrc sJ|jjpt|r nddD}fddfddt|DS)zList of registered tasks. Arguments: taskinfoitems (Sequence[str]): List of task attributes to include. Defaults to ``exchange,routing_key,rate_limit``. builtins (bool): Also include built-in tasks. css|] }|ds|VqdS)zcelery.N)rTrZrr!r!r$ s   zregistered..csBfddD}|rdd|D}djd|SjS)Ncs.i|]}t|ddur|tt|dqSr_)getattrrS)rZfieldrr!r$r\s z5registered.._extract_info..cSsg|]}d|qS)rq)r)rZfr!r!r$rsz5registered.._extract_info..z{} [{}] )rxrr:r)rfieldsrY)rrr$ _extract_infos z!registered.._extract_infocsg|]}|qSr!r!r)rregr!r$rszregistered..)rFrDEFAULT_TASK_INFO_ITEMSsorted)rrbuiltinsr*rr!)rrrr$ registereds   rgN@rnum max_depthz.[object_type=Request] [num=200 [max_depth=10]])rrr rcszddl}Wn tytdwtd|tjdddd$}||d||j|fd d |jd d |jiWdS1sGwYdS) aCreate graph of uncollected objects (memory-leak debugging). Arguments: num (int): Max number of objects to graph. max_depth (int): Traverse at most n levels deep. type (str): Name of object to graph. Default is ``"Request"``. rNzRequires the objgraph libraryzDumping graph for type %rcobjgz.pngF)prefixsuffixdeletecs|vSr_r!)vobjectsr!r$szobjgraph..)r highlightfilenamer) objgraph ImportErrorrrYtempfileNamedTemporaryFileby_type show_backrefsr:)rrrr _objgraphfhr!rr$rs$    $rcKsddlm}|S)z Sample current RSS memory usage.r) sample_mem)celery.utils.debugr)rr*rr!r!r$ memsamples rsamplesz[n_samples=10]cKs(ddlm}t}|j|d|S)z/Dump statistics of previous memsample requests.r)r)file) celery.utilsrioStringIOmemdumpgetvalue)rrr*routr!r!r$rs  rnz[N=1]cKs4|jjjr tdS|jj||j|tdS)z!Grow pool by n processes/threads.zJpool_grow is not supported with autoscale. Adjust autoscale range instead.zpool will grow)rr autoscalerr'rgrow_update_prefetch_countr rrr*r!r!r$ pool_grows  r cKs6|jjjr tdS|jj||j| tdS)z#Shrink pool by n processes/threads.zLpool_shrink is not supported with autoscale. Adjust autoscale range instead.zpool will shrink)rrrr'rshrinkr r r r!r!r$ pool_shrinks rcKs.|jjjr|jjj|||dtdStd)zRestart execution pool.)reloaderzreload startedzPool restarts not enabled)rFrNworker_pool_restartsrrreloadr r)rmodulesrrr*r!r!r$ pool_restart,s rmaxminz [max [min]]cCs:|jjj}|r|||\}}td|d|Std)zModify autoscale settings.zautoscale now max=z min=zAutoscale not enabled)rrrr~r r)rrrrmax_min_r!r!r$ autoscale6s rGot shutdown from remotecKst|t|)zShutdown worker(s).)rwarningr)rmsgr*r!r!r$shutdownCs rrr exchange_typerz' [exchange [type [routing_key]]]cKs2|jj|jj|||p d|fi|td|S)z2Tell worker(s) to consume from task queue by name.directz add consumer )r call_soonadd_task_queuer )rrrrroptionsr!r!r$ add_consumerLs  r"zcKs |j|jj|td|S)z9Tell worker(s) to stop consuming from task queue by name.zno longer consuming from )rrcancel_task_queuer )rr_r!r!r$cancel_consumer^sr%cCs |jjrdd|jjjDSgS)z:List the task queues a worker is currently consuming from.cSsg|] }t|jddqS)T)recurse)rkas_dict)rZrr!r!r$rnsz!active_queues..)r task_consumerqueuesrr!r!r$ active_queuesjs r*)F)FN)NNNr_)NF)rrr)r)r)NFN)NN)r)Ur1rr collectionsrrrbilliard.commonrkombu.utils.encodingrcelery.exceptionsrcelery.platformsrrucelery.utils.functionalr celery.utils.logr celery.utils.serializationr r celery.utils.timer r.rryrfr__all__rr0rrr r'rrBrErHrNrPr^requests __getitem__r]r| __contains__rrXrorrjrSrlrfloatrrrrrrrrrrrrdrcrrintrrrr rrrrr"r%r*r!r!r!r$s,                   6  $