feat: add support to ondemand workers and ff_send_out

domferr · domferr · commit 64c4e6e07479 · 2024-10-01T14:43:50.000+02:00
diff --git a/fastflow_module.cpp b/fastflow_module.cpp
@@ -68,17 +68,17 @@ fastflow_exec(PyObject *module)
     return 0;
 }
 
-PyDoc_STRVAR(ff_send_out_to_doc, "Send out to a node");
+PyDoc_STRVAR(ff_send_out_doc, "Send out to a node");
 
-PyObject* empty_ff_send_out_to(PyObject *self, PyObject *args) {
+PyObject* empty_ff_send_out(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 },
+    { "ff_send_out", (PyCFunction) empty_ff_send_out, METH_VARARGS, ff_send_out_doc },
     {NULL, NULL} /* Sentinel */
 };
 
diff --git a/include/process/base_process.hpp b/include/process/base_process.hpp
@@ -41,32 +41,27 @@ void process_body(std::string &node_ser, int read_fd, int send_fd, bool isMultiO
     py_ff_callback_object* callback = (py_ff_callback_object*) PyObject_CallObject(
         (PyObject *) &py_ff_callback_type, NULL
     );
-    callback->ff_send_out_to_callback = [isMultiOutput, &pickl, &messaging](PyObject* pydata, int index) {
+    callback->ff_send_out_callback = [isMultiOutput, &pickl, &messaging](PyObject* pydata, int index) {
         if (!isMultiOutput) {
             PyErr_SetString(PyExc_Exception, "Operation not available. This is not a multi output node");
             return (PyObject*) NULL;
         }
 
-        if (index < 0) {
-            PyErr_SetString(PyExc_Exception, "Index cannot be negative");
-            return (PyObject*) NULL;
-        }
-
         Message response;
         if (PyObject_TypeCheck(pydata, &py_ff_constant_type) != 0) {
             // we may have a fastflow constant as data to send out to index
             py_ff_constant_object* _const_result = reinterpret_cast<py_ff_constant_object*>(pydata);
-            int err = messaging.call_remote(response, "ff_send_out_to", index, _const_result->ff_const);
+            int err = messaging.call_remote(response, "ff_send_out", _const_result->ff_const, index);
             if (err <= 0)  {
-                PyErr_SetString(PyExc_Exception, "Error occurred sending ff_send_out_to request");
+                PyErr_SetString(PyExc_Exception, "Error occurred sending ff_send_out request");
                 return (PyObject*) NULL;
             }
         } else {
             auto data = pickl.pickle(pydata);
-            int err = messaging.call_remote(response, "ff_send_out_to", index, data);
+            int err = messaging.call_remote(response, "ff_send_out", data, index);
             if (PyErr_Occurred()) return (PyObject*) NULL;
             if (err <= 0)  {
-                PyErr_SetString(PyExc_Exception, "Error occurred sending ff_send_out_to request");
+                PyErr_SetString(PyExc_Exception, "Error occurred sending ff_send_out request");
                 return (PyObject*) NULL;
             }
         }
@@ -81,7 +76,7 @@ void process_body(std::string &node_ser, int read_fd, int send_fd, bool isMultiO
         CHECK_ERROR_THEN("PyDict_SetItemString failure: ", cleanup_exit();)
     }
     // if you access the methods by importing them from the module, replace each method with the delegate's one
-    if (PyDict_SetItemString(globals, "ff_send_out_to", PyObject_GetAttrString((PyObject*) callback, "ff_send_out_to")) == -1) {
+    if (PyDict_SetItemString(globals, "ff_send_out", PyObject_GetAttrString((PyObject*) callback, "ff_send_out")) == -1) {
         CHECK_ERROR_THEN("PyDict_SetItemString failure: ", cleanup_exit();)
     }
 
@@ -248,23 +243,27 @@ class base_process {
         
         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) {
+            // if the call of ff_send_out (as of today...)
+            if (response.f_name.compare("ff_send_out") != 0) {
                 handleError("got invalid f_name", );
                 return NULL;
             }
 
-            // parse received ff_send_out_to request
-            std::tuple<int, std::string> args = messaging.parse_data<int, std::string>(response.data);
+            // parse received ff_send_out request
+            std::tuple<std::string, int> args = messaging.parse_data<std::string, int>(response.data);
             // try to deserialize to constant. If it results into NULL, then it is NOT a FastFlow's constant
-            void* constant = deserialize<void*>(std::get<1>(args));
-            // finally perform ff_send_out_to
-            bool result = registered_callback->ff_send_out_to(constant == NULL ? (void*) new std::string(std::get<1>(args)):constant, std::get<0>(args));
+            void* constant = deserialize<void*>(std::get<0>(args));
+            // finally perform ff_send_out
+            int index = std::get<1>(args);
+            auto data = constant == NULL ? (void*) new std::string(std::get<0>(args)):constant;
+            bool result = index >= 0 ? 
+                registered_callback->ff_send_out_to(data, index):
+                registered_callback->ff_send_out(data);
             
-            // send ff_send_out_to result
+            // send ff_send_out result
             err = messaging.send_response(result);
             if (err <= 0) {
-                handleError("error sending ff_send_out_to response", );
+                handleError("error sending ff_send_out response", );
                 return NULL;
             }
 
diff --git a/include/py_ff_a2a.hpp b/include/py_ff_a2a.hpp
@@ -124,8 +124,30 @@ PyObject* py_ff_a2a_add_firstset(PyObject *self, PyObject *args, PyObject* kwds)
     py_ff_a2a_object* _self = reinterpret_cast<py_ff_a2a_object*>(self);
 
     PyObject *py_nodes = NULL;
-    bool use_main_thread = false;
-    if (parse_args(args, kwds, &py_nodes, &use_main_thread) == -1) return NULL;
+
+    PyObject *py_use_main_thread = Py_False;
+    PyObject *py_ondemand = Py_False;
+    static const char *kwlist[] = { "", "ondemand", "use_main_thread", NULL };
+    if (!PyArg_ParseTupleAndKeywords(args, kwds, "O|$OO", const_cast<char**>(kwlist), &py_nodes, &py_ondemand, &py_use_main_thread)) {
+        assert(PyErr_Occurred());
+        PyErr_SetString(PyExc_RuntimeError, "Tuple and keywords parsing failed");
+        return NULL;
+    }
+
+    if (py_use_main_thread != nullptr && !PyBool_Check(py_use_main_thread)) {
+        PyErr_Format(PyExc_TypeError, "A bool is required for 'use_main_thread' keyword (got type %s)",
+                     Py_TYPE(py_use_main_thread)->tp_name);
+        return NULL;
+    }
+
+    if (py_ondemand != nullptr && !PyBool_Check(py_ondemand)) {
+        PyErr_Format(PyExc_TypeError, "A bool is required for 'ondemand' keyword (got type %s)",
+                     Py_TYPE(py_ondemand)->tp_name);
+        return NULL;
+    }
+
+    bool use_main_thread = PyObject_IsTrue(py_use_main_thread) == 1;
+    bool ondemand = PyObject_IsTrue(py_ondemand) == 1;
 
     std::vector<ff::ff_node*> set;
     // Iterate over all the arguments
@@ -163,8 +185,7 @@ PyObject* py_ff_a2a_add_firstset(PyObject *self, PyObject *args, PyObject* kwds)
     }
     Py_DECREF(iterator);
 
-    int ondemand = 0;
-    int val = _self->a2a->add_firstset(set, ondemand, false);
+    int val = _self->a2a->add_firstset(set, ondemand ? 1:0, false);
     return PyLong_FromLong(val);
 }
 
diff --git a/include/py_ff_callback.hpp b/include/py_ff_callback.hpp
@@ -7,31 +7,48 @@
 
 typedef struct {
     PyObject_HEAD
-    std::function<PyObject*(PyObject*, int)> ff_send_out_to_callback;
+    std::function<PyObject*(PyObject*, int)> ff_send_out_callback;
 } py_ff_callback_object;
 
-PyDoc_STRVAR(py_ff_callback_ff_send_out_to_doc, "FF send out to");
+PyDoc_STRVAR(py_ff_callback_ff_send_out_doc, "FF send out to");
 
-PyObject* py_ff_callback_ff_send_out_to(PyObject *self, PyObject *args)
+PyObject* py_ff_callback_ff_send_out(PyObject *self, PyObject *args)
 {
     assert(self);
     py_ff_callback_object* _self = reinterpret_cast<py_ff_callback_object*>(self);
     
     PyObject *pydata = nullptr;
-    unsigned int index = 0;
-    if (!PyArg_ParseTuple(args, "OI", &pydata, &index)) return NULL;
+    PyObject *pyindex = nullptr;
+    // doc: http://web.mit.edu/people/amliu/vrut/python/ext/parseTuple.html
+    if (!PyArg_ParseTuple(args, "O|O", &pydata, &pyindex)) return NULL;
     
     if (pydata == nullptr) {
         PyErr_SetString(PyExc_TypeError, "Please provide valid data object");
         return NULL;
     }
 
-    return _self->ff_send_out_to_callback(pydata, index);
+    int index = -1;
+    if (pyindex != nullptr) {
+        if (PyLong_Check(pyindex) == 0) {
+            PyErr_Format(PyExc_TypeError, "Please provide the index as an integer object (got type %s)",
+                        Py_TYPE(pyindex)->tp_name);
+            return NULL;
+        }
+
+        index = PyLong_AsLong(pyindex);
+
+        if (index < 0) {
+            PyErr_SetString(PyExc_Exception, "Index cannot be negative");
+            return (PyObject*) NULL;
+        }
+    }
+
+    return _self->ff_send_out_callback(pydata, index);
 }
 
 static PyMethodDef py_ff_callback_methods[] = {
-    { "ff_send_out_to", (PyCFunction) py_ff_callback_ff_send_out_to, 
-        METH_VARARGS, py_ff_callback_ff_send_out_to_doc },
+    { "ff_send_out", (PyCFunction) py_ff_callback_ff_send_out, 
+        METH_VARARGS, py_ff_callback_ff_send_out_doc },
     {NULL, NULL} /* Sentinel */
 };
 
diff --git a/include/subint/base_subint.hpp b/include/subint/base_subint.hpp
@@ -84,8 +84,8 @@ class base_subint {
         py_ff_callback_object* callback = (py_ff_callback_object*) PyObject_CallObject(
             (PyObject *) &py_ff_callback_type, NULL
         );
-        callback->ff_send_out_to_callback = [this](PyObject* pydata, int index) {
-            return this->py_ff_send_out_to(pydata, index);
+        callback->ff_send_out_callback = [this](PyObject* pydata, int index) {
+            return this->py_ff_send_out(pydata, index);
         };
         
         int returnValue = 0;
@@ -96,7 +96,7 @@ class base_subint {
             CHECK_ERROR_THEN("PyDict_SetItemString failure: ", returnValue = -1;)
         }
         // if you access the methods by importing them from the module, replace each method with the callback's one
-        if (PyDict_SetItemString(globals, "ff_send_out_to", PyObject_GetAttrString((PyObject*) callback, "ff_send_out_to")) == -1) {
+        if (PyDict_SetItemString(globals, "ff_send_out", PyObject_GetAttrString((PyObject*) callback, "ff_send_out")) == -1) {
             CHECK_ERROR_THEN("PyDict_SetItemString failure: ", returnValue = -1;)
         }
 
@@ -205,27 +205,25 @@ class base_subint {
         LOGELAPSED("svc_end time ", svc_end_start_time);
     }
 
-    PyObject* py_ff_send_out_to(PyObject *py_data, int index) {        
-        if (registered_callback == NULL) {
+    PyObject* py_ff_send_out(PyObject *py_data, int index) {        
+        if (registered_callback == NULL && index != -1) {
             PyErr_SetString(PyExc_Exception, "Operation not available. This is not a multi output node");
             return NULL;
         }
 
-        if (index < 0) {
-            PyErr_SetString(PyExc_Exception, "Index cannot be negative");
-            return (PyObject*) NULL;
-        }
-
         // we may have a fastflow constant as data to send out to index
         if (PyObject_TypeCheck(py_data, &py_ff_constant_type) != 0) {
             py_ff_constant_object* _const_result = reinterpret_cast<py_ff_constant_object*>(py_data);
-            return registered_callback->ff_send_out_to(_const_result->ff_const, index) ? Py_True:Py_False;
+            auto res = index >= 0 ? 
+                registered_callback->ff_send_out_to(_const_result->ff_const, index):
+                registered_callback->ff_send_out(_const_result->ff_const);
+            return res ? Py_True:Py_False;
         }
 
         auto pickled_data_bytes = pickl->pickle_bytes(py_data);
         CHECK_ERROR_THEN("pickle send out data failure: ", return NULL;)
 
-        return registered_callback->ff_send_out_to(pickled_data_bytes, index) ? Py_True:Py_False;
+        return registered_callback->ff_send_out(pickled_data_bytes, index) ? Py_True:Py_False;
     }
 
     void register_callback(ff::ff_monode* cb_node) {
diff --git a/tests/python/a2a-ff-send-to.test.py b/tests/python/a2a-ff-send-to.test.py
@@ -1,4 +1,4 @@
-from fastflow import FFAllToAll, EOS, GO_ON, ff_send_out_to
+from fastflow import FFAllToAll, EOS, GO_ON, ff_send_out
 import sys
 
 """
@@ -21,7 +21,7 @@ def svc(self, *arg):
             return EOS
         self.counter += 1
 
-        ff_send_out_to(list([f"source{self.id}-to-sink1"]), 0)
+        ff_send_out(list([f"source{self.id}-to-sink1"]), 0)
 
 class sink():
     def __init__(self, id):
diff --git a/tests/python/a2a-ff-send.test.py b/tests/python/a2a-ff-send.test.py
@@ -0,0 +1,55 @@
+from fastflow import FFAllToAll, EOS, GO_ON, ff_send_out
+import sys
+
+"""
+    source _   _ sink
+            | |
+    source _|_|_ sink
+            | |
+    source _|_|_ sink
+            |
+    source _|
+"""
+
+class source():
+    def __init__(self, id):
+        self.counter = 1
+        self.id = id
+
+    def svc(self, *arg):
+        while self.counter < 5:
+            ff_send_out(list([f"source{self.id}-to-any"]))
+            self.counter += 1
+        return EOS
+
+class sink():
+    def __init__(self, id):
+        self.id = id
+
+    def svc(self, lis: list):
+        lis.append(f"sink{self.id}")
+        print(lis)
+
+def run_test(use_subinterpreters = True):
+    a2a = FFAllToAll(use_subinterpreters)
+    first_stage_size = 4
+    second_stage_size = 3
+    # build first stages
+    first_lis = [source(i+1) for i in range(first_stage_size)]
+    # build second stages
+    second_lis = [sink(i+1) for i in range(second_stage_size)]
+    # add first stages
+    a2a.add_firstset(first_lis)
+    # add second stages
+    a2a.add_secondset(second_lis)
+    a2a.run_and_wait_end()
+
+if __name__ == "__main__":
+    if sys.version_info[1] >= 12:
+        print("Subinterpreters")
+        run_test(use_subinterpreters = True)
+    else:
+        print("Skip subinterpreters test because python version is < 3.12")
+    print()
+    print("Processes")
+    run_test(use_subinterpreters = False)
diff --git a/tests/python/a2a-ondemand.test.py b/tests/python/a2a-ondemand.test.py
@@ -0,0 +1,60 @@
+from fastflow import FFAllToAll, EOS, GO_ON, ff_send_out
+import sys
+import time
+
+"""
+    source _   _ slowestsink
+            | |
+    source _|_|_ sink
+            | |
+    source _|_|_ sink
+            |
+    source _|
+"""
+
+class source():
+    def __init__(self, id):
+        self.counter = 1
+        self.id = id
+
+    def svc(self, *arg):
+        while self.counter < 8:
+            ff_send_out(list([f"source{self.id}-to-any"]))
+            self.counter += 1
+        return EOS
+
+class sink():
+    def __init__(self, id):
+        self.id = id
+
+    def svc(self, lis: list):
+        if self.id == 1:
+            time.sleep(0.1)
+            lis.append(f"slowestsink{self.id}")
+        else:
+            lis.append(f"sink{self.id}")
+        print(lis)
+
+def run_test(use_subinterpreters = True):
+    a2a = FFAllToAll(use_subinterpreters)
+    first_stage_size = 4
+    second_stage_size = 3
+    # build first stages
+    first_lis = [source(i+1) for i in range(first_stage_size)]
+    # build second stages
+    second_lis = [sink(i+1) for i in range(second_stage_size)]
+    # add first stages
+    a2a.add_firstset(first_lis, ondemand=True)
+    # add second stages
+    a2a.add_secondset(second_lis)
+    a2a.run_and_wait_end()
+
+if __name__ == "__main__":
+    if sys.version_info[1] >= 12:
+        print("Subinterpreters")
+        run_test(use_subinterpreters = True)
+    else:
+        print("Skip subinterpreters test because python version is < 3.12")
+    print()
+    print("Processes")
+    run_test(use_subinterpreters = False)
diff --git a/tests/python/pipe-with-multi-input-and-output.test.py b/tests/python/pipe-with-multi-input-and-output.test.py
diff --git a/tests/python/pipe-with-multi-output.test.py b/tests/python/pipe-with-multi-output.test.py
diff --git a/thesis/with-numpy.py b/thesis/with-numpy.py
