feat: set fastflow as module name

Author: domferr

README.md

@@ -3,7 +3,7 @@
 ### Farm
 
 ```py
-from fastflow_module import FFFarm, GO_ON
+from fastflow import FFFarm, GO_ON
 import sys
 
 class source():
@@ -51,7 +51,7 @@ farm.run_and_wait_end()
 ### Pipeline
 
 ```py
-from fastflow_module import FFPipeline, GO_ON
+from fastflow import FFPipeline, GO_ON
 
 # define a stage
 class stage():

benchmark/README.md

@@ -0,0 +1,26 @@
+# How to run the benchmarks
+
+1. From the root directory, create virtual environment via `python3.12 -m venv .venv`
+2. From the root directory, activate the virtual environment by running `source .venv/bin/activate`
+3. Build and install `busy_wait`. Read the README.md file inside busy_wait's directory
+4. Ensure you have built and installed `fastflow` Python API by running `make` from the root directory
+
+# Usage
+
+```
+usage: bench.py [-h] -tasks TASKS -workers WORKERS -ms MS -bytes BYTES [-blocking-mode] [-no-mapping] [-proc | -sub | -seq]
+
+Run a farm of <WORKERS> workers and <TASKS> tasks. Each task is <MS>ms long and has a size of <BYTES> bytes. Using subinterpreters or multiprocessing based strategy
+
+options:
+ -h, --help show this help message and exit
+ -tasks TASKS Number of tasks to process
+ -workers WORKERS Number of workers of the farm
+ -ms MS Duration, in milliseconds, of one task
+ -bytes BYTES The size, in bytes, of one task
+ -blocking-mode Enable blocking mode of the farm
+ -no-mapping Disable fastflow's mapping
+ -proc Use multiprocessing to process tasks
+ -sub Use subinterpreters to process tasks
+ -seq Run tasks sequentially
+```

benchmark/busy_wait/README.md

@@ -0,0 +1,5 @@
+# How to build
+
+1. ensure you have activated the virtual environment
+2. `cd` to busy_wait's directory
+3. run `./install.sh`

benchmark/busy_wait/install.sh

@@ -1,2 +1,3 @@
 rm -rf build
-python3.12 setup.py install
+pip install setuptools
+pip install .

benchmark/farm/bench.py

@@ -1,24 +1,8 @@
-from fastflow_module import FFFarm, GO_ON
+from fastflow import FFFarm, EOS, GO_ON
 import argparse
 import sys
 import busy_wait
 
-"""class DummyData:
- def __init__(self, size, val):
- self.astr = f"a string {val}"
- self.num = 1704 + val
- self.adict = {
- "val": 17 + val,
- "str": f"string inside a dictionary {val}"
- }
- self.atup = (4, 17, 16, val, 15, 30)
- self.aset = set([1, 2, 9, 6, val])
- 
- def __str__(self):
- return f"{self.astr}, {self.num}, {self.adict}, {self.atup}, {self.aset}"
- 
- __repr__ = __str__"""
-
 class DummyData:
  def __init__(self, data):
  self.data = [data]
@@ -35,7 +19,7 @@ def __init__(self, data_sample, n_tasks):
 
  def svc(self, *args):
  if self.n_tasks == 0:
- return
+ return EOS
  self.n_tasks -= 1
 
  return DummyData(self.data_sample)
