This document describes the current stable version of Celery (5.6). For development docs, go here.

# Source code for celery.concurrency.base

    """Base Execution Pool."""
    import logging
    import os
    import sys
    import time
    from typing import Any, Dict

    from billiard.einfo import ExceptionInfo
    from billiard.exceptions import WorkerLostError
    from kombu.utils.encoding import safe_repr

    from celery.exceptions import WorkerShutdown, WorkerTerminate, reraise
    from celery.utils import timer2
    from celery.utils.log import get_logger
    from celery.utils.text import truncate

    __all__ = ('BasePool', 'apply_target')

    logger = get_logger('celery.pool')

    def apply_target(target, args=(), kwargs=None, callback=None,
                     accept_callback=None, pid=None, getpid=os.getpid,
                     propagate=(), monotonic=time.monotonic, **_):
        """Apply function within pool context."""
        kwargs = {} if not kwargs else kwargs
        if accept_callback:
            accept_callback(pid or getpid(), monotonic())
        try:
            ret = target(*args, **kwargs)
        except propagate:
            raise
        except Exception:
            raise
        except (WorkerShutdown, WorkerTerminate):
            raise
        except BaseException as exc:
            try:
                reraise(WorkerLostError, WorkerLostError(repr(exc)),
                        sys.exc_info()[2])
            except WorkerLostError:
                callback(ExceptionInfo())
        else:
            callback(ret)

    class BasePool:
        """Task pool."""

        RUN = 0x1
        CLOSE = 0x2
        TERMINATE = 0x3

        Timer = timer2.Timer

        #: set to true if the pool can be shutdown from within
        #: a signal handler.
        signal_safe = True

        #: set to true if pool uses greenlets.
        is_green = False

        _state = None
        _pool = None
        _does_debug = True

        #: only used by multiprocessing pool
        uses_semaphore = False

        task_join_will_block = True
        body_can_be_buffer = False

        def __init__(self, limit=None, putlocks=True, forking_enable=True,
                     callbacks_propagate=(), app=None, **options):
            self.limit = limit
            self.putlocks = putlocks
            self.options = options
            self.forking_enable = forking_enable
            self.callbacks_propagate = callbacks_propagate
            self.app = app

        def on_start(self):
            pass

        def did_start_ok(self):
            return True

        def flush(self):
            pass

        def on_stop(self):
            pass

        def register_with_event_loop(self, loop):
            pass

        def on_apply(self, *args, **kwargs):
            pass

        def on_terminate(self):
            pass

        def on_soft_timeout(self, job):
            pass

        def on_hard_timeout(self, job):
            pass

        def maintain_pool(self, *args, **kwargs):
            pass

        def terminate_job(self, pid, signal=None):
            raise NotImplementedError(
                f'{type(self)} does not implement kill_job')

        def restart(self):
            raise NotImplementedError(
                f'{type(self)} does not implement restart')

        def stop(self):
            self.on_stop()
            self._state = self.TERMINATE

        def terminate(self):
            self._state = self.TERMINATE
            self.on_terminate()

        def start(self):
            self._does_debug = logger.isEnabledFor(logging.DEBUG)
            self.on_start()
            self._state = self.RUN

        def close(self):
            self._state = self.CLOSE
            self.on_close()

        def on_close(self):
            pass

        def apply_async(self, target, args=None, kwargs=None, **options):
            """Equivalent of the :func:`apply` built-in function.

            Callbacks should optimally return as soon as possible since
            otherwise the thread which handles the result will get blocked.
            """
            kwargs = {} if not kwargs else kwargs
            args = [] if not args else args
            if self._does_debug:
                logger.debug('TaskPool: Apply %s (args:%s kwargs:%s)',
                             target, truncate(safe_repr(args), 1024),
                             truncate(safe_repr(kwargs), 1024))

            return self.on_apply(target, args, kwargs,
                                 waitforslot=self.putlocks,
                                 callbacks_propagate=self.callbacks_propagate,
                                 **options)

        def _get_info(self) -> Dict[str, Any]:
            """
            Return configuration and statistics information. Subclasses should
            augment the data as required.

            :return: The returned value must be JSON-friendly.
            """
            return {
                'implementation': self.__class__.__module__ + ':' + self.__class__.__name__,
                'max-concurrency': self.limit,
            }

        @property
        def info(self):
            return self._get_info()

        @property
        def active(self):
            return self._state == self.RUN

        @property
        def num_processes(self):
            return self.limit