Merge pull request #165 from ylow/master

Restored incremental printout in IPython notebook

Author: Jay Gu

oss_src/unity/python/sframe/connect/server.py

@@ -13,6 +13,7 @@
 
 from ..util.config import DEFAULT_CONFIG as default_local_conf
 from ..connect import _get_metric_tracker
+from ..cython.cy_ipc import get_print_status_function_pointer
 from .. import sys_util as _sys_util
 import logging
 import os
@@ -123,7 +124,12 @@ def get_logger(self):
         return self.logger
 
     def set_log_progress(self, enable):
-        self.dll.set_log_progress(enable)
+        if enable:
+            ptr = get_print_status_function_pointer()
+            ptr = cast(ptr, POINTER(c_void_p))
+            self.dll.set_log_progress_callback(ptr)
+        else:
+            self.dll.set_log_progress(enable)
 
     def _load_dll_ok(self, root_path):
         server_env = _sys_util.make_unity_server_env()
@@ -145,6 +151,7 @@ def _load_dll_ok(self, root_path):
             self.dll.start_server.argtypes = [c_char_p, c_char_p, c_char_p, c_ulonglong, c_ulonglong]
             self.dll.get_client.restype = c_void_p
             self.dll.set_log_progress.argtypes = [c_bool]
+            self.dll.set_log_progress_callback.argtypes = [c_void_p]
             self.dll.stop_server
         except Exception as e:
             return (False, str(e))

oss_src/unity/python/sframe/cython/cy_ipc.pxd

@@ -25,3 +25,7 @@ cdef class PyCommClient:
     cdef comm_client *thisptr 
     cpdef stop(self)
     cpdef start(self)
+
+cdef void print_status(string status_string) nogil
+
+cpdef get_print_status_function_pointer()

oss_src/unity/python/sframe/cython/cy_ipc.pyx

@@ -10,7 +10,6 @@ from libcpp.string cimport string
 from libc.stdio cimport printf
 from .python_printer_callback import print_callback
 
-
 cdef class PyCommClient:
 
     def __cinit__(self):
@@ -33,3 +32,13 @@ def make_comm_client_from_existing_ptr(size_t client_ptr):
     ret = PyCommClient()
     ret.thisptr = <comm_client*>(client_ptr)
     return ret
+
+
+cdef void print_status(string status_string) nogil:
+    with gil:
+        status_string = status_string.rstrip()
+        print_callback(status_string)
+
+ctypedef void* void_p
+cpdef get_print_status_function_pointer():
+    return <size_t>(<void_p>(print_status))

oss_src/unity/python/sframe/cython/python_printer_callback.py

@@ -27,7 +27,7 @@ def print_callback(val):
         # I have to intrude rather deep into IPython to make it behave
         if have_ipython:
             if InteractiveShell.initialized():
-                IPython.display.publish_display_data('graphlab.callback', {'text/plain':val,'text/html':'<pre>' + val + '</pre>'})
+                IPython.display.publish_display_data({'text/plain':val,'text/html':'<pre>' + val + '</pre>'})
                 success = True
     except:
         pass

oss_src/unity/server/unity_server.cpp

@@ -79,6 +79,19 @@ void unity_server::start(const unity_server_initializer& server_initializer) {
   server->start();
   logstream(LOG_EMPH) << "Unity server listening on: " <<  options.server_address << std::endl;
   logstream(LOG_EMPH) << "Total System Memory Detected: " << total_mem() << std::endl;
+  log_thread.launch([=]() {
+                      do {
+                        std::pair<std::string, bool> queueelem = this->log_queue.dequeue();
+                        if (queueelem.second == false) {
+                          break;
+                        } else {
+                          // we need to read it before trying to do the callback
+                          // Otherwise we might accidentally call a null pointer
+                          volatile progress_callback_type cback = this->log_progress_callback;
+                          if (cback != nullptr) cback(queueelem.first);
+                        }
+                      } while(1);
+                    });
 }
 
 /**
@@ -88,6 +101,7 @@ void unity_server::stop() {
   delete server;
   server = nullptr;
   set_log_progress(false);
+  log_queue.stop_blocking();
   graphlab::global_teardown::get_instance().perform_teardown();
 }
 
@@ -142,5 +156,18 @@ void unity_server::set_log_progress(bool enable) {
   }
 }
 
+void unity_server::set_log_progress_callback(void (*callback)(std::string)) {
+  if (callback == nullptr) {
+    log_progress_callback = nullptr;
+    global_logger().add_observer(LOG_PROGRESS, NULL);
+  } else {
+    log_progress_callback = callback;
+    global_logger().add_observer(
+        LOG_PROGRESS,
+        [=](int lineloglevel, const char* buf, size_t len){
+          this->log_queue.enqueue(std::string(buf, len));
+        });
+  }
+}
 
 } // end of graphlab

oss_src/unity/server/unity_server.hpp

@@ -8,6 +8,8 @@
 #ifndef SFRAME_UNITY_SERVER_HPP
 #define SFRAME_UNITY_SERVER_HPP
 
+#include <parallel/pthread_tools.hpp>
+#include <util/blocking_queue.hpp>
 #include "unity_server_options.hpp"
 #include "unity_server_init.hpp"
 
@@ -22,6 +24,8 @@ class toolkit_class_registry;
 
 class unity_server {
  public:
+  typedef void(*progress_callback_type)(std::string);
+
   /**
    * Constructor
    */
@@ -40,7 +44,9 @@ class unity_server {
   /**
    * Enable or disable log progress stream.
    */
-  static void set_log_progress(bool enable);
+  void set_log_progress(bool enable);
+
+  void set_log_progress_callback(progress_callback_type callback);
 
   inline cppipc::comm_server* get_comm_server() { return server; }
 
@@ -59,6 +65,12 @@ class unity_server {
   cppipc::comm_server* server;
   toolkit_function_registry* toolkit_functions;
   toolkit_class_registry* toolkit_classes;
+
+ private:
+  volatile progress_callback_type log_progress_callback = nullptr;
+  graphlab::thread log_thread;
+  blocking_queue<std::string> log_queue;
+
 }; // end of class usenity_server
 } // end of namespace graphlab
 

oss_src/unity/server/unity_server_capi.cpp

@@ -99,7 +99,15 @@ EXPORT void stop_server() {
  * Enable or disable log progress stream.
  */
 EXPORT void set_log_progress(bool enable) {
-  graphlab::unity_server::set_log_progress(enable);
+  if (graphlab::SERVER) {
+    graphlab::SERVER->set_log_progress(enable);
+  }
+}
+
+EXPORT void set_log_progress_callback(void* callback) {
+  if (graphlab::SERVER) {
+    graphlab::SERVER->set_log_progress_callback(reinterpret_cast<void(*)(std::string)>(callback));
+  }
 }
 }
  // end of extern "C"