Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Base class to manage a running kernel"""
3# Copyright (c) Jupyter Development Team.
4# Distributed under the terms of the Modified BSD License.
5import asyncio
6import functools
7import os
8import re
9import signal
10import sys
11import typing as t
12import uuid
13import warnings
14from asyncio.futures import Future
15from concurrent.futures import Future as CFuture
16from contextlib import contextmanager
17from enum import Enum
19import zmq
20from jupyter_core.utils import run_sync
21from traitlets import (
22 Any,
23 Bool,
24 CaselessStrEnum,
25 Dict,
26 DottedObjectName,
27 Float,
28 Instance,
29 TraitError,
30 Type,
31 Unicode,
32 default,
33 observe,
34 observe_compat,
35 validate,
36)
37from traitlets.utils.importstring import import_item
39from . import kernelspec
40from .asynchronous import AsyncKernelClient
41from .blocking import BlockingKernelClient
42from .client import KernelClient
43from .connect import ConnectionFileMixin
44from .managerabc import KernelManagerABC
45from .provisioning import KernelProvisionerBase
46from .provisioning import KernelProvisionerFactory as KPF # noqa
48# After an upgrade to Sphinx 9 and myst 5, the doc build started to fail
49# with the following error: :8: (ERROR/3) Unexpected indentation.
50# This seems to be due to the docstring of the wrapper function inside
51# in_pending_state. However, removing the docstring doe snot fix the issue
52# since we use the :undoc-members: directive with automodule.
53# The workaround is to explicitly set what we want to document
55__all__ = [
56 "AsyncKernelManager",
57 "KernelManager",
58 "in_pending_state",
59 "run_kernel",
60 "start_new_async_kernel",
61 "start_new_kernel",
62]
65class _ShutdownStatus(Enum):
66 """
68 This is so far used only for testing in order to track the internal state of
69 the shutdown logic, and verifying which path is taken for which
70 missbehavior.
72 """
74 Unset = None
75 ShutdownRequest = "ShutdownRequest"
76 SigtermRequest = "SigtermRequest"
77 SigkillRequest = "SigkillRequest"
80F = t.TypeVar("F", bound=t.Callable[..., t.Any])
83def _get_future() -> t.Union[Future, CFuture]:
84 """Get an appropriate Future object"""
85 try:
86 asyncio.get_running_loop()
87 return Future()
88 except RuntimeError:
89 # No event loop running, use concurrent future
90 return CFuture()
93def in_pending_state(method: F) -> F:
94 """Sets the kernel to a pending state by
95 creating a fresh Future for the KernelManager's `ready`
96 attribute. Once the method is finished, set the Future's results.
97 """
99 @t.no_type_check
100 @functools.wraps(method)
101 async def wrapper(self: t.Any, *args: t.Any, **kwargs: t.Any) -> t.Any:
102 """Create a future for the decorated method."""
103 if self._attempted_start or not self._ready:
104 self._ready = _get_future()
105 try:
106 # call wrapped method, await, and set the result or exception.
107 out = await method(self, *args, **kwargs)
108 # Add a small sleep to ensure tests can capture the state before done
109 await asyncio.sleep(0.01)
110 if self.owns_kernel:
111 self._ready.set_result(None)
112 return out
113 except Exception as e:
114 self._ready.set_exception(e)
115 self.log.exception(self._ready.exception())
116 raise e
118 return t.cast(F, wrapper)
121class KernelManager(ConnectionFileMixin):
122 """Manages a single kernel in a subprocess on this host.
124 This version starts kernels with Popen.
125 """
127 _ready: t.Union[Future, CFuture] | None
129 def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
130 """Initialize a kernel manager."""
131 if args:
132 warnings.warn(
133 "Passing positional only arguments to "
134 "`KernelManager.__init__` is deprecated since jupyter_client"
135 " 8.6, and will become an error on future versions. Positional "
136 " arguments have been ignored since jupyter_client 7.0",
137 DeprecationWarning,
138 stacklevel=2,
139 )
140 self._owns_kernel = kwargs.pop("owns_kernel", True)
141 super().__init__(**kwargs)
142 self._shutdown_status = _ShutdownStatus.Unset
143 self._attempted_start = False
144 self._ready = None
146 _created_context: Bool = Bool(False)
148 # The PyZMQ Context to use for communication with the kernel.
149 context: Instance = Instance(zmq.Context)
151 @default("context")
152 def _context_default(self) -> zmq.Context:
153 self._created_context = True
154 return zmq.Context()
156 # the class to create with our `client` method
157 client_class: DottedObjectName = DottedObjectName(
158 "jupyter_client.blocking.BlockingKernelClient", config=True
159 )
160 client_factory: Type = Type(klass=KernelClient, config=True)
162 transport_encryption: CaselessStrEnum = CaselessStrEnum(
163 ["disabled", "auto", "required"],
164 default_value="disabled",
165 config=True,
166 help=(
167 "Transport encryption policy for manager-side provisioning of CurveZMQ server keys for kernels. "
168 "'disabled' (default) does not provision Curve credentials, 'auto' provisions when the kernelspec "
169 "declares support, and 'required' enforces provisioning and fails startup if transport encryption "
170 "cannot be applied."
171 ),
172 )
174 @validate("transport_encryption")
175 def _validate_transport_encryption(self, proposal: dict) -> str:
176 value = proposal["value"]
177 if value in ("auto", "required") and not zmq.has("curve"):
178 msg = (
179 f"transport_encryption={value!r} requires CurveZMQ support, "
180 "but zmq.has('curve') returned False. "
181 "Install pyzmq with libzmq compiled with libsodium to enable CurveZMQ."
182 )
183 raise TraitError(msg)
184 return value
186 def _transport_encryption_policy(self, value: str | None = None) -> str:
187 """Normalize transport encryption input into one of the supported policy values."""
188 if value is None:
189 value = self.transport_encryption
190 normalized = str(value).lower()
191 if normalized not in {"disabled", "auto", "required"}:
192 msg = (
193 "transport_encryption must be one of: 'disabled', 'auto', 'required' "
194 f"(got: {value!r})"
195 )
196 raise ValueError(msg)
197 return normalized
199 def _kernel_supports_curve_encryption(self) -> bool:
200 """Whether kernelspec metadata declares support for Curve transport encryption."""
201 if self.kernel_spec is None:
202 return False
203 metadata = getattr(self.kernel_spec, "metadata", {}) or {}
204 supported_encryption = metadata.get("supported_encryption")
205 if supported_encryption is None:
206 return False
207 if isinstance(supported_encryption, str):
208 return supported_encryption.strip().lower() == "curve"
209 if isinstance(supported_encryption, (list, tuple, set)):
210 normalized = {str(item).strip().lower() for item in supported_encryption}
211 return "curve" in normalized
212 return False
214 @default("client_factory")
215 def _client_factory_default(self) -> Type:
216 return import_item(self.client_class)
218 @observe("client_class")
219 def _client_class_changed(self, change: t.Dict[str, DottedObjectName]) -> None:
220 self.client_factory = import_item(str(change["new"]))
222 kernel_id: t.Union[str, Unicode] = Unicode(None, allow_none=True)
224 # The kernel provisioner with which this KernelManager is communicating.
225 # This will generally be a LocalProvisioner instance unless the kernelspec
226 # indicates otherwise.
227 provisioner: KernelProvisionerBase | None = None
229 kernel_spec_manager: Instance = Instance(kernelspec.KernelSpecManager)
231 @default("kernel_spec_manager")
232 def _kernel_spec_manager_default(self) -> kernelspec.KernelSpecManager:
233 return kernelspec.KernelSpecManager(data_dir=self.data_dir)
235 @observe("kernel_spec_manager")
236 @observe_compat
237 def _kernel_spec_manager_changed(self, change: t.Dict[str, Instance]) -> None:
238 self._kernel_spec = None
240 shutdown_wait_time: Float = Float(
241 5.0,
242 config=True,
243 help="Time to wait for a kernel to terminate before killing it, "
244 "in seconds. When a shutdown request is initiated, the kernel "
245 "will be immediately sent an interrupt (SIGINT), followed"
246 "by a shutdown_request message, after 1/2 of `shutdown_wait_time`"
247 "it will be sent a terminate (SIGTERM) request, and finally at "
248 "the end of `shutdown_wait_time` will be killed (SIGKILL). terminate "
249 "and kill may be equivalent on windows. Note that this value can be"
250 "overridden by the in-use kernel provisioner since shutdown times may"
251 "vary by provisioned environment.",
252 )
254 kernel_name: t.Union[str, Unicode] = Unicode(kernelspec.NATIVE_KERNEL_NAME)
256 @observe("kernel_name")
257 def _kernel_name_changed(self, change: t.Dict[str, str]) -> None:
258 self._kernel_spec = None
259 if change["new"] == "python":
260 self.kernel_name = kernelspec.NATIVE_KERNEL_NAME
262 _kernel_spec: kernelspec.KernelSpec | None = None
264 @property
265 def kernel_spec(self) -> kernelspec.KernelSpec | None:
266 if self._kernel_spec is None and self.kernel_name != "":
267 self._kernel_spec = self.kernel_spec_manager.get_kernel_spec(self.kernel_name)
268 return self._kernel_spec
270 cache_ports: Bool = Bool(
271 False,
272 config=True,
273 help="True if the MultiKernelManager should cache ports for this KernelManager instance",
274 )
276 @default("cache_ports")
277 def _default_cache_ports(self) -> bool:
278 return self.transport == "tcp"
280 @property
281 def ready(self) -> t.Union[CFuture, Future]:
282 """A future that resolves when the kernel process has started for the first time"""
283 if not self._ready:
284 self._ready = _get_future()
285 return self._ready
287 @property
288 def ipykernel(self) -> bool:
289 return self.kernel_name in {"python", "python2", "python3"}
291 # Protected traits
292 _launch_args: t.Optional["Dict[str, Any]"] = Dict(allow_none=True)
293 _control_socket: Any = Any()
295 _restarter: Any = Any()
297 autorestart: Bool = Bool(
298 True, config=True, help="""Should we autorestart the kernel if it dies."""
299 )
301 shutting_down: bool = False
303 def __del__(self) -> None:
304 self._close_control_socket()
305 self.cleanup_connection_file()
307 # --------------------------------------------------------------------------
308 # Kernel restarter
309 # --------------------------------------------------------------------------
311 def start_restarter(self) -> None:
312 """Start the kernel restarter."""
313 pass
315 def stop_restarter(self) -> None:
316 """Stop the kernel restarter."""
317 pass
319 def add_restart_callback(self, callback: t.Callable, event: str = "restart") -> None:
320 """Register a callback to be called when a kernel is restarted"""
321 if self._restarter is None:
322 return
323 self._restarter.add_callback(callback, event)
325 def remove_restart_callback(self, callback: t.Callable, event: str = "restart") -> None:
326 """Unregister a callback to be called when a kernel is restarted"""
327 if self._restarter is None:
328 return
329 self._restarter.remove_callback(callback, event)
331 # --------------------------------------------------------------------------
332 # create a Client connected to our Kernel
333 # --------------------------------------------------------------------------
335 def client(self, **kwargs: t.Any) -> BlockingKernelClient:
336 """Create a client configured to connect to our kernel"""
337 kw: dict = {}
338 kw.update(self.get_connection_info(session=True))
339 kw.update(
340 {
341 "connection_file": self.connection_file,
342 "parent": self,
343 }
344 )
346 # add kwargs last, for manual overrides
347 kw.update(kwargs)
348 return self.client_factory(**kw)
350 # --------------------------------------------------------------------------
351 # Kernel management
352 # --------------------------------------------------------------------------
354 def resolve_path(self, path: str) -> str | None:
355 """Resolve path to given file."""
356 assert self.provisioner is not None
357 return self.provisioner.resolve_path(path)
359 def update_env(self, *, env: t.Dict[str, str]) -> None:
360 """
361 Allow to update the environment of a kernel manager.
363 This will take effect only after kernel restart when the new env is
364 passed to the new kernel.
366 This is useful as some of the information of the current kernel reflect
367 the state of the session that started it, and those session information
368 (like the attach file path, or name), are mutable.
370 .. version-added: 8.5
371 """
372 # Mypy think this is unreachable as it see _launch_args as Dict, not t.Dict
373 if (
374 isinstance(self._launch_args, dict)
375 and "env" in self._launch_args
376 and isinstance(self._launch_args["env"], dict) # type: ignore [unreachable]
377 ):
378 self._launch_args["env"].update(env) # type: ignore [unreachable]
380 def format_kernel_cmd(self, extra_arguments: t.List[str] | None = None) -> t.List[str]:
381 """Replace templated args (e.g. {connection_file})"""
382 extra_arguments = extra_arguments or []
383 assert self.kernel_spec is not None
384 cmd = self.kernel_spec.argv + extra_arguments
386 if cmd and cmd[0] in {
387 "python",
388 "python%i" % sys.version_info[0],
389 "python%i.%i" % sys.version_info[:2],
390 }:
391 # executable is 'python' or 'python3', use sys.executable.
392 # These will typically be the same,
393 # but if the current process is in an env
394 # and has been launched by abspath without
395 # activating the env, python on PATH may not be sys.executable,
396 # but it should be.
397 cmd[0] = sys.executable
399 # Make sure to use the realpath for the connection_file
400 # On windows, when running with the store python, the connection_file path
401 # is not usable by non python kernels because the path is being rerouted when
402 # inside of a store app.
403 # See this bug here: https://bugs.python.org/issue41196
404 ns: t.Dict[str, t.Any] = {
405 "connection_file": os.path.realpath(self.connection_file),
406 "prefix": sys.prefix,
407 }
409 if self.kernel_spec: # type:ignore[truthy-bool]
410 ns["resource_dir"] = self.kernel_spec.resource_dir
411 assert isinstance(self._launch_args, dict)
413 ns.update(self._launch_args)
415 pat = re.compile(r"\{([A-Za-z0-9_]+)\}")
417 def from_ns(match: t.Any) -> t.Any:
418 """Get the key out of ns if it's there, otherwise no change."""
419 return ns.get(match.group(1), match.group())
421 return [pat.sub(from_ns, arg) for arg in cmd]
423 async def _async_launch_kernel(self, kernel_cmd: t.List[str], **kw: t.Any) -> None:
424 """actually launch the kernel
426 override in a subclass to launch kernel subprocesses differently
427 Note that provisioners can now be used to customize kernel environments
428 and
429 """
430 assert self.provisioner is not None
431 connection_info = await self.provisioner.launch_kernel(kernel_cmd, **kw)
432 assert self.provisioner.has_process
433 # Provisioner provides the connection information. Load into kernel manager
434 # and write the connection file, if not already done.
435 self._reconcile_connection_info(connection_info)
437 _launch_kernel = run_sync(_async_launch_kernel)
439 # Control socket used for polite kernel shutdown
441 def _connect_control_socket(self) -> None:
442 if self._control_socket is None:
443 self._control_socket = self._create_connected_socket("control")
444 self._control_socket.linger = 100
446 def _close_control_socket(self) -> None:
447 if self._control_socket is None:
448 return
449 self._control_socket.close()
450 self._control_socket = None
452 async def _async_pre_start_kernel(
453 self, *, transport_encryption: str | None = None, **kw: t.Any
454 ) -> t.Tuple[t.List[str], t.Dict[str, t.Any]]:
455 """Prepares a kernel for startup in a separate process.
457 If random ports (port=0) are being used, this method must be called
458 before the channels are created.
460 Parameters
461 ----------
462 `**kw` : optional
463 keyword arguments that are passed down to build the kernel_cmd
464 and launching the kernel (e.g. Popen kwargs).
465 """
466 self.shutting_down = False
467 if transport_encryption is not None:
468 self.transport_encryption = self._transport_encryption_policy(transport_encryption)
469 self.kernel_id = self.kernel_id or kw.pop("kernel_id", str(uuid.uuid4()))
470 # save kwargs for use in restart
471 # assigning Traitlets Dicts to Dict make mypy unhappy but is ok
472 self._launch_args = kw.copy()
473 if (
474 self._transport_encryption_policy() == "required"
475 and not self._kernel_supports_curve_encryption()
476 ):
477 msg = (
478 "transport_encryption='required' but kernelspec does not declare "
479 "metadata.supported_encryption='curve'."
480 )
481 raise RuntimeError(msg)
482 if self.provisioner is None: # will not be None on restarts
483 self.provisioner = KPF.instance(parent=self.parent).create_provisioner_instance(
484 self.kernel_id,
485 self.kernel_spec,
486 parent=self,
487 )
488 kw = await self.provisioner.pre_launch(**kw)
489 kernel_cmd = kw.pop("cmd")
490 return kernel_cmd, kw
492 pre_start_kernel = run_sync(_async_pre_start_kernel)
494 async def _async_post_start_kernel(self, **kw: t.Any) -> None:
495 """Performs any post startup tasks relative to the kernel.
497 Parameters
498 ----------
499 `**kw` : optional
500 keyword arguments that were used in the kernel process's launch.
501 """
502 self.start_restarter()
503 self._connect_control_socket()
504 assert self.provisioner is not None
505 await self.provisioner.post_launch(**kw)
507 post_start_kernel = run_sync(_async_post_start_kernel)
509 @in_pending_state
510 async def _async_start_kernel(self, **kw: t.Any) -> None:
511 """Starts a kernel on this host in a separate process.
513 If random ports (port=0) are being used, this method must be called
514 before the channels are created.
516 Parameters
517 ----------
518 `**kw` : optional
519 keyword arguments that are passed down to build the kernel_cmd
520 and launching the kernel (e.g. Popen kwargs).
521 """
522 self._attempted_start = True
523 kernel_cmd, kw = await self._async_pre_start_kernel(**kw)
525 # launch the kernel subprocess
526 self.log.debug("Starting kernel: %s", kernel_cmd)
527 await self._async_launch_kernel(kernel_cmd, **kw)
528 await self._async_post_start_kernel(**kw)
530 start_kernel = run_sync(_async_start_kernel)
532 async def _async_request_shutdown(self, restart: bool = False) -> None:
533 """Send a shutdown request via control channel"""
534 content = {"restart": restart}
535 msg = self.session.msg("shutdown_request", content=content)
536 # ensure control socket is connected
537 self._connect_control_socket()
538 self.session.send(self._control_socket, msg)
539 assert self.provisioner is not None
540 await self.provisioner.shutdown_requested(restart=restart)
541 self._shutdown_status = _ShutdownStatus.ShutdownRequest
543 request_shutdown = run_sync(_async_request_shutdown)
545 async def _async_finish_shutdown(
546 self,
547 waittime: float | None = None,
548 pollinterval: float = 0.1,
549 restart: bool = False,
550 ) -> None:
551 """Wait for kernel shutdown, then kill process if it doesn't shutdown.
553 This does not send shutdown requests - use :meth:`request_shutdown`
554 first.
555 """
556 if waittime is None:
557 waittime = max(self.shutdown_wait_time, 0)
558 if self.provisioner: # Allow provisioner to override
559 waittime = self.provisioner.get_shutdown_wait_time(recommended=waittime)
561 try:
562 await asyncio.wait_for(
563 self._async_wait(pollinterval=pollinterval), timeout=waittime / 2
564 )
565 except asyncio.TimeoutError:
566 self.log.debug("Kernel is taking too long to finish, terminating")
567 self._shutdown_status = _ShutdownStatus.SigtermRequest
568 await self._async_send_kernel_sigterm()
570 try:
571 await asyncio.wait_for(
572 self._async_wait(pollinterval=pollinterval), timeout=waittime / 2
573 )
574 except asyncio.TimeoutError:
575 self.log.debug("Kernel is taking too long to finish, killing")
576 self._shutdown_status = _ShutdownStatus.SigkillRequest
577 await self._async_kill_kernel(restart=restart)
578 else:
579 # Process is no longer alive, wait and clear
580 if self.has_kernel:
581 assert self.provisioner is not None
582 await self.provisioner.wait()
584 finish_shutdown = run_sync(_async_finish_shutdown)
586 async def _async_cleanup_resources(self, restart: bool = False) -> None:
587 """Clean up resources when the kernel is shut down"""
588 if not restart:
589 self.cleanup_connection_file()
591 self.cleanup_ipc_files()
592 self._close_control_socket()
593 self.session.parent = None
595 if self._created_context and not restart:
596 self.context.destroy(linger=100)
598 if self.provisioner:
599 await self.provisioner.cleanup(restart=restart)
601 cleanup_resources = run_sync(_async_cleanup_resources)
603 @in_pending_state
604 async def _async_shutdown_kernel(self, now: bool = False, restart: bool = False) -> None:
605 """Attempts to stop the kernel process cleanly.
607 This attempts to shutdown the kernels cleanly by:
609 1. Sending it a shutdown message over the control channel.
610 2. If that fails, the kernel is shutdown forcibly by sending it
611 a signal.
613 Parameters
614 ----------
615 now : bool
616 Should the kernel be forcible killed *now*. This skips the
617 first, nice shutdown attempt.
618 restart: bool
619 Will this kernel be restarted after it is shutdown. When this
620 is True, connection files will not be cleaned up.
621 """
622 if not self.owns_kernel:
623 return
625 self.shutting_down = True # Used by restarter to prevent race condition
626 # Stop monitoring for restarting while we shutdown.
627 self.stop_restarter()
629 if self.has_kernel:
630 await self._async_interrupt_kernel()
632 if now:
633 await self._async_kill_kernel()
634 else:
635 await self._async_request_shutdown(restart=restart)
636 # Don't send any additional kernel kill messages immediately, to give
637 # the kernel a chance to properly execute shutdown actions. Wait for at
638 # most 1s, checking every 0.1s.
639 await self._async_finish_shutdown(restart=restart)
641 await self._async_cleanup_resources(restart=restart)
643 shutdown_kernel = run_sync(_async_shutdown_kernel)
645 async def _async_restart_kernel(
646 self, now: bool = False, newports: bool = False, **kw: t.Any
647 ) -> None:
648 """Restarts a kernel with the arguments that were used to launch it.
650 Parameters
651 ----------
652 now : bool, optional
653 If True, the kernel is forcefully restarted *immediately*, without
654 having a chance to do any cleanup action. Otherwise the kernel is
655 given 1s to clean up before a forceful restart is issued.
657 In all cases the kernel is restarted, the only difference is whether
658 it is given a chance to perform a clean shutdown or not.
660 newports : bool, optional
661 If the old kernel was launched with random ports, this flag decides
662 whether the same ports and connection file will be used again.
663 If False, the same ports and connection file are used. This is
664 the default. If True, new random port numbers are chosen and a
665 new connection file is written. It is still possible that the newly
666 chosen random port numbers happen to be the same as the old ones.
668 `**kw` : optional
669 Any options specified here will overwrite those used to launch the
670 kernel.
671 """
672 if self._launch_args is None:
673 msg = "Cannot restart the kernel. No previous call to 'start_kernel'."
674 raise RuntimeError(msg)
676 # Stop currently running kernel.
677 await self._async_shutdown_kernel(now=now, restart=True)
679 if newports:
680 self.cleanup_random_ports()
682 # Start new kernel.
683 self._launch_args.update(kw)
684 await self._async_start_kernel(**self._launch_args)
686 restart_kernel = run_sync(_async_restart_kernel)
688 @property
689 def owns_kernel(self) -> bool:
690 return self._owns_kernel
692 @property
693 def has_kernel(self) -> bool:
694 """Has a kernel process been started that we are actively managing."""
695 return self.provisioner is not None and self.provisioner.has_process
697 async def _async_send_kernel_sigterm(self, restart: bool = False) -> None:
698 """similar to _kill_kernel, but with sigterm (not sigkill), but do not block"""
699 if self.has_kernel:
700 assert self.provisioner is not None
701 await self.provisioner.terminate(restart=restart)
703 _send_kernel_sigterm = run_sync(_async_send_kernel_sigterm)
705 async def _async_kill_kernel(self, restart: bool = False) -> None:
706 """Kill the running kernel.
708 This is a private method, callers should use shutdown_kernel(now=True).
709 """
710 if self.has_kernel:
711 assert self.provisioner is not None
712 await self.provisioner.kill(restart=restart)
714 # Wait until the kernel terminates.
715 try:
716 await asyncio.wait_for(self._async_wait(), timeout=5.0)
717 except asyncio.TimeoutError:
718 # Wait timed out, just log warning but continue - not much more we can do.
719 self.log.warning("Wait for final termination of kernel timed out - continuing...")
720 pass
721 else:
722 # Process is no longer alive, wait and clear
723 if self.has_kernel:
724 await self.provisioner.wait()
726 _kill_kernel = run_sync(_async_kill_kernel)
728 async def _async_interrupt_kernel(self) -> None:
729 """Interrupts the kernel by sending it a signal.
731 Unlike ``signal_kernel``, this operation is well supported on all
732 platforms.
733 """
734 if not self.has_kernel and self._ready is not None:
735 if isinstance(self._ready, CFuture):
736 ready = asyncio.ensure_future(t.cast(Future[t.Any], self._ready))
737 else:
738 ready = self._ready
739 # Wait for a shutdown if one is in progress.
740 if self.shutting_down:
741 await ready
742 # Wait for a startup.
743 await ready
745 if self.has_kernel:
746 assert self.kernel_spec is not None
747 interrupt_mode = self.kernel_spec.interrupt_mode
748 if interrupt_mode == "signal":
749 await self._async_signal_kernel(signal.SIGINT)
751 elif interrupt_mode == "message":
752 msg = self.session.msg("interrupt_request", content={})
753 self._connect_control_socket()
754 self.session.send(self._control_socket, msg)
755 else:
756 msg = "Cannot interrupt kernel. No kernel is running!"
757 raise RuntimeError(msg)
759 interrupt_kernel = run_sync(_async_interrupt_kernel)
761 async def _async_signal_kernel(self, signum: int) -> None:
762 """Sends a signal to the process group of the kernel (this
763 usually includes the kernel and any subprocesses spawned by
764 the kernel).
766 Note that since only SIGTERM is supported on Windows, this function is
767 only useful on Unix systems.
768 """
769 if self.has_kernel:
770 assert self.provisioner is not None
771 await self.provisioner.send_signal(signum)
772 else:
773 msg = "Cannot signal kernel. No kernel is running!"
774 raise RuntimeError(msg)
776 signal_kernel = run_sync(_async_signal_kernel)
778 async def _async_is_alive(self) -> bool:
779 """Is the kernel process still running?"""
780 if not self.owns_kernel:
781 return True
783 if self.has_kernel:
784 assert self.provisioner is not None
785 ret = await self.provisioner.poll()
786 if ret is None:
787 return True
788 return False
790 is_alive = run_sync(_async_is_alive)
792 async def _async_wait(self, pollinterval: float = 0.1) -> None:
793 # Use busy loop at 100ms intervals, polling until the process is
794 # not alive. If we find the process is no longer alive, complete
795 # its cleanup via the blocking wait(). Callers are responsible for
796 # issuing calls to wait() using a timeout (see _kill_kernel()).
797 while await self._async_is_alive():
798 await asyncio.sleep(pollinterval)
801class AsyncKernelManager(KernelManager):
802 """An async kernel manager."""
804 # the class to create with our `client` method
805 client_class: DottedObjectName = DottedObjectName(
806 "jupyter_client.asynchronous.AsyncKernelClient", config=True
807 )
808 client_factory: Type = Type(klass="jupyter_client.asynchronous.AsyncKernelClient", config=True)
810 # The PyZMQ Context to use for communication with the kernel.
811 context: Instance = Instance(zmq.asyncio.Context)
813 @default("context")
814 def _context_default(self) -> zmq.asyncio.Context:
815 self._created_context = True
816 return zmq.asyncio.Context()
818 def client( # type:ignore[override]
819 self, **kwargs: t.Any
820 ) -> AsyncKernelClient:
821 """Get a client for the manager."""
822 return super().client(**kwargs) # type:ignore[return-value]
824 _launch_kernel = KernelManager._async_launch_kernel # type:ignore[assignment]
825 start_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_start_kernel # type:ignore[assignment]
826 pre_start_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_pre_start_kernel # type:ignore[assignment]
827 post_start_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_post_start_kernel # type:ignore[assignment]
828 request_shutdown: t.Callable[..., t.Awaitable] = KernelManager._async_request_shutdown # type:ignore[assignment]
829 finish_shutdown: t.Callable[..., t.Awaitable] = KernelManager._async_finish_shutdown # type:ignore[assignment]
830 cleanup_resources: t.Callable[..., t.Awaitable] = KernelManager._async_cleanup_resources # type:ignore[assignment]
831 shutdown_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_shutdown_kernel # type:ignore[assignment]
832 restart_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_restart_kernel # type:ignore[assignment]
833 _send_kernel_sigterm = KernelManager._async_send_kernel_sigterm # type:ignore[assignment]
834 _kill_kernel = KernelManager._async_kill_kernel # type:ignore[assignment]
835 interrupt_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_interrupt_kernel # type:ignore[assignment]
836 signal_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_signal_kernel # type:ignore[assignment]
837 is_alive: t.Callable[..., t.Awaitable] = KernelManager._async_is_alive # type:ignore[assignment]
840KernelManagerABC.register(KernelManager)
843def start_new_kernel(
844 startup_timeout: float = 60, kernel_name: str = "python", **kwargs: t.Any
845) -> t.Tuple[KernelManager, BlockingKernelClient]:
846 """Start a new kernel, and return its Manager and Client"""
847 km = KernelManager(kernel_name=kernel_name)
848 km.start_kernel(**kwargs)
849 kc = km.client()
850 kc.start_channels()
851 try:
852 kc.wait_for_ready(timeout=startup_timeout)
853 except RuntimeError:
854 kc.stop_channels()
855 km.shutdown_kernel()
856 raise
858 return km, kc
861async def start_new_async_kernel(
862 startup_timeout: float = 60, kernel_name: str = "python", **kwargs: t.Any
863) -> t.Tuple[AsyncKernelManager, AsyncKernelClient]:
864 """Start a new kernel, and return its Manager and Client"""
865 km = AsyncKernelManager(kernel_name=kernel_name)
866 await km.start_kernel(**kwargs)
867 kc = km.client()
868 kc.start_channels()
869 try:
870 await kc.wait_for_ready(timeout=startup_timeout)
871 except RuntimeError:
872 kc.stop_channels()
873 await km.shutdown_kernel()
874 raise
876 return (km, kc)
879@contextmanager
880def run_kernel(**kwargs: t.Any) -> t.Iterator[KernelClient]:
881 """Context manager to create a kernel in a subprocess.
883 The kernel is shut down when the context exits.
885 Returns
886 -------
887 kernel_client: connected KernelClient instance
888 """
889 km, kc = start_new_kernel(**kwargs)
890 try:
891 yield kc
892 finally:
893 kc.stop_channels()
894 km.shutdown_kernel(now=True)