Spaces:
Runtime error
Runtime error
File size: 1,562 Bytes
6f72e2b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | """A thread for a shell channel."""
from __future__ import annotations
import asyncio
from threading import current_thread
from typing import Any
import zmq
from .subshell_manager import SubshellManager
from .thread import SHELL_CHANNEL_THREAD_NAME, BaseThread
class ShellChannelThread(BaseThread):
"""A thread for a shell channel.
Communicates with shell/subshell threads via pairs of ZMQ inproc sockets.
"""
def __init__(
self,
context: zmq.Context[Any],
shell_socket: zmq.Socket[Any],
**kwargs,
):
"""Initialize the thread."""
super().__init__(name=SHELL_CHANNEL_THREAD_NAME, **kwargs)
self._manager: SubshellManager | None = None
self._zmq_context = context # Avoid use of self._context
self._shell_socket = shell_socket
# Record the parent thread - the thread that started the app (usually the main thread)
self.parent_thread = current_thread()
self.asyncio_lock = asyncio.Lock()
@property
def manager(self) -> SubshellManager:
# Lazy initialisation.
if self._manager is None:
assert current_thread() == self.parent_thread
self._manager = SubshellManager(
self._zmq_context,
self.io_loop,
self._shell_socket,
)
return self._manager
def run(self) -> None:
"""Run the thread."""
try:
super().run()
finally:
if self._manager:
self._manager.close()
|