Try to capture stdout and err going directly to 1/2 filedescriptor.

This try to fix a long standing issue that stdout and stderr going
directly to the filedescriptor are not shown in notebooks.

This is annoying when using wrappers around c-libraries, or calling
system commands as those will not be seen from within notebook.

Here we redirect and split the filedescriptor and watch those in threads
and redirect both to the original FD (terminal), and ZMQ (notebook).
Thus output sent to fd 1 & 2 will be shown BOTH in terminal that
launched the notebook server and in notebook themselves.

One of the concern is that now logs and errors internal to ipykernel may
appear in the notebook themselves, so may confuse user; though these
should be limited to error and debug; and we can workaround this by
setting the log handler to not be stdout/err.

This still seem like a big hack to me, and I don't like thread.

I did not manage to make reading the FD non-blocking; so this cannot be
put in the io-thread – at least I'm not sure how. So adds 2 extra
threads to the kernel.

This might need to be turn off by default for now until further testing.

Locally this seem to work with things like:

  - os.system("echo HELLO WORLD")
  - c-extensions writing directly to fd 1 and 2 (when properly flushed).

I have no clue how filedescriptor work on windows, so this only change
behavior on linux and mac.

Author: Carreau

ipykernel/iostream.py

@@ -12,6 +12,7 @@
 import threading
 import warnings
 from weakref import WeakSet
+import traceback
 from io import StringIO, TextIOBase
 
 import zmq
@@ -285,7 +286,44 @@ class OutStream(TextIOBase):
     topic = None
     encoding = 'UTF-8'
 
-    def __init__(self, session, pub_thread, name, pipe=None, echo=None):
+    def _watch_pipe_fd(self):
+        """
+        We've redirected standards steams 0 and 1 into a pipe.
+
+        We need to watch in a thread and redirect them to the right places.
+
+        1) the ZMQ channels to show in notebook interfaces,
+        2) the original stdout/err, to capture errors in terminals.
+
+        We cannot schedule this on the ioloop thread, as this might be blocking.
+
+        """
+
+        try:
+            bts = os.read(self._fid, 1000)
+            while bts and self._should_watch:
+                self.write(bts.decode())
+                os.write(self._original_stdstream_copy, bts)
+                bts = os.read(self._fid, 1000)
+        except Exception:
+            self._exc = sys.exc_info()
+
+    def __init__(
+        self, session, pub_thread, name, pipe=None, echo=None, *, watchfd=True
+    ):
+        """
+        Parameters
+        ----------
+        name : str {'stderr', 'stdout'}
+            the name of the standard stream to replace
+        watchfd : bool (default, True)
+            Watch the file descripttor corresponding to the replaced stream.
+            This is useful if you know some underlying code will write directly
+            the file descriptor by its number. It will spawn a watching thread,
+            that will swap the give file descriptor for a pipe, read from the
+            pipe, and insert this into the current Stream.
+
+        """
         if pipe is not None:
             warnings.warn(
                 "pipe argument to OutStream is deprecated and ignored",
@@ -297,8 +335,12 @@ def __init__(self, session, pub_thread, name, pipe=None, echo=None):
         self.session = session
         if not isinstance(pub_thread, IOPubThread):
             # Backward-compat: given socket, not thread. Wrap in a thread.
-            warnings.warn("OutStream should be created with IOPubThread, not %r" % pub_thread,
-                DeprecationWarning, stacklevel=2)
+            warnings.warn(
+                "Since IPykernel 4.3, OutStream should be created with "
+                "IOPubThread, not %r" % pub_thread,
+                DeprecationWarning,
+                stacklevel=2,
+            )
             pub_thread = IOPubThread(pub_thread)
             pub_thread.start()
         self.pub_thread = pub_thread
@@ -312,19 +354,44 @@ def __init__(self, session, pub_thread, name, pipe=None, echo=None):
         self._new_buffer()
         self.echo = None
 
+        if watchfd and (
+            sys.platform.startswith("linux") or sys.platform.startswith("darwin")
+        ):
+            self._should_watch = True
+            self._setup_stream_redirects(name)
+
         if echo:
             if hasattr(echo, 'read') and hasattr(echo, 'write'):
                 self.echo = echo
             else:
                 raise ValueError("echo argument must be a file like object")
 
+    def _setup_stream_redirects(self, name):
+        pr, pw = os.pipe()
+        fno = getattr(sys, name).fileno()
+        self._original_stdstream_copy = os.dup(fno)
+        os.dup2(pw, fno)
+
+        self._fid = pr
+
+        self._exc = None
+        self.watch_fd_thread = threading.Thread(target=self._watch_pipe_fd)
+        self.watch_fd_thread.daemon = True
+        self.watch_fd_thread.start()
+
     def _is_master_process(self):
         return os.getpid() == self._master_pid
 
     def set_parent(self, parent):
         self.parent_header = extract_header(parent)
 
     def close(self):
+        if sys.platform.startswith("linux") or sys.platform.startswith("darwin"):
+            self._should_watch = False
+            self.watch_fd_thread.join()
+        if self._exc:
+            etype, value, tb = self._exc
+            traceback.print_exception(etype, value, tb)
         self.pub_thread = None
 
     @property

ipykernel/kernelbase.py

@@ -825,9 +825,9 @@ def _send_abort_reply(self, stream, msg, idents):
         """Send a reply to an aborted request"""
         self.log.info("Aborting:")
         self.log.info("%s", msg)
-        reply_type = msg['header']['msg_type'].rsplit('_', 1)[0] + '_reply'
-        status = {'status': 'aborted'}
-        md = {'engine': self.ident}
+        reply_type = msg["header"]["msg_type"].rsplit("_", 1)[0] + "_reply"
+        status = {"status": "aborted"}
+        md = {"engine": self.ident}
         md.update(status)
         self.session.send(
             stream, reply_type, metadata=md,