hooiv commented on PR #70163:
URL: https://github.com/apache/airflow/pull/70163#issuecomment-5076846557
Thanks for the review! I have pushed a commit that adds a unit test for this
fix in `tests/providers/celery/executors/test_celery_executor.py`.
Regarding how `billiard` triggers this bug, Celery's multiprocessing fork
(`billiard`) installs a global `SIGCHLD` handler upon initialization. This
handler loops and calls `waitpid(-1, os.WNOHANG)` to aggressively reap *any*
child process that exits, which intercepts the exit statuses of child processes
spawned by the Airflow Task SDK.
Here is a minimal reproducible example demonstrating the issue:
`python
import os
import time
import psutil
from billiard.pool import Pool
if __name__ == "__main__":
# 1. Starting a billiard pool installs a global SIGCHLD handler
# that aggressively calls waitpid(-1, WNOHANG) to reap ANY exiting
children
pool = Pool(1)
# 2. Airflow Task SDK forks a child process to run the task
pid = os.fork()
if pid == 0:
# Task subprocess does its work and exits
os._exit(42)
proc = psutil.Process(pid)
# Give the child time to exit and billiard's signal handler time to wake
up and reap it
time.sleep(0.2)
# 3. Airflow Task SDK tries to check the exit status, but wait() returns
None
# because billiard already stole the exit status!
print(f"Child PID {pid}")
print(f"psutil.wait(timeout=0) returned: {proc.wait(timeout=0)}")
pool.close()
pool.join()
`
If you run this script, `proc.wait(timeout=0)` will return `None` (meaning
the status has been consumed) instead of the actual exit code `42`. By
resetting `SIGCHLD` to `SIG_DFL` before executing the workload, we prevent
`billiard` from stealing the exit status of the Airflow tasks.
--
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]
