?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
switching current checkpoint so that if system crashes during checkpoint it pick
[csql.git] / src / storage / Database.cxx
blobe91f78ba1f4aac3a2835b373a9bc74c5a3e2c436
1 /***************************************************************************
2 * Copyright (C) 2007 by www.databasecache.com *
3 * Contact: praba_tuty@databasecache.com *
4 * *
5 * This program is free software; you can redistribute it and/or modify *
6 * it under the terms of the GNU General Public License as published by *
7 * the Free Software Foundation; either version 2 of the License, or *
8 * (at your option) any later version. *
9 * *
10 * This program is distributed in the hope that it will be useful, *
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of *
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
13 * GNU General Public License for more details. *
14 * *
15 ***************************************************************************/
16 #include<Database.h>
17 #include<os.h>
18 #include<CatalogTables.h>
19 #include<Transaction.h>
20 #include<Lock.h>
21 #include<Debug.h>
22 #include<Config.h>
23 #include<Process.h>
24
25 const char* Database::getName()
26 {
27 return metaData_->dbName_;
28 }
29
30 int Database::getDatabaseID()
31 {
32 return metaData_->dbID_;
33 }
34
35 long Database::getMaxSize()
36 {
37 return metaData_->maxSize_;
38 }
39
40 long Database::getCurrentSize()
41 {
42 return metaData_->curSize_;
43
44 }
45 Page* Database::getCurrentPage()
46 {
47 return metaData_->curPage_;
48
49 }
50 Page* Database::getFirstPage()
51 {
52 return metaData_->firstPage_;
53
54 }
55 int Database::getNoOfChunks()
56 {
57 return metaData_->noOfChunks_;
58 }
59 Chunk* Database::getHashIndexChunk()
60 {
61 return metaData_->hashIndexChunk_;
62 }
63
64 void Database::setDatabaseID(int id)
65 {
66 metaData_->dbID_ = id;
67 }
68 void Database::setName(const char *name)
69 {
70 strcpy(metaData_->dbName_ , name);
71 }
72 void Database::setCurrentSize(long size)
73 {
74 metaData_->curSize_ = size;
75 }
76 void Database::setCurrentPage(Page *page)
77 {
78 //metaData_->curPage_ = page;
79 Mutex::CASL((long*)&metaData_->curPage_, (long)metaData_->curPage_, (long)page);
80 }
81 void Database::setFirstPage(Page *page)
82 {
83 metaData_->firstPage_ = page;
84 }
85 void Database::setMaxSize(long size)
86 {
87 metaData_->maxSize_ = size;
88 }
89 void Database::setNoOfChunks(int chunks)
90 {
91 metaData_->noOfChunks_ = chunks;
92 }
93 void Database::setHashIndexChunk(Chunk *ch)
94 {
95 metaData_->hashIndexChunk_ = ch;
96 }
97
98
99 int Database::initAllocDatabaseMutex()
100 {
101 return metaData_->dbAllocMutex_.init("allocdb");
102 }
103 DbRetVal Database::getAllocDatabaseMutex(bool procAccount)
104 {
105 int ret= metaData_->dbAllocMutex_.getLock(procSlot, procAccount);
106 if (ret) return ErrLockTimeOut; else return OK;
107 }
108 DbRetVal Database::releaseAllocDatabaseMutex(bool procAccount)
109 {
110 metaData_->dbAllocMutex_.releaseLock(procSlot, procAccount);
111 return OK;
112 }
113
114 int Database::initPrepareStmtMutex()
115 {
116 return metaData_->dbPrepareStmtMutex_.init("prepstmt");
117 }
118 DbRetVal Database::getPrepareStmtMutex(bool procAccount)
119 {
120 int ret= metaData_->dbPrepareStmtMutex_.getLock(procSlot, procAccount);
121 if (ret) return ErrLockTimeOut; else return OK;
122 }
123 DbRetVal Database::releasePrepareStmtMutex(bool procAccount)
124 {
125 metaData_->dbPrepareStmtMutex_.releaseLock(procSlot, procAccount);
126 return OK;
127 }
128
129 int Database::initTransTableMutex()
130 {
131 return metaData_->dbTransTableMutex_.init("transtable");
132 }
133 DbRetVal Database::getTransTableMutex()
134 {
135 int ret = metaData_->dbTransTableMutex_.getLock(procSlot);
136 if (ret) return ErrLockTimeOut; else return OK;
137 }
138 DbRetVal Database::releaseTransTableMutex()
139 {
140 metaData_->dbTransTableMutex_.releaseLock(procSlot);
141 return OK;
142 }
143
144
145
146 int Database::initProcessTableMutex()
147 {
148 return metaData_->dbProcTableMutex_.init("proctable");
149 }
150 DbRetVal Database::getProcessTableMutex(bool procAccount)
151 {
152 int ret = metaData_->dbProcTableMutex_.getLock(procSlot, procAccount);
153 if (ret) return ErrLockTimeOut; else return OK;
154 }
155 DbRetVal Database::releaseProcessTableMutex(bool procAccount)
156 {
157 metaData_->dbProcTableMutex_.releaseLock(procSlot, procAccount);
158 return OK;
159 }
160
161
162
163 int Database::initCheckpointMutex()
164 {
165 return metaData_->ckptMutex_.init("checkpoint");
166 }
167 DbRetVal Database::getSCheckpointMutex(bool procAccount)
168 {
169 int ret = metaData_->ckptMutex_.getShareLock(procSlot, procAccount);
170 if (ret) return ErrLockTimeOut; else return OK;
171 }
172 DbRetVal Database::getXCheckpointMutex(bool procAccount)
173 {
174 int ret = metaData_->ckptMutex_.getExclusiveLock(procSlot, procAccount);
175 if (ret) return ErrLockTimeOut; else return OK;
176 }
177 DbRetVal Database::releaseCheckpointMutex(bool procAccount)
178 {
179 metaData_->ckptMutex_.releaseShareLock(procSlot, procAccount);
180 return OK;
181 }
182
183 // Gets the free page
184 // Each page is segmented by PAGE_SIZE, so it checks the pageInfo
185 // of each page to determine if the page is free
186 // Algorithm is to scan through the pageInfo objects stored at
187 // address (db start address + i * PAGE_SIZE) where i = 1..n till end
188 // database
189 // But in case of large tuples, pages are merged, so there wont be
190 // PageInfo object on pages which are merged.
191 // These pages are skipped by checking the nextPageAfterMerge_ of PageInfo
192
193 //NOTE::IMPORTANT::assumes alloc database lock is taken before calling this
194 Page* Database::getFreePage()
195 {
196 Page* page = getFirstPage();
197 //Page* page = getCurrentPage();
198 //printDebug(DM_Alloc, "Database::getFreePage firstPage:%x",page);
199 printDebug(DM_Alloc, "Database::getFreePage currentpage:%x",page);
200 PageInfo* pageInfo = ((PageInfo*)page);
201 char* endAddr = ((char*)getMetaDataPtr()) + getMaxSize();
202 int pageSize = PAGE_SIZE;
203 bool isEndAddchk=false;
204 while( 1 == pageInfo->isUsed_)
205 {
206 //If any pages are merged to store data larger than PAGE_SIZE
207 //move to the next page after the merge and check whether it is used
208 if ( pageInfo->nextPageAfterMerge_ == NULL) {
209 pageInfo = (PageInfo*)((char*)pageInfo + pageSize);
210 printDebug(DM_Alloc,"Normal Page:Moving to page:%x",pageInfo);
211 }
212 else {
213 pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
214 printDebug(DM_Alloc,"Merged Page:Moving to page:%x",pageInfo);
215 }
216
217 if((((char*) pageInfo) + pageSize) >= endAddr )
218 {
219 if(!isEndAddchk){ isEndAddchk=true; pageInfo=(PageInfo *)getFirstPage(); }
220 else
221 break;
222 }
223 if ((char*)pageInfo >= endAddr)
224 {
225 //printError(ErrSysInternal,"Invalid address %x",pageInfo);
226 return NULL;
227 }
228
229 }
230 if (!isValidAddress(((char*) pageInfo) + pageSize))
231 {
232 printError(ErrSysInternal, "Invalid address %x",((char*) pageInfo) + pageSize);
233 return NULL;
234 }
235 setCurrentPage((Page*) pageInfo);
236 printDebug(DM_Alloc,"Database::getFreePage returning page:%x",pageInfo);
237 return (Page*) pageInfo ;
238 }
239
240 //Used by tuples more than PAGE_SIZE
241 //NOTE::IMPORTANT::assumes alloc database lock is taken before calling this
242 Page* Database::getFreePage(size_t size)
243 {
244 Page* page = getFirstPage();
245 PageInfo* pageInfo = ((PageInfo*)page);
246 int multiple = size / PAGE_SIZE;
247 int offset = ((multiple + 1) * PAGE_SIZE);
248 printDebug(DM_Alloc, "Database::getFreePage firstPage:%x size:%ld",page, size);
249 char* endAddr = ((char*)getMetaDataPtr()) + getMaxSize();
250 int pageSize = PAGE_SIZE;
251 bool isEndAddchk = false;
252 while(true){
253 while( 1 == pageInfo->isUsed_)
254 {
255 //If any pages are merged to store data larger than PAGE_SIZE
256 //move to the next page after the merge and check whether it is used
257 if ( pageInfo->nextPageAfterMerge_ == NULL) {
258 pageInfo = (PageInfo*)((char*)pageInfo + pageSize);
259 printDebug(DM_Alloc,"Normal Page:Moving to page:%x",pageInfo);
260 }
261 else {
262 pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
263 printDebug(DM_Alloc,"Merged Page:Moving to page:%x",pageInfo);
264 }
265 if((((char*) pageInfo) + offset) >= endAddr )
266 {
267 if(!isEndAddchk){ isEndAddchk=true; pageInfo=(PageInfo *)getFirstPage(); }
268 else
269 break;
270 }
271 }
272 int i = 0;
273 PageInfo *pInfo = pageInfo;
274 if ((((char*)pInfo) + offset) >= endAddr)
275 {
276 printError(ErrSysInternal,"Invalid address %x",((char*)pInfo) + offset);
277 return NULL;
278 }
279 for (i = 0; i< multiple + 1; i++)
280 {
281 if (1 == pInfo->isUsed_) break;
282 pInfo = (PageInfo*)((char*)pInfo + pageSize);
283 }
284 if ( i == (multiple + 1)) break;
285 pageInfo = (PageInfo*)((char*)pInfo + pageSize);
286 }
287
288 printDebug(DM_Alloc,"Database::getFreePage returning page:%x",pageInfo);
289 setCurrentPage((Page*) pageInfo);
290 return (Page*) pageInfo ;
291 }
292
293 void Database::printStatistics()
294 {
295 Page* page = getFirstPage();
296 PageInfo* pageInfo = ((PageInfo*)page);
297 int usedPageCount =0, usedMergedPageCount =0, totalPages=0;
298 int totalDirtyPages=0;
299 printf("<DatabaseStatistics>\n");
300 printf(" <Database Name> %s </Database Name>\n", getName());
301 printf(" <Max Size> %ld </Max Size>\n", getMaxSize());
302 printf(" <First Page> %x </First Page>\n", getFirstPage());
303 while(isValidAddress((char*) pageInfo))
304 {
305 if (pageInfo == NULL) break;
306 //if (pageInfo > getCurrentPage()) break;
307 if (1 == pageInfo->isUsed_) {
308 if ( pageInfo->nextPageAfterMerge_ == NULL) {
309 if (BITSET(pageInfo->flags, IS_DIRTY)) totalDirtyPages++;
310 pageInfo = (PageInfo*)((char*)pageInfo + PAGE_SIZE);
311 usedPageCount++; totalPages++;
312 printDebug(DM_Alloc, "Normal Page:Moving to page:%x\n",pageInfo);
313 continue;
314 }
315 else {
316 if (BITSET(pageInfo->flags, IS_DIRTY)) totalDirtyPages++;
317 pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
318 usedMergedPageCount++; totalPages++;
319 printDebug(DM_Alloc,"Merged Page:Moving to page:%x\n",pageInfo);
320 continue;
321 }
322 } else if (BITSET(pageInfo->flags, IS_DIRTY)) totalDirtyPages++;
323 pageInfo = (PageInfo*)((char*)pageInfo + PAGE_SIZE);
324 printDebug(DM_Alloc,"Normal Page not used:Moving to page:%x\n",pageInfo);
325 totalPages++;
326 }
327 printf(" <Total Pages> %d </Total Pages>\n", totalPages);
328 if (Conf::config.useDurability())
329 printf(" <Dirty Pages> %d </Dirty Pages>\n", totalDirtyPages);
330 printf(" <Used Normal Pages> %d </Used Normal Pages>\n", usedPageCount);
331 printf(" <Used Merged Pages> %d </Used Merged Pages>\n", usedMergedPageCount);
332 printf(" <Chunks Used> %d </Chunks Used>\n", getNoOfChunks());
333 printf("</DatabaseStatistics>\n");
334
335 return ;
336 }
337
338
339 //called only in case of system database to create and initialize the chunk
340 //information
341 DbRetVal Database::createSystemDatabaseChunk(AllocType type, size_t size, int id)
342 {
343
344 Chunk *chunk;
345 if (-1 == id )
346 {
347 printError(ErrSysFatal, "Database ID corrupted");
348 return ErrSysFatal;
349 }
350 chunk = getSystemDatabaseChunk(id);
351
352 chunk->setChunkNameForSystemDB(id);
353
354 if (FixedSizeAllocator == type) chunk->setSize(size);
355 //getDatabaseMutex();
356 if (chunk->allocSize_ > PAGE_SIZE)
357 chunk->curPage_ = getFreePage(chunk->allocSize_);
358 else
359 chunk->curPage_ = getFreePage();
360 if ( chunk->curPage_ == NULL)
361 {
362 //releaseDatabaseMutex();
363 printError(ErrNoMemory, "No free pages in database: Database full");
364 return ErrNoMemory;
365 }
366
367 chunk->firstPage_ = chunk->curPage_;
368 PageInfo* firstPageInfo = ((PageInfo*)chunk->firstPage_);
369 firstPageInfo->setFirstPageAsUsed();
370 chunk->setChunkID(id);
371 chunk->setAllocType(type);
372 printDebug(DM_Database, "Creating System Database Chunk:%d Size:%d",id, chunk->allocSize_);
373 if (chunk->allocSize_ > PAGE_SIZE)
374 {
375 int multiple = os::floor(chunk->allocSize_ / PAGE_SIZE);
376 int offset = ((multiple + 1) * PAGE_SIZE);
377 firstPageInfo->nextPageAfterMerge_ = ((char*)firstPageInfo)+ offset;
378 }
379
380 if (0 == size)
381 {
382 VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
383 varInfo->isUsed_ = 0;
384 varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
385
386 }
387 incrementChunk();
388 //releaseDatabaseMutex();
389 return OK;
390 }
391
392 //This is never called currently. If situation arises will be coded later.
393 DbRetVal Database::deleteSystemDatabaseChunk(int id)
394 {
395
396 Chunk *chunk = getSystemDatabaseChunk(id);
397 chunk->setChunkID(-1);
398 chunk->setSize(0);
399 chunk->setAllocType(UnknownAllocator);
400 //TODO::
401 //chunk->pageList_
402 //walk though the pageList ptr and get all the page pointers
403 //then free all the pages used to store this by setting the
404 //start of page to notused
405 chunk->firstPage_ = NULL;
406 chunk->curPage_ = NULL;
407 decrementChunk();
408 return OK;
409 }
410
411
412 void Database::createAllCatalogTables()
413 {
414 //These are special chunks which hold catalog tables and other information
415 //
416 // chunk id 0 ->userChunkTable
417 // chunk id 1 ->lockBucketHash
418 // chunk id 2 ->lockTable
419 //
420 // chunk id 10->DATABASE
421 // chunk id 11->USER
422 // chunk id 12->TABLE
423 // chunk id 13->FIELD
424 // chunk id 14->ACCESS
425
426 createSystemTables();
427 createMetaDataTables();
428 }
429 void Database::createSystemTables()
430 {
431 createSystemDatabaseChunk(FixedSizeAllocator,
432 sizeof(Chunk), UserChunkTableId);
433 createSystemDatabaseChunk(FixedSizeAllocator,
434 sizeof(Bucket) * LOCK_BUCKET_SIZE,
435 LockTableHashBucketId);
436 createSystemDatabaseChunk(FixedSizeAllocator,
437 sizeof(Mutex)* LOCK_BUCKET_SIZE,
438 LockTableMutexId);
439 createSystemDatabaseChunk(FixedSizeAllocator,
440 sizeof(LockHashNode), LockTableId);
441 createSystemDatabaseChunk(FixedSizeAllocator,
442 sizeof(TransHasNode), TransHasTableId);
443
444 createSystemDatabaseChunk(VariableSizeAllocator,
445 0, UndoLogTableID);
446 }
447 void Database::createMetaDataTables()
448 {
449 createSystemDatabaseChunk(FixedSizeAllocator,
450 sizeof(CDATABASEFILE), DatabaseTableId);
451 createSystemDatabaseChunk(FixedSizeAllocator,
452 sizeof(CUSER), UserTableId);
453 createSystemDatabaseChunk(FixedSizeAllocator,
454 sizeof(CTABLE), TableTableId);
455 createSystemDatabaseChunk(FixedSizeAllocator,
456 sizeof(CFIELD), FieldTableId);
457 createSystemDatabaseChunk(FixedSizeAllocator,
458 sizeof(CACCESS), AccessTableId);
459 createSystemDatabaseChunk(FixedSizeAllocator,
460 sizeof(CINDEX), IndexTableId);
461 createSystemDatabaseChunk(FixedSizeAllocator,
462 sizeof(CINDEXFIELD), IndexFieldTableId);
463 createSystemDatabaseChunk(FixedSizeAllocator,
464 sizeof(CFK), ForeignKeyTableId);
465 createSystemDatabaseChunk(FixedSizeAllocator,
466 sizeof(CFKFIELD), ForeignKeyFieldTableId);
467 }
468
469 //used in case of system database
470 Chunk* Database::getSystemDatabaseChunk(int id)
471 {
472 size_t offset = os::alignLong(sizeof (DatabaseMetaData)) +
473 id * sizeof (Chunk);
474 return (Chunk*)(((char*) metaData_) + offset);
475 }
476
477
478 //used in case of system database
479 Transaction* Database::getSystemDatabaseTrans(int slot)
480 {
481 size_t offset = os::alignLong(sizeof (DatabaseMetaData)) +
482 os::alignLong(MAX_CHUNKS * sizeof (Chunk)) +
483 slot * sizeof (Transaction);
484 return (Transaction*)(((char*) metaData_) + offset);
485 }
486
487 bool Database::isValidAddress(void* addr)
488 {
489 if ((char*) addr >= ((char*)getMetaDataPtr()) + getMaxSize())
490 return false;
491 else
492 return true;
493 }
494
495 //should be called only on system database
496 void* Database::allocLockHashBuckets()
497 {
498 Chunk *chunk = getSystemDatabaseChunk(LockTableHashBucketId);
499 DbRetVal rv=OK;
500 void *ptr = chunk->allocate(this, &rv);
501 if (NULL == ptr)
502 {
503 printError(ErrNoMemory, "Chunk Allocation failed for lock hash bucket catalog table");
504 }
505 return ptr;
506 }
507
508 Bucket* Database::getLockHashBuckets()
509 {
510 Chunk *tChunk = getSystemDatabaseChunk(LockTableHashBucketId);
511 ChunkIterator iter = tChunk->getIterator();
512 return (Bucket*)iter.nextElement();
513 }
514 void Database::setUniqueChunkID(int id)
515 {
516 (metaData_->chunkUniqueID_).setID(id);
517 }
518
519 int Database::getUniqueIDForChunk()
520 {
521 return ((metaData_->chunkUniqueID_).getID());
522 }
523
524 DbRetVal Database::recoverMutex(Mutex *mut)
525 {
526 //TODO: operations need to be undone before recovering the mutex.
527 mut->recoverMutex();
528 return OK;
529 }
530 DbRetVal Database::writeDirtyPages(char *dataFile)
531 {
532 int fd = open(dataFile, O_WRONLY|O_CREAT, 0644);
533 lseek(fd, 0, SEEK_SET);
534 void *buf = (void *) metaData_;
535 int sizeToWrite = os::alignLong(sizeof(DatabaseMetaData));
536 ssize_t retSize = os::write(fd, (char*)buf, sizeToWrite);
537 if (-1 == retSize)
538 {
539 printError(ErrWarning, "Warning:Unable to write metadata");
540 return ErrSysInternal;
541 }
542 PageInfo *pageInfo = (PageInfo*) getFirstPage();
543 long pageSize =PAGE_SIZE;
544 int pagesWritten=0, writeOffset=0;
545 long long totalBytesWritten=0;
546 while(isValidAddress((char*) pageInfo))
547 {
548 if ( NULL == pageInfo ) break;
549 if (pageInfo > getCurrentPage()) {
550 char *a="0";
551 lseek(fd, getMaxSize() -1, SEEK_SET);
552 if ( -1 == os::write(fd, a, 1)) {
553 printError(ErrSysInternal, "Unable to extend chkpt file");
554 close(fd);
555 return ErrSysInternal;
556 }
557 break;
558 }
559 if (BITSET(pageInfo->flags, IS_DIRTY)) {
560 if (NULL == pageInfo->nextPageAfterMerge_)
561 pageSize = PAGE_SIZE;
562 else
563 pageSize = (long)pageInfo->nextPageAfterMerge_ - (long)pageInfo;
564 writeOffset = (long) pageInfo - (long) metaData_;
565 lseek(fd, writeOffset, SEEK_SET);
566 CLEARBIT(pageInfo->flags, IS_DIRTY);
567 retSize = os::write(fd, (char*)pageInfo, pageSize);
568 if ( -1 == retSize ) {
569 printError(ErrSysInternal, "Unable to write dirty page %x", pageInfo);
570 close(fd);
571 return ErrSysInternal;
572 }
573 totalBytesWritten= totalBytesWritten + retSize;
574 pagesWritten++;
575 }
576 if ( pageInfo->nextPageAfterMerge_ == NULL) {
577 pageInfo = (PageInfo*)((char*)pageInfo + PAGE_SIZE);
578 } else {
579 pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
580 }
581 }
582 printf("Total Dirty pages written %d %lld\n", pagesWritten, totalBytesWritten);
583 logFine(Conf::logger, "Total Dirty pages written %d\n", pagesWritten);
584 close(fd);
585 return OK;
586 }
587
588 DbRetVal Database::checkPoint()
589 {
590 char dataFile[MAX_FILE_LEN];
591 char cmd[MAX_FILE_LEN];
592 char dbRedoFileName[MAX_FILE_LEN];
593 sprintf(dbRedoFileName, "%s/csql.db.cur", Conf::config.getDbFile());
594 if (!Conf::config.useMmap()) {
595 // sprintf(dataFile, "%s/db.chkpt.data1", Conf::config.getDbFile());
596 sprintf(dataFile, "%s/db.chkpt.data", Conf::config.getDbFile());
597 FILE *fp = NULL;
598 if (fp = fopen(dataFile, "r")) {
599 fclose(fp);
600 int ret = unlink(dataFile);
601 if (ret != OK) {
602 printError(ErrOS, "Unable to delete old chkpt file. Failure");
603 return ErrOS;
604 }
605 }
606 int fd = open(dataFile, O_WRONLY|O_CREAT, 0644);
607 void *buf = (void *) metaData_;
608 lseek(fd, 0, SEEK_SET);
609 write(fd, buf, Conf::config.getMaxDbSize());
610 close(fd);
611 sprintf(cmd, "cp -f %s/db.chkpt.data %s/db.chkpt.data1", Conf::config.getDbFile(), Conf::config.getDbFile());
612 int ret = system(cmd);
613 if (ret != 0) {
614 printError(ErrOS, "Unable to take checkpoint back up file");
615 return ErrOS;
616 }
617 } else {
618 int fd = getChkptfd();
619 if (!os::fdatasync(fd)) {
620 logFine(Conf::logger, "fsync succedded");
621 }
622 int ret = unlink(dbRedoFileName);
623 if (ret != 0) {
624 close(fd);
625 printError(ErrSysInternal, "Unable to delete redo log file");
626 printError(ErrSysInternal, "Delete %s manually and restart the server", dbRedoFileName);
627 return ErrOS;
628 }
629 //switch the checkpoint so that during recovery, fsynced checkpoint is
630 //used during recovery if the below step(writeDirtyPages)
631 //is not completed succesfully.
632 if (Database::getCheckpointID() == 0)
633 Database::setCheckpointID(1);
634 else
635 Database::setCheckpointID(0);
636
637 int val=Database::getCheckpointID();
638
639 sprintf(dataFile, "%s/db.chkpt.data%d", Conf::config.getDbFile(), val);
640 DbRetVal rv = writeDirtyPages(dataFile);
641 if (OK != rv)
642 {
643 printError(ErrSysInternal, "Unable to write dirty pages");
644 close(fd);
645 return rv;
646 }
647
648 //Note: do not change order, chkpt id should be switched only after
649 //all dirty pages are written to disk. otherwise(if server crashes
650 //when it writes these dirty pages) recovery should use
651 //mapped file as fsync is already done on that file.
652 if (Database::getCheckpointID() == 0)
653 Database::setCheckpointID(1);
654 else
655 Database::setCheckpointID(0);
656
657 close(fd);
658 return OK;
659 }
660 int ret = unlink(dbRedoFileName);
661 if (ret != 0) {
662 printError(ErrSysInternal, "Unable to delete redo log file. Delete and restart the server\n");
663 return ErrOS;
664 }
665 return OK;
666 }
667 int Database::getCheckpointID()
668 {
669 int id=0;
670 char curCkptFile[MAX_FILE_LEN];
671 sprintf(curCkptFile, "%s/db.chkpt.cur", Conf::config.getDbFile());
672 FILE *fp = fopen(curCkptFile, "r");
673 if (NULL == fp) { setCheckpointID(0); return 0; }
674 fscanf(fp, "%d", &id);
675 fclose(fp);
676 return id;
677 }
678 void Database::setCheckpointID(int id)
679 {
680 char curCkptFile[MAX_FILE_LEN];
681 sprintf(curCkptFile, "%s/db.chkpt.cur", Conf::config.getDbFile());
682 FILE *fp = fopen(curCkptFile, "w");
683 if (NULL == fp) {
684
685 printError(ErrSysInternal, "Unable to set checkpointID");
686 return;
687 }
688 fprintf(fp, "%d", id);
689 logFine(Conf::logger, "Current checkpoint set to %d", id);
690 fclose(fp);
691 return;
692 }
693
694
695 //used only by the user database not the system database
696 DbRetVal Database::recoverUserDB()
697 {
698 char dataFile[MAX_FILE_LEN];
699 char cmd[MAX_FILE_LEN];
700 sprintf(dataFile, "%s/db.chkpt.data", Conf::config.getDbFile());
701 int fd = open(dataFile, O_RDONLY);
702 if (-1 == fd) { return OK; }
703 void *buf = (void *) metaData_;
704 read(fd, buf, Conf::config.getMaxDbSize());
705 close(fd);
706 return OK;
707 }
708
709 //used only by the system database
710 DbRetVal Database::recoverSystemDB()
711 {
712 char mapFile[MAX_FILE_LEN];
713 sprintf(mapFile, "%s/db.chkpt.map", Conf::config.getDbFile());
714 int fd = open(mapFile, O_RDONLY);
715 if (-1 == fd) { return OK; }
716 CatalogTableTABLE cTable(this);
717 CatalogTableINDEX cIndex(this);
718 struct Object buf;
719 while (read(fd, &buf, sizeof(buf))) {
720 if (buf.type == Tbl) {
721 cTable.setChunkPtr(buf.name, buf.firstPage, buf.curPage);
722 }
723
724 else if (buf.type == hIdx || buf.type == tIdx) {
725 cIndex.setChunkPtr(buf.name, buf.type, buf.bucketChunk, buf.firstPage, buf.curPage);
726 }
727 }
728 return OK;
729 }
Main memory database SQL engine (cvs import)