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)
349 for key in ("curve_publickey", "curve_secretkey"):
350 if isinstance(kw.get(key), str):
351 kw[key] = kw[key].encode("ascii")
353 return self.client_factory(**kw)
355 # --------------------------------------------------------------------------
356 # Kernel management
357 # --------------------------------------------------------------------------
359 def resolve_path(self, path: str) -> str | None:
360 """Resolve path to given file."""
361 assert self.provisioner is not None
362 return self.provisioner.resolve_path(path)
364 def update_env(self, *, env: t.Dict[str, str]) -> None:
365 """
366 Allow to update the environment of a kernel manager.
368 This will take effect only after kernel restart when the new env is
369 passed to the new kernel.
371 This is useful as some of the information of the current kernel reflect
372 the state of the session that started it, and those session information
373 (like the attach file path, or name), are mutable.
375 .. version-added: 8.5
376 """
377 # Mypy think this is unreachable as it see _launch_args as Dict, not t.Dict
378 if (
379 isinstance(self._launch_args, dict)
380 and "env" in self._launch_args
381 and isinstance(self._launch_args["env"], dict) # type: ignore [unreachable]
382 ):
383 self._launch_args["env"].update(env) # type: ignore [unreachable]
385 def format_kernel_cmd(self, extra_arguments: t.List[str] | None = None) -> t.List[str]:
386 """Replace templated args (e.g. {connection_file})"""
387 extra_arguments = extra_arguments or []
388 assert self.kernel_spec is not None
389 cmd = self.kernel_spec.argv + extra_arguments
391 if cmd and cmd[0] in {
392 "python",
393 "python%i" % sys.version_info[0],
394 "python%i.%i" % sys.version_info[:2],
395 }:
396 # executable is 'python' or 'python3', use sys.executable.
397 # These will typically be the same,
398 # but if the current process is in an env
399 # and has been launched by abspath without
400 # activating the env, python on PATH may not be sys.executable,
401 # but it should be.
402 cmd[0] = sys.executable
404 # Make sure to use the realpath for the connection_file
405 # On windows, when running with the store python, the connection_file path
406 # is not usable by non python kernels because the path is being rerouted when
407 # inside of a store app.
408 # See this bug here: https://bugs.python.org/issue41196
409 ns: t.Dict[str, t.Any] = {
410 "connection_file": os.path.realpath(self.connection_file),
411 "prefix": sys.prefix,
412 }
414 if self.kernel_spec: # type:ignore[truthy-bool]
415 ns["resource_dir"] = self.kernel_spec.resource_dir
416 assert isinstance(self._launch_args, dict)
418 ns.update(self._launch_args)
420 pat = re.compile(r"\{([A-Za-z0-9_]+)\}")
422 def from_ns(match: t.Any) -> t.Any:
423 """Get the key out of ns if it's there, otherwise no change."""
424 return ns.get(match.group(1), match.group())
426 return [pat.sub(from_ns, arg) for arg in cmd]
428 async def _async_launch_kernel(self, kernel_cmd: t.List[str], **kw: t.Any) -> None:
429 """actually launch the kernel
431 override in a subclass to launch kernel subprocesses differently
432 Note that provisioners can now be used to customize kernel environments
433 and
434 """
435 assert self.provisioner is not None
436 connection_info = await self.provisioner.launch_kernel(kernel_cmd, **kw)
437 assert self.provisioner.has_process
438 # Provisioner provides the connection information. Load into kernel manager
439 # and write the connection file, if not already done.
440 self._reconcile_connection_info(connection_info)
442 _launch_kernel = run_sync(_async_launch_kernel)
444 # Control socket used for polite kernel shutdown
446 def _connect_control_socket(self) -> None:
447 if self._control_socket is None:
448 self._control_socket = self._create_connected_socket("control")
449 self._control_socket.linger = 100
451 def _close_control_socket(self) -> None:
452 if self._control_socket is None:
453 return
454 self._control_socket.close()
455 self._control_socket = None
457 async def _async_pre_start_kernel(
458 self, *, transport_encryption: str | None = None, **kw: t.Any
459 ) -> t.Tuple[t.List[str], t.Dict[str, t.Any]]:
460 """Prepares a kernel for startup in a separate process.
462 If random ports (port=0) are being used, this method must be called
463 before the channels are created.
465 Parameters
466 ----------
467 `**kw` : optional
468 keyword arguments that are passed down to build the kernel_cmd
469 and launching the kernel (e.g. Popen kwargs).
470 """
471 self.shutting_down = False
472 if transport_encryption is not None:
473 self.transport_encryption = self._transport_encryption_policy(transport_encryption)
474 self.kernel_id = self.kernel_id or kw.pop("kernel_id", str(uuid.uuid4()))
475 # save kwargs for use in restart
476 # assigning Traitlets Dicts to Dict make mypy unhappy but is ok
477 self._launch_args = kw.copy()
478 if (
479 self._transport_encryption_policy() == "required"
480 and not self._kernel_supports_curve_encryption()
481 ):
482 msg = (
483 "transport_encryption='required' but kernelspec does not declare "
484 "'curve' in `metadata.supported_encryption`."
485 )
486 raise RuntimeError(msg)
487 if self.provisioner is None: # will not be None on restarts
488 self.provisioner = KPF.instance(parent=self.parent).create_provisioner_instance(
489 self.kernel_id,
490 self.kernel_spec,
491 parent=self,
492 )
493 kw = await self.provisioner.pre_launch(**kw)
494 kernel_cmd = kw.pop("cmd")
495 return kernel_cmd, kw
497 pre_start_kernel = run_sync(_async_pre_start_kernel)
499 async def _async_post_start_kernel(self, **kw: t.Any) -> None:
500 """Performs any post startup tasks relative to the kernel.
502 Parameters
503 ----------
504 `**kw` : optional
505 keyword arguments that were used in the kernel process's launch.
506 """
507 self.start_restarter()
508 self._connect_control_socket()
509 assert self.provisioner is not None
510 await self.provisioner.post_launch(**kw)
512 post_start_kernel = run_sync(_async_post_start_kernel)
514 @in_pending_state
515 async def _async_start_kernel(self, **kw: t.Any) -> None:
516 """Starts a kernel on this host in a separate process.
518 If random ports (port=0) are being used, this method must be called
519 before the channels are created.
521 Parameters
522 ----------
523 `**kw` : optional
524 keyword arguments that are passed down to build the kernel_cmd
525 and launching the kernel (e.g. Popen kwargs).
526 """
527 self._attempted_start = True
528 kernel_cmd, kw = await self._async_pre_start_kernel(**kw)
530 # launch the kernel subprocess
531 self.log.debug("Starting kernel: %s", kernel_cmd)
532 await self._async_launch_kernel(kernel_cmd, **kw)
533 await self._async_post_start_kernel(**kw)
535 start_kernel = run_sync(_async_start_kernel)
537 async def _async_request_shutdown(self, restart: bool = False) -> None:
538 """Send a shutdown request via control channel"""
539 content = {"restart": restart}
540 msg = self.session.msg("shutdown_request", content=content)
541 # ensure control socket is connected
542 self._connect_control_socket()
543 self.session.send(self._control_socket, msg)
544 assert self.provisioner is not None
545 await self.provisioner.shutdown_requested(restart=restart)
546 self._shutdown_status = _ShutdownStatus.ShutdownRequest
548 request_shutdown = run_sync(_async_request_shutdown)
550 async def _async_finish_shutdown(
551 self,
552 waittime: float | None = None,
553 pollinterval: float = 0.1,
554 restart: bool = False,
555 ) -> None:
556 """Wait for kernel shutdown, then kill process if it doesn't shutdown.
558 This does not send shutdown requests - use :meth:`request_shutdown`
559 first.
560 """
561 if waittime is None:
562 waittime = max(self.shutdown_wait_time, 0)
563 if self.provisioner: # Allow provisioner to override
564 waittime = self.provisioner.get_shutdown_wait_time(recommended=waittime)
566 try:
567 await asyncio.wait_for(
568 self._async_wait(pollinterval=pollinterval), timeout=waittime / 2
569 )
570 except asyncio.TimeoutError:
571 self.log.debug("Kernel is taking too long to finish, terminating")
572 self._shutdown_status = _ShutdownStatus.SigtermRequest
573 await self._async_send_kernel_sigterm()
575 try:
576 await asyncio.wait_for(
577 self._async_wait(pollinterval=pollinterval), timeout=waittime / 2
578 )
579 except asyncio.TimeoutError:
580 self.log.debug("Kernel is taking too long to finish, killing")
581 self._shutdown_status = _ShutdownStatus.SigkillRequest
582 await self._async_kill_kernel(restart=restart)
583 else:
584 # Process is no longer alive, wait and clear
585 if self.has_kernel:
586 assert self.provisioner is not None
587 await self.provisioner.wait()
589 finish_shutdown = run_sync(_async_finish_shutdown)
591 async def _async_cleanup_resources(self, restart: bool = False) -> None:
592 """Clean up resources when the kernel is shut down"""
593 if not restart:
594 self.cleanup_connection_file()
596 self.cleanup_ipc_files()
597 self._close_control_socket()
598 self.session.parent = None
600 if self._created_context and not restart:
601 self.context.destroy(linger=100)
603 if self.provisioner:
604 await self.provisioner.cleanup(restart=restart)
606 cleanup_resources = run_sync(_async_cleanup_resources)
608 @in_pending_state
609 async def _async_shutdown_kernel(self, now: bool = False, restart: bool = False) -> None:
610 """Attempts to stop the kernel process cleanly.
612 This attempts to shutdown the kernels cleanly by:
614 1. Sending it a shutdown message over the control channel.
615 2. If that fails, the kernel is shutdown forcibly by sending it
616 a signal.
618 Parameters
619 ----------
620 now : bool
621 Should the kernel be forcible killed *now*. This skips the
622 first, nice shutdown attempt.
623 restart: bool
624 Will this kernel be restarted after it is shutdown. When this
625 is True, connection files will not be cleaned up.
626 """
627 if not self.owns_kernel:
628 return
630 self.shutting_down = True # Used by restarter to prevent race condition
631 # Stop monitoring for restarting while we shutdown.
632 self.stop_restarter()
634 if self.has_kernel:
635 await self._async_interrupt_kernel()
637 if now:
638 await self._async_kill_kernel()
639 else:
640 await self._async_request_shutdown(restart=restart)
641 # Don't send any additional kernel kill messages immediately, to give
642 # the kernel a chance to properly execute shutdown actions. Wait for at
643 # most 1s, checking every 0.1s.
644 await self._async_finish_shutdown(restart=restart)
646 await self._async_cleanup_resources(restart=restart)
648 shutdown_kernel = run_sync(_async_shutdown_kernel)
650 async def _async_restart_kernel(
651 self, now: bool = False, newports: bool = False, **kw: t.Any
652 ) -> None:
653 """Restarts a kernel with the arguments that were used to launch it.
655 Parameters
656 ----------
657 now : bool, optional
658 If True, the kernel is forcefully restarted *immediately*, without
659 having a chance to do any cleanup action. Otherwise the kernel is
660 given 1s to clean up before a forceful restart is issued.
662 In all cases the kernel is restarted, the only difference is whether
663 it is given a chance to perform a clean shutdown or not.
665 newports : bool, optional
666 If the old kernel was launched with random ports, this flag decides
667 whether the same ports and connection file will be used again.
668 If False, the same ports and connection file are used. This is
669 the default. If True, new random port numbers are chosen and a
670 new connection file is written. It is still possible that the newly
671 chosen random port numbers happen to be the same as the old ones.
673 `**kw` : optional
674 Any options specified here will overwrite those used to launch the
675 kernel.
676 """
677 if self._launch_args is None:
678 msg = "Cannot restart the kernel. No previous call to 'start_kernel'."
679 raise RuntimeError(msg)
681 # Stop currently running kernel.
682 await self._async_shutdown_kernel(now=now, restart=True)
684 if newports:
685 self.cleanup_random_ports()
687 # Start new kernel.
688 self._launch_args.update(kw)
689 await self._async_start_kernel(**self._launch_args)
691 restart_kernel = run_sync(_async_restart_kernel)
693 @property
694 def owns_kernel(self) -> bool:
695 return self._owns_kernel
697 @property
698 def has_kernel(self) -> bool:
699 """Has a kernel process been started that we are actively managing."""
700 return self.provisioner is not None and self.provisioner.has_process
702 async def _async_send_kernel_sigterm(self, restart: bool = False) -> None:
703 """similar to _kill_kernel, but with sigterm (not sigkill), but do not block"""
704 if self.has_kernel:
705 assert self.provisioner is not None
706 await self.provisioner.terminate(restart=restart)
708 _send_kernel_sigterm = run_sync(_async_send_kernel_sigterm)
710 async def _async_kill_kernel(self, restart: bool = False) -> None:
711 """Kill the running kernel.
713 This is a private method, callers should use shutdown_kernel(now=True).
714 """
715 if self.has_kernel:
716 assert self.provisioner is not None
717 await self.provisioner.kill(restart=restart)
719 # Wait until the kernel terminates.
720 try:
721 await asyncio.wait_for(self._async_wait(), timeout=5.0)
722 except asyncio.TimeoutError:
723 # Wait timed out, just log warning but continue - not much more we can do.
724 self.log.warning("Wait for final termination of kernel timed out - continuing...")
725 pass
726 else:
727 # Process is no longer alive, wait and clear
728 if self.has_kernel:
729 await self.provisioner.wait()
731 _kill_kernel = run_sync(_async_kill_kernel)
733 async def _async_interrupt_kernel(self) -> None:
734 """Interrupts the kernel by sending it a signal.
736 Unlike ``signal_kernel``, this operation is well supported on all
737 platforms.
738 """
739 if not self.has_kernel and self._ready is not None:
740 if isinstance(self._ready, CFuture):
741 ready = asyncio.ensure_future(t.cast(Future[t.Any], self._ready))
742 else:
743 ready = self._ready
744 # Wait for a shutdown if one is in progress.
745 if self.shutting_down:
746 await ready
747 # Wait for a startup.
748 await ready
750 if self.has_kernel:
751 assert self.kernel_spec is not None
752 interrupt_mode = self.kernel_spec.interrupt_mode
753 if interrupt_mode == "signal":
754 await self._async_signal_kernel(signal.SIGINT)
756 elif interrupt_mode == "message":
757 msg = self.session.msg("interrupt_request", content={})
758 self._connect_control_socket()
759 self.session.send(self._control_socket, msg)
760 else:
761 msg = "Cannot interrupt kernel. No kernel is running!"
762 raise RuntimeError(msg)
764 interrupt_kernel = run_sync(_async_interrupt_kernel)
766 async def _async_signal_kernel(self, signum: int) -> None:
767 """Sends a signal to the process group of the kernel (this
768 usually includes the kernel and any subprocesses spawned by
769 the kernel).
771 Note that since only SIGTERM is supported on Windows, this function is
772 only useful on Unix systems.
773 """
774 if self.has_kernel:
775 assert self.provisioner is not None
776 await self.provisioner.send_signal(signum)
777 else:
778 msg = "Cannot signal kernel. No kernel is running!"
779 raise RuntimeError(msg)
781 signal_kernel = run_sync(_async_signal_kernel)
783 async def _async_is_alive(self) -> bool:
784 """Is the kernel process still running?"""
785 if not self.owns_kernel:
786 return True
788 if self.has_kernel:
789 assert self.provisioner is not None
790 ret = await self.provisioner.poll()
791 if ret is None:
792 return True
793 return False
795 is_alive = run_sync(_async_is_alive)
797 async def _async_wait(self, pollinterval: float = 0.1) -> None:
798 # Use busy loop at 100ms intervals, polling until the process is
799 # not alive. If we find the process is no longer alive, complete
800 # its cleanup via the blocking wait(). Callers are responsible for
801 # issuing calls to wait() using a timeout (see _kill_kernel()).
802 while await self._async_is_alive():
803 await asyncio.sleep(pollinterval)
806class AsyncKernelManager(KernelManager):
807 """An async kernel manager."""
809 # the class to create with our `client` method
810 client_class: DottedObjectName = DottedObjectName(
811 "jupyter_client.asynchronous.AsyncKernelClient", config=True
812 )
813 client_factory: Type = Type(klass="jupyter_client.asynchronous.AsyncKernelClient", config=True)
815 # The PyZMQ Context to use for communication with the kernel.
816 context: Instance = Instance(zmq.asyncio.Context)
818 @default("context")
819 def _context_default(self) -> zmq.asyncio.Context:
820 self._created_context = True
821 return zmq.asyncio.Context()
823 def client( # type:ignore[override]
824 self, **kwargs: t.Any
825 ) -> AsyncKernelClient:
826 """Get a client for the manager."""
827 return super().client(**kwargs) # type:ignore[return-value]
829 _launch_kernel = KernelManager._async_launch_kernel # type:ignore[assignment]
830 start_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_start_kernel # type:ignore[assignment]
831 pre_start_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_pre_start_kernel # type:ignore[assignment]
832 post_start_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_post_start_kernel # type:ignore[assignment]
833 request_shutdown: t.Callable[..., t.Awaitable] = KernelManager._async_request_shutdown # type:ignore[assignment]
834 finish_shutdown: t.Callable[..., t.Awaitable] = KernelManager._async_finish_shutdown # type:ignore[assignment]
835 cleanup_resources: t.Callable[..., t.Awaitable] = KernelManager._async_cleanup_resources # type:ignore[assignment]
836 shutdown_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_shutdown_kernel # type:ignore[assignment]
837 restart_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_restart_kernel # type:ignore[assignment]
838 _send_kernel_sigterm = KernelManager._async_send_kernel_sigterm # type:ignore[assignment]
839 _kill_kernel = KernelManager._async_kill_kernel # type:ignore[assignment]
840 interrupt_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_interrupt_kernel # type:ignore[assignment]
841 signal_kernel: t.Callable[..., t.Awaitable] = KernelManager._async_signal_kernel # type:ignore[assignment]
842 is_alive: t.Callable[..., t.Awaitable] = KernelManager._async_is_alive # type:ignore[assignment]
845KernelManagerABC.register(KernelManager)
848def start_new_kernel(
849 startup_timeout: float = 60, kernel_name: str = "python", **kwargs: t.Any
850) -> t.Tuple[KernelManager, BlockingKernelClient]:
851 """Start a new kernel, and return its Manager and Client"""
852 km = KernelManager(kernel_name=kernel_name)
853 km.start_kernel(**kwargs)
854 kc = km.client()
855 kc.start_channels()
856 try:
857 kc.wait_for_ready(timeout=startup_timeout)
858 except RuntimeError:
859 kc.stop_channels()
860 km.shutdown_kernel()
861 raise
863 return km, kc
866async def start_new_async_kernel(
867 startup_timeout: float = 60, kernel_name: str = "python", **kwargs: t.Any
868) -> t.Tuple[AsyncKernelManager, AsyncKernelClient]:
869 """Start a new kernel, and return its Manager and Client"""
870 km = AsyncKernelManager(kernel_name=kernel_name)
871 await km.start_kernel(**kwargs)
872 kc = km.client()
873 kc.start_channels()
874 try:
875 await kc.wait_for_ready(timeout=startup_timeout)
876 except RuntimeError:
877 kc.stop_channels()
878 await km.shutdown_kernel()
879 raise
881 return (km, kc)
884@contextmanager
885def run_kernel(**kwargs: t.Any) -> t.Iterator[KernelClient]:
886 """Context manager to create a kernel in a subprocess.
888 The kernel is shut down when the context exits.
890 Returns
891 -------
892 kernel_client: connected KernelClient instance
893 """
894 km, kc = start_new_kernel(**kwargs)
895 try:
896 yield kc
897 finally:
898 kc.stop_channels()
899 km.shutdown_kernel(now=True)