Release GIL for table(), table.update(), table.remove(), async …

…conflated updates
finos · Sep 21, 2020 · c1bbbf4 · c1bbbf4
1 parent e21b5a1
commit c1bbbf4
Show file tree

Hide file tree

Showing 21 changed files with 312 additions and 370 deletions.
diff --git a/cpp/perspective/CMakeLists.txt b/cpp/perspective/CMakeLists.txt
@@ -514,6 +514,7 @@ set (SOURCE_FILES
 	${PSP_CPP_SRC}/src/cpp/pool.cpp
 	${PSP_CPP_SRC}/src/cpp/port.cpp
 	${PSP_CPP_SRC}/src/cpp/process_state.cpp
+	${PSP_CPP_SRC}/src/cpp/pyutils.cpp
 	${PSP_CPP_SRC}/src/cpp/raii.cpp
 	${PSP_CPP_SRC}/src/cpp/raii_impl_linux.cpp
 	${PSP_CPP_SRC}/src/cpp/raii_impl_osx.cpp

diff --git a/cpp/perspective/src/cpp/gnode.cpp b/cpp/perspective/src/cpp/gnode.cpp
@@ -20,6 +20,10 @@
 #include <perspective/logtime.h>
 #include <perspective/utils.h>
 
+#ifdef PSP_ENABLE_PYTHON
+#include <perspective/pyutils.h>
+#endif
+
 namespace perspective {
 
 t_tscalar
@@ -250,7 +254,7 @@ t_gnode::_process_table(t_uindex port_id) {
         return result;
     }
 
-    std::shared_ptr<t_port> input_port = m_input_ports[port_id];
+    std::shared_ptr<t_port>& input_port = m_input_ports[port_id];
 
     if (input_port->get_table()->size() == 0) {
         return result;
@@ -288,7 +292,7 @@ t_gnode::_process_table(t_uindex port_id) {
         // contexts obliquely read gnode state at various points.
         _update_contexts_from_state(flattened);
 
-        release_inputs();
+        input_port->release();
         release_outputs();
 
     #ifdef PSP_GNODE_VERIFY
@@ -593,6 +597,9 @@ bool
 t_gnode::process(t_uindex port_id) {
     PSP_TRACE_SENTINEL();
     PSP_VERBOSE_ASSERT(m_init, "Cannot `process` on an uninited gnode.");
+#ifdef PSP_ENABLE_PYTHON
+    PerspectiveScopedGILRelease acquire(m_event_loop_thread_id);
+#endif
 
     t_process_table_result result = _process_table(port_id);
 
@@ -1347,6 +1354,13 @@ t_gnode::repr() const {
     return ss.str();
 }
 
+#ifdef PSP_ENABLE_PYTHON
+void 
+t_gnode::set_event_loop_thread_id(std::thread::id id) {
+    m_event_loop_thread_id = id;
+}
+#endif
+
 void
 t_gnode::register_context(const std::string& name, std::shared_ptr<t_ctx0> ctx) {
     _register_context(name, ZERO_SIDED_CONTEXT, reinterpret_cast<std::int64_t>(ctx.get()));

diff --git a/cpp/perspective/src/cpp/pool.cpp b/cpp/perspective/src/cpp/pool.cpp
@@ -13,7 +13,9 @@
 #include <perspective/update_task.h>
 #include <perspective/compat.h>
 #include <perspective/env_vars.h>
+#ifdef PSP_ENABLE_PYTHON
 #include <thread>
+#endif
 #include <chrono>
 
 namespace perspective {
@@ -48,6 +50,7 @@ empty_callback() {
 
 t_pool::t_pool()
     : m_update_delegate(empty_callback())
+    , m_event_loop_thread_id(std::thread::id())
     , m_sleep(0) {
         m_run.clear();
     }
@@ -83,6 +86,11 @@ t_pool::register_gnode(t_gnode* node) {
     t_uindex id = m_gnodes.size() - 1;
     node->set_id(id);
     node->set_pool_cleanup([this, id]() { this->m_gnodes[id] = 0; });
+#ifdef PSP_ENABLE_PYTHON
+    if (m_event_loop_thread_id != std::thread::id()) {
+        node->set_event_loop_thread_id(m_event_loop_thread_id);
+    }
+#endif
 
     if (t_env::log_progress()) {
         std::cout << "t_pool.register_gnode node => " << node << " rv => " << id << std::endl;
@@ -124,27 +132,32 @@ t_pool::send(t_uindex gnode_id, t_uindex port_id, const t_data_table& table) {
     }
 }
 
+#ifdef PSP_ENABLE_PYTHON
+void t_pool::set_event_loop() {
+    m_event_loop_thread_id = std::this_thread::get_id();
+    for (auto node : m_gnodes) {
+        node->set_event_loop_thread_id(m_event_loop_thread_id);
+    }
+}
+
+std::thread::id t_pool::get_event_loop_thread_id() const {
+    return m_event_loop_thread_id;
+}
+#endif
+
 void
-t_pool::_process_helper() {
+t_pool::_process() {
     auto work_to_do = m_data_remaining.load();
     if (work_to_do) {
         t_update_task task(*this);
         task.run();
     }
 }
 
-void
-t_pool::_process() {
-    t_uindex sleep_time = m_sleep.load();
-    _process_helper();
-    std::this_thread::sleep_for(std::chrono::milliseconds(sleep_time));
-}
-
 void
 t_pool::stop() {
     m_run.clear(std::memory_order_release);
-    _process_helper();
-
+    _process();
     if (t_env::log_progress()) {
         std::cout << "t_pool.stop" << std::endl;
     }

diff --git a/cpp/perspective/src/cpp/pyutils.cpp b/cpp/perspective/src/cpp/pyutils.cpp
@@ -8,39 +8,28 @@
  */
 
 #include <perspective/first.h>
+#include <perspective/base.h>
 #include <perspective/pyutils.h>
-#include <frameobject.h>
-#include <iostream>
-#include <perspective/pythonhelpers.h>
-
+#ifdef PSP_ENABLE_PYTHON
 namespace perspective {
 
-bool
-curthread_has_gil() {
-    auto tstate = _PyThreadState_Current;
-    return tstate && (tstate == PyGILState_GetThisThreadState());
-}
-
-void
-print_python_stack() {
-    auto tstate = PyThreadState_GET();
-    if (0 != tstate && 0 != tstate->frame) {
-        auto f = tstate->frame;
-        std::cout << "Python stack trace:" << std::endl;
-        while (0 != f) {
-            auto line = PyCode_Addr2Line(f->f_code, f->f_lasti);
-            auto filename = PyString_AsString(f->f_code->co_filename);
-            auto funcname = PyString_AsString(f->f_code->co_name);
-            std::cout << "\t" << filename << ":" << line << " : " << funcname << std::endl;
-            f = f->f_back;
+PerspectiveScopedGILRelease::PerspectiveScopedGILRelease(std::thread::id event_loop_thread_id) : m_thread_state(NULL) {
+    if (event_loop_thread_id != std::thread::id()) {
+        if (std::this_thread::get_id() != event_loop_thread_id) {
+            std::stringstream err;
+            err << "Perspective called from wrong thread; Expected " << event_loop_thread_id << "; Got " << std::this_thread::get_id() << std::endl;
+            PSP_COMPLAIN_AND_ABORT(err.str());
         }
+        m_thread_state = PyEval_SaveThread();
     }
 }
 
-std::string
-repr(PyObject* pyo) {
-    PyObjectPtr repr(PyObject_Repr(pyo));
-    return std::string(PyString_AsString(repr.get()));
+PerspectiveScopedGILRelease::~PerspectiveScopedGILRelease() {
+     if (m_thread_state != NULL) {
+        PyEval_RestoreThread(m_thread_state);
+    }
 }
 
-} // end namespace perspective
+} // end namespace perspective
+
+#endif
diff --git a/cpp/perspective/src/cpp/view.cpp b/cpp/perspective/src/cpp/view.cpp
@@ -770,6 +770,14 @@ View<CTX_T>::is_column_only() const {
     return m_view_config->is_column_only();
 }
 
+#ifdef PSP_ENABLE_PYTHON
+template <typename CTX_T>
+std::thread::id 
+View<CTX_T>::get_event_loop_thread_id() const {
+    return m_table->get_pool()->get_event_loop_thread_id();
+};
+#endif
+
 /******************************************************************************
  *
  * Private

diff --git a/cpp/perspective/src/include/perspective/gnode.h b/cpp/perspective/src/include/perspective/gnode.h
@@ -25,6 +25,9 @@
 #include <perspective/computed_column_map.h>
 #include <perspective/computed_function.h>
 #include <tsl/ordered_map.h>
+#ifdef PSP_ENABLE_PYTHON
+#include <thread>
+#endif
 #ifdef PSP_PARALLEL_FOR
 #include <tbb/parallel_sort.h>
 #include <tbb/tbb.h>
@@ -190,6 +193,10 @@ class PERSPECTIVE_EXPORT t_gnode {
     void pprint() const;
     std::string repr() const;
 
+#ifdef PSP_ENABLE_PYTHON
+    void set_event_loop_thread_id(std::thread::id id);
+#endif
+
 protected:
     /**
      * @brief Given `tbl`, notify each registered context with `tbl`.
@@ -390,6 +397,10 @@ class PERSPECTIVE_EXPORT t_gnode {
     std::vector<t_custom_column> m_custom_columns;
     std::function<void()> m_pool_cleanup;
     bool m_was_updated;
+
+#ifdef PSP_ENABLE_PYTHON
+    std::thread::id m_event_loop_thread_id;
+#endif
 };
 
 /**

diff --git a/cpp/perspective/src/include/perspective/pool.h b/cpp/perspective/src/include/perspective/pool.h
@@ -15,6 +15,10 @@
 #include <mutex>
 #include <atomic>
 
+#ifdef PSP_ENABLE_PYTHON
+#include <thread>
+#endif
+
 #if defined PSP_ENABLE_WASM
     #include <emscripten/val.h>
     typedef emscripten::val t_val;
@@ -57,6 +61,11 @@ class PERSPECTIVE_EXPORT t_pool {
         t_uindex gnode_id, const std::string& name, t_ctx_type type, std::int64_t ptr);
 #endif
 
+#ifdef PSP_ENABLE_PYTHON
+    void set_event_loop();
+    std::thread::id get_event_loop_thread_id() const;
+#endif
+
     /**
      * @brief Call the binding language's `update_callback` method,
      * set at initialize time.
@@ -74,7 +83,7 @@ class PERSPECTIVE_EXPORT t_pool {
     void send(t_uindex gnode_id, t_uindex port_id, const t_data_table& table);
 
     void _process();
-    void _process_helper();
+
     void init();
     void stop();
     void set_sleep(t_uindex ms);
@@ -102,6 +111,9 @@ class PERSPECTIVE_EXPORT t_pool {
     bool validate_gnode_id(t_uindex gnode_id) const;
 
 private:
+#ifdef PSP_ENABLE_PYTHON
+    std::thread::id m_event_loop_thread_id;
+#endif
     std::mutex m_mtx;
     std::vector<t_gnode*> m_gnodes;
 

diff --git a/cpp/perspective/src/include/perspective/pyutils.h b/cpp/perspective/src/include/perspective/pyutils.h
@@ -0,0 +1,30 @@
+/******************************************************************************
+ *
+ * Copyright (c) 2017, the Perspective Authors.
+ *
+ * This file is part of the Perspective library, distributed under the terms of
+ * the Apache License 2.0.  The full license can be found in the LICENSE file.
+ *
+ */
+
+#pragma once
+#include <perspective/first.h>
+
+#ifdef PSP_ENABLE_PYTHON
+#include <thread>
+
+namespace perspective {
+
+class PerspectiveScopedGILRelease {
+    public:
+        PerspectiveScopedGILRelease(std::thread::id event_loop_thread_id);
+        ~PerspectiveScopedGILRelease();
+    private:
+        PyThreadState* m_thread_state;
+};
+
+
+void _set_event_loop();
+
+} // namespace perspective
+#endif
diff --git a/cpp/perspective/src/include/perspective/view.h b/cpp/perspective/src/include/perspective/view.h
@@ -21,6 +21,9 @@
 #include <cstddef>
 #include <memory>
 #include <map>
+#ifdef PSP_ENABLE_PYTHON
+#include <thread>
+#endif
 
 namespace perspective {
 
@@ -224,6 +227,9 @@ class PERSPECTIVE_EXPORT View {
     t_stepdelta get_step_delta(t_index bidx, t_index eidx) const;
     t_dtype get_column_dtype(t_uindex idx) const;
     bool is_column_only() const;
+#ifdef PSP_ENABLE_PYTHON
+    std::thread::id get_event_loop_thread_id() const;
+#endif
 
 private:
     /**

diff --git a/python/perspective/perspective/include/perspective/python.h b/python/perspective/perspective/include/perspective/python.h
@@ -30,6 +30,7 @@
 #include <perspective/binding.h>
 #include <perspective/exception.h>
 #include <perspective/exports.h>
+#include <perspective/pyutils.h>
 #include <perspective/python/accessor.h>
 #include <perspective/python/base.h>
 #include <perspective/python/computed.h>
@@ -255,6 +256,7 @@ PYBIND11_MODULE(libbinding, m)
         .def(py::init<>())
         .def("set_update_delegate", &t_pool::set_update_delegate)
         .def("unregister_gnode", &t_pool::unregister_gnode)
+        .def("set_event_loop", &t_pool::set_event_loop)
         .def("_process", &t_pool::_process);
 
     /******************************************************************************
@@ -384,7 +386,6 @@ PYBIND11_MODULE(libbinding, m)
     m.def("make_computations", &make_computations);
     m.def("scalar_to_py", &scalar_to_py);
     m.def("_set_nthreads", &_set_nthreads);
-    m.def("_set_event_loop", &_set_event_loop);
 }
 
 #endif
diff --git a/python/perspective/perspective/include/perspective/python/serialization.h b/python/perspective/perspective/include/perspective/python/serialization.h
@@ -11,6 +11,7 @@
 
 #include <perspective/base.h>
 #include <perspective/binding.h>
+#include <perspective/pyutils.h>
 #include <perspective/python/base.h>
 #include <perspective/python/utils.h>
 

diff --git a/python/perspective/perspective/include/perspective/python/table.h b/python/perspective/perspective/include/perspective/python/table.h
@@ -11,6 +11,7 @@
 
 #include <perspective/base.h>
 #include <perspective/binding.h>
+#include <perspective/pyutils.h>
 #include <perspective/python/base.h>
 
 namespace perspective {

diff --git a/python/perspective/perspective/include/perspective/python/utils.h b/python/perspective/perspective/include/perspective/python/utils.h
@@ -19,16 +19,6 @@
 namespace perspective {
 namespace binding {
 
-class PerspectiveScopedGILRelease {
-    public:
-        PerspectiveScopedGILRelease();
-        ~PerspectiveScopedGILRelease();
-    private:
-        PyThreadState* thread_state;
-};
-
-
-void _set_event_loop();
 
 void _set_nthreads(int nthreads);
 

diff --git a/python/perspective/perspective/include/perspective/python/view.h b/python/perspective/perspective/include/perspective/python/view.h
@@ -11,6 +11,7 @@
 
 #include <perspective/base.h>
 #include <perspective/binding.h>
+#include <perspective/pyutils.h>
 #include <perspective/python/base.h>
 #include <perspective/python/utils.h>