File: //usr/local/lib/python3.9/site-packages/celery/worker/__pycache__/autoscale.cpython-39.pyc
a
X>h� � @ s� d Z ddlZddlZddlmZmZ ddlmZ ddlm Z ddl
m Z ddl m
Z
dd lmZ dd
lmZ d Ze e�Zejejej ZZZeej�d d
��ZG dd� de j �Z G dd� de
�Z dS )z�Pool Autoscaling.
This module implements the internal thread responsible
for growing and shrinking the pool according to the
current autoscale settings.
The autoscale thread is only enabled if
the :option:`celery worker --autoscale` option is used.
� N)� monotonic�sleep)� DummyLock)� bootsteps)�
get_logger)�bgThread� )�state)�Pool)�
Autoscaler�WorkerComponent�AUTOSCALE_KEEPALIVE� c @ s> e Zd ZdZdZdZefZdd� Zdd� Z dd � Z
d
d � Z d S )
r z?Bootstep that starts the autoscaler thread/timer in the worker.r Tc K s |j | _d |_d S �N)Z autoscale�enabled�
autoscaler)�self�w�kwargs� r �A/usr/local/lib/python3.9/site-packages/celery/worker/autoscale.py�__init__& s zWorkerComponent.__init__c C s> | j |j|j|j|j||jr"t� nd d� }|_|js:|S d S )N)�worker�mutex)Z instantiateZautoscaler_cls�pool�max_concurrency�min_concurrencyZ
use_eventloopr r )r r Zscalerr r r �create* s � zWorkerComponent.createc C s* |j j�|jj� |�|jj|jj� d S r )�consumerZon_task_message�addr � maybe_scaleZcall_repeatedly� keepalive)r r Zhubr r r �register_with_event_loop2 s �z(WorkerComponent.register_with_event_loopc C s d|j �� iS )zReturn `Autoscaler` info.r )r �info)r r r r r r# 8 s zWorkerComponent.infoN) �__name__�
__module__� __qualname__�__doc__�labelZ conditionalr
�requiresr r r" r# r r r r r s r c s� e Zd ZdZddedf� fdd� Zdd� Zd dd �Zd d
d �Zd d d
�Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zedd� �Zed d � �Z� ZS )!r z,Background thread to autoscale pool workers.r Nc sN t � �� || _|pt�� | _|| _|| _|| _d | _ || _
| jsJJ d��d S )Nzcannot scale down too fast.) �superr r � threading�Lockr r r r! �_last_scale_upr )r r r r r r! r �� __class__r r r @ s
zAutoscaler.__init__c C s: | j � | �� W d � n1 s$0 Y td� d S )Ng �?)r r r �r r r r �bodyN s &zAutoscaler.bodyc C sZ | j }t| j| j�}||kr.| �|| � dS t| j| j�}||k rV| �|| � dS d S )NT)� processes�min�qtyr �scale_up�maxr �
scale_down)r �reqZprocs�curr r r � _maybe_scaleS s zAutoscaler._maybe_scalec C s | � |�r| j�� d S r )r: r Z
maintain_pool)r r8 r r r r ^ s
zAutoscaler.maybe_scalec C s� | j �v |d ur:|| jk r*| �| j| � | �|� || _|d urb|| jkr\| �|| j � || _| j| jfW d � S 1 s�0 Y d S r )r r2 �_shrink� _update_consumer_prefetch_countr �_growr )r r6 r3 r r r �updateb s
zAutoscaler.updatec C s t � | _| �|�S r )r r- r= �r �nr r r r5 o s zAutoscaler.scale_upc C s& | j r"t� | j | jkr"| �|�S d S r )r- r r! r; r? r r r r7 s s �zAutoscaler.scale_downc C s t d|� | j�|� d S )NzScaling up %s processes.)r# r Zgrowr? r r r r= x s
zAutoscaler._growc
C sj t d|� z| j�|� W nJ ty4 td� Y n2 tyd } ztd|dd� W Y d }~n
d }~0 0 d S )NzScaling down %s processes.z0Autoscaler won't scale down: all processes busy.zAutoscaler: scale_down: %rT)�exc_info)r# r �shrink�
ValueError�debug� Exception�error)r r@ �excr r r r; | s
zAutoscaler._shrinkc C s || j }|r | jj�|� d S r )r r r Z_update_prefetch_count)r Znew_max�diffr r r r< � s
�z*Autoscaler._update_consumer_prefetch_countc C s | j | j| j| jd�S )N)r6 r3 �currentr4 )r r r2 r4 r0 r r r r# � s
�zAutoscaler.infoc C s
t tj�S r )�lenr Zreserved_requestsr0 r r r r4 � s zAutoscaler.qtyc C s | j jS r )r Z
num_processesr0 r r r r2 � s zAutoscaler.processes)N)N)NN)r$ r% r& r' r
r r1 r: r r> r5 r7 r= r; r< r# �propertyr4 r2 �
__classcell__r r r. r r = s$ �
r ) r' �osr+ �timer r Z kombu.asynchronous.semaphorer Zceleryr Zcelery.utils.logr Zcelery.utils.threadsr � r �
componentsr
�__all__r$ �loggerrD r# rF �float�environ�getr
Z
StartStopStepr r r r r r �<module> s