Spin off another thread to handle callbacks. Otherwise we might double lock the GIL.

Author: Yucheng Low

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();
 }
 
@@ -144,12 +158,14 @@ 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){
-          callback(std::string(buf, len));
+          this->log_queue.enqueue(std::string(buf, len));
         });
   }
 }

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,9 +44,9 @@ class unity_server {
   /**
    * Enable or disable log progress stream.
    */
-  static void set_log_progress(bool enable);
+  void set_log_progress(bool enable);
 
-  static void set_log_progress_callback(void (*callback)(std::string));
+  void set_log_progress_callback(progress_callback_type callback);
 
   inline cppipc::comm_server* get_comm_server() { return server; }
 
@@ -61,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,11 +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) {
-  graphlab::unity_server::set_log_progress_callback(reinterpret_cast<void(*)(std::string)>(callback));
+  if (graphlab::SERVER) {
+    graphlab::SERVER->set_log_progress_callback(reinterpret_cast<void(*)(std::string)>(callback));
+  }
 }
 }
  // end of extern "C"