This document describes the current stable version of Celery (5.7). For development docs, go here.
Source code for celery.concurrency.prefork
"""Prefork execution pool.
Pool implementation using :mod:`multiprocessing`.
"""
import os
import threading
import time
from billiard import forking_enable, get_start_method, set_start_method
from billiard.common import REMAP_SIGTERM, TERM_SIGNAME
from billiard.pool import CLOSE, RUN
from billiard.pool import Pool as BlockingPool
from kombu.asynchronous import get_event_loop
from kombu.message import Message
from celery import platforms, signals
from celery._state import _set_task_join_will_block, set_default_app
from celery.app import base as _app_base
from celery.app import trace
from celery.concurrency.base import BasePool
from celery.utils.functional import noop
from celery.utils.log import get_logger
from .asynpool import AsynPool
__all__ = ('TaskPool', 'process_initializer', 'process_destructor')
#: List of signals to reset when a child process starts.
WORKER_SIGRESET = {
'SIGTERM', 'SIGHUP', 'SIGTTIN', 'SIGTTOU', 'SIGUSR1',
}
#: List of signals to ignore when a child process starts.
if REMAP_SIGTERM:
WORKER_SIGIGNORE = {'SIGINT', TERM_SIGNAME}
else:
WORKER_SIGIGNORE = {'SIGINT'}
logger = get_logger(__name__)
warning, debug = logger.warning, logger.debug
#: Host name of the worker this pool process belongs to, set by
#: :func:`process_initializer` for :func:`process_destructor`.
_process_hostname = None
[docs]
def process_initializer(app, hostname):
"""Pool child process initializer.
Initialize the child pool process to ensure the correct
app instance is used and things like logging works.
"""
# Each running worker gets SIGKILL by OS when main process exits.
platforms.set_pdeathsig('SIGKILL')
_set_task_join_will_block(True)
platforms.signals.reset(*WORKER_SIGRESET)
platforms.signals.ignore(*WORKER_SIGIGNORE)
platforms.set_mp_process_title('celeryd', hostname=hostname)
if get_start_method() != 'fork':
# billiard started this child as a fresh interpreter on its own (for
# example the macOS default since billiard 4.3), so announce it the
# same way the parent does for worker_pool_start_method 'spawn'.
# Both have to be set before init_worker() imports the task modules:
# the environment variable for whatever this process starts later,
# and the flag in the already-imported module that decides how
# @app.task binds.
os.environ['FORKED_BY_MULTIPROCESSING'] = '1'
_app_base.USING_EXECV = True
# This is for Windows and other platforms not supporting
# fork(). Note that init_worker makes sure it's only
# run once per process.
app.loader.init_worker()
app.loader.init_worker_process()
logfile = os.environ.get('CELERY_LOG_FILE') or None
if logfile and '%i' in logfile.lower():
# logfile path will differ so need to set up logging again.
app.log.already_setup = False
app.log.setup(int(os.environ.get('CELERY_LOG_LEVEL', 0) or 0),
logfile,
bool(os.environ.get('CELERY_LOG_REDIRECT', False)),
str(os.environ.get('CELERY_LOG_REDIRECT_LEVEL')),
hostname=hostname)
if os.environ.get('FORKED_BY_MULTIPROCESSING'):
# the child is a fresh interpreter (spawn, forkserver or execv)
trace.setup_worker_optimizations(app, hostname)
else:
app.set_current()
set_default_app(app)
app.finalize()
trace._tasks = app._tasks # enables fast_trace_task optimization.
# rebuild execution handler for all tasks.
from celery.app.trace import build_tracer
for name, task in app.tasks.items():
task.__trace__ = build_tracer(name, task, app.loader, hostname,
app=app)
from celery.worker import state as worker_state
worker_state.reset_state()
global _process_hostname
_process_hostname = hostname
signals.worker_process_init.send(sender=hostname)
[docs]
def process_destructor(pid, exitcode):
"""Pool child process destructor.
Dispatch the :signal:`worker_process_shutdown` signal.
"""
signals.worker_process_shutdown.send(
sender=_process_hostname, pid=pid, exitcode=exitcode,
)
def _is_ack_callback(callback):
# The consumer queues ``promise(message.ack_log_error)`` and
# ``promise(message.reject_log_error)``. Match the bound method by owner
# type and name so a transport's Message subclass overriding them still
# counts; anything unrecognised is left for ``hub.close()``.
fun = getattr(callback, 'fun', None)
return (isinstance(getattr(fun, '__self__', None), Message) and
getattr(fun, '__name__', None) in ('ack_log_error', 'reject_log_error'))
def _run_ready_callbacks(hub):
"""Run the task ack/reject callbacks queued with ``hub.call_soon()``.
With ``task_acks_late`` the acknowledgement of a task that finished
while the pool is being joined is one of these, and nothing else runs
them before ``hub.close()``, which waits for the join. Only the
``Message.ack_log_error`` / ``reject_log_error`` promises are run: the
queue can also hold the transport's socket reader, which re-queues
itself after every read and would hand deliveries to the closed pool,
so anything else is put back for ``hub.close()``. Pop one at a time:
the main thread may still add to the set we drained
(``Consumer.on_send_event_buffered`` bypasses the lock) and must not
abort the batch.
"""
ready = hub._pop_ready()
while ready:
callback = ready.pop()
if not _is_ack_callback(callback):
hub.call_soon(callback)
continue
try:
callback()
except Exception as exc: # pylint: disable=broad-except
hub.on_callback_error(callback, exc)
[docs]
class TaskPool(BasePool):
"""Multiprocessing Pool implementation."""
Pool = AsynPool
BlockingPool = BlockingPool
uses_semaphore = True
write_stats = None
[docs]
def on_start(self):
if self.forking_enable:
forking_enable(True)
else:
# billiard's forking_enable(False) maps to the legacy execv
# mechanism, which depends on the optional _billiard C extension
# and is unavailable on CPython 3 (always warns and silently
# stays on fork). Use the modern 'spawn' start method instead,
# which is implemented in pure Python and actually takes effect.
#
# Each spawned child is a fresh interpreter and therefore must
# re-run the worker optimizations (task registry shortcut used by
# fast_trace_task, Django model validation, etc.). The rest of
# Celery keys this "fresh interpreter" behavior off the
# FORKED_BY_MULTIPROCESSING environment variable (historically set
# by billiard's execv path), so set it for the spawned children.
os.environ['FORKED_BY_MULTIPROCESSING'] = '1'
set_start_method('spawn', force=True)
Pool = (self.BlockingPool if self.options.get('threads', True)
else self.Pool)
proc_alive_timeout = (
self.app.conf.worker_proc_alive_timeout if self.app
else None
)
P = self._pool = Pool(processes=self.limit,
initializer=process_initializer,
on_process_exit=process_destructor,
enable_timeouts=True,
synack=False,
proc_alive_timeout=proc_alive_timeout,
**self.options)
# Create proxy methods
self.on_apply = P.apply_async
self.maintain_pool = P.maintain_pool
self.terminate_job = P.terminate_job
self.grow = P.grow
self.shrink = P.shrink
self.flush = getattr(P, 'flush', None) # FIXME add to billiard
[docs]
def register_with_event_loop(self, loop):
try:
reg = self._pool.register_with_event_loop
except AttributeError:
return
return reg(loop)
[docs]
def on_stop(self):
"""Gracefully stop the pool."""
if self._pool is not None and self._pool._state in (RUN, CLOSE):
self._pool.close()
# Keep firing timers (for heartbeats on async transports) while
# the pool drains. If not using an async transport, no hub exists
# and the timer thread is not created.
hub = get_event_loop()
if hub is not None:
shutdown_event = threading.Event()
def fire_timers_loop():
while not shutdown_event.is_set():
try:
hub.fire_timers()
_run_ready_callbacks(hub)
except Exception:
logger.warning(
"Exception in timer thread during prefork on_stop()",
exc_info=True,
)
# 0.5 seconds was chosen as a balance between joining quickly
# after the pool join is complete and sleeping long enough to
# avoid excessive CPU usage.
time.sleep(0.5)
timer_thread = threading.Thread(
target=fire_timers_loop,
daemon=True,
name="prefork-timer-shutdown",
)
timer_thread.start()
try:
self._pool.join()
finally:
shutdown_event.set()
timer_thread.join(timeout=1.0)
if timer_thread.is_alive():
logger.warning(
"Timer thread in prefork on_stop() did not terminate cleanly"
)
else:
# The results the join waited for queued their acks
# after the thread's last pass.
try:
_run_ready_callbacks(hub)
except Exception:
logger.warning(
"Exception running ready callbacks after prefork pool join",
exc_info=True,
)
else:
self._pool.join()
self._pool = None
[docs]
def on_terminate(self):
"""Force terminate the pool."""
if self._pool is not None:
self._pool.terminate()
self._pool = None
[docs]
def on_close(self):
if self._pool is not None and self._pool._state == RUN:
self._pool.close()
def _get_info(self):
write_stats = getattr(self._pool, 'human_write_stats', None)
info = super()._get_info()
info.update({
'max-concurrency': self.limit,
'processes': [p.pid for p in self._pool._pool],
'max-tasks-per-child': self._pool._maxtasksperchild or 'N/A',
'put-guarded-by-semaphore': self.putlocks,
'timeouts': (self._pool.soft_timeout or 0,
self._pool.timeout or 0),
'writes': write_stats() if write_stats is not None else 'N/A',
})
return info
@property
def num_processes(self):
return self._pool._processes
