[FIX] changed the code to avoid some underlying bugs of FastFlow using A2As

Author: mencagli

wf/accumulator.hpp

@@ -39,7 +39,6 @@
 #include<unordered_map>
 #include<ff/node.hpp>
 #include<ff/pipeline.hpp>
-#include<ff/multinode.hpp>
 #include<ff/farm.hpp>
 #include<basic.hpp>
 #include<context.hpp>
@@ -82,7 +81,7 @@ class Accumulator: public ff::ff_farm, public Basic_Operator
     size_t parallelism; // internal parallelism of the Accumulator
     bool used; // true if the Accumulator has been added/chained in a MultiPipe
     // class Accumulator_Node
-    class Accumulator_Node: public ff::ff_minode_t<tuple_t, result_t>
+    class Accumulator_Node: public ff::ff_node_t<tuple_t, result_t>
     {
 private:
         acc_func_t acc_func; // reduce/fold function

wf/filter.hpp

@@ -38,7 +38,6 @@
 #include<string>
 #include<ff/node.hpp>
 #include<ff/pipeline.hpp>
-#include<ff/multinode.hpp>
 #include<ff/farm.hpp>
 #include<basic.hpp>
 #include<context.hpp>
@@ -87,7 +86,7 @@ class Filter: public ff::ff_farm, public Basic_Operator
     bool keyed; // flag stating whether the Filter is configured with keyBy or not
     bool used; // true if the Filter has been added/chained in a MultiPipe
     // class Filter_Node
-    class Filter_Node: public ff::ff_minode_t<tuple_t, result_t>
+    class Filter_Node: public ff::ff_node_t<tuple_t, result_t>
     {
 private:
         filter_func_t filter_func; // filter function (with boolean return type)

wf/flatmap.hpp

@@ -38,7 +38,6 @@
 #include<string>
 #include<ff/node.hpp>
 #include<ff/pipeline.hpp>
-#include<ff/multinode.hpp>
 #include<ff/farm.hpp>
 #include<basic.hpp>
 #include<shipper.hpp>
@@ -80,7 +79,7 @@ class FlatMap: public ff::ff_farm, public Basic_Operator
     bool keyed; // flag stating whether the FlatMap is configured with keyBy or not
     bool used; // true if the FlatMap has been added/chained in a MultiPipe
     // class FlatMap_Node
-    class FlatMap_Node: public ff::ff_minode_t<tuple_t, result_t>
+    class FlatMap_Node: public ff::ff_node_t<tuple_t, result_t>
     {
 private:
         flatmap_func_t flatmap_func; // flatmap function

wf/map.hpp

@@ -38,7 +38,6 @@
 #include<string>
 #include<ff/node.hpp>
 #include<ff/pipeline.hpp>
-#include<ff/multinode.hpp>
 #include<ff/farm.hpp>
 #include<basic.hpp>
 #include<context.hpp>
@@ -83,7 +82,7 @@ class Map: public ff::ff_farm, public Basic_Operator
     bool keyed; // flag stating whether the Map is configured with keyBy or not
     bool used; // true if the Map has been added/chained in a MultiPipe
     // class Map_Node
-    class Map_Node: public ff::ff_minode_t<tuple_t, result_t>
+    class Map_Node: public ff::ff_node_t<tuple_t, result_t>
     {
 private:
         map_func_ip_t func_ip; // in-place map function

wf/multipipe.hpp

@@ -268,6 +268,10 @@ class MultiPipe: public ff::ff_pipeline
                     collector_t *collector = new collector_t(_ordering, atomic_num_dropped);
                     combine_with_firststage(*stage, collector, true); // add the ordering_node / kslack_node
                 }
+                else {
+                    dummy_mi *collector = new dummy_mi();
+                    combine_with_firststage(*stage, collector, true); // dummy multi-input node
+                }
                 first_set.push_back(stage);
             }
             matrioska->add_firstset(first_set, 0, true);
@@ -318,6 +322,10 @@ class MultiPipe: public ff::ff_pipeline
                     collector_t *collector = new collector_t(_ordering, atomic_num_dropped);
                     combine_with_firststage(*stage, collector, true); // add the ordering_node / kslack_node
                 }
+                else {
+                    dummy_mi *collector = new dummy_mi();
+                    combine_with_firststage(*stage, collector, true); // dummy multi-input node
+                }
                 first_set.push_back(stage);
             }
             matrioska->add_firstset(first_set, 0, true);

