feat: map None to ff::EOS

Author: domferr

fastflow_module.cpp

@@ -13,15 +13,6 @@
 #include "process/ff_node_process.hpp"
 #include "py_ff_constant.hpp"
 
-PyDoc_STRVAR(ff_send_out_to_doc, "Send out to a node");
-
-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) 
@@ -70,9 +61,22 @@ fastflow_module_exec(PyObject *module)
  return -1;
  }
 
+ if (PyModule_AddObject(module, EOS_CONSTANT_NAME, build_py_ff_constant(ff::FF_EOS)) < 0) {
+ return -1;
+ }
+
  return 0;
 }
 
+PyDoc_STRVAR(ff_send_out_to_doc, "Send out to a node");
+
+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;
+}
+
 static PyMethodDef module_methods[] = {
  { "ff_send_out_to", (PyCFunction) empty_ff_send_out_to, METH_VARARGS, ff_send_out_to_doc },
  {NULL, NULL} /* Sentinel */

include/process/base_process.hpp

@@ -121,9 +121,12 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
 
  // we may have a fastflow constant as result
  if (PyObject_TypeCheck(py_result, &py_ff_constant_type) != 0) {
- // py_ff_constant_object* _const_result = reinterpret_cast<py_ff_constant_object*>(py_result);
- // the only constant available is FF_GO_ON
- err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_RESPONSE_GO_ON, .data = "", .f_name = "" });
+ py_ff_constant_object* _const_result = reinterpret_cast<py_ff_constant_object*>(py_result);
+ err = sendMessage(read_fd, send_fd, { 
+ .type = _const_result->ff_const == ff::FF_EOS ? MESSAGE_TYPE_EOS:MESSAGE_TYPE_GO_ON, 
+ .data = "", 
+ .f_name = "" 
+ });
  if (err <= 0) handleError("[child] send constant response", cleanup_exit());
  LOG("[child] sent constant response");
  } else {
@@ -305,9 +308,9 @@ class base_process {
  }
 
  // got response of svc
- if (response.type == MESSAGE_TYPE_RESPONSE_GO_ON) return ff::FF_GO_ON;
- if (response.type == MESSAGE_TYPE_RESPONSE_EOS) return ff::FF_EOS;
- if (response.data.compare(none_str) == 0) return ff::FF_EOS;
+ if (response.type == MESSAGE_TYPE_GO_ON) return ff::FF_GO_ON;
+ if (response.type == MESSAGE_TYPE_EOS) return ff::FF_EOS;
+ if (response.data.compare(none_str) == 0) return ff::FF_GO_ON;
 
  return new std::string(response.data);
  }

include/process/messaging.hpp

@@ -10,8 +10,8 @@
 enum message_type {
  MESSAGE_TYPE_RESPONSE = 1,
  MESSAGE_TYPE_REMOTE_PROCEDURE_CALL,
- MESSAGE_TYPE_RESPONSE_GO_ON,
- MESSAGE_TYPE_RESPONSE_EOS,
+ MESSAGE_TYPE_GO_ON,
+ MESSAGE_TYPE_EOS,
  MESSAGE_TYPE_ACK
 };
 
