# -*- coding: utf-8 -*- # # Copyright © Spyder Project Contributors # Licensed under the terms of the MIT License # (see spyder/__init__.py for details) """ In addition to the remote_call mechanism implemented in CommBase: - Send a message to a debugging kernel """ from contextlib import contextmanager import logging import os import pickle import jupyter_client from qtpy.QtCore import QEventLoop, QObject, QTimer, Signal import zmq from zmq.ssh import tunnel as zmqtunnel from spyder_kernels.comms.commbase import CommBase, CommError from spyder.plugins.ipythonconsole.utils.ssh import openssh_tunnel logger = logging.getLogger(__name__) # Patch jupyter_client to define 'comm' as a socket type jupyter_client.connect.channel_socket_types['comm'] = zmq.DEALER class KernelComm(CommBase, QObject): """ Class with the necessary attributes and methods to handle communications with a console. """ _sig_got_reply = Signal() _sig_comm_port_changed = Signal() sig_exception_occurred = Signal(dict) def __init__(self): super(KernelComm, self).__init__() self.remote_comm_port = None self.kernel_client = None # Register handlers self.register_call_handler('_async_error', self._async_error) self.register_call_handler('_set_comm_port', self._set_comm_port) def ssh_tunnel(self, *args, **kwargs): if os.name == 'nt': return zmqtunnel.paramiko_tunnel(*args, **kwargs) else: return openssh_tunnel(None, *args, **kwargs) def _set_comm_port(self, port): """Set comm port.""" if port is None or port == self.remote_comm_port: return self.remote_comm_port = port client = self.kernel_client if hasattr(client, 'ssh_parameters'): # Need to tunnel port hostname, sshkey, password = client.ssh_parameters local_port = zmqtunnel.select_random_ports(1)[0] remote_port = port remote_ip = client.ip self.ssh_tunnel( local_port, remote_port, hostname, remote_ip, sshkey, password, timeout=10) port = local_port if not (hasattr(client, 'comm_port') and client.comm_port == port): client.comm_port = port identity = client.session.bsession socket = client._create_connected_socket( 'comm', identity=identity) client.comm_channel = client.shell_channel_class( socket, client.session, client.ioloop) # We emit in case we are waiting on this self._sig_comm_port_changed.emit() def shutdown_comm_channel(self): """Shutdown the comm channel.""" # This is necessary to avoid an error when closing the console. # Fixes spyder-ide/spyder#15645 try: channel = self.kernel_client.comm_channel except AttributeError: channel = None if channel: id_list = self.get_comm_id_list() for comm_id in id_list: msg = self.kernel_client.session.msg( 'shutdown_request', {"comm_id": comm_id}) channel.send(msg) self.kernel_client.comm_channel = None def comm_channel_connected(self): """Check if the comm channel is connected.""" return self.kernel_client.comm_channel is not None @contextmanager def comm_channel_manager(self, comm_id, queue_message=False): """Use comm_channel instead of shell_channel.""" if queue_message: # Send without comm_channel yield return if not self.comm_channel_connected(): # Ask again for comm config self.remote_call()._send_comm_config() raise CommError("Comm not connected.") id_list = self.get_comm_id_list(comm_id) for comm_id in id_list: self._comms[comm_id]['comm']._send_channel = ( self.kernel_client.comm_channel) try: yield finally: id_list = self.get_comm_id_list(comm_id) for comm_id in id_list: self._comms[comm_id]['comm']._send_channel = ( self.kernel_client.shell_channel) def _set_call_return_value(self, call_dict, data, is_error=False): """Override to use the comm_channel for all replies.""" # Avoid crash if comm channel not connected queue_message = not self.comm_channel_connected() with self.comm_channel_manager(self.calling_comm_id, queue_message): super(KernelComm, self)._set_call_return_value( call_dict, data, is_error) def remove(self, comm_id=None): """ Remove the comm without notifying the other side. Use when the other side is already down. """ id_list = self.get_comm_id_list(comm_id) for comm_id in id_list: del self._comms[comm_id] def close(self, comm_id=None, shutdown_channel=True): """Ask kernel to close comm and send confirmation.""" if shutdown_channel: # Only shutdown the channel if the kernel closes as well self.shutdown_comm_channel() id_list = self.get_comm_id_list(comm_id) for comm_id in id_list: # Send comm_close directly to avoid really closing the comm self._comms[comm_id]['comm']._send_msg( 'comm_close', {}, None, None, None) def open_comm(self, kernel_client): """Open comm through the kernel client.""" self.kernel_client = kernel_client self.kernel_client.comm_channel = None try: self._register_comm( # Create new comm and send the highest protocol kernel_client.comm_manager.new_comm(self._comm_name, data={ 'pickle_protocol': pickle.HIGHEST_PROTOCOL})) except AttributeError: logger.info( "Unable to open comm due to unexistent comm manager: " + "kernel_client.comm_manager=" + str(kernel_client.comm_manager) ) def remote_call(self, interrupt=False, blocking=False, callback=None, comm_id=None, timeout=None, display_error=False): """Get a handler for remote calls.""" return super(KernelComm, self).remote_call( interrupt=interrupt, blocking=blocking, callback=callback, comm_id=comm_id, timeout=timeout, display_error=display_error) # ---- Private ----- def on_incoming_call(self, call_dict): """A call was received""" if "comm_port" in call_dict: self._set_comm_port(call_dict["comm_port"]) return super(KernelComm, self).on_incoming_call(call_dict) def _get_call_return_value(self, call_dict, call_data, comm_id): """ Interupt the kernel if needed. """ settings = call_dict['settings'] blocking = 'blocking' in settings and settings['blocking'] if not self.kernel_client.is_alive(): if blocking: raise RuntimeError("Kernel is dead") else: # The user has other problems logger.info( "Dropping message because kernel is dead: %s", str(call_dict) ) return settings = call_dict['settings'] interrupt = 'interrupt' in settings and settings['interrupt'] interrupt = interrupt or blocking # Need to make sure any blocking call is replied rapidly. if interrupt and not self.comm_channel_connected(): # Ask again for comm config self.remote_call()._send_comm_config() # Can not interrupt if comm not connected interrupt = False logger.debug( "Dropping interrupt because comm is disconnected: " + str(call_dict) ) if blocking: raise CommError("Cannot block on a disconnected comm") try: with self.comm_channel_manager( comm_id, queue_message=not interrupt): return super(KernelComm, self)._get_call_return_value( call_dict, call_data, comm_id) except RuntimeError as e: if blocking: raise else: # The user has other problems logger.info( "Dropping message because of exception: ", str(e), str(call_dict) ) return def _wait_reply(self, call_id, call_name, timeout): """Wait for the other side reply.""" def got_reply(): return call_id in self._reply_inbox timeout_msg = "Timeout while waiting for {}".format( self._reply_waitlist) self._wait(got_reply, self._sig_got_reply, timeout_msg, timeout) def _wait(self, condition, signal, timeout_msg, timeout): """ Wait until condition() is True by running an event loop. signal: qt signal that should interrupt the event loop. timeout_msg: Message to display in case of a timeout. timeout: time in seconds before a timeout """ # Exit if condition is fulfilled or the kernel is dead. if condition(): return if not self.kernel_client.is_alive(): raise RuntimeError("Kernel is dead") # Create event loop to wait with wait_loop = QEventLoop(None) wait_timeout = QTimer(self) wait_timeout.setSingleShot(True) # Connect signals to stop kernel loop wait_timeout.timeout.connect(wait_loop.quit) self.kernel_client.hb_channel.kernel_died.connect(wait_loop.quit) signal.connect(wait_loop.quit) # Wait until the kernel returns the value wait_timeout.start(timeout * 1000) while not condition(): if not wait_timeout.isActive(): signal.disconnect(wait_loop.quit) self.kernel_client.hb_channel.kernel_died.disconnect( wait_loop.quit) if condition(): return if not self.kernel_client.is_alive(): raise RuntimeError("Kernel is dead") raise TimeoutError(timeout_msg) wait_loop.exec_() wait_timeout.stop() signal.disconnect(wait_loop.quit) self.kernel_client.hb_channel.kernel_died.disconnect( wait_loop.quit) def _handle_remote_call_reply(self, msg_dict, buffer): """ A blocking call received a reply. """ super(KernelComm, self)._handle_remote_call_reply( msg_dict, buffer) self._sig_got_reply.emit() def _async_error(self, error_wrapper): """ Handle an error that was raised on the other side and sent back. """ for line in error_wrapper.format_error(): self.sig_exception_occurred.emit( dict(text=line, is_traceback=True) )