Merge pull request #132 from hoytak/pylambda-cythonization-and-a-hypernewt

Complete removal of boost::python.

Author: TobyRoseman

local_cmake/UseCython.cmake

@@ -250,7 +250,7 @@ function( cython_add_module _name )
   foreach( _file ${ARGN} )
     if( ${_file} MATCHES ".*\\.py[x]?$" )
       list( APPEND pyx_module_sources ${_file} )
-    else()
+    else() 
       list( APPEND other_module_sources ${_file} )
     endif()
   endforeach()

oss_local_scripts/install_python_toolchain.sh

@@ -61,12 +61,20 @@ fi
 if [[ $haspython == 0 ]]; then
         if [[ $OSTYPE == darwin* ]]; then
                 if [ ! -e miniconda.sh ]; then
-                        download_file http://repo.continuum.io/miniconda/Miniconda-latest-MacOSX-x86_64.sh miniconda.sh
+                        if [[ ${PYTHON_VERSION} == "python3.4m" ]]; then
+                                download_file http://repo.continuum.io/miniconda/Miniconda3-3.16.0-MacOSX-x86_64.sh miniconda.sh
+                        else
+                                download_file http://repo.continuum.io/miniconda/Miniconda-latest-MacOSX-x86_64.sh miniconda.sh
+                        fi
                 fi
                 bash ./miniconda.sh -p $PWD/deps/conda -b
         elif [[ "$OSTYPE" == "msys" ]]; then
                 if [ ! -e miniconda.exe ]; then
-                        download_file http://repo.continuum.io/miniconda/Miniconda-latest-Windows-x86_64.exe miniconda.exe
+                        if [[ ${PYTHON_VERSION} == "python3.4m" ]]; then
+                                download_file https://repo.continuum.io/miniconda/Miniconda3-3.16.0-Windows-x86_64.exe miniconda.exe
+                        else
+                                download_file http://repo.continuum.io/miniconda/Miniconda-latest-Windows-x86_64.exe miniconda.exe
+                        fi
                 fi
                 cmd /C "miniconda.exe /S /RegisterPython=0 /AddToPath=0 /D=`cygpath -w $PWD/deps/conda/bin`"
                 mkdir -p $PWD/deps/conda/lib

oss_src/CMakeLists.txt

@@ -32,6 +32,7 @@ subdirs(
   sframe
   sgraph
   luastate
+  python_callbacks
   lambda
   graph
   lazy_eval

oss_src/lambda/CMakeLists.txt

@@ -3,34 +3,25 @@ project(lambda)
 make_library(pylambda
   SOURCES
     lambda_constants.cpp
-    pylambda_function.cpp
     lambda_master.cpp
+    pylambda_function.cpp
     graph_pylambda_master.cpp
     # lualambda_master.cpp
   REQUIRES
-    flexible_type cppipc sframe shmipc
+    flexible_type cppipc sframe shmipc python_callbacks process
     # luastate luajit
     EXTERNAL_VISIBILITY
 )
 
 
-make_library(pylambda_worker_lib
-  SOURCES
-  python_api.cpp
-  python_import_modules.cpp
-  pyflexible_type.cpp
-  pylambda.cpp
-  graph_pylambda.cpp
-  REQUIRES
-    flexible_type cppipc sframe shmipc python_header_only boost_python
-)
-
 make_library(pylambda_worker
   SOURCES
+  pylambda.cpp
   pylambda_worker.cpp
+  graph_pylambda.cpp
   REQUIRES
-    logger process pylambda_worker_lib python_header_only
-  SHARED
+    flexible_type cppipc sframe shmipc python_callbacks process
+    EXTERNAL_VISIBILITY
 )
 
 

oss_src/lambda/graph_pylambda.cpp

@@ -5,22 +5,14 @@
  * This software may be modified and distributed under the terms
  * of the BSD license. See the LICENSE file for details.
  */
-#include <lambda/graph_pylambda.hpp>
-#include <lambda/pyflexible_type.hpp>
-#include <lambda/python_api.hpp>
-#include <lambda/python_thread_guard.hpp>
 #include <lambda/lambda_utils.hpp>
-#include <Python.h>
-#include <boost/python.hpp>
+#include <lambda/graph_pylambda.hpp>
+#include <lambda/pylambda.hpp>
+#include <python_callbacks/python_callbacks.hpp>
 
 namespace graphlab {
 namespace lambda {
 
-namespace python = boost::python;
-
-typedef python::dict py_vertex_object;
-typedef python::dict py_edge_object;
-
 /**************************************************************************/
 /*                                                                        */
 /*                          pysgraph_synchronize                          */
@@ -85,13 +77,13 @@ vertex_partition_exchange pysgraph_synchronize::get_vertex_partition_exchange(si
 /*                                                                        */
 /**************************************************************************/
 graph_pylambda_evaluator::graph_pylambda_evaluator() {
-  python_thread_guard guard;
-  m_current_lambda = new python::object;
 }
 
 graph_pylambda_evaluator::~graph_pylambda_evaluator() {
-  python_thread_guard guard;
-  delete m_current_lambda;
+  if(m_lambda_id != size_t(-1)) {
+    std::cerr << "destructor: Clearing lambda " << m_lambda_id << std::endl; 
+    release_lambda(m_lambda_id); 
+  }
 }
 
 void graph_pylambda_evaluator::init(const std::string& lambda,
@@ -103,7 +95,15 @@ void graph_pylambda_evaluator::init(const std::string& lambda,
   clear();
 
   // initialize members
-  make_lambda(lambda);
+  size_t new_lambda_id = make_lambda(lambda);
+
+  // If it has changed, release the old one.
+  if(m_lambda_id != size_t(-1) && new_lambda_id != m_lambda_id) {
+    release_lambda(m_lambda_id);
+  }
+  
+  m_lambda_id = new_lambda_id; 
+  
   m_vertex_keys = vertex_fields;
   m_edge_keys = edge_fields;
   m_srcid_column = src_column_id;
@@ -117,24 +117,21 @@ void graph_pylambda_evaluator::clear() {
   m_graph_sync.clear();
   m_srcid_column = -1;
   m_dstid_column = -1;
-  *m_current_lambda = python::object();
 }
 
 std::vector<sgraph_edge_data> graph_pylambda_evaluator::eval_triple_apply(
     const std::vector<sgraph_edge_data>& all_edge_data,
     size_t src_partition, size_t dst_partition,
     const std::vector<size_t>& mutated_edge_field_ids) {
 
+  std::lock_guard<mutex> lg(m_mutex);
+  
   logstream(LOG_INFO) << "graph_lambda_worker eval triple apply " << src_partition 
                       << ", " << dst_partition << std::endl;
-  python_thread_guard guard;
+  
   DASSERT_TRUE(is_loaded(src_partition));
   DASSERT_TRUE(is_loaded(dst_partition));
 
-  py_edge_object edge_object;
-  py_vertex_object source_object;
-  py_vertex_object target_object;
-
   auto& source_partition = m_graph_sync.get_partition(src_partition);
   auto& target_partition = m_graph_sync.get_partition(dst_partition);
 
@@ -144,65 +141,23 @@ std::vector<sgraph_edge_data> graph_pylambda_evaluator::eval_triple_apply(
   }
 
   std::vector<sgraph_edge_data> ret(all_edge_data.size());
-  try {
-    size_t cnt = 0;
-    for (const auto& edata: all_edge_data) {
-      PyDict_UpdateFromFlex(edge_object, m_edge_keys, edata);
-      size_t srcid = edata[m_srcid_column];
-      size_t dstid = edata[m_dstid_column];
-
-      auto& source_vertex = source_partition[srcid];
-      auto& target_vertex = target_partition[dstid];
-
-      PyDict_UpdateFromFlex(source_object, m_vertex_keys, source_vertex);
-      PyDict_UpdateFromFlex(target_object, m_vertex_keys, target_vertex);
-
-      python::object lambda_ret = (*m_current_lambda)(source_object, edge_object, target_object);
-      if (lambda_ret.is_none() || !PyTuple_Check(lambda_ret.ptr()) || python::len(lambda_ret) != 3) {
-        throw(std::string("Lambda must return a tuple of the form (source_data, edge_data, target_data)."));
-      }
-
-      for (size_t i = 0; i < m_vertex_keys.size(); ++i) {
-        source_vertex[i] = PyObject_AsFlex(lambda_ret[0][m_vertex_keys[i]]);
-        target_vertex[i] = PyObject_AsFlex(lambda_ret[2][m_vertex_keys[i]]);
-      }
-
-      if (!mutated_edge_field_ids.empty()) {
-        edge_object.update(lambda_ret[1]);
-        for (auto& key: mutated_edge_keys)
-          ret[cnt].push_back(PyObject_AsFlex(edge_object[key]));
-      }
-      ++cnt;
-    }
-  } catch (python::error_already_set const& e) {
-    std::string error_string = parse_python_error();
-    throw(error_string);
-  } catch (std::exception& e) {
-    throw(std::string(e.what()));
-  } catch (const char* e) {
-    throw(e);
-  } catch (std::string& e) {
-    throw(e);
-  } catch (...) {
-    throw("Unknown exception from python lambda evaluation.");
-  }
-  return ret;
-}
 
-void graph_pylambda_evaluator::make_lambda(const std::string& pylambda_str) {
-  python_thread_guard guard;
-  try {
-    python::object pickle = python::import("pickle");
-    PyObject* lambda_bytes = PyByteArray_FromStringAndSize(pylambda_str.c_str(), pylambda_str.size());
-    *m_current_lambda = python::object((pickle.attr("loads")(python::object(python::handle<>(lambda_bytes)))));
-  } catch (python::error_already_set const& e) {
-    std::string error_string = parse_python_error();
-    throw(error_string);
-  } catch (std::exception& e) {
-    throw(std::string(e.what()));
-  } catch (...) {
-    throw("Unknown exception from python lambda evaluation.");
-  }
+  lambda_graph_triple_apply_data lgt;
+
+  lgt.all_edge_data = &all_edge_data;
+  lgt.out_edge_data = &ret;
+  lgt.source_partition = &source_partition;
+  lgt.target_partition = &target_partition;
+  lgt.vertex_keys = &m_vertex_keys;
+  lgt.edge_keys = &m_edge_keys;
+  lgt.mutated_edge_keys = &mutated_edge_keys;
+  lgt.srcid_column = m_srcid_column;
+  lgt.dstid_column = m_dstid_column;
+
+  evaluation_functions.eval_graph_triple_apply(m_lambda_id, &lgt);
+  python::check_for_python_exception();
+  
+  return ret;
 }
 
 }

oss_src/lambda/graph_pylambda.hpp

@@ -7,18 +7,11 @@
  */
 #ifndef GRAPHLAB_LAMBDA_GRAPH_PYLAMBDA_EVALUATOR_HPP
 #define GRAPHLAB_LAMBDA_GRAPH_PYLAMBDA_EVALUATOR_HPP
-#include<lambda/graph_lambda_interface.hpp>
-#include<flexible_type/flexible_type.hpp>
-#include<vector>
-
-// Forward delcaration
-namespace boost {
-  namespace python {
-    namespace api {
-      class object;
-    }
-  }
-}
+#include <lambda/graph_lambda_interface.hpp>
+#include <flexible_type/flexible_type.hpp>
+#include <vector>
+#include <parallel/mutex.hpp>
+#include <atomic>
 
 namespace graphlab {
 
@@ -150,14 +143,15 @@ class graph_pylambda_evaluator : public graph_lambda_evaluator_interface {
                                                   const std::vector<size_t>& mutated_edge_field_ids = {});
 
  private:
-  void make_lambda(const std::string& pylambda_str);
-
+  mutex m_mutex;
+  
+  size_t m_lambda_id = size_t(-1);
+  
   std::vector<std::string> m_vertex_keys;
   std::vector<std::string> m_edge_keys;
   size_t m_srcid_column;
   size_t m_dstid_column;
-  boost::python::api::object* m_current_lambda = NULL;
-
+  
   pysgraph_synchronize m_graph_sync;
 };