LdbCoordinator.C

Go to the documentation of this file.

/*****************************************************************************
 * $Source: /home/cvs/namd/cvsroot/namd2/src/LdbCoordinator.C,v $
 * $Author: bhatele $
 * $Date: 2008/09/24 19:39:25 $
 * $Revision: 1.88 $
 *****************************************************************************/

#include <stdlib.h>

#include "InfoStream.h"
#include "NamdCentLB.h"
#include "NamdNborLB.h"

#include "HomePatch.h"
#include "LdbCoordinator.decl.h"
#include "LdbCoordinator.h"
#include "NamdTypes.h"
#include "Node.h"
#include "SimParameters.h"
#include "PatchMap.inl"
#include "ComputeMap.h"
//#define DEBUGM
#define MIN_DEBUG_LEVEL 3
#include "Debug.h"
#include "Controller.h"
#include "Sequencer.h"
#include "RefineOnly.h"
#include "ComputeMgr.h"
#include "packmsg.h"

#include "elements.h"
#include "ComputeMgr.decl.h"

#define DEBUG_LEVEL 4

#if CONVERSE_VERSION_ELAN
extern "C" void enableBlockingReceives();
extern "C" void disableBlockingReceives();
#endif

void LdbCoordinator_initproc() {
#if CHARM_VERSION >= 50804
  // Set the load balancing period (in seconds).  Without this the
  // load balancing framework will hang until 1 second has passed
  // since the last load balancing, causing hiccups in very fast runs.
  // This is duplicated below for older versions, but putting it here
  // also fixes the first load balance.
  LBSetPeriod(1.0e-5);
#endif
}

void LdbCoordinator::staticMigrateFn(LDObjHandle handle, int dest)
{
#if CHARM_VERSION > 050606
   LdbCoordinator *ldbCoordinator = (LdbCoordinator *)LDOMUserData(handle.omhandle);
   ldbCoordinator->Migrate(handle,dest);
#else
   ((LdbCoordinator*)handle.omhandle.user_ptr)->Migrate(handle,dest);
#endif
}

void LdbCoordinator::Migrate(LDObjHandle handle, int dest)
{
  LdbMigrateMsg* msg = new LdbMigrateMsg;
  msg->handle = handle;
  msg->from = CkMyPe();
  msg->to = dest;
#if CHARM_VERSION > 050402
  CProxy_LdbCoordinator ldbProxy(thisgroup);
  ldbProxy[CkMyPe()].RecvMigrate(msg);
#else
  CProxy_LdbCoordinator(thisgroup).RecvMigrate(msg,CkMyPe());
#endif
}

void LdbCoordinator::staticStatsFn(LDOMHandle h, int state)
{
  CkPrintf("I'm supposed to set stats\n");
}

void LdbCoordinator::staticQueryEstLoadFn(LDOMHandle h)
{
  CkPrintf("I'm supposed to query load\n");
}

void LdbCoordinator::staticReceiveAtSync(void* data)
{

#if CONVERSE_VERSION_ELAN
    //disableBlockingReceives();
#endif

  ((LdbCoordinator*)data)->ReceiveAtSync();
}

void LdbCoordinator::ReceiveAtSync()
{
  theLbdb->RegisteringObjects(myHandle);
}

void LdbCoordinator::staticResumeFromSync(void* data)
{
  ((LdbCoordinator*)data)->ResumeFromSync();
}

void LdbCoordinator::ResumeFromSync()
{
  theLbdb->DoneRegisteringObjects(myHandle);
#if CMK_PERSISTENT_COMM
  if (takingLdbData) {
//CmiPrintf("[%d] CmiDestoryPersistent\n", CkMyPe());

    HomePatchList *hpl = PatchMap::Object()->homePatchList();
    ResizeArrayIter<HomePatchElem> ai(*hpl);
    for (ai=ai.begin(); ai != ai.end(); ai++) {
      HomePatch *patch = (*ai).patch;
      patch->destoryPersistComm();
    }
  }
#endif
  CProxy_LdbCoordinator cl(thisgroup);
#if CHARM_VERSION > 050402
  cl[0].nodeDone();
#else
  cl.nodeDone(0);
#endif
}

