File: //usr/local/lib/python3.9/site-packages/celery/worker/__pycache__/components.cpython-39.pyc
a
X>hI � @ s d Z ddlZddlZddlmZ ddlmZmZ ddlm Z m
Z
ddl m Z
ddlmZ ddlmZ dd lmZ dd
lmZ dd lmZ d Zd
dhZdZdZ G dd� dej �Z G dd� dej �ZG dd� dej �Z G dd� dej �Z G dd� dej �Z!G dd � d ej �Z"dS ) zWorker-level Bootsteps.� N)�Hub)�get_event_loop�set_event_loop)� DummyLock�LaxBoundedSemaphore)�Timer)� bootsteps)�_set_task_join_will_block)�ImproperlyConfigured)�
IS_WINDOWS)�
worker_logger)r r �Pool�Beat�StateDB�Consumer�eventlet�geventzO-B option doesn't work with eventlet/gevent pools: use standalone beat instead.z�
The worker_pool setting shouldn't be used to select the eventlet/gevent
pools, instead you *must use the -P* argument so that patches are applied
as early as possible.
c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) r zTimer bootstep.c C sD |j rtdd�|_n,|js$|jj|_| j|j|j| j| j d�|_d S )Ng $@)� max_interval)r Zon_errorZon_tick)
�
use_eventloop�_Timer�timerZ timer_cls�pool_clsr � instantiateZtimer_precision�on_timer_error�
on_timer_tick��self�w� r �B/usr/local/lib/python3.9/site-packages/celery/worker/components.py�create# s
�z Timer.createc C s t jd|dd� d S )NzTimer error: %rT)�exc_info)�logger�error)r �excr r r r 1 s zTimer.on_timer_errorc C s t �d|� d S )Nz Timer wake-up! Next ETA %s secs.)r" �debug)r �delayr r r r 4 s zTimer.on_timer_tickN)�__name__�
__module__� __qualname__�__doc__r r r r r r r r s r c sV e Zd ZdZefZ� fdd�Zdd� Zdd� Zdd � Z d
d � Z
d d
� Z dd� Z � Z
S )r z Worker starts the event loop.c s d |_ t� j|fi |�� d S �N)�hub�super�__init__�r r �kwargs�� __class__r r r. = s z Hub.__init__c C s |j S r+ )r r r r r �
include_ifA s zHub.include_ifc C sF t � |_|jd u r8t|jdd �}t|r*|nt|j��|_| �|� | S )NZ requires_hub)r r, �getattrZ _conninfor �_Hubr �_patch_thread_primitives)r r Z required_hubr r r r D s
�
z
Hub.createc C s d S r+ r r r r r �startM s z Hub.startc C s |j �� d S r+ �r, �closer r r r �stopP s zHub.stopc C s |j �� d S r+ r8 r r r r � terminateS s z
Hub.terminatec C s: t � |jj_zddlm} W n ty. Y n0 t |_d S )Nr )�pool)r �app�clock�mutexZbilliardr< � ImportError�Lock)r r r< r r r r6 V s z Hub._patch_thread_primitives)r' r( r) r* r �requiresr. r3 r r7 r: r; r6 �
__classcell__r r r1 r r 8 s r c sP e Zd ZdZefZd� fdd� Zdd� Zdd� Zd d
� Z d d � Z
d
d� Z � Z S )r
a
Bootstep managing the worker pool.
Describes how to initialize the worker pool, and starts and stops
the pool during worker start-up/shutdown.
Adds attributes:
* autoscale
* pool
* max_concurrency
* min_concurrency
Nc s� d |_ d |_|j|_|j| _t|t�rN|�d�\}}}t|�|rHt|�pJdg}||_ |j rh|j \|_|_t
� j |fi |�� d S )N�,r ) r< �max_concurrency� concurrency�min_concurrency� optimization�
isinstance�str� partition�int� autoscaler- r. )r r rM r0 Zmax_c�_Zmin_cr1 r r r. r s
z
Pool.__init__c C s |j r|j �� d S r+ )r< r9 r r r r r9 s z
Pool.closec C s |j r|j �� d S r+ )r< r; r r r r r; � s zPool.terminatec C s� d }d }|j jjtv r$t�tt�� |j p.t }|j
}|j |_ |s~t
|� }|_|jj|_|jj|_d}|jr~|jjr~|j|_ |j}| j|j|j
|j |jf|j|j|j |j |jo�||j |||d|| j |j d� }|_ t!|j"� |S )N�d T)ZinitargsZmaxtasksperchild�max_memory_per_child�timeoutZ soft_timeoutZputlocksZlost_worker_timeout�threads� max_restarts�
allow_restartZforking_enable� semaphoreZsched_strategyr= )#r= �confZ worker_pool� GREEN_POOLS�warnings�warn� UserWarning�W_POOL_SETTINGr r rG Z
_process_task� process_taskr rU �acquireZ_quick_acquire�releaseZ_quick_releaseZ
pool_putlocksr Zuses_semaphoreZ_process_task_semZ
pool_restartsr �hostnameZmax_tasks_per_childrP Z
time_limitZsoft_time_limitZworker_lost_waitrH r< r Ztask_join_will_block)r r rU rS ZthreadedZprocsrT r< r r r r � sD
�
z Pool.createc C s d|j r|j jndiS )Nr< zN/A)r< �infor r r r r` � s z Pool.infoc C s |j �|� d S r+ )r< �register_with_event_loop)r r r, r r r ra � s z Pool.register_with_event_loop)N)
r' r( r) r* r rB r. r9 r; r r` ra rC r r r1 r r
b s
$r
c s2 e Zd ZdZd ZdZd� fdd� Zdd� Z� ZS ) r zWStep used to embed a beat process.
Enabled when the ``beat`` argument is set.
TFc s. | | _ |_d |_t� j|fd|i|�� d S )N�beat)�enabledrb r- r. )r r rb r0 r1 r r r. � s z
Beat.__init__c C s@ ddl m} |jj�d�r"tt��||j|j|j d� }|_
|S )Nr )�EmbeddedService)r r )�schedule_filenameZ
scheduler_cls) Z celery.beatrd r r( �endswithr
� ERR_B_GREENr= re Z schedulerrb )r r rd �br r r r � s � z Beat.create)F) r' r( r) r* �labelZ conditionalr. r rC r r r1 r r � s
r c s( e Zd ZdZ� fdd�Zdd� Z� ZS )r z:Bootstep that sets up between-restart state database file.c s& |j | _d |_t� j|fi |�� d S r+ )�statedbrc � _persistencer- r. r/ r1 r r r. � s zStateDB.__init__c C s, |j �|j |j|jj�|_t�|jj� d S r+ ) �stateZ
Persistentrj r= r> rk �atexit�register�saver r r r r � s zStateDB.create)r' r( r) r* r. r rC r r r1 r r � s r c @ s e Zd ZdZdZdd� ZdS )r z)Bootstep starting the Consumer blueprint.Tc C sn |j rt|j d�|j }n |j|j }| j|j|j|j|j|j ||j
|j |j ||j
|j|j|jd� }|_|S )N� ) r_ � task_eventsZ
init_callbackZinitial_prefetch_countr< r r= �
controllerr, Zworker_options�disable_rate_limits�prefetch_multiplier)rE �maxrt rF r Z consumer_clsr\ r_ rq Zready_callbackr< r r= r, �optionsrs Zconsumer)r r Zprefetch_count�cr r r r � s& � zConsumer.createN)r' r( r) r* �lastr r r r r r � s r )#r* rm rX Zkombu.asynchronousr r5 r r Z kombu.asynchronous.semaphorer r Zkombu.asynchronous.timerr r Zceleryr Z
celery._stater Zcelery.exceptionsr
Zcelery.platformsr Zcelery.utils.logr r" �__all__rW rg r[ ZStepZ
StartStopStepr
r r r r r r r �<module> s* *P