added functionality to track communication volume

lib/CommStats.cpp

+
+/*
+ * CommStats.cpp
+ *
+ *  Created on: 4 Feb 2020
+ *      Author: Philipp Samfass
+ */
+
+#include "CommStats.h"
+#include "Logging.h"
+#include "Rank.h"
+#include "RankControl.h"
+
+#include <fstream>
+#include <map>
+#include <set>
+#include <sstream>
+#include <string>
+#include <utility>
+#include <stddef.h>
+#include <bitset>
+#include <unistd.h>
+#include <list>
+#include <vector>
+#include <atomic>
+
+struct CommunicationStats {
+  std::atomic<size_t> sentBytes; 
+  std::atomic<size_t> receivedBytes;
+} stats; 
+
+size_t CommunicationStatistics::computeCommunicationVolume(MPI_Datatype datatype, int count) {
+  int size;
+  MPI_Type_size(datatype, &size);
+  return static_cast<size_t>(count)*size;
+}
+
+void CommunicationStatistics::trackSend(MPI_Datatype datatype, int count) {
+  stats.sentBytes += computeCommunicationVolume(datatype, count); 
+}
+
+void CommunicationStatistics::trackReceive(MPI_Datatype datatype, int count) {
+  stats.receivedBytes += computeCommunicationVolume(datatype, count);
+}
+
+void CommunicationStatistics::outputCommunicationStatistics() {  
+  std::cout.flush(); \
+  std::cout << "[TMPI]    [rank " << getTeamRank() << "/" << getWorldRank() << "]  bytes sent: "<< stats.sentBytes.load()
+                                                                               <<" bytes received: " << stats.receivedBytes.load()<< std::endl; \
+ 
+  std::cout.flush(); 
+
+  std::string filenamePrefix = getEnvString("TMPI_STATS_FILE");
+  std::string outputPathPrefix = getEnvString("TMPI_STATS_OUTPUT_PATH");
+
+  if (!filenamePrefix.empty()) {
+    // Write Generic Sync points to files
+    char sep = ',';
+    std::ostringstream filename;
+    std::string outputFolder(outputPathPrefix.empty() ? "tmpi-statistics" : outputPathPrefix);
+    filename << outputFolder << "/"
+        << filenamePrefix << "-"
+        << getWorldRank() << "-"
+        << getTeamRank() << "-"
+        << getTeam()
+        << ".csv";
+    std::ofstream f;
+    f.open(filename.str().c_str());
+
+    logInfo("Writing statistics to " << filename);
+
+    f << "sent received"<<"\n";
+    f << stats.sentBytes.load() << " "<< stats.receivedBytes.load();
+
+    f.close();
+  } 
+}

lib/CommStats.h

+/*
+ * CommStats.h
+ *
+ *  Created on: 4 Feb 2020
+ *      Author: Philipp Samfass
+ */
+
+#ifndef STATISTICS_H_
+#define STATISTICS_H_
+
+#include <mpi.h>
+
+
+namespace CommunicationStatistics {
+
+ size_t computeCommunicationVolume(MPI_Datatype, int count);
+
+ void trackSend(MPI_Datatype datatype, int count);
+ void trackReceive(MPI_Datatype datatype, int count); 
+
+ void outputCommunicationStatistics();
+
+}
+
+#endif /* STATISTICS_H_ */
+

lib/Makefile

@@ -2,8 +2,8 @@ CC=mpiicpc
 CFLAGS += -fPIC -g -Wall -std=c++11
 LDFLAGS += -shared
 
-SRC = Rank.cpp RankControl.cpp Timing.cpp Wrapper.cpp teaMPI.cpp
-DEP = Rank.h RankControl.h Timing.h Wrapper.h Logging.h teaMPI.h
+SRC = Rank.cpp RankControl.cpp Timing.cpp Wrapper.cpp teaMPI.cpp CommStats.cpp
+DEP = Rank.h RankControl.h Timing.h Wrapper.h Logging.h teaMPI.h CommStats.h
 OBJECTS = $(SRC:.cpp=.o)
 
 TARGET = libtmpi.so

lib/Wrapper.cpp

@@ -6,6 +6,7 @@
 #include "Logging.h"
 #include "Rank.h"
 #include "Timing.h"
+#include "CommStats.h"
 
 int MPI_Init(int *argc, char*** argv) {
   int err = PMPI_Init(argc, argv);
@@ -58,6 +59,9 @@ int MPI_Comm_free(MPI_Comm *comm) {
 
 int MPI_Send(const void *buf, int count, MPI_Datatype datatype, int dest,
               int tag, MPI_Comm comm) {
+#if COMM_STATS
+  CommunicationStatistics::trackSend(datatype,  count);
+#endif
   //assert(comm == MPI_COMM_WORLD);
   int err = PMPI_Send(buf, count, datatype, dest, tag, getTeamComm(comm));
   logInfo("Send to rank " << dest << "/" << mapTeamToWorldRank(dest) << " with tag " << tag);
@@ -67,6 +71,9 @@ int MPI_Send(const void *buf, int count, MPI_Datatype datatype, int dest,
 int MPI_Recv(void *buf, int count, MPI_Datatype datatype, int source, int tag,
               MPI_Comm comm, MPI_Status *status) {
   //assert(comm == MPI_COMM_WORLD);
+#if COMM_STATS
+  CommunicationStatistics::trackReceive(datatype,  count);
+#endif
   int err = PMPI_Recv(buf, count, datatype, source, tag, getTeamComm(comm), status);
   logInfo("Receive from rank " << source << "/" << mapTeamToWorldRank(source) << " with tag " << tag);
   return err;
@@ -93,6 +100,9 @@ int MPI_Iallgather(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
 int MPI_Isend(const void *buf, int count, MPI_Datatype datatype, int dest,
                int tag, MPI_Comm comm, MPI_Request *request) {
   //assert(comm == MPI_COMM_WORLD);
+#if COMM_STATS
+  CommunicationStatistics::trackSend(datatype,  count);
+#endif
   int err = PMPI_Isend(buf, count, datatype, dest, tag, getTeamComm(comm), request);
   logInfo("Isend to rank " << dest << "/" << mapTeamToWorldRank(dest) << " with tag " << tag);
   return err;
@@ -101,6 +111,9 @@ int MPI_Isend(const void *buf, int count, MPI_Datatype datatype, int dest,
 int MPI_Irecv(void *buf, int count, MPI_Datatype datatype, int source, int tag,
                MPI_Comm comm, MPI_Request *request) {
   //assert(comm == MPI_COMM_WORLD);
+#if COMM_STATS
+  CommunicationStatistics::trackReceive(datatype,  count);
+#endif
   int err = PMPI_Irecv(buf, count, datatype, source, tag, getTeamComm(comm), request);
   logInfo("Receive from rank " << source << "/" << mapTeamToWorldRank(source) << " with tag " << tag);
   return err;
@@ -209,6 +222,9 @@ int MPI_Finalize() {
   PMPI_Barrier(MPI_COMM_WORLD);
   freeTeamComm();
   Timing::outputTiming();
+#if COMM_STATS
+  CommunicationStatistics::outputCommunicationStatistics();
+#endif
 #ifdef DirtyCleanUp
   return MPI_SUCCESS;
 #endif