[FIX] Fixing of persistent operators

mencagli · mencagli · commit bb8d3027122c · 2024-02-27T21:37:49.000+01:00
diff --git a/tests/rocksdb_tests/rocksdb_common.hpp b/tests/rocksdb_tests/rocksdb_common.hpp
@@ -467,7 +467,7 @@ class WinSink_Functor
         if (out) {
             received++;
             totalsum += (*out).value;
-            //std::cout << "Ricevuto risultato finestra wid: " << (*out).wid << ", chiave: " << (*out).key << ", valore: " << (*out).value << std::endl;
+            std::cout << "Ricevuto risultato finestra wid: " << (*out).wid << ", chiave: " << (*out).key << ", valore: " << (*out).value << std::endl;
         }
         else {
             // printf("Received: %ld results, total sum: %ld\n", received, totalsum);
diff --git a/tests/rocksdb_tests/test_rocksdb_6.cpp b/tests/rocksdb_tests/test_rocksdb_6.cpp
@@ -107,15 +107,6 @@ int main(int argc, char *argv[])
         cout << "|  | (" << source_degree << ") +-->+ (" << filter_degree << ") +-->+  (" << flatmap_degree << ") +-->+ (" << map_degree << ") +-->+  (" << kw_degree << ")   +-->+ (" << sink_degree << ") |  |" << endl;
         cout << "|  +-----+   +-----+   +------+   +-----+   +--------+   +-----+  |" << endl;
         cout << "+-----------------------------------------------------------------+" << endl;
-        auto tuple_serializer = [](tuple_t &t) -> std::string {
-            return std::to_string(t.key) + "," + std::to_string(t.value);
-        };
-        auto tuple_deserializer = [](std::string &s) -> tuple_t {
-            tuple_t t;
-            t.key = atoi(s.substr(0, s.find(",")).c_str());
-            t.value = atoi(s.substr(s.find(",")+1, s.length()-1).c_str());
-            return t;
-        };
         auto result_serializer = [](result_t &r) -> std::string {
             return std::to_string(r.key) + "," + std::to_string(r.value) + ";" + std::to_string(r.wid);
         };
@@ -160,7 +151,6 @@ int main(int argc, char *argv[])
                                     .withParallelism(kw_degree)
                                     .withKeyBy([](const tuple_t &t) -> size_t { return t.key; })
                                     .withCBWindows(win_len, win_slide)
-                                    .withTupleSerializerAndDeserializer(tuple_serializer, tuple_deserializer)
                                     .withResultSerializerAndDeserializer(result_serializer, result_deserializer)
                                     .build();
 #else
diff --git a/tests/rocksdb_tests/test_rocksdb_8.cpp b/tests/rocksdb_tests/test_rocksdb_8.cpp
@@ -108,15 +108,6 @@ int main(int argc, char *argv[])
         cout << "|  | (" << source_degree << ") +-->+ (" << filter_degree << ") +-->+  (" << flatmap_degree << ") +-->+ (" << map_degree << ") +-->+  (" << kw_degree << ")   +-->+ (" << sink_degree << ") |  |" << endl;
         cout << "|  +-----+   +-----+   +------+   +-----+   +--------+   +-----+  |" << endl;
         cout << "+-----------------------------------------------------------------+" << endl;
-        auto tuple_serializer = [](tuple_t &t) -> std::string {
-            return std::to_string(t.key) + "," + std::to_string(t.value);
-        };
-        auto tuple_deserializer = [](std::string &s) -> tuple_t {
-            tuple_t t;
-            t.key = atoi(s.substr(0, s.find(",")).c_str());
-            t.value = atoi(s.substr(s.find(",")+1, s.length()-1).c_str());
-            return t;
-        };
         auto result_serializer = [](result_t &r) -> std::string {
             return std::to_string(r.key) + "," + std::to_string(r.value) + ";" + std::to_string(r.wid);
         };
@@ -165,7 +156,6 @@ int main(int argc, char *argv[])
                                     .withParallelism(kw_degree)
                                     .withKeyBy([](const tuple_t &t) -> size_t { return t.key; })
                                     .withTBWindows(microseconds(win_len), microseconds(win_slide))
-                                    .withTupleSerializerAndDeserializer(tuple_serializer, tuple_deserializer)
                                     .withResultSerializerAndDeserializer(result_serializer, result_deserializer)
                                     .build();
 #else
diff --git a/wf/persistent/builders_rocksdb.hpp b/wf/persistent/builders_rocksdb.hpp
@@ -1273,7 +1273,7 @@ class P_Keyed_Windows_Builder: public P_Basic_Builder<P_Keyed_Windows_Builder, w
     keyextr_func_t key_extr = [](const tuple_t &t) -> key_t { return key_t(); }; // key extractor
     bool isKeyBySet = false; // true if a key extractor has been provided
     size_t frag_bytes = sizeof(tuple_t) * 16; // size in bytes of each archive fragment of the stream
-    bool results_in_memory = true; // flag stating if results must be kepts on memory or on RocksDB
+    bool results_in_memory = true; // flag stating if results must be kepts in memory or on RocksDB
     bool isTupleFunctions = false; // flag stating if the tuple serializer/deserializer have been provided
     bool isResultFunctions = false; // flag stating if the result serializer/deserializer have been provided
     uint64_t win_len=0; // window length in number of tuples or in time units
@@ -1428,6 +1428,7 @@ class P_Keyed_Windows_Builder: public P_Basic_Builder<P_Keyed_Windows_Builder, w
         result_serialize = _result_serialize;
         result_deserialize = _result_deserialize;
         isResultFunctions = true;
+        results_in_memory = false;
         return *this;
     }
 
