Actually build the LDMS service (#538)

Author: daboehme

caliper-config.h.in

@@ -24,6 +24,7 @@
 #cmakedefine CALIPER_HAVE_ROCTRACER
 #cmakedefine CALIPER_HAVE_UMPIRE
 #cmakedefine CALIPER_HAVE_CRAYPAT
+#cmakedefine CALIPER_HAVE_LDMS
 
 #cmakedefine CALIPER_REDUCED_CONSTEXPR_USAGE
 

src/services/CMakeLists.txt

@@ -115,6 +115,9 @@ endif()
 if (CALIPER_HAVE_CRAYPAT)
   add_subdirectory(craypat)
 endif()
+if (CALIPER_HAVE_LDMS)
+  add_subdirectory(ldms)
+endif()
 if (CALIPER_HAVE_MPI)
   include_directories(${MPI_CXX_INCLUDE_PATH})
   add_subdirectory(mpireport)

src/services/ldms/LdmsForwarder.cpp

@@ -94,7 +94,7 @@ ldms_t setup_connection(const char *xprt, const char *host,
                 ts.tv_nsec = 0;
         }
 
-        ldms_g = ldms_xprt_new_with_auth(xprt, auth, NULL);
+        ldms_g = ldms_xprt_new_with_auth(xprt, NULL, auth, NULL);
         if (!ldms_g) {
                 printf("Error %d creating the '%s' transport\n",
                        errno, xprt);
@@ -127,8 +127,8 @@ void caliper_ldms_connector_initialize()
     /* Check/set LDMS transport type */
     if (!env_ldms_xprt || !env_ldms_host || !env_ldms_port || !env_ldms_auth){
 	Log(1).stream() << "Either the transport, host, port or authentication is not given. Setting to default.\n";
-  
-    	if (env_ldms_xprt == NULL) 
+
+    	if (env_ldms_xprt == NULL)
         	env_ldms_xprt = "sock";
 
     	if (env_ldms_host == NULL)
@@ -139,7 +139,7 @@ void caliper_ldms_connector_initialize()
 
     	if (env_ldms_auth == NULL)
 		env_ldms_auth = "munge";
-    	   
+
     }
 
     pthread_mutex_lock(&ln_lock);
@@ -158,28 +158,26 @@ void caliper_ldms_connector_initialize()
     return;
 }
 
-std::ostream& write_ldms_record(std::ostream& os, int mpi_rank, RegionProfile& profile)
+void write_ldms_record(int mpi_rank, RegionProfile& profile)
 {
-
-    caliper_ldms_connector_initialize();	
+    caliper_ldms_connector_initialize();
 
     std::map<std::string, double> region_times;
     double total_time = 0;
-    
 
     int buffer_size = 4096;
     char* buffer = (char*) malloc (sizeof(char) * buffer_size);
-    
+
     const char*   env_ldms_jobid_str            = getenv("SLURM_JOB_ID");
     const char*   env_ldms_procid               = getenv("SLURM_PROCID");
     const char*   env_ldms_slurm_nodelist       = getenv("SLURM_JOB_NODELIST");
     const char*   env_ldms_caliper_verbose_str  = getenv("CALIPER_LDMS_VERBOSE");
 
     int env_ldms_jobid = env_ldms_jobid_str == NULL ? 0 : atoi( env_ldms_jobid_str );
-    int env_ldms_caliper_verbose = env_ldms_caliper_verbose_str == NULL ? 0 : atoi( env_ldms_caliper_verbose_str );    
+    int env_ldms_caliper_verbose = env_ldms_caliper_verbose_str == NULL ? 0 : atoi( env_ldms_caliper_verbose_str );
 
     std::tie(region_times, std::ignore, total_time) =
-        profile.inclusive_path_profile();
+        profile.inclusive_region_times();
 
     double unix_ts = 1e-6 * std::chrono::duration_cast<std::chrono::microseconds>(
         std::chrono::system_clock::now().time_since_epoch()).count();
@@ -189,78 +187,63 @@ std::ostream& write_ldms_record(std::ostream& os, int mpi_rank, RegionProfile& p
         if (p.second < 0.05 * total_time)
             continue;
 
-        os << "{ \"timestamp\": " << std::fixed << unix_ts
-           << ", \"duration\": "  << std::fixed << p.second;
-
-        util::write_esc_string(os << ", \"path\": \"", p.first) << "\"";
-
-	std::string path_msg = "" + p.first;
-        std::string path = path_msg.c_str();
+    	// std::string path_msg = "" + p.first;
+        const char* path = p.first.c_str();
 
         if (mpi_rank >= 0) {
-            os << ", \"rank\": " << mpi_rank;
-	    snprintf(buffer, buffer_size, "{ \"timestamp\": %f, \"jobid\" : %d, \"rank\" : %d, \"procid\" : %d, \"nodelist\" : %s, \"caliper-perf-data\", \"duration\": %f, \"path\": \"%s\"} \n", unix_ts, env_ldms_jobid, mpi_rank, env_ldms_procid, env_ldms_slurm_nodelist, p.second, path); 
-	} else {
-	    snprintf(buffer, buffer_size, "{ \"timestamp\": %f, \"jobid\" : %d, \"rank\": %d, \"procid\" : %d, \"nodelist\" : %s, \"caliper-perf-data\", \"duration\": %f, \"path\": \"%s\"} \n", unix_ts, env_ldms_jobid, 0, env_ldms_procid, env_ldms_slurm_nodelist, p.second, path);	
-	}
-	os << "} \n";
-	
-	if (env_ldms_caliper_verbose > 0)
-		puts(buffer);
+            snprintf(buffer, buffer_size, "{ \"timestamp\": %f, \"jobid\" : %d, \"rank\" : %d, \"procid\" : %s, \"nodelist\" : %s, \"caliper-perf-data\", \"duration\": %f, \"path\": \"%s\"} \n", unix_ts, env_ldms_jobid, mpi_rank, env_ldms_procid, env_ldms_slurm_nodelist, p.second, path);
+	    } else {
+	        snprintf(buffer, buffer_size, "{ \"timestamp\": %f, \"jobid\" : %d, \"rank\": %d, \"procid\" : %s, \"nodelist\" : %s, \"caliper-perf-data\", \"duration\": %f, \"path\": \"%s\"} \n", unix_ts, env_ldms_jobid, 0, env_ldms_procid, env_ldms_slurm_nodelist, p.second, path);
+	    }
 
-	int rc = ldmsd_stream_publish( ldms_cali, "caliper-perf-data", LDMSD_STREAM_JSON, buffer, strlen(buffer) + 1);
-	if (rc)
-		 Log(0).stream() << "Error " << rc << " publishing data.\n";
+	    if (env_ldms_caliper_verbose > 0)
+		    puts(buffer);
 
-	else if (env_ldms_caliper_verbose > 0)
-		 Log(2).stream() << "Caliper Message published successfully\n";
-	}
+    	int rc = ldmsd_stream_publish( ldms_cali, "caliper-perf-data", LDMSD_STREAM_JSON, buffer, strlen(buffer) + 1);
 
-	return os;
+	    if (rc)
+		     Log(0).stream() << "Error " << rc << " publishing data.\n";
+    	else if (env_ldms_caliper_verbose > 0)
+	    	 Log(2).stream() << "Caliper Message published successfully\n";
+	}
 }
 
 class LdmsForwarder
 {
     RegionProfile profile;
-    OutputStream  stream;
-
-    std::string   filename;
 
     void snapshot(Caliper* c, Channel*) {
         Entry e = c->get(c->get_attribute("mpi.rank"));
         int rank = e.empty() ? -1 : e.value().to_int();
 
-        write_ldms_record(*stream.stream(), rank, profile) << "\n";
+        write_ldms_record(rank, profile);
 
         profile.clear(); // reset profile - skip to create a cumulative profile
     }
 
     void post_init(Caliper* c, Channel* channel) {
-        std::vector<Entry> rec;
-        stream.set_filename(filename.c_str(), *c, rec);
         profile.start();
     }
 
-    LdmsForwarder(const char* fname)
-        : filename { fname }
+    LdmsForwarder()
     { }
 
 public:
-    
+
     static const char* s_spec;
 
     static void create(Caliper* c, Channel* channel) {
-        ConfigSet cfg = 
+        ConfigSet cfg =
             services::init_config_from_spec(channel->config(), s_spec);
 
-        LdmsForwarder* instance = new LdmsForwarder(cfg.get("filename").to_string().c_str());
+        LdmsForwarder* instance = new LdmsForwarder();
 
         channel->events().post_init_evt.connect(
             [instance](Caliper* c, Channel* channel){
                 instance->post_init(c, channel);
             });
         channel->events().snapshot.connect(
-            [instance](Caliper* c, Channel* channel, int scope, SnapshotView, SnapshotBuilder&){
+            [instance](Caliper* c, Channel* channel, SnapshotView, SnapshotBuilder&){
                 instance->snapshot(c, channel);
             });
         channel->events().finish_evt.connect(
@@ -271,21 +254,15 @@ class LdmsForwarder
         Log(1).stream() << channel->name() << "Initialized LDMS forwarder\n";
     }
 
-	
+
 };
 
 const char* LdmsForwarder::s_spec = R"json(
-{   
+{
  "name"        : "ldms",
  "description" : "Forward Caliper regions to LDMS (prototype)",
  "config"      :
  [
-  {
-    "name"        : "filename",
-    "description" : "Output file name, or stdout/stderr",
-    "type"        : "string",
-    "value"       : "stdout"
-  }
  ]
 }
 )json";