Skip to content

misc: create new process group by set_new_process_group utility #24371

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
potiuk merged 1 commit into from
Jun 12, 2022
Merged

misc: create new process group by set_new_process_group utility #24371

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion airflow/dag_processing/manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,7 +217,7 @@ def _run_processor_manager(
# Make this process start as a new process group - that makes it easy
# to kill all sub-process of this at the OS-level, rather than having
# to iterate the child processes
os.setpgid(0, 0)
set_new_process_group()

setproctitle("airflow scheduler -- DagFileProcessorManager")
# Reload configurations and settings to avoid collision with parent process.
Expand Down
4 changes: 2 additions & 2 deletions airflow/task/task_runner/standard_task_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@

from airflow.settings import CAN_FORK
from airflow.task.task_runner.base_task_runner import BaseTaskRunner
from airflow.utils.process_utils import reap_process_group
from airflow.utils.process_utils import reap_process_group, set_new_process_group


class StandardTaskRunner(BaseTaskRunner):
Expand Down Expand Up @@ -53,7 +53,7 @@ def _start_by_fork(self):
return psutil.Process(pid)
else:
# Start a new process group
os.setpgid(0, 0)
set_new_process_group()
import signal

signal.signal(signal.SIGINT, signal.SIG_DFL)
Expand Down
5 changes: 3 additions & 2 deletions airflow/utils/process_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +61,9 @@ def reap_process_group(
a SIGKILL will be send.

:param process_group_id: process group id to kill.
The process that wants to create the group should run `os.setpgid(0, 0)` as the first
command it executes which will set group id = process_id. Effectively the process that is the
The process that wants to create the group should run
`airflow.utils.process_utils.set_new_process_group()` as the first command
it executes which will set group id = process_id. Effectively the process that is the
"root" of the group has pid = gid and all other processes in the group have different
pids but the same gid (equal the pid of the root process)
:param logger: log handler
Expand Down