divaltor opened a new issue, #77019:
URL: https://github.com/apache/airflow/issues/77019
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.4.0
### What happened and how to reproduce it?
In Airflow 3.4.0 (Task SDK 1.4.0), `ActivitySubprocess.start` imports the
task runner in the supervisor (`_import_task_runner_before_fork`).
`WatchedSubprocess.start` then forks the task process with a bare `os.fork()`.
After the fork, the cyclic GC in each process writes to the headers of the
inherited objects. This splits the copy-on-write pages into private copies, so
the supervisor and the task process each end up with their own copy of most of
the inherited heap.
With KubernetesExecutor, every task pod runs this pair of processes, so the
duplicated heap is paid once per running task.
`gc.freeze()` before the bare fork, with no unfreeze afterwards (the same
pattern as #76016 for the triggerer), lowers the combined PSS of the two
processes by 27-49 MB:
| DAG file | PSS without freeze | PSS with freeze |
|---|---|---|
| One `EmptyOperator`, no other imports | 178-200 MB | 151 MB |
| Same, plus `import pandas` at top level | 225-234 MB | 198 MB |
Three runs each on `apache/airflow:slim-3.4.0-python3.12` (linux/arm64,
pandas 3.1.1 installed for the second DAG). With the freeze, the result was the
same on every run. The script below copies the supervisor's import and fork
order, then reads `Pss` from `/proc/<pid>/smaps_rollup` after both processes
run `gc.collect()`. This is a snapshot after the DAG parse. In a long task,
each later full GC in the supervisor splits more pages, so the saving can grow
over the task's runtime.
### What you think should happen instead?
<details><summary>Repro script</summary>
```python
import gc, os, sys, time
import airflow.sdk.execution_time.supervisor # noqa: F401
import airflow.sdk.execution_time.task_runner # noqa: F401 (as
_import_task_runner_before_fork does)
def pss(pid):
with open(f"/proc/{pid}/smaps_rollup") as f:
return next(int(l.split()[1]) // 1024 for l in f if
l.startswith("Pss:"))
mode, dag_file = sys.argv[1], sys.argv[2]
if mode == "freeze":
gc.freeze()
r, w = os.pipe()
pid = os.fork()
if pid == 0:
from airflow.dag_processing.dagbag import DagBag
DagBag(dag_folder=dag_file)
gc.collect()
os.write(w, b"x")
time.sleep(3)
os._exit(0)
os.read(r, 1)
gc.collect()
s, t = pss(os.getpid()), pss(pid)
print(f"{mode} supervisor={s} task={t} total={s + t}")
os.waitpid(pid, 0)
```
Run it as `python repro.py nofreeze dag.py`, then as `python repro.py freeze
dag.py`.
</details>
The same fix already exists for the LocalExecutor (#58365), the Dag
processor (#60505), the Celery worker (#62212) and the triggerer (#76016). I
found no issue or PR for the task supervisor.
A possible placement is in `ActivitySubprocess.start`, so the Dag
processor's per-file forks through `WatchedSubprocess.start` are not affected:
```python
use_exec = target is _subprocess_main and task_process_uses_exec()
if target is _subprocess_main and not use_exec:
_import_task_runner_before_fork()
# Forked once per supervisor and never unfrozen: an unfreeze would let
the next
# full GC split the pages the task process shares copy-on-write.
gc.freeze()
```
Why it looks safe:
- The supervisor forks the task process once. The parent memory growth from
repeated freeze/unfreeze cycles in the Dag processor (#60505) needs many forks.
- Objects frozen in the supervisor are never collected. The supervisor lives
for one task and keeps these objects anyway.
- The freeze is skipped when `task_process_uses_exec()` is true (macOS,
`execute_tasks_new_python_interpreter`), because `posix_spawn` shares no pages.
- The task process still collects every object it creates after the fork.
### Operating System
_No response_
### Deployment
None
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