@@ -62,14 +62,15 @@ int receiveMessage(int read_fd, int send_fd, Message& message) {
  uint32_t dataSize;
  res = read(read_fd, &dataSize, sizeof(dataSize));
  if (res <= 0) handleError("read data size", return res);
+
  char* bufferData = new char[dataSize + 1];
  if (dataSize > 0) {
  res = readn(read_fd, bufferData, dataSize);
  if (res <= 0) handleError("read data", return res);
  }
 
  bufferData[dataSize] = '\0';
- message.data = std::string(bufferData);
+ //message.data = std::string(bufferData);
  message.data.assign(bufferData, dataSize);
  delete[] bufferData;
 
@@ -124,7 +125,7 @@ void create_message_ff_send_out_to(Message &message, int index, void* &constant,
  message.data.append(index_str);
  message.data.append("~");
  message.data.append(constant != NULL ? "t":"f");
- message.data.append(constant != NULL ? std::to_string(constant == ff::FF_EOS ? MESSAGE_TYPE_RESPONSE_EOS:MESSAGE_TYPE_RESPONSE_GO_ON):data);
+ message.data.append(constant != NULL ? std::to_string(constant == ff::FF_EOS ? MESSAGE_TYPE_EOS:MESSAGE_TYPE_GO_ON):data);
 }
 
 void parse_message_ff_send_out_to(Message &message, void **constant, int *index, std::string *data) {
@@ -133,10 +134,10 @@ void parse_message_ff_send_out_to(Message &message, void **constant, int *index,
  if (message.data.at(dividerPos+1) == 't') {
  *data = "";
  std::string inner_data = message.data.substr(dividerPos+2, message.data.length() - dividerPos - 1);
- if (inner_data.compare(std::to_string(MESSAGE_TYPE_RESPONSE_EOS)) == 0) {
+ if (inner_data.compare(std::to_string(MESSAGE_TYPE_EOS)) == 0) {
  *constant = ff::FF_EOS;
  }
- if (inner_data.compare(std::to_string(MESSAGE_TYPE_RESPONSE_GO_ON)) == 0) {
+ if (inner_data.compare(std::to_string(MESSAGE_TYPE_GO_ON)) == 0) {
  *constant = ff::FF_GO_ON;
  }
  } else {

include/py_ff_a2a.hpp

@@ -138,8 +138,18 @@ PyObject* py_ff_a2a_add_firstset(PyObject *self, PyObject *args, PyObject* kwds)
  py_ff_pipeline_object* _pipe = reinterpret_cast<py_ff_pipeline_object*>(item);
  if (!_pipe->pipeline->isMultiOutput()) {
  ff::ff_node* last_stage = _pipe->pipeline->get_laststage();
- ff::ff_node* new_node = new ff::internal_mo_transformer(last_stage, false);
- _pipe->pipeline->change_node(last_stage, new_node, true, true);
+ // let's replace the last stage with a multi output variant
+ PyObject *last_stage_py_node = _self->use_subinterpreters ? 
+ (reinterpret_cast<ff_node_subint*>(last_stage))->get_python_object():
+ (reinterpret_cast<ff_node_process*>(last_stage))->get_python_object();
+ if (last_stage_py_node == NULL) {
+ PyErr_SetString(PyExc_RuntimeError, "Unable to transform previous stage to multi output (null pointer)");
+ return NULL;
+ }
+ ff::ff_node* new_last_stage = _self->use_subinterpreters ? 
+ (ff::ff_node*) new ff_monode_subint(last_stage_py_node):
+ (ff::ff_node*) new ff_monode_process(last_stage_py_node);
+ _pipe->pipeline->change_node(last_stage, new_last_stage, true, true);
  }
  node = _pipe->pipeline;
  } else if (_self->use_subinterpreters) {

include/py_ff_constant.hpp

@@ -19,6 +19,7 @@ static PyTypeObject py_ff_constant_type = {
 };
 
 #define GO_ON_CONSTANT_NAME "GO_ON"
+#define EOS_CONSTANT_NAME "EOS"
 
 PyObject* build_py_ff_constant(void *val) {
  py_ff_constant_object* constant = (py_ff_constant_object*) PyObject_CallObject(

include/subint/base_subint.hpp

@@ -141,6 +141,10 @@ for [k, v] in glb:
  CHECK_ERROR_THEN("PyDict_SetItemString failure: ", returnValue = -1;)
  }
 
+ if (PyDict_SetItemString(globals, EOS_CONSTANT_NAME, build_py_ff_constant(ff::FF_EOS)) == -1) {
+ CHECK_ERROR_THEN("PyDict_SetItemString failure: ", returnValue = -1;)
+ }
+
  if (PyObject_HasAttrString(node, "svc_init")) {
  PyObject* svc_init_func = PyObject_GetAttrString(node, "svc_init");
  if (svc_init_func) {
@@ -162,20 +166,22 @@ for [k, v] in glb:
  }
 
  void * svc(void *arg) {
- auto is_ff_marker = arg == ff::FF_GO_ON;
  TIMESTART(svc_start_time);
- PyObject* pickled_bytes = is_ff_marker || arg == NULL ? NULL:reinterpret_cast<PyObject*>(arg);
+ // arg may be equal to ff::FF_GO_ON in case of a node of a first set of an a2a that hasn't input channels
+ PyObject* pickled_bytes = arg == ff::FF_GO_ON || arg == NULL ? NULL:reinterpret_cast<PyObject*>(arg);
 
- PyObject* py_args = is_ff_marker || arg == NULL ? nullptr:pickl->unpickle_bytes(pickled_bytes);
+ PyObject* py_args = arg == ff::FF_GO_ON || arg == NULL ? nullptr:pickl->unpickle_bytes(pickled_bytes);
  CHECK_ERROR_THEN("unpickle serialized data failure: ", return NULL;)
  //if (pickled_bytes) Py_DECREF(pickled_bytes);
 
  PyObject* py_result = py_args != nullptr && PyTuple_Check(py_args) == 1 ? PyObject_CallObject(svc_func, py_args):PyObject_CallFunctionObjArgs(svc_func, py_args, nullptr);
  CHECK_ERROR_THEN("PyObject_CallObject failure: ", return NULL;)
-
+ 
+ // map None to ff::FF_GO_ON
+ // we have None also if the svc function returned nothing (e.g. void function in c++)
  if (py_result == Py_None) {
  Py_DECREF(py_result);
- return NULL;
+ return ff::FF_GO_ON;
  }
 
  // we may have a fastflow constant as result

tests/python/a2a-ff-send-to.test.py

@@ -1,4 +1,4 @@
-from fastflow_module import FFAllToAll, GO_ON, ff_send_out_to
+from fastflow_module import FFAllToAll, EOS, ff_send_out_to
 import sys
 
 """
@@ -18,21 +18,18 @@ def __init__(self, id):
 
  def svc(self, *arg):
  if self.counter > 5:
- return
+ return EOS
  self.counter += 1
 
  ff_send_out_to(list([f"source{self.id}-to-sink1"]), 0)
 
- return GO_ON
-
 class sink():
  def __init__(self, id):
  self.id = id
 
  def svc(self, lis: list):
  lis.append(f"sink{self.id}")
  print(lis)
- return GO_ON
 
 def run_test(use_subinterpreters = True):
  a2a = FFAllToAll(use_subinterpreters)

tests/python/a2a-with-pipe.test.py

@@ -1,4 +1,4 @@
-from fastflow_module import FFAllToAll, FFPipeline, GO_ON
+from fastflow_module import FFAllToAll, FFPipeline, EOS
 import sys
 
 """
@@ -18,7 +18,7 @@ def __init__(self, id):
 
  def svc(self, *arg):
  if self.counter > 5:
- return
+ return EOS
  self.counter += 1
 
  return list([self.id])
@@ -46,7 +46,6 @@ def __init__(self, id):
  def svc(self, lis: list):
  lis.append(self.id)
  print(lis)
- return GO_ON
 
 def run_test(use_subinterpreters = True):
  a2a = FFAllToAll(use_subinterpreters)

tests/python/a2a.test.py

@@ -1,4 +1,4 @@
-from fastflow_module import FFAllToAll, GO_ON
+from fastflow_module import FFAllToAll, EOS
 import sys
 
 """
@@ -18,7 +18,7 @@ def __init__(self, id):
 
  def svc(self, *arg):
  if self.counter > 5:
- return
+ return EOS
  self.counter += 1
 
  return list([self.id])
@@ -30,7 +30,6 @@ def __init__(self, id):
  def svc(self, lis: list):
  lis.append(self.id)
  print(lis)
- return GO_ON
 
 def run_test(use_subinterpreters = True):
  a2a = FFAllToAll(use_subinterpreters)

tests/python/farm.test.py

@@ -1,4 +1,4 @@
-from fastflow_module import FFFarm, GO_ON
+from fastflow_module import FFFarm, EOS
 import sys
 
 class source():
@@ -7,7 +7,7 @@ def __init__(self):
 
  def svc(self, *arg):
  if self.counter == 12:
- return
+ return EOS
  self.counter += 1
 
  return list(['source'])
@@ -23,7 +23,6 @@ def svc(self, lis: list):
 class sink(): 
  def svc(self, lis: list):
  print(lis)
- return GO_ON
 
 def run_test(use_subinterpreters = True):
  farm = FFFarm(use_subinterpreters)