Home Research Papers Posters Manuals Talks Download People Help Internal

ck-com/ComlibManager.h

Go to the documentation of this file.

#ifndef COMLIBMANAGER_H
#define COMLIBMANAGER_H


#include <math.h>
#include <map>
#include <set>

#include "convcomlibmanager.h"
#include "ComlibStrategy.h"
#include "ComlibArrayListener.h"
#include "cksection.h"

#define CHARM_MPI 0 
#define MAX_NSTRAT 1024
#define LEARNING_PERIOD 1000 //Number of iterations after which the
                             //learning framework will discover 
                             //the appropriate strategy, not completely 
                             //implemented
#include "ComlibStats.h"

#include "comlib.decl.h"

CkpvExtern(CkGroupID, cmgrID);
class ComlibDummyMsg: public CMessage_ComlibDummyMsg {
    int dummy;
};



// TO BE MOVED TO MULTICAST SPECIFIC
struct ComlibMulticastIndexCount {
  int pe;
  int count;
};

inline int indexCountCompare(const void *a, const void *b) {
  ComlibMulticastIndexCount a1 = *(ComlibMulticastIndexCount*) a;
  ComlibMulticastIndexCount b1 = *(ComlibMulticastIndexCount*) b;

  if(a1.pe < b1.pe)
    return -1;
  
  if(a1.pe == b1.pe)
    return 0;
  
  if(a1.pe > b1.pe)
    return 1;
  
  return 0;
}


class ComlibMulticastMsg : public CkMcastBaseMsg, 
               public CMessage_ComlibMulticastMsg {
  public:
    int nPes;
    ComlibMulticastIndexCount *indicesCount;
    CkArrayIndex *indices;
    char *usrMsg;        
};


void ComlibAssociateProxy(ComlibInstanceHandle cinst, CProxy &proxy);
void ComlibAssociateProxy(Strategy *strat, CProxy &proxy); 

ComlibInstanceHandle ComlibRegister(Strategy *strat);

void ComlibBegin(CProxy &proxy, int iteration);    
void ComlibEnd(CProxy &proxy, int iteration);    

inline void ComlibResetSectionProxy(CProxySection_ArrayBase &sproxy) {
  sproxy.ckGetSectionInfo().info.sInfo.cInfo.id = 0;
}


void ComlibInitSectionID(CkSectionID &sid);

class LBMigrateMsg;

class ComlibManager: public CkDelegateMgr {
  //friend class ComlibInstanceHandle;

    int *bcast_pelist;  //Pelist passed to all broadcast operations

    int section_send_event;

    CkArrayIndex dummyArrayIndex;

    ConvComlibManager *converseManager;


    std::map<ComlibInstanceHandle, std::set<CharmMessageHolder*> > delayMessageSendBuffer;
   
 
    int setupComplete;
  
    int prioEndIterationFlag;

    ComlibGlobalStats clib_gstats; 
    int numStatsReceived;

    int curComlibController;   //Processor where strategies are created
    int clibIteration;         //Number of such learning iterations,
                               //each of which is triggered by a
                               //loadbalancing operation

    void init(); 

    //charm_message for multicast for a section of that group
    void multicast(CharmMessageHolder *cmsg, int instid); //charm message holder here
    //void multicast(void *charm_msg, int npes, int *pelist);

    //The following funtions can be accessed only from the user provided hooks
    friend void ComlibBegin(CProxy&, int iteration);
    friend void ComlibEnd(CProxy&, int iteration);

    void beginIteration(int instid, int iteration);
    
    void endIteration(int instid, int iteration);
    
    void printPeList(const char* note, int *peList);

    
    bool shouldBufferMessagesNow(int instid);
    void sendBufferedMessages(int instid, int step=-1);
    void sendBufferedMessagesAllStrategies();

 public:

    ComlibLocalStats clib_stats;   //To store statistics of
                                   //communication operations
    
    ComlibManager();  //Recommended constructor

    ComlibManager(CkMigrateMessage *m) { }
    int useDefCtor(void){ return 1; } //Use default constructor should
    //be pupped and store all the strategies.

    /* Initialization routines */
    
    void barrier(void);
    void resumeFromSetupBarrier();
   
    /* The delegation framework reimplemented functions */

    void ArraySend(CkDelegateData *pd,int ep, void *msg, 
                   const CkArrayIndex &idx, CkArrayID a);
    void GroupSend(CkDelegateData *pd, int ep, void *msg, int onpe, 
                   CkGroupID gid);
    void ArrayBroadcast(CkDelegateData *pd,int ep,void *m,CkArrayID a);
    void GroupBroadcast(CkDelegateData *pd,int ep,void *m,CkGroupID g);
    void ArraySectionSend(CkDelegateData *pd, int ep ,void *m, 
                                  int nsid, CkSectionID *s, int opts);
    CkDelegateData* ckCopyDelegateData(CkDelegateData *data); 
    CkDelegateData *DelegatePointerPup(PUP::er &p,CkDelegateData *pd);