@@ -75,8 +59,9 @@ def build_farm(n_tasks, task_ms, nworkers, data_sample, use_subinterpreters = Fa
 
  return farm
 
-def run_farm(n_tasks, task_ms, nworkers, data_sample, use_subinterpreters = False, blocking_mode = None, no_mapping = False, use_main_thread = False):
- print(f"run farm of {nworkers} workers and {n_tasks} tasks", file=sys.stderr)
+def run_farm(n_tasks, task_ms, nworkers, data_sample, data_size, use_subinterpreters = False, blocking_mode = None, no_mapping = False, use_main_thread = False):
+ strategy = "Not using any strategy" if use_main_thread else f"Using {'subinterpreters' if use_subinterpreters else 'processes'}-based strategy"
+ print(f"Running a farm of {nworkers} workers and {n_tasks} tasks. Each task is {task_ms}ms long and has a size of {data_size} bytes. {strategy}", file=sys.stderr)
  farm = build_farm(n_tasks, task_ms, nworkers, data_sample, use_subinterpreters, use_main_thread)
  if blocking_mode is not None:
  farm.blocking_mode(blocking_mode)
@@ -90,55 +75,30 @@ def get_data_sample(task_bytes):
  result = ascii_letters * int(task_bytes / len(ascii_letters))
  return result
 
-def print_res(title, res, args):
- print(title)
- print(f"res[{args.bytes}] = res.get({args.bytes}, []); res.get({args.bytes}).append(({args.workers}, {res})) # bytes =", args.bytes, "ms =", args.ms)
-
 if __name__ == "__main__":
- parser = argparse.ArgumentParser(description='Process some tasks.')
+ parser = argparse.ArgumentParser(description='Run a farm of <WORKERS> workers and <TASKS> tasks. Each task is <MS>ms long and has a size of <BYTES> bytes. Using subinterpreters or multiprocessing based strategy')
  parser.add_argument('-tasks', type=int, help='Number of tasks to process', required=True)
  parser.add_argument('-workers', type=int, help='Number of workers of the farm', required=True)
- parser.add_argument('-ms', type=int, help='Duration in milliseconds of one task', required=True)
+ parser.add_argument('-ms', type=int, help='Duration, in milliseconds, of one task', required=True)
  parser.add_argument('-bytes', type=int, help='The size, in bytes, of one task', required=True)
- parser.add_argument('-blocking-mode', help='The blocking mode of the farm', required=False, action='store_true', default=None)
+ parser.add_argument('-blocking-mode', help='Enable blocking mode of the farm', required=False, action='store_true', default=None)
  parser.add_argument('-no-mapping', help="Disable fastflow's mapping", required=False, action='store_true')
- group = parser.add_mutually_exclusive_group(required=False)
+ group = parser.add_mutually_exclusive_group(required=True)
  group.add_argument('-proc', action='store_true', help='Use multiprocessing to process tasks')
  group.add_argument('-sub', action='store_true', help='Use subinterpreters to process tasks')
  group.add_argument('-seq', action='store_true', help='Run tasks sequentially')
  args = parser.parse_args()
 
  # test the serialization to adjust the number of bytes
  data_sample = get_data_sample(args.bytes)
+ 
+ res = run_farm(args.tasks, args.ms, args.workers, data_sample, args.bytes, use_subinterpreters = args.sub and not args.proc, blocking_mode = args.blocking_mode, no_mapping = args.no_mapping, use_main_thread = args.seq)
+ print(f"res[{args.bytes}] = res.get({args.bytes}, []); res.get({args.bytes}).append(({args.workers}, {res})) # bytes =", args.bytes, "ms =", args.ms)
+
 
- if args.proc:
- processes = [[],[]]
- res = run_farm(args.tasks, args.ms, args.workers, data_sample, use_subinterpreters = False, blocking_mode = args.blocking_mode, no_mapping = args.no_mapping)
- print(f"done in {res}ms")
- processes[0].append(args.workers) # x
- processes[1].append(res) # y
-
- print_res("processes", res, args)
- elif args.sub:
- subinterpreters = [[],[]]
- res = run_farm(args.tasks, args.ms, args.workers, data_sample, use_subinterpreters = True, blocking_mode = args.blocking_mode, no_mapping = args.no_mapping)
- print(f"done in {res}ms")
- subinterpreters[0].append(args.workers) # x
- subinterpreters[1].append(res) # y
-
- print_res("subinterpreters", res, args)
- else:
- standard = [[],[]]
- res = run_farm(args.tasks, args.ms, args.workers, data_sample, use_subinterpreters = False, use_main_thread = True)
- print(f"done in {res}ms")
- standard[0].append(args.workers) # x
- standard[1].append(res) # y
-
- print_res("standard", res, args)
-
-
-# for i in 1 2 4 8 10 12 16 20 26 30 36 42 48 54 60 64; do for size in 1024 4096 8192 16384 32768 65536 524288 1048576; do echo $i $size; done; done
-# python3.12 benchmark/farm/bench.py -tasks 128 -workers 4 -ms 100 -bytes 512000 -sub
-# python3.12 benchmark/farm/bench.py -tasks 128 -workers 4 -ms 100 -bytes 320024 -sub 2>1 | grep subinterp
-
-# for i in 1 2 4 8 10 12 16 20 26 30 36 42 48 54 60 64; do for size in 1024 4096 8192 16384 32768 65536 524288 1048576; do python3.12 benchmark/farm/bench.py -tasks 512 -workers $i -ms 500 -bytes $size -sub 2>1 | grep subinterp; done; done
+"""
+Some examples to run this benchmark
+- python3.12 benchmark/farm/bench.py -tasks 128 -workers 4 -ms 100 -bytes 512000 -sub
+- python3.12 benchmark/farm/bench.py -tasks 128 -workers 4 -ms 100 -bytes 320024 -sub 2>1 | grep subinterp
+- for i in 1 2 4 8 10 12 16 20 26 30 36 42 48 54 60 64; do for size in 1024 4096 8192 16384 32768 65536 524288 1048576; do python3.12 benchmark/farm/bench.py -tasks 512 -workers $i -ms 500 -bytes $size -sub 2>1 | grep subinterp; done; done
+"""

fastflow_module.cpp

@@ -15,7 +15,7 @@
 
 /* initialization function */
 static int
-fastflow_module_exec(PyObject *module) 
+fastflow_exec(PyObject *module) 
 { 
  // add FFPipeline
  if (PyType_Ready(&py_ff_pipeline_type) < 0)
@@ -83,7 +83,7 @@ static PyMethodDef module_methods[] = {
 };
 
 static PyModuleDef_Slot module_slots[] = {
- {Py_mod_exec, (void*) fastflow_module_exec},
+ {Py_mod_exec, (void*) fastflow_exec},
 #if PY_MINOR_VERSION >= 12
  {Py_mod_multiple_interpreters, Py_MOD_PER_INTERPRETER_GIL_SUPPORTED},
 #endif
@@ -92,15 +92,15 @@ static PyModuleDef_Slot module_slots[] = {
 
 static struct PyModuleDef moduledef = {
  .m_base = PyModuleDef_HEAD_INIT,
- .m_name = "fastflow_module",
+ .m_name = "fastflow",
  .m_doc = "This is FastFlow's docstring",
  .m_size = 0,
  .m_methods = module_methods,
  .m_slots = module_slots,
 };
 
 PyMODINIT_FUNC
-PyInit_fastflow_module(void) {
+PyInit_fastflow(void) {
  return PyModuleDef_Init(&moduledef);
 }
 

include/process/base_process.hpp

@@ -12,6 +12,9 @@
 #include "object.h"
 #include "../py_ff_constant.hpp"
 
+#define SERIALIZED_EMPTY_TUPLE "(t."
+#define SERIALIZED_NONE "N."
+
 void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  Message message;
 
@@ -91,7 +94,7 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  CHECK_ERROR_THEN("PyModule_GetDict failure: ", cleanup_exit();)
 
  // if you access the methods from the module itself, replace it with the callback
- if (PyDict_SetItemString(globals, "fastflow_module", (PyObject*) callback) == -1) {
+ if (PyDict_SetItemString(globals, "fastflow", (PyObject*) callback) == -1) {
  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
@@ -121,18 +124,19 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  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 = "" 
+ .type = _const_result->ff_const == ff::FF_EOS ? MESSAGE_TYPE_EOS:MESSAGE_TYPE_GO_ON
  });
  if (err <= 0) handleError("[child] send constant response", cleanup_exit());
  } else {
+ std::string result_str = SERIALIZED_NONE;
  // serialize response
- auto result_str = pickl.pickle(py_result);
- CHECK_ERROR_THEN("[child] pickle result failure: ", cleanup_exit();)
+ if (py_result != Py_None) {
+ 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 = "" });
+ err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_RESPONSE, .data = result_str });
  if (err <= 0) handleError("[child] send response", cleanup_exit());
 
  Py_DECREF(py_result);
@@ -146,8 +150,6 @@ void process_body(int read_fd, int send_fd, bool isMultiOutput) {
  cleanup_exit();
 }
 
-#define SERIALIZED_EMPTY_TUPLE "(t."
-
 class base_process {
 public: 
  base_process(PyObject* node): node(node), send_fd(-1), read_fd(-1), registered_callback(NULL) {
@@ -175,18 +177,19 @@ class base_process {
  auto node_str = pickl.pickle(node);
  CHECK_ERROR_THEN("pickle node failure: ", return -1;)
 
- none_str = std::string(pickl.pickle(Py_None));
- CHECK_ERROR_THEN("pickle None failure: ", return -1;)
-
  int mainToChildFD[2]; // data to be sent from main process to child process
  int childToMainFD[2]; // data to be sent from child process to main process
  // create pipes
  if (pipe(mainToChildFD) == -1) {
- perror("pipe mainToChildFD");
+ std::string msg = "Failed to create pipe. ";
+ msg.append(strerror(errno));
+ PyErr_SetString(PyExc_Exception, msg.c_str());
  return -1;
  }
  if (pipe(childToMainFD) == -1) {
- perror("pipe childToMainFD");
+ std::string msg = "Failed to create pipe. ";
+ msg.append(strerror(errno));
+ PyErr_SetString(PyExc_Exception, msg.c_str());
  return -1;
  }
 
@@ -196,7 +199,9 @@ class base_process {
  PyOS_BeforeFork();
  pid = fork();
  if (pid == -1) {
- perror("fork failed");
+ std::string msg = "Failed to fork. ";
+ msg.append(strerror(errno));
+ PyErr_SetString(PyExc_Exception, msg.c_str());
  returnValue = -1;
  } else if (pid == 0) { // child
  PyOS_AfterFork_Child();
@@ -205,7 +210,9 @@ class base_process {
  close(childToMainFD[0]); // Close read end of childToMainFD
 
  process_body(mainToChildFD[0], childToMainFD[1], registered_callback != NULL);
- perror("[child] shouldn't be here...");
+ std::string msg = "[child] shouldn't be here... ";
+ msg.append(strerror(errno));
+ PyErr_SetString(PyExc_Exception, msg.c_str());
  returnValue = -1;
  } else { // parent
  PyOS_AfterFork_Parent();
@@ -219,7 +226,8 @@ class base_process {
  send_fd = mainToChildFD[1];
  read_fd = childToMainFD[0];
 
- int err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_REMOTE_PROCEDURE_CALL, .data = node_str, .f_name = "" });
+ // send serialized node
+ int err = sendMessage(read_fd, send_fd, { .data = node_str });
  if (err <= 0) handleError("send serialized node", returnValue = -1);
 
  if (err > 0 && has_svc_init) {
@@ -277,12 +285,11 @@ class base_process {
  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(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 = "" });
+ err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_RESPONSE, .data = result ? "t":"f" });
  if (err <= 0) {
  handleError("error sending ff_send_out_to response", );
  return NULL;
@@ -298,7 +305,7 @@ class base_process {
 
  // got response of svc
  if (response.type == MESSAGE_TYPE_EOS) return ff::FF_EOS;
- if (response.type == MESSAGE_TYPE_GO_ON || response.data.compare(none_str) == 0) {
+ if (response.type == MESSAGE_TYPE_GO_ON || response.data.compare(SERIALIZED_NONE) == 0) {
  return ff::FF_GO_ON;
  }
 
@@ -314,7 +321,7 @@ class base_process {
  }
 
  // 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 = "" });
+ int err = sendMessage(read_fd, send_fd, { .type = MESSAGE_TYPE_END_OF_LIFE });
 
  // Acquire the main GIL
  PyEval_RestoreThread(tstate);
@@ -347,7 +354,6 @@ class base_process {
  int send_fd;
  int read_fd;
  pid_t pid;
- std::string none_str;
  ff::ff_monode* registered_callback;
 };
 

include/process/messaging.hpp

@@ -18,8 +18,8 @@ enum message_type {
 
 struct Message {
  message_type type;
- std::string data;
- std::string f_name;
+ std::string data = "";
+ std::string f_name = "";
 };
 
 int sendMessage(int read_fd, int send_fd, const Message& message) {

include/py_ff_a2a.hpp

@@ -278,7 +278,7 @@ static PyType_Spec spec_py_ff_a2a = {
 
 PyTypeObject py_ff_a2a_type = {
  PyVarObject_HEAD_INIT(NULL, 0)
- .tp_name = "fastflow_module.FFAllToAll",
+ .tp_name = "fastflow.FFAllToAll",
  .tp_basicsize = sizeof(py_ff_a2a_object) + sizeof(ff::ff_a2a),
  .tp_itemsize = 0,
  .tp_dealloc = (destructor) py_ff_a2a_dealloc,

include/py_ff_callback.hpp

@@ -37,7 +37,7 @@ static PyMethodDef py_ff_callback_methods[] = {
 
 static PyTypeObject py_ff_callback_type = {
  PyVarObject_HEAD_INIT(NULL, 0)
- .tp_name = "fastflow_module.FFCallback",
+ .tp_name = "fastflow.FFCallback",
  .tp_basicsize = sizeof(py_ff_callback_object),
  .tp_itemsize = 0,
  .tp_flags = Py_TPFLAGS_DEFAULT,