LdbCoordinator::LdbCoordinator()
{
  if (CkpvAccess(LdbCoordinator_instance) == NULL) {
    CkpvAccess(LdbCoordinator_instance) = this;
  } else {
    iout << iFILE << iERROR << iPE 
         << "LdbCoordinator instanced twice on same node!" << endi;
    CkExit();
  }
  
#if 0
  // Create a load balancer
  if (CkMyPe() == 0) {
    //   CreateCentralLB();
    CreateNamdCentLB();
    //   CreateNamdNborLB();
  }
#endif

  ldbCycleNum = 1;
  takingLdbData = 1;
  totalStepsDone = 0;
  nLocalComputes = nLocalPatches = 0;
  patchNAtoms = (int *) NULL;
  sequencerThreads = (Sequencer **) NULL;
  ldbStatsFP = NULL;
  computeArray = NULL;
  patchArray = NULL;
  processorArray = NULL;

  nodesDone = 0;

  // Register self as an object manager for new charm++ balancer framework
#if CHARM_VERSION > 050610
  theLbdb = LBDatabase::Object(); 
#else
  theLbdb = CProxy_LBDatabase::ckLocalBranch(lbdb);
#endif

  // Set the load balancing period (in seconds).  Without this the
  // load balancing framework will hang until 1 second has passed
  // since the last load balancing, causing hiccups in very fast runs.
  // Unfortunately, the clock is already set for the first load
  // balancing, but only +LBPeriod 1.0e-5 can fix that in older charm.
  // For newer versions this is handled in initproc above.

  theLbdb->SetLBPeriod(1.0e-5);

#if CHARM_VERSION > 050403
  myOMid.id.idx = 1;
#else
  myOMid.id = 1;
#endif
  LDCallbacks cb = { (LDMigrateFn)staticMigrateFn,
                     (LDStatsFn)staticStatsFn,
                     (LDQueryEstLoadFn)staticQueryEstLoadFn
                   };
  myHandle = theLbdb->RegisterOM(myOMid,(void*)this,cb);

  // Add myself as a local barrier receiver, so I know when I might
  // be registering objects.
  theLbdb->AddLocalBarrierReceiver((LDBarrierFn)staticReceiveAtSync,
                                   (void*)this);;

  // Also, add a local barrier client, to trigger load balancing
  ldBarrierHandle = theLbdb->
    AddLocalBarrierClient((LDResumeFn)staticResumeFromSync,
                          (void*)this);
  objHandles = 0;
  reg_all_objs = 1;

}

LdbCoordinator::~LdbCoordinator(void)
{
  delete [] patchNAtoms;
  delete [] sequencerThreads;
  delete [] objHandles;
  if (CkMyPe() == 0)
  {
    delete [] computeArray;
    delete [] patchArray;
    delete [] processorArray;
  }
  if (ldbStatsFP)
    fclose(ldbStatsFP);

}

void LdbCoordinator::createLoadBalancer()
{
  if (CkMyPe()==0) 
    iout << "LDB: Measuring processor speeds ..." << endi;
  const SimParameters *simParams = Node::Object()->simParameters;
  /*if (simParams->ldbStrategy == LDBSTRAT_ALGNBOR) 
    CreateNamdNborLB();
  else {*/
    CreateNamdCentLB();
  if (CkMyPe()==0)
    iout << " Done.\n" << endi;
}

