upstream/ipython Files · IPython/kernel/blocking/client.py

Catch interrupted poll() in terminal console...

Catch interrupted poll() in terminal console Alternative to my own PR - catch ZMQError in run_cell, and if it's caused by an interrupt, ignore it. more complex, especially if we want to handle the timeout nicely as proposed in the comments, but it's possibly also more convenient for other users of that API. Or perhaps not - I'm not sure what makes sense for other API consumers in this case. Fixes gh-8105

Jessica B. Hamrick - - Load All Authors

File last commit:

r19453:3cc62d01


                r20836:0b3b28de

Download file

             client.py
        
                    38 lines
            
             | 1.2 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / IPython / kernel / blocking / client.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      """Implements a fully blocking kernel client.

      Useful for test suites and blocking terminal interfaces.

      """

      # Copyright (c) IPython Development Team.

      # Distributed under the terms of the Modified BSD License.

      try:

          from queue import Empty  # Python 3

      except ImportError:

          from Queue import Empty  # Python 2

      from IPython.utils.traitlets import Type

      from IPython.kernel.channels import HBChannel

      from IPython.kernel.client import KernelClient

      from .channels import ZMQSocketChannel

      class BlockingKernelClient(KernelClient):

          def wait_for_ready(self):

              # Wait for kernel info reply on shell channel

              while True:

                  msg = self.shell_channel.get_msg(block=True)

                  if msg['msg_type'] == 'kernel_info_reply':

                      self._handle_kernel_info_reply(msg)

                      break

              # Flush IOPub channel

              while True:

                  try:

                      msg = self.iopub_channel.get_msg(block=True, timeout=0.2)

                  except Empty:

                      break

          # The classes to use for the various channels

          shell_channel_class = Type(ZMQSocketChannel)

          iopub_channel_class = Type(ZMQSocketChannel)

          stdin_channel_class = Type(ZMQSocketChannel)

          hb_channel_class = Type(HBChannel)

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

				"""Implements a fully blocking kernel client.

				Useful for test suites and blocking terminal interfaces.
				"""
				# Copyright (c) IPython Development Team.
				# Distributed under the terms of the Modified BSD License.

				try:
				from queue import Empty # Python 3
				except ImportError:
				from Queue import Empty # Python 2

				from IPython.utils.traitlets import Type
				from IPython.kernel.channels import HBChannel
				from IPython.kernel.client import KernelClient
				from .channels import ZMQSocketChannel

				class BlockingKernelClient(KernelClient):
				def wait_for_ready(self):
				# Wait for kernel info reply on shell channel
				while True:
				msg = self.shell_channel.get_msg(block=True)
				if msg['msg_type'] == 'kernel_info_reply':
				self._handle_kernel_info_reply(msg)
				break

				# Flush IOPub channel
				while True:
				try:
				msg = self.iopub_channel.get_msg(block=True, timeout=0.2)
				except Empty:
				break

				# The classes to use for the various channels
				shell_channel_class = Type(ZMQSocketChannel)
				iopub_channel_class = Type(ZMQSocketChannel)
				stdin_channel_class = Type(ZMQSocketChannel)
				hb_channel_class = Type(HBChannel)