Pup numZerocopyROops in RODataMsg after recovering from a failure
[charm.git] / src / ck-core / ckcheckpoint.C
1 /*
2 Charm++ File: Checkpoint Library
3 added 01/03/2003 by Chao Huang, chuang10@uiuc.edu
4
5 More documentation goes here...
6 --- Updated 12/14/2003 by Gengbin, gzheng@uiuc.edu
7     see ckcheckpoint.h for change log
8 */
9
10 #include <stdio.h>
11 #include <stdlib.h>
12 #ifndef _WIN32
13 #include <unistd.h>
14 #endif
15 #include <string.h>
16 #include <sstream>
17 using std::ostringstream;
18 #include <errno.h>
19 #include "charm++.h"
20 #include "ck.h"
21 #include "ckcheckpoint.h"
22 #include "CkCheckpoint.decl.h"
23
24 void noopit(const char*, ...)
25 {}
26
27 //#define DEBCHK   CkPrintf
28 #define DEBCHK noopit
29
30 #define DEBUGC(x) x
31 //#define DEBUGC(x) 
32
33 CkGroupID _sysChkptMgr;
34
35 typedef struct _GroupInfo{
36         CkGroupID gID;
37         int MigCtor;
38         char name[256];
39         bool present;
40 } GroupInfo;
41 PUPbytes(GroupInfo)
42 PUPmarshall(GroupInfo)
43
44 bool _inrestart = false;
45 bool _restarted = false;
46 int _oldNumPes = 0;
47 bool _chareRestored = false;
48 double chkptStartTimer = 0;
49 #if CMK_SHRINK_EXPAND
50 int originalnumGroups = -1;
51 extern int Cmi_isOldProcess;
52 extern int Cmi_myoldpe;
53 extern char *_shrinkexpand_basedir;
54 #endif
55
56 #if CMK_ONESIDED_IMPL
57 extern UInt numZerocopyROops; // Required for broadcasting RO Data after recovering from failure
58 #endif
59
60 void CkCreateLocalChare(int epIdx, envelope *env);
61
62 // helper class to get number of array elements
63 class ElementCounter : public CkLocIterator {
64 private:
65         int count;
66 public:
67         ElementCounter():count(0){};
68         void addLocation(CkLocation &loc)  { count++; }
69         int getCount() { return count; }
70 };
71
72 // helper class to pup all elements that belong to same ckLocMgr
73 class ElementCheckpointer : public CkLocIterator {
74 private:
75         CkLocMgr *locMgr;
76         PUP::er &p;
77 public:
78         ElementCheckpointer(CkLocMgr* mgr_, PUP::er &p_):locMgr(mgr_),p(p_){};
79         void addLocation(CkLocation &loc) {
80                 CkArrayIndex idx=loc.getIndex();
81                 CkGroupID gID = locMgr->ckGetGroupID();
82                 CmiUInt8 id = loc.getID();
83                 p|gID;      // store loc mgr's GID as well for easier restore
84                 p|idx;
85                 p|id;
86                 p|loc;
87                 //CkPrintf("[%d] addLocation: ", CkMyPe()), idx.print();
88         }
89 };
90
91
92 extern void _initDone();
93
94 static void bdcastRO(void){
95         int i;
96         // Determine the size of the RODataMessage
97         PUP::sizer ps;
98         for(i=0;i<_readonlyTable.size();i++) _readonlyTable[i]->pupData(ps);
99
100         // Allocate and fill out the RODataMessage
101         envelope *env = _allocEnv(RODataMsg, ps.size());
102         PUP::toMem pp((char *)EnvToUsr(env));
103 #if CMK_ONESIDED_IMPL
104         pp|numZerocopyROops; // Messages of type 'RODataMsg' need to have numZerocopyROops pupped in order
105                        // to be processed inside _processRODataMsg
106 #endif
107         for(i=0;i<_readonlyTable.size();i++) _readonlyTable[i]->pupData(pp);
108         
109         env->setCount(++_numInitMsgs);
110         env->setSrcPe(CkMyPe());
111         CmiSetHandler(env, _roRestartHandlerIdx);
112         CmiSyncBroadcastAndFree(env->getTotalsize(), (char *)env);
113 }
114
115 #if CMK_SHRINK_EXPAND
116 static void bdcastROGroupData(void){
117         int i;
118         //Determine the size of the RODataMessage
119         PUP::sizer ps, ps1;
120         CkPupROData(ps);
121         int ROSize = ps.size();
122
123         CkPupGroupData(ps1);
124         int GroupSize = ps1.size();
125
126         char *msg = (char *)CmiAlloc(CmiMsgHeaderSizeBytes + 2*sizeof(int) + ps.size() + ps1.size());
127         char *payloadOffset = msg + CmiMsgHeaderSizeBytes;
128
129         // how much data to send
130         *(int*)payloadOffset = ps.size();
131         payloadOffset += sizeof(int);
132         *(int*)payloadOffset = ps1.size();
133         payloadOffset += sizeof(int);
134
135         //Allocate and fill out the RODataMessage
136         PUP::toMem pp((char *)payloadOffset);
137         CkPupROData(pp);
138
139         CkPupGroupData(pp);
140
141         CmiSetHandler(msg, _ROGroupRestartHandlerIdx);
142         CmiSyncBroadcastAllAndFree(CmiMsgHeaderSizeBytes + 2*sizeof(int) + pp.size(), msg);
143 }
144 #endif
145
146 // Print out an array index to this string as decimal fields
147 // separated by underscores.
148 void printIndex(const CkArrayIndex &idx,char *dest) {
149         const int *idxData=idx.data();
150         for (int i=0;i<idx.nInts;i++) {
151                 sprintf(dest,"%s%d",i==0?"":"_", idxData[i]);
152                 dest+=strlen(dest);
153         }
154 }
155
156 static bool checkpointOne(const char* dirname, CkCallback& cb, bool requestStatus);
157
158 static void addPartitionDirectory(ostringstream &path) {
159         if (CmiNumPartitions() > 1) {
160           path << "/part-" << CmiMyPartition() << '/';
161         }
162 }
163
164 static FILE* openCheckpointFile(const char *dirname, const char *basename,
165                                 const char *mode, int id = -1) {
166         ostringstream out;
167         out << dirname << '/';
168         addPartitionDirectory(out);
169         out << basename;
170         if (id != -1)
171                 out << '_' << id;
172         out << ".dat";
173
174         FILE *fp = CmiFopen(out.str().c_str(), mode);
175         if (!fp) {
176                 ostringstream error;
177                 error << "PE " << CkMyPe() << " failed to open checkpoint file: " << out.str()
178                       << ", mode: " << mode << " status: " << strerror(errno);
179                 CkAbort(error.str().c_str());
180         }
181         return fp;
182 }
183
184 /**
185  * There is only one Checkpoint Manager in the whole system
186 **/
187 class CkCheckpointMgr : public CBase_CkCheckpointMgr {
188 private:
189         CkCallback restartCB;
190         double chkptStartTimer;
191         bool requestStatus;
192         int chkpStatus;
193 public:
194         CkCheckpointMgr() { }
195         CkCheckpointMgr(CkMigrateMessage *m):CBase_CkCheckpointMgr(m) { }
196         void Checkpoint(const char *dirname, CkCallback cb, bool requestStatus = false);
197         void SendRestartCB(void);
198         void pup(PUP::er& p){ p|restartCB; }
199 };
200
201 // broadcast
202 void CkCheckpointMgr::Checkpoint(const char *dirname, CkCallback cb, bool _requestStatus){
203         chkptStartTimer = CmiWallTimer();
204         requestStatus = _requestStatus;
205         // make dir on all PEs in case it is a local directory
206         CmiMkdir(dirname);
207         bool success = true;
208         if (CmiNumPartitions() > 1) {
209           ostringstream partDir;
210           partDir << dirname;
211           addPartitionDirectory(partDir);
212           CmiMkdir(partDir.str().c_str());
213         }
214
215         if (CkMyPe() == 0) {
216 #if CMK_SHRINK_EXPAND
217     if (pending_realloc_state == REALLOC_IN_PROGRESS) {
218       // After restarting from this AtSync checkpoint, resume execution along the
219       // normal path (i.e. whatever the user defined as ResumeFromSync.)
220       CkCallback resumeFromSyncCB(CkIndex_LBDatabase::ResumeClients(), _lbdb);
221       success &= checkpointOne(dirname, resumeFromSyncCB, requestStatus);
222     } else
223 #endif
224     {
225       success &= checkpointOne(dirname, cb, requestStatus);
226     }
227   }
228
229 #ifndef CMK_CHARE_USE_PTR
230         // save plain singleton chares into Chares.dat
231         FILE* fChares = openCheckpointFile(dirname, "Chares", "wb", CkMyPe());
232         PUP::toDisk pChares(fChares);
233         CkPupChareData(pChares);
234         if(pChares.checkError())
235           success = false;
236         if(CmiFclose(fChares)!=0)
237           success = false;
238 #endif
239
240         // save groups into Groups.dat
241         // content of the file: numGroups, GroupInfo[numGroups], _groupTable(PUP'ed), groups(PUP'ed)
242         FILE* fGroups = openCheckpointFile(dirname, "Groups", "wb", CkMyPe());
243         PUP::toDisk pGroups(fGroups);
244 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
245         CkPupGroupData(pGroups, true);
246 #else
247         CkPupGroupData(pGroups);
248 #endif
249         if(pGroups.checkError())
250           success = false;
251         if(CmiFclose(fGroups)!=0)
252           success = false;
253
254         // save nodegroups into NodeGroups.dat
255         // content of the file: numNodeGroups, GroupInfo[numNodeGroups], _nodeGroupTable(PUP'ed), nodegroups(PUP'ed)
256         if (CkMyRank() == 0) {
257           FILE* fNodeGroups = openCheckpointFile(dirname, "NodeGroups", "wb", CkMyNode());
258           PUP::toDisk pNodeGroups(fNodeGroups);
259 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
260           CkPupNodeGroupData(pNodeGroups, true);
261 #else
262           CkPupNodeGroupData(pNodeGroups);
263 #endif
264           if(pNodeGroups.checkError())
265             success = false;
266           if(CmiFclose(fNodeGroups)!=0)
267             success = false;
268         }
269
270         //DEBCHK("[%d]CkCheckpointMgr::Checkpoint called dirname={%s}\n",CkMyPe(),dirname);
271         FILE *datFile = openCheckpointFile(dirname, "arr", "wb", CkMyPe());
272         PUP::toDisk  p(datFile);
273         CkPupArrayElementsData(p);
274         if(p.checkError())
275           success = false;
276         if(CmiFclose(datFile)!=0)
277           success = false;
278
279 #if ! CMK_DISABLE_SYNC
280 #if CMK_HAS_SYNC_FUNC
281         sync();
282 #elif CMK_HAS_SYNC
283         system("sync");
284 #endif
285 #endif
286         chkpStatus = success?CK_CHECKPOINT_SUCCESS:CK_CHECKPOINT_FAILURE;
287         restartCB = cb;
288         DEBCHK("[%d]restartCB installed\n",CkMyPe());
289
290         // Use barrier instead of contribute here:
291         // barrier is stateless and multiple calls to it do not overlap.
292         barrier(CkCallback(CkReductionTarget(CkCheckpointMgr, SendRestartCB), 0, thisgroup));
293 }
294
295 void CkCheckpointMgr::SendRestartCB(void){
296         DEBCHK("[%d]Sending out the cb\n",CkMyPe());
297         CkPrintf("Checkpoint to disk finished in %fs, sending out the cb...\n", CmiWallTimer() - chkptStartTimer);
298         if(requestStatus)
299         {
300           CkCheckpointStatusMsg * m = new CkCheckpointStatusMsg(chkpStatus);
301           restartCB.send(m);
302         }
303         else
304           restartCB.send();
305 }
306
307 void CkPupROData(PUP::er &p)
308 {
309         int _numReadonlies = 0;
310         int _numReadonlyMsgs = 0;
311         if (!p.isUnpacking()) _numReadonlies=_readonlyTable.size();
312
313         p|_numReadonlies;
314
315         if (p.isUnpacking()) {
316           if (_numReadonlies != _readonlyTable.size())
317             CkAbort("You cannot add readonlies and restore from checkpoint...");
318         }
319         for(int i=0;i<_numReadonlies;i++) _readonlyTable[i]->pupData(p);
320         if (!p.isUnpacking()) _numReadonlyMsgs=_readonlyMsgs.size();
321         p|_numReadonlyMsgs;
322         for(int i=0;i<_numReadonlyMsgs; i++){
323                 ReadonlyMsgInfo *c = _readonlyMsgs[i];
324                 CkPupMessage(p,c->pMsg);
325         }
326 }
327
328 // handle main chare
329 void CkPupMainChareData(PUP::er &p, CkArgMsg *args)
330 {
331         int nMains=_mainTable.size();
332         DEBCHK("[%d] CkPupMainChareData %s: nMains = %d\n", CkMyPe(),p.typeString(),nMains);
333         for(int i=0;i<nMains;i++){  /* Create all mainchares */
334                 ChareInfo *entry = _chareTable[_mainTable[i]->chareIdx];
335                 int entryMigCtor = entry->getMigCtor();
336                 if(entryMigCtor!=-1) {
337                         Chare* obj;
338                         if (p.isUnpacking()) {
339                                 int size = entry->size;
340                                 DEBCHK("MainChare PUP'ed: name = %s, idx = %d, size = %d\n", entry->name, i, size);
341                                 obj = (Chare*)malloc(size);
342                                 _MEMCHECK(obj);
343                                 _mainTable[i]->setObj(obj);
344                                 //void *m = CkAllocSysMsg();
345                                 _entryTable[entryMigCtor]->call(args, obj);
346                         }
347                         else 
348                                 obj = (Chare *)_mainTable[i]->getObj();
349                         obj->virtual_pup(p);
350                 }
351         }
352         // to update mainchare proxy
353         // only readonly variables of Chare Proxy are taken care of here;
354         // in general, if chare proxy is contained in some data structure,
355         // such as CkCallback, it is user's responsibility to
356         // update them after restarting
357 #if !CMK_SHRINK_EXPAND
358         if (p.isUnpacking() && CkMyPe()==0)
359                 bdcastRO();
360 #endif
361
362 }
363
364 #ifndef CMK_CHARE_USE_PTR
365
366 CkpvExtern(std::vector<void *>, chare_objs);
367 CkpvExtern(std::vector<int>, chare_types);
368 CkpvExtern(std::vector<VidBlock *>, vidblocks);
369
370 // handle plain non-migratable chare
371 void CkPupChareData(PUP::er &p)
372 {
373   int i, n = 0;
374   if (!p.isUnpacking()) n = CkpvAccess(chare_objs).size();
375   p|n;
376   for (i=0; i<n; i++) {
377         int chare_type = 0;
378         if (!p.isUnpacking()) {
379                 chare_type = CkpvAccess(chare_types)[i];
380         }
381         p | chare_type;
382         bool pup_flag = true;
383         if (!p.isUnpacking()) {
384           if(CkpvAccess(chare_objs)[i] == NULL){
385             pup_flag = false;
386           }
387         }
388         p|pup_flag;
389         if(pup_flag)
390         {
391           if (p.isUnpacking()) {
392                   int migCtor = _chareTable[chare_type]->migCtor;
393                   if(migCtor==-1) {
394                           char buf[512];
395                           sprintf(buf,"Chare %s needs a migration constructor and PUP'er routine for restart.\n", _chareTable[chare_type]->name);
396                           CkAbort(buf);
397                   }
398                   void *m = CkAllocSysMsg();
399                   envelope* env = UsrToEnv((CkMessage *)m);
400                   CkCreateLocalChare(migCtor, env);
401                   CkFreeSysMsg(m);
402           }
403           Chare *obj = (Chare*)CkpvAccess(chare_objs)[i];
404           obj->virtual_pup(p);
405         }
406         else
407         {
408           CkpvAccess(chare_objs)[i] = NULL;
409         }
410   }
411
412   if (!p.isUnpacking()) n = CkpvAccess(vidblocks).size();
413   p|n;
414   for (i=0; i<n; i++) {
415         VidBlock *v;
416         bool pup_flag = true;
417         if (!p.isUnpacking()) {
418           if(CkpvAccess(vidblocks)[i]==NULL)
419           {
420             pup_flag = false;
421           }
422         }
423         p|pup_flag;
424         if(pup_flag)
425         {
426           if (p.isUnpacking()) {
427                   v = new VidBlock();
428                   CkpvAccess(vidblocks).push_back(v);
429           }
430           else{
431                   v = CkpvAccess(vidblocks)[i];
432           }
433           v->pup(p);
434         }
435   }
436 }
437 #else
438 void CkPupChareData(PUP::er &p)
439 {
440    // not implemented
441 }
442 #endif
443
444 typedef void GroupCreationFn(CkGroupID groupID, int constructorIdx, envelope *env);
445
446 static void CkPupPerPlaceData(PUP::er &p, GroupIDTable *idTable, GroupTable *objectTable,
447                               unsigned int &numObjects, int constructionMsgType,
448                               GroupCreationFn creationFn
449 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
450                               , bool create
451 #endif
452                              )
453 {
454   int numGroups = 0, i;
455
456   if (!p.isUnpacking()) {
457     numGroups = idTable->size();
458   }
459   p|numGroups;
460   if (p.isUnpacking()) {
461     if(CkMyPe()==0)  
462       numObjects = numGroups+1; 
463     else 
464       numObjects = 1;
465   }
466   DEBCHK("[%d] CkPupPerPlaceData %s: numGroups = %d\n", CkMyPe(),p.typeString(),numGroups);
467
468   std::vector<GroupInfo> tmpInfo(numGroups);
469   if (!p.isUnpacking()) {
470     for (i = 0; i < numGroups; i++) {
471       tmpInfo[i].gID = (*idTable)[i];
472       TableEntry ent = objectTable->find(tmpInfo[i].gID);
473       tmpInfo[i].present = ent.getObj() != NULL;
474       tmpInfo[i].MigCtor = _chareTable[ent.getcIdx()]->migCtor;
475       strncpy(tmpInfo[i].name,_chareTable[ent.getcIdx()]->name,255);
476       //CkPrintf("[%d] CkPupPerPlaceData: %s group %s \n", CkMyPe(), p.typeString(), tmpInfo[i].name);
477
478       if(tmpInfo[i].MigCtor==-1) {
479         char buf[512];
480         sprintf(buf,"(Node)Group %s needs a migration constructor and PUP'er routine for restart.\n", tmpInfo[i].name);
481         CkAbort(buf);
482       }
483     }
484   }
485   p|tmpInfo;
486
487   for (i = 0; i < numGroups; i++) 
488   {
489     if (!tmpInfo[i].present)
490       continue;
491
492     CkGroupID gID = tmpInfo[i].gID;
493     if (p.isUnpacking()) {
494       int eIdx = tmpInfo[i].MigCtor;
495       if (eIdx == -1) {
496         CkPrintf("[%d] ERROR> (Node)Group %s's migration constructor is not defined!\n", CkMyPe(), tmpInfo[i].name);
497         CkAbort("Abort");
498       }
499       void *m = CkAllocSysMsg();
500       envelope* env = UsrToEnv((CkMessage *)m);
501       env->setMsgtype(constructionMsgType);
502
503 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
504       if(create)
505 #endif
506       {
507         creationFn(gID, eIdx, env);
508       }
509     }   // end of unPacking
510     IrrGroup *gobj = objectTable->find(gID).getObj();
511
512 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
513     if(creationFn == CkCreateLocalGroup && !create)
514     {
515       gobj->mlogData->teamRecoveryFlag = 1;
516     }
517 #endif
518
519     // if using migration constructor, you'd better have a pup
520     gobj->virtual_pup(p);
521   }
522 }
523
524 void CkPupGroupData(PUP::er &p
525 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
526                     , bool create
527 #endif
528   )
529 {
530         CkPupPerPlaceData(p, CkpvAccess(_groupIDTable), CkpvAccess(_groupTable),
531                           CkpvAccess(_numGroups), BocInitMsg, &CkCreateLocalGroup
532 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
533                           , create
534 #endif
535                          );
536 }
537
538 void CkPupNodeGroupData(PUP::er &p
539 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
540                         , bool create
541 #endif
542   )
543 {
544           CkPupPerPlaceData(p, &CksvAccess(_nodeGroupIDTable),
545                             CksvAccess(_nodeGroupTable), CksvAccess(_numNodeGroups),
546                             NodeBocInitMsg, &CkCreateLocalNodeGroup
547 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
548                             , create
549 #endif
550                            );
551 }
552
553 // handle chare array elements for this processor
554 void CkPupArrayElementsData(PUP::er &p, int notifyListeners)
555 {
556         int i;
557         // safe in both packing/unpacking at this stage
558         int numGroups = CkpvAccess(_groupIDTable)->size();
559
560         // number of array elements on this processor
561         int numElements = 0;
562         if (!p.isUnpacking()) {
563           ElementCounter  counter;
564           CKLOCMGR_LOOP(mgr->iterate(counter););
565           numElements = counter.getCount();
566         }
567         p|numElements;
568
569         DEBCHK("[%d] CkPupArrayElementsData %s numGroups:%d numElements:%d \n",CkMyPe(),p.typeString(), numGroups, numElements);
570
571         if (!p.isUnpacking())
572         {
573           // let CkLocMgr iterate over and store every array element
574           CKLOCMGR_LOOP(ElementCheckpointer chk(mgr, p); mgr->iterate(chk););
575         }
576         else {
577           // loop and create all array elements ourselves
578           //CkPrintf("total chare array cnts: %d\n", numElements);
579           for (int i=0; i<numElements; i++) {
580                 CkGroupID gID;
581                 CkArrayIndex idx;
582                 CmiUInt8 id;
583                 p|gID;
584                 p|idx;
585                 p|id;
586                 CkLocMgr *mgr = (CkLocMgr*)CkpvAccess(_groupTable)->find(gID).getObj();
587                 if (notifyListeners){
588                   mgr->resume(idx, id, p, true);
589                 }
590                 else{
591                   mgr->restore(idx, id, p);
592                 }
593           }
594         }
595         // finish up
596         if (notifyListeners)
597         for(i=0;i<numGroups;i++) {
598                 IrrGroup *obj = CkpvAccess(_groupTable)->find((*CkpvAccess(_groupIDTable))[i]).getObj();
599                 if (obj)
600                   obj->ckJustMigrated();
601         }
602 }
603
604 #if __FAULT__
605 int  CkCountArrayElements(){
606     int numGroups = CkpvAccess(_groupIDTable)->size();
607     int i;
608     ElementCounter  counter;
609     CKLOCMGR_LOOP(mgr->iterate(counter););
610   int numElements = counter.getCount();
611     return numElements;
612 }
613 #endif
614
615 void CkPupProcessorData(PUP::er &p)
616 {
617     // save readonlys, and callback BTW
618     if(CkMyRank()==0) {
619         CkPupROData(p);
620     }
621
622     // save mainchares into MainChares.dat
623     if(CkMyPe()==0) {
624       CkPupMainChareData(p, NULL);
625     }
626         
627     // save non-migratable chare
628     CkPupChareData(p);
629
630     // save groups 
631 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
632     CkPupGroupData(p,true);
633 #else
634     CkPupGroupData(p);
635 #endif
636
637     // save nodegroups
638     if(CkMyRank()==0) {
639 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
640         CkPupNodeGroupData(p,true);     
641 #else
642         CkPupNodeGroupData(p);
643 #endif
644     }
645
646     // pup array elements
647     CkPupArrayElementsData(p);
648 }
649
650 // called only on pe 0
651 static bool checkpointOne(const char* dirname, CkCallback& cb, bool requestStatus){
652         CmiAssert(CkMyPe()==0);
653         char filename[1024];
654         
655         // save readonlys, and callback BTW
656         FILE* fRO = openCheckpointFile(dirname, "RO", "wb", -1);
657         PUP::toDisk pRO(fRO);
658         int _numPes = CkNumPes();
659         pRO|_numPes;
660         int _numNodes = CkNumNodes();
661
662         pRO|_numNodes;
663         pRO|cb;
664         CkPupROData(pRO);
665         pRO|requestStatus;
666
667         if(pRO.checkError())
668         {
669           return false;
670         }
671
672         if(CmiFclose(fRO)!=0)
673         {
674           return false;
675         }
676
677         // save mainchares into MainChares.dat
678         {
679                 FILE* fMain = openCheckpointFile(dirname, "MainChares", "wb", -1);
680                 PUP::toDisk pMain(fMain);
681                 CkPupMainChareData(pMain, NULL);
682                 if(pMain.checkError())
683                 {
684                   return false;
685                 }
686                 if(CmiFclose(fMain) != 0)
687                 {
688                   return false;
689                 }
690         }
691         return true;
692 }
693
694 void CkRemoveArrayElements()
695 {
696   int i;
697   int numGroups = CkpvAccess(_groupIDTable)->size();
698   CKLOCMGR_LOOP(mgr->flushAllRecs(););
699 /*  GroupTable *gTbl = CkpvAccess(_groupTable);
700   for(i=0; i<numGroups; i++){
701     IrrGroup *obj = CkpvAccess(_groupTable)->find((*CkpvAccess(_groupIDTable))[i]).getObj();
702     if(obj->isLocMgr()) {
703         CkLocMgr *mgr = (CkLocMgr *)obj;
704         mgr->flushAllRecs();
705     }
706   }*/
707 }
708
709 /*
710 void CkTestArrayElements()
711 {
712   int i;
713   int numGroups = CkpvAccess(_groupIDTable)->size();
714   //CKLOCMGR_LOOP(mgr->flushAllRecs(););
715   GroupTable *gTbl = CkpvAccess(_groupTable);
716   for(i=0; i<numGroups; i++){
717     IrrGroup *obj = CkpvAccess(_groupTable)->find((*CkpvAccess(_groupIDTable))[i]).getObj();
718     CkPrintf("An object at [%d]: %p | isLocMgr: %d\n", i, obj, obj->isLocMgr());
719   }
720 }
721 */
722
723 void CkStartCheckpoint(const char* dirname,const CkCallback& cb, bool requestStatus)
724 {
725   if(cb.isInvalid()) 
726     CkAbort("callback after checkpoint is not set properly");
727
728   if(cb.containsPointer())
729     CkAbort("Cannot restart from a callback based on a pointer");
730
731
732         CkPrintf("[%d] Checkpoint starting in %s\n", CkMyPe(), dirname);
733         
734         // hand over to checkpoint managers for per-processor checkpointing
735         CProxy_CkCheckpointMgr(_sysChkptMgr).Checkpoint(dirname, cb, requestStatus);
736 }
737
738 /**
739   * Restart: There's no such object as restart manager is created
740   *          because a group cannot restore itself anyway.
741   *          The mechanism exists as converse code and get invoked by
742   *          broadcast message.
743   **/
744
745 CkCallback cb;
746 void CkRestartMain(const char* dirname, CkArgMsg *args){
747         int i;
748         char filename[1024];
749         
750         if (CmiMyRank() == 0) {
751           _inrestart = true;
752           _restarted = true;
753           CkMemCheckPT::inRestarting = true;
754         }
755
756         // restore readonlys
757         FILE* fRO = openCheckpointFile(dirname, "RO", "rb", -1);
758         int _numPes = -1;
759         PUP::fromDisk pRO(fRO);
760         pRO|_numPes;
761         int _numNodes = -1;
762         pRO|_numNodes;
763         pRO|cb;
764         if (CmiMyRank() == 0) CkPupROData(pRO);
765         bool requestStatus = false;
766         pRO|requestStatus;
767         CmiFclose(fRO);
768         DEBCHK("[%d]CkRestartMain: readonlys restored\n",CkMyPe());
769         _oldNumPes = _numPes;
770
771         CmiNodeBarrier();
772
773         // restore mainchares
774         FILE* fMain = openCheckpointFile(dirname, "MainChares", "rb");
775         if(fMain && CkMyPe()==0){ // only main chares have been checkpointed, we restart on PE0
776                 PUP::fromDisk pMain(fMain);
777                 CkPupMainChareData(pMain, args);
778                 CmiFclose(fMain);
779                 DEBCHK("[%d]CkRestartMain: mainchares restored\n",CkMyPe());
780                 //bdcastRO(); // moved to CkPupMainChareData()
781         }
782         
783 #ifndef CMK_CHARE_USE_PTR
784         // restore chares only when number of pes is the same 
785         if(CkNumPes() == _numPes) {
786                 FILE* fChares = openCheckpointFile(dirname, "Chares", "rb", CkMyPe());
787                 PUP::fromDisk pChares(fChares);
788                 CkPupChareData(pChares);
789                 CmiFclose(fChares);
790                 if (CmiMyRank() == 0) _chareRestored = true;
791         }
792 #endif
793
794         // restore groups
795         // content of the file: numGroups, GroupInfo[numGroups], _groupTable(PUP'ed), groups(PUP'ed)
796         // restore from PE0's copy if shrink/expand
797         FILE* fGroups = openCheckpointFile(dirname, "Groups", "rb",
798                                      (CkNumPes() == _numPes) ? CkMyPe() : 0);
799         PUP::fromDisk pGroups(fGroups);
800 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
801     CkPupGroupData(pGroups,true);
802 #else
803     CkPupGroupData(pGroups);
804 #endif
805         CmiFclose(fGroups);
806
807         // restore nodegroups
808         // content of the file: numNodeGroups, GroupInfo[numNodeGroups], _nodeGroupTable(PUP'ed), nodegroups(PUP'ed)
809         if(CkMyRank()==0){
810                 FILE* fNodeGroups = openCheckpointFile(dirname, "NodeGroups", "rb",
811                                                        (CkNumNodes() == _numNodes) ? CkMyNode() : 0);
812                 PUP::fromDisk pNodeGroups(fNodeGroups);
813 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
814         CkPupNodeGroupData(pNodeGroups,true);
815 #else
816         CkPupNodeGroupData(pNodeGroups);
817 #endif
818                 CmiFclose(fNodeGroups);
819         }
820
821         // for each location, restore arrays
822         //DEBCHK("[%d]Trying to find location manager\n",CkMyPe());
823         DEBCHK("[%d]Number of PE: %d -> %d\n",CkMyPe(),_numPes,CkNumPes());
824         if(CkMyPe() < _numPes)  // in normal range: restore, otherwise, do nothing
825           for (i=0; i<_numPes;i++) {
826             if (i%CkNumPes() == CkMyPe()) {
827               FILE *datFile = openCheckpointFile(dirname, "arr", "rb", i);
828               PUP::fromDisk  p(datFile);
829               CkPupArrayElementsData(p);
830               CmiFclose(datFile);
831             }
832           }
833
834         _inrestart = false;
835
836         if (CmiMyRank()==0) _initDone();  // this rank will trigger other ranks
837         //_initDone();
838         CkMemCheckPT::inRestarting = false;
839         if(CkMyPe()==0) {
840                 CmiPrintf("[%d]CkRestartMain done. sending out callback.\n",CkMyPe());
841                 if(requestStatus)
842                 {
843                   CkCheckpointStatusMsg * m = new CkCheckpointStatusMsg(CK_CHECKPOINT_SUCCESS);
844                   cb.send(m); 
845                 }
846                 else
847                 {
848                   cb.send();
849                 }
850         }
851 }
852
853 #if CMK_SHRINK_EXPAND
854 // after resume and getting message
855 void CkResumeRestartMain(char * msg) {
856   int i;
857   char filename[1024];
858   const char * dirname = "";
859   _inrestart = true;
860   _restarted = true;
861   CkMemCheckPT::inRestarting = true;
862   CmiPrintf("[%d]CkResumeRestartMain: Inside Resume Restart\n",CkMyPe());
863   CmiPrintf("[%d]CkResumeRestartMain: Group restored %d\n",CkMyPe(), CkpvAccess(_numGroups)-1);
864
865   int _numPes = -1;
866   if(CkMyPe()!=0) {
867     PUP::fromMem pRO((char *)(msg+CmiMsgHeaderSizeBytes+2*sizeof(int)));
868
869     CkPupROData(pRO);
870     CmiPrintf("[%d]CkRestartMain: readonlys restored\n",CkMyPe());
871
872 #if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
873     CkPupGroupData(pRO,true);
874 #else
875     CkPupGroupData(pRO);
876 #endif
877     CmiPrintf("[%d]CkResumeRestartMain: Group restored %d\n",CkMyPe(), CkpvAccess(_numGroups)-1);
878   }
879
880   CmiFree(msg);
881   CmiNodeBarrier();
882   if(Cmi_isOldProcess) {
883     /* CmiPrintf("[%d] For shrinkexpand newpe=%d, oldpe=%d \n",Cmi_myoldpe, CkMyPe(), Cmi_myoldpe); */
884     // non-shrink files would be empty since LB would take care
885     FILE *datFile = openCheckpointFile(dirname, "arr", "rb", Cmi_myoldpe);
886     PUP::fromDisk  p(datFile);
887     CkPupArrayElementsData(p);
888     CmiFclose(datFile);
889   }
890   _initDone();
891   _inrestart = false;
892   CkMemCheckPT::inRestarting = false;
893   if(CkMyPe()==0) {
894     CmiPrintf("[%d]CkResumeRestartMain done. sending out callback.\n",CkMyPe());
895     CkPrintf("Restart from shared memory  finished in %fs, sending out the cb...\n", CmiWallTimer() - chkptStartTimer);
896     cb.send();
897   }
898 }
899 #endif
900
901 // Main chare: initialize system checkpoint manager
902 class CkCheckpointInit : public Chare {
903 public:
904   CkCheckpointInit(CkArgMsg *msg) {
905     _sysChkptMgr = CProxy_CkCheckpointMgr::ckNew();
906     delete msg;
907   }
908   CkCheckpointInit(CkMigrateMessage *m) {delete m;}
909 };
910
911 #include "CkCheckpoint.def.h"
912 #include "CkCheckpointStatus.def.h"
913