Minor improvements:

	Head node logging now reports number of events of each type encountered in the step
	Periodic perf output now records accumulated number of events of each type, number of outliers detected and number of IO steps since last write
	ZMQNet send/recv now throw errors if a zmq_error was detected, and the error message is reported
	Perf filenames now include program index rather than just rank
	PerfPeriodic now uses map rather than unordered_map so that output columns order doesn't change due to rehashing
	Driver now returns exit code of 1 if an error was caught
	ZMQNet uses the error functionality rather than just throw
	Anomaly count written to perf output

Author: giltirn

app/driver.cpp

@@ -101,7 +101,7 @@ void printHelp(){
   std::cout << "Usage: driver <Trace engine type> <Trace directory> <Trace file prefix> <Options>\n" 
 	    << "Where <Trace engine type> : BPFile or SST\n"
 	    << "      <Trace directory>   : The directory in which the BPFile or SST file is located\n"
-	    << "      <Trace file prefix> : The prefix of the file eg \"tau-metrics-[mybinary]\"\n"
+	    << "      <Trace file prefix> : The prefix of the file (the trace file name without extension e.g. \"tau-metrics-mybinary\" for \"tau-metrics-mybinary.bp\")\n"
 	    << "      <Options>           : Optional arguments as described below.\n";
   getOptionalArgsParser().help(std::cout);
 }
