This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit caffeafaa258bfb5af33466924ce02c8b5dee04e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 30 21:27:08 2026 +0530
[v3-4-test] Freeze the GC before the triggerer's forks so memory stays flat
(#76016) (#76029)
* 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.
* Trim the gc.freeze comment to the non-obvious point
* Keep the GC frozen after forking serve_logs (per review)
Unfreezing right after the fork let the main process's next full GC split
the
COW-shared pages, cancelling the benefit. Freeze and stay frozen under the
fork
start method only (forks happen once, like the Dag processor manager); skip
the
freeze entirely for forkserver/spawn.
* Move the gc.freeze to before the trigger-runner fork
Freezing before the serve_logs fork was too early -- the heavy object graph
is
built afterward, so the pages shared with the later trigger-runner fork were
still split by the next full GC and memory still stepped (~246->431 MB PSS).
Freeze right before the runner's bare fork instead, gated on not
_should_use_exec()
(no COW sharing when the child execs), and stay frozen. Reverts the
serve_logs
change; measured flat (~245->246 MB PSS).
* Freeze the GC before the serve_logs fork too, not only the runner fork
Ephraim's 4-way PSS measurement on 3.4.0a4 (fork start method, GC forced in
all three triggerer processes) showed the runner-fork freeze alone still
steps
(245 -> 341 MB): serve_logs is forked earlier, shares the heap, and its own
full GC copies it. Freezing before the serve_logs fork keeps it flat and the
runner inherits the frozen generation. Keep the runner-fork freeze for
--skip-serve-logs and forkserver, where nothing is frozen beforehand.
---------
(cherry picked from commit 5af4c5bf12da93f975d3329c92deb660578462e9)
Co-authored-by: Rahul Vats <[email protected]>
Co-authored-by: Claude <[email protected]>
---
.../src/airflow/cli/commands/triggerer_command.py | 9 +++++++-
.../src/airflow/jobs/triggerer_job_runner.py | 7 ++++++
.../unit/cli/commands/test_triggerer_command.py | 27 ++++++++++++++++++++++
airflow-core/tests/unit/jobs/test_triggerer_job.py | 24 +++++++++++++++++++
4 files changed, 66 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..feeb50fbd74 100644
--- a/airflow-core/src/airflow/cli/commands/triggerer_command.py
+++ b/airflow-core/src/airflow/cli/commands/triggerer_command.py
@@ -18,10 +18,11 @@
from __future__ import annotations
+import gc
from collections.abc import Generator
from contextlib import contextmanager
from functools import partial
-from multiprocessing import Process
+from multiprocessing import Process, get_start_method
from airflow.cli.commands.daemon_utils import run_command_with_daemon_option
from airflow.configuration import conf
@@ -43,6 +44,12 @@ 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))
+ if get_start_method() == "fork":
+ # Freeze and stay frozen: unfreezing would let this process's next
full GC split the
+ # pages the forked child shares copy-on-write (a memory jump with
no new allocation).
+ # The triggerer forks its children once, like the Dag processor
manager. No effect
+ # under forkserver/spawn, where the child re-imports instead of
sharing.
+ gc.freeze()
sub_proc.start()
try:
yield
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 4c6cc56ceea..c6399e2027f 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -19,6 +19,7 @@ from __future__ import annotations
import asyncio
import contextvars
import functools
+import gc
import logging
import math
import os
@@ -566,6 +567,12 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
**kwargs,
):
proc_id = job.id if job is not None else uuid4()
+ if not supervisor._should_use_exec():
+ # Freeze before the bare fork and stay frozen: unfreezing would
let this process's next
+ # full GC split the pages the runner child shares copy-on-write (a
memory jump with no
+ # new allocation). Forked once, like the Dag processor manager.
Skipped when the child
+ # execs, which discards the inherited pages.
+ gc.freeze()
# Triggers run user code that polls APIs / watches queues / hits HTTP
# endpoints -- almost always network calls that trigger macOS-unsafe
ObjC
# initialization. Fork+exec a clean interpreter for the runner child.
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..472c8927db3 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,33 @@ class TestTriggererCommand:
execute_callable=mock_trigger_job_runner.return_value._execute,
)
+ @pytest.mark.parametrize(
+ ("start_method", "should_freeze"),
+ [("fork", True), ("forkserver", False), ("spawn", False)],
+ )
+ @mock.patch("airflow.cli.commands.triggerer_command.gc", autospec=True)
+ @mock.patch("airflow.cli.commands.triggerer_command.get_start_method",
autospec=True)
+ @mock.patch("airflow.cli.commands.triggerer_command.Process",
autospec=True)
+ def test_serve_logs_freezes_gc_before_fork_only_under_fork(
+ self, mock_process, mock_get_start_method, mock_gc, start_method,
should_freeze
+ ):
+ """Under ``fork`` serve_logs is forked with the GC frozen and left
frozen (no unfreeze, or a
+ later GC would split the COW-shared pages); other start methods skip
the freeze entirely."""
+ mock_get_start_method.return_value = start_method
+ manager = mock.Mock()
+ manager.attach_mock(mock_gc.freeze, "freeze")
+ manager.attach_mock(mock_process.return_value.start, "start")
+
+ with triggerer_command._serve_logs(skip_serve_logs=False):
+ pass
+
+ mock_gc.unfreeze.assert_not_called()
+ if should_freeze:
+ assert manager.mock_calls == [mock.call.freeze(),
mock.call.start()]
+ else:
+ mock_gc.freeze.assert_not_called()
+ mock_process.return_value.start.assert_called_once()
+
@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"""
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index bbf01eba0f1..c1d09ac3b9d 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -336,6 +336,30 @@ def test_start_opts_into_fork_exec(monkeypatch, mocker,
platform_uses_exec):
assert base_start.call_args.kwargs["target"] ==
TriggerRunnerSupervisor.run_in_process
[email protected]("platform_uses_exec", [True, False])
+def test_start_freezes_gc_before_bare_fork(monkeypatch, mocker,
platform_uses_exec):
+ """On bare-fork platforms the runner is forked with the GC frozen and left
frozen, so a later
+ full GC can't split the COW-shared pages; when the child execs there is no
sharing, so skip it."""
+ monkeypatch.setattr(supervisor, "_should_use_exec", lambda:
platform_uses_exec)
+ mock_gc = mocker.patch("airflow.jobs.triggerer_job_runner.gc",
autospec=True)
+ base_start = mocker.patch(
+ "airflow.sdk.execution_time.supervisor.WatchedSubprocess.start",
return_value=MagicMock()
+ )
+ manager = mocker.MagicMock()
+ manager.attach_mock(mock_gc.freeze, "freeze")
+ manager.attach_mock(base_start, "start")
+
+ TriggerRunnerSupervisor.start(job=Job(id=999), capacity=10)
+
+ mock_gc.unfreeze.assert_not_called()
+ if platform_uses_exec:
+ mock_gc.freeze.assert_not_called()
+ base_start.assert_called_once()
+ else:
+ # Freeze must happen before the fork (super().start) and never
unfreeze.
+ assert [c[0] for c in manager.mock_calls] == ["freeze", "start"]
+
+
@pytest.mark.parametrize("option_value", ["True", "False"])
def test_start_ignores_execute_tasks_new_python_interpreter(mocker,
option_value):
"""