void LdbCoordinator::initialize(PatchMap *pMap, ComputeMap *cMap, int reinit)
{
  const SimParameters *simParams = Node::Object()->simParameters;

#if 0
  static int lbcreated = 0; // XXX static variables are unsafe for SMP
  // PE0 first time Create a load balancer
  if (CkMyPe() == 0 && !lbcreated) {
    if (simParams->ldbStrategy == LDBSTRAT_ALGNBOR) 
      CreateNamdNborLB();
    else {
      //   CreateCentralLB();
      CreateNamdCentLB();
    }
    lbcreated = 1;
  }
#endif

  //  DebugM(10,"stepsPerLdbCycle initialized\n");
  stepsPerLdbCycle = simParams->ldbPeriod;
  firstLdbStep = simParams->firstLdbStep;
  int lastLdbStep = simParams->lastLdbStep;

  computeMap = cMap;
  patchMap = pMap;

  // Set the number of received messages correctly for node 0

  nStatsMessagesExpected = Node::Object()->numNodes();
  nStatsMessagesReceived = 0;

  if (patchNAtoms) 
    delete [] patchNAtoms;  // Depends on delete NULL to do nothing
  nPatches = patchMap->numPatches();
  patchNAtoms = new int[nPatches];

  typedef Sequencer *seqPtr;

  if ( ! reinit ) {
    delete [] sequencerThreads;  // Depends on delete NULL to do nothing
    sequencerThreads = new seqPtr[nPatches];
  }

  nLocalPatches=0;

  int i;
  for(i=0;i<nPatches;i++)
  {
    if (patchMap->node(i) == Node::Object()->myid())
    {
      nLocalPatches++;
      patchNAtoms[i]=0;
    } else {
      patchNAtoms[i]=-1;
    }
    if ( ! reinit ) sequencerThreads[i]=NULL;
  }
  if ( ! reinit ) controllerThread = NULL;
  if (nLocalPatches != patchMap->numHomePatches())
    NAMD_die("Disaggreement in patchMap data.\n");
 
  nLocalComputes = 0;
  numComputes = cMap->numComputes();

  for(i=0;i<numComputes;i++)  {
    if ( (computeMap->node(i) == Node::Object()->myid())
         && ( (computeMap->type(i) == computeNonbondedPairType)
              || (computeMap->type(i) == computeSelfBondsType)
              || (computeMap->type(i) == computeSelfAnglesType)
              || (computeMap->type(i) == computeSelfDihedralsType)
              || (computeMap->type(i) == computeSelfImpropersType)
              || (computeMap->type(i) == computeSelfCrosstermsType)
              || (computeMap->type(i) == computeNonbondedSelfType) ) ) {
      nLocalComputes++;
    }
  }
  
  // New LB frameworks registration

  // Allocate data structure to save incoming migrations.  Processor
  // zero will get all migrations

  // If this is the first time through, we need it register patches
  if (reg_all_objs) {
    // Tell the lbdb that I'm registering objects, until I'm done
    // registering them.
    theLbdb->RegisteringObjects(myHandle);
    
    patchHandles = new LDObjHandle[nLocalPatches];
    int patch_count=0;
    int i;
    for(i=0;i<nPatches;i++)
      if (patchMap->node(i) == Node::Object()->myid()) {
        LDObjid elemID;
        elemID.id[0] = i;
        elemID.id[1] = elemID.id[2] = elemID.id[3] = -2;

        if (patch_count >= nLocalPatches) {
          iout << iFILE << iERROR << iPE 
               << "LdbCoordinator found too many local patches!" << endi;
          CkExit();
        }
        patchHandles[patch_count] 
          = theLbdb->RegisterObj(myHandle,elemID,0,0);
        patch_count++;
      }
  
    // Allocate new object handles
    if (objHandles == 0) {
      objHandles = new LDObjHandle[numComputes];
      for(i=0; i<numComputes; i++)
        objHandles[i].id.id[0] = -1; // Use -1 to mark unused entries

      // Register computes
      for(i=0; i<numComputes; i++)  {
        if ( (computeMap->node(i) == Node::Object()->myid())
             && ( (computeMap->type(i) == computeNonbondedPairType)
                  || (computeMap->type(i) == computeSelfBondsType)
                  || (computeMap->type(i) == computeSelfAnglesType)
                  || (computeMap->type(i) == computeSelfDihedralsType)
                  || (computeMap->type(i) == computeSelfImpropersType)
                  || (computeMap->type(i) == computeSelfCrosstermsType)
                  || (computeMap->type(i) == computeNonbondedSelfType) ) ) {
          // Register the object with the load balancer
          // Store the depended patch IDs in the rest of the element ID
          LDObjid elemID;
          elemID.id[0] = i;
        
          if (cMap->numPids(i) > 2)
            elemID.id[3] = cMap->pid(i,2);
          else elemID.id[3] = -1;

          if (cMap->numPids(i) > 1)
            elemID.id[2] =  cMap->pid(i,1);
          else elemID.id[2] = -1;

          if (cMap->numPids(i) > 0)
            elemID.id[1] =  cMap->pid(i,0);
          else elemID.id[1] = -1;

          objHandles[i] = theLbdb->RegisterObj(myHandle,elemID,0,1);
        }
      }
    }
    theLbdb->DoneRegisteringObjects(myHandle);
    reg_all_objs = 0;
  }

  // Fixup to take care of the extra timestep at startup
  // This is pretty ugly here, but it makes the count correct
  
  // iout << "LDB Cycle Num: " << ldbCycleNum << "\n";

  if (ldbCycleNum==1)
  {
    totalStepsDone += firstLdbStep;
    numStepsToRun = firstLdbStep;
    takingLdbData = 0;
    theLbdb->CollectStatsOff();
  }
  else if ( (ldbCycleNum <= 4) || !takingLdbData )
  {
    totalStepsDone += firstLdbStep;
    if(lastLdbStep != -1 && totalStepsDone > lastLdbStep) {
      numStepsToRun = -1;
      takingLdbData = 0;
      theLbdb->CollectStatsOff();
    } else {
      numStepsToRun = firstLdbStep;
      takingLdbData = 1;
      theLbdb->CollectStatsOn();
    }
  }
  else 
  {
    totalStepsDone += stepsPerLdbCycle - firstLdbStep;
    if(lastLdbStep != -1 && totalStepsDone > lastLdbStep) {
      numStepsToRun = -1;
      takingLdbData = 0;
      theLbdb->CollectStatsOff();
    } else {
      numStepsToRun = stepsPerLdbCycle - firstLdbStep;
      takingLdbData = 0;
      theLbdb->CollectStatsOff();
    }
  }

/*------------------------------------------------------------------------------*
 * ---------------------------------------------------------------------------- *
 * Comments inserted by Abhinav to clarify relation between ldbCycleNum,        *
 * load balancing step numbers (printed by the step() function) and             *
 * tracing of the steps                                                         *
 * ---------------------------------------------------------------------------- *
 * If trace is turned off in the beginning, then tracing is turned on           *
 * at ldbCycleNum = 4 and turned off at ldbCycleNum = 8. ldbCycleNum can        *
 * be adjusted by specifying firstLdbStep and ldbPeriod which are set by        *
 * default to 5*stepspercycle and 200*stepspercycle if not specified.           *
 *                                                                              *
 * If we choose firstLdbStep = 20 and ldbPeriod = 100, we have the              *
 * following timeline (for these particular numbers):                           *
 *                                                                              *
 * Tracing         :  <------ off ------><------------- on ------------><-- off *
 * Ldb Step() No   :              1     2     3        4      5        6      7 *
 * Iteration Steps : 00====20====40====60====80======160====180======260====280 *
 * ldbCycleNum     :  1     2     3     4     5        6      7        8      9 *
 * Instrumention   :          Inst  Inst  Inst           Inst            Inst   *
 * LDB Strategy    :             Alg7  Ref   Ref             Ref            Ref *
 *                                                                              *
 * Alg7 = AlgSeven                                                              *
 * Ref  = Refine (NamdCentLB.C, Rebalancer.C)                                   *
 * Inst = Instrumentation Phase (no real load balancing)                        *
 * ---------------------------------------------------------------------------- *
 *------------------------------------------------------------------------------*
 */

#if CHARM_VERSION >= 050606
  if (traceAvailable()) {
    static int specialTracing = 0; // XXX static variables are unsafe for SMP
    if (ldbCycleNum == 1 && traceIsOn() == 0)  specialTracing = 1;
    if (specialTracing) {
      if (ldbCycleNum == 4) traceBegin();
      if (ldbCycleNum == 8) traceEnd();
    }
  }
#endif

  nPatchesReported = 0;
  nPatchesExpected = nLocalPatches;
  nComputesReported = 0;
  nComputesExpected = nLocalComputes * numStepsToRun;
  controllerReported = 0;
  controllerExpected = ! CkMyPe();

  if (CkMyPe() == 0)
  {
    if (computeArray == NULL)
      computeArray = new computeInfo[numComputes];
    if (patchArray == NULL)
      patchArray = new patchInfo[nPatches];
    if (processorArray == NULL)
      processorArray = new processorInfo[CkNumPes()];
  }
    
  theLbdb->ClearLoads();
}

