support for medium resilience
[charm.git] / src / ck-core / cklocation.C
1 /** \file cklocation.C
2  *  \addtogroup CkArrayImpl
3  *
4  *  The location manager keeps track of an indexed set of migratable objects.
5  *  It is used by the array manager to locate array elements, interact with the
6  *  load balancer, and perform migrations.
7  *
8  *  Orion Sky Lawlor, olawlor@acm.org 9/29/2001
9  */
10
11 #include "charm++.h"
12 #include "register.h"
13 #include "ck.h"
14 #include "trace.h"
15 #include "TopoManager.h"
16 #include <vector>
17 #include<sstream>
18
19 #if CMK_LBDB_ON
20 #include "LBDatabase.h"
21 #include "MetaBalancer.h"
22 #if CMK_GLOBAL_LOCATION_UPDATE
23 #include "BaseLB.h"
24 #include "init.h"
25 #endif
26 #endif // CMK_LBDB_ON
27
28 #if CMK_GRID_QUEUE_AVAILABLE
29 CpvExtern(void *, CkGridObject);
30 #endif
31
32 static const char *idx2str(const CkArrayMessage *m) {
33   return idx2str(((CkArrayMessage *)m)->array_index());
34 }
35
36 #define ARRAY_DEBUG_OUTPUT 0
37
38 #if ARRAY_DEBUG_OUTPUT 
39 #   define DEB(x) CkPrintf x  //General debug messages
40 #   define DEBI(x) CkPrintf x  //Index debug messages
41 #   define DEBC(x) CkPrintf x  //Construction debug messages
42 #   define DEBS(x) CkPrintf x  //Send/recv/broadcast debug messages
43 #   define DEBM(x) CkPrintf x  //Migration debug messages
44 #   define DEBL(x) CkPrintf x  //Load balancing debug messages
45 #   define DEBK(x) CkPrintf x  //Spring Cleaning debug messages
46 #   define DEBB(x) CkPrintf x  //Broadcast debug messages
47 #   define AA "LocMgr on %d: "
48 #   define AB ,CkMyPe()
49 #   define DEBUG(x) CkPrintf x
50 #   define DEBAD(x) CkPrintf x
51 #else
52 #   define DEB(X) /*CkPrintf x*/
53 #   define DEBI(X) /*CkPrintf x*/
54 #   define DEBC(X) /*CkPrintf x*/
55 #   define DEBS(x) /*CkPrintf x*/
56 #   define DEBM(X) /*CkPrintf x*/
57 #   define DEBL(X) /*CkPrintf x*/
58 #   define DEBK(x) /*CkPrintf x*/
59 #   define DEBB(x) /*CkPrintf x*/
60 #   define str(x) /**/
61 #   define DEBUG(x)   /**/
62 #   define DEBAD(x) /*CkPrintf x*/
63 #endif
64
65 //whether to use block mapping in the SMP node level
66 bool useNodeBlkMapping;
67
68 #if CMK_LBDB_ON
69 /*LBDB object handles are fixed-sized, and not necc.
70 the same size as ArrayIndices.
71 */
72 LDObjid idx2LDObjid(const CkArrayIndex &idx)
73 {
74   LDObjid r;
75   int i;
76   const int *data=idx.data();
77   if (OBJ_ID_SZ>=idx.nInts) {
78     for (i=0;i<idx.nInts;i++)
79       r.id[i]=data[i];
80     for (i=idx.nInts;i<OBJ_ID_SZ;i++)
81       r.id[i]=0;
82   } else {
83     //Must hash array index into LBObjid
84     int j;
85     for (j=0;j<OBJ_ID_SZ;j++)
86         r.id[j]=data[j];
87     for (i=0;i<idx.nInts;i++)
88       for (j=0;j<OBJ_ID_SZ;j++)
89         r.id[j]+=circleShift(data[i],22+11*i*(j+1))+
90           circleShift(data[i],21-9*i*(j+1));
91   }
92
93 #if CMK_GLOBAL_LOCATION_UPDATE
94   r.dimension = idx.dimension;
95   r.nInts = idx.nInts; 
96   r.isArrayElement = 1; 
97 #endif
98
99   return r;
100 }
101
102 #if CMK_GLOBAL_LOCATION_UPDATE
103 void UpdateLocation(MigrateInfo& migData) {
104
105   if (migData.obj.id.isArrayElement == 0) {
106     return;
107   }
108
109   CkArrayIndex idx; 
110   idx.dimension = migData.obj.id.dimension; 
111   idx.nInts = migData.obj.id.nInts; 
112
113   for (int i = 0; i < idx.nInts; i++) {
114     idx.data()[i] = migData.obj.id.id[i];    
115   }
116
117   CkGroupID locMgrGid;
118   locMgrGid.idx = migData.obj.id.locMgrGid;
119   CkLocMgr *localLocMgr = (CkLocMgr *) CkLocalBranch(locMgrGid);
120   localLocMgr->updateLocation(idx, migData.to_pe); 
121 }
122 #endif
123
124 #endif
125
126 /*********************** Array Messages ************************/
127 CkArrayIndex &CkArrayMessage::array_index(void)
128 {
129     return UsrToEnv((void *)this)->getsetArrayIndex();
130 }
131 unsigned short &CkArrayMessage::array_ep(void)
132 {
133         return UsrToEnv((void *)this)->getsetArrayEp();
134 }
135 unsigned short &CkArrayMessage::array_ep_bcast(void)
136 {
137         return UsrToEnv((void *)this)->getsetArrayBcastEp();
138 }
139 unsigned char &CkArrayMessage::array_hops(void)
140 {
141         return UsrToEnv((void *)this)->getsetArrayHops();
142 }
143 unsigned int CkArrayMessage::array_getSrcPe(void)
144 {
145         return UsrToEnv((void *)this)->getsetArraySrcPe();
146 }
147 unsigned int CkArrayMessage::array_ifNotThere(void)
148 {
149         return UsrToEnv((void *)this)->getArrayIfNotThere();
150 }
151 void CkArrayMessage::array_setIfNotThere(unsigned int i)
152 {
153         UsrToEnv((void *)this)->setArrayIfNotThere(i);
154 }
155
156 /*********************** Array Map ******************
157 Given an array element index, an array map tells us 
158 the index's "home" Pe.  This is the Pe the element will
159 be created on, and also where messages to this element will
160 be forwarded by default.
161 */
162
163 CkArrayMap::CkArrayMap(void) { }
164 CkArrayMap::~CkArrayMap() { }
165 int CkArrayMap::registerArray(CkArrayIndex& numElements,CkArrayID aid)
166 {return 0;}
167
168 #define CKARRAYMAP_POPULATE_INITIAL(POPULATE_CONDITION) \
169         int i; \
170         for (int i1=0; i1<numElements.data()[0]; i1++) { \
171           if (numElements.dimension == 1) { \
172             /* Make 1D indices */ \
173             i = i1; \
174             CkArrayIndex1D idx(i1); \
175             if (POPULATE_CONDITION) \
176               mgr->insertInitial(idx,CkCopyMsg(&ctorMsg)); \
177           } else { \
178             /* higher dimensionality */ \
179             for (int i2=0; i2<numElements.data()[1]; i2++) { \
180               if (numElements.dimension == 2) { \
181                 /* Make 2D indices */ \
182                 i = i1 * numElements.data()[1] + i2; \
183                 CkArrayIndex2D idx(i1, i2); \
184                 if (POPULATE_CONDITION) \
185                   mgr->insertInitial(idx,CkCopyMsg(&ctorMsg)); \
186               } else { \
187                 /* higher dimensionality */ \
188                 CkAssert(numElements.dimension == 3); \
189                 for (int i3=0; i3<numElements.data()[2]; i3++) { \
190                   /* Make 3D indices */ \
191                   i = (i1 * numElements.data()[1] + i2) * numElements.data()[2] + i3; \
192                   CkArrayIndex3D idx(i1, i2, i3 ); \
193                   if (POPULATE_CONDITION) \
194                     mgr->insertInitial(idx,CkCopyMsg(&ctorMsg)); \
195                 } \
196               } \
197             } \
198           } \
199         }
200
201 void CkArrayMap::populateInitial(int arrayHdl,CkArrayIndex& numElements,void *ctorMsg,CkArrMgr *mgr)
202 {
203         if (numElements.nInts==0) {
204           CkFreeMsg(ctorMsg);
205           return;
206         }
207         int thisPe=CkMyPe();
208         /* The CkArrayIndex is supposed to have at most 3 dimensions, which
209            means that all the fields are ints, and numElements.nInts represents
210            how many of them are used */
211         CKARRAYMAP_POPULATE_INITIAL(procNum(arrayHdl,idx)==thisPe);
212
213 #if CMK_BIGSIM_CHARM
214         BgEntrySplit("split-array-new-end");
215 #endif
216
217         mgr->doneInserting();
218         CkFreeMsg(ctorMsg);
219 }
220
221 CkGroupID _defaultArrayMapID;
222 CkGroupID _fastArrayMapID;
223
224 class RRMap : public CkArrayMap
225 {
226 public:
227   RRMap(void)
228   {
229           DEBC((AA"Creating RRMap\n"AB));
230   }
231   RRMap(CkMigrateMessage *m):CkArrayMap(m){}
232   int procNum(int /*arrayHdl*/, const CkArrayIndex &i)
233   {
234 #if 1
235     if (i.nInts==1) {
236       //Map 1D integer indices in simple round-robin fashion
237       int ans= (i.data()[0])%CkNumPes();
238       while(!CmiNodeAlive(ans) || (ans == CkMyPe() && CkpvAccess(startedEvac))){
239         ans = (ans +1 )%CkNumPes();
240       }
241       return ans;
242     }
243     else 
244 #endif
245       {
246         //Map other indices based on their hash code, mod a big prime.
247         unsigned int hash=(i.hash()+739)%1280107;
248         int ans = (hash % CkNumPes());
249         while(!CmiNodeAlive(ans)){
250                 ans = (ans +1 )%CkNumPes();
251         }
252         return ans;
253
254       }
255   }
256 };
257
258 /** 
259  * Class used to store the dimensions of the array and precalculate numChares,
260  * binSize and other values for the DefaultArrayMap -- ASB
261  */
262 class arrayMapInfo {
263 public:
264   CkArrayIndex _nelems;
265   int _binSizeFloor;            /* floor of numChares/numPes */
266   int _binSizeCeil;             /* ceiling of numChares/numPes */
267   int _numChares;               /* initial total number of chares */
268   int _remChares;               /* numChares % numPes -- equals the number of
269                                    processors in the first set */
270   int _numFirstSet;             /* _remChares X (_binSize + 1) -- number of
271                                    chares in the first set */
272
273   int _nBinSizeFloor;           /* floor of numChares/numNodes */
274   int _nRemChares;              /* numChares % numNodes -- equals the number of
275                                    nodes in the first set */
276   int _nNumFirstSet;            /* _remChares X (_binSize + 1) -- number of
277                                    chares in the first set of nodes */
278
279   /** All processors are divided into two sets. Processors in the first set
280    *  have one chare more than the processors in the second set. */
281
282   arrayMapInfo(void) { }
283
284   arrayMapInfo(CkArrayIndex& n) : _nelems(n), _numChares(0) {
285     compute_binsize();
286   }
287
288   ~arrayMapInfo() {}
289   
290   void compute_binsize()
291   {
292     int numPes = CkNumPes();
293     //Now assuming homogenous nodes where each node has the same number of PEs
294     int numNodes = CkNumNodes();
295
296     if (_nelems.nInts == 1) {
297       _numChares = _nelems.data()[0];
298     } else if (_nelems.nInts == 2) {
299       _numChares = _nelems.data()[0] * _nelems.data()[1];
300     } else if (_nelems.nInts == 3) {
301       _numChares = _nelems.data()[0] * _nelems.data()[1] * _nelems.data()[2];
302     }
303
304     _remChares = _numChares % numPes;
305     _binSizeFloor = (int)floor((double)_numChares/(double)numPes);
306     _binSizeCeil = (int)ceil((double)_numChares/(double)numPes);
307     _numFirstSet = _remChares * (_binSizeFloor + 1);
308
309     _nRemChares = _numChares % numNodes;
310     _nBinSizeFloor = _numChares/numNodes;
311     _nNumFirstSet = _nRemChares * (_nBinSizeFloor +1);
312   }
313
314   void pup(PUP::er& p){
315     p|_nelems;
316     p|_binSizeFloor;
317     p|_binSizeCeil;
318     p|_numChares;
319     p|_remChares;
320     p|_numFirstSet;
321     p|_nRemChares;
322     p|_nBinSizeFloor;
323     p|_nNumFirstSet;
324   }
325 }c;
326
327
328 /**
329  * The default map object -- This does blocked mapping in the general case and
330  * calls the round-robin procNum for the dynamic insertion case -- ASB
331  */
332 class DefaultArrayMap : public RRMap
333 {
334 public:
335   /** This array stores information about different chare arrays in a Charm
336    *  program (dimensions, binsize, numChares etc ... ) */
337   CkPupPtrVec<arrayMapInfo> amaps;
338
339 public:
340   DefaultArrayMap(void) {
341     DEBC((AA"Creating DefaultArrayMap\n"AB));
342   }
343
344   DefaultArrayMap(CkMigrateMessage *m) : RRMap(m){}
345
346   int registerArray(CkArrayIndex& numElements, CkArrayID aid)
347   {
348     int idx = amaps.size();
349     amaps.resize(idx+1);
350     amaps[idx] = new arrayMapInfo(numElements);
351     return idx;
352   }
353  
354   int procNum(int arrayHdl, const CkArrayIndex &i) {
355     int flati;
356     if (amaps[arrayHdl]->_nelems.nInts == 0) {
357       return RRMap::procNum(arrayHdl, i);
358     }
359
360     if (i.nInts == 1) {
361       flati = i.data()[0];
362     } else if (i.nInts == 2) {
363       flati = i.data()[0] * amaps[arrayHdl]->_nelems.data()[1] + i.data()[1];
364     } else if (i.nInts == 3) {
365       flati = (i.data()[0] * amaps[arrayHdl]->_nelems.data()[1] + i.data()[1]) * amaps[arrayHdl]->_nelems.data()[2] + i.data()[2];
366     }
367 #if CMK_ERROR_CHECKING
368     else {
369       CkAbort("CkArrayIndex has more than 3 integers!");
370     }
371 #endif
372
373     if(useNodeBlkMapping){
374       if(flati < amaps[arrayHdl]->_numChares){
375         int numCharesOnNode = amaps[arrayHdl]->_nBinSizeFloor;
376         int startNodeID, offsetInNode;
377         if(flati < amaps[arrayHdl]->_nNumFirstSet){
378           numCharesOnNode++;
379           startNodeID = flati/numCharesOnNode;
380           offsetInNode = flati%numCharesOnNode;
381         }else{
382           startNodeID = amaps[arrayHdl]->_nRemChares+(flati-amaps[arrayHdl]->_nNumFirstSet)/numCharesOnNode;
383           offsetInNode = (flati-amaps[arrayHdl]->_nNumFirstSet)%numCharesOnNode;
384         }
385         int nodeSize = CkMyNodeSize(); //assuming every node has same number of PEs
386         int elemsPerPE = numCharesOnNode/nodeSize;
387         int remElems = numCharesOnNode%nodeSize;
388         int firstSetPEs = remElems*(elemsPerPE+1);
389         if(offsetInNode<firstSetPEs){
390           return CkNodeFirst(startNodeID)+offsetInNode/(elemsPerPE+1);
391         }else{
392           return CkNodeFirst(startNodeID)+remElems+(offsetInNode-firstSetPEs)/elemsPerPE;
393         }
394       } else
395           return (flati % CkNumPes());
396     }
397     //regular PE-based block mapping
398     if(flati < amaps[arrayHdl]->_numFirstSet)
399       return (flati / (amaps[arrayHdl]->_binSizeFloor + 1));
400     else if (flati < amaps[arrayHdl]->_numChares)
401       return (amaps[arrayHdl]->_remChares + (flati - amaps[arrayHdl]->_numFirstSet) / (amaps[arrayHdl]->_binSizeFloor));
402     else
403       return (flati % CkNumPes());
404   }
405
406   void pup(PUP::er& p){
407     RRMap::pup(p);
408     int npes = CkNumPes();
409     p|npes;
410     p|amaps;
411     if (p.isUnpacking() && npes != CkNumPes())  {   // binSize needs update
412       for (int i=0; i<amaps.size(); i++)
413         amaps[i]->compute_binsize();
414     }
415   }
416 };
417
418 /**
419  *  A fast map for chare arrays which do static insertions and promise NOT
420  *  to do late insertions -- ASB
421  */
422 class FastArrayMap : public DefaultArrayMap
423 {
424 public:
425   FastArrayMap(void) {
426     DEBC((AA"Creating FastArrayMap\n"AB));
427   }
428
429   FastArrayMap(CkMigrateMessage *m) : DefaultArrayMap(m){}
430
431   int registerArray(CkArrayIndex& numElements, CkArrayID aid)
432   {
433     int idx;
434     idx = DefaultArrayMap::registerArray(numElements, aid);
435
436     return idx;
437   }
438
439   int procNum(int arrayHdl, const CkArrayIndex &i) {
440     int flati;
441     if (amaps[arrayHdl]->_nelems.nInts == 0) {
442       return RRMap::procNum(arrayHdl, i);
443     }
444
445     if (i.nInts == 1) {
446       flati = i.data()[0];
447     } else if (i.nInts == 2) {
448       flati = i.data()[0] * amaps[arrayHdl]->_nelems.data()[1] + i.data()[1];
449     } else if (i.nInts == 3) {
450       flati = (i.data()[0] * amaps[arrayHdl]->_nelems.data()[1] + i.data()[1]) * amaps[arrayHdl]->_nelems.data()[2] + i.data()[2];
451     }
452 #if CMK_ERROR_CHECKING
453     else {
454       CkAbort("CkArrayIndex has more than 3 integers!");
455     }
456 #endif
457
458     /** binSize used in DefaultArrayMap is the floor of numChares/numPes
459      *  but for this FastArrayMap, we need the ceiling */
460     return (flati / amaps[arrayHdl]->_binSizeCeil);
461   }
462
463   void pup(PUP::er& p){
464     DefaultArrayMap::pup(p);
465   }
466 };
467
468
469 /**
470  * This map can be used for topology aware mapping when the mapping is provided
471  * through a file -- ASB
472  */
473 class ReadFileMap : public DefaultArrayMap
474 {
475 private:
476   CkVec<int> mapping;
477
478 public:
479   ReadFileMap(void) {
480     DEBC((AA"Creating ReadFileMap\n"AB));
481   }
482
483   ReadFileMap(CkMigrateMessage *m) : DefaultArrayMap(m){}
484
485   int registerArray(CkArrayIndex& numElements, CkArrayID aid)
486   {
487     int idx;
488     idx = DefaultArrayMap::registerArray(numElements, aid);
489
490     if(mapping.size() == 0) {
491       int numChares;
492
493       if (amaps[idx]->_nelems.nInts == 1) {
494         numChares = amaps[idx]->_nelems.data()[0];
495       } else if (amaps[idx]->_nelems.nInts == 2) {
496         numChares = amaps[idx]->_nelems.data()[0] * amaps[idx]->_nelems.data()[1];
497       } else if (amaps[idx]->_nelems.nInts == 3) {
498         numChares = amaps[idx]->_nelems.data()[0] * amaps[idx]->_nelems.data()[1] * amaps[idx]->_nelems.data()[2];
499       } else {
500         CkAbort("CkArrayIndex has more than 3 integers!");
501       }
502
503       mapping.resize(numChares);
504       FILE *mapf = fopen("mapfile", "r");
505       TopoManager tmgr;
506       int x, y, z, t;
507
508       for(int i=0; i<numChares; i++) {
509         (void) fscanf(mapf, "%d %d %d %d", &x, &y, &z, &t);
510         mapping[i] = tmgr.coordinatesToRank(x, y, z, t);
511       }
512       fclose(mapf);
513     }
514
515     return idx;
516   }
517
518   int procNum(int arrayHdl, const CkArrayIndex &i) {
519     int flati;
520
521     if (i.nInts == 1) {
522       flati = i.data()[0];
523     } else if (i.nInts == 2) {
524       flati = i.data()[0] * amaps[arrayHdl]->_nelems.data()[1] + i.data()[1];
525     } else if (i.nInts == 3) {
526       flati = (i.data()[0] * amaps[arrayHdl]->_nelems.data()[1] + i.data()[1]) * amaps[arrayHdl]->_nelems.data()[2] + i.data()[2];
527     } else {
528       CkAbort("CkArrayIndex has more than 3 integers!");
529     }
530
531     return mapping[flati];
532   }
533
534   void pup(PUP::er& p){
535     DefaultArrayMap::pup(p);
536     p|mapping;
537   }
538 };
539
540 class BlockMap : public RRMap
541 {
542 public:
543   BlockMap(void){
544         DEBC((AA"Creating BlockMap\n"AB));
545   }
546   BlockMap(CkMigrateMessage *m):RRMap(m){ }
547   void populateInitial(int arrayHdl,CkArrayIndex& numElements,void *ctorMsg,CkArrMgr *mgr){
548         if (numElements.nInts==0) {
549           CkFreeMsg(ctorMsg);
550           return;
551         }
552         int thisPe=CkMyPe();
553         int numPes=CkNumPes();
554         int binSize;
555         if (numElements.nInts == 1) {
556           binSize = (int)ceil((double)numElements.data()[0]/(double)numPes);
557         } else if (numElements.nInts == 2) {
558           binSize = (int)ceil((double)(numElements.data()[0]*numElements.data()[1])/(double)numPes);
559         } else if (numElements.nInts == 3) {
560           binSize = (int)ceil((double)(numElements.data()[0]*numElements.data()[1]*numElements.data()[2])/(double)numPes);
561         } else {
562           CkAbort("CkArrayIndex has more than 3 integers!");
563         }
564         CKARRAYMAP_POPULATE_INITIAL(i/binSize==thisPe);
565
566         /*
567         CkArrayIndex idx;
568         for (idx=numElements.begin(); idx<numElements; idx.getNext(numElements)) {
569           //for (int i=0;i<numElements;i++) {
570                 int binSize = (int)ceil((double)numElements.getCombinedCount()/(double)numPes);
571                 if (i/binSize==thisPe)
572                         mgr->insertInitial(idx,CkCopyMsg(&ctorMsg));
573         }*/
574         mgr->doneInserting();
575         CkFreeMsg(ctorMsg);
576   }
577 };
578
579 /**
580  * map object-- use seed load balancer.  
581  */
582 class CldMap : public CkArrayMap
583 {
584 public:
585   CldMap(void)
586   {
587           DEBC((AA"Creating CldMap\n"AB));
588   }
589   CldMap(CkMigrateMessage *m):CkArrayMap(m){}
590   int homePe(int /*arrayHdl*/, const CkArrayIndex &i)
591   {
592     if (i.nInts==1) {
593       //Map 1D integer indices in simple round-robin fashion
594       return (i.data()[0])%CkNumPes();
595     }
596     else 
597       {
598         //Map other indices based on their hash code, mod a big prime.
599         unsigned int hash=(i.hash()+739)%1280107;
600         return (hash % CkNumPes());
601       }
602   }
603   int procNum(int arrayHdl, const CkArrayIndex &i)
604   {
605      return CLD_ANYWHERE;   // -1
606   }
607   void populateInitial(int arrayHdl,CkArrayIndex& numElements,void *ctorMsg,CkArrMgr *mgr)  {
608         if (numElements.nInts==0) {
609           CkFreeMsg(ctorMsg);
610           return;
611         }
612         int thisPe=CkMyPe();
613         int numPes=CkNumPes();
614         //CkArrayIndex idx;
615
616         CKARRAYMAP_POPULATE_INITIAL(i%numPes==thisPe);
617         /*for (idx=numElements.begin(); idx<numElements; idx.getNext(numElements)) {
618           //for (int i=0;i<numElements;i++)
619                         if((idx.getRank(numElements))%numPes==thisPe)
620                                 mgr->insertInitial(CkArrayIndex1D(i),CkCopyMsg(&ctorMsg),0);
621         }*/
622         mgr->doneInserting();
623         CkFreeMsg(ctorMsg);
624   }
625
626 };
627
628
629 /// A class responsible for parsing the command line arguments for the PE
630 /// to extract the format string passed in with +ConfigurableRRMap
631 class ConfigurableRRMapLoader {
632 public:
633   
634   int *locations;
635   int objs_per_block;
636   int PE_per_block;
637
638   /// labels for states used when parsing the ConfigurableRRMap from ARGV
639   enum ConfigurableRRMapLoadStatus{
640     not_loaded,
641     loaded_found,
642     loaded_not_found
643   };
644   
645   enum ConfigurableRRMapLoadStatus state;
646   
647   ConfigurableRRMapLoader(){
648     state = not_loaded;
649     locations = NULL;
650     objs_per_block = 0;
651     PE_per_block = 0;
652   }
653   
654   /// load configuration if possible, and return whether a valid configuration exists
655   bool haveConfiguration() {
656     if(state == not_loaded) {
657       DEBUG(("[%d] loading ConfigurableRRMap configuration\n", CkMyPe()));
658       char **argv=CkGetArgv();
659       char *configuration = NULL;
660       bool found = CmiGetArgString(argv, "+ConfigurableRRMap", &configuration);
661       if(!found){
662         DEBUG(("Couldn't find +ConfigurableRRMap command line argument\n"));
663         state = loaded_not_found;
664         return false;
665       } else {
666
667         DEBUG(("Found +ConfigurableRRMap command line argument in %p=\"%s\"\n", configuration, configuration));
668
669         std::istringstream instream(configuration);
670         CkAssert(instream.good());
671          
672         // Example line:
673         // 10 8 0 1 2 3 4 5 6 7 7 7 7
674         // Map 10 objects to 8 PEs, with each object's index among the 8 PEs.
675         
676         // extract first integer
677         instream >> objs_per_block >> PE_per_block;
678         CkAssert(instream.good());
679         CkAssert(objs_per_block > 0);
680         CkAssert(PE_per_block > 0);
681         locations = new int[objs_per_block];
682         for(int i=0;i<objs_per_block;i++){
683           locations[i] = 0;
684           CkAssert(instream.good());
685           instream >> locations[i];
686           CkAssert(locations[i] < PE_per_block);
687         }
688         state = loaded_found;
689         return true;
690       }
691
692     } else {
693       DEBUG(("[%d] ConfigurableRRMap has already been loaded\n", CkMyPe()));
694       return state == loaded_found;
695     }      
696      
697   }
698   
699 };
700
701 CkpvDeclare(ConfigurableRRMapLoader, myConfigRRMapState);
702
703 void _initConfigurableRRMap(){
704   CkpvInitialize(ConfigurableRRMapLoader, myConfigRRMapState);
705 }
706
707
708 /// Try to load the command line arguments for ConfigurableRRMap
709 bool haveConfigurableRRMap(){
710   DEBUG(("haveConfigurableRRMap()\n"));
711   ConfigurableRRMapLoader &loader =  CkpvAccess(myConfigRRMapState);
712   return loader.haveConfiguration();
713 }
714
715 class ConfigurableRRMap : public RRMap
716 {
717 public:
718   ConfigurableRRMap(void){
719         DEBC((AA"Creating ConfigurableRRMap\n"AB));
720   }
721   ConfigurableRRMap(CkMigrateMessage *m):RRMap(m){ }
722
723
724   void populateInitial(int arrayHdl,CkArrayIndex& numElements,void *ctorMsg,CkArrMgr *mgr){
725     // Try to load the configuration from command line argument
726     CkAssert(haveConfigurableRRMap());
727     ConfigurableRRMapLoader &loader =  CkpvAccess(myConfigRRMapState);
728     if (numElements.nInts==0) {
729       CkFreeMsg(ctorMsg);
730       return;
731     }
732     int thisPe=CkMyPe();
733     int maxIndex = numElements.data()[0];
734     DEBUG(("[%d] ConfigurableRRMap: index=%d,%d,%d\n", CkMyPe(),(int)numElements.data()[0], (int)numElements.data()[1], (int)numElements.data()[2]));
735
736     if (numElements.nInts != 1) {
737       CkAbort("ConfigurableRRMap only supports dimension 1!");
738     }
739         
740     for (int index=0; index<maxIndex; index++) {        
741       CkArrayIndex1D idx(index);                
742       
743       int cyclic_block = index / loader.objs_per_block;
744       int cyclic_local = index % loader.objs_per_block;
745       int l = loader.locations[ cyclic_local ];
746       int PE = (cyclic_block*loader.PE_per_block + l) % CkNumPes();
747
748       DEBUG(("[%d] ConfigurableRRMap: index=%d is located on PE %d l=%d\n", CkMyPe(), (int)index, (int)PE, l));
749
750       if(PE == thisPe)
751         mgr->insertInitial(idx,CkCopyMsg(&ctorMsg));
752
753     }
754     //        CKARRAYMAP_POPULATE_INITIAL(PE == thisPe);
755         
756     mgr->doneInserting();
757     CkFreeMsg(ctorMsg);
758   }
759 };
760
761
762 CkpvStaticDeclare(double*, rem);
763
764 class arrInfo {
765  private:
766    CkArrayIndex _nelems;
767    int *_map;
768    void distrib(int *speeds);
769  public:
770    arrInfo(void):_map(NULL){}
771    arrInfo(CkArrayIndex& n, int *speeds)
772    {
773      _nelems = n;
774      _map = new int[_nelems.getCombinedCount()];
775      distrib(speeds);
776    }
777    ~arrInfo() { delete[] _map; }
778    int getMap(const CkArrayIndex &i);
779    void pup(PUP::er& p){
780      p|_nelems;
781      int totalElements = _nelems.getCombinedCount();
782      if(p.isUnpacking()){
783        _map = new int[totalElements];
784      }
785      p(_map,totalElements);
786    }
787 };
788
789 static int cmp(const void *first, const void *second)
790 {
791   int fi = *((const int *)first);
792   int si = *((const int *)second);
793   return ((CkpvAccess(rem)[fi]==CkpvAccess(rem)[si]) ?
794           0 :
795           ((CkpvAccess(rem)[fi]<CkpvAccess(rem)[si]) ?
796           1 : (-1)));
797 }
798
799 void
800 arrInfo::distrib(int *speeds)
801 {
802   int _nelemsCount = _nelems.getCombinedCount();
803   double total = 0.0;
804   int npes = CkNumPes();
805   int i,j,k;
806   for(i=0;i<npes;i++)
807     total += (double) speeds[i];
808   double *nspeeds = new double[npes];
809   for(i=0;i<npes;i++)
810     nspeeds[i] = (double) speeds[i] / total;
811   int *cp = new int[npes];
812   for(i=0;i<npes;i++)
813     cp[i] = (int) (nspeeds[i]*_nelemsCount);
814   int nr = 0;
815   for(i=0;i<npes;i++)
816     nr += cp[i];
817   nr = _nelemsCount - nr;
818   if(nr != 0)
819   {
820     CkpvAccess(rem) = new double[npes];
821     for(i=0;i<npes;i++)
822       CkpvAccess(rem)[i] = (double)_nelemsCount*nspeeds[i] - cp[i];
823     int *pes = new int[npes];
824     for(i=0;i<npes;i++)
825       pes[i] = i;
826     qsort(pes, npes, sizeof(int), cmp);
827     for(i=0;i<nr;i++)
828       cp[pes[i]]++;
829     delete[] pes;
830     delete[] CkpvAccess(rem);
831   }
832   k = 0;
833   for(i=0;i<npes;i++)
834   {
835     for(j=0;j<cp[i];j++)
836       _map[k++] = i;
837   }
838   delete[] cp;
839   delete[] nspeeds;
840 }
841
842 int
843 arrInfo::getMap(const CkArrayIndex &i)
844 {
845   if(i.nInts==1)
846     return _map[i.data()[0]];
847   else
848     return _map[((i.hash()+739)%1280107)%_nelems.getCombinedCount()];
849 }
850
851 //Speeds maps processor number to "speed" (some sort of iterations per second counter)
852 // It is initialized by processor 0.
853 static int* speeds;
854
855 #if CMK_USE_PROP_MAP
856 typedef struct _speedmsg
857 {
858   char hdr[CmiMsgHeaderSizeBytes];
859   int node;
860   int speed;
861 } speedMsg;
862
863 static void _speedHdlr(void *m)
864 {
865   speedMsg *msg=(speedMsg *)m;
866   if (CmiMyRank()==0)
867     for (int pe=0;pe<CmiNodeSize(msg->node);pe++)
868       speeds[CmiNodeFirst(msg->node)+pe] = msg->speed;  
869   CmiFree(m);
870 }
871
872 // initnode call
873 void _propMapInit(void)
874 {
875   speeds = new int[CkNumPes()];
876   int hdlr = CkRegisterHandler((CmiHandler)_speedHdlr);
877   CmiPrintf("[%d]Measuring processor speed for prop. mapping...\n", CkMyPe());
878   int s = LDProcessorSpeed();
879   speedMsg msg;
880   CmiSetHandler(&msg, hdlr);
881   msg.node = CkMyNode();
882   msg.speed = s;
883   CmiSyncBroadcastAllAndFree(sizeof(msg), &msg);
884   for(int i=0;i<CkNumNodes();i++)
885     CmiDeliverSpecificMsg(hdlr);
886 }
887 #else
888 void _propMapInit(void)
889 {
890   speeds = new int[CkNumPes()];
891   int i;
892   for(i=0;i<CkNumPes();i++)
893     speeds[i] = 1;
894 }
895 #endif
896 /**
897  * A proportional map object-- tries to map more objects to
898  * faster processors and fewer to slower processors.  Also
899  * attempts to ensure good locality by mapping nearby elements
900  * together.
901  */
902 class PropMap : public CkArrayMap
903 {
904 private:
905   CkPupPtrVec<arrInfo> arrs;
906 public:
907   PropMap(void)
908   {
909     CkpvInitialize(double*, rem);
910     DEBC((AA"Creating PropMap\n"AB));
911   }
912   PropMap(CkMigrateMessage *m) {}
913   int registerArray(CkArrayIndex& numElements,CkArrayID aid)
914   {
915     int idx = arrs.size();
916     arrs.resize(idx+1);
917     arrs[idx] = new arrInfo(numElements, speeds);
918     return idx;
919   }
920   int procNum(int arrayHdl, const CkArrayIndex &i)
921   {
922     return arrs[arrayHdl]->getMap(i);
923   }
924   void pup(PUP::er& p){
925     p|arrs;
926   }
927 };
928
929 class CkMapsInit : public Chare
930 {
931 public:
932         CkMapsInit(CkArgMsg *msg) {
933                 _defaultArrayMapID = CProxy_DefaultArrayMap::ckNew();
934                 _fastArrayMapID = CProxy_FastArrayMap::ckNew();
935                 delete msg;
936         }
937
938         CkMapsInit(CkMigrateMessage *m) {}
939 };
940
941 // given an envelope of a Charm msg, find the recipient object pointer
942 CkMigratable * CkArrayMessageObjectPtr(envelope *env) {
943   if (env->getMsgtype()!=ForArrayEltMsg) return NULL;   // not an array msg
944
945   CkArrayID aid = env->getsetArrayMgr();
946   CkArray *mgr=(CkArray *)_localBranch(aid);
947   if (mgr) {
948     CkLocMgr *locMgr = mgr->getLocMgr();
949     if (locMgr) {
950       return locMgr->lookup(env->getsetArrayIndex(),aid);
951     }
952   }
953   return NULL;
954 }
955
956 /****************************** Out-of-Core support ********************/
957
958 #if CMK_OUT_OF_CORE
959 CooPrefetchManager CkArrayElementPrefetcher;
960 CkpvDeclare(int,CkSaveRestorePrefetch);
961
962 /**
963  * Return the out-of-core objid (from CooRegisterObject)
964  * that this Converse message will access.  If the message
965  * will not access an object, return -1.
966  */
967 int CkArrayPrefetch_msg2ObjId(void *msg) {
968   envelope *env=(envelope *)msg;
969   CkMigratable *elt = CkArrayMessageObjectPtr(env);
970   return elt?elt->prefetchObjID:-1;
971 }
972
973 /**
974  * Write this object (registered with RegisterObject)
975  * to this writable file.
976  */
977 void CkArrayPrefetch_writeToSwap(FILE *swapfile,void *objptr) {
978   CkMigratable *elt=(CkMigratable *)objptr;
979
980   //Save the element's data to disk:
981   PUP::toDisk p(swapfile);
982   elt->pup(p);
983
984   //Call the element's destructor in-place (so pointer doesn't change)
985   CkpvAccess(CkSaveRestorePrefetch)=1;
986   elt->~CkMigratable(); //< because destuctor is virtual, destroys user class too.
987   CkpvAccess(CkSaveRestorePrefetch)=0;
988 }
989         
990 /**
991  * Read this object (registered with RegisterObject)
992  * from this readable file.
993  */
994 void CkArrayPrefetch_readFromSwap(FILE *swapfile,void *objptr) {
995   CkMigratable *elt=(CkMigratable *)objptr;
996   //Call the element's migration constructor in-place
997   CkpvAccess(CkSaveRestorePrefetch)=1;
998   int ctorIdx=_chareTable[elt->thisChareType]->migCtor;
999   elt->myRec->invokeEntry(elt,(CkMigrateMessage *)0,ctorIdx,CmiTrue);
1000   CkpvAccess(CkSaveRestorePrefetch)=0;
1001   
1002   //Restore the element's data from disk:
1003   PUP::fromDisk p(swapfile);
1004   elt->pup(p);
1005 }
1006
1007 static void _CkMigratable_prefetchInit(void) 
1008 {
1009   CkpvExtern(int,CkSaveRestorePrefetch);
1010   CkpvAccess(CkSaveRestorePrefetch)=0;
1011   CkArrayElementPrefetcher.msg2ObjId=CkArrayPrefetch_msg2ObjId;
1012   CkArrayElementPrefetcher.writeToSwap=CkArrayPrefetch_writeToSwap;
1013   CkArrayElementPrefetcher.readFromSwap=CkArrayPrefetch_readFromSwap;
1014   CooRegisterManager(&CkArrayElementPrefetcher, _charmHandlerIdx);
1015 }
1016 #endif
1017
1018 /****************************** CkMigratable ***************************/
1019 /**
1020  * This tiny class is used to convey information to the 
1021  * newly created CkMigratable object when its constructor is called.
1022  */
1023 class CkMigratable_initInfo {
1024 public:
1025         CkLocRec_local *locRec;
1026         int chareType;
1027         CmiBool forPrefetch; /* If true, this creation is only a prefetch restore-from-disk.*/
1028 };
1029
1030 CkpvStaticDeclare(CkMigratable_initInfo,mig_initInfo);
1031
1032
1033 void _CkMigratable_initInfoInit(void) {
1034   CkpvInitialize(CkMigratable_initInfo,mig_initInfo);
1035 #if CMK_OUT_OF_CORE
1036   _CkMigratable_prefetchInit();
1037 #endif
1038 }
1039
1040 void CkMigratable::commonInit(void) {
1041         CkMigratable_initInfo &i=CkpvAccess(mig_initInfo);
1042 #if CMK_OUT_OF_CORE
1043         isInCore=CmiTrue;
1044         if (CkpvAccess(CkSaveRestorePrefetch))
1045                 return; /* Just restoring from disk--don't touch object */
1046         prefetchObjID=-1; //Unregistered
1047 #endif
1048         myRec=i.locRec;
1049         thisIndexMax=myRec->getIndex();
1050         thisChareType=i.chareType;
1051         usesAtSync=CmiFalse;
1052         usesAutoMeasure=CmiTrue;
1053         barrierRegistered=CmiFalse;
1054
1055   local_state = OFF;
1056   prev_load = 0.0;
1057   can_reset = false;
1058
1059 #if CMK_LBDB_ON
1060   if (_lb_args.metaLbOn()) {
1061     atsync_iteration = myRec->getMetaBalancer()->get_iteration();
1062   }
1063 #endif
1064
1065         /*
1066         FAULT_EVAC
1067         */
1068         AsyncEvacuate(CmiTrue);
1069 }
1070
1071 CkMigratable::CkMigratable(void) {
1072         DEBC((AA"In CkMigratable constructor\n"AB));
1073         commonInit();
1074 }
1075 CkMigratable::CkMigratable(CkMigrateMessage *m): Chare(m) {
1076         commonInit();
1077 }
1078
1079 int CkMigratable::ckGetChareType(void) const {return thisChareType;}
1080
1081 void CkMigratable::pup(PUP::er &p) {
1082         DEBM((AA"In CkMigratable::pup %s\n"AB,idx2str(thisIndexMax)));
1083         Chare::pup(p);
1084         p|thisIndexMax;
1085         p(usesAtSync);
1086   p(can_reset);
1087         p(usesAutoMeasure);
1088 #if CMK_LBDB_ON 
1089         int readyMigrate;
1090         if (p.isPacking()) readyMigrate = myRec->isReadyMigrate();
1091         p|readyMigrate;
1092         if (p.isUnpacking()) myRec->ReadyMigrate(readyMigrate);
1093 #endif
1094         if(p.isUnpacking()) barrierRegistered=CmiFalse;
1095         /*
1096                 FAULT_EVAC
1097         */
1098         p | asyncEvacuate;
1099         if(p.isUnpacking()){myRec->AsyncEvacuate(asyncEvacuate);}
1100         
1101         p | chkp_cb;
1102         p | atsync_chkp_iter;   
1103         if(p.isUnpacking()){
1104                 resetForChkp();
1105         }
1106
1107         ckFinishConstruction();
1108 }
1109
1110 void CkMigratable::ckDestroy(void) {
1111         DEBC((AA"In CkMigratable::ckDestroy %s\n"AB,idx2str(thisIndexMax)));
1112         myRec->destroy();
1113 }
1114
1115 void CkMigratable::ckAboutToMigrate(void) { }
1116 void CkMigratable::ckJustMigrated(void) { }
1117 void CkMigratable::ckJustRestored(void) { }
1118
1119 CkMigratable::~CkMigratable() {
1120         DEBC((AA"In CkMigratable::~CkMigratable %s\n"AB,idx2str(thisIndexMax)));
1121 #if CMK_OUT_OF_CORE
1122         isInCore=CmiFalse;
1123         if (CkpvAccess(CkSaveRestorePrefetch)) 
1124                 return; /* Just saving to disk--don't deregister anything. */
1125         /* We're really leaving or dying-- unregister from the ooc system*/
1126         if (prefetchObjID!=-1) {
1127                 CooDeregisterObject(prefetchObjID);
1128                 prefetchObjID=-1;
1129         }
1130 #endif
1131         /*Might want to tell myRec about our doom here--
1132         it's difficult to avoid some kind of circular-delete, though.
1133         */
1134 #if CMK_LBDB_ON 
1135         if (barrierRegistered) {
1136           DEBL((AA"Removing barrier for element %s\n"AB,idx2str(thisIndexMax)));
1137           //CkPrintf("Removing barrier for element %s\n",idx2str(thisIndexMax));
1138           if (usesAtSync)
1139                 myRec->getLBDB()->RemoveLocalBarrierClient(ldBarrierHandle);
1140           else
1141                 myRec->getLBDB()->RemoveLocalBarrierReceiver(ldBarrierRecvHandle);
1142         }
1143 #endif
1144         //To detect use-after-delete
1145         thisIndexMax.nInts=-12345;
1146         thisIndexMax.dimension=-12345;
1147 }
1148
1149 void CkMigratable::CkAbort(const char *why) const {
1150         CkError("CkMigratable '%s' aborting:\n",_chareTable[thisChareType]->name);
1151         ::CkAbort(why);
1152 }
1153
1154 void CkMigratable::ResumeFromSync(void)
1155 {
1156 //      CkAbort("::ResumeFromSync() not defined for this array element!\n");
1157 }
1158
1159 void CkMigratable::UserSetLBLoad() {
1160         CkAbort("::UserSetLBLoad() not defined for this array element!\n");
1161 }
1162
1163 #if CMK_LBDB_ON  //For load balancing:
1164 // user can call this helper function to set obj load (for model-based lb)
1165 void CkMigratable::setObjTime(double cputime) {
1166         myRec->setObjTime(cputime);
1167 }
1168 double CkMigratable::getObjTime() {
1169         return myRec->getObjTime();
1170 }
1171
1172 void CkMigratable::clearMetaLBData() {
1173 //  if (can_reset) {
1174     local_state = OFF;
1175     atsync_iteration = -1;
1176     prev_load = 0.0;
1177     can_reset = false;
1178 //  }
1179 }
1180
1181 void CkMigratable::recvLBPeriod(void *data) {
1182   if (atsync_iteration < 0) {
1183     return;
1184   }
1185   int lb_period = *((int *) data);
1186  DEBAD(("\t[obj %s] Received the LB Period %d current iter %d state %d on PE %d\n",
1187      idx2str(thisIndexMax), lb_period, atsync_iteration, local_state, CkMyPe()));
1188
1189   bool is_tentative;
1190   if (local_state == LOAD_BALANCE) {
1191     CkAssert(lb_period == myRec->getMetaBalancer()->getPredictedLBPeriod(is_tentative));
1192     return;
1193   }
1194
1195   if (local_state == PAUSE) {
1196     if (atsync_iteration < lb_period) {
1197       local_state = DECIDED;
1198       ResumeFromSync();
1199       return;
1200     }
1201     local_state = LOAD_BALANCE;
1202
1203     can_reset = true;
1204     //myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
1205     return;
1206   }
1207   local_state = DECIDED;
1208 }
1209
1210 void CkMigratable::metaLBCallLB() {
1211         myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
1212 }
1213
1214 void CkMigratable::ckFinishConstruction(void)
1215 {
1216 //      if ((!usesAtSync) || barrierRegistered) return;
1217         myRec->setMeasure(usesAutoMeasure);
1218         if (barrierRegistered) return;
1219         DEBL((AA"Registering barrier client for %s\n"AB,idx2str(thisIndexMax)));
1220         //CkPrintf("Registering barrier client for %s\n",idx2str(thisIndexMax));
1221         if (usesAtSync)
1222           ldBarrierHandle = myRec->getLBDB()->AddLocalBarrierClient(
1223                 (LDBarrierFn)staticResumeFromSync,(void*)(this));
1224         else
1225           ldBarrierRecvHandle = myRec->getLBDB()->AddLocalBarrierReceiver(
1226                 (LDBarrierFn)staticResumeFromSync,(void*)(this));
1227         barrierRegistered=CmiTrue;
1228 }
1229
1230
1231 void CkMigratable::setChkpResumeClient(CkCallback & _cb)
1232 {
1233         chkp_cb = _cb;
1234         nextChkpIter = -1;
1235         nextChkpDecided = false;
1236         atsync_chkp_iter = -1;
1237         local_chkp_pause = false;
1238 }
1239
1240 void CkMigratable::AtChkpSync()
1241 {
1242         if(CmiNumPartition()==1){
1243                 chkp_cb.send();
1244                 return;
1245         }
1246         atsync_chkp_iter++;
1247         myRec->getChkpMgr()->recvIter(atsync_chkp_iter);
1248         //reduction to decide the current maximum
1249         if(nextChkpDecided){
1250                 if(atsync_chkp_iter<nextChkpIter){
1251                         chkp_cb.send();
1252                 }else if(atsync_chkp_iter == nextChkpIter){
1253                         resetForChkp();
1254                         myRec->getChkpMgr()->reachChkpIter();
1255                 }else{
1256                         CkAbort("Impossible state\n");
1257                 }
1258         }
1259         else if(myRec->getChkpMgr()->localDecided){
1260                 int localIter = myRec->getChkpMgr()->localMaxIter;
1261                 if(atsync_chkp_iter==localIter){
1262                         local_chkp_pause = true;
1263                 }
1264                 else if(atsync_chkp_iter<localIter){
1265                         chkp_cb.send();
1266                 }
1267                 else{
1268                         CkAbort("local Impossible state\n");
1269                 }
1270         }       
1271         else{
1272                 chkp_cb.send();
1273         }
1274 }
1275
1276 void CkMigratable::resetForChkp(){
1277         nextChkpDecided = false;
1278         local_chkp_pause = false;
1279 }
1280
1281 void CkMigratable::ResumeFromChkp(){
1282         chkp_cb.send();
1283 }
1284
1285 void CkMigratable::recvChkpIter(void * _iter){
1286         int iter = *(int *)_iter;
1287         nextChkpIter = iter;
1288         nextChkpDecided = true;
1289         if(atsync_chkp_iter>nextChkpIter){
1290                 CkAbort("impossible state in notify\n");
1291         }
1292         else if(atsync_chkp_iter==nextChkpIter){
1293                 resetForChkp();
1294                 myRec->getChkpMgr()->reachChkpIter();
1295         }
1296         else{
1297                 if(local_chkp_pause)
1298                         chkp_cb.send(); 
1299         }
1300 }
1301
1302 void CkMigratable::AtSync(int waitForMigration)
1303 {
1304         if (!usesAtSync)
1305                 CkAbort("You must set usesAtSync=CmiTrue in your array element constructor to use AtSync!\n");
1306 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
1307         mlogData->toResumeOrNot=1;
1308 #endif
1309         myRec->AsyncMigrate(!waitForMigration);
1310         if (waitForMigration) ReadyMigrate(CmiTrue);
1311         ckFinishConstruction();
1312   //DEBL((AA"Element %s going to sync\n"AB,idx2str(thisIndexMax)));
1313   //if(CmiMyPartition()==1)
1314 //      CkPrintf("Element %s going to sync\n",idx2str(thisIndexMax));
1315   // model-based load balancing, ask user to provide cpu load
1316   if (usesAutoMeasure == CmiFalse) UserSetLBLoad();
1317
1318   if (!_lb_args.metaLbOn()) {
1319     myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
1320     return;
1321   }
1322
1323   // When MetaBalancer is turned on
1324
1325   if (atsync_iteration == -1) {
1326     can_reset = false;
1327     local_state = OFF;
1328     prev_load = 0.0;
1329   }
1330
1331   atsync_iteration++;
1332   //CkPrintf("[pe %s] atsync_iter %d && predicted period %d state: %d\n",
1333   //    idx2str(thisIndexMax), atsync_iteration,
1334   //    myRec->getMetaBalancer()->getPredictedLBPeriod(), local_state);
1335   double tmp = prev_load;
1336   prev_load = myRec->getObjTime();
1337   double current_load = prev_load - tmp;
1338   if (atsync_iteration <= myRec->getMetaBalancer()->get_finished_iteration()) {
1339     CkPrintf("[%d:%s] Error!! Contributing to iter %d < current iter %d\n",
1340       CkMyPe(), idx2str(thisIndexMax), atsync_iteration,
1341       myRec->getMetaBalancer()->get_finished_iteration());
1342     CkAbort("Not contributing to the right iteration\n");
1343   }
1344
1345   if (atsync_iteration != 0) {
1346     myRec->getMetaBalancer()->AddLoad(atsync_iteration, current_load);
1347   }
1348
1349   bool is_tentative;
1350   if (atsync_iteration < myRec->getMetaBalancer()->getPredictedLBPeriod(is_tentative)) {
1351     ResumeFromSync();
1352   } else if (is_tentative) {
1353     local_state = PAUSE;
1354   } else if (local_state == DECIDED) {
1355     DEBAD(("[%d:%s] Went to load balance iter %d\n", CkMyPe(), idx2str(thisIndexMax), atsync_iteration));
1356     local_state = LOAD_BALANCE;
1357     can_reset = true;
1358     //myRec->getLBDB()->AtLocalBarrier(ldBarrierHandle);
1359   } else {
1360     DEBAD(("[%d:%s] Went to pause state iter %d\n", CkMyPe(), idx2str(thisIndexMax), atsync_iteration));
1361     local_state = PAUSE;
1362   }
1363 }
1364
1365 void CkMigratable::ReadyMigrate(CmiBool ready)
1366 {
1367         myRec->ReadyMigrate(ready);
1368 }
1369
1370 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
1371     extern int globalResumeCount;
1372 #endif
1373
1374 void CkMigratable::staticResumeFromSync(void* data)
1375 {
1376         CkMigratable *el=(CkMigratable *)data;
1377 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
1378     if(el->mlogData->toResumeOrNot ==0 || el->mlogData->resumeCount >= globalResumeCount){
1379         return;
1380     }
1381 #endif
1382         DEBL((AA"Element %s resuming from sync\n"AB,idx2str(el->thisIndexMax)));
1383 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
1384     CpvAccess(_currentObj) = el;
1385 #endif
1386
1387   if (_lb_args.metaLbOn()) {
1388         el->clearMetaLBData();
1389         }
1390         el->ResumeFromSync();
1391 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
1392     el->mlogData->resumeCount++;
1393 #endif
1394 }
1395 void CkMigratable::setMigratable(int migratable) 
1396 {
1397         myRec->setMigratable(migratable);
1398 }
1399
1400 struct CkArrayThreadListener {
1401         struct CthThreadListener base;
1402         CkMigratable *mig;
1403 };
1404
1405 extern "C"
1406 void CkArrayThreadListener_suspend(struct CthThreadListener *l)
1407 {
1408         CkArrayThreadListener *a=(CkArrayThreadListener *)l;
1409         a->mig->ckStopTiming();
1410 }
1411
1412 extern "C"
1413 void CkArrayThreadListener_resume(struct CthThreadListener *l)
1414 {
1415         CkArrayThreadListener *a=(CkArrayThreadListener *)l;
1416         a->mig->ckStartTiming();
1417 }
1418
1419 extern "C"
1420 void CkArrayThreadListener_free(struct CthThreadListener *l)
1421 {
1422         CkArrayThreadListener *a=(CkArrayThreadListener *)l;
1423         delete a;
1424 }
1425
1426 void CkMigratable::CkAddThreadListeners(CthThread tid, void *msg)
1427 {
1428         Chare::CkAddThreadListeners(tid, msg);   // for trace
1429         CthSetThreadID(tid, thisIndexMax.data()[0], thisIndexMax.data()[1], 
1430                        thisIndexMax.data()[2]);
1431         CkArrayThreadListener *a=new CkArrayThreadListener;
1432         a->base.suspend=CkArrayThreadListener_suspend;
1433         a->base.resume=CkArrayThreadListener_resume;
1434         a->base.free=CkArrayThreadListener_free;
1435         a->mig=this;
1436         CthAddListener(tid,(struct CthThreadListener *)a);
1437 }
1438 #else
1439 void CkMigratable::setObjTime(double cputime) {}
1440 double CkMigratable::getObjTime() {return 0.0;}
1441
1442 /* no load balancer: need dummy implementations to prevent link error */
1443 void CkMigratable::CkAddThreadListeners(CthThread tid, void *msg)
1444 {
1445 }
1446 #endif
1447
1448
1449 /*CkMigratableList*/
1450 CkMigratableList::CkMigratableList() {}
1451 CkMigratableList::~CkMigratableList() {}
1452
1453 void CkMigratableList::setSize(int s) {
1454         el.resize(s);
1455 }
1456
1457 void CkMigratableList::put(CkMigratable *v,int atIdx) {
1458 #if CMK_ERROR_CHECKING
1459         if (atIdx>=length())
1460                 CkAbort("Internal array manager error (CkMigrableList::put index out of bounds)");
1461 #endif
1462         el[atIdx]=v;
1463 }
1464
1465
1466 /************************** Location Records: *********************************/
1467
1468 //---------------- Base type:
1469 void CkLocRec::weAreObsolete(const CkArrayIndex &idx) {}
1470 CkLocRec::~CkLocRec() { }
1471 void CkLocRec::beenReplaced(void)
1472     {/*Default: ignore replacement*/}
1473
1474 //Return the represented array element; or NULL if there is none
1475 CkMigratable *CkLocRec::lookupElement(CkArrayID aid) {return NULL;}
1476
1477 //Return the last known processor; or -1 if none
1478 int CkLocRec::lookupProcessor(void) {return -1;}
1479
1480
1481 /*----------------- Local:
1482 Matches up the array index with the local index, an
1483 interfaces with the load balancer on behalf of the
1484 represented array elements.
1485 */
1486 CkLocRec_local::CkLocRec_local(CkLocMgr *mgr,CmiBool fromMigration,
1487   CmiBool ignoreArrival, const CkArrayIndex &idx_,int localIdx_)
1488         :CkLocRec(mgr),idx(idx_),localIdx(localIdx_),
1489          running(CmiFalse),deletedMarker(NULL)
1490 {
1491 #if CMK_LBDB_ON
1492         DEBL((AA"Registering element %s with load balancer\n"AB,idx2str(idx)));
1493         //BIGSIM_OOC DEBUGGING
1494         //CkPrintf("LocMgr on %d: Registering element %s with load balancer\n", CkMyPe(), idx2str(idx));
1495         nextPe = -1;
1496         asyncMigrate = CmiFalse;
1497         readyMigrate = CmiTrue;
1498         enable_measure = CmiTrue;
1499         bounced  = CmiFalse;
1500         the_lbdb=mgr->getLBDB();
1501         the_metalb=mgr->getMetaBalancer();
1502         the_chkp = mgr->getChkpMgr();
1503         LDObjid ldid = idx2LDObjid(idx);
1504 #if CMK_GLOBAL_LOCATION_UPDATE
1505         ldid.locMgrGid = mgr->getGroupID().idx;
1506 #endif        
1507         ldHandle=the_lbdb->RegisterObj(mgr->getOMHandle(),
1508                 ldid,(void *)this,1);
1509         if (fromMigration) {
1510                 DEBL((AA"Element %s migrated in\n"AB,idx2str(idx)));
1511                 if (!ignoreArrival)  {
1512                         the_lbdb->Migrated(ldHandle, CmiTrue);
1513                   // load balancer should ignore this objects movement
1514                 //  AsyncMigrate(CmiTrue);
1515                 }
1516         }
1517 #endif
1518         the_chkp = mgr->getChkpMgr();
1519         /*
1520                 FAULT_EVAC
1521         */
1522         asyncEvacuate = CmiTrue;
1523 }
1524 CkLocRec_local::~CkLocRec_local()
1525 {
1526         if (deletedMarker!=NULL) *deletedMarker=CmiTrue;
1527         myLocMgr->reclaim(idx,localIdx);
1528 #if CMK_LBDB_ON
1529         stopTiming();
1530         DEBL((AA"Unregistering element %s from load balancer\n"AB,idx2str(idx)));
1531         the_lbdb->UnregisterObj(ldHandle);
1532 #endif
1533 }
1534 void CkLocRec_local::migrateMe(int toPe) //Leaving this processor
1535 {
1536         //This will pack us up, send us off, and delete us
1537 //      printf("[%d] migrating migrateMe to %d \n",CkMyPe(),toPe);
1538         myLocMgr->emigrate(this,toPe);
1539 }
1540
1541 #if CMK_LBDB_ON
1542 void CkLocRec_local::informIdealLBPeriod(int lb_ideal_period) {
1543   myLocMgr->informLBPeriod(this, lb_ideal_period);
1544 }
1545
1546 void CkLocRec_local::metaLBCallLB() {
1547         myLocMgr->metaLBCallLB(this);
1548 }
1549
1550 void CkLocRec_local::startTiming(int ignore_running) {
1551         if (!ignore_running) running=CmiTrue;
1552         DEBL((AA"Start timing for %s at %.3fs {\n"AB,idx2str(idx),CkWallTimer()));
1553         if (enable_measure) the_lbdb->ObjectStart(ldHandle);
1554 }
1555 void CkLocRec_local::stopTiming(int ignore_running) {
1556         DEBL((AA"} Stop timing for %s at %.3fs\n"AB,idx2str(idx),CkWallTimer()));
1557         if ((ignore_running || running) && enable_measure) the_lbdb->ObjectStop(ldHandle);
1558         if (!ignore_running) running=CmiFalse;
1559 }
1560 void CkLocRec_local::setObjTime(double cputime) {
1561         the_lbdb->EstObjLoad(ldHandle, cputime);
1562 }
1563 double CkLocRec_local::getObjTime() {
1564         LBRealType walltime, cputime;
1565         the_lbdb->GetObjLoad(ldHandle, walltime, cputime);
1566         return walltime;
1567 }
1568 #endif
1569
1570 void CkLocRec_local::destroy(void) //User called destructor
1571 {
1572         //Our destructor does all the needed work
1573         delete this;
1574 }
1575 //Return the represented array element; or NULL if there is none
1576 CkMigratable *CkLocRec_local::lookupElement(CkArrayID aid) {
1577         return myLocMgr->lookupLocal(localIdx,aid);
1578 }
1579
1580 //Return the last known processor; or -1 if none
1581 int CkLocRec_local::lookupProcessor(void) {
1582         return CkMyPe();
1583 }
1584
1585 CkLocRec::RecType CkLocRec_local::type(void)
1586 {
1587         return local;
1588 }
1589
1590 void CkLocRec_local::addedElement(void) 
1591 {
1592         //Push everything in the half-created queue into the system--
1593         // anything not ready yet will be put back in.
1594         while (!halfCreated.isEmpty()) 
1595                 CkArrayManagerDeliver(CkMyPe(),halfCreated.deq());
1596 }
1597
1598 CmiBool CkLocRec_local::isObsolete(int nSprings,const CkArrayIndex &idx_)
1599
1600         int len=halfCreated.length();
1601         if (len!=0) {
1602                 /* This is suspicious-- the halfCreated queue should be extremely
1603                  transient.  It's possible we just looked at the wrong time, though;
1604                  so this is only a warning. 
1605                 */
1606                 CkPrintf("CkLoc WARNING> %d messages still around for uncreated element %s!\n",
1607                          len,idx2str(idx));
1608         }
1609         //A local element never expires
1610         return CmiFalse;
1611 }
1612
1613 /**********Added for cosmology (inline function handling without parameter marshalling)***********/
1614
1615 LDObjHandle CkMigratable::timingBeforeCall(int* objstopped){
1616
1617         LDObjHandle objHandle;
1618 #if CMK_LBDB_ON
1619         if (getLBDB()->RunningObject(&objHandle)) {
1620                 *objstopped = 1;
1621                 getLBDB()->ObjectStop(objHandle);
1622         }
1623         myRec->startTiming(1);
1624 #endif
1625
1626   //DEBS((AA"   Invoking entry %d on element %s\n"AB,epIdx,idx2str(idx)));
1627         //CmiBool isDeleted=CmiFalse; //Enables us to detect deletion during processing
1628         //deletedMarker=&isDeleted;
1629 /*#ifndef CMK_OPTIMIZE
1630         if (msg) {  Tracing: 
1631                 envelope *env=UsrToEnv(msg);
1632         //      CkPrintf("ckLocation.C beginExecuteDetailed %d %d \n",env->getEvent(),env->getsetArraySrcPe());
1633                 if (_entryTable[epIdx]->traceEnabled)
1634                         _TRACE_BEGIN_EXECUTE_DETAILED(env->getEvent(),
1635                                  ForChareMsg,epIdx,env->getsetArraySrcPe(), env->getTotalsize(), idx.getProjectionID(((CkGroupID)env->getsetArrayMgr())).idx);
1636         }
1637 #endif*/
1638
1639   return objHandle;
1640 }
1641
1642 void CkMigratable::timingAfterCall(LDObjHandle objHandle,int *objstopped){
1643   
1644 /*#ifndef CMK_OPTIMIZE
1645         if (msg) {  Tracing: 
1646                 if (_entryTable[epIdx]->traceEnabled)
1647                         _TRACE_END_EXECUTE();
1648         }
1649 #endif*/
1650 //#if CMK_LBDB_ON
1651 //        if (!isDeleted) checkBufferedMigration();   // check if should migrate
1652 //#endif
1653 //      if (isDeleted) return CmiFalse;//We were deleted
1654 //      deletedMarker=NULL;
1655 //      return CmiTrue;
1656         myRec->stopTiming(1);
1657 #if CMK_LBDB_ON
1658         if (*objstopped) {
1659                  getLBDB()->ObjectStart(objHandle);
1660         }
1661 #endif
1662
1663  return;
1664 }
1665 /****************************************************************************/
1666
1667
1668 CmiBool CkLocRec_local::invokeEntry(CkMigratable *obj,void *msg,
1669         int epIdx,CmiBool doFree) 
1670 {
1671
1672         DEBS((AA"   Invoking entry %d on element %s\n"AB,epIdx,idx2str(idx)));
1673         CmiBool isDeleted=CmiFalse; //Enables us to detect deletion during processing
1674         deletedMarker=&isDeleted;
1675         startTiming();
1676
1677
1678 #if CMK_TRACE_ENABLED
1679         if (msg) { /* Tracing: */
1680                 envelope *env=UsrToEnv(msg);
1681         //      CkPrintf("ckLocation.C beginExecuteDetailed %d %d \n",env->getEvent(),env->getsetArraySrcPe());
1682                 if (_entryTable[epIdx]->traceEnabled)
1683                         _TRACE_BEGIN_EXECUTE_DETAILED(env->getEvent(),
1684                                  ForChareMsg,epIdx,env->getsetArraySrcPe(), env->getTotalsize(), idx.getProjectionID((((CkGroupID)env->getsetArrayMgr())).idx));
1685         }
1686 #endif
1687
1688         if (doFree) 
1689            CkDeliverMessageFree(epIdx,msg,obj);
1690         else /* !doFree */
1691            CkDeliverMessageReadonly(epIdx,msg,obj);
1692
1693
1694 #if CMK_TRACE_ENABLED
1695         if (msg) { /* Tracing: */
1696                 if (_entryTable[epIdx]->traceEnabled)
1697                         _TRACE_END_EXECUTE();
1698         }
1699 #endif
1700 #if CMK_LBDB_ON
1701         if (!isDeleted) checkBufferedMigration();   // check if should migrate
1702 #endif
1703         if (isDeleted) return CmiFalse;//We were deleted
1704         deletedMarker=NULL;
1705         stopTiming();
1706         return CmiTrue;
1707 }
1708
1709 CmiBool CkLocRec_local::deliver(CkArrayMessage *msg,CkDeliver_t type,int opts)
1710 {
1711
1712         if (type==CkDeliver_queue) { /*Send via the message queue */
1713                 if (opts & CK_MSG_KEEP)
1714                         msg = (CkArrayMessage *)CkCopyMsg((void **)&msg);
1715                 CkArrayManagerDeliver(CkMyPe(),msg,opts);
1716                 return CmiTrue;
1717         }
1718         else
1719         {
1720                 CkMigratable *obj=myLocMgr->lookupLocal(localIdx,
1721                         UsrToEnv(msg)->getsetArrayMgr());
1722                 if (obj==NULL) {//That sibling of this object isn't created yet!
1723                         if (opts & CK_MSG_KEEP)
1724                                 msg = (CkArrayMessage *)CkCopyMsg((void **)&msg);
1725                         if (msg->array_ifNotThere()!=CkArray_IfNotThere_buffer) {
1726                                 return myLocMgr->demandCreateElement(msg,CkMyPe(),type);
1727                         }
1728                         else {
1729                                 DEBS((AA"   BUFFERING message for nonexistent element %s!\n"AB,idx2str(this->idx)));
1730                                 halfCreated.enq(msg);
1731                                 return CmiTrue;
1732                         }
1733                 }
1734                         
1735                 if (msg->array_hops()>1)
1736                         myLocMgr->multiHop(msg);
1737                 CmiBool doFree = (CmiBool)!(opts & CK_MSG_KEEP);
1738 #if CMK_LBDB_ON
1739                 // if there is a running obj being measured, stop it temporarily
1740                 LDObjHandle objHandle;
1741                 int objstopped = 0;
1742                 if (the_lbdb->RunningObject(&objHandle)) {
1743                         objstopped = 1;
1744                         the_lbdb->ObjectStop(objHandle);
1745                 }
1746 #endif
1747 #if CMK_GRID_QUEUE_AVAILABLE
1748                 // retain a pointer to the sending object (needed later)
1749                 CpvAccess(CkGridObject) = obj;
1750 #endif
1751
1752         CmiBool status = invokeEntry(obj,(void *)msg,msg->array_ep(),doFree);
1753         
1754 #if CMK_GRID_QUEUE_AVAILABLE
1755                 CpvAccess(CkGridObject) = NULL;
1756 #endif
1757 #if CMK_LBDB_ON
1758                 if (objstopped) the_lbdb->ObjectStart(objHandle);
1759 #endif
1760                 return status;
1761         }
1762
1763
1764 }
1765
1766 #if CMK_LBDB_ON
1767
1768 void CkLocRec_local::staticMetaLBResumeWaitingChares(LDObjHandle h, int lb_ideal_period) {
1769         CkLocRec_local *el=(CkLocRec_local *)LDObjUserData(h);
1770         DEBL((AA"MetaBalancer wants to resume waiting chare %s\n"AB,idx2str(el->idx)));
1771         el->metaLBResumeWaitingChares(lb_ideal_period);
1772 }
1773
1774 void CkLocRec_local::metaLBResumeWaitingChares(int lb_ideal_period) {
1775   informIdealLBPeriod(lb_ideal_period);
1776 }
1777
1778 void CkLocRec_local::staticMetaLBCallLBOnChares(LDObjHandle h) {
1779         CkLocRec_local *el=(CkLocRec_local *)LDObjUserData(h);
1780         DEBL((AA"MetaBalancer wants to call LoadBalance on chare %s\n"AB,idx2str(el->idx)));
1781         el->metaLBCallLBOnChares();
1782 }
1783
1784 void CkLocRec_local::metaLBCallLBOnChares() {
1785   metaLBCallLB();
1786 }
1787
1788 void CkLocRec_local::staticMigrate(LDObjHandle h, int dest)
1789 {
1790         CkLocRec_local *el=(CkLocRec_local *)LDObjUserData(h);
1791         DEBL((AA"Load balancer wants to migrate %s to %d\n"AB,idx2str(el->idx),dest));
1792         el->recvMigrate(dest);
1793 }
1794
1795 void CkLocRec_local::recvMigrate(int toPe)
1796 {
1797         // we are in the mode of delaying actual migration
1798         // till readyMigrate()
1799         if (readyMigrate) { migrateMe(toPe); }
1800         else nextPe = toPe;
1801 }
1802
1803 void CkLocRec_local::AsyncMigrate(CmiBool use)  
1804 {
1805         asyncMigrate = use; 
1806         the_lbdb->UseAsyncMigrate(ldHandle, use);
1807 }
1808
1809 CmiBool CkLocRec_local::checkBufferedMigration()
1810 {
1811         // we don't migrate in user's code when calling ReadyMigrate(true)
1812         // we postphone the action to here until we exit from the user code.
1813         if (readyMigrate && nextPe != -1) {
1814             int toPe = nextPe;
1815             nextPe = -1;
1816             // don't migrate inside the object call
1817             migrateMe(toPe);
1818             // don't do anything
1819             return CmiTrue;
1820         }
1821         return CmiFalse;
1822 }
1823
1824 int CkLocRec_local::MigrateToPe()
1825 {
1826         int pe = nextPe;
1827         nextPe = -1;
1828         return pe;
1829 }
1830
1831 void CkLocRec_local::setMigratable(int migratable)
1832 {
1833         if (migratable)
1834           the_lbdb->Migratable(ldHandle);
1835         else
1836           the_lbdb->NonMigratable(ldHandle);
1837 }
1838 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
1839 void CkLocRec_local::Migrated(){
1840     the_lbdb->Migrated(ldHandle, CmiTrue);
1841 }
1842 #endif
1843 #endif
1844
1845 /**
1846  * Represents a deleted array element (and prevents re-use).
1847  * These are a debugging aid, usable only by uncommenting a line in
1848  * the element destruction code.
1849  */
1850 class CkLocRec_dead:public CkLocRec {
1851 public:
1852         CkLocRec_dead(CkLocMgr *Narr):CkLocRec(Narr) {}
1853   
1854         virtual RecType type(void) {return dead;}
1855   
1856         virtual CmiBool deliver(CkArrayMessage *msg,CkDeliver_t type,int opts=0) {
1857                 CkPrintf("Dead array element is %s.\n",idx2str(msg->array_index()));
1858                 CkAbort("Send to dead array element!\n");
1859                 return CmiFalse;
1860         }
1861         virtual void beenReplaced(void) 
1862                 {CkAbort("Can't re-use dead array element!\n");}
1863   
1864         //Return if this element is now obsolete (it isn't)
1865         virtual CmiBool isObsolete(int nSprings,const CkArrayIndex &idx) {return CmiFalse;}     
1866 };
1867
1868 /**
1869  * This is the abstract superclass of arrayRecs that keep track of their age,
1870  * and eventually expire. Its kids are remote and buffering.
1871  */
1872 class CkLocRec_aging:public CkLocRec {
1873 private:
1874         int lastAccess;//Age when last accessed
1875 protected:
1876         //Update our access time
1877         inline void access(void) {
1878                 lastAccess=myLocMgr->getSpringCount();
1879         }
1880         //Return if we are "stale"-- we were last accessed a while ago
1881         CmiBool isStale(void) {
1882                 if (myLocMgr->getSpringCount()-lastAccess>3) return CmiTrue;
1883                 else return CmiFalse;
1884         }
1885 public:
1886         CkLocRec_aging(CkLocMgr *Narr):CkLocRec(Narr) {
1887                 lastAccess=myLocMgr->getSpringCount();
1888         }
1889         //Return if this element is now obsolete
1890         virtual CmiBool isObsolete(int nSprings,const CkArrayIndex &idx)=0;
1891         //virtual void pup(PUP::er &p) { CkLocRec::pup(p); p(lastAccess); }
1892 };
1893
1894
1895 /**
1896  * Represents a remote array element.  This is just a PE number.
1897  */
1898 class CkLocRec_remote:public CkLocRec_aging {
1899 private:
1900         int onPe;//The last known Pe for this element
1901 public:
1902         CkLocRec_remote(CkLocMgr *Narr,int NonPe)
1903                 :CkLocRec_aging(Narr)
1904                 {
1905                         onPe=NonPe;
1906 #if CMK_ERROR_CHECKING
1907                         if (onPe==CkMyPe())
1908                                 CkAbort("ERROR!  'remote' array element on this Pe!\n");
1909 #endif
1910                 }
1911         //Return the last known processor for this element
1912         int lookupProcessor(void) {
1913                 return onPe;
1914         }  
1915         virtual RecType type(void) {return remote;}
1916   
1917         //Send a message for this element.
1918         virtual CmiBool deliver(CkArrayMessage *msg,CkDeliver_t type,int opts=0) {
1919                 /*FAULT_EVAC*/
1920                 int destPE = onPe;
1921                 if((!CmiNodeAlive(onPe) && onPe != allowMessagesOnly)){
1922 //                      printf("Delivery failed because process %d is invalid\n",onPe);
1923                         /*
1924                                 Send it to its home processor instead
1925                         */
1926                         const CkArrayIndex &idx=msg->array_index();
1927                         destPE = getNextPE(idx);
1928                 }
1929                 access();//Update our modification date
1930                 msg->array_hops()++;
1931                 DEBS((AA"   Forwarding message for element %s to %d (REMOTE)\n"AB,
1932                       idx2str(msg->array_index()),destPE));
1933                 if (opts & CK_MSG_KEEP)
1934                         msg = (CkArrayMessage *)CkCopyMsg((void **)&msg);
1935                 CkArrayManagerDeliver(destPE,msg,opts);
1936                 return CmiTrue;
1937         }
1938         //Return if this element is now obsolete
1939         virtual CmiBool isObsolete(int nSprings,const CkArrayIndex &idx) {
1940                 if (myLocMgr->isHome(idx)) 
1941                         //Home elements never become obsolete
1942                         // if they did, we couldn't deliver messages to that element.
1943                         return CmiFalse;
1944                 else if (isStale())
1945                         return CmiTrue;//We haven't been used in a long time
1946                 else
1947                         return CmiFalse;//We're fairly recent
1948         }
1949         //virtual void pup(PUP::er &p) { CkLocRec_aging::pup(p); p(onPe); }
1950 };
1951
1952
1953 /**
1954  * Buffers messages until record is replaced in the hash table, 
1955  * then delivers all messages to the replacing record.  This is 
1956  * used when a message arrives for a local element that has not
1957  * yet been created, buffering messages until the new element finally
1958  * checks in.
1959  *
1960  * It's silly to send a message to an element you won't ever create,
1961  * so this kind of record causes an abort "Stale array manager message!"
1962  * if it's left undelivered too long.
1963  */
1964 class CkLocRec_buffering:public CkLocRec_aging {
1965 private:
1966         CkQ<CkArrayMessage *> buffer;//Buffered messages.
1967 public:
1968         CkLocRec_buffering(CkLocMgr *Narr):CkLocRec_aging(Narr) {}
1969         virtual ~CkLocRec_buffering() {
1970                 if (0!=buffer.length()) {
1971                         CkPrintf("[%d] Warning: Messages abandoned in array manager buffer!\n", CkMyPe());
1972                         CkArrayMessage *m;
1973                         while (NULL!=(m=buffer.deq()))  {
1974                                 delete m;
1975                         }
1976                 }
1977         }
1978   
1979         virtual RecType type(void) {return buffering;}
1980   
1981         //Buffer a message for this element.
1982         virtual CmiBool deliver(CkArrayMessage *msg,CkDeliver_t type,int opts=0) {
1983                 DEBS((AA" Queued message for %s\n"AB,idx2str(msg->array_index())));
1984                 if (opts & CK_MSG_KEEP)
1985                         msg = (CkArrayMessage *)CkCopyMsg((void **)&msg);
1986                 buffer.enq(msg);
1987 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
1988                 envelope *env = UsrToEnv(msg);
1989                 env->sender = CpvAccess(_currentObj)->mlogData->objID;
1990 #endif
1991                 return CmiTrue;
1992         }
1993  
1994         //This is called when this ArrayRec is about to be replaced.
1995         // We dump all our buffered messages off on the next guy,
1996         // who should know what to do with them.
1997         virtual void beenReplaced(void) {
1998                 DEBS((AA" Delivering queued messages:\n"AB));
1999                 CkArrayMessage *m;
2000                 while (NULL!=(m=buffer.deq())) {
2001 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))         
2002                 DEBUG(CmiPrintf("[%d] buffered message being sent\n",CmiMyPe()));
2003                 envelope *env = UsrToEnv(m);
2004                 Chare *oldObj = CpvAccess(_currentObj);
2005                 CpvAccess(_currentObj) =(Chare *) env->sender.getObject();
2006                 env->sender.type = TypeInvalid;
2007 #endif
2008                 DEBS((AA"Sending buffered message to %s\n"AB,idx2str(m->array_index())));
2009                 myLocMgr->deliverViaQueue(m);
2010 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))         
2011                 CpvAccess(_currentObj) = oldObj;
2012 #endif
2013                 }
2014         }
2015   
2016         //Return if this element is now obsolete
2017         virtual CmiBool isObsolete(int nSprings,const CkArrayIndex &idx) {
2018                 if (isStale() && buffer.length()>0) {
2019                         /*This indicates something is seriously wrong--
2020                           buffers should be short-lived.*/
2021                         CkPrintf("[%d] WARNING: %d stale array message(s) found!\n",CkMyPe(),buffer.length());
2022                         CkArrayMessage *msg=buffer[0];
2023                         CkPrintf("Addressed to: ");
2024                         CkPrintEntryMethod(msg->array_ep());
2025                         CkPrintf(" index %s\n",idx2str(idx));
2026                         if (myLocMgr->isHome(idx)) 
2027                                 CkPrintf("is this an out-of-bounds array index, or was it never created?\n");
2028                         else //Idx is a remote-home index
2029                                 CkPrintf("why weren't they forwarded?\n");
2030                         
2031                         // CkAbort("Stale array manager message(s)!\n");
2032                 }
2033                 return CmiFalse;
2034         }
2035   
2036 /*  virtual void pup(PUP::er &p) {
2037     CkLocRec_aging::pup(p);
2038     CkArray::pupArrayMsgQ(buffer, p);
2039     }*/
2040 };
2041
2042 /*********************** Spring Cleaning *****************/
2043 /**
2044  * Used to periodically flush out unused remote element pointers.
2045  *
2046  * Cleaning often will free up memory quickly, but slow things
2047  * down because the cleaning takes time and some not-recently-referenced
2048  * remote element pointers might be valid and used some time in 
2049  * the future.
2050  *
2051  * Also used to determine if buffered messages have become stale.
2052  */
2053 inline void CkLocMgr::springCleaning(void)
2054 {
2055   nSprings++;
2056
2057   //Poke through the hash table for old ArrayRecs.
2058   void *objp;
2059   void *keyp;
2060   
2061   CkHashtableIterator *it=hash.iterator();
2062   CmiImmediateLock(hashImmLock);
2063   while (NULL!=(objp=it->next(&keyp))) {
2064     CkLocRec *rec=*(CkLocRec **)objp;
2065     CkArrayIndex &idx=*(CkArrayIndex *)keyp;
2066     if (rec->isObsolete(nSprings,idx)) {
2067       //This record is obsolete-- remove it from the table
2068       DEBK((AA"Cleaning out old record %s\n"AB,idx2str(idx)));
2069       hash.remove(*(CkArrayIndex *)&idx);
2070       delete rec;
2071       it->seek(-1);//retry this hash slot
2072     }
2073   }
2074   CmiImmediateUnlock(hashImmLock);
2075   delete it;
2076 }
2077 void CkLocMgr::staticSpringCleaning(void *forWhom,double curWallTime) {
2078         DEBK((AA"Starting spring cleaning at %.2f\n"AB,CkWallTimer()));
2079         ((CkLocMgr *)forWhom)->springCleaning();
2080 }
2081 //doesn't delete if there is extra pe
2082 void CkLocMgr::flushLocalRecs(void)
2083 {
2084   void *objp;
2085   void *keyp;
2086   CkHashtableIterator *it=hash.iterator();
2087   CmiImmediateLock(hashImmLock);
2088   while (NULL!=(objp=it->next(&keyp))) {
2089     CkLocRec *rec=*(CkLocRec **)objp;
2090     CkArrayIndex &idx=*(CkArrayIndex *)keyp;
2091     if (rec->type() == CkLocRec::local) {
2092         callMethod((CkLocRec_local*)rec, &CkMigratable::ckDestroy);
2093         it->seek(-1);//retry this hash slot
2094     }
2095   }
2096   delete it;
2097   CmiImmediateUnlock(hashImmLock);
2098 }
2099
2100 void CkLocMgr::recvChkpIter(int iter)
2101 {
2102   void *objp;
2103   void *keyp;
2104   CkHashtableIterator *it=hash.iterator();
2105   CmiImmediateLock(hashImmLock);
2106   while (NULL!=(objp=it->next(&keyp))) {
2107     CkLocRec *rec=*(CkLocRec **)objp;
2108     CkArrayIndex &idx=*(CkArrayIndex *)keyp;
2109     if (rec->type() == CkLocRec::local) {
2110         callMethod((CkLocRec_local*)rec, &CkMigratable::recvChkpIter,&iter);
2111     }
2112   }
2113   delete it;
2114   CmiImmediateUnlock(hashImmLock);
2115 }
2116
2117 void CkLocMgr::resumeFromChkp(){
2118   void *objp;
2119   void *keyp;
2120   CkHashtableIterator *it=hash.iterator();
2121   CmiImmediateLock(hashImmLock);
2122   while (NULL!=(objp=it->next(&keyp))) {
2123     CkLocRec *rec=*(CkLocRec **)objp;
2124     CkArrayIndex &idx=*(CkArrayIndex *)keyp;
2125     if (rec->type() == CkLocRec::local) {
2126         callMethod((CkLocRec_local*)rec, &CkMigratable::ResumeFromChkp);
2127     }
2128   }
2129   delete it;
2130   CmiImmediateUnlock(hashImmLock);
2131 }
2132
2133 // clean all buffer'ed messages and also free local objects
2134 void CkLocMgr::flushAllRecs(void)
2135 {
2136   void *objp;
2137   void *keyp;
2138   CkHashtableIterator *it=hash.iterator();
2139   CmiImmediateLock(hashImmLock);
2140   while (NULL!=(objp=it->next(&keyp))) {
2141     CkLocRec *rec=*(CkLocRec **)objp;
2142     CkArrayIndex &idx=*(CkArrayIndex *)keyp;
2143     if (rec->type() != CkLocRec::local) {
2144       //In the case of taking core out of memory (in BigSim's emulation)
2145       //the meta data in the location manager are not deleted so we need
2146       //this condition
2147       
2148       if(_BgOutOfCoreFlag!=1){
2149         hash.remove(*(CkArrayIndex *)&idx);
2150         delete rec;
2151         it->seek(-1);//retry this hash slot
2152       }
2153     }
2154     else {
2155         callMethod((CkLocRec_local*)rec, &CkMigratable::ckDestroy);
2156         it->seek(-1);//retry this hash slot
2157     }
2158   }
2159   delete it;
2160   CmiImmediateUnlock(hashImmLock);
2161 }
2162
2163
2164 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
2165 void CkLocMgr::callForAllRecords(CkLocFn fnPointer,CkArray *arr,void *data){
2166         void *objp;
2167         void *keyp;
2168
2169         CkHashtableIterator *it = hash.iterator();
2170         while (NULL!=(objp=it->next(&keyp))) {
2171                 CkLocRec *rec=*(CkLocRec **)objp;
2172                 CkArrayIndex &idx=*(CkArrayIndex *)keyp;
2173                 fnPointer(arr,data,rec,&idx);
2174         }
2175
2176         // releasing iterator memory
2177         delete it;
2178 }
2179 #endif
2180
2181 /*************************** LocMgr: CREATION *****************************/
2182 CkLocMgr::CkLocMgr(CkGroupID mapID_,CkGroupID lbdbID_,CkGroupID metalbID_,CkArrayIndex& numInitial)
2183         :thisProxy(thisgroup),thislocalproxy(thisgroup,CkMyPe()),
2184          hash(17,0.3)
2185 {
2186         DEBC((AA"Creating new location manager %d\n"AB,thisgroup));
2187 // moved to _CkMigratable_initInfoInit()
2188 //      CkpvInitialize(CkMigratable_initInfo,mig_initInfo);
2189
2190         managers.init();
2191         nManagers=0;
2192         firstManager=NULL;
2193         firstFree=localLen=0;
2194         duringMigration=CmiFalse;
2195         nSprings=0;
2196 #if !CMK_GLOBAL_LOCATION_UPDATE
2197         CcdCallOnConditionKeepOnPE(CcdPERIODIC_1minute,staticSpringCleaning,(void *)this, CkMyPe());
2198 #endif
2199
2200 //Register with the map object
2201         mapID=mapID_;
2202         map=(CkArrayMap *)CkLocalBranch(mapID);
2203         if (map==NULL) CkAbort("ERROR!  Local branch of array map is NULL!");
2204         mapHandle=map->registerArray(numInitial,thisgroup);
2205
2206 //Find and register with the load balancer
2207         lbdbID = lbdbID_;
2208   metalbID = metalbID_;
2209         initLB(lbdbID_, metalbID_);
2210         hashImmLock = CmiCreateImmediateLock();
2211         chkpID = ckCheckPTGroupID;
2212         the_chkp = (CkMemCheckPT *)CkLocalBranch(chkpID);
2213 }
2214
2215 CkLocMgr::CkLocMgr(CkGroupID mapID_,CkGroupID lbdbID_,CkArrayIndex& numInitial)
2216         :thisProxy(thisgroup),thislocalproxy(thisgroup,CkMyPe()),
2217          hash(17,0.3)
2218 {
2219 #if !CMK_LBDB_ON
2220         CkGroupID _metalb;
2221 #endif
2222         CkLocMgr(mapID_, lbdbID_, _metalb, numInitial);
2223 }
2224
2225
2226 CkLocMgr::CkLocMgr(CkMigrateMessage* m)
2227         :IrrGroup(m),thisProxy(thisgroup),thislocalproxy(thisgroup,CkMyPe()),hash(17,0.3)
2228 {
2229         managers.init();
2230         nManagers=0;
2231         firstManager=NULL;
2232         firstFree=localLen=0;
2233         duringMigration=CmiFalse;
2234         nSprings=0;
2235 #if !CMK_GLOBAL_LOCATION_UPDATE
2236         CcdCallOnConditionKeepOnPE(CcdPERIODIC_1minute,staticSpringCleaning,(void *)this, CkMyPe());
2237 #endif
2238         hashImmLock = CmiCreateImmediateLock();
2239 }
2240
2241 void CkLocMgr::pup(PUP::er &p){
2242         IrrGroup::pup(p);
2243         p|mapID;
2244         p|mapHandle;
2245         p|lbdbID;
2246   p|metalbID;
2247    p|chkpID;
2248         mapID = _defaultArrayMapID;
2249         if(p.isUnpacking()){
2250                 thisProxy=thisgroup;
2251                 CProxyElement_CkLocMgr newlocalproxy(thisgroup,CkMyPe());
2252                 thislocalproxy=newlocalproxy;
2253                 //Register with the map object
2254                 map=(CkArrayMap *)CkLocalBranch(mapID);
2255                 if (map==NULL) CkAbort("ERROR!  Local branch of array map is NULL!");
2256                 CkArrayIndex emptyIndex;
2257                 map->registerArray(emptyIndex,thisgroup);
2258                 // _lbdb is the fixed global groupID
2259                 initLB(lbdbID, metalbID);
2260                 the_chkp = (CkMemCheckPT *)CkLocalBranch(chkpID);
2261 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_)) ||CMK_MEM_CHECKPOINT    
2262         int count;
2263         p | count;
2264         DEBUG(CmiPrintf("[%d] Unpacking Locmgr %d has %d home elements\n",CmiMyPe(),thisgroup.idx,count));
2265 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))    
2266         homeElementCount = count;
2267 #endif
2268         for(int i=0;i<count;i++){
2269             CkArrayIndex idx;
2270             int pe;
2271             idx.pup(p);
2272             p | pe;
2273   //          CmiPrintf("[%d] idx %s is a home element exisiting on pe %d\n",CmiMyPe(),idx2str(idx),pe);
2274             inform(idx,pe);
2275             CkLocRec *rec = elementNrec(idx);
2276             CmiAssert(rec!=NULL);
2277             CmiAssert(lastKnown(idx) == pe);
2278         }
2279 #endif
2280                 // delay doneInserting when it is unpacking during restart.
2281                 // to prevent load balancing kicking in
2282                 if (!CkInRestarting()) 
2283                         doneInserting();
2284         }else{
2285  /**
2286  * pack the indexes of elements which have their homes on this processor
2287  * but dont exist on it.. needed for broadcast after a restart
2288  * indexes of local elements dont need to be packed
2289  * since they will be recreated later anyway
2290  */
2291 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))  || CMK_MEM_CHECKPOINT   
2292                 int count=0,count1=0;
2293         void *objp;
2294         void *keyp;
2295         CkVec<int> pe_list;
2296         CkVec<CkArrayIndex> idx_list;
2297         CkHashtableIterator *it = hash.iterator();
2298       while (NULL!=(objp=it->next(&keyp))) {
2299           CkLocRec *rec=*(CkLocRec **)objp;
2300           CkArrayIndex &idx=*(CkArrayIndex *)keyp;
2301             if(rec->type() != CkLocRec::local){
2302                 if(homePe(idx) == CmiMyPe()){
2303                   int pe;
2304                   CkArrayIndex max = idx;
2305                   pe = rec->lookupProcessor();
2306                   idx_list.push_back(max);
2307                   pe_list.push_back(pe);
2308                     count++;
2309                 }
2310             }
2311         }
2312         p | count;
2313
2314                 // releasing iterator memory
2315                 delete it;
2316
2317       for(int i=0;i<pe_list.length();i++){
2318         CkArrayIndex max = idx_list[i];
2319         max.pup(p);
2320         p|pe_list[i];
2321       }
2322     /*    it = hash.iterator();
2323       while (NULL!=(objp=it->next(&keyp))) {
2324       CkLocRec *rec=*(CkLocRec **)objp;
2325         CkArrayIndex &idx=*(CkArrayIndex *)keyp;
2326             CkArrayIndex max = idx;
2327             if(rec->type() != CkLocRec::local){
2328                 if(homePe(idx) == CmiMyPe()){
2329                     int pe;
2330                     max.pup(p);
2331                     pe = rec->lookupProcessor();
2332                     p | pe;
2333                     count1++;
2334                 }
2335             }
2336         }
2337       //  CmiAssert(count == count1);
2338
2339                 // releasing iterator memory
2340                 delete it;*/
2341
2342 #endif
2343
2344         }
2345 }
2346
2347 void _CkLocMgrInit(void) {
2348   /* Don't trace our deliver method--it does its own tracing */
2349   CkDisableTracing(CkIndex_CkLocMgr::deliverInline(0));
2350 }
2351
2352 /// Add a new local array manager to our list.
2353 /// Returns a new CkMigratableList for the manager to store his
2354 /// elements in.
2355 CkMigratableList *CkLocMgr::addManager(CkArrayID id,CkArrMgr *mgr)
2356 {
2357         CK_MAGICNUMBER_CHECK
2358         DEBC((AA"Adding new array manager\n"AB));
2359         //Link new manager into list
2360         ManagerRec *n=new ManagerRec;
2361         managers.find(id)=n;
2362         n->next=firstManager;
2363         n->mgr=mgr;
2364         n->elts.setSize(localLen);
2365         nManagers++;
2366         firstManager=n;
2367         return &n->elts;
2368 }
2369
2370 /// Return the next unused local element index.
2371 int CkLocMgr::nextFree(void) {
2372         if (firstFree>=localLen)
2373         {//Need more space in the local index arrays-- enlarge them
2374                 int oldLen=localLen;
2375                 localLen=localLen*2+8;
2376                 DEBC((AA"Growing the local list from %d to %d...\n"AB,oldLen,localLen));
2377                 for (ManagerRec *m=firstManager;m!=NULL;m=m->next)
2378                         m->elts.setSize(localLen);
2379                 //Update the free list
2380                 freeList.resize(localLen);
2381                 for (int i=oldLen;i<localLen;i++)
2382                         freeList[i]=i+1;
2383         }
2384         int localIdx=firstFree;
2385         if (localIdx==-1) CkAbort("CkLocMgr free list corrupted!");
2386         firstFree=freeList[localIdx];
2387         freeList[localIdx]=-1; //Mark as used
2388         return localIdx;
2389 }
2390
2391 CkLocRec_remote *CkLocMgr::insertRemote(const CkArrayIndex &idx,int nowOnPe)
2392 {
2393         DEBS((AA"Remote element %s lives on %d\n"AB,idx2str(idx),nowOnPe));
2394         CkLocRec_remote *rem=new CkLocRec_remote(this,nowOnPe);
2395         insertRec(rem,idx);
2396         return rem;
2397 }
2398
2399 //This element now lives on the given Pe
2400 void CkLocMgr::inform(const CkArrayIndex &idx,int nowOnPe)
2401 {
2402         if (nowOnPe==CkMyPe())
2403                 return; //Never insert a "remote" record pointing here
2404         CkLocRec *rec=elementNrec(idx);
2405         if (rec!=NULL && rec->type()==CkLocRec::local){
2406 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
2407         CmiPrintf("[%d]WARNING!!! Element %d:%s is local but is being told it exists on %d\n",CkMyPe(),idx.dimension,idx2str(idx), nowOnPe);
2408 #endif
2409                 return; //Never replace a local element's record!
2410         }
2411         insertRemote(idx,nowOnPe);
2412 }
2413
2414 //Tell this element's home processor it now lives "there"
2415 void CkLocMgr::informHome(const CkArrayIndex &idx,int nowOnPe)
2416 {
2417         int home=homePe(idx);
2418         if (home!=CkMyPe() && home!=nowOnPe) {
2419                 //Let this element's home Pe know it lives here now
2420                 DEBC((AA"  Telling %s's home %d that it lives on %d.\n"AB,idx2str(idx),home,nowOnPe));
2421 //#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
2422 //#if defined(_FAULT_MLOG_)
2423 //        informLocationHome(thisgroup,idx,home,CkMyPe());
2424 //#else
2425                 thisProxy[home].updateLocation(idx,nowOnPe);
2426 //#endif
2427         }
2428 }
2429
2430 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
2431 CkLocRec_local *CkLocMgr::createLocal(const CkArrayIndex &idx,
2432         CmiBool forMigration, CmiBool ignoreArrival,
2433         CmiBool notifyHome,int dummy)
2434 {
2435     int localIdx=nextFree();
2436     DEBC((AA"Adding new record for element %s at local index %d\n"AB,idx2str(idx),localIdx));
2437     CkLocRec_local *rec=new CkLocRec_local(this,forMigration,ignoreArrival,idx,localIdx);
2438     if(!dummy){
2439         insertRec(rec,idx); //Add to global hashtable
2440     }   
2441     if (notifyHome) informHome(idx,CkMyPe());
2442     return rec; 
2443 }
2444 #else
2445 CkLocRec_local *CkLocMgr::createLocal(const CkArrayIndex &idx, 
2446                 CmiBool forMigration, CmiBool ignoreArrival,
2447                 CmiBool notifyHome)
2448 {
2449         //CkPrintf("Adding new record for element %s\n",idx2str(idx));
2450         int localIdx=nextFree();
2451         DEBC((AA"Adding new record for element %s at local index %d\n"AB,idx2str(idx),localIdx));
2452         //CkPrintf("Adding new record for element %s at local index %d\n",idx2str(idx),localIdx);
2453         CkLocRec_local *rec=new CkLocRec_local(this,forMigration,ignoreArrival,idx,localIdx);
2454         insertRec(rec,idx); //Add to global hashtable
2455         if (notifyHome) informHome(idx,CkMyPe());
2456         return rec;
2457 }
2458 #endif
2459
2460 //Add a new local array element, calling element's constructor
2461 CmiBool CkLocMgr::addElement(CkArrayID id,const CkArrayIndex &idx,
2462                 CkMigratable *elt,int ctorIdx,void *ctorMsg)
2463 {
2464         CK_MAGICNUMBER_CHECK
2465         CkLocRec *oldRec=elementNrec(idx);
2466         CkLocRec_local *rec;
2467         if (oldRec==NULL||oldRec->type()!=CkLocRec::local) 
2468         { //This is the first we've heard of that element-- add new local record
2469                 rec=createLocal(idx,CmiFalse,CmiFalse,CmiTrue);
2470 #if CMK_GLOBAL_LOCATION_UPDATE
2471                 if (homePe(idx) != CkMyPe()) {
2472                   DEBC((AA"Global location broadcast for new element idx %s "
2473                         "assigned to %d \n"AB, idx2str(idx), CkMyPe()));
2474                   thisProxy.updateLocation(idx, CkMyPe());  
2475                 }
2476 #endif
2477                 
2478         } else 
2479         { //rec is *already* local-- must not be the first insertion    
2480                 rec=((CkLocRec_local *)oldRec);
2481                 rec->addedElement();
2482         }
2483         if (!addElementToRec(rec,managers.find(id),elt,ctorIdx,ctorMsg)) return CmiFalse;
2484         elt->ckFinishConstruction();
2485         return CmiTrue;
2486 }
2487
2488 //As above, but shared with the migration code
2489 CmiBool CkLocMgr::addElementToRec(CkLocRec_local *rec,ManagerRec *m,
2490                 CkMigratable *elt,int ctorIdx,void *ctorMsg)
2491 {//Insert the new element into its manager's local list
2492         int localIdx=rec->getLocalIndex();
2493         if (m->elts.get(localIdx)!=NULL) CkAbort("Cannot insert array element twice!");
2494         m->elts.put(elt,localIdx); //Local element table
2495
2496 //Call the element's constructor
2497         DEBC((AA"Constructing element %s of array\n"AB,idx2str(rec->getIndex())));
2498         CkMigratable_initInfo &i=CkpvAccess(mig_initInfo);
2499         i.locRec=rec;
2500         i.chareType=_entryTable[ctorIdx]->chareIdx;
2501         if (!rec->invokeEntry(elt,ctorMsg,ctorIdx,CmiTrue)) return CmiFalse;
2502
2503 #if CMK_OUT_OF_CORE
2504         /* Register new element with out-of-core */
2505         PUP::sizer p_getSize; elt->pup(p_getSize);
2506         elt->prefetchObjID=CooRegisterObject(&CkArrayElementPrefetcher,p_getSize.size(),elt);
2507 #endif
2508         
2509         return CmiTrue;
2510 }
2511 void CkLocMgr::updateLocation(const CkArrayIndex &idx,int nowOnPe) {
2512   inform(idx,nowOnPe);
2513         CProxy_CkMemCheckPT checkptMgr(ckCheckPTGroupID);
2514 }
2515
2516 /*************************** LocMgr: DELETION *****************************/
2517 /// This index will no longer be used-- delete the associated elements
2518 void CkLocMgr::reclaim(const CkArrayIndex &idx,int localIdx) {
2519         CK_MAGICNUMBER_CHECK
2520         DEBC((AA"Destroying element %s (local %d)\n"AB,idx2str(idx),localIdx));
2521         //Delete, and mark as empty, each array element
2522         for (ManagerRec *m=firstManager;m!=NULL;m=m->next) {
2523                 delete m->elts.get(localIdx);
2524                 m->elts.empty(localIdx);
2525         }
2526         
2527         removeFromTable(idx);
2528         
2529         //Link local index into free list
2530         freeList[localIdx]=firstFree;
2531         firstFree=localIdx;
2532         
2533                 
2534         if (!duringMigration) 
2535         { //This is a local element dying a natural death
2536             #if CMK_BIGSIM_CHARM
2537                 //After migration, reclaimRemote will be called through 
2538                 //the CkRemoveArrayElement in the pupping routines for those 
2539                 //objects that are not on the home processors. However,
2540                 //those remote records should not be deleted since the corresponding
2541                 //objects are not actually deleted but on disk. If deleted, msgs
2542                 //that seeking where is the object will be accumulated (a circular
2543                 //msg chain) and causes the program no progress
2544                 if(_BgOutOfCoreFlag==1) return; 
2545             #endif
2546                 int home=homePe(idx);
2547                 if (home!=CkMyPe())
2548 #if CMK_MEM_CHECKPOINT
2549                 if (!CkInRestarting()) // all array elements are removed anyway
2550 #endif
2551                         thisProxy[home].reclaimRemote(idx,CkMyPe());
2552         /*      //Install a zombie to keep the living from re-using this index.
2553                 insertRecN(new CkLocRec_dead(this),idx); */
2554         }
2555 }
2556
2557 void CkLocMgr::reclaimRemote(const CkArrayIndex &idx,int deletedOnPe) {
2558         DEBC((AA"Our element %s died on PE %d\n"AB,idx2str(idx),deletedOnPe));
2559         CkLocRec *rec=elementNrec(idx);
2560         if (rec==NULL) return; //We never knew him
2561         if (rec->type()==CkLocRec::local) return; //He's already been reborn
2562         removeFromTable(idx);
2563         delete rec;
2564 }
2565 void CkLocMgr::removeFromTable(const CkArrayIndex &idx) {
2566 #if CMK_ERROR_CHECKING
2567         //Make sure it's actually in the table before we delete it
2568         if (NULL==elementNrec(idx))
2569                 CkAbort("CkLocMgr::removeFromTable called on invalid index!");
2570 #endif
2571         CmiImmediateLock(hashImmLock);
2572         hash.remove(*(CkArrayIndex *)&idx);
2573         CmiImmediateUnlock(hashImmLock);
2574 #if CMK_ERROR_CHECKING
2575         //Make sure it's really gone
2576         if (NULL!=elementNrec(idx))
2577                 CkAbort("CkLocMgr::removeFromTable called, but element still there!");
2578 #endif
2579 }
2580
2581 /************************** LocMgr: MESSAGING *************************/
2582 /// Deliver message to this element, going via the scheduler if local
2583 /// @return 0 if object local, 1 if not
2584 int CkLocMgr::deliver(CkMessage *m,CkDeliver_t type,int opts) {
2585         DEBS((AA"deliver \n"AB));
2586         CK_MAGICNUMBER_CHECK
2587         CkArrayMessage *msg=(CkArrayMessage *)m;
2588
2589
2590         const CkArrayIndex &idx=msg->array_index();
2591         DEBS((AA"deliver %s\n"AB,idx2str(idx)));
2592         if (type==CkDeliver_queue)
2593                 _TRACE_CREATION_DETAILED(UsrToEnv(m),msg->array_ep());
2594         CkLocRec *rec=elementNrec(idx);
2595         if(rec != NULL){
2596                 DEBS((AA"deliver %s of type %d \n"AB,idx2str(idx),rec->type()));
2597         }else{
2598                 DEBS((AA"deliver %s rec is null\n"AB,idx2str(idx)));
2599         }
2600 //#if (!defined(_FAULT_MLOG_) && !defined(_FAULT_CAUSAL_))
2601 //#if !defined(_FAULT_MLOG_)
2602 #if CMK_LBDB_ON
2603
2604         LDObjid ldid = idx2LDObjid(idx);
2605 #if CMK_GLOBAL_LOCATION_UPDATE
2606         ldid.locMgrGid = thisgroup.idx;
2607 #endif        
2608         if (type==CkDeliver_queue) {
2609                 if (!(opts & CK_MSG_LB_NOTRACE) && the_lbdb->CollectingCommStats()) {
2610                 if(rec!=NULL) the_lbdb->Send(myLBHandle,ldid,UsrToEnv(msg)->getTotalsize(), rec->lookupProcessor(), 1);
2611                 else /*rec==NULL*/ the_lbdb->Send(myLBHandle,ldid,UsrToEnv(msg)->getTotalsize(),homePe(msg->array_index()), 1);
2612                 }
2613         }
2614 #endif
2615 //#endif
2616 #if CMK_GRID_QUEUE_AVAILABLE
2617         int gridSrcPE;
2618         int gridSrcCluster;
2619         int gridDestPE;
2620         int gridDestCluster;
2621         CkMigratable *obj;
2622         ArrayElement *obj2;
2623         CkGroupID gid;
2624         int *data;
2625
2626         obj = (CkMigratable *) CpvAccess(CkGridObject);   // CkGridObject is a pointer to the sending object (retained earlier)
2627         if (obj != NULL) {
2628           obj2 = dynamic_cast<ArrayElement *> (obj);
2629           if (obj2 > 0) {
2630             // Get the sending object's array gid and indexes.
2631             // These are guaranteed to exist due to the succeeding dynamic cast above.
2632             gid = obj2->ckGetArrayID ();
2633             data = obj2->thisIndexMax.data ();
2634
2635             // Get the source PE and destination PE.
2636             gridSrcPE = CkMyPe ();
2637             if (rec != NULL) {
2638               gridDestPE = rec->lookupProcessor ();
2639             } else {
2640               gridDestPE = homePe (msg->array_index ());
2641             }
2642
2643             // Get the source cluster and destination cluster.
2644             gridSrcCluster = CmiGetCluster (gridSrcPE);
2645             gridDestCluster = CmiGetCluster (gridDestPE);
2646
2647             // If the Grid queue interval is greater than zero, it means that the more complicated
2648             // technique for registering border objects that exceed a specified threshold of
2649             // cross-cluster messages within a specified interval (and deregistering border objects
2650             // that do not meet this threshold) is used.  Otherwise a much simpler technique is used
2651             // where a border object is registered immediately upon sending a single cross-cluster
2652             // message (and deregistered when load balancing takes place).
2653             if (obj2->grid_queue_interval > 0) {
2654               // Increment the sending object's count of all messages.
2655               obj2->msg_count += 1;
2656
2657               // If the source cluster and destination cluster differ, this is a Grid message.
2658               // (Increment the count of all Grid messages.)
2659               if (gridSrcCluster != gridDestCluster) {
2660                 obj2->msg_count_grid += 1;
2661               }
2662
2663               // If the number of messages exceeds the interval, check to see if the object has
2664               // sent enough cross-cluster messages to qualify as a border object.
2665               if (obj2->msg_count >= obj2->grid_queue_interval) {
2666                 if (obj2->msg_count_grid >= obj2->grid_queue_threshold) {
2667                   // The object is a border object; if it is not already registered, register it.
2668                   if (!obj2->border_flag) {
2669                     CmiGridQueueRegister (gid.idx, obj2->thisIndexMax.nInts, data[0], data[1], data[2]);
2670                   }
2671                   obj2->border_flag = 1;
2672                 } else {
2673                   // The object is not a border object; if it is registered, deregister it.
2674                   if (obj2->border_flag) {
2675                     CmiGridQueueDeregister (gid.idx, obj2->thisIndexMax.nInts, data[0], data[1], data[2]);
2676                   }
2677                   obj2->border_flag = 0;
2678                 }
2679                 // Reset the counts.
2680                 obj2->msg_count = 0;
2681                 obj2->msg_count_grid = 0;
2682               }
2683             } else {
2684               if (gridSrcCluster != gridDestCluster) {
2685                 CmiGridQueueRegister (gid.idx, obj2->thisIndexMax.nInts, data[0], data[1], data[2]);
2686               }
2687             }
2688           }
2689
2690           // Reset the CkGridObject pointer.
2691           CpvAccess(CkGridObject) = NULL;
2692         }
2693 #endif
2694         /**FAULT_EVAC*/
2695         if (rec!=NULL){
2696                 CmiBool result = rec->deliver(msg,type,opts);
2697                 // if result is CmiFalse, than rec is not valid anymore, as the object
2698                 // the message was just delivered to has died or migrated out.
2699                 // Therefore rec->type() cannot be invoked!
2700                 if (result==CmiTrue && rec->type()==CkLocRec::local) return 0;
2701                 else return 1;
2702                 /*if(!result){
2703                         //DEBS((AA"deliver %s failed type %d \n"AB,idx2str(idx),rec->type()));
2704                         DEBS((AA"deliver %s failed \n"AB,idx2str(idx)));
2705                         if(rec->type() == CkLocRec::remote){
2706                                 if (opts & CK_MSG_KEEP)
2707                                         msg = (CkArrayMessage *)CkCopyMsg((void **)&msg);
2708                                 deliverUnknown(msg,type);
2709                         }
2710                 }*/
2711         }else /* rec==NULL*/ {
2712                 if (opts & CK_MSG_KEEP)
2713                         msg = (CkArrayMessage *)CkCopyMsg((void **)&msg);
2714                 deliverUnknown(msg,type,opts);
2715                 return 1;
2716         }
2717
2718 }
2719
2720 /// This index is not hashed-- somehow figure out what to do.
2721 CmiBool CkLocMgr::deliverUnknown(CkArrayMessage *msg,CkDeliver_t type,int opts)
2722 {
2723         CK_MAGICNUMBER_CHECK
2724         const CkArrayIndex &idx=msg->array_index();
2725         int onPe=homePe(idx);
2726         if (onPe!=CkMyPe()) 
2727         {// Forward the message to its home processor
2728                 DEBM((AA"Forwarding message for unknown %s to home %d \n"AB,idx2str(idx),onPe));
2729                 msg->array_hops()++;
2730                 CkArrayManagerDeliver(onPe,msg,opts);
2731                 return CmiTrue;
2732         }
2733         else
2734         { // We *are* the home processor:
2735         //Check if the element's array manager has been registered yet:
2736           CkArrMgr *mgr=managers.find(UsrToEnv((void *)msg)->getsetArrayMgr())->mgr;
2737           if (!mgr) { //No manager yet-- postpone the message (stupidly)
2738             if (CkInRestarting()) {
2739               // during restarting, this message should be ignored
2740               delete msg;
2741             }
2742             else {
2743               CkArrayManagerDeliver(CkMyPe(),msg); 
2744             }
2745           }
2746           else { // Has a manager-- must buffer the message
2747             DEBC((AA"Adding buffer for unknown element %s\n"AB,idx2str(idx)));
2748             CkLocRec *rec=new CkLocRec_buffering(this);
2749             insertRecN(rec,idx);
2750             rec->deliver(msg,type);
2751           
2752             if (msg->array_ifNotThere()!=CkArray_IfNotThere_buffer) 
2753             { //Demand-create the element:
2754               return demandCreateElement(msg,-1,type);
2755             }
2756           }
2757           return CmiTrue;
2758         }
2759 }
2760
2761 CmiBool CkLocMgr::demandCreateElement(CkArrayMessage *msg,int onPe,CkDeliver_t type)
2762 {
2763         CK_MAGICNUMBER_CHECK
2764         const CkArrayIndex &idx=msg->array_index();
2765         int chareType=_entryTable[msg->array_ep()]->chareIdx;
2766         int ctor=_chareTable[chareType]->getDefaultCtor();
2767         if (ctor==-1) CkAbort("Can't create array element to handle message--\n"
2768                               "The element has no default constructor in the .ci file!\n");
2769         if (onPe==-1) 
2770         { //Decide where element needs to live
2771                 if (msg->array_ifNotThere()==CkArray_IfNotThere_createhere) 
2772                         onPe=UsrToEnv(msg)->getsetArraySrcPe();
2773                 else //Createhome
2774                         onPe=homePe(idx);
2775         }
2776         
2777         //Find the manager and build the element
2778         DEBC((AA"Demand-creating element %s on pe %d\n"AB,idx2str(idx),onPe));
2779         CkArrMgr *mgr=managers.find(UsrToEnv((void *)msg)->getsetArrayMgr())->mgr;
2780         if (!mgr) CkAbort("Tried to demand-create for nonexistent arrMgr");
2781         return mgr->demandCreateElement(idx,onPe,ctor,type);
2782 }
2783
2784 //This message took several hops to reach us-- fix it
2785 void CkLocMgr::multiHop(CkArrayMessage *msg)
2786 {
2787
2788         CK_MAGICNUMBER_CHECK
2789         int srcPe=msg->array_getSrcPe();
2790         if (srcPe==CkMyPe())
2791                 DEB((AA"Odd routing: local element %s is %d hops away!\n"AB,idx2str(msg),msg->array_hops()));
2792         else
2793         {//Send a routing message letting original sender know new element location
2794                 DEBS((AA"Sending update back to %d for element\n"AB,srcPe,idx2str(msg)));
2795                 thisProxy[srcPe].updateLocation(msg->array_index(),CkMyPe());
2796         }
2797 }
2798
2799 /************************** LocMgr: ITERATOR *************************/
2800 CkLocation::CkLocation(CkLocMgr *mgr_, CkLocRec_local *rec_)
2801         :mgr(mgr_), rec(rec_) {}
2802         
2803 const CkArrayIndex &CkLocation::getIndex(void) const {
2804         return rec->getIndex();
2805 }
2806
2807 void CkLocation::destroyAll() {
2808         mgr->callMethod(rec, &CkMigratable::ckDestroy);
2809 }
2810
2811 void CkLocation::pup(PUP::er &p) {
2812         mgr->pupElementsFor(p,rec,CkElementCreation_migrate);
2813 }
2814
2815 CkLocIterator::~CkLocIterator() {}
2816
2817 /// Iterate over our local elements:
2818 void CkLocMgr::iterate(CkLocIterator &dest) {
2819   //Poke through the hash table for local ArrayRecs.
2820   void *objp;
2821   CkHashtableIterator *it=hash.iterator();
2822   CmiImmediateLock(hashImmLock);
2823
2824   while (NULL!=(objp=it->next())) {
2825     CkLocRec *rec=*(CkLocRec **)objp;
2826     if (rec->type()==CkLocRec::local) {
2827       CkLocation loc(this,(CkLocRec_local *)rec);
2828       dest.addLocation(loc);
2829     }
2830   }
2831   CmiImmediateUnlock(hashImmLock);
2832   delete it;
2833 }
2834
2835
2836
2837
2838 /************************** LocMgr: MIGRATION *************************/
2839 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
2840 void CkLocMgr::pupElementsFor(PUP::er &p,CkLocRec_local *rec,
2841         CkElementCreation_t type, CmiBool create, int dummy)
2842 {
2843     p.comment("-------- Array Location --------");
2844     register ManagerRec *m;
2845     int localIdx=rec->getLocalIndex();
2846     CkVec<CkMigratable *> dummyElts;
2847     for (m=firstManager;m!=NULL;m=m->next) {
2848         int elCType;
2849         if (!p.isUnpacking())
2850         { //Need to find the element's existing type
2851             CkMigratable *elt=m->element(localIdx);
2852             if (elt) elCType=elt->ckGetChareType();
2853             else elCType=-1; //Element hasn't been created
2854         }
2855         p(elCType);
2856         if (p.isUnpacking() && elCType!=-1) {
2857             CkMigratable *elt=m->mgr->allocateMigrated(elCType,rec->getIndex(),type);
2858             int migCtorIdx=_chareTable[elCType]->getMigCtor();
2859                         if(!dummy){
2860                                 if(create)
2861                                 if (!addElementToRec(rec,m,elt,migCtorIdx,NULL)) return;
2862                                 }else{
2863                     CkMigratable_initInfo &i=CkpvAccess(mig_initInfo);
2864                     i.locRec=rec;
2865                     i.chareType=_entryTable[migCtorIdx]->chareIdx;
2866                     dummyElts.push_back(elt);
2867                     if (!rec->invokeEntry(elt,NULL,migCtorIdx,CmiTrue)) return ;
2868                 }
2869         }
2870         }
2871     if(!dummy){
2872         for (m=firstManager;m!=NULL;m=m->next) {
2873             CkMigratable *elt=m->element(localIdx);
2874             if (elt!=NULL)
2875                 {
2876                        elt->pup(p);
2877                 }
2878         }
2879     }else{
2880             for(int i=0;i<dummyElts.size();i++){
2881                 CkMigratable *elt = dummyElts[i];
2882                 if (elt!=NULL){
2883             elt->pup(p);
2884                         }
2885                 delete elt;
2886             }
2887                         for (ManagerRec *m=firstManager;m!=NULL;m=m->next) {
2888                 m->elts.empty(localIdx);
2889             }
2890         freeList[localIdx]=firstFree;
2891         firstFree=localIdx;
2892     }
2893 }
2894 #else
2895 void CkLocMgr::pupElementsFor(PUP::er &p,CkLocRec_local *rec,
2896                 CkElementCreation_t type,CmiBool rebuild,CmiBool create)
2897 {
2898         p.comment("-------- Array Location --------");
2899         register ManagerRec *m;
2900         int localIdx=rec->getLocalIndex();
2901
2902         //First pup the element types
2903         // (A separate loop so ckLocal works even in element pup routines)
2904         for (m=firstManager;m!=NULL;m=m->next) {
2905                 int elCType;
2906                 if (!p.isUnpacking()&&!p.isChecking())
2907                 { //Need to find the element's existing type
2908                         CkMigratable *elt=m->element(localIdx);
2909                         if (elt) elCType=elt->ckGetChareType();
2910                         else elCType=-1; //Element hasn't been created
2911                 }
2912
2913                 p(elCType);
2914                 if (p.isUnpacking() && elCType!=-1) {
2915                         //Create the element
2916                         CkMigratable *elt=m->mgr->allocateMigrated(elCType,rec->getIndex(),type);
2917                         int migCtorIdx=_chareTable[elCType]->getMigCtor();
2918                         //Insert into our tables and call migration constructor
2919                         if(create)
2920                                 if (!addElementToRec(rec,m,elt,migCtorIdx,NULL)) return;
2921                 }
2922         }
2923
2924         //Next pup the element data
2925         for (m=firstManager;m!=NULL;m=m->next) {
2926                 CkMigratable *elt=m->element(localIdx);
2927                 if (elt!=NULL)
2928                 {
2929                         elt->pup(p);
2930 #if CMK_ERROR_CHECKING
2931                         if (p.isUnpacking()) elt->sanitycheck();
2932 #endif
2933                 }
2934         }
2935
2936 #if CMK_MEM_CHECKPOINT
2937         if(rebuild){
2938           ArrayElement *elt;
2939           CkVec<CkMigratable *> list;
2940           migratableList(rec, list);
2941           CmiAssert(list.length() > 0);
2942           for (int l=0; l<list.length(); l++) {
2943                 //    reset, may not needed now
2944                 // for now.
2945                 for (int i=0; i<CK_ARRAYLISTENER_MAXLEN; i++) {
2946                         ArrayElement * elt = (ArrayElement *)list[l];
2947                   contributorInfo *c=(contributorInfo *)&elt->listenerData[i];
2948                   if (c) c->redNo = 0;
2949                 }
2950           }
2951         }
2952 #endif
2953 }
2954 #endif
2955
2956 /// Call this member function on each element of this location:
2957 void CkLocMgr::callMethod(CkLocRec_local *rec,CkMigratable_voidfn_t fn)
2958 {
2959         int localIdx=rec->getLocalIndex();
2960         for (ManagerRec *m=firstManager;m!=NULL;m=m->next) {
2961                 CkMigratable *el=m->element(localIdx);
2962                 if (el) (el->* fn)();
2963         }
2964 }
2965
2966 /// Call this member function on each element of this location:
2967 void CkLocMgr::callMethod(CkLocRec_local *rec,CkMigratable_voidfn_arg_t fn,     void * data)
2968 {
2969         int localIdx=rec->getLocalIndex();
2970         for (ManagerRec *m=firstManager;m!=NULL;m=m->next) {
2971                 CkMigratable *el=m->element(localIdx);
2972                 if (el) (el->* fn)(data);
2973         }
2974 }
2975
2976 /// return a list of migratables in this local record
2977 void CkLocMgr::migratableList(CkLocRec_local *rec, CkVec<CkMigratable *> &list)
2978 {
2979         register ManagerRec *m;
2980         int localIdx=rec->getLocalIndex();
2981
2982         for (m=firstManager;m!=NULL;m=m->next) {
2983                 CkMigratable *elt=m->element(localIdx);
2984                 if (elt) list.push_back(elt);
2985         }
2986 }
2987
2988 /// Migrate this local element away to another processor.
2989 void CkLocMgr::emigrate(CkLocRec_local *rec,int toPe)
2990 {
2991         CK_MAGICNUMBER_CHECK
2992         if (toPe==CkMyPe()) return; //You're already there!
2993         /*
2994                 FAULT_EVAC
2995                 if the toProcessor is already marked as invalid, dont emigrate
2996                 Shouldn't happen but might
2997         */
2998         if(!CmiNodeAlive(toPe)){
2999                 return;
3000         }
3001         CkArrayIndex idx=rec->getIndex();
3002
3003 #if CMK_OUT_OF_CORE
3004         int localIdx=rec->getLocalIndex();
3005         /* Load in any elements that are out-of-core */
3006         for (ManagerRec *m=firstManager;m!=NULL;m=m->next) {
3007                 CkMigratable *el=m->element(localIdx);
3008                 if (el) if (!el->isInCore) CooBringIn(el->prefetchObjID);
3009         }
3010 #endif
3011
3012         //Let all the elements know we're leaving
3013         callMethod(rec,&CkMigratable::ckAboutToMigrate);
3014         /*EVAC*/
3015
3016 //First pass: find size of migration message
3017         int bufSize;
3018         {
3019                 PUP::sizer p;
3020                 p(nManagers);
3021                 pupElementsFor(p,rec,CkElementCreation_migrate);
3022                 bufSize=p.size(); 
3023         }
3024
3025 //Allocate and pack into message
3026         int doubleSize=bufSize/sizeof(double)+1;
3027         CkArrayElementMigrateMessage *msg = 
3028                 new (doubleSize, 0) CkArrayElementMigrateMessage;
3029         msg->idx=idx;
3030         msg->length=bufSize;
3031 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_)) 
3032     msg->gid = ckGetGroupID();
3033 #endif
3034 #if CMK_LBDB_ON
3035         msg->ignoreArrival = rec->isAsyncMigrate()?1:0;
3036 #endif
3037         /*
3038                 FAULT_EVAC
3039         */
3040         msg->bounced = rec->isBounced();
3041         {
3042                 PUP::toMem p(msg->packData); 
3043                 p.becomeDeleting(); 
3044                 p(nManagers);
3045                 pupElementsFor(p,rec,CkElementCreation_migrate);
3046                 if (p.size()!=bufSize) {
3047                         CkError("ERROR! Array element claimed it was %d bytes to a "
3048                                 "sizing PUP::er, but copied %d bytes into the packing PUP::er!\n",
3049                                 bufSize,p.size());
3050                         CkAbort("Array element's pup routine has a direction mismatch.\n");
3051                 }
3052         }
3053
3054         DEBM((AA"Migrated index size %s to %d \n"AB,idx2str(idx),toPe));        
3055
3056 //#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
3057 //#if defined(_FAULT_MLOG_)
3058 //    sendMlogLocation(toPe,UsrToEnv(msg));
3059 //#else
3060         //Send off message and delete old copy
3061         thisProxy[toPe].immigrate(msg);
3062 //#endif
3063
3064         duringMigration=CmiTrue;
3065         delete rec; //Removes elements, hashtable entries, local index
3066         
3067         
3068         duringMigration=CmiFalse;
3069         //The element now lives on another processor-- tell ourselves and its home
3070         inform(idx,toPe);
3071 //#if (!defined(_FAULT_MLOG_) && !defined(_FAULT_CAUSAL_))    
3072 //#if !defined(_FAULT_MLOG_)    
3073         informHome(idx,toPe);
3074 //#endif
3075
3076 #if !CMK_LBDB_ON && CMK_GLOBAL_LOCATION_UPDATE
3077         DEBM((AA"Global location update. idx %s " 
3078               "assigned to %d \n"AB,idx2str(idx),toPe));
3079         thisProxy.updateLocation(idx, toPe);                        
3080 #endif
3081
3082         CK_MAGICNUMBER_CHECK
3083 }
3084
3085 #if CMK_LBDB_ON
3086 void CkLocMgr::informLBPeriod(CkLocRec_local *rec, int lb_ideal_period) {
3087         callMethod(rec,&CkMigratable::recvLBPeriod, (void *)&lb_ideal_period);
3088 }
3089
3090 void CkLocMgr::metaLBCallLB(CkLocRec_local *rec) {
3091         callMethod(rec, &CkMigratable::metaLBCallLB);
3092 }
3093 #endif
3094
3095 /**
3096   Migrating array element is arriving on this processor.
3097 */
3098 void CkLocMgr::immigrate(CkArrayElementMigrateMessage *msg)
3099 {
3100         const CkArrayIndex &idx=msg->idx;
3101                 
3102         PUP::fromMem p(msg->packData); 
3103         
3104         int nMsgMan;
3105         p(nMsgMan);
3106         if (nMsgMan<nManagers)
3107                 CkAbort("Array element arrived from location with fewer managers!\n");
3108         if (nMsgMan>nManagers) {
3109                 //Some array managers haven't registered yet-- throw it back
3110                 DEBM((AA"Busy-waiting for array registration on migrating %s\n"AB,idx2str(idx)));
3111                 thisProxy[CkMyPe()].immigrate(msg);
3112                 return;
3113         }
3114
3115         //Create a record for this element
3116 //#if (!defined(_FAULT_MLOG_) && !defined(_FAULT_CAUSAL_))    
3117 //#if !defined(_FAULT_MLOG_)     
3118         CkLocRec_local *rec=createLocal(idx,CmiTrue,msg->ignoreArrival,CmiFalse /* home told on departure */ );
3119 //#else
3120 //    CkLocRec_local *rec=createLocal(idx,CmiTrue,CmiTrue,CmiFalse /* home told on departure */ );
3121 //#endif
3122         
3123         //Create the new elements as we unpack the message
3124         pupElementsFor(p,rec,CkElementCreation_migrate);
3125         if (p.size()!=msg->length) {
3126                 CkError("ERROR! Array element claimed it was %d bytes to a"
3127                         "packing PUP::er, but %d bytes in the unpacking PUP::er!\n",
3128                         msg->length,p.size());
3129                 CkError("(I have %d managers; he claims %d managers)\n",
3130                         nManagers,nMsgMan);
3131                 
3132                 CkAbort("Array element's pup routine has a direction mismatch.\n");
3133         }
3134         /*
3135                 FAULT_EVAC
3136                         if this element came in as a result of being bounced off some other process,
3137                         then it needs to be resumed. It is assumed that it was bounced because load 
3138                         balancing caused it to move into a processor which later crashed
3139         */
3140         if(msg->bounced){
3141                 callMethod(rec,&CkMigratable::ResumeFromSync);
3142         }
3143         
3144         //Let all the elements know we've arrived
3145         callMethod(rec,&CkMigratable::ckJustMigrated);
3146         /*FAULT_EVAC
3147                 If this processor has started evacuating array elements on it 
3148                 dont let new immigrants in. If they arrive send them to what
3149                 would be their correct homePE.
3150                 Leave a record here mentioning the processor where it got sent
3151         */
3152         
3153         if(CkpvAccess(startedEvac)){
3154                 int newhomePE = getNextPE(idx);
3155                 DEBM((AA"Migrated into failed processor index size %s resent to %d \n"AB,idx2str(idx),newhomePE));      
3156                 CkLocMgr *mgr = rec->getLocMgr();
3157                 int targetPE=getNextPE(idx);
3158                 //set this flag so that load balancer is not informed when
3159                 //this element migrates
3160                 rec->AsyncMigrate(CmiTrue);
3161                 rec->Bounced(CmiTrue);
3162                 mgr->emigrate(rec,targetPE);
3163                 
3164         }
3165
3166         delete msg;
3167 }
3168
3169 void CkLocMgr::restore(const CkArrayIndex &idx, PUP::er &p)
3170 {
3171         //This is in broughtIntoMem during out-of-core emulation in BigSim,
3172         //informHome should not be called since such information is already
3173         //immediately updated real migration
3174 #if CMK_ERROR_CHECKING
3175         if(_BgOutOfCoreFlag!=2)
3176             CmiAbort("CkLocMgr::restore should only be used in out-of-core emulation for BigSim and be called when object is brought into memory!\n");
3177 #endif
3178         CkLocRec_local *rec=createLocal(idx,CmiFalse,CmiFalse,CmiFalse);
3179         
3180         //BIGSIM_OOC DEBUGGING
3181         //CkPrintf("Proc[%d]: Registering element %s with LDB\n", CkMyPe(), idx2str(idx));
3182
3183         //Create the new elements as we unpack the message
3184         pupElementsFor(p,rec,CkElementCreation_restore);
3185
3186         callMethod(rec,&CkMigratable::ckJustRestored);
3187 }
3188
3189
3190 /// Insert and unpack this array element from this checkpoint (e.g., from CkLocation::pup)
3191 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
3192 void CkLocMgr::resume(const CkArrayIndex &idx, PUP::er &p, CmiBool create, int dummy)
3193 {
3194         CkLocRec_local *rec;
3195         CkLocRec *recGlobal;    
3196
3197         if(create){
3198                 rec = createLocal(idx,CmiFalse,CmiFalse,CmiTrue && !dummy /* home doesn't know yet */,dummy );
3199         }else{
3200                 recGlobal = elementNrec(idx);
3201                 if(recGlobal == NULL) 
3202                         CmiAbort("Local object not found");
3203                 if(recGlobal->type() != CkLocRec::local)
3204                         CmiAbort("Local object not local, :P");
3205                 rec = (CkLocRec_local *)recGlobal;
3206         }
3207         
3208     pupElementsFor(p,rec,CkElementCreation_resume,create,dummy);
3209
3210     if(!dummy){
3211         callMethod(rec,&CkMigratable::ckJustMigrated);
3212     }
3213 }
3214 #else
3215 void CkLocMgr::resume(const CkArrayIndex &idx, PUP::er &p, CmiBool notify,CmiBool rebuild,CmiBool create)
3216 {
3217         CkLocRec_local *rec;
3218         CkLocRec *recGlobal;    
3219
3220         if(create){
3221                 //Create the new elements as we unpack the message
3222                 rec=createLocal(idx,CmiFalse,CmiFalse,notify /* home doesn't know yet */ );
3223         }else{
3224                 recGlobal = elementNrec(idx);
3225                 if(recGlobal == NULL) 
3226                         CmiAbort("Local object not found");
3227                 if(recGlobal->type() != CkLocRec::local)
3228                         CmiAbort("Local object not local, :P");
3229                 rec = (CkLocRec_local *)recGlobal;
3230         }
3231         pupElementsFor(p,rec,CkElementCreation_resume,rebuild,create);
3232         if(!p.isChecking())
3233                 callMethod(rec,&CkMigratable::ckJustMigrated);
3234 }
3235 #endif
3236
3237 /********************* LocMgr: UTILITY ****************/
3238 void CkMagicNumber_impl::badMagicNumber(
3239         int expected,const char *file,int line,void *obj) const
3240 {
3241         CkError("FAILURE on pe %d, %s:%d> Expected %p's magic number "
3242                 "to be 0x%08x; but found 0x%08x!\n", CkMyPe(),file,line,obj,
3243                 expected, magic);
3244         CkAbort("Bad magic number detected!  This implies either\n"
3245                 "the heap or a message was corrupted!\n");
3246 }
3247 CkMagicNumber_impl::CkMagicNumber_impl(int m) :magic(m) { }
3248
3249 //Look up the object with this array index, or return NULL
3250 CkMigratable *CkLocMgr::lookup(const CkArrayIndex &idx,CkArrayID aid) {
3251         CkLocRec *rec=elementNrec(idx);
3252         if (rec==NULL) return NULL;
3253         else return rec->lookupElement(aid);
3254 }
3255 //"Last-known" location (returns a processor number)
3256 int CkLocMgr::lastKnown(const CkArrayIndex &idx) {
3257         CkLocMgr *vthis=(CkLocMgr *)this;//Cast away "const"
3258         CkLocRec *rec=vthis->elementNrec(idx);
3259         int pe=-1;
3260         if (rec!=NULL) pe=rec->lookupProcessor();
3261         if (pe==-1) return homePe(idx);
3262         else{
3263                 /*
3264                         FAULT_EVAC
3265                         if the lastKnownPE is invalid return homePE and delete this record
3266                 */
3267                 if(!CmiNodeAlive(pe)){
3268                         removeFromTable(idx);
3269                         return homePe(idx);
3270                 }
3271                 return pe;
3272         }       
3273 }
3274 /// Return true if this array element lives on another processor
3275 bool CkLocMgr::isRemote(const CkArrayIndex &idx,int *onPe) const
3276 {
3277         CkLocMgr *vthis=(CkLocMgr *)this;//Cast away "const"
3278         CkLocRec *rec=vthis->elementNrec(idx);
3279         if (rec==NULL || rec->type()!=CkLocRec::remote) 
3280                 return false; /* not definitely a remote element */
3281         else /* element is indeed remote */
3282         {
3283                 *onPe=rec->lookupProcessor();
3284                 return true;
3285         }
3286 }
3287
3288 static const char *rec2str[]={
3289     "base (INVALID)",//Base class (invalid type)
3290     "local",//Array element that lives on this Pe
3291     "remote",//Array element that lives on some other Pe
3292     "buffering",//Array element that was just created
3293     "dead"//Deleted element (for debugging)
3294 };
3295
3296 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
3297 void CkLocMgr::setDuringMigration(CmiBool _duringMigration){
3298     duringMigration = _duringMigration;
3299 }
3300 #endif
3301
3302
3303 //Add given element array record at idx, replacing the existing record
3304 void CkLocMgr::insertRec(CkLocRec *rec,const CkArrayIndex &idx) {
3305         CkLocRec *old=elementNrec(idx);
3306         insertRecN(rec,idx);
3307         if (old!=NULL) {
3308                 DEBC((AA"  replaces old rec(%s) for %s\n"AB,rec2str[old->type()],idx2str(idx)));
3309                 //There was an old element at this location
3310                 if (old->type()==CkLocRec::local && rec->type()==CkLocRec::local) {
3311                     if (!CkInRestarting()) {    // ok if it is restarting
3312                         CkPrintf("ERROR! Duplicate array index: %s\n",idx2str(idx));
3313                         CkAbort("Duplicate array index used");
3314                     }
3315                 }
3316                 old->beenReplaced();
3317                 delete old;
3318         }
3319 }
3320
3321 //Add given record, when there is guarenteed to be no prior record
3322 void CkLocMgr::insertRecN(CkLocRec *rec,const CkArrayIndex &idx) {
3323         DEBC((AA"  adding new rec(%s) for %s\n"AB,rec2str[rec->type()],idx2str(idx)));
3324         CmiImmediateLock(hashImmLock);
3325         hash.put(*(CkArrayIndex *)&idx)=rec;
3326         CmiImmediateUnlock(hashImmLock);
3327 }
3328
3329 //Call this on an unrecognized array index
3330 static void abort_out_of_bounds(const CkArrayIndex &idx)
3331 {
3332   CkPrintf("ERROR! Unknown array index: %s\n",idx2str(idx));
3333   CkAbort("Array index out of bounds\n");
3334 }
3335
3336 //Look up array element in hash table.  Index out-of-bounds if not found.
3337 CkLocRec *CkLocMgr::elementRec(const CkArrayIndex &idx) {
3338 #if ! CMK_ERROR_CHECKING
3339 //Assume the element will be found
3340         return hash.getRef(*(CkArrayIndex *)&idx);
3341 #else
3342 //Include an out-of-bounds check if the element isn't found
3343         CkLocRec *rec=elementNrec(idx);
3344         if (rec==NULL) abort_out_of_bounds(idx);
3345         return rec;
3346 #endif
3347 }
3348
3349 //Look up array element in hash table.  Return NULL if not there.
3350 CkLocRec *CkLocMgr::elementNrec(const CkArrayIndex &idx) {
3351         return hash.get(*(CkArrayIndex *)&idx);
3352 }
3353
3354 struct LocalElementCounter :  public CkLocIterator
3355 {
3356     unsigned int count;
3357     LocalElementCounter() : count(0) {}
3358     void addLocation(CkLocation &loc)
3359         { ++count; }
3360 };
3361
3362 unsigned int CkLocMgr::numLocalElements()
3363 {
3364     LocalElementCounter c;
3365     iterate(c);
3366     return c.count;
3367 }
3368
3369
3370 /********************* LocMgr: LOAD BALANCE ****************/
3371
3372 #if !CMK_LBDB_ON
3373 //Empty versions of all load balancer calls
3374 void CkLocMgr::initLB(CkGroupID lbdbID_, CkGroupID metalbID_) {}
3375 void CkLocMgr::startInserting(void) {}
3376 void CkLocMgr::doneInserting(void) {}
3377 void CkLocMgr::dummyAtSync(void) {}
3378 #endif
3379
3380
3381 #if CMK_LBDB_ON
3382 void CkLocMgr::initLB(CkGroupID lbdbID_, CkGroupID metalbID_)
3383 { //Find and register with the load balancer
3384         the_lbdb = (LBDatabase *)CkLocalBranch(lbdbID_);
3385         if (the_lbdb == 0)
3386                 CkAbort("LBDatabase not yet created?\n");
3387         DEBL((AA"Connected to load balancer %p\n"AB,the_lbdb));
3388         the_metalb = (MetaBalancer *)CkLocalBranch(metalbID_);
3389         if (the_metalb == 0)
3390                 CkAbort("MetaBalancer not yet created?\n");
3391
3392         // Register myself as an object manager
3393         LDOMid myId;
3394         myId.id = thisgroup;
3395         LDCallbacks myCallbacks;
3396         myCallbacks.migrate = (LDMigrateFn)CkLocRec_local::staticMigrate;
3397         myCallbacks.setStats = NULL;
3398         myCallbacks.queryEstLoad = NULL;
3399   myCallbacks.metaLBResumeWaitingChares =
3400       (LDMetaLBResumeWaitingCharesFn)CkLocRec_local::staticMetaLBResumeWaitingChares;
3401   myCallbacks.metaLBCallLBOnChares =
3402       (LDMetaLBCallLBOnCharesFn)CkLocRec_local::staticMetaLBCallLBOnChares;
3403         myLBHandle = the_lbdb->RegisterOM(myId,this,myCallbacks);
3404
3405         // Tell the lbdb that I'm registering objects
3406         the_lbdb->RegisteringObjects(myLBHandle);
3407
3408         /*Set up the dummy barrier-- the load balancer needs
3409           us to call Registering/DoneRegistering during each AtSync,
3410           and this is the only way to do so.
3411         */
3412         the_lbdb->AddLocalBarrierReceiver(
3413                 (LDBarrierFn)staticRecvAtSync,(void*)(this));
3414         dummyBarrierHandle = the_lbdb->AddLocalBarrierClient(
3415                 (LDResumeFn)staticDummyResumeFromSync,(void*)(this));
3416         dummyAtSync();
3417 }
3418 void CkLocMgr::dummyAtSync(void)
3419 {
3420         DEBL((AA"dummyAtSync called\n"AB));
3421         the_lbdb->AtLocalBarrier(dummyBarrierHandle);
3422 }
3423
3424 void CkLocMgr::staticDummyResumeFromSync(void* data)
3425 {      ((CkLocMgr*)data)->dummyResumeFromSync(); }
3426 void CkLocMgr::dummyResumeFromSync()
3427 {
3428         DEBL((AA"DummyResumeFromSync called\n"AB));
3429         the_lbdb->DoneRegisteringObjects(myLBHandle);
3430         dummyAtSync();
3431 }
3432 void CkLocMgr::staticRecvAtSync(void* data)
3433 {      ((CkLocMgr*)data)->recvAtSync(); }
3434 void CkLocMgr::recvAtSync()
3435 {
3436         DEBL((AA"recvAtSync called\n"AB));
3437         the_lbdb->RegisteringObjects(myLBHandle);
3438 }
3439
3440 void CkLocMgr::startInserting(void)
3441 {
3442         the_lbdb->RegisteringObjects(myLBHandle);
3443 }
3444 void CkLocMgr::doneInserting(void)
3445 {
3446         the_lbdb->DoneRegisteringObjects(myLBHandle);
3447 }
3448 #endif
3449
3450 #include "CkLocation.def.h"
3451
3452