wf/sink.hpp

@@ -45,7 +45,6 @@
 #include<ff/node.hpp>
 #include<ff/combine.hpp>
 #include<ff/pipeline.hpp>
-#include<ff/multinode.hpp>
 #include<ff/farm.hpp>
 #include<basic.hpp>
 #include<context.hpp>
@@ -90,7 +89,7 @@ class Sink: public ff::ff_farm, public Basic_Operator
     bool keyed; // flag stating whether the Sink is configured with keyBy or not
     bool used; // true if the Sink has been added/chained in a MultiPipe
     // class Sink_Node
-    class Sink_Node: public ff::ff_minode_t<tuple_t>
+    class Sink_Node: public ff::ff_node_t<tuple_t>
     {
     private:
         sink_func_t sink_func; // sink function (receiving a reference to an optional containing the input)

wf/test_mp_gpu_kff_cb.cpp

@@ -41,7 +41,6 @@
 #include<unordered_map>
 #include<math.h>
 #include<ff/node.hpp>
-#include<ff/multinode.hpp>
 #include<meta.hpp>
 #include<window.hpp>
 #include<context.hpp>
@@ -55,7 +54,7 @@ namespace wf {
 
 // Win_Seq class
 template<typename tuple_t, typename result_t, typename input_t>
-class Win_Seq: public ff::ff_minode_t<input_t, result_t>
+class Win_Seq: public ff::ff_node_t<input_t, result_t>
 {
 public:
     // type of the non-incremental window processing function
@@ -485,9 +484,9 @@ class Win_Seq: public ff::ff_minode_t<input_t, result_t>
     {
         eos_received++;
         // check the number of received EOS messages
-        if ((eos_received != this->get_num_inchannels()) && (this->get_num_inchannels() != 0)) { // workaround due to FastFlow
-            return;
-        }
+        //if ((eos_received != this->get_num_inchannels()) && (this->get_num_inchannels() != 0)) { // workaround due to FastFlow
+        //    return;
+        //}
         // iterate over all the keys
         for (auto &k: keyMap) {
             auto &wins = (k.second).wins;
@@ -568,13 +567,13 @@ class Win_Seq: public ff::ff_minode_t<input_t, result_t>
     // method to start the node execution asynchronously
     int run(bool) override
     {
-        return ff::ff_minode::run();
+        return ff::ff_node::run();
     }
 
     // method to wait the node termination
     int wait() override
     {
-        return ff::ff_minode::wait();
+        return ff::ff_node::wait();
     }
 };
 

wf/win_seq.hpp

@@ -42,7 +42,6 @@
 #include<unordered_map>
 #include<math.h>
 #include<ff/node.hpp>
-#include<ff/multinode.hpp>
 #include<meta.hpp>
 #include<window.hpp>
 #include<meta_gpu.hpp>
@@ -82,7 +81,7 @@ __global__ void ComputeBatch_Kernel(void *input_data,
 
 // Win_Seq_GPU class
 template<typename tuple_t, typename result_t, typename win_F_t, typename input_t>
-class Win_Seq_GPU: public ff::ff_minode_t<input_t, result_t>
+class Win_Seq_GPU: public ff::ff_node_t<input_t, result_t>
 {
 private:
     // type of the stream archive used by the Win_Seq_GPU node
@@ -614,9 +613,9 @@ class Win_Seq_GPU: public ff::ff_minode_t<input_t, result_t>
     {
         eos_received++;
         // check the number of received EOS messages
-        if ((eos_received != this->get_num_inchannels()) && (this->get_num_inchannels() != 0)) { // workaround due to FastFlow
-            return;
-        }
+        //if ((eos_received != this->get_num_inchannels()) && (this->get_num_inchannels() != 0)) { // workaround due to FastFlow
+        //    return;
+        //}
         // emit results of the previously running kernel on the GPU
         waitAndFlush();
         // allocate on the CPU the scratchpad_memory
@@ -709,13 +708,13 @@ class Win_Seq_GPU: public ff::ff_minode_t<input_t, result_t>
     // method to start the node execution asynchronously
     int run(bool) override
     {
-        return ff::ff_minode::run();
+        return ff::ff_node::run();
     }
 
     // method to wait the node termination
     int wait() override
     {
-        return ff::ff_minode::wait();
+        return ff::ff_node::wait();
     }
 };