feat: add support to sending out constants

Author: domferr

include/process/base_process.hpp

@@ -54,12 +54,20 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  return (PyObject*) NULL;
  }
 
- // serialize pydata
- auto pydata_str = pickl.pickle(pydata);
- if (PyErr_Occurred()) return (PyObject*) NULL;
+ // we may have a fastflow constant as data to send out to index
+ void *constant = NULL;
+ std::string pydata_str = "";
+ if (PyObject_TypeCheck(pydata, &py_ff_constant_type) != 0) {
+ py_ff_constant_object* _const_result = reinterpret_cast<py_ff_constant_object*>(pydata);
+ constant = _const_result->ff_const;
+ } else {
+ // serialize pydata
+ pydata_str = pickl.pickle(pydata);
+ if (PyErr_Occurred()) return (PyObject*) NULL;
+ }
 
  Message message;
- create_message_ff_send_out_to(message, pydata_str, index);
+ create_message_ff_send_out_to(message, index, constant, pydata_str);
 
  // send response
  int err = sendMessage(read_fd, send_fd, message);
@@ -140,7 +148,7 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  cleanup_exit();
 }
 
-#define EMPTY_TUPLE_STR "(t."
+#define SERIALIZED_EMPTY_TUPLE "(t."
 
 class base_process {
 public: 
@@ -214,22 +222,22 @@ class base_process {
  send_fd = mainToChildFD[1];
  read_fd = childToMainFD[0];
 
- int err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_REMOTE_FUNCTION_CALL, .data = node_str, .f_name = "" });
+ int err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_REMOTE_PROCEDURE_CALL, .data = node_str, .f_name = "" });
  if (err <= 0) handleError("send serialized node", returnValue = -1);
 
  if (err > 0 && has_svc_init) {
  Message response;
- auto empty_tuple = std::string(EMPTY_TUPLE_STR);
- int err = remote_function_call(send_fd, read_fd, empty_tuple, "svc_init", response);
+ auto empty_tuple = std::string(SERIALIZED_EMPTY_TUPLE);
+ int err = remote_procedure_call(send_fd, read_fd, empty_tuple, "svc_init", response);
  if (err <= 0) {
  handleError("read result of remote call of svc_init", );
  } else {
  // Hold the main GIL
  PyEval_RestoreThread(tstate);
  PyObject *svc_init_result = pickl.unpickle(response.data);
  CHECK_ERROR_THEN("unpickle svc_init_result failure: ", returnValue = -1;)
+
  returnValue = 0;
- // if we are here, then the GIL was acquired before
  if (PyLong_Check(svc_init_result)) {
  long res_as_long = PyLong_AsLong(svc_init_result);
  returnValue = static_cast<int>(res_as_long);
@@ -249,31 +257,35 @@ class base_process {
  }
 
  void* svc(void *arg) {
- auto is_ff_marker = arg == ff::FF_GO_ON;
- std::string serialized_data = is_ff_marker || arg == NULL ? EMPTY_TUPLE_STR:*reinterpret_cast<std::string*>(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
+ std::string serialized_data = arg == NULL || arg == ff::FF_GO_ON ? SERIALIZED_EMPTY_TUPLE:*reinterpret_cast<std::string*>(arg);
 
  Message response;
- int err = remote_function_call(send_fd, read_fd, serialized_data, "svc", response);
+ 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;
  }
 
- while(response.type == MESSAGE_TYPE_REMOTE_FUNCTION_CALL) {
+ while(response.type == MESSAGE_TYPE_REMOTE_PROCEDURE_CALL) {
+ // the only supported remote procedure call from the child 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;
  }
 
  // parse received ff_send_out_to request
- int sendout_index = 0;
- std::string* ser_data = new std::string();
- parse_message_ff_send_out_to(response, ser_data, &sendout_index);
+ int index;
+ void *constant = NULL;
+ std::string *data = new std::string();
+ parse_message_ff_send_out_to(response, &constant, &index, data);
 
+ if (constant != NULL) free(data);
  // finally perform ff_send_out_to
- bool result = registered_callback->ff_send_out_to(ser_data, sendout_index);
+ bool result = registered_callback->ff_send_out_to(constant != NULL ? constant:data, index);
 
  // send ff_send_out_to result
  err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_RESPONSE, .data = result ? "t":"f", .f_name = "" });
@@ -294,6 +306,7 @@ 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;
 
  return new std::string(response.data);
@@ -317,8 +330,8 @@ class base_process {
 
  if (has_svc_end) {
  Message response;
- auto empty_tuple = std::string(EMPTY_TUPLE_STR);
- int err = remote_function_call(send_fd, read_fd, empty_tuple, "svc_end", 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", );
  }
 

include/process/messaging.hpp

@@ -7,13 +7,16 @@
 
 #define handleError(msg, then) do { if (errno < 0) { perror(msg); } else { LOG(msg": errno == 0, fd closed" << std::endl); } then; } while(0)
 
-#define MESSAGE_TYPE_RESPONSE '1'
-#define MESSAGE_TYPE_RESPONSE_GO_ON '2'
-#define MESSAGE_TYPE_REMOTE_FUNCTION_CALL '3'
-#define MESSAGE_TYPE_ACK '4'
+enum message_type {
+ MESSAGE_TYPE_RESPONSE = 1,
+ MESSAGE_TYPE_REMOTE_PROCEDURE_CALL,
+ MESSAGE_TYPE_RESPONSE_GO_ON,
+ MESSAGE_TYPE_RESPONSE_EOS,
+ MESSAGE_TYPE_ACK
+};
 
 struct Message {
- char type;
+ message_type type;
  std::string data;
  std::string f_name;
 };
@@ -34,7 +37,7 @@ int sendMessage(int read_fd, int send_fd, const Message& message) {
  }
 
  // send f_name
- uint32_t fnameSize = message.f_name.size();
+ uint32_t fnameSize = message.f_name.length();
  if (write(send_fd, &fnameSize, sizeof(fnameSize)) == -1) {
  handleError("write f_name size", return -1);
  }
@@ -92,36 +95,54 @@ int receiveMessage(int read_fd, int send_fd, Message& message) {
  return 1; // 0 = EOF, -1 = ERROR, 1 = SUCCESS
 }
 
-void create_message_ff_send_out_to(Message &message, std::string &data, int index) {
- message.type = MESSAGE_TYPE_REMOTE_FUNCTION_CALL;
- message.f_name = "ff_send_out_to";
- std::string index_str = std::to_string(index);
- message.data.erase();
- message.data.reserve(index_str.length() + 3 + data.length());
- message.data.append(index_str);
- message.data.append("~");
- message.data.append(data);
-}
-
-void parse_message_ff_send_out_to(Message &message, std::string *data, int* sendout_index) {
- int dividerPos = message.data.find('~');
- *sendout_index = std::stoi(message.data.substr(0, dividerPos));
- *data = message.data.substr(dividerPos+1, message.data.length() - dividerPos);
-}
-
-int remote_function_call(int send_fd, int read_fd, std::string &data, const char *f_name, Message &response) {
+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;
  }
 
  int err = sendMessage(read_fd, send_fd, { 
- .type = MESSAGE_TYPE_REMOTE_FUNCTION_CALL, 
+ .type = MESSAGE_TYPE_REMOTE_PROCEDURE_CALL, 
  .data = data, 
  .f_name = f_name });
  if (err <= 0) return err;
 
  return receiveMessage(read_fd, send_fd, response);
 }
 
+void create_message_ff_send_out_to(Message &message, int index, void* &constant, std::string &data) {
+ if (constant != NULL && constant != ff::FF_EOS 
+ && constant != ff::FF_GO_ON) {
+ throw "fastflow constant not supported";
+ }
+
+ message.type = MESSAGE_TYPE_REMOTE_PROCEDURE_CALL;
+ message.f_name = "ff_send_out_to";
+ std::string index_str = std::to_string(index);
+ message.data.erase();
+ message.data.reserve(index_str.length() + 8 + data.length()); // 8 is just to be sure there is enough space for the 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);
+}
+
+void parse_message_ff_send_out_to(Message &message, void **constant, int *index, std::string *data) {
+ int dividerPos = message.data.find('~');
+ *index = std::stoi(message.data.substr(0, dividerPos));
+ 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) {
+ *constant = ff::FF_EOS;
+ }
+ if (inner_data.compare(std::to_string(MESSAGE_TYPE_RESPONSE_GO_ON)) == 0) {
+ *constant = ff::FF_GO_ON;
+ }
+ } else {
+ *constant = NULL;
+ *data = message.data.substr(dividerPos+2, message.data.length() - dividerPos - 1);
+ }
+}
+
 #endif //MESSAGING_HPP

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

@@ -2,13 +2,13 @@
 import sys
 
 """
- first ______ second
- |
- first ___| second
- |
- first ___| second
- |
- first ___|
+ source ______ sink
+  |
+ source ___| sink
+  |
+ source ___| sink
+  |
+ source ___|
 """
 
 class source():

tests/python/a2a.test.py

@@ -2,13 +2,13 @@
 import sys
 
 """
- first _ _ second
- | |
- first _|_|_ second
- | |
- first _|_|_ second
- |
- first _|
+ source _ _ sink
+  | |
+ source _|_|_ sink
+  | |
+ source _|_|_ sink
+  |
+ source _|
 """
 
 class source():
@@ -18,7 +18,7 @@ def __init__(self, id):
 
  def svc(self, *arg):
  if self.counter > 5:
- return None
+ return
  self.counter += 1
 
  return list([self.id])