feat: add EOS constant and map None to EOS, fix processes closing procedures

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

@@ -24,6 +24,7 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  close(send_fd);
  close(read_fd);
  pickl.~pickling();
+ // finalize python interpreter
  Py_Finalize();
  exit(0);
  };
@@ -33,12 +34,10 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  // receive serialized node
  int err = receiveMessage(read_fd, send_fd, message);
  if (err <= 0) handleError("[child] read serialized node", cleanup_exit());
- LOG("[child] read serialized node");
 
  // deserialize Python node
  auto node = pickl.unpickle(message.data);
  CHECK_ERROR_THEN("[child] unpickle node failure: ", cleanup_exit();)
- LOG("[child] deserialized node");
 
  py_ff_callback_object* callback = (py_ff_callback_object*) PyObject_CallObject(
  (PyObject *) &py_ff_callback_type, NULL
@@ -76,8 +75,6 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  return (PyObject*) NULL;
  }
 
- LOG("[child] sent ff_send_out_to message");
- 
  // receive response
  err = receiveMessage(read_fd, send_fd, message);
  if (err <= 0) {
@@ -104,41 +101,42 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
 
  while(err > 0) {
  err = receiveMessage(read_fd, send_fd, message);
- if (err < 0) handleError("[child] read next", cleanup_exit());
-
- if (err > 0) {
- // deserialize data
- auto py_args_tuple = pickl.unpickle(message.data);
- CHECK_ERROR_THEN("[child] deserialize data failure: ", cleanup_exit();)
- // call function
- PyObject *py_func = PyObject_GetAttrString(node, message.f_name.c_str());
- CHECK_ERROR_THEN("[child] get node function: ", cleanup_exit();)
-
- if (py_func) {
- // finally call the function
- PyObject* py_result = PyTuple_Check(py_args_tuple) == 1 ? PyObject_CallObject(py_func, py_args_tuple):PyObject_CallFunctionObjArgs(py_func, py_args_tuple, nullptr);
- CHECK_ERROR_THEN("[child] call function failure: ", cleanup_exit();)
-
- // 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 = "" });
- if (err <= 0) handleError("[child] send constant response", cleanup_exit());
- LOG("[child] sent constant response");
- } else {
- // serialize response
- auto result_str = pickl.pickle(py_result);
- CHECK_ERROR_THEN("[child] pickle result failure: ", cleanup_exit();)
-
- // send response
- err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_RESPONSE, .data = result_str, .f_name = "" });
- if (err <= 0) handleError("[child] send response", cleanup_exit());
- LOG("[child] sent response");
-
- Py_DECREF(py_result);
- Py_DECREF(py_func);
- }
+ if (err <= 0) handleError("[child] read next", cleanup_exit());
+ 
+ if (message.type == MESSAGE_TYPE_END_OF_LIFE) break;
+ 
+ // deserialize data
+ auto py_args_tuple = pickl.unpickle(message.data);
+ CHECK_ERROR_THEN("[child] deserialize data failure: ", cleanup_exit();)
+ // call function
+ PyObject *py_func = PyObject_GetAttrString(node, message.f_name.c_str());
+ CHECK_ERROR_THEN("[child] get node function: ", cleanup_exit();)
+
+ if (py_func) {
+ // finally call the function
+ PyObject* py_result = PyTuple_Check(py_args_tuple) == 1 ? PyObject_CallObject(py_func, py_args_tuple):PyObject_CallFunctionObjArgs(py_func, py_args_tuple, nullptr);
+ CHECK_ERROR_THEN("[child] call function failure: ", cleanup_exit();)
+
+ // 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);
+ 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());
+ } else {
+ // serialize response
+ auto result_str = pickl.pickle(py_result);
+ CHECK_ERROR_THEN("[child] pickle result failure: ", cleanup_exit();)
+
+ // send response
+ err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_RESPONSE, .data = result_str, .f_name = "" });
+ if (err <= 0) handleError("[child] send response", cleanup_exit());
+
+ Py_DECREF(py_result);
+ Py_DECREF(py_func);
  }
  }
  }
@@ -159,7 +157,6 @@ class base_process {
  }
 
  int svc_init() {
- TIMESTART(svc_init_start_time);
  // associate a new thread state with ff_node's thread
  PyThreadState* cached_tstate = tstate;
  tstate = PyThreadState_New(cached_tstate->interp);
@@ -252,7 +249,6 @@ class base_process {
  }
 
  // from here the GIL is NOT acquired
- LOGELAPSED("svc_init time ", svc_init_start_time);
  return returnValue;
  }
 
@@ -264,7 +260,6 @@ class base_process {
  int err = remote_procedure_call(send_fd, read_fd, serialized_data, "svc", response);
  if (err <= 0) {
  handleError("remote call of svc", );
- LOG("an error occurred, abort.");
  return NULL;
  }
 
@@ -273,7 +268,6 @@ class base_process {
  // if the call of ff_send_out_to (as of today...)
  if (response.f_name.compare("ff_send_out_to") != 0) {
  handleError("got invalid f_name", );
- LOG("an error occurred, got invalid f_name. Abort.");
  return NULL;
  }
 
@@ -291,63 +285,51 @@ class base_process {
  err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_RESPONSE, .data = result ? "t":"f", .f_name = "" });
  if (err <= 0) {
  handleError("error sending ff_send_out_to response", );
- LOG("an error occurred, sending ff_send_out_to. Abort.");
  return NULL;
  }
 
  // prepare for next iteration
  err = receiveMessage(read_fd, send_fd, response);
  if (err <= 0) {
  handleError("waiting for svc response", );
- LOG("an error occurred, abort.");
  return NULL;
  }
  }
 
  // 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_EOS) return ff::FF_EOS;
