Skip to content

Clean up cluster process reaping #6840

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
merged 1 commit into from
Aug 18, 2022
Merged

Clean up cluster process reaping #6840

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
26 changes: 12 additions & 14 deletions distributed/utils_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -568,14 +568,14 @@ def security():
return tls_only_security()


def _kill_join(proc, timeout):
proc.kill()
proc.join(timeout)
if proc.is_alive():
raise multiprocessing.TimeoutError(
f"Process {proc} did not shut down within {timeout}s"
)
proc.close()
def _kill_join_processes(processes):
# Join may hang or cause issues, so make sure all are killed first.
# Note that we don't use a timeout, but rely on the overall pytest timeout.
for proc in processes:
proc.kill()
for proc in processes:
proc.join()
proc.close()


def _close_queue(q):
Expand All @@ -590,15 +590,13 @@ def cluster(
nanny=False,
worker_kwargs=None,
active_rpc_timeout=10,
shutdown_timeout=20,
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unused in codebase

scheduler_kwargs=None,
config=None,
):
worker_kwargs = worker_kwargs or {}
scheduler_kwargs = scheduler_kwargs or {}
config = config or {}

ws = weakref.WeakSet()
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unused in function (and workers_by_pid already held strong references to the Process instances anyway). Plus we do want strong references, so that we can kill them at the end. It's not like garbage-collecting a Process instance means the actual OS process will be cleaned up.

enable_proctitle_on_children()

with check_process_leak(check=True), check_instances(), config_for_cluster_tests():
Expand All @@ -608,6 +606,8 @@ def cluster(
_run_worker = run_worker

with contextlib.ExitStack() as stack:
processes = []
stack.callback(_kill_join_processes, processes)
# The scheduler queue will receive the scheduler's address
scheduler_q = get_mp_context().Queue()
stack.callback(_close_queue, scheduler_q)
Expand All @@ -620,9 +620,8 @@ def cluster(
kwargs=scheduler_kwargs,
daemon=True,
)
ws.add(scheduler)
scheduler.start()
stack.callback(_kill_join, scheduler, shutdown_timeout)
processes.append(scheduler)

# Launch workers
workers_by_pid = {}
Expand All @@ -642,9 +641,8 @@ def cluster(
args=(q, scheduler_q, config),
kwargs=kwargs,
)
ws.add(proc)
proc.start()
stack.callback(_kill_join, proc, shutdown_timeout)
processes.append(proc)
workers_by_pid[proc.pid] = {"proc": proc}

saddr_or_exception = scheduler_q.get()
Expand Down