File: //usr/local/lib/python3.9/site-packages/celery/worker/__pycache__/worker.cpython-39.pyc
a
X>h�= � @ s@ d Z ddlZddlZddlmZmZ ddlmZ ddlmZ ddl m
Z
ddl m Z ddl m
Z dd l mZ dd
lmZmZ dd lmZmZmZ dd lmZmZ dd
lmZ ddl m Z ddl m Z ddl m!Z!m"Z" ddl#m$Z$ ddl%m&Z& ddl'm(Z( z ddl)Z)W n e*�y dZ)Y n0 dZ+dZ,dZ-dZ.G dd� d�Z/dS )a� WorkController can be used to instantiate in-process workers.
The command-line interface for the worker is in :mod:`celery.bin.worker`,
while the worker program is in :mod:`celery.apps.worker`.
The worker program is responsible for adding signal handlers,
setting up logging, etc. This is a bare-bones worker without
global side-effects (i.e., except for the global state stored in
:mod:`celery.worker.state`).
The worker consists of several components, all managed by bootsteps
(mod:`celery.bootsteps`).
� N)�datetime�timezone)�sleep)� cpu_count)�detect_environment)� bootsteps)� concurrency)�signals)�RUN� TERMINATE)�ImproperlyConfigured�TaskRevokedError�WorkerTerminate)�
EX_FAILURE�create_pidlock)�reload_from_cwd)�mlevel)�
worker_logger)�default_nodename�
worker_direct)� str_to_list)�default_socket_timeout� ��state)�WorkControllerg @z�
Trying to select queue subset of {0!r}, but queue {1} isn't
defined in the `task_queues` setting.
If you want to automatically declare unknown queues you can
enable the `task_create_missing_queues` setting.
ze
Trying to deselect queue subset of {0!r}, but queue {1} isn't
defined in the `task_queues` setting.
c @ sZ e Zd ZdZdZdZdZdZdZdZ G dd� de
j �Z dJdd�Z dKdd�Z
d d
� Zd d � Zd
d� Zdd� Zdd� Zdd� Zdd� ZdLdd�Zdd� Zdd � Zd d � Zd d � Zd!d"� Zd#d$� Zd%d&� Z d'd(� Z d)d*� Z dMd,d-�Z dNd.d/�Z dOd1d2�Z!dPd3d4�Z"dQd5d6�Z#dRd7d8�Z$d9d:� Z%d;d<� Z&d=d>� Z'd?d@� Z(dAdB� Z)e*dCdD� �Z+dSdFdG�Z,dHdI� Z-dS )Tr zUnmanaged worker instance.Nc @ s e Zd ZdZdZh d�ZdS )zWorkController.BlueprintzWorker bootstep blueprint.ZWorker> z celery.worker.components:Hubz!celery.worker.components:Consumerz celery.worker.components:Beatz celery.worker.components:Poolz celery.worker.components:Timerz celery.worker.components:StateDBz'celery.worker.autoscale:WorkerComponentN)�__name__�
__module__� __qualname__�__doc__�nameZ
default_steps� r! r! �>/usr/local/lib/python3.9/site-packages/celery/worker/worker.py� BlueprintL s r# c K s� |p| j | _ t|�| _t�tj�| _| j j� � | j
f i |�� | j f i |�� | j f i |�� | j
f i | jf i |���� d S �N)�appr �hostnamer �nowr �utc� startup_time�loaderZ init_worker�on_before_init�setup_defaults�
on_after_init�setup_instance� prepare_args)�selfr% r& �kwargsr! r! r" �__init__Z s
zWorkController.__init__c K s� || _ | �||� | �t|�� | jsLz t� | _W n tyJ d| _Y n0 t| j�| _|p`| j | _
| j � � | _
|d u r�| �� n|| _|| _tjj| d� t�| j�| _g | _| �� | j| j jd | j| j| j d�| _ | j j | fi |�� d S )N� �ZsenderZworker)�steps�on_start�on_close�
on_stopped) �pidfile� setup_queues�setup_includesr r r �NotImplementedErrorr �loglevel�on_consumer_ready�ready_callbackr% Zconnection_for_read� _conninfo�should_use_eventloop�
use_eventloop�optionsr Z worker_init�send� _concurrencyZget_implementation�pool_clsr5 �on_init_blueprintr# r6 r7 r8 � blueprint�apply)r0 �queuesr? r9 �includerB Zexclude_queuesr1 r! r! r" r. e s4 �
�z WorkController.setup_instancec C s d S r$ r! �r0 r! r! r" rG � s z WorkController.on_init_blueprintc K s d S r$ r! �r0 r1 r! r! r" r+ � s z WorkController.on_before_initc K s d S r$ r! rM r! r! r" r- � s z WorkController.on_after_initc C s | j rt| j �| _d S r$ )r9 r �pidlockrL r! r! r" r6 � s zWorkController.on_startc C s d S r$ r! )r0 �consumerr! r! r" r> � s z WorkController.on_consumer_readyc C s | j j�� d S r$ )r% r* Zshutdown_workerrL r! r! r" r7 � s zWorkController.on_closec C s( | j �� | j�� | jr$| j�� d S r$ )�timer�stoprO �shutdownrN �releaserL r! r! r" r8 � s
zWorkController.on_stoppedc
C s� t |�}t |�}z| jjj�|� W n8 ty\ } z tt�� � ||���W Y d }~n
d }~0 0 z| jjj�
|� W n8 ty� } z tt �� � ||���W Y d }~n
d }~0 0 | jj j
r�| jjj�t
| j�� d S r$ )r r% ZamqprJ �select�KeyErrorr �SELECT_UNKNOWN_QUEUE�strip�formatZdeselect�DESELECT_UNKNOWN_QUEUE�confr Z
select_addr& )r0 rK �exclude�excr! r! r" r: � s ��
zWorkController.setup_queuesc sf t � jjj�}|r0|t |�7 }� fdd�|D � |� _dd� � jj�� D �}t t|�|B �� jj_d S )Nc s g | ]}� j j�|��qS r! )r% r* Zimport_task_module��.0�mrL r! r" �
<listcomp>� � z1WorkController.setup_includes.<locals>.<listcomp>c S s h | ] }|j j�qS r! )� __class__r )r^ �taskr! r! r" � <setcomp>� s �z0WorkController.setup_includes.<locals>.<setcomp>)�tupler% rZ rK �tasks�values�set)r0 Zincludes�prev� task_modulesr! rL r" r; � s
�z WorkController.setup_includesc K s |S r$ r! rM r! r! r" r/ � s zWorkController.prepare_argsc C s t jj| d� d S )Nr4 )r Zworker_shutdownrD rL r! r! r" �_send_worker_shutdown� s z$WorkController._send_worker_shutdownc
C s� z| j �| � W n� ty* | �� Y n� tyh } z(tjd|dd� | jtd� W Y d }~nVd }~0 t y� } z| j|j
d� W Y d }~n&d }~0 t y� | jtd� Y n0 d S )NzUnrecoverable error: %rT)�exc_info)�exitcode) rH �startr � terminate� Exception�logger�criticalrQ r �
SystemExit�code�KeyboardInterrupt)r0 r\ r! r! r" rn � s " zWorkController.startc C s | j j| d|fdd� d S )N�register_with_event_loopz hub.register)�args� description)rH Zsend_all)r0 Zhubr! r! r" rv � s �z'WorkController.register_with_event_loopc C s | � | j|�S r$ )Z_quick_acquire�
_process_task�r0 �reqr! r! r" �_process_task_sem� s z WorkController._process_task_semc C sH z|� | j� W n2 tyB z | �� W n ty< Y n0 Y n0 dS )z2Process task by sending it to the pool of workers.N)Zexecute_using_pool�poolr
Z_quick_release�AttributeErrorrz r! r! r" ry � s z WorkController._process_taskc C s& z| j �� W n ty Y n0 d S r$ )rO �closer~ rL r! r! r" �signal_consumer_close� s z$WorkController.signal_consumer_closec C s t � dko | jjjjo | jj S )N�default)r r@ � transportZ
implementsZ asynchronousr% Z
IS_WINDOWSrL r! r! r" rA � s
��z#WorkController.should_use_eventloopFc C sF |dur|| _ | jjtkr:| �� |r.| jjr:| jdd� | �� dS )z7Graceful shutdown of the worker server (Warm shutdown).NT��warm) rm rH r r
r� r} � signal_safe� _shutdownrk )r0 �
in_sighandlerrm r! r! r" rQ � s zWorkController.stopc C s0 | j jtkr,| �� |r | jjr,| jdd� dS )z>Not so graceful shutdown of the worker server (Cold shutdown).Fr� N)rH r r r� r} r� r� )r0 r� r! r! r" ro � s zWorkController.terminateTc C sR | j d urNtt��, | j j| | d� | j �� W d � n1 sD0 Y d S )N)ro )rH r �SHUTDOWN_SOCKET_TIMEOUTrQ �join)r0 r� r! r! r" r� s
zWorkController._shutdownc C sT t | j|||d�� | jr.| j�� | j�� z| j�� W n tyN Y n0 d S )N)� force_reload�reloader)�list�_reload_modulesrO Zupdate_strategiesZreset_rate_limitsr} Zrestartr< )r0 �modules�reloadr� r! r! r" r�
s �
zWorkController.reloadc s. � �fdd�t |d u r �jjjn|p&d�D �S )Nc 3 s | ]}�j |fi � ��V qd S r$ )�_maybe_reload_moduler] �r1 r0 r! r" � <genexpr> s �z1WorkController._reload_modules.<locals>.<genexpr>r! )rh r% r* rj )r0 r� r1 r! r� r" r� s � ��z WorkController._reload_modulesc C sH |t jvr$t�d|� | jj�|�S |rDt�d|� tt j| |�S d S )Nzimporting module %szreloading module %s)�sysr� rq �debugr% r* Zimport_from_cwdr )r0 �moduler� r� r! r! r" r� s
z#WorkController._maybe_reload_modulec C s8 t �tj�| j }| jjt�� t | j
j �t |�
� �d�S )N)�total�pid�clock�uptime)r r' r r( r) r Z total_count�os�getpid�strr% r� �round�
total_seconds)r0 r� r! r! r" �info( s
�zWorkController.infoc C sb t d u rtd��t �t j�}|j|j|j|j|j|j |j
|j |j |j
|j|j|j|j|j|jd�S )Nz%rusage not supported by this platform)�utimeZstimeZmaxrssZixrssZidrssZisrssZminfltZmajfltZnswapZinblockZoublockZmsgsndZmsgrcvZnsignalsZnvcswZnivcsw)�resourcer< Z getrusageZ RUSAGE_SELFZru_utimeZru_stimeZ ru_maxrssZru_ixrssZru_idrssZru_isrssZ ru_minfltZ ru_majfltZru_nswapZ
ru_inblockZ
ru_oublockZ ru_msgsndZ ru_msgrcvZ ru_nsignalsZru_nvcswZ ru_nivcsw)r0 �sr! r! r" �rusage/ s( �zWorkController.rusagec C s` | � � }|�| j� | �� |�| jj� | j�� z| �� |d< W n tyZ d|d< Y n0 |S )Nr� zN/A)r� �updaterH rO r� r< )r0 r� r! r! r" �statsF s zWorkController.statsc C s dj | | jr| j�� ndd�S )z``repr(worker)``.z#<Worker: {self.hostname} ({state})>ZINIT)r0 r )rX rH Z human_staterL r! r! r" �__repr__P s �zWorkController.__repr__c C s | j S )z#``str(worker) == worker.hostname``.)r& rL r! r! r" �__str__W s zWorkController.__str__c C s t S r$ r rL r! r! r" r [ s zWorkController.state�WARNc K s | j j} || _|| _| d|�| _| d|�| _| d||�| _| d|�| _| d|�| _| d|�| _ | pd|
| _
| d| �| _ | d|
�| _ | d | �| _
| d
||�| _| d |�| _| d ||�| _| d
||�| _| d||�| _| d|�| _| d| �| _t| d|��| _| d|�| _| d|�| _d S )NZworker_concurrencyZworker_send_task_eventsZ worker_poolZworker_consumerZ worker_timerZworker_timer_precisionZworker_autoscalerZworker_pool_putlocksZworker_pool_restartsZworker_state_dbZbeat_schedule_filenameZbeat_scheduler�task_time_limit�task_soft_time_limitZworker_max_tasks_per_childZworker_max_memory_per_childZworker_prefetch_multiplierZworker_disable_rate_limits�worker_lost_wait)r% �eitherr= �logfiler � task_eventsrF � consumer_cls� timer_cls�timer_precision� optimization�autoscaler_cls�
pool_putlocks�
pool_restarts�statedb�schedule_filename� scheduler�
time_limit�soft_time_limit�max_tasks_per_child�max_memory_per_child�int�prefetch_multiplier�disable_rate_limitsr� ) r0 r r= r� r� r} r� r� r� r� r� r� r� �Or� r� r� r� rF Zstate_dbr� r� Z
scheduler_clsr� r� r� r� r� r� Z_kwr� r! r! r" r, _ sN �
�������z WorkController.setup_defaultsc C sV | j }ttj�}|jjr d}|jjdkrR|rRd|jj� d�}t�|� t |jj� dS )a� Wait :setting:`worker_soft_shutdown_timeout` if soft shutdown is enabled.
To enable soft shutdown, set the :setting:`worker_soft_shutdown_timeout` in the
configuration. Soft shutdown can be used to allow the worker to finish processing
few more tasks before initiating a cold shutdown. This mechanism allows the worker
to finish short tasks that are already in progress and requeue long-running tasks
to be picked up by another worker.
.. warning::
If there are no tasks in the worker, the worker will not wait for the
soft shutdown timeout even if it is set as it makes no sense to wait for
the timeout when there are no tasks to process.
Tr z)Initiating Soft Shutdown, terminating in z secondsN)
r% re r Zactive_requestsrZ Z#worker_enable_soft_shutdown_on_idleZ worker_soft_shutdown_timeoutrq �warningr )r0 r% �requests�logr! r! r" �wait_for_soft_shutdown� s
z%WorkController.wait_for_soft_shutdown)NN)NNNNNN)N)FN)F)T)NFN)N)FN) Nr� NNNNNNNNNNNNNNNNNNNNNNNNNN).r r r r r% rN rH r} Z semaphorerm r r# r2 r. rG r+ r- r6 r> r7 r8 r: r; r/ rk rn rv r| ry r� rA rQ ro r� r� r� r� r� r� r� r� r� �propertyr r, r� r! r! r! r" r ? s�
�
(
�
=r )0r r� r� r r �timer Zbilliardr Zkombu.utils.compatr Zceleryr r rE r Zcelery.bootstepsr
r Zcelery.exceptionsr r
r Zcelery.platformsr r Zcelery.utils.importsr Zcelery.utils.logr r rq Zcelery.utils.nodenamesr r Zcelery.utils.textr Zcelery.utils.threadsr � r r� � ImportError�__all__r� rV rY r r! r! r! r" �<module> s8