@@ -201,6 +201,8 @@ int main(int argc, char ** argv){
   verboseStream << "Driver rank " << params.rank << ": waiting at pre-run barrier" << std::endl;
   MPI_Barrier(MPI_COMM_WORLD);
 
+  bool error = false;
+
   try 
     {
       //Instantiate Chimbuko
@@ -276,23 +278,20 @@ int main(int argc, char ** argv){
 				  << "Total function/comm events         : " << total_n_func_events + total_n_comm_events
 				  << std::endl;
     }
-  catch (std::invalid_argument &e)
-    {
-      std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : caught invalid argument:" << std::endl;
-      std::cout << e.what() << std::endl;
-    }
-  catch (std::ios_base::failure &e)
-    {
-      std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : I/O base exception caught\n";
-      std::cout << e.what() << std::endl;
-    }
-  catch (std::exception &e)
-    {
-      std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : Exception caught\n";
-      std::cout << e.what() << std::endl;
-    }
+  catch (const std::invalid_argument &e){
+    std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : caught invalid argument: " << e.what() << std::endl;
+    error = true;
+  }
+  catch (const std::ios_base::failure &e){
+    std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : I/O base exception caught: " << e.what() << std::endl;
+    error = true;
+  }
+  catch (const std::exception &e){
+    std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : Exception caught: " << e.what() << std::endl;
+    error = true;
+  }
 
   MPI_Finalize();
   headProgressStream(params.rank) << "Driver is exiting" << std::endl;
-  return 0;
+  return error ? 1 : 0;
 }

include/chimbuko/util/PerfStats.hpp

@@ -3,7 +3,7 @@
 #ifdef _PERF_METRIC
 #include <chrono>
 #include <sstream>
-#include <unordered_map>
+#include <map>
 #include "chimbuko/util/RunMetric.hpp"
 #endif
 
@@ -103,7 +103,7 @@ namespace chimbuko{
 #ifdef _PERF_METRIC
     std::string m_outputpath;
     std::string m_filename;
-    std::unordered_map<std::string, std::string> m_data;
+    std::map<std::string, std::string> m_data;
     bool m_first_write; /**< Is this the first write?*/
 #endif
   public:    

src/ad/ADNetClient.cpp

@@ -84,10 +84,10 @@ void ADNetClient::connect_ps(int rank, int srank, std::string sname) {
     msg.clear();
 
     verboseStream << "ADNetClient waiting for handshake response from server" << std::endl;
-    int ret = ZMQNet::recv(m_socket, strmsg);
-    if(ret == -1){
-      if(errno == EAGAIN){ recoverable_error("Connect error to parameter server: handshake timed out"); }
-      else{ recoverable_error("Connect error to parameter server: receive returned an error"); }
+    try{
+      ZMQNet::recv(m_socket, strmsg);
+    }catch(const std::exception &err){
+      recoverable_error("Connect error to parameter server, ADNetClient failed to initialize");
       return;
     }
 
@@ -166,8 +166,7 @@ std::string ADNetClient::send_and_receive(const Message &msg){
   ZMQNet::send(m_socket, send_msg);
 
   //Receive global parameters from PS
-  int ret = ZMQNet::recv(m_socket, recv_msg);   
-  if(ret == -1) throw std::runtime_error("ADNetClient::send_and_receive: receive call returned an error");
+  ZMQNet::recv(m_socket, recv_msg);   
 #endif
 
 #ifdef _PERF_METRIC

src/chimbuko.cpp

@@ -320,9 +320,9 @@ bool Chimbuko::parseInputStep(int &step,
   m_parser->endStep();
 
   // count total number of events
-  n_func_events += (unsigned long long)m_parser->getNumFuncData();
-  n_comm_events += (unsigned long long)m_parser->getNumCommData();
-  n_counter_events += (unsigned long long)m_parser->getNumCounterData();
+  n_func_events = (unsigned long long)m_parser->getNumFuncData();
+  n_comm_events = (unsigned long long)m_parser->getNumCommData();
+  n_counter_events = (unsigned long long)m_parser->getNumCounterData();
 
   //Parse the new metadata for any attributes we want to maintain
   m_metadata_parser->addData(m_parser->getNewMetaData());
@@ -458,10 +458,10 @@ void Chimbuko::run(unsigned long long& n_func_events,
   int step = m_parser->getCurrentStep(); //gives -1 as initial value. step+1 is the expected value of step in parseInputStep and is used as a (non-fatal) check
   unsigned long first_event_ts, last_event_ts; //earliest and latest timestamps in io frame
 
-  std::string ad_perf = "ad_perf_" + std::to_string(m_params.rank) + ".json";
+  std::string ad_perf = stringize("ad_perf.%d.%d.json", m_params.program_idx, m_params.rank);
   m_perf.setWriteLocation(m_params.perf_outputpath, ad_perf);
 
-  std::string ad_perf_prd = "ad_perf_prd_" + std::to_string(m_params.rank) + ".log";
+  std::string ad_perf_prd = stringize("ad_perf_prd.%d.%d.log", m_params.program_idx, m_params.rank);
   m_perf_prd.setWriteLocation(m_params.perf_outputpath, ad_perf_prd);
 
 #if defined(_PERF_METRIC) && defined(ENABLE_PROVDB)
@@ -471,13 +471,28 @@ void Chimbuko::run(unsigned long long& n_func_events,
 #endif
 
   PerfTimer step_timer, timer;
+  
+  unsigned long long n_func_events_step, n_comm_events_step, n_counter_events_step; //event count in present step
+  unsigned long long n_func_events_accum_prd = 0, n_comm_events_accum_prd = 0, n_counter_events_accum_prd = 0; //accumulated event counts since last write of periodic data
+  unsigned long n_outliers_accum_prd = 0; //accumulated outliers detected since last write of periodic data 
+  int n_steps_accum_prd = 0; //number of steps since last write of periodic data
 
   //Loop until we lose connection with the application
-  while ( parseInputStep(step, n_func_events, n_comm_events, n_counter_events) ) {
+  while ( parseInputStep(step, n_func_events_step, n_comm_events_step, n_counter_events_step) ) {
+    //Increment total events
+    n_func_events += n_func_events_step;
+    n_comm_events += n_comm_events_step;
+    n_counter_events += n_counter_events_step;
+
+    n_func_events_accum_prd += n_func_events_step;
+    n_comm_events_accum_prd += n_comm_events_step;
+    n_counter_events_accum_prd += n_counter_events_step;
+    
     //Decide whether to report step progress
     bool do_step_report = enableVerboseLogging() || (m_params.step_report_freq > 0 && step % m_params.step_report_freq == 0);
 
-    if(do_step_report){ headProgressStream(m_params.rank) << "driver rank " << m_params.rank << " starting analysis of step " << step << std::endl; }
+    if(do_step_report){ headProgressStream(m_params.rank) << "driver rank " << m_params.rank << " starting analysis of step " << step 
+							  << ". Event count: func=" << n_func_events_step << " comm=" << n_comm_events_step << " counter=" << n_counter_events_step << std::endl; }
     step_timer.start();
 
     //Extract counters and put into counter manager
@@ -494,9 +509,13 @@ void Chimbuko::run(unsigned long long& n_func_events,
     timer.start();
     Anomalies anomalies = m_outlier->run(step);
     m_perf.add("ad_run_anom_detection_time_ms", timer.elapsed_ms());
+    m_perf.add("ad_run_anomaly_count", anomalies.nEvents(Anomalies::EventType::Outlier));
 
-    n_outliers += anomalies.nEvents(Anomalies::EventType::Outlier);
+    int nout = anomalies.nEvents(Anomalies::EventType::Outlier);
+    n_outliers += nout;
+    n_outliers_accum_prd += nout;
     frames++;
+    n_steps_accum_prd++;
 
     //Generate anomaly provenance for detected anomalies and send to DB
     timer.start();
@@ -542,6 +561,23 @@ void Chimbuko::run(unsigned long long& n_func_events,
       size_t total, resident;
       getMemUsage(total, resident);
       m_perf_prd.add("ad_mem_usage_kB", resident);
+      
+      m_perf_prd.add("io_steps", n_steps_accum_prd);
+
+      //Write accumulated outlier count
+      m_perf_prd.add("outlier_count", n_outliers_accum_prd);
+
+      //Write accumulated event counts
+      m_perf_prd.add("event_count_func", n_func_events_accum_prd);
+      m_perf_prd.add("event_count_comm", n_comm_events_accum_prd);
+      m_perf_prd.add("event_count_counter", n_counter_events_accum_prd);
+
+      //Reset the counts
+      n_func_events_accum_prd = 0;
+      n_comm_events_accum_prd = 0;
+      n_counter_events_accum_prd = 0;
+      n_outliers_accum_prd = 0;
+      n_steps_accum_prd = 0;
 
       //These only write if both filename and output path is set
       m_perf_prd.write();

src/net/zmq_net.cpp

@@ -56,7 +56,7 @@ class NetPayloadClientDisconnectWithCount: public NetPayloadBase{
     response.set_msg("", false);
     std::lock_guard<std::mutex> _(m);
     clients--;
-    if(clients < 0) throw std::runtime_error("ZMQNet registered clients < 0! Likely a client did not perform a handshake");
+    if(clients < 0) fatal_error("ZMQNet registered clients < 0! Likely a client did not perform a handshake");
 
     verboseStream << "ZMQNet received disconnect notification, #clients is now " << clients << std::endl;
   };
@@ -77,11 +77,11 @@ void ZMQNet::init(int* /*argc*/, char*** /*argv*/, int nt)
 {
     m_context  = zmq_ctx_new();
     if(zmq_ctx_set(m_context, ZMQ_IO_THREADS, m_io_threads) != 0)
-      throw std::runtime_error("ZMQNet::init couldn't set number of io threads to requested amount");   
+      fatal_error("ZMQNet::init couldn't set number of io threads to requested amount");   
 
     //Check worker_idx = 0 for all payloads
     for(auto const &e : m_payloads)
-      if(e.first != 0) throw std::runtime_error("ZMQNet all payloads must have worker_idx=0");
+      if(e.first != 0) fatal_error("ZMQNet all payloads must have worker_idx=0");
 
     init_thread_pool(nt);
 }
@@ -134,9 +134,9 @@ void doWork(void* context,
 
     timer.start();
     auto kit = payloads.find((MessageKind)msg.kind());
-    if(kit == payloads.end()) throw std::runtime_error("ZMQNet::doWork : No payload associated with the message kind provided. Message: " + strmsg + " (did you add the payload to the server?)");
+    if(kit == payloads.end()) fatal_error("ZMQNet::doWork : No payload associated with the message kind provided. Message: " + strmsg + " (did you add the payload to the server?)");
     auto pit = kit->second.find((MessageType)msg.type());
-    if(pit == kit->second.end()) throw std::runtime_error("ZMQNet::doWork : No payload associated with the message type provided. Mesage: " + strmsg + " (did you add the payload to the server?)");
+    if(pit == kit->second.end()) fatal_error("ZMQNet::doWork : No payload associated with the message type provided. Mesage: " + strmsg + " (did you add the payload to the server?)");
     perf.add(perf_prefix + "message_kind_type_lookup_ms", timer.elapsed_ms());
 
     //Apply the payload
@@ -394,7 +394,7 @@ int ZMQNet::recvAndSend(void* skFrom, void* skTo, int max_msg){
     if(len == -1 && errno == EAGAIN){ //no more messages on queue
       zmq_msg_close(&msg);
       return msg_count;
-    }else if(len == -1) throw std::runtime_error("ZMQNet::recvAndSend error receiving: " + std::string(zmq_strerror(errno)));
+    }else if(len == -1) fatal_error("ZMQNet::recvAndSend error receiving: " + std::string(zmq_strerror(errno)));
     int more;
     zmq_getsockopt(skFrom, ZMQ_RCVMORE, &more, &more_size); //check whether message is part of a multi-part message
 
@@ -429,6 +429,9 @@ int ZMQNet::send(void* socket, const std::string& strmsg)
     memcpy(zmq_msg_data(&msg), (const void*)strmsg.data(), strmsg.size());
     ret = zmq_msg_send(&msg, socket, 0);
     zmq_msg_close(&msg);
+
+    if(ret == -1) fatal_error("Send failed with error: " + std::string(zmq_strerror(errno)) );
+
     return ret;
 }
 
@@ -440,6 +443,9 @@ int ZMQNet::recv(void* socket, std::string& strmsg)
     ret = zmq_msg_recv(&msg, socket, 0);    
     if(ret != -1) strmsg.assign((char*)zmq_msg_data(&msg), ret);
     zmq_msg_close(&msg);
+
+    if(ret == -1) fatal_error("Receive failed with error: " + std::string(zmq_strerror(errno)) );
+
     return ret;
 }
 #endif