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
The following commit(s) were added to refs/heads/gc-freeze-triggerer-serve-logs
by this push:
new 14109e13edd Move the gc.freeze to before the trigger-runner fork
14109e13edd is described below
commit 14109e13edd93c2f6ca276dd3cc54e3f20df6ef9
Author: Rahul Vats <[email protected]>
AuthorDate: Fri Oct 30 14:42:50 2026 +0530
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).
---
.../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, 32 insertions(+), 35 deletions(-)
diff --git a/airflow-core/src/airflow/cli/commands/triggerer_command.py
b/airflow-core/src/airflow/cli/commands/triggerer_command.py
index feeb50fbd74..3f881e35c51 100644
--- a/airflow-core/src/airflow/cli/commands/triggerer_command.py
+++ b/airflow-core/src/airflow/cli/commands/triggerer_command.py
@@ -18,11 +18,10 @@
from __future__ import annotations
-import gc
from collections.abc import Generator
from contextlib import contextmanager
from functools import partial
-from multiprocessing import Process, get_start_method
+from multiprocessing import Process
from airflow.cli.commands.daemon_utils import run_command_with_daemon_option
from airflow.configuration import conf
@@ -44,12 +43,6 @@ 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 1c67986f564..2680759dbfb 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 472c8927db3..449ed61626c 100644
--- a/airflow-core/tests/unit/cli/commands/test_triggerer_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_triggerer_command.py
@@ -84,33 +84,6 @@ 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 6cd5c17a226..888213d2753 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):
"""
