FM / worker

/opt/imh-python/lib/python3.9/site-packages/celery/worker/

/opt/imh-python/lib/python3.9/site-packages/celery/worker

Name	Size	Mode	Actions
consumer/	-	0755	rm
__pycache__/	-	0755	rm
autoscale.py	4593	0644	edit dl rm
components.py	7497	0644	edit dl rm
control.py	19882	0644	edit dl rm
heartbeat.py	2107	0644	edit dl rm
loops.py	4433	0644	edit dl rm
pidbox.py	3630	0644	edit dl rm
request.py	27229	0644	edit dl rm
state.py	8583	0644	edit dl rm
strategy.py	7324	0644	edit dl rm
worker.py	14488	0644	edit dl rm
__init__.py	95	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/celery/worker/loops.py (4433B)

"""The consumers highly-optimized inner loop."""
import errno
import socket

from celery import bootsteps
from celery.exceptions import WorkerLostError
from celery.utils.log import get_logger

from . import state

__all__ = ('asynloop', 'synloop')

# pylint: disable=redefined-outer-name
# We cache globals and attribute lookups, so disable this warning.

logger = get_logger(__name__)

def _quick_drain(connection, timeout=0.1):
    try:
        connection.drain_events(timeout=timeout)
    except Exception as exc:  # pylint: disable=broad-except
        exc_errno = getattr(exc, 'errno', None)
        if exc_errno is not None and exc_errno != errno.EAGAIN:
            raise

def _enable_amqheartbeats(timer, connection, rate=2.0):
    heartbeat_error = [None]

if not connection:
        return heartbeat_error

heartbeat = connection.get_heartbeat_interval()  # negotiated
    if not (heartbeat and connection.supports_heartbeats):
        return heartbeat_error

def tick(rate):
        try:
            connection.heartbeat_check(rate)
        except Exception as e:
            # heartbeat_error is passed by reference can be updated
            # no append here list should be fixed size=1
            heartbeat_error[0] = e

timer.call_repeatedly(heartbeat / rate, tick, (rate,))
    return heartbeat_error

def asynloop(obj, connection, consumer, blueprint, hub, qos,
             heartbeat, clock, hbrate=2.0):
    """Non-blocking event loop."""
    RUN = bootsteps.RUN
    update_qos = qos.update
    errors = connection.connection_errors

on_task_received = obj.create_task_handler()

heartbeat_error = _enable_amqheartbeats(hub.timer, connection, rate=hbrate)

consumer.on_message = on_task_received
    obj.controller.register_with_event_loop(hub)
    obj.register_with_event_loop(hub)
    consumer.consume()
    obj.on_ready()

# did_start_ok will verify that pool processes were able to start,
    # but this will only work the first time we start, as
    # maxtasksperchild will mess up metrics.
    if not obj.restart_count and not obj.pool.did_start_ok():
        raise WorkerLostError('Could not start worker processes')

# consumer.consume() may have prefetched up to our
    # limit - drain an event so we're in a clean state
    # prior to starting our event loop.
    if connection.transport.driver_type == 'amqp':
        hub.call_soon(_quick_drain, connection)

# FIXME: Use loop.run_forever
    # Tried and works, but no time to test properly before release.
    hub.propagate_errors = errors
    loop = hub.create_loop()

try:
        while blueprint.state == RUN and obj.connection:
            state.maybe_shutdown()
            if heartbeat_error[0] is not None:
                raise heartbeat_error[0]

# We only update QoS when there's no more messages to read.
            # This groups together qos calls, and makes sure that remote
            # control commands will be prioritized over task messages.
            if qos.prev != qos.value:
                update_qos()

try:
                next(loop)
            except StopIteration:
                loop = hub.create_loop()
    finally:
        try:
            hub.reset()
        except Exception as exc:  # pylint: disable=broad-except
            logger.exception(
                'Error cleaning up after event loop: %r', exc)

def synloop(obj, connection, consumer, blueprint, hub, qos,
            heartbeat, clock, hbrate=2.0, **kwargs):
    """Fallback blocking event loop for transports that doesn't support AIO."""
    RUN = bootsteps.RUN
    on_task_received = obj.create_task_handler()
    perform_pending_operations = obj.perform_pending_operations
    heartbeat_error = [None]
    if getattr(obj.pool, 'is_green', False):
        heartbeat_error = _enable_amqheartbeats(obj.timer, connection, rate=hbrate)
    consumer.on_message = on_task_received
    consumer.consume()

obj.on_ready()

while blueprint.state == RUN and obj.connection:
        state.maybe_shutdown()
        if heartbeat_error[0] is not None:
            raise heartbeat_error[0]
        if qos.prev != qos.value:
            qos.update()
        try:
            perform_pending_operations()
            connection.drain_events(timeout=2.0)
        except socket.timeout:
            pass
        except OSError:
            if blueprint.state == RUN:
                raise

cmd: