feat: ff_send_out_to support, more conversions in pipeline, refactor of nodes

Author: domferr

Makefile

@@ -1,14 +1,14 @@
 .PHONY: clean config repara-copy titanic-copy build
 
-clean:
-rm -rf build
+# run source .venv/bin/activate before running make build
+build: clean
+pip install .
 
 repara-copy:
-rsync -av -e ssh --exclude 'fastflow' --exclude '.git' --exclude '.venv' --exclude '*.egg-info' ./ dferraro@repara.unipi.it:/home/dferraro/fastflow-python
+rsync -av -e ssh --exclude 'fastflow' --exclude '.git' --exclude '.venv' --exclude '*.egg-info' --exclude 'build' ./ dferraro@repara.unipi.it:/home/dferraro/fastflow-python
 
 titanic-copy:
-rsync -av -e ssh --exclude 'fastflow' --exclude '.git' --exclude '.venv' --exclude '*.egg-info' ./ dferraro@titanic.unipi.it:/home/dferraro/fastflow-python
-
-build: clean
-pip install .
+rsync -av -e ssh --exclude 'fastflow' --exclude '.git' --exclude '.venv' --exclude '*.egg-info' --exclude 'build' ./ dferraro@titanic.unipi.it:/home/dferraro/fastflow-python
 
+clean:
+rm -rf build

fastflow_module.cpp

@@ -4,60 +4,100 @@
 #include <cstring>
 #include <ostream>
 #include <sstream>
+#include <ff/ff.hpp>
 #include "py_ff_a2a.hpp"
 #include "py_ff_pipeline.hpp"
 #include "py_ff_farm.hpp"
+#include "py_ff_callback.hpp"
+#include "process/ff_monode_process.hpp"
+#include "process/ff_node_process.hpp"
+#include "py_ff_constant.hpp"
 
-// A struct contains the definition of a module
-PyModuleDef moduledef = {
- PyModuleDef_HEAD_INIT,
- "fastflow_module", // Module name
- "This is FastFlow's docstring",
- -1, // Optional size of the module state memory
- NULL, // Optional module methods
- NULL, // Optional slot definitions
- NULL, // Optional traversal function
- NULL, // Optional clear function
- NULL // Optional module deallocation function
-};
+PyDoc_STRVAR(ff_send_out_to_doc, "Send out to a node");
 
-PyMODINIT_FUNC
-PyInit_fastflow_module(void) {
- PyObject* module = PyModule_Create(&moduledef);
- if (module == NULL)
- return NULL;
+PyObject* empty_ff_send_out_to(PyObject *self, PyObject *args) {
+ assert(self);
 
+ PyErr_SetString(PyExc_Exception, "Operation not available. This is not a multi output node");
+ return NULL;
+}
+
+/* initialization function */
+static int
+fastflow_module_exec(PyObject *module) 
+{ 
  // add FFPipeline
  if (PyType_Ready(&py_ff_pipeline_type) < 0)
- return NULL;
+ return -1;
 
  if (PyModule_AddObject(module, "FFPipeline", (PyObject *) &py_ff_pipeline_type) < 0) {
- Py_DECREF(module);
- return NULL;
+ return -1;
  }
 
  // add FFFarm
  PyObject *py_ff_farm = PyType_FromSpec(&spec_py_ff_farm);
  if (py_ff_farm == NULL){
- return NULL;
+ return -1;
  }
  Py_INCREF(py_ff_farm);
 
- if (PyModule_AddObject(module, "FFFarm", py_ff_farm) < 0){
+ if (PyModule_AddObject(module, "FFFarm", py_ff_farm) < 0) {
  Py_DECREF(py_ff_farm);
- Py_DECREF(module);
- return NULL;
+ return -1;
  }
 
  // add FFAllToAll
  if (PyType_Ready(&py_ff_a2a_type) < 0)
- return NULL;
+ return -1;
 
  if (PyModule_AddObject(module, "FFAllToAll", (PyObject *) &py_ff_a2a_type) < 0) {
- Py_DECREF(module);
- return NULL;
+ return -1;
+ }
+
+ // add the callback object. Do not AddObject, 
+ // to avoid the user instantiating it
+ if (PyType_Ready(&py_ff_callback_type) < 0) {
+ return -1;
+ }
+
+ // add fastflow's constant type. Do not AddObject, 
+ // to avoid the user instantiating it
+ if (PyType_Ready(&py_ff_constant_type) < 0) {
+ return -1;
+ }
+
+ if (PyModule_AddObject(module, GO_ON_CONSTANT_NAME, build_py_ff_constant(ff::FF_GO_ON)) < 0) {
+ return -1;
  }
- return module;
+
+ return 0;
+}
+
+static PyMethodDef module_methods[] = {
+ { "ff_send_out_to", (PyCFunction) empty_ff_send_out_to, METH_VARARGS, ff_send_out_to_doc },
+ {NULL, NULL} /* Sentinel */
+};
+
+static PyModuleDef_Slot module_slots[] = {
+ {Py_mod_exec, (void*) fastflow_module_exec},
+#if PY_MINOR_VERSION >= 12
+ {Py_mod_multiple_interpreters, Py_MOD_PER_INTERPRETER_GIL_SUPPORTED},
+#endif
+ {0, NULL},
+};
+
+static struct PyModuleDef moduledef = {
+ .m_base = PyModuleDef_HEAD_INIT,
+ .m_name = "fastflow_module",
+ .m_doc = "This is FastFlow's docstring",
+ .m_size = 0,
+ .m_methods = module_methods,
+ .m_slots = module_slots,
+};
+
+PyMODINIT_FUNC
+PyInit_fastflow_module(void) {
+ return PyModuleDef_Init(&moduledef);
 }
 
 // Useful reading

include/error_macros.hpp

@@ -1,7 +1,6 @@
 #ifndef ERROR_MACROS
 #define ERROR_MACROS
 
-
 #if PY_MINOR_VERSION >= 12
  #define PRINT_ERROR(descr) \
 { PyObject* err = PyErr_GetRaisedException(); \
@@ -12,8 +11,7 @@
  PyErr_Clear(); }
 #else
 #define PRINT_ERROR(descr) \
-{ std::cerr << descr << "TODO: get raised exception" << std::endl; \
- PyErr_Clear(); }
+{ std::cerr << descr << std::endl; PyErr_Print(); }
 #endif
 
 #define CHECK_ERROR_THEN(descr, then) if (PyErr_Occurred()) { PRINT_ERROR(descr) { then } }

include/node_utils.hpp

@@ -4,8 +4,10 @@
 #include <Python.h>
 #include <ff/ff.hpp>
 #include "py_ff_node.hpp"
-#include "py_ff_node_subint.hpp"
-#include "py_ff_node_process.hpp"
+#include "subint/ff_monode_subint.hpp"
+#include "subint/ff_node_subint.hpp"
+#include "process/ff_monode_process.hpp"
+#include "process/ff_node_process.hpp"
 
 int parse_args(PyObject *args, PyObject *kwds, PyObject **py_node, bool *use_main_thread) {
  *use_main_thread = false;
@@ -30,18 +32,18 @@ int parse_args(PyObject *args, PyObject *kwds, PyObject **py_node, bool *use_mai
  return 0;
 }
 
-ff::ff_node* args_to_node(PyObject *args, PyObject *kwds, bool use_subints) {
+ff::ff_node* args_to_node(PyObject *args, PyObject *kwds, bool use_subints, bool multi_output = false) {
  PyObject *py_node = NULL;
  bool use_main_thread = false;
  if (parse_args(args, kwds, &py_node, &use_main_thread) == -1) return NULL;
 
  if (use_subints) {
- return new py_ff_node_subint(py_node);
+ return multi_output ? (ff::ff_node*)new ff_monode_subint(py_node):new ff_node_subint(py_node);
  } else if (use_main_thread) {
  return new py_ff_node(py_node);
  }
 
- return new py_ff_node_process(py_node);
+ return multi_output ? (ff::ff_node*)new ff_monode_process(py_node):new ff_node_process(py_node);
 }
 
 #endif // NODE_UTILS