From acce95183b3411355feaad5a40b20718e8ce5a73 Mon Sep 17 00:00:00 2001
From: Matthew Rocklin <mrocklin@gmail.com>
Date: Tue, 20 Dec 2022 09:42:50 -0600
Subject: [PATCH] WorkerPorcess blocks on kill if still starting

Originally authored by @fjetter
---
 distributed/nanny.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/distributed/nanny.py b/distributed/nanny.py
index 4ae0e28b48f..1acdcf279e4 100644
--- a/distributed/nanny.py
+++ b/distributed/nanny.py
@@ -796,8 +796,13 @@ async def kill(
         if self.status == Status.stopping:
             await self.stopped.wait()
             return
+        # If the process is not properly up it will not watch the closing queue
+        # and we may end up leaking this process
+        # Therefore wait for it to be properly started before killing it
+        if self.status == Status.starting:
+            await self.running.wait()
+
         assert self.status in (
-            Status.starting,
             Status.running,
             Status.failed,  # process failed to start, but hasn't been joined yet
         ), self.status