Unified sync_param method for HBOS and sstd, implementation now in base class

Fixed error HbosParam::update where internal params were not being correctly updated

Author: giltirn

include/chimbuko/ad/ADOutlier.hpp

@@ -92,12 +92,12 @@ namespace chimbuko {
 					   const unsigned long func_id, std::vector<CallListIterator_t>& data) = 0;
 
     /**
-     * @brief abstract method to update local parameters and get global ones
+     * @brief synchronize local parameters with global parameters
      *
      * @param[in] param local parameters
      * @return std::pair<size_t, size_t> [sent, recv] message size
      */
-    virtual std::pair<size_t, size_t> sync_param(ParamInterface const* param) = 0;
+    virtual std::pair<size_t, size_t> sync_param(ParamInterface const* param);
 
 
     /**
@@ -170,13 +170,6 @@ namespace chimbuko {
     unsigned long compute_outliers(Anomalies &outliers,
 				   const unsigned long func_id, std::vector<CallListIterator_t>& data) override;
 
-
-    /**
-     * @brief Send the local statistics to the parameter server and update the stored global parameters with the resulting updated statistics
-     * @param param The local statistics
-     */
-    std::pair<size_t, size_t> sync_param(ParamInterface const* param) override;
-
     /**
      * @brief Compute the anomaly score (probability) for an event assuming a Gaussian distribution
      */
@@ -232,13 +225,6 @@ namespace chimbuko {
     unsigned long compute_outliers(Anomalies &outliers,
 				   const unsigned long func_id, std::vector<CallListIterator_t>& data) override;
 
-
-    /**
-     * @brief Send the local statistics to the parameter server and update the stored global parameters with the resulting updated statistics
-     * @param param The local statistics
-     */
-    std::pair<size_t, size_t> sync_param(ParamInterface const* param) override;
-
     /**
      * @brief Scott's rule for bin_width estimation during histogram formation
      */

src/ad/ADOutlier.cpp

@@ -54,6 +54,29 @@ double ADOutlier::getStatisticValue(const ExecData_t &e) const{
   }
 }
 
+std::pair<size_t,size_t> ADOutlier::sync_param(ParamInterface const* param)
+{
+  if (!m_use_ps) {
+    verboseStream << "m_use_ps not USED!" << std::endl;
+    m_param->update(param->serialize());
+    return std::make_pair(0, 0);
+  }
+  else {
+    Message msg;
+    msg.set_info(m_net_client->get_client_rank(), m_net_client->get_server_rank(), MessageType::REQ_ADD, MessageKind::PARAMETERS);
+    msg.set_msg(param->serialize(), false);
+    size_t sent_sz = msg.size();
+   
+    m_net_client->send_and_receive(msg, msg);
+    size_t recv_sz = msg.size();
+    m_param->assign(msg.buf());
+    return std::make_pair(sent_sz, recv_sz);
+  }
+}
+
+
+
+
 /* ---------------------------------------------------------------------------
  * Implementation of ADOutlierSSTD class
  * --------------------------------------------------------------------------- */
@@ -64,28 +87,6 @@ ADOutlierSSTD::ADOutlierSSTD(OutlierStatistic stat, double sigma) : ADOutlier(st
 ADOutlierSSTD::~ADOutlierSSTD() {
 }
 
-std::pair<size_t,size_t> ADOutlierSSTD::sync_param(ParamInterface const* param)
-{
-  SstdParam& g = *(SstdParam*)m_param; //global parameter set
-  const SstdParam & l = *(SstdParam const*)param; //local parameter set
-
-    if (!m_use_ps) {
-        g.update(l);
-        return std::make_pair(0, 0);
-    }
-    else {
-        Message msg;
-        msg.set_info(m_net_client->get_client_rank(), m_net_client->get_server_rank(), MessageType::REQ_ADD, MessageKind::PARAMETERS);
-        msg.set_msg(l.serialize(), false);
-        size_t sent_sz = msg.size();
-
-	m_net_client->send_and_receive(msg, msg);
-        size_t recv_sz = msg.size();
-        g.assign(msg.buf());
-        return std::make_pair(sent_sz, recv_sz);
-    }
-}
-
 Anomalies ADOutlierSSTD::run(int step) {
   Anomalies outliers;
   if (m_execDataMap == nullptr) return outliers;
@@ -217,29 +218,6 @@ ADOutlierHBOS::~ADOutlierHBOS() {
     m_param->clear();
 }
 
-std::pair<size_t,size_t> ADOutlierHBOS::sync_param(ParamInterface const* param)
-{
-  HbosParam& g = *(HbosParam*)m_param; //global parameter set
-  const HbosParam & l = *(HbosParam const*)param; //local parameter set
-
-    if (!m_use_ps) {
-        verboseStream << "m_use_ps not USED!" << std::endl;
-        g.update(l);
-        return std::make_pair(0, 0);
-    }
-    else {
-        Message msg;
-        msg.set_info(m_net_client->get_client_rank(), m_net_client->get_server_rank(), MessageType::REQ_ADD, MessageKind::PARAMETERS);
-        msg.set_msg(l.serialize(), false);
-        size_t sent_sz = msg.size();
-   
-        m_net_client->send_and_receive(msg, msg);
-        size_t recv_sz = msg.size();
-        g.assign(msg.buf());
-        return std::make_pair(sent_sz, recv_sz);
-    }
-}
-
 Anomalies ADOutlierHBOS::run(int step) {
   Anomalies outliers;
   if (m_execDataMap == nullptr) return outliers;

src/param/hbos_param.cpp

@@ -94,11 +94,14 @@ using namespace chimbuko;
  std::string HbosParam::update(const std::string& parameters, bool return_update)
  {
      std::unordered_map<unsigned long, Histogram> hbosstats;
-
      deserialize_cerealpb(parameters, hbosstats);
-     update_and_return(hbosstats); //update hbosstats to reflect changes
-
-     return (return_update) ? serialize_cerealpb(hbosstats): "";
+     if(return_update){
+       update_and_return(hbosstats); //update hbosstats to reflect changes
+       return serialize_cerealpb(hbosstats);
+     }else{
+       update(hbosstats);
+       return "";
+     }
  }
 
  void HbosParam::update(const std::unordered_map<unsigned long, Histogram>& hbosstats)
@@ -113,7 +116,7 @@ using namespace chimbuko;
  {
     std::lock_guard<std::mutex> _(m_mutex);
      for (auto& pair: hbosstats) {
-         hbosstats[pair.first] += pair.second;
+         m_hbosstats[pair.first] += pair.second;
          pair.second = hbosstats[pair.first];
      }
 

src/param/sstd_param.cpp

@@ -76,11 +76,14 @@ void SstdParam::deserialize_cerealpb(const std::string& parameters,  std::unorde
 std::string SstdParam::update(const std::string& parameters, bool return_update)
 {
     std::unordered_map<unsigned long, RunStats> runstats;
-
     deserialize_cerealpb(parameters, runstats);
-    update_and_return(runstats);  //update runstats to reflect changes
-
-    return (return_update) ? serialize_cerealpb(runstats): "";
+    if(return_update){
+      update_and_return(runstats);  //update runstats to reflect changes
+      return serialize_cerealpb(runstats);
+    }else{
+      update(runstats);
+      return "";
+    }
 }
 
 void SstdParam::assign(const std::unordered_map<unsigned long, RunStats>& runstats)