+ if (response.type == MESSAGE_TYPE_GO_ON || response.data.compare(none_str) == 0) {
+ return ff::FF_GO_ON;
+ }
 
  return new std::string(response.data);
  }
 
- void cleanup() {
- // Cleanup of objects created
- Py_DECREF(node);
- node = nullptr;
- tstate = nullptr;
- PyEval_SaveThread();
-
- if (send_fd > 0) close(send_fd);
- send_fd = -1;
- if (read_fd > 0) close(read_fd);
- send_fd = -1;
- }
-
  void svc_end() {
- TIMESTART(svc_end_start_time);
-
  if (has_svc_end) {
  Message response;
  auto empty_tuple = std::string(SERIALIZED_EMPTY_TUPLE);
  int err = remote_procedure_call(send_fd, read_fd, empty_tuple, "svc_end", response);
  if (err <= 0) handleError("read result of remote call of svc_end", );
  }
 
- // close the pipes, so the process can stop meanwhile we acquire the GIL and cleanup everything
+ // send end of life. Meanwhile we acquire the GIL and cleanup, the process will stop
+ int err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_END_OF_LIFE, .data = "", .f_name = "" });
+ 
+ // Acquire the main GIL
+ PyEval_RestoreThread(tstate);
+ // Cleanup of objects created
+ Py_DECREF(node);
+ node = nullptr;
+ tstate = nullptr;
+ // Release the main GIL
+ PyEval_SaveThread();
+ waitpid(pid, nullptr, 0);
+
  if (send_fd > 0) close(send_fd);
  send_fd = -1;
  if (read_fd > 0) close(read_fd);
  send_fd = -1;
-
- waitpid(pid, nullptr, 0);
-
- // Acquire the main GIL
- PyEval_RestoreThread(tstate);
- cleanup();
-
- LOGELAPSED("svc_end time ", svc_end_start_time);
  }
 
  void register_callback(ff::ff_monode* cb_node) {

include/process/messaging.hpp

@@ -5,14 +5,15 @@
 #include "debugging.hpp"
 #include <ff/distributed/ff_network.hpp> // import writen and readn
 
-#define handleError(msg, then) do { if (errno < 0) { perror(msg); } else { LOG(msg": errno == 0, fd closed" << std::endl); } then; } while(0)
+#define handleError(msg, then) do { perror(msg); then; } while(0)
 
 enum message_type {
  MESSAGE_TYPE_RESPONSE = 1,
  MESSAGE_TYPE_REMOTE_PROCEDURE_CALL,
- MESSAGE_TYPE_RESPONSE_GO_ON,
- MESSAGE_TYPE_RESPONSE_EOS,
- MESSAGE_TYPE_ACK
+ MESSAGE_TYPE_GO_ON,
+ MESSAGE_TYPE_EOS,
+ MESSAGE_TYPE_ACK,
+ MESSAGE_TYPE_END_OF_LIFE
 };
 
 struct Message {
@@ -46,8 +47,8 @@ int sendMessage(int read_fd, int send_fd, const Message& message) {
  handleError("write f_name", return -1);
  }
 
- int dummy; 
- int err = read(read_fd, &dummy, sizeof(dummy)); 
+ int ack; 
+ int err = read(read_fd, &ack, sizeof(ack));
  if (err <= 0) handleError("error receiving ACK", return err);
 
  return 1; // 0 = EOF, -1 = ERROR, 1 = SUCCESS
@@ -62,14 +63,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;
 
@@ -88,23 +90,22 @@ int receiveMessage(int read_fd, int send_fd, Message& message) {
  message.f_name = std::string(bufferFname);
  delete[] bufferFname;
 
- int dummy = 17; 
- res = write(send_fd, &dummy, sizeof(dummy)); 
+ int ack = 17; 
+ res = write(send_fd, &ack, sizeof(ack)); 
  if (res == -1) handleError("error sending ACK", return res);
 
  return 1; // 0 = EOF, -1 = ERROR, 1 = SUCCESS
 }
 
 int remote_procedure_call(int send_fd, int read_fd, std::string &data, const char *f_name, Message &response) {
- if (send_fd == -1 || read_fd == -1) {
- // they are -1 if an error occurred during svc_init or svc
- return -1;
- }
+ // they are -1 if an error occurred during svc_init or svc
+ if (send_fd == -1 || read_fd == -1) return -1;
 
  int err = sendMessage(read_fd, send_fd, { 
  .type = MESSAGE_TYPE_REMOTE_PROCEDURE_CALL, 
  .data = data, 
- .f_name = f_name });
+ .f_name = f_name 
+ });
  if (err <= 0) return err;
 
  return receiveMessage(read_fd, send_fd, response);
@@ -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(