From 35390371968e5af17686a01c2bad0529a0f044a2 2014-05-07 20:04:37
From: MinRK <benjaminrk@gmail.com>
Date: 2014-05-07 20:04:37
Subject: [PATCH] Backport PR #5669: allows threadsafe sys.stdout.flush from background threads

allows threadsafe flush from background threads

---

diff --git a/IPython/kernel/zmq/iostream.py b/IPython/kernel/zmq/iostream.py
index fdb4d15..e89c404 100644
--- a/IPython/kernel/zmq/iostream.py
+++ b/IPython/kernel/zmq/iostream.py
@@ -1,12 +1,7 @@
-"""wrappers for stdout/stderr forwarding over zmq
-"""
+"""Wrappers for forwarding stdout/stderr over zmq"""
 
-#-----------------------------------------------------------------------------
-#  Copyright (C) 2013  The IPython Development Team
-#
-#  Distributed under the terms of the BSD License.  The full license is in
-#  the file COPYING, distributed as part of this software.
-#-----------------------------------------------------------------------------
+# Copyright (c) IPython Development Team.
+# Distributed under the terms of the Modified BSD License.
 
 import os
 import threading
@@ -15,6 +10,7 @@ import uuid
 from io import StringIO, UnsupportedOperation
 
 import zmq
+from zmq.eventloop.ioloop import IOLoop
 
 from .session import extract_header
 
@@ -130,6 +126,17 @@ class OutStream(object):
             else:
                 break
     
+    def _schedule_flush(self):
+        """schedule a flush in the main thread
+        
+        only works with a tornado/pyzmq eventloop running
+        """
+        if IOLoop.initialized():
+            IOLoop.instance().add_callback(self.flush)
+        else:
+            # no async loop, at least force the timer
+            self._start = 0
+    
     def flush(self):
         """trigger actual zmq send"""
         if self.pub_socket is None:
@@ -140,9 +147,9 @@ class OutStream(object):
         if mp_mode != CHILD:
             # we are master
             if not self._is_master_thread():
-                # sub-threads must not trigger flush,
-                # but at least they can force the timer.
-                self._start = 0
+                # sub-threads must not trigger flush directly,
+                # but at least they can schedule an async flush, or force the timer.
+                self._schedule_flush()
                 return
             
             self._flush_from_subprocesses()