bingqin2 commented on issue #70117:
URL: https://github.com/apache/airflow/issues/70117#issuecomment-6757886979
I'd like to work on this. The hang reproduces deterministically on main
(d23f21fa31), without Celery. It happens whenever the task process's exit
status has already been collected by something else. In the reproducer below,
`SIGCHLD` is set to `SIG_IGN`, which makes the kernel reap children
automatically and is inherited across `fork` and `exec`. The task process exits
right after the startup message:
| `SIGCHLD` in the supervisor | Task process | `ActivitySubprocess.wait()`
after 15 s |
| --- | --- | --- |
| default | exits, gone | returned 0 |
| `SIG_IGN` | exits, gone | still waiting, `_exit_code` is `None` |
With the status gone, `psutil.Process.wait(timeout=0)` returns `None`
instead of an exit code.
- `_check_subprocess_exit` stores that `None` as `_exit_code` and takes the
"exited" branch, which overwrites `_process_exit_monotonic` on every call.
- `_service_subprocess` returns the `None`, so `_monitor_subprocess` treats
the process as alive and keeps sending heartbeats.
- Its loop, `while self._exit_code is None or self._open_sockets`, never
ends.
- The `socket_cleanup_timeout` check is guarded by `self._exit_code is not
None`, so it never runs.
I found no `SIGCHLD` handler or `waitpid(-1)` call in Celery 5 or billiard
4. So I can't confirm what collects the status in your deployment. An inherited
`SIG_IGN` is one way it can happen. Because of that, I'd fix this in the
supervisor, so the fix holds whatever the cause:
- When `wait()` returns `None` and the process no longer exists, treat it as
exited with an unknown status, log a warning that says so, and record the exit
time only once. That keeps the `socket_cleanup_timeout` safety net for normal
exits, which was the problem with #70163.
- Finish with the state the task already reported, and fail the task
instance if it reported none.
- Add regression tests with a real child process under `SIG_IGN`, next to
the existing normal-exit tests.
<details>
<summary>Reproducer (task-sdk/tests/task_sdk/execution_time/)</summary>
```python
from __future__ import annotations
import os
import signal
import threading
import pytest
from uuid6 import uuid7
from airflow.sdk.api.datamodels._generated import TaskInstance
from airflow.sdk.execution_time.comms import CommsDecoder
from airflow.sdk.execution_time.supervisor import ActivitySubprocess
from task_sdk import FAKE_BUNDLE
from task_sdk.execution_time.test_supervisor import client_with_ti_start #
noqa: F401
@pytest.mark.parametrize("sigchld", [signal.SIG_DFL, signal.SIG_IGN],
ids=["default", "auto-reaped"])
def test_wait_returns_after_task_process_exits(client_with_ti_start,
sigchld): # noqa: F811
def subprocess_main():
CommsDecoder()._get_response()
old = signal.signal(signal.SIGCHLD, sigchld)
try:
proc = ActivitySubprocess.start(
dag_rel_path=os.devnull,
bundle_info=FAKE_BUNDLE,
what=TaskInstance(
id=uuid7(), task_id="b", dag_id="c", run_id="d",
try_number=1,
dag_version_id=uuid7(), queue="default",
),
client=client_with_ti_start,
target=subprocess_main,
)
result = {}
waiter = threading.Thread(target=lambda:
result.update(rc=proc.wait()), daemon=True)
waiter.start()
waiter.join(15)
print(f"\nSIGCHLD={sigchld!r}: wait() returned={not
waiter.is_alive()} rc={result.get('rc')} "
f"_exit_code={proc._exit_code} child_gone={not
__import__('psutil').pid_exists(proc.pid)}")
assert not waiter.is_alive(), "supervisor still waiting 15 s after
the task process exited"
finally:
signal.signal(signal.SIGCHLD, old)
if waiter.is_alive():
proc._exit_code = -1 # let the leaked monitor loop end
```
</details>
--
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]
