Home Research Papers Posters Manuals Talks Download People Help Internal

ck-cp/pathHistory.h

Go to the documentation of this file.

#ifndef __PATH_HISTORY_H__
#define __PATH_HISTORY_H__


#if CMK_WITH_CONTROLPOINT


#include <vector>
#include <map>
#include <utility>
#include <cmath>
#include "PathHistory.decl.h"
#include "envelope.h"

#include <pup_stl.h>



void initializeCriticalPath(void);

void useThisCriticalPathForPriorities();
void automaticallySetMessagePriority(envelope *env);


class pathHistoryManager : public CBase_pathHistoryManager {
 private:
  std::map< std::pair<int,int>, int> criticalPathForPriorityCounts;
  
  pathInformationMsg *pathForUser; // A place to store the path while we perform a broadcast and reduction to save projections user events.

 public:

  pathHistoryManager();

  pathHistoryManager(CkMigrateMessage *m){
    CkAbort("pathHistoryManager does not have a working migration constructor."); 
  }
  
  void pup(PUP::er &p) { 
    CkAbort("pathHistoryManager cannot be pupped.");
  } 



 void traceCriticalPathBackStepByStep(pathInformationMsg *msg);



 void broadcastCriticalPathProjections(pathInformationMsg *msg);
 void criticalPathProjectionsDone(CkReductionMsg *msg);

 void saveCriticalPathForPriorities(pathInformationMsg *msg);


 void useCriticalPathForPriories();

 const std::map< std::pair<int,int>, int> & getCriticalPathForPriorityCounts() const { 
   return criticalPathForPriorityCounts;
 }
  
};



class MergeablePathHistory {
 public:
  double timeEntryMethodStarted;

  double preceding_path_time;

  int sender_pe;  // for traversing back over the PAG
  int sender_history_table_idx; // for traversing back over the PAG
  
  int local_ep;  // The locally executing EP
  int local_arr; // The locally executing array

#ifdef USE_CRITICAL_PATH_HEADER_ARRAY
 MergeablePathHistory(const envelope *env) 
   : sender_pe(env->getSrcPe()), 
    sender_history_table_idx(env->pathHistory.get_sender_history_table_idx()), 
    local_ep(env->getEpIdx()),
    local_arr(env->getArrayMgrIdx()),
    preceding_path_time(env->pathHistory.getTime()),
    timeEntryMethodStarted(0.0)
      {
        // No body
      }
#else
 MergeablePathHistory(const envelope *env) 
   : sender_pe(env->getSrcPe()), 
    local_ep(env->getEpIdx()),
    local_arr(env->getArrayMgrIdx()),
    timeEntryMethodStarted(0.0)
      {
        // No body
      }
#endif

  MergeablePathHistory() {
    reset();
  }
  
  void reset(){
    sender_pe = -1; 
    sender_history_table_idx = -1; 
    local_ep = -1;
    preceding_path_time = 0.0;
    timeEntryMethodStarted = 0.0;
  }
  
  void sanity_check(){
    if(sender_history_table_idx > -1){ 
      CkAssert(sender_pe < CkNumPes()); 
      CkAssert(sender_pe >= -1);
    }
/*     CkAssert(sender_history_table_idx < 100000); */
/*     CkAssert(sender_history_table_idx >= 0); */
   
/*     CkAssert(local_ep < 500); */
/*     CkAssert(local_ep >= 0); */

  }

  void updateMax(const MergeablePathHistory& other){
    if(preceding_path_time < other.preceding_path_time)
      *this = other;
  }


  void updateMax(const envelope* env){
    updateMax(MergeablePathHistory(env));
  }
  
  double getTotalTime() const{
    return preceding_path_time;
  }

  void printHTMLToString(char* buf) const{
#ifdef USE_CRITICAL_PATH_HEADER_ARRAY
    buf[0] = '\0';
    sprintf(buf+strlen(buf), "MergeablePathHistory time=%lf send pe=%d idx=%d timeEntryStarted=%lf", (double)preceding_path_time, (int)sender_pe, (int)sender_history_table_idx, (double)timeEntryMethodStarted);
#endif
  }

  void setDebug100(){
    preceding_path_time = 100.0;
    CkPrintf("Setting path length to 100\n");
  }

};


class PathHistoryTableEntry {
#ifdef USE_CRITICAL_PATH_HEADER_ARRAY

 public:
  int sender_pe;
  int sender_history_table_idx;
  int local_ep;
  int local_arr;
  int local_pe;

 private:
  double start_time;
  double local_path_time;
  double preceding_path_time;

 public:
  
 PathHistoryTableEntry() 
   : sender_pe(-1), 
    sender_history_table_idx(-1), 
    start_time(0.0),
    local_path_time(0.0), 
    preceding_path_time(0.0),     
    local_ep(-1),
    local_arr(-1),
    local_pe(CkMyPe())
      {
        // No body
      }
  