@@ -1452,6 +1453,25 @@ class P_Keyed_Windows_Builder: public P_Basic_Builder<P_Keyed_Windows_Builder, w
             std::cerr << RED << "WindFlow Error: P_Keyed_Windows with paralellism > 1 requires a key extractor" << DEFAULT_COLOR << std::endl;
             exit(EXIT_FAILURE);
         }
+        // check the presence of the tuple/result serializer/deserializer according to type of the window processing logic
+        if constexpr (std::is_invocable<decltype(func), const Iterable<tuple_t> &, result_t &>::value ||
+                      std::is_invocable<decltype(func), const Iterable<tuple_t> &, result_t &, RuntimeContext &>::value) {
+            if (!isTupleFunctions) {
+                std::cerr << RED << "WindFlow Error: P_Keyed_Windows instantiated with a non-incremental logic without tuple serializer/serializer" << DEFAULT_COLOR << std::endl;
+                exit(EXIT_FAILURE);
+            }
+        }
+        else if constexpr (std::is_invocable<decltype(func), const tuple_t &, result_t &>::value ||
+                           std::is_invocable<decltype(func), const tuple_t &, result_t &, RuntimeContext &>::value) {
+            if (isTupleFunctions) {
+                std::cerr << RED << "WindFlow Error: P_Keyed_Windows receives tuple serializer/deserializer with an incremental logic" << DEFAULT_COLOR << std::endl;
+                exit(EXIT_FAILURE);
+            }
+            if (!isResultFunctions) {
+                std::cerr << RED << "WindFlow Error: P_Keyed_Windows instantiated with an incremental logic without result serializer/serializer" << DEFAULT_COLOR << std::endl;
+                exit(EXIT_FAILURE);
+            }
+        }
         return p_keyed_wins_t(func,
                               key_extr,
                               this->parallelism,
diff --git a/wf/persistent/p_window_replica.hpp b/wf/persistent/p_window_replica.hpp
@@ -230,6 +230,7 @@ class P_Window_Replica: public Basic_Replica
         return final_range;
     }
 
+#if 0
     // getDistance method
     size_t getDistance(const wrapper_t &_w1,
                        const wrapper_t &_w2,
@@ -254,6 +255,7 @@ class P_Window_Replica: public Basic_Replica
         its.second = history_buffer.end();
         return std::distance(its.first, its.second);
     }
+#endif
 
     // getEnd method
     input_iterator_t getEnd(Key_Descriptor &_kd)
@@ -336,17 +338,6 @@ class P_Window_Replica: public Basic_Replica
         delete mydb_results;
     }
 
-    /*
-        template<typename X>
-        rocksdb::DB *get_operator_internal_db()
-        {
-            if (std::is_same<X, tuple_t>::value) {
-                return mydb_wrappers->get_internal_db();
-            }
-            return mydb_results->get_internal_db();
-        }
-    */
-
     // svc (utilized by the FastFlow runtime)
     void *svc(void *_in) override
     {
@@ -504,9 +495,9 @@ class P_Window_Replica: public Basic_Replica
                             (this->context).setContextParameters(_timestamp, _watermark); // set the parameter of the RuntimeContext
                             func(iter, res, this->context);
                         }
-                    }
-                    if (t_s) { // purge tuples from the archive
-                        purge(*t_s, key_d, key);
+                        if (t_s) { // purge tuples from the archive
+                            purge(*t_s, key_d, key);
+                        }
                     }
                     cnt_fired++;
                     key_d.last_lwid++;
diff --git a/wf/window_replica.hpp b/wf/window_replica.hpp
@@ -322,9 +322,9 @@ class Window_Replica: public Basic_Replica
                             (this->context).setContextParameters(_timestamp, _watermark); // set the parameter of the RuntimeContext
                             func(iter, win.getResult(), this->context);
                         }
-                    }
-                    if (t_s) { // purge tuples from the archive
-                        (key_d.archive).purge(*t_s);
+                        if (t_s) { // purge tuples from the archive
+                            (key_d.archive).purge(*t_s);
+                        }
                     }
                     cnt_fired++;
                     key_d.last_lwid++;

Original file line number	Diff line number	Diff line change
`@@ -467,7 +467,7 @@ class WinSink_Functor`
`467`	`467`	`if (out) {`
`468`	`468`	`received++;`
`469`	`469`	`totalsum += (*out).value;`
`470`		`- //std::cout << "Ricevuto risultato finestra wid: " << (out).wid << ", chiave: " << (out).key << ", valore: " << (*out).value << std::endl;`
	`470`	`+ std::cout << "Ricevuto risultato finestra wid: " << (out).wid << ", chiave: " << (out).key << ", valore: " << (*out).value << std::endl;`
`471`	`471`	`}`
`472`	`472`	`else {`
`473`	`473`	`// printf("Received: %ld results, total sum: %ld\n", received, totalsum);`
Original file line number	Diff line number	Diff line change
`@@ -322,9 +322,9 @@ class Window_Replica: public Basic_Replica`
`322`	`322`	`(this->context).setContextParameters(_timestamp, _watermark); // set the parameter of the RuntimeContext`
`323`	`323`	`func(iter, win.getResult(), this->context);`
`324`	`324`	`}`
`325`		`- }`
`326`		`- if (t_s) { // purge tuples from the archive`
`327`		`- (key_d.archive).purge(*t_s);`
	`325`	`+ if (t_s) { // purge tuples from the archive`
	`326`	`+ (key_d.archive).purge(*t_s);`
	`327`	`+ }`
`328`	`328`	`}`
`329`	`329`	`cnt_fired++;`
`330`	`330`	`key_d.last_lwid++;`