void LdbCoordinator::patchLoad(PatchID id, int nAtoms, int /* timestep */)
{
  CmiAssert( id >=0 && id < nPatches);
  if (patchNAtoms[id] != -1) {
    patchNAtoms[id] = nAtoms;
    nPatchesReported++;
  } else {
    DebugM(10, "::patchLoad() Unexpected patch reporting in\n");
  }
}

void LdbCoordinator::startWork(ComputeID id, int /* timestep */ )
{
  CmiAssert(id >=0 && id < numComputes);
  theLbdb->ObjectStart(objHandles[id]);
}

void LdbCoordinator::endWork(ComputeID id, int /* timestep */)
{
  CmiAssert(id >=0 && id < numComputes);
  theLbdb->ObjectStop(objHandles[id]);
  nComputesReported++;
}

void LdbCoordinator::rebalance(Sequencer *seq, PatchID pid)
{
  if (Node::Object()->simParameters->ldbStrategy == LDBSTRAT_NONE)
    return;

  sequencerThreads[pid] = seq;
  seq->suspend();
}

void LdbCoordinator::rebalance(Controller *c)
{
  if (Node::Object()->simParameters->ldbStrategy == LDBSTRAT_NONE)
    return;

  iout << "LDB: ============= START OF LOAD BALANCING ============== " << CmiWallTimer() << "\n" << endi;
  DebugM(3, "Controller reached load balance barrier.\n");
  controllerReported = 1;
  controllerThread = c;

  CProxy_LdbCoordinator(thisgroup).barrier();

  CthSuspend();
}