    inline Strategy *getStrategy(int instid)
        {return converseManager->getStrategy(instid);}

    inline StrategyTableEntry *getStrategyTableEntry(int instid)
        {return converseManager->getStrategyTable(instid);}

    // Entry methods used by bracketed strategies when there is some object
    // migration. The comlib Manager realizes there is an error, it enters error
    // mode by suspending the delivery of the strategy if it has already
    // started, and it globally reduces the number of elements which already
    // called endIteration. When this number matches the number of elements
    // involved in the operation (or a multiple of it in case the user overlaps
    // iterations), the system ri-deliver the doneInserting to the strategy.
    //  void bracketedFinishSetup(int instid);
    void bracketedCatchUpPE(int instid, int step);
    void bracketedReceiveCount(int instid, int pe, int count, int isFirst, int step);
    void bracketedStartErrorRecoveryProcess(int instid, int step);
    void bracketedErrorDetected(int instid, int step);
    void bracketedConfirmCount(int instid, int step);
    void bracketedCountConfirmed(int instid, int count, int step);
    void bracketedReceiveNewCount(int instid, int step);
    void bracketedReceiveNewPeList(int instid, int step, int *count);

    void bracketedFinalBarrier(int instid, int step);
    void bracketedReleaseCount(int instid, int step);
    void bracketedReleaseBufferedMessages(int instid, int step);

    void bracketedStartDiscovery(int instid);
    void bracketedDiscover(int instid, CkArrayID aid, CkArrayIndex &idx, int isSrc);
    void bracketedContributeDiscovery(int instid, int pe, int nsrc, int ndest, int step);


    // TODO: Delete the following two methods!!!!
    void AtSync();           //User program called loadbalancer
    void lbUpdate(LBMigrateMsg *); //loadbalancing updates

    //Learning functions
    //void learnPattern(int totalMessageCount, int totalBytes);
    //void switchStrategy(int strat);

    //void collectStats(ComlibLocalStats &s, int src,CkVec<ClibGlobalArrayIndex>&);
    void collectStats(ComlibLocalStats &s, int src);
 

    void printDiagnostics(int cinst);

    void printDiagnostics();   
       
}; 

class ComlibDelegateData : public CkDelegateData {
 private:
  int _instid; 

  friend void ComlibAssociateProxy(ComlibInstanceHandle, CProxy&);
  friend CkDelegateData *ComlibManager::ckCopyDelegateData(CkDelegateData*);
  ComlibDelegateData(int instid);

 public:
  ComlibDelegateData(CkMigrateMessage *) : CkDelegateData() { ref(); }
  
  void beginIteration();
  void endIteration();

  inline int getID() {return _instid;}
  
  void pup(PUP::er &p) {
    p | _instid;
  }

};

void ComlibAtSync(void *msg);
void ComlibNotifyMigrationDoneHandler(void *msg);
void ComlibLBMigrationUpdate(LBMigrateMsg *);


#ifdef filippo
// The old interface
#define RECORD_SENDSTATS(sid, bytes, dest) {             \
        CProxy_ComlibManager cgproxy(CkpvAccess(cmgrID));               \
        cgproxy.ckLocalBranch()->clib_stats.recordSend(sid, bytes, dest); \
}\

#define RECORD_RECV_STATS(sid, bytes, src) { \
        CProxy_ComlibManager cgproxy(CkpvAccess(cmgrID)); \
        cgproxy.ckLocalBranch()->clib_stats.recordRecv(sid, bytes, src); \
}\

#define RECORD_SENDM_STATS(sid, bytes, dest_arr, ndest) {       \
        CProxy_ComlibManager cgproxy(CkpvAccess(cmgrID)); \
        cgproxy.ckLocalBranch()->clib_stats.recordSendM(sid, bytes, dest_arr, ndest); \
}\

#define RECORD_RECVM_STATS(sid, bytes, src_arr, nsrc) {        \
        CProxy_ComlibManager cgproxy(CkpvAccess(cmgrID)); \
        cgproxy.ckLocalBranch()->clib_stats.recordRecvM(sid, bytes, src_arr, nsrc); \
}\

#else
// the new version of comlib does not do anything with these yet
#define RECORD_SEND_STATS(sid, bytes, dest) /* empty */
#define RECORD_RECV_STATS(sid, bytes, src) /* empty */
#define RECORD_SENDM_STATS(sid, bytes, dest_arr, ndest) /* empty */
#define RECORD_RECVM_STATS(sid, bytes, src_arr, nsrc) /* empty */
#endif

#endif

---