U bhM @s dZddlZddlZddlmZmZmZddlmZddl m Z ddl m Z ddl mZddl mZdd lmZdd lmZdd lmZmZdd lmZd dlmZd dlmZdZdZ ee!Z"edddddddddgZ#ddZ$ddZ%Gdd d eZ&d!d"Z'd#d$Z(e(d%d&Z)e(d'd(d)efgd*dd,d-Z*d.d/Z+e(d0d1d2d3d4Z,ej-j.fd5d6Z/ej0j1ej2j1fd7d8Z3e'd9d1d2dd:d;Z4e'dd?Z5dd@dAZ6e'd9dBe7fgdCdDdEdFZ8e'dGe7fdHe7fgdIdJdKdHZ9e'dGe7fdLe:fdMe:fgdNdJddOdPZ;e(dQdRZe'dWdXZ?e'dYdZZ@e(d+d[dd\d]ZAe(d^d_d`daZBe(dbdcZCe(dddedfdgZDdhdiZEe(djdedkdlZFe(dmdeddndoZGe(dpdedqdrZHe(dsdtdudvddwdxZIe(dyde7fdzeJfd{eJfgd|d}dddZKe(ddZLe(deJfgddJdddZMe'deJfgddJdddZNe'deJfgddJdddZOe'dddZPe'deJfdeJfgddJdddZQe'dddZRe'de7fde7fde7fde7fgddJdddZSe'de7fgddJddZTe(ddZUdS)z.Worker remote control command implementations.N)UserDict defaultdict namedtuple) TERM_SIGNAME) safe_repr)WorkerShutdown)EX_OK)signals) maybe_list) get_logger)jsonify strtobool)ratestate)Request)Panel)exchange routing_key rate_limitcontroller_info_taliastypevisibledefault_timeouthelp signatureargsvariadiccCsd|iS)Nokvaluer!r!9/tmp/pip-unpacked-wheel-kcem4wq5/celery/worker/control.pyr sr cCsd|iS)Nerrorr!r"r!r!r$nok#sr&c @s2eZdZdZiZiZeddZed dd ZdS) rz+Global registry of remote control commands.cOs |r|jf||S|jf|SN) _register)clsrkwargsr!r!r$register-szPanel.registerNcontrolT?c s" f dd} | S)Nc s^p|j}p$|jpddd}|j|<t |j|<rZ|j<|S)N r)__name____doc__stripsplitdatarmeta)ZfunZ control_nameZ_help rrr)rrnamerrrrr!r$_inner8s     zPanel._register.._innerr!) r)r7rrrrrrrrr8r!r6r$r(3s zPanel._register) NNr,Tr-NNNN) r0 __module__ __qualname__r1r4r5 classmethodr+r(r!r!r!r$r's rcKstjfddi|S)Nrr,rr+r*r!r!r$control_commandEsr>cKstjfddi|S)Nrinspectr<r=r!r!r$inspect_commandIsr@cCst|jS)z6Information about Celery installation for bug reports.)r appZ bugreportrr!r!r$reportOsrBZ dump_confz[include_defaults=False] with_defaults)rrrFcKst|jjj|dttdS)zList configuration.)rC)Z keyfilterZunknown_type_filter)r rAconftable_wanted_config_keyr)rrCr*r!r!r$rDUsrDcCst|to|d S)N__) isinstancestr startswith)keyr!r!r$rFasrFidsz[id1 [id2 [... [idN]]]])rrcKsddtt|DS)z!Query for task information by id.cSs i|]}|jt||fqSr!)id_state_of_taskinfo).0reqr!r!r$ mszquery_task..)_find_requests_by_idr )rrLr*r!r!r$ query_taskgs rTc cs2|D](}z||VWqtk r*YqXqdSr')KeyError)rL get_requesttask_idr!r!r$rSss rScCs||r dS||rdSdS)Nactivereservedreadyr!)requestZ is_active is_reservedr!r!r$rN|s rNrWcKsNtt|p gd}}t||||f|}t|tr>d|kr>|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 _revokerHdictr )rrW terminatesignalr*task_idsr!r!r$revokes rcheadersz/[key1=value1 [key2=value2 [... [keyN=valueN]]]]cKs.t|p t}t|tr&dd|D}|D]2\}}ttj |pFgtt|}|tj|<q.|svt d|dSttj } t t } | D]z} t| dr| jr|D]\\} } | | jkrt| } t| j| }t |t | @}|r| | || j|jj|dqq| 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)rPhr!r!r$rRs z-revoke_by_stamped_headers..zheaders z' flagged as revoked, but not terminatedstampsraz were not terminatedz revoked)_signalssignumrrHlistitemsr worker_stateZrevoked_stampsgetr active_requestsrr]hasattrrgupdater`consumerpool)rrdr`rar*rjheaderrgZupdated_stampsroZ#terminated_scheme_to_stamps_mappingrQZexpected_header_keyZexpected_header_valueZ actual_headerZmatching_stamps_for_requestr!r!r$revoke_by_stamped_headerss.     ruc Kst|}t}tj||rt|p(t}t|D]L}|j |kr4| |j t d|j ||j |jj|dt||kr4qq4|stdStdd|Sd|} t d| |S)NzTerminating %s (%s)rhzterminate: tasks unknownz terminate: {}z, zTasks flagged as revoked: %s)lenr]rmrevokedrqrirjrrSrMaddloggerrOr`rrrsr formatjoin) rrbr`rar*sizeZ terminatedrjr[Zidstrr!r!r$r^s$       r^raz [id1 [id2 [... [idN]]]])rrrcKst||d|dS)z+Terminate task by task id (or list of ids).T)r`ra)rc)rrarWr*r!r!r$r`sr` task_namerz0 )rrc Ksz t|Wn4tk r@}ztd|WYSd}~XYnXz||jj|_Wn,tk rtjd|ddtdYSX|j |st 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&rAtasksrrUryr%rrZreset_rate_limitsrOr )rr}rr*excr!r!r$rs* $  softhardz# [hard_secs]cKsbz|jj|}Wn,tk r<tjd|ddtdYSX||_||_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 %sTr~rz5New time limits for tasks of type %s: soft=%s hard=%sztime limits set successfully) rArrUryr%r&Zsoft_time_limit time_limitrOr )rr}rrr*taskr!r!r$r s rcKsd|jjjiS)z Get current logical clock value.clock)rArr#rr*r!r!r$r>srcKs|jjr|jj|||dS)zHold election. Arguments: id (str): Unique election id. topic (str): Election topic. action (str): Action to take for elected actor. N)rrZgossipelection)rrMZtopicactionr*r!r!r$rDs rcCs>|jj}|jr6d|jkr6|jdtdtdStdS)z+Tell worker(s) to send task-related events.rz)Events of group {task} enabled by remote.ztask events enabledztask events already enabled)rrevent_dispatchergroupsrxryrOr r dispatcherr!r!r$ enable_eventsQs   rcCs8|jj}d|jkr0|jdtdtdStdS)z3Tell worker(s) to stop sending task-related events.rz*Events of group {task} disabled by remote.ztask events disabledztask events already disabled)rrrrdiscardryrOr 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)rydebugrrrsendrmZ SOFTWARE_INFOrr!r!r$ heartbeatgs r)rcKsJ||jkrFtd||r&tj|tjtjj|jj dSdS)zRequest mingle sync-data.z sync with %s)rwrN) hostnameryrOrmrwrqpurge_datarArZforward)rZ from_noderwr*r!r!r$helloqs     rg?)rcKstdS)zPing worker(s).Zpong)r rr!r!r$pingsrcKs |jjS)z&Request worker statistics/information.)rr controllerstatsrr!r!r$rsrZ dump_schedule)rcKstt|jjS)z0List of currently scheduled ETA/countdown tasks.)rk_iter_schedule_requestsrrtimerrr!r!r$ scheduledsrc csn|jjD]`}z|jjd}Wnttfk r8YqYqXt|tr|jrT|j nd|j | dVqdS)Nr)etapriorityr[) schedulequeueentryr IndexError TypeErrorrHrr isoformatrrO)rZwaitingZarg0r!r!r$rs   rZ dump_reservedcKs.|tj|tj}|s gSdd|DS)zAList of currently reserved tasks, not including scheduled/active.cSsg|] }|qSr!rOrPr[r!r!r$ szreserved..)tsetrmreserved_requestsro)rr*Zreserved_tasksr!r!r$rYs  rYZ dump_activec sfdd|tjDS)z'List of tasks currently being executed.csg|]}|jdqS)saferrrr!r$rszactive..)rrmro)rrr*r!rr$rXs  rXZ dump_revokedcKs ttjS)zList of revoked task-ids.)rkrmrwrr!r!r$rwsrwZ dump_tasks taskinfoitemsz[attr1 [attr2 [... [attrN]]]])rrrc sJ|jjpt|rn ddD}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)rJrPrr!r!r$ s zregistered..csBfddD}|r._extract_info..cSsg|]}d|qS)re)r{)rPfr!r!r$rsz5registered.._extract_info..z{} [{}] )rlrzr7r{)rfieldsrO)rrr$ _extract_infos z!registered.._extract_infocsg|]}|qSr!r!r)rregr!r$rszregistered..)rArDEFAULT_TASK_INFO_ITEMSsorted)rrbuiltinsr*rr!)rrrr$ registereds  rgN@num max_depthz.[object_type=Request] [num=200 [max_depth=10]])rrr rc sz ddl}Wntk r(tdYnXtd|tjddddF}||d||j|fd d |jd d |jiW5QRSQRXdS) 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 %rZcobjgz.pngF)prefixsuffixdeletecs|kSr'r!)vZobjectsr!r$zobjgraph..)rZ highlightfilenamer) objgraph ImportErrorryrOtempfileNamedTemporaryFileZby_typeZ show_backrefsr7)rrrrZ _objgraphfhr!rr$rs$    rcKsddlm}|S)z Sample current RSS memory usage.r) sample_mem)Zcelery.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)Z celery.utilsrioStringIOmemdumpgetvalue)rrr*routr!r!r$rs  rnz[N=1]cKs4|jjjrtdS|jj||j|tdS)z!Grow pool by n processes/threads.zJpool_grow is not supported with autoscale. Adjust autoscale range instead.zpool will grow)rrr autoscalerr&rsZgrow_update_prefetch_countr rrr*r!r!r$ pool_grows   rcKs6|jjjrtdS|jj||j| tdS)z#Shrink pool by n processes/threads.zLpool_shrink is not supported with autoscale. Adjust autoscale range instead.zpool will shrink)rrrrr&rsshrinkrr rr!r!r$ pool_shrinks  rcKs2|jjjr&|jjj|||dtdStddS)zRestart execution pool.)reloaderzreload startedzPool restarts not enabledN)rArDZworker_pool_restartsrrrreloadr r)rmodulesrrr*r!r!r$ pool_restart-s rmaxminz [max [min]]cCs>|jjj}|r2|||\}}td|d|StddS)zModify autoscale settings.zautoscale now max=z min=zAutoscale not enabledN)rrrrrqr r)rrrrZmax_Zmin_r!r!r$ autoscale7s  rGot shutdown from remotecKst|ttdS)zShutdown worker(s).N)rywarningrr)rmsgr*r!r!r$shutdownDs rrr exchange_typerz' [exchange [type [routing_key]]]cKs.|jj|jj|||pd|f|td|S)z2Tell worker(s) to consume from task queue by name.directz add consumer )rr call_soonZadd_task_queuer )rrrrroptionsr!r!r$ add_consumerMs rzcKs |j|jj|td|S)z9Tell worker(s) to stop consuming from task queue by name.zno longer consuming from )rrrZcancel_task_queuer )rr_r!r!r$cancel_consumer_s rcCs |jjrdd|jjjDSgS)z:List the task queues a worker is currently consuming from.cSsg|]}t|jddqS)T)Zrecurse)r_Zas_dict)rPrr!r!r$rosz!active_queues..)rrZ task_consumerZqueuesrr!r!r$ active_queuesks r)F)FN)FN)FN)NNN)N)N)F)NF)rrr)r)r)r)NFN)NN)r)NNN)Vr1rr collectionsrrrZbilliard.commonrZkombu.utils.encodingrZcelery.exceptionsrZcelery.platformsrr riZcelery.utils.functionalr Zcelery.utils.logr Zcelery.utils.serializationr r Zcelery.utils.timerr.rrmr[r__all__rr0ryrr r&rr>r@rBrDrFrTrequests __getitem__rSro __contains__rrNrcrur^rIr`rfloatrrrrrrrrrrrrYrXrwrintrrrrrrrrrrrr!r!r!r$s>                 6   $