 PathHistoryTableEntry(const MergeablePathHistory& p, double localContribution = 0.0)
    : sender_pe(p.sender_pe), 
    sender_history_table_idx(p.sender_history_table_idx), 
    local_path_time(localContribution), 
    preceding_path_time(p.preceding_path_time),
    start_time(p.timeEntryMethodStarted),
    local_ep(p.local_ep),
    local_arr(p.local_arr),
    local_pe(CkMyPe())
      {
        // No body
      }

  PathHistoryTableEntry(const MergeablePathHistory& p, double start, double finish)
    : sender_pe(p.sender_pe), 
    sender_history_table_idx(p.sender_history_table_idx), 
    local_path_time(finish-start), 
    preceding_path_time(p.preceding_path_time),
    start_time(start),
    local_ep(p.local_ep),
    local_arr(p.local_arr),
    local_pe(CkMyPe())
      {
        // No body
      }

  void printInfo(char *prefix = ""){
    CkPrintf("%s [sender pe=%d table idx=%d] [local path contribution=%lf ep=%d] [Time= %lf + %lf]\n", prefix,  
             sender_pe, sender_history_table_idx, local_path_time, local_ep, preceding_path_time, local_path_time);  
  }


  int addToTableAndEnvelope(envelope *env);
  
  int addToTable();
  
  double getTotalTime(){
    return local_path_time + preceding_path_time;
  }

  double get_start_time() const {return start_time;}
  double get_local_path_time() const {return local_path_time;}
  double get_preceding_path_time() const {return preceding_path_time;}
#endif

};



void  saveCurrentPathAsUserEvent(const char* prefix="");

void setCurrentlyExecutingPathTo100(void);

void  printPathInMsg(void* msg);

void  printEPInfo();


void traceCriticalPathBack(CkCallback cb, bool saveToProjectionsTraces = false);


class pathInformationMsg : public CMessage_pathInformationMsg {
 public:
  PathHistoryTableEntry *history;
  int historySize;
  int saveAsProjectionsUserEvents;
  CkCallback cb;
  int table_idx;

  void printme(){
#ifdef USE_CRITICAL_PATH_HEADER_ARRAY
    CkPrintf("Path contains %d entries\n", historySize);
    for(int i=historySize-1;i>=0;i--){
      CkPrintf("\tPath Step %d: local_path_time=%lf arr=%d ep=%d starttime=%lf preceding path time=%lf pe=%d\n",i, history[i].get_local_path_time(),  history[i].local_arr, history[i].local_ep, history[i].get_start_time(), history[i].get_preceding_path_time(), history[i].local_pe);
    }
#endif    
  }
  
};


CkpvExtern(MergeablePathHistory, currentlyExecutingPath); // The maximal incoming path for the node
CkpvExtern(double, timeEntryMethodStarted);


#ifdef USE_CRITICAL_PATH_HEADER_ARRAY

typedef std::map<int, PathHistoryTableEntry> PathHistoryTableType;
CkpvExtern(PathHistoryTableType, pathHistoryTable);
CkpvExtern(int, pathHistoryTableLastIdx);




// Reset the counts for the currently executing message. Cut the incoming path
extern void resetThisEntryPath();

extern void criticalPath_start(envelope * env); 
extern void criticalPath_send(envelope * sendingEnv);
extern void criticalPath_end();
extern void criticalPath_split(); 





#define MERGE_PATH_DECLARE(x) MergeablePathHistory merge_path_##x

#define MERGE_PATH_RESET(x) merge_path_##x.reset()

#define MERGE_PATH_MAX(x) merge_path_##x.updateMax(CkpvAccess(currentlyExecutingPath)); CkpvAccess(currentlyExecutingPath) = merge_path_##x; 



#define MERGE_PATH_DECLARE_D(x) std::map<int,MergeablePathHistory> merge_path_D_##x

#define MERGE_PATH_RESET_D(x,n) merge_path_D_##x[n].reset()

#define MERGE_PATH_DELETE_D(x,n) merge_path_D_##x.erase(n)

#define MERGE_PATH_DELETE_ALL_D(x) merge_path_D_##x.clear()

#define MERGE_PATH_MAX_D(x,n) merge_path_D_##x[n].updateMax(CkpvAccess(currentlyExecutingPath)); CkpvAccess(currentlyExecutingPath) = merge_path_D_##x[n]; 



#else



#define MERGE_PATH_DECLARE(x) ;
#define MERGE_PATH_RESET(x) ;
#define MERGE_PATH_MAX(x) ;
#define MERGE_PATH_DECLARE_D(x) ;
#define MERGE_PATH_RESET_D(x,n) ;
#define MERGE_PATH_DELETE_D(x,n) ;
#define MERGE_PATH_DELETE_ALL_D(x) ;
#define MERGE_PATH_MAX_D(x,n) ;


#endif




#endif
#endif

---