Added a class that does a combined net send of data for both ADLocalFuncStatistics and ADLocalCounterStatistics + unit test

Used the above to combine the two send calls into a single call
Fixed (recoverable) errors occuring when assigning comm and counter events whose timestamps are later than the function exit time
Added comparison operators for several pserver data classes
In Chimbuko main class, moved code that gathers and sends data to PS into a separate method

Author: giltirn

app/pserver.cpp

@@ -187,8 +187,7 @@ int main (int argc, char ** argv){
 
     net.add_payload(new NetPayloadUpdateParams(param, args.freeze_params));
     net.add_payload(new NetPayloadGetParams(param));
-    net.add_payload(new NetPayloadUpdateAnomalyStats(&global_func_stats));
-    net.add_payload(new NetPayloadUpdateCounterStats(&global_counter_stats));
+    net.add_payload(new NetPayloadRecvCombinedADdata(&global_func_stats, &global_counter_stats));
     net.add_payload(new NetPayloadGlobalFunctionIndexMapBatched(&global_func_index_map));
     net.init(nullptr, nullptr, args.nt);
 

include/chimbuko/AD.hpp

@@ -10,6 +10,7 @@
 #include "chimbuko/ad/ADNetClient.hpp"
 #include "chimbuko/ad/ADLocalFuncStatistics.hpp"
 #include "chimbuko/ad/ADLocalCounterStatistics.hpp"
+#include "chimbuko/ad/ADcombinedPSdata.hpp"
 #include "chimbuko/ad/ADProvenanceDBclient.hpp"
 #include "chimbuko/ad/ADMetadataParser.hpp"
 #include "chimbuko/ad/ADglobalFunctionIndexMap.hpp"

include/chimbuko/ad/ADLocalCounterStatistics.hpp

@@ -140,6 +140,16 @@ namespace chimbuko{
      */
     int getIOstep() const{ return m_step; }
 
+    /**
+     * @brief Comparison operator
+     */
+    bool operator==(const ADLocalCounterStatistics &r) const{ return m_program_idx == r.m_program_idx && m_step == r.m_step && m_stats == r.m_stats; }
+
+    /**
+     * @brief Inequality operator
+     */
+    bool operator!=(const ADLocalCounterStatistics &r) const{ return !( *this == r ); }
+
   protected:
     /**
      * @brief update (send) counter statistics gathered during this io step to the connected parameter server

include/chimbuko/ad/ADcombinedPSdata.hpp

@@ -0,0 +1,87 @@
+#pragma once
+#include <chimbuko_config.h>
+#include <chimbuko/ad/ADLocalFuncStatistics.hpp>
+#include <chimbuko/ad/ADLocalCounterStatistics.hpp>
+
+namespace chimbuko{
+  /** 
+   * @brief A class that combines all data that is sent to the PS into a single send transmission
+   */
+  class ADcombinedPSdata{
+  public:
+    ADcombinedPSdata(ADLocalFuncStatistics &func_stats, ADLocalCounterStatistics &counter_stats, PerfStats* perf=nullptr): 
+      m_func_stats(func_stats), m_counter_stats(counter_stats), m_perf(perf){}
+    
+    ADLocalCounterStatistics &get_counter_stats(){ return m_counter_stats; }
+    const ADLocalCounterStatistics &get_counter_stats() const{ return m_counter_stats; }
+
+    ADLocalFuncStatistics &get_func_stats(){ return m_func_stats; }
+    const ADLocalFuncStatistics &get_func_stats() const{ return m_func_stats; }
+
+    /**
+     * @brief The State object is what is serialized for the communication
+     */
+    struct State{
+      ADLocalFuncStatistics::State func_stats_state;
+      ADLocalCounterStatistics::State counter_stats_state;
+
+      /**
+       * @brief Serialize using cereal
+       */
+      template<class Archive>
+      void serialize(Archive & archive){
+	archive(func_stats_state , counter_stats_state);
+      }
+
+      /**
+       * Serialize into Cereal portable binary format
+       */
+      std::string serialize_cerealpb() const;
+      
+      /**
+       * Serialize from Cereal portable binary format
+       */     
+      void deserialize_cerealpb(const std::string &strstate);
+    };
+
+    /**
+     * @brief Get the current state as a state object
+     *
+     * The string dump of this object is the serialized form sent to the parameter server
+     */    
+    State get_state() const;
+
+
+    /**
+     * @brief Set the internal variables to the given state object
+     */
+    void set_state(const State &s);
+
+
+    /**
+     * @brief Serialize this class for communication over the network
+     */
+    std::string net_serialize() const;
+
+    /**
+     * @brief Unserialize this class after communication over the network
+     */
+    void net_deserialize(const std::string &s);
+      
+      
+    /**
+     * @brief Send the data to the pserver
+     * @param net_client The network client object
+     * @return std::pair<size_t, size_t> [sent, recv] message size
+     */
+    std::pair<size_t, size_t> send(ADNetClient &net_client) const;
+  
+
+  private:
+    ADLocalFuncStatistics &m_func_stats;
+    ADLocalCounterStatistics &m_counter_stats;
+    PerfStats* m_perf;
+  };
+
+
+}

include/chimbuko/chimbuko.hpp

@@ -194,6 +194,12 @@ namespace chimbuko {
 				  const unsigned long last_event_ts) const;
 
 
+    /**
+     * @brief Gather and send the required data to the pserver
+     */
+    void gatherAndSendPSdata(const Anomalies &anomalies,
+			     const int step) const;
+
     /**
      * @brief Send new metadata entries collected during current fram to provenance DB
      */

include/chimbuko/message.hpp

@@ -31,7 +31,8 @@ enum MessageKind {
     PARAMETERS    = 2,
     ANOMALY_STATS = 3,
     COUNTER_STATS = 4,
-    FUNCTION_INDEX = 5
+    FUNCTION_INDEX = 5,
+    AD_PS_COMBINED_STATS = 6
 };
 
 enum MessageCmd {
@@ -204,6 +205,7 @@ class Message {
 	case 3: return "ANOMALY_STATS";
 	case 4: return "COUNTER_STATS";
 	case 5: return "FUNCTION_INDEX";
+	case 6: return "AD_PS_COMBINED_STATS";
 	default: return "UNKNOWN";
         }
     }

include/chimbuko/pserver.hpp

@@ -3,3 +3,4 @@
 #include "pserver/GlobalAnomalyStats.hpp"
 #include "pserver/GlobalCounterStats.hpp"
 #include "pserver/PSglobalFunctionIndexMap.hpp"
+#include "pserver/NetPayloadRecvCombinedADdata.hpp"

include/chimbuko/pserver/AggregateAnomalyData.hpp

@@ -69,6 +69,16 @@ namespace chimbuko {
      */
     size_t get_n_data() const;
 
+    /**
+     * @brief Comparison operator
+     */
+    bool operator==(const AggregateAnomalyData &r) const;
+    /**
+     * @brief Inequality operator
+     */
+    bool operator!=(const AggregateAnomalyData &r) const{ return !( *this == r ); }
+
+
   private:
     mutable std::mutex               m_mutex;
     RunStats                 m_stats; /** Statistics on the number of anomalies collected per io step since start of run as well as count of total anomalies*/

include/chimbuko/pserver/AggregateFuncStats.hpp

@@ -53,6 +53,19 @@ namespace chimbuko {
      */
     const RunStats & get_exclusive() const{ return m_exclusive; }
 
+    /**
+     * @brief Comparison operator
+     */
+    bool operator==(const AggregateFuncStats &r) const{ return m_pid == r.m_pid && m_fid == r.m_fid && m_func == r.m_func &&
+	                                                       m_func_anomaly == r.m_func_anomaly && m_inclusive == r.m_inclusive && 
+	                                                       m_exclusive == r.m_exclusive; 
+    }
+    /**
+     * @brief Inequality operator
+     */
+    bool operator!=(const AggregateFuncStats &r) const{ return !( *this == r ); }
+
+
   private:
     int m_pid; /**< Program idx */
     int m_fid; /**< Function idx */

include/chimbuko/pserver/GlobalAnomalyStats.hpp

@@ -100,6 +100,16 @@ namespace chimbuko{
      * @return JSON object containing anomaly and function data
      */
     nlohmann::json collect();
+    
+    /**
+     * @brief Comparison operator
+     */
+    bool operator==(const GlobalAnomalyStats &r) const{ return m_anomaly_stats == r.m_anomaly_stats && m_funcstats == r.m_funcstats; }
+
+    /**
+     * @brief Inequality operator
+     */
+    bool operator!=(const GlobalAnomalyStats &r) const{ return !( *this == r ); }
 
   protected:    
     std::unordered_map<int, std::unordered_map<unsigned long, AggregateAnomalyData> > m_anomaly_stats; /**< Map of program index and rank to the statistics of the number of anomalies per step and the AnomalyData objects that have been added by that AD instance since the last flush */