src/ck-perf/trace-utilization.h

   1 /**
   2  * \addtogroup CkPerf
   3  */
   4 /*@{*/
   5
   6
   7 #ifndef _TRACE_UTILIZATION_H
   8 #define _TRACE_UTILIZATION_H
   9
  10 #include <stdio.h>
  11 #include <errno.h>
  12 #include <deque>
  13
  14 #include "charm++.h"
  15
  16
  17 #include "trace.h"
  18 #include "envelope.h"
  19 #include "register.h"
  20 #include "trace-common.h"
  21 #include "ckcallback-ccs.h"
  22
  23 #include "TraceUtilization.decl.h"
  24
  25 #define INVALIDEP     -2
  26 #define TRACEON_EP    -3
  27 #define NUM_DUMMY_EPS 9
  28
  29
  30 // initial bin size, time in seconds
  31 #define  BIN_PER_SEC    1000
  32 #define  BIN_SIZE       0.001
  33 #define NUM_BINS      32768
  34
  35
  36 /** Define the types used in the gathering of sum detail statistics for use with CCS */
  37 #define numBins_T int
  38 #define numProcs_T int
  39 #define entriesInBin_T short
  40 #define ep_T short
  41 #define utilization_T unsigned char
  42 #define other_EP 10000
  43
  44
  45 /* readonly */ extern CProxy_TraceUtilizationBOC traceUtilizationGroupProxy;
  46
  47 // This function has unused arguments to match the type of
  48 // CcdVoidFn, which CcdCallOnConditionKeep takes
  49 void collectUtilizationData(void *, double);
  50
  51
  52
  53 /** A main chare that can create the BOC/group */
  54 class TraceUtilizationInit : public Chare {
  55  public:
  56   TraceUtilizationInit(CkArgMsg *m) {
  57     delete m;
  58     CkPrintf("[%d] TraceUtilizationInit creating traceUtilizationGroupProxy");
  59     fflush(stdout);
  60
  61     traceUtilizationGroupProxy = CProxy_TraceUtilizationBOC::ckNew();
  62
  63     CkPrintf("Trace Summary now listening in for CCS Client\n");
  64     CcsRegisterHandler("CkPerfSumDetail compressed", CkCallback(CkIndex_TraceUtilizationBOC::ccsRequestSumDetailCompressed(NULL), traceUtilizationGroupProxy[0]));
  65
  66     CkPrintf("[%d] Setting up periodic startCollectData callback\n", CkMyPe());
  67     CcdCallOnConditionKeep(CcdPERIODIC_1second, collectUtilizationData, (void *)NULL);
  68
  69   }
  70 };
  71
  72
  73
  74
  75
  76
  77 /**
  78     A class that reads/writes a buffer out of different types of data.
  79
  80     This class exists because I need to get references to parts of the buffer
  81     that have already been used so that I can increment counters inside the buffer.
  82 */
  83
  84 class compressedBuffer {
  85  public:
  86   char* buf;
  87   int pos; ///<< byte position just beyond the previously read/written data
  88
  89   compressedBuffer(){
  90     buf = NULL;
  91     pos = 0;
  92   }
  93
  94   compressedBuffer(int bytes){
  95     buf = (char*)malloc(bytes);
  96     pos = 0;
  97   }
  98
  99   compressedBuffer(void *buffer){
 100     buf = (char*)buffer;
 101     pos = 0;
 102   }
 103
 104   void init(void *buffer){
 105     buf = (char*)buffer;
 106     pos = 0;
 107   }
 108
 109   inline void * currentPtr(){
 110     return (void*)(buf+pos);
 111   }
 112
 113   template <typename T>
 114     T read(int offset){
 115     // to resolve unaligned writes causing bus errors, need memcpy
 116     T v;
 117     memcpy(&v, buf+offset, sizeof(T));
 118     return v;
 119   }
 120
 121   template <typename T>
 122     void write(T v, int offset){
 123     T v2 = v; // on stack
 124     // to resolve unaligned writes causing bus errors, need memcpy
 125     memcpy(buf+offset, &v2, sizeof(T));
 126   }
 127
 128   template <typename T>
 129     void increment(int offset){
 130     T temp;
 131     temp = read<T>(offset);
 132     temp ++;
 133     write<T>(temp, offset);
 134   }
 135
 136   template <typename T>
 137     void accumulate(T v, int offset){
 138     T temp;
 139     temp = read<T>(offset);
 140     temp += v;
 141     write<T>(temp, offset);
 142   }
 143
 144   template <typename T>
 145     int push(T v){
 146     int oldpos = pos;
 147     write<T>(v, pos);
 148     pos += sizeof(T);
 149     return oldpos;
 150   }
 151
 152   template <typename T>
 153     T pop(){
 154     T temp = read<T>(pos);
 155     pos += sizeof(T);
 156     return temp;
 157   }
 158
 159   template <typename T>
 160     T peek(){
 161     T temp = read<T>(pos);
 162     return temp;
 163   }
 164
 165   template <typename T0, typename T>
 166     T peekSecond(){
 167     T temp;
 168     memcpy(&temp, buf+pos+sizeof(T0), sizeof(T));
 169     return temp;
 170   }
 171
 172   int datalength(){
 173     return pos;
 174   }
 175
 176   void * buffer(){
 177     return (void*) buf;
 178   }
 179
 180   void freeBuf(){
 181     free(buf);
 182   }
 183
 184   ~compressedBuffer(){
 185     // don't free the buf because the user my have supplied the buffer
 186   }
 187
 188 };
 189
 190
 191
 192 compressedBuffer compressAvailableNewSumDetail(int max=10000);
 193 void mergeCompressedBin(compressedBuffer *srcBufferArray, int numSrcBuf, int *numProcsRepresentedInMessage, int totalProcsAcrossAllMessages, compressedBuffer &destBuffer);
 194 //void printSumDetailInfo(int desiredBinsToSend);
 195 CkReductionMsg *sumDetailCompressedReduction(int nMsg,CkReductionMsg **msgs);
 196 void printCompressedBuf(compressedBuffer b);
 197 compressedBuffer fakeCompressedMessage();
 198 compressedBuffer emptyCompressedBuffer();
 199 void sanityCheckCompressedBuf(compressedBuffer b);
 200 bool isCompressedBufferSane(compressedBuffer b);
 201 double averageUtilizationInBuffer(compressedBuffer b);
 202
 203
 204
 205
 206
 207 class TraceUtilization : public Trace {
 208  public:
 209   int execEp; // the currently executing EP
 210   unsigned int epInfoSize;
 211   double start; // the start time for the currently executing EP
 212
 213   double *cpuTime;     // NUM_BINS*epInfoSize
 214   int lastBinUsed;
 215   unsigned int numBinsSent;
 216   unsigned int previouslySentBins;
 217
 218
 219   TraceUtilization() {
 220     execEp = TRACEON_EP;
 221     cpuTime = NULL;
 222     lastBinUsed = -1;
 223     numBinsSent = 0;
 224   }
 225
 226
 227   /// Initialize memory after the number of EPs has been determined
 228   void initMem(){
 229     int _numEntries=_entryTable.size();
 230     epInfoSize = _numEntries + NUM_DUMMY_EPS + 1; // keep a spare EP
 231     //    CkPrintf("allocating cpuTime[%d]\n", (int) (NUM_BINS*epInfoSize));
 232     cpuTime = new double[NUM_BINS*epInfoSize];
 233     _MEMCHECK(cpuTime);
 234
 235     if(CkMyPe() == 0)
 236       writeSts();
 237
 238   }
 239
 240   void writeSts(void);
 241
 242   void creation(envelope *e, int epIdx, int num=1) {}
 243
 244   void beginExecute(envelope *e, void *obj);
 245   void beginExecute(CmiObjId  *tid);
 246   void beginExecute(int event,int msgType,int ep,int srcPe, int mlen=0, CmiObjId *idx=NULL, void *obj=NULL);
 247   void endExecute(void);
 248   void beginIdle(double currT) {}
 249   void endIdle(double currT) {}
 250   void beginPack(void){}
 251   void endPack(void) {}
 252   void beginUnpack(void) {}
 253   void endUnpack(void) {}
 254   void beginComputation(void) {  initMem(); }
 255   void endComputation(void) {}
 256   void traceClearEps() {}
 257   void traceWriteSts() {}
 258   void traceClose() {}
 259
 260   void addEventType(int eventType);
 261
 262   int cpuTimeEntriesAvailable() const { return lastBinUsed+1; }
 263   int cpuTimeEntriesSentSoFar() const { return numBinsSent; }
 264   void incrementNumCpuTimeEntriesSent(int n) { numBinsSent += n; }
 265
 266
 267   double sumUtilization(int startBin, int endBin);
 268
 269
 270   void updateCpuTime(int epIdx, double startTime, double endTime){
 271
 272     //    CkPrintf("updateCpuTime(startTime=%lf endTime=%lf)\n", startTime, endTime);
 273
 274     if (epIdx >= epInfoSize) {
 275       CkPrintf("WARNING: epIdx=%d >=  epInfoSize=%d\n", (int)epIdx, (int)epInfoSize );
 276       return;
 277     }
 278
 279     int startingBinIdx = (int)(startTime/BIN_SIZE);
 280     int endingBinIdx = (int)(endTime/BIN_SIZE);
 281
 282     if (startingBinIdx == endingBinIdx) {
 283       addToCPUtime(startingBinIdx, epIdx, endTime - startTime);
 284     } else if (startingBinIdx < endingBinIdx) { // EP spans intervals
 285       addToCPUtime(startingBinIdx, epIdx, (startingBinIdx+1)*BIN_SIZE - startTime);
 286       while(++startingBinIdx < endingBinIdx)
 287         addToCPUtime(startingBinIdx, epIdx, BIN_SIZE);
 288       addToCPUtime(endingBinIdx, epIdx, endTime - endingBinIdx*BIN_SIZE);
 289     }
 290   }
 291
 292
 293   /// Zero out all entries from (lastBinUsed+1) up to and including interval.
 294   inline void zeroIfNecessary(unsigned int interval){
 295     for(unsigned int i=lastBinUsed+1; i<= interval; i++){
 296       // zero all eps for this bin
 297       for(unsigned int j=0;j<epInfoSize;j++){
 298         cpuTime[(i%NUM_BINS)*epInfoSize+j] = 0.0;
 299       }
 300     }
 301     lastBinUsed = interval;
 302   }
 303
 304   UInt getEpInfoSize() { return epInfoSize; }
 305
 306   /// for Summary-Detail
 307   inline double getCPUtime(unsigned int interval, unsigned int ep) {
 308     CkAssert(ep < epInfoSize);
 309     if(cpuTime != NULL && interval <= lastBinUsed)
 310       return cpuTime[(interval%NUM_BINS)*epInfoSize+ep];
 311     else {
 312       CkPrintf("getCPUtime called with invalid options: cpuTime=%p lastBinUsed=%d interval=%d ep=%d\n", cpuTime, lastBinUsed, (int)interval, (int)ep);
 313       return 0.0;
 314     }
 315   }
 316
 317   inline void addToCPUtime(unsigned int interval, unsigned int ep, double val){
 318     //    CkAssert(ep < epInfoSize);
 319     zeroIfNecessary(interval);
 320     //    CkPrintf("addToCPUtime interval=%d\n", (int)interval);
 321     cpuTime[(interval%NUM_BINS)*epInfoSize+ep] += val;
 322   }
 323
 324
 325   inline double getUtilization(int interval, int ep){
 326     return getCPUtime(interval, ep) * 100.0 * (double)BIN_PER_SEC;
 327   }
 328
 329
 330   compressedBuffer compressNRecentSumDetail(int desiredBinsToSend);
 331
 332
 333 };
 334
 335
 336
 337
 338
 339
 340 class TraceUtilizationBOC : public CBase_TraceUtilizationBOC {
 341
 342   std::deque<CkReductionMsg *> storedSumDetailResults;
 343
 344  public:
 345   TraceUtilizationBOC() {}
 346   TraceUtilizationBOC(CkMigrateMessage* msg) {}
 347   ~TraceUtilizationBOC() {}
 348
 349
 350   /// Entry methods:
 351   void ccsRequestSumDetailCompressed(CkCcsRequestMsg *m);
 352   void collectSumDetailData();
 353   void sumDetailDataCollected(CkReductionMsg *);
 354
 355 };
 356
 357
 358
 359
 360
 361 #endif
 362
 363 /*@}*/