Main Page | Namespace List | Class Hierarchy | Alphabetical List | Class List | File List | Class Members | File Members

Sync.C

Go to the documentation of this file.
00001 
00008 /*
00009     Sync will ensure that all homepatches finished updating before Computes starts and all proxies finished updating themselves.
00010 */
00011 
00012 #if !defined(WIN32) || defined(__CYGWIN__)
00013 #include <unistd.h>
00014 #endif
00015 #include <stdio.h>
00016 
00017 #include "InfoStream.h"
00018 #include "Patch.h"
00019 #include "PatchMap.h"
00020 #include "ProxyMgr.h"
00021 #include "Compute.h"
00022 #include "ComputeMap.h"
00023 
00024 #include "Sync.h"
00025 
00026 #define MIN_DEBUG_LEVEL 3
00027 //#define DEBUGM
00028 #include "Debug.h"
00029 
00030 // make sure all HomePatches get their positions data and sendProxyData to 
00031 // their proxies before computes get positionsReady.
00032 int useSync = 1;
00033 
00034 // useProxySync will make sure all proxies get updated before computes' 
00035 // positionsReady triggered so that real computation starts.
00036 // when these two combined, it will make sure that homepatch get all its force 
00037 // and positions data, and proxies receive its updated data before all 
00038 // computes start.
00039 int useProxySync = 0;
00040 
00041 Sync::Sync(): INCREASE(600), step(0), counter(0), homeReady(0)
00042 {
00043     if (CkpvAccess(Sync_instance) == NULL) {
00044         CkpvAccess(Sync_instance) = this;
00045     } else {
00046         iout << iFILE << iERROR << iPE
00047           << "Sync instanced twice on same processor!" << endi;
00048         CkExit();
00049     }
00050     capacity = INCREASE;
00051     clist = new _clist[capacity];
00052     cnum = 0;
00053     nPatcheReady = 0;
00054     numPatches = -1;
00055 }
00056 
00057 Sync::~Sync()
00058 {
00059   delete [] clist;
00060 }
00061 
00062 void Sync::openSync(void)
00063 {
00064   if (useSync) {
00065     // if use proxy spanning tree, proxy sync is forced
00066     if (!useProxySync && (proxySendSpanning || proxyRecvSpanning)) {
00067 #if !CMK_IMMEDIATE_MSG
00068       // Dont need proxy sync when immediate messges are turned on
00069       // If on BG/P, useProxySync should not be turned on for better performance
00070       #if !CMK_BLUEGENEP
00071       // CmiPrintf("[%d] useProxySync is turned on. \n", CkMyPe());
00072       useProxySync = 1;
00073       #endif
00074 #endif
00075     }
00076     // no proxies on this node, no need to use proxy sync.
00077     if (useProxySync && ProxyMgr::Object()->numProxies() == 0) {
00078       // CmiPrintf("[%d] useProxySync is turned off because no proxy. \n", CkMyPe());
00079       useProxySync = 0;
00080     }
00081     // if no proxy sync and no home patch, then disable home patch sync as well
00082     if (!useProxySync && PatchMap::Object()->numHomePatches() == 0) useSync = 0;
00083   }
00084   if(CkMyPe() == 0)
00085     CmiPrintf("[%d] useSync: %d, useProxySync: %d\n", CkMyPe(), useSync, useProxySync);
00086 }    
00087 
00088 // called from Patch::positionsReady()
00089 int Sync::holdComputes(PatchID pid, ComputeIDListIter cid, int doneMigration)
00090 {
00091   if (!useProxySync) {
00092     // only hold when homepatches are not ready
00093     PatchMap *patchMap = PatchMap::Object();
00094     if (homeReady) {
00095       triggerCompute();
00096       return 0;
00097     }
00098   }
00099 
00100   int slot = 0;
00101   for (; slot < cnum; slot++)
00102      if (clist[slot].pid == -1) break;
00103   if (slot == cnum) {
00104     cnum++;
00105     // table is full, expand the list
00106     if (cnum == capacity) {
00107       capacity += INCREASE;
00108       struct _clist *tmp = new _clist[capacity];
00109       memcpy(tmp, clist, cnum*sizeof(_clist));
00110       delete [] clist;
00111       clist = tmp;
00112       CmiPrintf("[%d] Info:: Sync buffer overflow and expanded!\n", CkMyPe());
00113     }
00114   }
00115 
00116   clist[slot].cid = cid;
00117   clist[slot].pid = pid;
00118   clist[slot].doneMigration  = doneMigration;
00119   clist[slot].step = PatchMap::Object()->patch(pid)->flags.sequence;
00120 
00121 //  CkPrintf("REG[%d]: patch:%d step:%d-%d slot:%d\n", CkMyPe(), pid, patchMap->patch(pid)->flags.sequence, step, slot);
00122 
00123   if (clist[slot].step == step) {
00124       nPatcheReady++;
00125       triggerCompute();
00126   }
00127   return 1;
00128 }
00129 
00130 // called from HomePatch::positionsReady()
00131 void Sync::PatchReady(void)
00132 {
00133   counter ++;
00134   triggerCompute();
00135 }
00136 
00137 void Sync::releaseComputes()
00138 {
00139   PatchMap *patchMap = PatchMap::Object();
00140   ComputeMap *computeMap = ComputeMap::Object();
00141 
00142   nPatcheReady = 0;
00143   for (int i= 0; i<cnum; i++) {
00144     int &pid = clist[i].pid;
00145     if (pid == -1) continue;
00146     if (clist[i].step != step) {
00147       // count for next step
00148       if (clist[i].step == step + 1) nPatcheReady++;
00149       continue;
00150     }
00151     //         CkPrintf(" %d-%d-%d ",
00152     //   clist[i].pid, clist[i].step,
00153     //      patchMap->patch(pid)->flags.sequence);
00154 
00155     ComputeIDListIter cid = clist[i].cid;
00156 
00157     int compute_count = 0;
00158     for(cid = cid.begin(); cid != cid.end(); cid++) {
00159       compute_count++;
00160       computeMap->compute(*cid)->patchReady(pid,clist[i].doneMigration,step);
00161     }
00162     if (compute_count == 0 && patchMap->node(pid) != CkMyPe()) {
00163            iout << iINFO << "PATCH_COUNT-Sync step " << step
00164                 << "]: Patch " << pid << " on PE " 
00165                 << CkMyPe() <<" home patch " 
00166                 << patchMap->node(pid) << " does not have any computes\n" 
00167                 << endi;
00168     }
00169     pid = -1;
00170   }
00171 //  CkPrintf("\n");
00172 }
00173 
00174 void Sync::triggerCompute()
00175 {
00176   PatchMap *patchMap = PatchMap::Object();
00177 
00178   if (numPatches == -1) 
00179     numPatches = ProxyMgr::Object()->numProxies() + patchMap->numHomePatches();
00180 
00181 //  CkPrintf("SYNC[%d]: PATCHREADY:%d %d patches:%d %d\n", CkMyPe(), counter, PatchMap::Object()->numHomePatches(), nPatcheReady, numPatches);
00182 
00183   if (homeReady == 0 && counter == patchMap->numHomePatches()) {
00184     homeReady = 1;
00185     if (!useProxySync)  releaseComputes();
00186   }
00187 
00188   if (homeReady && nPatcheReady == numPatches)
00189   {
00190 //     CkPrintf("TRIGGERED[%d]\n", CkMyPe());
00191     if (useProxySync) releaseComputes();
00192 
00193     // reset counter
00194     counter = 0;
00195     numPatches = -1;
00196     step++;
00197     homeReady = 0;
00198   }
00199 }
00200 
00201 
00202 #if 0
00203 // hack for SUN MPCC
00204 // force compiler to instantiate ArrayElementT
00205 void *frightenCompilerIntoInstantiatingHack(void) {
00206   return new ArrayElementT<int>;
00207 }
00208 #endif
00209 
00210 
00211 #include "Sync.def.h"

Generated on Sat Aug 30 04:07:41 2008 for NAMD by  doxygen 1.3.9.1