Servers cannot restart again

Author: fjetter

distributed/core.py

@@ -310,30 +310,34 @@ async def start_unsafe(self):
         return self
 
     async def start(self):
-        async with self._startup_lock:
-            if self.status == Status.failed:
-                assert self.__startup_exc is not None
-                raise self.__startup_exc
-            elif self.status != Status.init:
-                return self
-            timeout = getattr(self, "death_timeout", None)
-
-            async def _close_on_failure(exc: Exception) -> None:
-                await self.close()
-                self.status = Status.failed
-                self.__startup_exc = exc
+        try:
+            async with self._startup_lock:
+                if self.status == Status.failed:
+                    assert self.__startup_exc is not None
+                    raise self.__startup_exc
+                elif self.status != Status.init:
+                    return self
+                timeout = getattr(self, "death_timeout", None)
+
+                async def _close_on_failure(exc: Exception) -> None:
+                    await self.close()
+                    self.status = Status.failed
+                    self.__startup_exc = exc
 
-            try:
-                await asyncio.wait_for(self.start_unsafe(), timeout=timeout)
-            except asyncio.TimeoutError as exc:
-                await _close_on_failure(exc)
-                raise asyncio.TimeoutError(
-                    f"{type(self).__name__} start timed out after {timeout}s."
-                ) from exc
-            except Exception as exc:
-                await _close_on_failure(exc)
-                raise RuntimeError(f"{type(self).__name__} failed to start.") from exc
-            self.status = Status.running
+                try:
+                    await asyncio.wait_for(self.start_unsafe(), timeout=timeout)
+                except asyncio.TimeoutError as exc:
+                    await _close_on_failure(exc)
+                    raise asyncio.TimeoutError(
+                        f"{type(self).__name__} start timed out after {timeout}s."
+                    ) from exc
+                except Exception as exc:
+                    await _close_on_failure(exc)
+                    raise RuntimeError(
+                        f"{type(self).__name__} failed to start."
+                    ) from exc
+                self.status = Status.running
+        finally:
             self.__started.set()
         return self
 
@@ -515,7 +519,7 @@ async def handle_comm(self, comm):
         logger.debug("Connection from %r to %s", address, type(self).__name__)
         self._comms[comm] = op
 
-        await self
+        await self.__started.wait()
         try:
             while not self.__stopped:
                 try: