src/storage/DatabaseManagerImpl.cxx

   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<DatabaseManager.h>
  18 #include<DatabaseManagerImpl.h>
  19 #include<os.h>
  20 #include<Table.h>
  21 #include<TableImpl.h>
  22 #include<Transaction.h>
  23 #include<CatalogTables.h>
  24 #include<Index.h>
  25 #include<Lock.h>
  26 #include<Debug.h>
  27 #include<Config.h>
  28 //#include<TableConfig.h>
  29 #include<Process.h>
  30
  31
  32 DatabaseManagerImpl::~DatabaseManagerImpl()
  33 {
  34     //Note:Databases are closed by the session interface
  35     delete tMgr_;
  36     delete lMgr_;
  37 }
  38
  39 void DatabaseManagerImpl::createLockManager()
  40 {
  41     lMgr_ = new LockManager(systemDatabase_);
  42     return;
  43 }
  44
  45 void DatabaseManagerImpl::createTransactionManager()
  46 {
  47
  48     tMgr_ = new TransactionManager();
  49     tMgr_->setFirstTrans(systemDatabase_->getSystemDatabaseTrans(0));
  50     return;
  51 }
  52 void DatabaseManagerImpl::setProcSlot()
  53 {
  54     systemDatabase_->setProcSlot(procSlot);
  55     db_->setProcSlot(procSlot);
  56 }
  57 DbRetVal DatabaseManagerImpl::openSystemDatabase()
  58 {
  59     DbRetVal rv = openDatabase(SYSTEMDB);
  60     if (rv != OK) return rv;
  61     systemDatabase_ = db_;
  62     db_ = NULL;
  63     printDebug(DM_Database, "Opened system database");
  64     return OK;
  65 }
  66
  67 DbRetVal DatabaseManagerImpl::closeSystemDatabase()
  68 {
  69     Database *db  = db_;
  70     //make them to point to system database file descriptor
  71     //and database pointer
  72     db_ = systemDatabase_;
  73     closeDatabase();
  74     db_ = db;
  75     printDebug(DM_Database, "Closed system database");
  76     return OK;
  77 }
  78
  79 DbRetVal DatabaseManagerImpl::createDatabase(const char *name, size_t size)
  80 {
  81     bool isMmapNeeded = Conf::config.useMmap();
  82     file_desc fd = (file_desc)-1;
  83     char cmd[1024];
  84     char dbMapFile[MAX_FILE_LEN];
  85     struct stat st;
  86     long fixAddr = MAP_ADDR_VALUE;
  87     bool firstTimeServer = false;
  88     if (NULL != db_ )
  89     {
  90         printError(ErrAlready, "Database is already created");
  91         return ErrAlready;
  92     }
  93     caddr_t rtnAddr = (caddr_t) NULL;
  94     void *mapAddr = NULL;
  95     shared_memory_id shm_id = 0;
  96
  97     char *startaddr = (char*)Conf::config.getMapAddress();
  98     shared_memory_key key = 0;
  99     if (0 == strcmp(name, SYSTEMDB))
 100     {
 101
 102         key = Conf::config.getSysDbKey();
 103     }
 104     else
 105     {
 106         startaddr = startaddr + Conf::config.getMaxSysDbSize();
 107         key = Conf::config.getUserDbKey();
 108     }
 109     if (!isMmapNeeded || (isMmapNeeded && 0 == strcmp(name, SYSTEMDB))) {
 110         shm_id = os::shm_create(key, size, 0660);
 111         if (-1 == shm_id) {
 112             if (errno == EEXIST) {
 113 #if (defined MMDB && defined EMBED)
 114                 printError(ErrOS, "One application is already running.");
 115                 return ErrOS;
 116 #else
 117                 printError(ErrOS, "Shared Memory already exists");
 118 #endif
 119             }
 120             printError(ErrOS, "Shared memory create failed");
 121             exit(0);
 122         }
 123     } else {
 124         //switch the checkpoint
 125         if (Database::getCheckpointID() == 0)
 126             Database::setCheckpointID(1);
 127         else
 128             Database::setCheckpointID(0);
 129         int chkptID=Database::getCheckpointID();
 130
 131
 132         sprintf(dbMapFile, "%s/db.chkpt.data%d", Conf::config.getDbFile(), chkptID);
 133         fd = os::openFile(dbMapFile, fileOpenCreat, 0660);
 134         if ((file_desc) -1 == fd) {
 135             printError(ErrOS, "Mmap file could not be opened");
 136             return ErrOS;
 137         }
 138         if(::stat(dbMapFile, &st) == -1) {
 139             printf("Unable to retrieve the db File data\n");
 140             os::closeFile(fd);
 141             db_->setChkptfd((file_desc)-1);
 142             return ErrOS;
 143         }
 144 #ifdef WINNT
 145         int localfd = os::open(dbMapFile, fileOpenCreat,0);
 146 #else
 147         int localfd = fd;
 148 #endif
 149         if (st.st_size == 0 || st.st_size < size) {
 150             firstTimeServer = true;
 151             off_t flSize = os::lseek(localfd, size - 1, SEEK_SET);
 152         }
 153         char *a = "0";
 154         int wSize = os::write(localfd, a, 1);
 155 #ifdef WINNT
 156         os::close(localfd);
 157 #endif
 158         mapAddr = os::mmap((void *)(fixAddr + Conf::config.getMaxSysDbSize()), size,
 159                                 mapProtRead | mapProtWrite, mapFixed | mapShared, fd, 0);
 160         rtnAddr = (caddr_t) mapAddr;
 161         printDebug(DM_Database, "Mapped db file address = %x", mapAddr);
 162     }
 163     void *shm_ptr = NULL;
 164     if (!isMmapNeeded || isMmapNeeded && 0 == strcmp(name, SYSTEMDB)) {
 165         shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 166         rtnAddr  = (caddr_t) shm_ptr;
 167         if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
 168         {
 169             printError(ErrOS, "Shared memory attach returned -ve value %d", rtnAddr);
 170             return ErrOS;
 171         }
 172 # if (defined MMDB && defined EMBED)
 173         if (0 == strcmp(name, SYSTEMDB)) ProcessManager::sysAddr = rtnAddr;
 174         else ProcessManager::usrAddr = rtnAddr;
 175 # endif
 176     }
 177     db_ = new Database();
 178     printDebug(DM_Database, "Creating database:%s",name);
 179
 180     /*if (!isMmapNeeded || isMmapNeeded && 0 == strcmp(name, SYSTEMDB)) {
 181         memset(shm_ptr, 0, size );
 182     }*/
 183
 184     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 185     db_->setDatabaseID(1);
 186     db_->setName(name);
 187     db_->setMaxSize(size);
 188     initMutexes(db_);
 189     db_->setUniqueChunkID(100);
 190     //compute the first page after book keeping information
 191     size_t offset = os::alignLong(sizeof (DatabaseMetaData));
 192     //Only for system db chunk array, trans array and proc array will be there
 193     if (0 == strcmp(name, SYSTEMDB))
 194     {
 195         db_->setCanTakeCheckPoint(true);
 196         offset = computeSysDbOffset();
 197     }
 198     int multiple = os::floor(offset / PAGE_SIZE);
 199     char *curPage = (((char*)rtnAddr) + ((multiple + 1) * PAGE_SIZE));
 200
 201     db_->setCurrentPage(curPage);
 202     db_->setFirstPage(curPage);
 203
 204     return OK;
 205 }
 206 size_t DatabaseManagerImpl::computeSysDbOffset()
 207 {
 208     size_t offset = os::alignLong(sizeof (DatabaseMetaData));
 209     offset = offset + os::alignLong(MAX_CHUNKS  * sizeof (Chunk));
 210     offset = offset + os::alignLong(Conf::config.getMaxProcs() * sizeof(Transaction));
 211     offset = offset + os::alignLong(Conf::config.getMaxProcs() * sizeof(ThreadInfo));
 212     return offset;
 213 }
 214 void DatabaseManagerImpl::initMutexes(Database *db_)
 215 {
 216     //TODO:for user database do not have transtable and processtable mutex
 217     db_->setNoOfChunks(0);
 218     db_->initAllocDatabaseMutex();
 219     db_->initTransTableMutex();
 220     db_->initCheckpointMutex();
 221     db_->initProcessTableMutex();
 222     db_->initPrepareStmtMutex();
 223     return;
 224 }
 225
 226 DbRetVal DatabaseManagerImpl::deleteDatabase(const char *name)
 227 {
 228     shared_memory_id shm_id = 0;
 229     if (0 == strcmp(name, SYSTEMDB))
 230     {
 231         shm_id = os::shm_open(Conf::config.getSysDbKey(), 100, 0660);
 232         os::shm_remove(shm_id);
 233                 delete systemDatabase_;
 234                 systemDatabase_ = NULL;
 235     } else {
 236         shm_id = os::shm_open(Conf::config.getUserDbKey(), 100, 0660);
 237         os::shm_remove(shm_id);
 238                 delete db_;
 239                 db_ = NULL;
 240     }
 241     return OK;
 242 }
 243
 244 DbRetVal DatabaseManagerImpl::openDatabase(const char *name)
 245 {
 246     bool isMmapNeeded = Conf::config.useMmap();
 247     char dbMapFile[MAX_FILE_LEN];
 248     file_desc fd = (file_desc)-1;
 249     long size = Conf::config.getMaxSysDbSize();
 250     char *startaddr = (char*)Conf::config.getMapAddress();
 251     long fixAddr = MAP_ADDR_VALUE;
 252     if (0 == strcmp(name , SYSTEMDB))
 253     {
 254         if (NULL !=systemDatabase_)
 255         {
 256             printError(ErrAlready, "System Database already open");
 257             return ErrAlready;
 258         }
 259     }
 260     else
 261     {
 262         if (NULL ==systemDatabase_)
 263         {
 264             printError(ErrNotOpen, "System Database not open");
 265             return ErrNotOpen;
 266         }
 267         size = Conf::config.getMaxDbSize();
 268         startaddr = startaddr +  Conf::config.getMaxSysDbSize();
 269         fixAddr += Conf::config.getMaxSysDbSize();
 270     }
 271     if (NULL != db_)
 272     {
 273         printError(ErrAlready, "User Database already open");
 274         return ErrAlready;
 275     }
 276     //system db should be opened before user database files
 277     caddr_t rtnAddr = (caddr_t) NULL;
 278
 279     shared_memory_id shm_id = 0;
 280     shared_memory_key key = 0;
 281
 282     if (0 == strcmp(name, SYSTEMDB))
 283         key = Conf::config.getSysDbKey();
 284     else
 285         key = Conf::config.getUserDbKey();
 286
 287
 288     void *shm_ptr = NULL;
 289     void *mapAddr = NULL;
 290     bool firstThread = false;
 291     if ( ( ProcessManager::noThreads == 0 && 0 == strcmp(name, SYSTEMDB) ||
 292           ProcessManager::noThreads == 1 && 0 != strcmp(name, SYSTEMDB) ) )
 293    {
 294         if(isMmapNeeded && 0 != strcmp(name, SYSTEMDB)){
 295             //: Attach to Map File
 296             int curChkptID = Database::getCheckpointID();
 297             sprintf(dbMapFile, "%s/db.chkpt.data%d", Conf::config.getDbFile(),
 298                                                      curChkptID);
 299             fd = os::openFile(dbMapFile, fileOpenReadWrite, 0660);
 300             if ((file_desc)-1 == fd) {
 301                 printError(ErrOS, "Mmap file could not be opened");
 302                 return ErrOS;
 303             }
 304             mapAddr = os::mmap((void *)fixAddr, size, mapProtRead | mapProtWrite,
 305                                 mapFixed | mapShared, fd, 0);
 306
 307             shm_ptr= (caddr_t) mapAddr;
 308             printDebug(DM_Database, "Mapped db file address = %x", mapAddr);
 309         }else
 310         {
 311             shm_id = os::shm_open(key, size, 0660);
 312             if (shm_id == -1 )
 313             {
 314                 printError(ErrOS, "Shared memory open failed");
 315                 return ErrOS;
 316             }
 317             shm_ptr = os::shm_attach(shm_id, startaddr, SHM_RND);
 318         }
 319         if (0 == strcmp(name, SYSTEMDB))
 320         {
 321             firstThread = true;
 322             ProcessManager::sysAddr = (char*) shm_ptr;
 323         }
 324         else
 325         {
 326             ProcessManager::usrAddr = (char*) shm_ptr;
 327         }
 328     } else {
 329         if (0 == strcmp(name, SYSTEMDB)) shm_ptr = ProcessManager::sysAddr;
 330         else shm_ptr = ProcessManager::usrAddr;
 331     }
 332
 333     rtnAddr  = (caddr_t) shm_ptr;
 334 #if defined (x86_64)
 335     if (rtnAddr < 0 || shm_ptr == (char*)0xffffffffffffffff)
 336     {
 337         printError(ErrOS, "Shared memory attach returned -ve value %x %d", shm_ptr, errno);
 338         return ErrOS;
 339     }
 340 #else
 341     if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
 342     {
 343         printError(ErrOS, "Shared memory attach returned -ve value %x %d", shm_ptr, errno);
 344         return ErrOS;
 345     }
 346 #endif
 347
 348     db_ = new Database();
 349     db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
 350     db_->setChkptfd(fd);
 351
 352     if (firstThread) ProcessManager::systemDatabase = db_;
 353
 354     printDebug(DM_Database, "Opening database: %s", name);
 355     return OK;
 356 }
 357
 358 DbRetVal DatabaseManagerImpl::closeDatabase()
 359 {
 360
 361     if (NULL == db_)
 362     {
 363         //Database is already closed
 364         return OK;
 365     }
 366     printDebug(DM_Database, "Closing database: %s",(char*)db_->getName());
 367     //check if this is the last thread to be deregistered
 368     int ret =0;// ProcessManager::mutex.getLock(-1, false);
 369     //If you are not getting lock ret !=0, it means somebody else is there.
 370     //he will close the database.
 371     if (0 != strcmp((char*)db_->getName(),  SYSTEMDB)) {
 372         file_desc fd = db_->getChkptfd();
 373         os::closeFile(fd);
 374     }
 375     if (ret == 0) {
 376     if (ProcessManager::noThreads == 0 && 0 == strcmp((char*)db_->getName(), SYSTEMDB)
 377        || ProcessManager::noThreads == 1 && 0 != strcmp((char*)db_->getName(),  SYSTEMDB) )  {
 378                 os::shm_detach((char*)db_->getMetaDataPtr());
 379             }
 380     }
 381     // ProcessManager::mutex.releaseLock(-1, false);
 382     delete db_;
 383     db_ = NULL;
 384     return OK;
 385 }
 386
 387 //Assumes that system database mutex is taken before calling this.
 388 Chunk* DatabaseManagerImpl::createUserChunk(size_t size)
 389 {
 390     //Allocate new node in system database to store
 391     Chunk *chunk = getSystemTableChunk(UserChunkTableId);
 392     DbRetVal rv = OK;
 393     void *ptr = chunk->allocate(systemDatabase_, &rv);
 394     if (NULL == ptr)
 395     {
 396          printError(rv, "Allocation failed for User chunk catalog table");
 397          return NULL;
 398     }
 399     Chunk *chunkInfo = (Chunk*)ptr;
 400     int id = db_->getUniqueIDForChunk();
 401     db_->incrementChunk();
 402     chunkInfo->initMutex(id);
 403     if (0 != size) chunkInfo->setSize(size);
 404     if (chunkInfo->allocSize_ > PAGE_SIZE)
 405         chunkInfo->curPage_ = db_->getFreePage(chunkInfo->allocSize_);
 406     else
 407         chunkInfo->curPage_ = db_->getFreePage();
 408     if ( NULL == chunkInfo->curPage_)
 409     {
 410         chunkInfo->destroyMutex();
 411         chunk->free(db_, ptr);
 412         printError(ErrNoMemory, "Database full: No space to allocate from database");
 413         return NULL;
 414     }
 415     PageInfo* firstPageInfo = ((PageInfo*)chunkInfo->curPage_);
 416     if (chunkInfo->allocSize_ > PAGE_SIZE)
 417     {
 418         int multiple = os::floor(chunkInfo->allocSize_ / PAGE_SIZE);
 419         int offset = ((multiple + 1) * PAGE_SIZE);
 420         firstPageInfo->setPageAsUsed(offset);
 421     }
 422     else
 423     {
 424         firstPageInfo->setPageAsUsed(chunkInfo->allocSize_);
 425         char *data = ((char*)firstPageInfo) + sizeof(PageInfo);
 426         *(InUse*)data =0;
 427     }
 428     if (0 == size)
 429     {
 430         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
 431         varInfo->isUsed_ = 0;
 432         varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
 433
 434     }
 435     chunkInfo->firstPage_ = chunkInfo->curPage_;
 436
 437     if (0 == size)
 438         chunkInfo->setAllocType(VariableSizeAllocator);
 439     else
 440         chunkInfo->setAllocType(FixedSizeAllocator);
 441
 442     chunkInfo->setChunkID(id);
 443     chunkInfo->setPageDirty(firstPageInfo);
 444     printDebug(DM_Database, "Creating new User chunk chunkID:%d size: %d firstPage:%x",
 445                                -1, chunkInfo->allocSize_, firstPageInfo);
 446
 447     return chunkInfo;
 448 }
 449
 450 //Assumes that system database mutex is taken before calling this.
 451 DbRetVal DatabaseManagerImpl::deleteUserChunk(Chunk *chunk)
 452 {
 453     //Go to the pages and set them to notUsed
 454     Page *page = chunk->firstPage_;
 455     PageInfo* pageInfo = ((PageInfo*)page);
 456     //Here...sure that atleast one page will be there even no tuples
 457     //are inserted.so not checking if pageInfo == NULL
 458     while( pageInfo->nextPage_ != NULL)
 459     {
 460         PageInfo *prev = pageInfo;
 461         pageInfo = (PageInfo*)(pageInfo->nextPage_);
 462         //sets pageInfo->isUsed_ = 0 and pageInfo->hasFreeSpace_ = 0
 463         //and initializes the page content to zero
 464         if(NULL == pageInfo->nextPageAfterMerge_){
 465             os::memset(prev, 0, PAGE_SIZE);
 466             SETBIT(prev->flags, IS_DIRTY);
 467
 468         }
 469         else
 470         {
 471             int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 472             char *iter = (char*)prev, *end=(char*)pageInfo->nextPageAfterMerge_;
 473             os::memset(prev, 0, size);
 474             //set dirty bit for all pages in merged pages
 475             while(iter <= end)
 476             {
 477                 PageInfo *info = (PageInfo*) iter;
 478                 SETBIT(info->flags, IS_DIRTY);
 479                 iter = iter  + PAGE_SIZE;
 480             }
 481         }
 482         printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, prev);
 483     }
 484     //The above loop wont execute for the last page
 485     //and for the case where table has only one page
 486     if(NULL == pageInfo->nextPageAfterMerge_) {
 487         os::memset(pageInfo, 0, PAGE_SIZE);
 488         SETBIT(pageInfo->flags, IS_DIRTY);
 489     }
 490     else
 491     {
 492         int size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
 493         char *iter = (char*)pageInfo, *end=(char*)pageInfo->nextPageAfterMerge_;
 494         os::memset(pageInfo, 0, size);
 495         //set dirty bit for all pages in merged pages
 496         while(iter <= end)
 497         {
 498             PageInfo *info = (PageInfo*) iter;
 499             SETBIT(info->flags, IS_DIRTY);
 500             iter = iter  + PAGE_SIZE;
 501         }
 502     }
 503     printDebug(DM_Database,"deleting user chunk:%x clearing page %x",chunk, pageInfo);
 504     chunk->chunkID_ = -1;
 505     chunk->allocSize_ = 0;
 506     chunk->curPage_ = NULL;
 507     chunk->firstPage_ = NULL;
 508     chunk->destroyMutex();
 509     db_->decrementChunk();
 510     Chunk *userChunk = getSystemTableChunk(UserChunkTableId);
 511     userChunk->free(systemDatabase_,chunk);
 512     printDebug(DM_Database,"deleting user chunk:%x",chunk);
 513     return OK;
 514 }
 515
 516 DbRetVal DatabaseManagerImpl::registerThread()
 517 {
 518     DbRetVal rv = OK;
 519     if (pMgr_ != NULL)
 520     {
 521         printError(ErrAlready, "Process already registered\n");
 522         return ErrAlready;
 523     }
 524     pMgr_ = new ProcessManager();
 525     rv = pMgr_->registerThread();
 526     if (rv ==OK) {
 527         procSlot = pMgr_->getProcSlot();
 528         systemDatabase_->setProcSlot(procSlot);
 529         printDebug(DM_Process, "Process registed with slot %d\n", procSlot);
 530     }
 531     return rv;
 532 }
 533
 534 DbRetVal DatabaseManagerImpl::deregisterThread()
 535 {
 536     DbRetVal rv = OK;
 537     if (pMgr_ != NULL)
 538     {
 539         rv = pMgr_->deregisterThread(procSlot);
 540         delete pMgr_;
 541         pMgr_ = NULL;
 542     }
 543     return rv;
 544 }
 545
 546 bool DatabaseManagerImpl::isAnyOneRegistered()
 547 {
 548     if (pMgr_ != NULL) return pMgr_->isAnyOneRegistered();
 549     return true;
 550 }
 551
 552
 553 void DatabaseManagerImpl::printUsageStatistics()
 554 {
 555     pMgr_->printUsageStatistics();
 556     tMgr_->printUsageStatistics();
 557     lMgr_->printUsageStatistics();
 558 }
 559
 560 void DatabaseManagerImpl::printDebugLockInfo()
 561 {
 562     lMgr_->printDebugInfo();
 563 }
 564
 565 void DatabaseManagerImpl::printDebugTransInfo()
 566 {
 567     tMgr_->printDebugInfo(systemDatabase_);
 568 }
 569 void DatabaseManagerImpl::printDebugProcInfo()
 570 {
 571     pMgr_->printDebugInfo();
 572 }
 573 void DatabaseManagerImpl::printDebugMutexInfo()
 574 {
 575     Database *db = sysDb();
 576     db->printDebugMutexInfo();
 577     Chunk *chunk;
 578     int id=1;
 579     printf("<Chunk Mutexes>\n");
 580     chunk=db->getSystemDatabaseChunk(UserChunkTableId);
 581     while(id<MAX_CHUNKS)
 582     {
 583        chunk=db->getSystemDatabaseChunk(id);
 584        if((chunk->getChunkID())!=0){
 585           chunk->printMutexInfo();
 586        }
 587        id++;
 588     }
 589     chunk=db->getSystemDatabaseChunk(UserChunkTableId);
 590     size_t size=chunk->getSize();
 591     int noOfDataNodes=os::floor((PAGE_SIZE - sizeof(PageInfo))/size);
 592     Page* page=chunk->getFirstPage();
 593     int i=0;
 594     Chunk *chk;
 595     while(page)
 596     {
 597         char *data = ((char*)page) + sizeof(PageInfo);
 598         for (i = 0; i< noOfDataNodes; i++)
 599         {
 600             if (*((InUse*)data) == 1)
 601             {
 602                 chk=(Chunk*)((InUse*)data+1);
 603                 chk->printMutexInfo();
 604             }
 605             data = data + size;
 606         }
 607         page = (PageInfo*)(((PageInfo*)page)->nextPage_) ;
 608     }
 609     printf("</Chunk Mutexes>\n");
 610     lMgr_->printMutexInfo();
 611 }
 612 void DatabaseManagerImpl::printDebugChunkInfo()
 613 {
 614     Database *db = sysDb();
 615     Chunk *chunk;
 616     int id=1;
 617     printf("<Chunk information>\n");
 618     printf("  <System Chunk >\n");
 619     chunk=db->getSystemDatabaseChunk(UserChunkTableId);
 620     chunk->print();
 621     while(id<MAX_CHUNKS)
 622     {
 623        chunk=db->getSystemDatabaseChunk(id);
 624        if((chunk->getChunkID())!=0){
 625           chunk->print();
 626        }
 627        id++;
 628     }
 629     printf("  </System Chunk >\n");
 630     printf("  <User Chunk >\n");
 631     chunk=db->getSystemDatabaseChunk(UserChunkTableId);
 632     size_t size=chunk->getSize();
 633     int noOfDataNodes=os::floor((PAGE_SIZE - sizeof(PageInfo))/size);
 634     Page* page=chunk->getFirstPage();
 635     int i=0;
 636     Chunk *chk;
 637     while(page)
 638     {
 639         char *data = ((char*)page) + sizeof(PageInfo);
 640         for (i = 0; i< noOfDataNodes; i++)
 641         {
 642             if (*((InUse*)data) == 1)
 643             {
 644                 chk=(Chunk*)((InUse*)data+1);
 645                 chk->print();
 646             }
 647             data = data + size;
 648         }
 649         page = (PageInfo*)(((PageInfo*)page)->nextPage_) ;
 650     }
 651     printf("  </User Chunk >\n");
 652     printf("</Chunk information>\n");
 653     return;
 654 }
 655 ChunkIterator DatabaseManagerImpl::getSystemTableIterator(CatalogTableID id)
 656 {
 657     Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(id);
 658     return fChunk->getIterator();
 659 }
 660
 661 Chunk* DatabaseManagerImpl::getSystemTableChunk(CatalogTableID id)
 662 {
 663     return systemDatabase_->getSystemDatabaseChunk(id);
 664 }
 665
 666 DbRetVal DatabaseManagerImpl::loadRecords(char *tblName, char *buffer)
 667 {
 668     // buffer should be as big as the no of pages occupied by the records
 669     Table *tbl = openTable(tblName);
 670     if (NULL == tbl) {
 671         printError(ErrSysInternal, "Unable to open table %s", tblName);
 672         return ErrSysInternal;
 673     }
 674     TableImpl *tb = (TableImpl *) tbl;
 675     char *bufIter = buffer;
 676     int pages = *(int *) bufIter; bufIter += sizeof(int);
 677     Page *firstPage = ((Chunk *)(tb->chunkPtr_))->getFirstPage();
 678     PageInfo *pi = (PageInfo *) firstPage;
 679     memcpy(bufIter, pi, PAGE_SIZE);
 680     bufIter += PAGE_SIZE;
 681     for (int i = 0; i < pages - 1; i++) {
 682         Page *nPage = pi->nextPage_;
 683         memcpy(bufIter, nPage, PAGE_SIZE);
 684         bufIter += PAGE_SIZE;
 685         pi = (PageInfo *) nPage;
 686     }
 687     closeTable(tbl);
 688     return OK;
 689 }
 690
 691 DbRetVal DatabaseManagerImpl::pasteRecords(char *tblName, void *buffer)
 692 {
 693     // buffer should be as big as the no of pages occupied by the records
 694     Table *tbl = openTable(tblName);
 695     if (NULL == tbl) {
 696         printError(ErrSysInternal, "Unable to open table %s", tblName);
 697         return ErrSysInternal;
 698     }
 699     TableImpl *tb = (TableImpl *) tbl;
 700     Database *db = tb->getDB();
 701     char *bufIter = (char *) buffer;
 702     int pages = *(int *) bufIter;
 703     bufIter += sizeof(int);
 704
 705     Page *firstPage = ((Chunk *)(tb->chunkPtr_))->getFirstPage();
 706     PageInfo *pi = (PageInfo *) firstPage;
 707     memcpy(pi, bufIter, PAGE_SIZE);
 708     bufIter += PAGE_SIZE;
 709     while (--pages != 0) {
 710         //get a new page allocated
 711         Page *newPage = db->getFreePage();
 712         memcpy(newPage, bufIter, PAGE_SIZE);
 713         pi->nextPage_ = newPage;
 714         pi = (PageInfo *) newPage;
 715     }
 716     // initialize chunk details and pageInfo
 717     ((Chunk *)tb->chunkPtr_)->curPage_ = pi;
 718     closeTable(tbl);
 719     return OK;
 720 }
 721
 722 void DatabaseManagerImpl::setCanTakeCheckPoint(bool ctcp)
 723 { systemDatabase_->setCanTakeCheckPoint(ctcp); }
 724
 725 bool DatabaseManagerImpl::getCanTakeCheckPoint()
 726 { return systemDatabase_->getCanTakeCheckPoint(); }
 727
 728 DbRetVal DatabaseManagerImpl::checkPoint()
 729 {
 730     if (!systemDatabase_->getCanTakeCheckPoint()) {
 731         printf("Load / Cache / Replication process might be running. CheckPoint not taken\n");
 732         return ErrLoadingOn;
 733     }
 734     DbRetVal rv = systemDatabase_->getXCheckpointMutex();
 735     if (OK != rv ) {
 736         printError(rv, "Unable to get checkpoint mutex");
 737         return ErrLockTimeOut;
 738     }
 739     if (tMgr_ && !tMgr_->isTransactionConsistent(systemDatabase_)) {
 740         printf("not in transaction consistent point\n");
 741         systemDatabase_->releaseCheckpointMutex();
 742         return ErrLockTimeOut;
 743     }
 744     rv = writeSchemaFile();
 745     if (rv != OK) {
 746         printError(ErrSysInternal, "checkpoint error");
 747     }
 748     rv = db()->checkPoint();
 749     systemDatabase_->releaseCheckpointMutex();
 750     return rv;
 751 }
 752
 753 DbRetVal DatabaseManagerImpl::writeSchemaFile()
 754 {
 755     DbRetVal rv = OK;
 756     FILE *fp = NULL;
 757     FILE *fp1 = NULL;
 758     int fd = -1;
 759     char schFile[MAX_FILE_LEN];
 760     char mapFile[MAX_FILE_LEN];
 761     sprintf(schFile, "%s/db.chkpt.schema1", Conf::config.getDbFile());
 762     sprintf(mapFile, "%s/db.chkpt.map1", Conf::config.getDbFile());
 763     fp = fopen(schFile, "r");
 764     if (fp != NULL) {
 765         fclose(fp);
 766         int ret = unlink(schFile);
 767         if( ret != 0) {
 768             printError(ErrOS, "checkpoint: delete schema file failed");
 769             return ErrOS;
 770         }
 771     }
 772     fp = fopen(schFile, "w+");
 773     if (fp == NULL) {
 774             printError(ErrOS, "Unable to create schema file for chkpt.");
 775             return ErrOS;
 776     }
 777     fp1 = fopen(mapFile, "r");
 778     if (fp1 != NULL) {
 779         fclose(fp1);
 780         int ret = unlink(mapFile);
 781         if( ret != 0) {
 782             printError(ErrOS, "checkpoint: delete schema file failed");
 783             return ErrOS;
 784         }
 785     }
 786     fd = open(mapFile, O_WRONLY|O_CREAT, 0644);
 787     if (fd == -1) {
 788         printError(ErrOS, "checkpoint: Unable to create map file.");
 789         return ErrOS;
 790     }
 791     List tableList = getAllTableNames();
 792     ListIterator iter = tableList.getIterator();
 793     Identifier *elem = NULL;
 794     int count =0;
 795     while (iter.hasElement()) {
 796         elem = (Identifier*) iter.nextElement();
 797 //        if (TableConf::config.isTableCached(elem->name) == OK) continue;
 798         fprintf(fp, "CREATE TABLE %s (", elem->name);
 799         Table *table = openTable(elem->name);
 800         if (NULL == table) {
 801             printError(ErrSysInternal, "Unable to open table %s", elem->name);
 802             return ErrSysInternal;
 803         }
 804         void *chunk = NULL; void *tptr = NULL; void *vcchunk = NULL;
 805         CatalogTableTABLE cTable(systemDatabase_);
 806         rv = cTable.getChunkAndTblPtr(elem->name, chunk, tptr, vcchunk);
 807         struct Object obj;
 808         strcpy(obj.name, elem->name);
 809         obj.type = Tbl;
 810         obj.bucketChunk = NULL;
 811         obj.firstPage = ((Chunk *)chunk)->getFirstPage();
 812         obj.curPage = ((Chunk *)chunk)->getCurrentPage();
 813         void *buf = &obj;
 814         write(fd, buf, sizeof(obj));
 815         FieldInfo *info = new FieldInfo();
 816         List fNameList = table->getFieldNameList();
 817         ListIterator fNameIter = fNameList.getIterator();
 818         count++;
 819         bool firstField=true;
 820         char fieldName[IDENTIFIER_LENGTH];
 821         while (fNameIter.hasElement()) {
 822             elem = (Identifier*) fNameIter.nextElement();
 823             Table::getFieldNameAlone(elem->name, fieldName);
 824             rv = table->getFieldInfo(elem->name, info);
 825             if (rv !=OK)  {
 826                 printf("unable to retrive info for table %s\n", elem->name);
 827             }
 828             if (firstField) {
 829                 fprintf(fp, "%s %s ", fieldName, AllDataType::getSQLString(info->type));
 830                 firstField = false;
 831             } else
 832                 fprintf(fp, ", %s %s ", fieldName, AllDataType::getSQLString(info->type));
 833             if (info->type == typeString || info->type == typeVarchar ||
 834                                                       info->type == typeBinary)
 835                 fprintf(fp, "(%d)",info->length);
 836             if (info->isNull) fprintf(fp, " NOT NULL ");
 837             if (info->isDefault) fprintf(fp, " DEFAULT '%s' ", info->defaultValueBuf);
 838             if (info->isAutoIncrement) fprintf(fp, " AUTO_INCREMENT ");
 839         }
 840         fNameIter.reset();
 841         while (fNameIter.hasElement())
 842             delete ((Identifier *) fNameIter.nextElement());
 843         fNameList.reset();
 844
 845         fprintf(fp, ");\n");
 846         table->printSQLIndexString(fp, fd);
 847         delete info;
 848         closeTable(table);
 849     }
 850     ListIterator tIter = tableList.getIterator();
 851     tIter.reset();
 852     while (tIter.hasElement())
 853         delete ((Identifier *) tIter.nextElement());
 854     tableList.reset();
 855
 856     fclose(fp);
 857     close(fd);
 858     return OK;
 859 }
 860
 861 DbRetVal DatabaseManagerImpl::recover()
 862 {
 863     DbRetVal rv = OK;
 864     rv = sysDb()->recoverSystemDB();
 865     if (rv != OK) return rv;
 866     if (!Conf::config.useMmap())rv = db()->recoverUserDB();
 867     return rv;
 868 }
 869
 870 void DatabaseManagerImpl::sendSignal(int signal)
 871 {
 872     ThreadInfo* tInfo = sysDb()->getThreadInfo(0);
 873     for (int i=0; i < Conf::config.getMaxProcs(); i++)
 874     {
 875         if (tInfo->pid_ !=0) os::kill(tInfo->pid_, signal);
 876         tInfo++;
 877     }
 878 }