void LdbCoordinator::barrier(void)
{
  if ( (nPatchesReported != nPatchesExpected) 
       || (nComputesReported != nComputesExpected)
       || (controllerReported != controllerExpected) )
  {
    NAMD_bug("Load balancer received wrong number of events.\n");
  }

  theLbdb->AtLocalBarrier(ldBarrierHandle);
}

void LdbCoordinator::nodeDone(void)
{
  nodesDone++;

  if (nodesDone==Node::Object()->numNodes()) {
    iout << "LDB: ============== END OF LOAD BALANCING =============== " << CmiWallTimer() << "\n\n" << endi;
    nodesDone=0;
    ExecuteMigrations();
  }
}

void LdbCoordinator::ExecuteMigrations(void)
{
 // computeMgr->updateComputes() call only on Node(0) i.e. right here
  // This will barrier for all Nodes - (i.e. Computes must be
  // here and with proxies before anyone can start up

  CProxy_ComputeMgr cm(CkpvAccess(BOCclass_group).computeMgr);
  ComputeMgr *computeMgr = cm.ckLocalBranch();
#if CHARM_VERSION > 050402
  computeMgr->updateComputes(CkIndex_LdbCoordinator::
                             updateComputesReady(),thisgroup);
#else
  computeMgr->updateComputes(CProxy_LdbCoordinator::
                             ckIdx_updateComputesReady(),thisgroup);
#endif
}

void LdbCoordinator::RecvMigrate(LdbMigrateMsg* m)
{
  // This method receives the migration from the framework,
  // unregisters it, and sends it to the destination PE
  const int id = m->handle.id.id[0];

  theLbdb->UnregisterObj(objHandles[id]);
  objHandles[id].id.id[0] = -1;

#if CHARM_VERSION > 050402
  CProxy_LdbCoordinator  ldbProxy(thisgroup);
  ldbProxy[m->to].ExpectMigrate(m);
#else
  CProxy_LdbCoordinator(thisgroup).ExpectMigrate(m,m->to);
#endif
}

void LdbCoordinator::ExpectMigrate(LdbMigrateMsg* m)
{
  objHandles[m->handle.id.id[0]] 
    = theLbdb->RegisterObj(myHandle,m->handle.id,0,1);

  theLbdb->Migrated(objHandles[m->handle.id.id[0]]);

  delete m;
}

