Merge pull request #26 from CODARcode/sm_release

Passing unit tests now and fixed detection loop for copod

Author: giltirn

configure.ac

@@ -40,12 +40,19 @@ AC_COMPILE_IFELSE( [AC_LANG_PROGRAM([[]],
 AC_ARG_ENABLE([mpi],
     AS_HELP_STRING([--disable-mpi], [Disable MPI. User will need to manually assign the rank index to the AD instances.]), [], [enable_mpi=yes])
 
-AS_IF([test "x$enable_mpi" != "xno"], [
+AS_IF(
+[test "x$enable_mpi" != "xno"], [
     #Check we can compile with MPI
     AC_MSG_NOTICE([MPI use is enabled, checking compilation with MPI is possible])
     AC_CHECK_HEADER(mpi.h, [], [AC_MSG_FAILURE([["Cannot find MPI header"]])] )
     AC_DEFINE([USE_MPI],[1],[Use MPI])
-])
+    LIBS+="-lpthread"
+],
+[test "x$enable_mpi" == "xno"], [
+    AC_MSG_NOTICE([MPI use is disabled])
+    LIBS+="-lpthread"    		       
+]
+)
 
 
 #Check for curl config and library

spack/repo/chimbuko/packages/chimbuko-performance-analysis/package.py

@@ -17,8 +17,9 @@ class ChimbukoPerformanceAnalysis(AutotoolsPackage):
     version('master', branch='master')
 
     variant('perf-metric', default=True, description='Build with performance monitoring')
-
-    depends_on('mpi')
+    variant('mpi', default=True, description='Enable building Chimbuko with MPI. If disabled the user must manually provide the rank index to the OAD.')
+    
+    depends_on('mpi', when="+mpi")
     depends_on('cereal')
     depends_on('adios2')
     depends_on('googletest')
@@ -34,12 +35,15 @@ class ChimbukoPerformanceAnalysis(AutotoolsPackage):
 
 
     def setup_environment(self, spack_env, run_env):
-        spack_env.set('CXX', self.spec['mpi'].mpicxx)
+        if '+mpi' in self.spec:
+            spack_env.set('CXX', self.spec['mpi'].mpicxx)
 
     def configure_args(self):
         args = ["--with-network=ZMQ", "--with-adios2=%s" % self.spec['adios2'].prefix ]
 
         if '+perf-metric' in self.spec:
                args.append('--with-perf-metric')
-
+        if '+mpi' not in self.spec:
+               args.append('--disable-mpi')
+               
         return args

src/ad/ADNetClient.cpp

@@ -428,8 +428,9 @@ void ADThreadNetClient::run(bool local){
 	size_t nwork = getNwork();
 	while(nwork > 0){
 	  ClientAction* work_item = getWorkItem();
+	  //Ask if shutdown is to be done *before* calling perform as blocking actions unlock the parent thread in perform which destroys the ClientAction object making the pointer invalid!
+	  shutdown = shutdown || work_item->shutdown_worker(); 
 	  work_item->perform(*client);
-	  shutdown = shutdown || work_item->shutdown_worker();
 
 	  if(work_item->do_delete()) delete work_item;
 	  nwork = getNwork();

src/ad/ADOutlier.cpp

@@ -592,11 +592,17 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
   std::vector<double> recon_p_runtimes = std::vector<double>(tot_runtimes, 0.0);
   std::vector<double> recon_n_runtimes = std::vector<double>(tot_runtimes, 0.0);
   int recon_idx = 0;
+  //verboseStream << "Unwrapping Merged Histogram. Size: " << param[func_id].counts().size() << std::endl;
   for(int i=0; i < param[func_id].counts().size(); i++){
     int count = param[func_id].counts().at(i);
+    //verboseStream << "Count: " << count << ", Value: " << param[func_id].bin_edges().at(i) << std::endl;
     for(int j=0; j<count; j++){
-      recon_p_runtimes.at(recon_idx++) = param[func_id].bin_edges().at(i);
-      recon_n_runtimes.at(recon_idx++) = -1 * param[func_id].bin_edges().at(i);
+
+      recon_p_runtimes.at(recon_idx) = param[func_id].bin_edges().at(i);
+      recon_n_runtimes.at(recon_idx) = -1 * param[func_id].bin_edges().at(i);
+      //verboseStream << "recon_p_runtimes.at(recon_idx): " << recon_p_runtimes.at(recon_idx) << ", recon_n_runtimes.at(recon_idx): " << recon_n_runtimes.at(recon_idx) << std::endl;
+      //verboseStream << "recon_idx: " << recon_idx << std::endl;
+      recon_idx++;
     }
   }
 
@@ -664,8 +670,9 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
 
       const double runtime_i = this->getStatisticValue(*itt); //runtimes.push_back(this->getStatisticValue(*itt));
       double ad_score;
-      
-      if (mean_pn_ecdf.at(running_idx++) > 0)
+
+      //verboseStream << "mean_pn_ecdf.at(running_idx++): " << mean_pn_ecdf.at(running_idx) << std::endl;
+      if (mean_pn_ecdf.at(running_idx++) < 0.99)
 	      ad_score = l_threshold + 1;
       else
 	      ad_score = l_threshold - 1;
@@ -686,8 +693,8 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
         //Capture maximum of one normal execution per io step
         itt->set_label(1);
         if(outliers.nFuncEvents(func_id, Anomalies::EventType::Normal) == 0) {
-	  verboseStream << "Detected normal event on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << runtime_i << std::endl;
-	  outliers.insert(itt, Anomalies::EventType::Normal);
+      	   verboseStream << "Detected normal event on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << runtime_i << std::endl;
+      	   outliers.insert(itt, Anomalies::EventType::Normal);
         }
 
       }
@@ -720,9 +727,9 @@ int ADOutlierCOPOD::np_digitize_get_bin_inds(const double& X, const std::vector<
 }
 
 std::vector<double> ADOutlierCOPOD::empiricalCDF(const std::vector<double>& runtimes, const bool sorted) {
-  
-  std::vector<double> tmp_runtimes = runtimes;	
-  auto ecdf = boost::math::empirical_cumulative_distribution_function(std::move(tmp_runtimes));	
+
+  std::vector<double> tmp_runtimes = runtimes;
+  auto ecdf = boost::math::empirical_cumulative_distribution_function(std::move(tmp_runtimes));
   std::vector<double> result_ecdf = std::vector<double>(runtimes.size(), 0.0);
   for(int i=0; i < runtimes.size(); i++) {
     result_ecdf.at(i) = ecdf(runtimes.at(i));

src/param.cpp

@@ -1,6 +1,7 @@
 #include "chimbuko/param.hpp"
 #include "chimbuko/param/sstd_param.hpp"
 #include "chimbuko/param/hbos_param.hpp"
+#include "chimbuko/param/copod_param.hpp"
 #include <chrono>
 
 using namespace chimbuko;
@@ -17,6 +18,9 @@ ParamInterface *ParamInterface::set_AdParam(const std::string & ad_algorithm) {
   else if (ad_algorithm == "sstd" || ad_algorithm == "SSTD") {
     return new SstdParam();
   }
+  else if (ad_algorithm == "copod" || ad_algorithm == "COPOD") {
+    return new CopodParam();
+  }
   else {
     return nullptr;
   }