This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 3f600661aac Add Celery worker mp_start_method config to curb Python
3.14 memory (#69015)
3f600661aac is described below
commit 3f600661aac6534a4de2ecf32bc1030358919d64
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Fri Jun 26 22:09:31 2026 +0900
Add Celery worker mp_start_method config to curb Python 3.14 memory (#69015)
Python 3.14 changed the Unix multiprocessing default from fork to forkserver
(gh-84559). The Celery worker starts its log server and the
stale-bundle-cleanup
loop (and the optional SecretCache manager) as stdlib
multiprocessing.Process
children. Under forkserver each re-imports Airflow and spins up extra
forkserver/resource-tracker processes, so a 3.13 to 3.14 upgrade inflates
the
worker's resident memory even though nothing about the workload changed.
A new [celery] mp_start_method (falling back to [core] mp_start_method)
lets a
deployment pin fork to restore the pre-3.14 behaviour. It governs only the
worker's stdlib multiprocessing helpers; Celery's prefork pool is driven by
billiard, which keeps its own fork default and is unaffected either way.
---
providers/celery/provider.yaml | 29 ++++++++++++++++++++++
.../airflow/providers/celery/cli/celery_command.py | 20 ++++++++++++++-
.../airflow/providers/celery/get_provider_info.py | 14 +++++++++++
.../tests/unit/celery/cli/test_celery_command.py | 27 +++++++++++++++++++-
4 files changed, 88 insertions(+), 2 deletions(-)
diff --git a/providers/celery/provider.yaml b/providers/celery/provider.yaml
index 66e2a053452..352a414e0a5 100644
--- a/providers/celery/provider.yaml
+++ b/providers/celery/provider.yaml
@@ -148,6 +148,35 @@ config:
type: string
example: ~
default: "16"
+ mp_start_method:
+ description: |
+ The ``multiprocessing`` start method the ``airflow celery worker``
process uses for the
+ standard-library ``multiprocessing`` helpers it starts: the log
server (``serve_logs``),
+ the stale-bundle-cleanup process, and the optional ``[secrets]
use_cache`` manager. Must
+ be one of the values returned by
``multiprocessing.get_all_start_methods()`` on your
+ platform (typically ``fork``, ``forkserver`` or ``spawn``). When
unset (the default) it
+ falls back to ``[core] mp_start_method`` and then to the platform
default.
+
+ Python 3.14 changed the Unix default from ``fork`` to
``forkserver``. ``forkserver`` and
+ ``spawn`` re-import Airflow in each helper and start extra
forkserver/resource-tracker
+ processes, which increases the worker's resident memory; set this to
``fork`` to restore
+ the pre-3.14 behaviour. This setting governs the standard-library
``multiprocessing``
+ helpers only: Celery's ``prefork`` pool is driven by ``billiard`` (a
separate fork of
+ ``multiprocessing``) and always uses ``fork``, so it is unaffected
either way.
+ version_added: ~
+ type: string
+ example: "fork"
+ default: ~
+ mp_forkserver_preload:
+ description: |
+ Comma-separated list of modules the ``forkserver`` process should
import up front, so the
+ worker's ``multiprocessing`` helpers inherit them copy-on-write
instead of re-importing
+ them. Only used when the effective ``mp_start_method`` is
``forkserver``. Falls back to
+ ``[core] mp_forkserver_preload`` when unset.
+ version_added: ~
+ type: string
+ example: "airflow"
+ default: ~
worker_autoscale:
description: |
The maximum and minimum number of pool processes that will be used
to dynamically resize
diff --git
a/providers/celery/src/airflow/providers/celery/cli/celery_command.py
b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
index 0e2b66aabf1..3aebe0c2088 100644
--- a/providers/celery/src/airflow/providers/celery/cli/celery_command.py
+++ b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
@@ -37,7 +37,11 @@ from lockfile.pidlockfile import read_pid_from_pidfile,
remove_existing_pidfile
from airflow import settings
from airflow.cli.simple_table import AirflowConsole
from airflow.exceptions import AirflowConfigException
-from airflow.providers.celery.version_compat import AIRFLOW_V_3_0_PLUS,
AIRFLOW_V_3_2_PLUS
+from airflow.providers.celery.version_compat import (
+ AIRFLOW_V_3_0_PLUS,
+ AIRFLOW_V_3_2_PLUS,
+ AIRFLOW_V_3_3_PLUS,
+)
from airflow.providers.common.compat.sdk import conf
from airflow.utils import cli as cli_utils
from airflow.utils.cli import setup_locations
@@ -193,6 +197,20 @@ def logger_setup_handler(logger, **kwargs):
@_providers_configuration_loaded
def worker(args):
"""Start Airflow Celery worker."""
+ # Apply the configured multiprocessing start method before the worker
creates any stdlib
+ # multiprocessing objects -- the serve_logs and stale-bundle-cleanup
helper Processes started
+ # below, and the optional SecretCache Manager. CPython 3.14 switched the
Unix default from fork
+ # to forkserver (gh-84559); under forkserver those helpers re-import
Airflow and spin up extra
+ # forkserver/resource_tracker processes, inflating the worker's resident
memory. Setting
+ # [celery] mp_start_method = fork (or [core] mp_start_method) restores the
pre-3.14 behaviour.
+ # This governs stdlib multiprocessing only; Celery's prefork pool is
driven by billiard, which
+ # keeps its own fork default and is unaffected. Guarded because
set_component_mp_start_method
+ # only exists on Airflow 3.3+.
+ if AIRFLOW_V_3_3_PLUS:
+ from airflow.utils.process_utils import set_component_mp_start_method
+
+ set_component_mp_start_method("celery")
+
team_config = None
if hasattr(args, "team") and args.team:
# Multi-team is enabled, create team-specific Celery app and use team
based config
diff --git a/providers/celery/src/airflow/providers/celery/get_provider_info.py
b/providers/celery/src/airflow/providers/celery/get_provider_info.py
index ce59a55918f..b1a663e3495 100644
--- a/providers/celery/src/airflow/providers/celery/get_provider_info.py
+++ b/providers/celery/src/airflow/providers/celery/get_provider_info.py
@@ -75,6 +75,20 @@ def get_provider_info():
"example": None,
"default": "16",
},
+ "mp_start_method": {
+ "description": "The ``multiprocessing`` start method
the ``airflow celery worker`` process uses for the\nstandard-library
``multiprocessing`` helpers it starts: the log server (``serve_logs``),\nthe
stale-bundle-cleanup process, and the optional ``[secrets] use_cache`` manager.
Must\nbe one of the values returned by
``multiprocessing.get_all_start_methods()`` on your\nplatform (typically
``fork``, ``forkserver`` or ``spawn``). When unset (the default) it\nfalls ba
[...]
+ "version_added": None,
+ "type": "string",
+ "example": "fork",
+ "default": None,
+ },
+ "mp_forkserver_preload": {
+ "description": "Comma-separated list of modules the
``forkserver`` process should import up front, so the\nworker's
``multiprocessing`` helpers inherit them copy-on-write instead of
re-importing\nthem. Only used when the effective ``mp_start_method`` is
``forkserver``. Falls back to\n``[core] mp_forkserver_preload`` when unset.\n",
+ "version_added": None,
+ "type": "string",
+ "example": "airflow",
+ "default": None,
+ },
"worker_autoscale": {
"description": "The maximum and minimum number of pool
processes that will be used to dynamically resize\nthe pool based on
load.Enable autoscaling by providing max_concurrency,min_concurrency\nwith the
``airflow celery worker`` command (always keep minimum processes,\nbut grow to
maximum if necessary).\nPick these numbers based on resources on worker box and
the nature of the task.\nIf autoscale option is available, worker_concurrency
will be ignored.\nhttps://do [...]
"version_added": None,
diff --git a/providers/celery/tests/unit/celery/cli/test_celery_command.py
b/providers/celery/tests/unit/celery/cli/test_celery_command.py
index 99a9506f3d6..49495e818a5 100644
--- a/providers/celery/tests/unit/celery/cli/test_celery_command.py
+++ b/providers/celery/tests/unit/celery/cli/test_celery_command.py
@@ -36,7 +36,11 @@ from airflow.providers.celery.cli.celery_command import
_bundle_cleanup_main, _r
from airflow.providers.common.compat.sdk import conf
from tests_common.test_utils.config import conf_vars
-from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS,
AIRFLOW_V_3_2_PLUS
+from tests_common.test_utils.version_compat import (
+ AIRFLOW_V_3_0_PLUS,
+ AIRFLOW_V_3_2_PLUS,
+ AIRFLOW_V_3_3_PLUS,
+)
PY313 = sys.version_info >= (3, 13)
@@ -195,6 +199,27 @@ class TestWorkerStart:
]
)
+ @pytest.mark.skipif(
+ not AIRFLOW_V_3_3_PLUS, reason="set_component_mp_start_method only
exists on Airflow 3.3+"
+ )
+ @mock.patch("airflow.utils.process_utils.set_component_mp_start_method")
+
@mock.patch("airflow.providers.celery.cli.celery_command.kombu.pools.reset")
+ @mock.patch("airflow.providers.celery.cli.celery_command.Celery")
+ @mock.patch("airflow.providers.celery.cli.celery_command.setup_locations")
+ @mock.patch("airflow.providers.celery.cli.celery_command.Process")
+ @mock.patch("airflow.providers.celery.executors.celery_executor.app")
+ def test_worker_applies_celery_mp_start_method(
+ self, mock_celery_app, mock_popen, mock_locations, mock_celery_cls,
mock_pools_reset, mock_set_mp
+ ):
+ # The worker pins its stdlib multiprocessing start method (serve_logs
/ bundle-cleanup /
+ # SecretCache Manager) from [celery] mp_start_method before spawning
any helper process.
+ mock_locations.return_value = ("pid_file", None, None, None)
+ args = self.parser.parse_args(["celery", "worker", "--concurrency",
"1", "--queues", "queue"])
+
+ celery_command.worker(args)
+
+ mock_set_mp.assert_called_once_with("celery")
+
@pytest.mark.backend("mysql", "postgres")
@pytest.mark.usefixtures("conf_stale_bundle_cleanup_disabled")