void LdbCoordinator::updateComputesReady() {
  DebugM(3,"updateComputesReady()\n");

  CProxy_LdbCoordinator(thisgroup).resume();
#if CHARM_VERSION > 050402
  CkStartQD(CkIndex_LdbCoordinator::resumeReady((CkQdMsg*)0),&thishandle);
#else
  CkStartQD(CProxy_LdbCoordinator::ckIdx_resumeReady((CkQdMsg*)0),&thishandle);
#endif
}

void LdbCoordinator::resume(void)
{
  DebugM(3,"resume()\n");

  //  printLocalLdbReport();

  ldbCycleNum++;
  initialize(PatchMap::Object(),ComputeMap::Object(),1);
}

void LdbCoordinator::resumeReady(CkQdMsg *msg) {

  DebugM(3,"resumeReady()\n");
  delete msg;

  CProxy_LdbCoordinator(thisgroup).resume2();
}

void LdbCoordinator::resume2(void)
{
  DebugM(3,"resume2()\n");

#if CONVERSE_VERSION_ELAN
  //  enableBlockingReceives();
#endif

  awakenSequencers();
}

void LdbCoordinator::awakenSequencers()
{
  if (controllerThread)
  {
    controllerThread->awaken();
    controllerThread = NULL;
  }
  for(int i=0; i < patchMap->numPatches(); i++)
  {
    if (sequencerThreads[i])
    {
      sequencerThreads[i]->awaken();
    }
    sequencerThreads[i]= NULL;
  }
}

// Figure out which proxies we will definitely create on other
// nodes, without regard for non-bonded computes.  This code is swiped
// from ProxyMgr, and changes there probable need to be propagated here.

int LdbCoordinator::requiredProxies(PatchID id, int neighborNodes[])
{
  enum proxyHere { No, Yes };
  int numNodes = CkNumPes();
  proxyHere *proxyNodes = new proxyHere[numNodes];
  int nProxyNodes;
  int i;

  // Note all home patches.
  for ( i = 0; i < numNodes; ++i )
  {
    proxyNodes[i] = No;
  }
  nProxyNodes=0;

  // Check all two-away neighbors.
  // This is really just one-away neighbors, since 
  // two-away always returns zero: RKB
  PatchID neighbors[1 + PatchMap::MaxOneAway + PatchMap::MaxTwoAway];

  int myNode = patchMap->node(id);
  neighbors[0] = id;
  int numNeighbors = 1 + patchMap->downstreamNeighbors(id,neighbors+1);
  for ( i = 0; i < numNeighbors; ++i )
  {
    const int proxyNode = patchMap->basenode(neighbors[i]);
    if (proxyNode != myNode)
      if (proxyNodes[proxyNode] == No)
      {
        proxyNodes[proxyNode] = Yes;
        neighborNodes[nProxyNodes] = proxyNode;
        nProxyNodes++;
      }
  }

  delete [] proxyNodes;
  return nProxyNodes;
}

void LdbCoordinator::printLocalLdbReport(void)
{
  char outputBuf[255];
  char *curLoc;

  CkPrintf("%d:Patch report:\n",CkMyPe());
  
  curLoc = outputBuf;
  int i,j=0;
  for(i=0; i<patchMap->numPatches(); i++)
  {
    if (patchNAtoms[i] != -1)
    {
      curLoc += sprintf(curLoc,"%5d: %5d ",i,patchNAtoms[i]);
      j++;
    } 
    if (((j % 4) == 0) && j)
    {
      curLoc = outputBuf;
      CkPrintf("[%d]%s\n",CkMyPe(),outputBuf);
      j=0;
    }
  }

  CkPrintf("%d:Compute report:\n",CkMyPe());
  
  curLoc = outputBuf;
  j=0;
}

void LdbCoordinator::printRequiredProxies(PatchID id, FILE *fp)
{
  // Check all two-away neighbors.
  // This is really just one-away neighbors, since 
  // two-away always returns zero: RKB
  int neighborNodes[PatchMap::MaxOneAway + PatchMap::MaxTwoAway];
  const int nProxyNodes = requiredProxies(id,neighborNodes);

  fprintf(fp,"%4d ",nProxyNodes);

  for(int i=0;i<nProxyNodes;i++)
    fprintf(fp,"%4d ",neighborNodes[i]);
}

#include "LdbCoordinator.def.h"

---