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