wip

Author: blink1073

jupyter_client/client.py

@@ -2,10 +2,10 @@
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
 import asyncio
-import atexit
 import sys
 import time
 import typing as t
+import weakref
 from functools import partial
 from getpass import getpass
 from queue import Empty
@@ -95,7 +95,8 @@ class KernelClient(ConnectionFileMixin):
 
     def _context_default(self) -> zmq.asyncio.Context:
         context = zmq.asyncio.Context()
-        atexit.register(context.destroy)
+        # Use a finalizer to destroy the context.
+        self._finalizer = weakref.finalize(self, context.destroy)
         return context
 
     # The classes to use for the various channels

jupyter_client/multikernelmanager.py

@@ -6,6 +6,7 @@
 import socket
 import typing as t
 import uuid
+import weakref
 
 import zmq
 from traitlets import Any
@@ -95,8 +96,6 @@ def create_kernel_manager(*args: Any, **kwargs: Any) -> KernelManager:
         help="Share a single zmq.Context to talk to all my kernels",
     ).tag(config=True)
 
-    _created_context = Bool(False)
-
     context = Instance("zmq.Context")
 
     _pending_kernels = Dict()
@@ -108,20 +107,18 @@ def _starting_kernels(self):
 
     @default("context")  # type:ignore[misc]
     def _context_default(self) -> zmq.Context:
-        self._created_context = True
-        return zmq.Context()
-
-    def __del__(self):
-        if self._created_context and self.context and not self.context.closed:
-            if self.log:
-                self.log.debug("Destroying zmq context for %s", self)
-            self.context.destroy()
-        try:
-            super_del = super().__del__
-        except AttributeError:
-            pass
-        else:
-            super_del()
+        context = zmq.Context()
+        # Use a finalizer to destroy the context.
+        # self._finalizer = weakref.finalize(self, context.destroy)
+
+        raise ValueError('Notes here')
+        """
+        The finalizer hangs because the context can't be destroyed.
+        There are two other things that are probably contributing:
+        - There is an open kernel subprocess at shutdown that is raising a warning on __del__
+        - We are not properly canceling the _add_kernel_when_ready task
+        """
+        return context
 
     connection_dir = Unicode("")
 
@@ -209,15 +206,15 @@ async def _async_start_kernel(self, kernel_name: t.Optional[str] = None, **kwarg
         kwargs['kernel_id'] = kernel_id  # Make kernel_id available to manager and provisioner
 
         starter = ensure_async(km.start_kernel(**kwargs))
-        fut = asyncio.ensure_future(self._add_kernel_when_ready(kernel_id, km, starter))
-        self._pending_kernels[kernel_id] = fut
+        task = asyncio.create_task(self._add_kernel_when_ready(kernel_id, km, starter))
+        self._pending_kernels[kernel_id] = task
         # Handling a Pending Kernel
         if self._using_pending_kernels():
             # If using pending kernels, do not block
             # on the kernel start.
             self._kernels[kernel_id] = km
         else:
-            await fut
+            await task
             # raise an exception if one occurred during kernel startup.
             if km.ready.exception():
                 raise km.ready.exception()  # type: ignore
@@ -246,9 +243,9 @@ async def _async_shutdown_kernel(
         self.log.info("Kernel shutdown: %s" % kernel_id)
         # If the kernel is still starting, wait for it to be ready.
         if kernel_id in self._pending_kernels:
-            kernel = self._pending_kernels[kernel_id]
+            task = self._pending_kernels[kernel_id]
             try:
-                await kernel
+                await task
                 km = self.get_kernel(kernel_id)
                 await t.cast(asyncio.Future, km.ready)
             except Exception:
@@ -311,7 +308,9 @@ async def _async_shutdown_all(self, now: bool = False) -> None:
             for km in kms:
                 try:
                     await km.ready
-                except Exception:
+                except asyncio.exceptions.CancelledError:
+                    self._pending_kernels[km.kernel_id].cancel()
+                except Exception as e:
                     # Will have been logged in _add_kernel_when_ready
                     pass