This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch gc-freeze-triggerer-serve-logs in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9244e61165799068b9c0eb508658e9f361780e0f Author: Rahul Vats <[email protected]> AuthorDate: Fri Oct 30 13:20:01 2026 +0530 Apply gc.freeze before forking serve_logs in the triggerer The triggerer forks serve_logs from its main process. Under the fork start method the child shares the parent's pages copy-on-write; a later full GC writes to every object's header and splits those shared pages into private copies, inflating anonymous memory with no new allocation. Freeze the GC before the fork and unfreeze right after, mirroring the executors (#58365), the Dag processor manager (#60505), and the Celery worker (#62212). No-op under forkserver/spawn. --- airflow-core/src/airflow/cli/commands/triggerer_command.py | 13 ++++++++++++- .../tests/unit/cli/commands/test_triggerer_command.py | 14 ++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/cli/commands/triggerer_command.py b/airflow-core/src/airflow/cli/commands/triggerer_command.py index 3f881e35c51..00b43552a5d 100644 --- a/airflow-core/src/airflow/cli/commands/triggerer_command.py +++ b/airflow-core/src/airflow/cli/commands/triggerer_command.py @@ -18,6 +18,7 @@ from __future__ import annotations +import gc from collections.abc import Generator from contextlib import contextmanager from functools import partial @@ -43,7 +44,17 @@ def _serve_logs(skip_serve_logs: bool = False) -> Generator[None, None, None]: if skip_serve_logs is False: port = conf.getint("logging", "trigger_log_server_port", fallback=8794) sub_proc = Process(target=partial(serve_logs, port=port)) - sub_proc.start() + # Freeze the GC before forking so existing objects move to the permanent generation and + # are not written to by a later full GC, which would split the COW-shared pages into + # private copies and inflate memory without any new allocation. Mirrors the executors and + # the Dag processor manager, which freeze before forking workers. Only has an effect under + # the ``fork`` start method; a no-op for ``forkserver``/``spawn`` (children re-import). + # Ref: https://docs.python.org/3/library/gc.html#gc.freeze + gc.freeze() + try: + sub_proc.start() + finally: + gc.unfreeze() try: yield finally: diff --git a/airflow-core/tests/unit/cli/commands/test_triggerer_command.py b/airflow-core/tests/unit/cli/commands/test_triggerer_command.py index 449ed61626c..37b1ebabc8f 100644 --- a/airflow-core/tests/unit/cli/commands/test_triggerer_command.py +++ b/airflow-core/tests/unit/cli/commands/test_triggerer_command.py @@ -84,6 +84,20 @@ class TestTriggererCommand: execute_callable=mock_trigger_job_runner.return_value._execute, ) + @mock.patch("airflow.cli.commands.triggerer_command.gc") + @mock.patch("airflow.cli.commands.triggerer_command.Process") + def test_serve_logs_freezes_gc_around_fork(self, mock_process, mock_gc): + """serve_logs must be forked with the GC frozen so a later GC can't split COW-shared pages.""" + manager = mock.Mock() + manager.attach_mock(mock_gc.freeze, "freeze") + manager.attach_mock(mock_gc.unfreeze, "unfreeze") + manager.attach_mock(mock_process.return_value.start, "start") + + with triggerer_command._serve_logs(skip_serve_logs=False): + pass + + assert manager.mock_calls == [mock.call.freeze(), mock.call.start(), mock.call.unfreeze()] + @mock.patch("airflow.cli.hot_reload.run_with_reloader") def test_triggerer_with_dev_flag(self, mock_reloader): """Ensure that triggerer with --dev flag uses hot-reload"""
