[ADD] Add files for release 4.2.2

Author: mencagli

AUTHORS

@@ -19,3 +19,4 @@ Alberto Ottimo, contributor
 Matteo Della Bartola, contributor
 Simone Frassinelli, contributor
 Yuriy Rymarchuk, contributor
+Andrea Filippi, contributor

tests/join_tests/join_common.hpp

@@ -42,17 +42,6 @@ struct tuple_t
     int64_t value;
 };
 
-template<>
-struct std::hash<tuple_t>
-{
-    size_t operator()(const tuple_t &t) const
-    {
-        size_t h1 = std::hash<int64_t>()(t.value);
-        size_t h2 = std::hash<size_t>()(t.key);
-        return h1 ^ h2;
-    }
-};
-
 struct res_t
 {
     size_t key;

wf/interval_join.hpp

@@ -78,22 +78,16 @@ class IJoin_Replica: public Basic_Replica
     using iterator_t = typename container_t::iterator; // iterator type for accessing wrapped tuples in the archive
     using compare_func_t = std::function<bool(const wrapper_t &, const uint64_t &)>; // function type to compare wrapped tuple to an uint64
 
-    /** 
-     *  @brief Structure to store statistics about an archive.
-     */ 
-    struct Archive_Stats
+    struct Archive_Stats // structure to store statistics about an archive
     {
-        // Total size of the archive.
-        size_t size;
+        size_t size; // total size of the archive
+        uint64_t size_count; // number of times the size of the archive has been recorded
 
-        // Number of times the size of the archive has been recorded
-        uint64_t size_count;
-
-        // Default constructor for Archive_Stats
+        // Constructor
         Archive_Stats():
                       size(0),
                       size_count(0) {}
- 
+
         // Records the size of the archive
         void recordSize(uint64_t _size)
         {
@@ -114,13 +108,15 @@ class IJoin_Replica: public Basic_Replica
     {
         JoinArchive<tuple_t, compare_func_t> archiveA; // archive of stream A tuples of this key
         JoinArchive<tuple_t, compare_func_t> archiveB; // archive of stream B tuples of this key
-        Archive_Stats archive_metrics;
+        Archive_Stats archive_metrics; // archive of statistics for this key
+        uint64_t partitioning_counter; // counter used in DP mode to establish which replica will save the given tuple
 
         // Constructor
         Key_Descriptor(compare_func_t _compare_func):
                        archiveA(_compare_func),
                        archiveB(_compare_func),
-                       archive_metrics(Archive_Stats()) {}
+                       archive_metrics(Archive_Stats()),
+                       partitioning_counter(0) {}
 
         // recordSize method
         void recordSize()
@@ -139,21 +135,6 @@ class IJoin_Replica: public Basic_Replica
     size_t id_inner; // id_inner value
     size_t num_inner; // num_inner value
 
-    // Calculates the FNV-1a hash value for the given key
-    const size_t fnv1a_hash(const void* key,
-                            const size_t len = sizeof(uint64_t))
-    {
-        const char* data = (char *)key;
-        const size_t prime = 0x1000193;
-        size_t hash = 0x811c9dc5;
-        for(int i = 0; i<len; i++) {
-            uint8_t value = data[i];
-            hash = hash ^ value;
-            hash *= prime;
-        }
-        return hash;
-    }
-
     // Checks if the given Join_Stream_t is Stream A
     bool isStreamA(Join_Stream_t stream) const
     {
@@ -334,21 +315,9 @@ class IJoin_Replica: public Basic_Replica
             insertIntoBuffer(key_d, wrapper_t(_tuple, _timestamp), _tag);
         }
         else if (joinMode == Join_Mode_t::DP) {
-            if constexpr(if_defined_hash<tuple_t>) {
-                // compute the hash index of the tuple given a defined hash function specialization for the tuple_t
-                size_t hash = std::hash<tuple_t>()(_tuple);
-                size_t hash_idx = (hash % num_inner);
-                if (hash_idx == id_inner) {
-                    insertIntoBuffer(key_d, wrapper_t(_tuple, _timestamp), _tag);
-                }
-            }
-            else {
-                // compute the hash index of the tuple using FNV-1a hash function using the timestamp
-                size_t hash = fnv1a_hash(&_timestamp);
-                size_t hash_idx = (hash % num_inner);
-                if (hash_idx == id_inner) {
-                    insertIntoBuffer(key_d, wrapper_t(_tuple, _timestamp), _tag);
-                }
+            key_d.partitioning_counter++;
+            if (key_d.partitioning_counter % num_inner == id_inner) {
+                insertIntoBuffer(key_d, wrapper_t(_tuple, _timestamp), _tag);
             }
         }
         if (this->execution_mode == Execution_Mode_t::DEFAULT) {
@@ -363,17 +332,6 @@ class IJoin_Replica: public Basic_Replica
                 last_time = _timestamp;
             }
         }
-#if defined (WF_JOIN_MEASUREMENT)
-        // Measure the size of the archives every 200ms
-        uint64_t delta = (current_time_nsecs() - last_measured_size_time) / 1e06; //ms
-        if (delta >= 200) {
-            for (auto &k: keyMap) {
-                Key_Descriptor &key_m_d = (k.second);
-                (key_m_d.recordSize());
-            }
-            last_measured_size_time = current_time_nsecs();
-        }
-#endif
     }
 
     double getArchiveMeanSize() const
@@ -576,61 +534,11 @@ class Interval_Join: public Basic_Operator
     // Destructor
     ~Interval_Join() override
     {
-#if defined(WF_JOIN_MEASUREMENT)
-        if (this->isTerminated()) {
-            printArchivePerKeyStats();
-        }
-#endif
         for (auto *r: replicas) { // delete all the replicas
             delete r;
         }
     }
 
-    /** 
-     *  @brief Print statistics of the archive per key.
-     *  
-     *  This function calculates and prints various statistics about the archive per key.
-     *  It calculates the mean archive size for each replica and checks the distribution
-     *  of the archive sizes and determines if it is balanced or not.
-     *  
-     *  @note This function assumes that the `replicas` vector is already populated with
-     *  valid replica objects.
-     *  
-     *  @note The balance check is determined based on the coefficient of variation (cv) value.
-     *        If the cv is less than 20, it is considered balanced and marked with a checkmark,
-     *        otherwise it is considered unbalanced and marked with a cross.
-     *  
-     *  @note The function uses the `std::cout` stream to print the statistics.
-     */ 
-    void printArchivePerKeyStats()
-    {
-        std::cout << "***" << std::endl;
-        std::cout << "Archive Stats: " << std::endl;
-        uint64_t num_replicas = replicas.size();
-        double acc_mean = 0.0;
-        int i = 0;
-        for (auto *r: replicas) {
-            auto mean = r->getArchiveMeanSize();
-            std::cout << (i+1) << " Replica mean -> " << mean << std::endl;
-            acc_mean += mean;
-            i++;
-        }
-        double mean_size = acc_mean / num_replicas;
-        double size_in_mb = mean_size * sizeof(tuple_t) / 1024;
-        std::cout << "Global Mean Archive Size -> " << mean_size << " | " << size_in_mb << " KB" << std::endl;
-        // Check distribution
-        double variance = 0;
-        for (auto *r: replicas) {
-            variance += std::pow(r->getArchiveMeanSize() - mean_size, 2);
-        }
-        variance /= num_replicas;
-        double cv = variance != 0 ? sqrt(variance) / mean_size * 100 : 0.0; //coefficient of variation
-        std::string check_balance = cv < 20 ? " ✔ " : " ✘ ";
-        std::cout << std::fixed << std::setprecision(2);
-        std::cout << "Variance -> " << variance << " | Coefficient of variation -> " << cv << "| Balanced Distribution ->" << check_balance << std::endl;
-        std::cout << "***" << std::endl;
-    }
-
     /** 
      *  \brief Get the type of the Interval Join as a string
      *  \return type of the Interval Join

wf/persistent/builders_rocksdb.hpp

@@ -1,5 +1,5 @@
 /**************************************************************************************
- *  Copyright (c) 2024- Gabriele Mencagli and Simone Frassinelli
+ *  Copyright (c) 2019- Gabriele Mencagli and Simone Frassinelli
  *  
  *  This file is part of WindFlow.
  *  
@@ -1283,6 +1283,7 @@ class P_Keyed_Windows_Builder: public P_Basic_Builder<P_Keyed_Windows_Builder, w
     uint64_t slide_len=0; // slide length in number of tuples or in time units
     uint64_t lateness=0; // lateness in time units
     Win_Type_t winType=Win_Type_t::CB; // window type (CB or TB)
+    size_t cacheCapacity = 0; // capacity of the internal cache
 
 public:
     /** 
@@ -1386,6 +1387,7 @@ class P_Keyed_Windows_Builder: public P_Basic_Builder<P_Keyed_Windows_Builder, w
         new_builder.slide_len = slide_len;
         new_builder.lateness = lateness;
         new_builder.winType = winType;
+        new_builder.cacheCapacity = cacheCapacity;
         return new_builder;
     }
 
@@ -1451,6 +1453,18 @@ class P_Keyed_Windows_Builder: public P_Basic_Builder<P_Keyed_Windows_Builder, w
         return *this;
     }
 
+    /** 
+     *  \brief Set the cache capacity
+     *  
+     *  \param _cacheCapacity number of keys to be cached  by each replica of this operator
+     *  \return a reference to the builder object
+     */ 
+    auto &withCacheCapacity(size_t _cacheCapacity)
+    {
+        cacheCapacity = _cacheCapacity;
+        return *this;
+    }
+
     /** 
      *  \brief Create the P_Keyed_Windows
      *  
@@ -1505,7 +1519,8 @@ class P_Keyed_Windows_Builder: public P_Basic_Builder<P_Keyed_Windows_Builder, w
                               win_len,
                               slide_len,
                               lateness,
-                              winType);
+                              winType,
+                              cacheCapacity);
     }
 };
 

wf/persistent/cache/cache.hpp

@@ -0,0 +1,76 @@
+/**************************************************************************************
+ *  Copyright (c) 2019- Gabriele Mencagli and Andrea Filippi
+ *  
+ *  This file is part of WindFlow.
+ *  
+ *  WindFlow is free software dual licensed under the GNU LGPL or MIT License.
+ *  You can redistribute it and/or modify it under the terms of the
+ *    * GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License, or
+ *      (at your option) any later version
+ *    OR
+ *    * MIT License: https://github.com/ParaGroup/WindFlow/blob/master/LICENSE.MIT
+ *  
+ *  WindFlow is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU Lesser General Public License for more details.
+ *  You should have received a copy of the GNU Lesser General Public License and
+ *  the MIT License along with WindFlow. If not, see <http://www.gnu.org/licenses/>
+ *  and <http://opensource.org/licenses/MIT/>.
+ **************************************************************************************
+ */
+
+/** 
+ *  @file    cache.hpp
+ *  @author  Gabriele Mencagli and Andrea Filippi
+ *  
+ *  @brief class defining the general interface of a cache used by the 
+ *         P_Keyed_Windows with non-incremental processing
+ *  
+ *  @section Cache (Description)
+ *  
+ *  Abstract class of a cache used by the P_Keyed_Windows operator with non-incremental
+ *  processing functions.
+ */ 
+
+#ifndef CACHE_H
+#define CACHE_H
+
+#include<optional>
+
+namespace wf {
+
+// class Cache
+template<typename key_t, typename value_t>
+class Cache
+{
+public:
+    // Destructor
+    virtual ~Cache() = default;
+
+    // Insert a new value associated with key in the cache
+    virtual void put(const key_t &_key, const value_t &_value) = 0;
+
+    // Read the value associated with a key (returns std::nullopt otherwise)
+    virtual std::optional<value_t> get(const key_t &_key) = 0;
+
+    // Remove a key entry from the cache
+    virtual void remove(const key_t &_key) = 0;
+
+    // Empty the cache
+    virtual void clear() = 0;
+
+    // Check if a key is present in the cache
+    virtual bool exists(const key_t &_key) const = 0;
+
+    // Get the number of elements in the cache
+    virtual size_t size() const = 0;
+
+    // Get the maximum capacity of the cache
+    virtual size_t capacity() const = 0;
+};
+
+} // namespace wf
+
+#endif