Class Tasks

celery/worker/consumer/tasks.py:20–119

Bootstep starting the task message consumer.

class Tasks(bootsteps.StartStopStep):
    """Bootstep starting the task message consumer."""

    requires = (Mingle,)

    def __init__(self, c, **kwargs):
        c.task_consumer = c.qos = None
        super().__init__(c, **kwargs)

    def start(self, c):
        """Start task consumer."""
        c.update_strategies()

        qos_global = self.qos_global(c)

        # set initial prefetch count
        c.connection.default_channel.basic_qos(
            0, c.initial_prefetch_count, qos_global,
        )

        c.task_consumer = c.app.amqp.TaskConsumer(
            c.connection, on_decode_error=c.on_decode_error,
        )

        def set_prefetch_count(prefetch_count):
            return c.task_consumer.qos(
                prefetch_count=prefetch_count,
                apply_global=qos_global,
            )
        eta_task_limit = c.app.conf.worker_eta_task_limit
        c.qos = QoS(
            set_prefetch_count, c.initial_prefetch_count, max_prefetch=eta_task_limit
        )

        if c.app.conf.worker_disable_prefetch:
            # Only apply disable-prefetch for Redis brokers
            is_redis_broker = c.connection.transport.driver_type == 'redis'
            if not is_redis_broker:
                logger.warning(
                    f"worker_disable_prefetch is only supported for Redis brokers. "
                    f"Current broker transport: {c.connection.transport.driver_type}. "
                    f"Ignoring disable_prefetch setting."
                )
                return

            from types import MethodType

            from celery.worker import state
            channel_qos = c.task_consumer.channel.qos
            original_can_consume = channel_qos.can_consume

            def can_consume(self):
                # Prefer autoscaler's max_concurrency if set; otherwise fall back to pool size
                limit = getattr(c.controller, "max_concurrency", None) or c.pool.num_processes
                if len(state.reserved_requests) >= limit:
                    return False
                return original_can_consume()