src/storage/Chunk.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<Allocator.h>
  17 #include<Database.h>
  18 #include<os.h>
  19 #include<Debug.h>
  20 #include<Config.h>
  21 #include<CatalogTables.h>
  22
  23 // sets the size of the Chunk allocator for fixed size
  24 // allocator
  25 // we need one integer to store book keeping information
  26 // whether the storage allocation unit is used of not
  27 // when it is deleted this flag is only set to unused
  28 void Chunk::setSize(size_t size)
  29 {
  30
  31     size_t needSize = size + sizeof(InUse);
  32     size_t multiple = (size_t) os::floor(needSize / sizeof(size_t));
  33     size_t rem = needSize % sizeof(size_t);
  34     if (0 == rem)
  35         allocSize_ = needSize;
  36     else
  37         allocSize_  = (multiple + 1) * sizeof(size_t);
  38 }
  39
  40 void* Chunk::allocateForLargeDataSize(Database *db)
  41 {
  42     PageInfo* pageInfo = ((PageInfo*)curPage_);
  43     DbRetVal ret = db->getAllocDatabaseMutex();
  44     if (ret != 0)
  45     {
  46         printError(ErrLockTimeOut,"Unable to acquire alloc database Mutex");
  47         return NULL;
  48     }
  49
  50     //check whether we have space in curPage
  51     if (BITSET(pageInfo->flags, HAS_SPACE))
  52     {
  53         char *data = ((char*)curPage_) + sizeof(PageInfo);
  54         InUse oldVal = pageInfo->flags;
  55         InUse newVal = oldVal;
  56         CLEARBIT(newVal, HAS_SPACE);
  57         int retVal = Mutex::CASGen(&pageInfo->flags, oldVal, newVal);
  58         if (retVal !=0) printError(ErrSysFatal, "Unable to set flags");
  59         *((InUse*)data) = 1;
  60         db->releaseAllocDatabaseMutex();
  61         return data + sizeof(InUse);
  62     }
  63
  64     //no space in curpage , get new page from database
  65     pageInfo = (PageInfo*)db->getFreePage(allocSize_);
  66     if (NULL == pageInfo)
  67     {
  68         db->releaseAllocDatabaseMutex();
  69         printError(ErrNoMemory,"No more free pages in the database");
  70         return NULL;
  71     }
  72     printDebug(DM_Alloc, "Chunk ID:%d Large Data Item newPage:%x",
  73                                        chunkID_, pageInfo);
  74     int multiple = (int) os::floor(allocSize_ / PAGE_SIZE);
  75     int offset = ((multiple + 1) * PAGE_SIZE);
  76
  77     pageInfo->setPageAsUsed(offset);
  78     setPageDirty(pageInfo);
  79
  80
  81     //create the link
  82     //((PageInfo*)curPage_)->nextPage_ = (Page*) pageInfo;
  83     int retVal = Mutex::CASL((long*) &(((PageInfo*)curPage_)->nextPage_),
  84                (long)(((PageInfo*)curPage_)->nextPage_), (long)pageInfo);
  85     if(retVal !=0) {
  86         printError(ErrLockTimeOut, "Fatal: Unable to create page link.");
  87     }
  88
  89     //Make this as current page
  90     //curPage_ = (Page*) pageInfo;
  91     retVal = Mutex::CASL((long*) &curPage_, (long)curPage_, (long)pageInfo);
  92     if(retVal !=0) {
  93         printError(ErrLockTimeOut, "Fatal:Unable to set current page");
  94     }
  95
  96     char* data = ((char*)curPage_) + sizeof(PageInfo);
  97     InUse oldVal = pageInfo->flags;
  98     InUse newVal = oldVal;
  99     CLEARBIT(newVal, HAS_SPACE);
 100     retVal = Mutex::CASGen(&pageInfo->flags, oldVal, newVal);
 101     if(retVal !=0) {
 102         printError(ErrLockTimeOut, "Fatal:Unable to set flags");
 103     }
 104     *((InUse*)data) = 1;
 105     db->releaseAllocDatabaseMutex();
 106     return data + sizeof(InUse);
 107
 108 }
 109
 110 void* Chunk::allocateFromFirstPage(Database *db, int noOfDataNodes, DbRetVal *status)
 111 {
 112     PageInfo *pageIter = ((PageInfo*)firstPage_);
 113     printDebug(DM_Alloc, "Chunk ID:%d. No free page in database",
 114                                                        chunkID_);
 115     printDebug(DM_Alloc, "Scan from firstPage:%x for free nodes",
 116                                                      firstPage_);
 117     char *data = NULL;
 118     int i = 0;
 119     //scan from first page to locate a free node available
 120     while(NULL != pageIter)
 121     {
 122         data = ((char*)pageIter) + sizeof(PageInfo);
 123         if (BITSET(pageIter->flags, HAS_SPACE))
 124         {
 125             for (i = 0; i< noOfDataNodes ; i++)
 126             {
 127                 if (1 == *((InUse*)data))
 128                     data = data + allocSize_;
 129                 else break;
 130             }
 131             if (i != noOfDataNodes) break;
 132         }
 133         printDebug(DM_Alloc, "Chunk ID: %d Page :%x does not have free nodes",
 134                                                            chunkID_, pageIter);
 135         pageIter = (PageInfo*)((PageInfo*) pageIter)->nextPage_;
 136     }
 137     if (NULL == pageIter) {
 138         *status = ErrNoMemory;
 139         return NULL;
 140     }
 141     printDebug(DM_Alloc,"ChunkID:%d Scan for free node End:Page :%x",
 142                                                   chunkID_, pageIter);
 143     int ret = Mutex::CASGen(data, 0, 1);
 144     if(ret !=0) {
 145         *status = ErrLockTimeOut;
 146         //printError(ErrLockTimeOut, "Unable to allocate from first page. Retry...");
 147         return NULL;
 148     }
 149     setPageDirty(pageIter);
 150     return data + sizeof(InUse);
 151 }
 152
 153 void* Chunk::allocateFromNewPage(Database *db, DbRetVal *status)
 154 {
 155     DbRetVal ret = db->getAllocDatabaseMutex();
 156     if (ret != 0)
 157     {
 158         printDebug(DM_Warning,"Unable to acquire alloc database Mutex Chunkid:%d", chunkID_);
 159         *status = ErrLockTimeOut;
 160         return NULL;
 161     }
 162     //get a new page from db
 163     Page *page = db->getFreePage();
 164     if (page == NULL) {
 165         printError(ErrNoMemory, "Unable to allocate page");
 166         db->releaseAllocDatabaseMutex();
 167         *status = ErrNoMemory;
 168         return NULL;
 169     }
 170     printDebug(DM_Alloc, "ChunkID:%d Normal Data Item newPage:%x",
 171                                                   chunkID_, page);
 172     //Initialize pageInfo for this new page
 173     PageInfo *pInfo = (PageInfo*)page;
 174     pInfo->setPageAsUsed(0);
 175     setPageDirty(pInfo);
 176
 177     char* data = ((char*)page) + sizeof(PageInfo);
 178     *((InUse*)data) = 1;
 179
 180     //create the link between old page and the newly created page
 181     PageInfo* pageInfo = ((PageInfo*)curPage_);
 182
 183     long oldPage = (long)pageInfo->nextPage_;
 184     //pageInfo->nextPage_ = page;
 185     int retVal = Mutex::CASL((long*)&pageInfo->nextPage_ , (long)pageInfo->nextPage_, (long)page);
 186     if(retVal !=0) {
 187         *((InUse*)data) = 0;
 188         pInfo->setPageAsFree();
 189         printDebug(DM_Warning, "Unable to get lock to set chunk list.");
 190         *status = ErrLockTimeOut;
 191         db->releaseAllocDatabaseMutex();
 192         return NULL;
 193     }
 194
 195     //make this new page as the current page
 196     //curPage_ = page;
 197     retVal = Mutex::CASL((long*)&curPage_ , (long)curPage_, (long)page);
 198     if(retVal !=0) {
 199         *((InUse*)data) = 0;
 200         pInfo->setPageAsFree();
 201         retVal = Mutex::CASL((long*)&pageInfo->nextPage_ , (long)pageInfo->nextPage_, (long)oldPage);
 202         if (retVal !=0) printError(ErrSysFatal, "Fatal: Unable to reset the nextPage");
 203         printDebug(DM_Warning, "Unable to get lock to set curPage");
 204         *status = ErrLockTimeOut;
 205         db->releaseAllocDatabaseMutex();
 206         return NULL;
 207     }
 208
 209     db->releaseAllocDatabaseMutex();
 210     return data + sizeof(InUse);
 211 }
 212
 213 //Allocates memory to store data
 214 //TODO::check whether it is locked before allocating.
 215 //delete tuple will set the usedflag to true, but locks will be held
 216 //till commit and it shall be rolledback.So make sure that it does not
 217 //allocate deleted tuple which is yet to be commited.
 218
 219 void* Chunk::allocate(Database *db, DbRetVal *status)
 220 {
 221     PageInfo* pageInfo = ((PageInfo*)curPage_);
 222
 223     int noOfDataNodes=(int) os::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
 224     char *data = ((char*)curPage_) + sizeof(PageInfo);
 225     printDebug(DM_Alloc, "Chunk::allocate id:%d curPage:%x noOfDataNodes:%d",
 226                          chunkID_, curPage_, noOfDataNodes);
 227
 228     //1.scan through data list and find if any is free to use in current page
 229     //2.If there is none then
 230     //    a) get new free page from db. set the prev->next to point
 231     //       to this new page
 232     //4.  b) initialize the free page to zero and get first data ptr.
 233     //5.If there is one, return that
 234
 235     //For allocation more than PAGE_SIZE
 236     if (0 == noOfDataNodes)
 237     {
 238         data = (char*) allocateForLargeDataSize(db);
 239         return data;
 240     }
 241
 242     //Chunk Mutex is required as atomic instructions are not giving
 243     // consistent results.  Need more investigation to improve concurrency
 244     int ret = getChunkMutex(db->procSlot);
 245     if (ret != 0)
 246     {
 247         if (status != NULL) *status = ErrLockTimeOut;
 248         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 249         return NULL;
 250     }
 251     int i = noOfDataNodes;
 252     if (BITSET(pageInfo->flags, HAS_SPACE))
 253     {
 254         for (i = 1; i< noOfDataNodes; i++)
 255         {
 256             if (*((InUse*)data) == 1) data = data + allocSize_;
 257             else break;
 258         }
 259
 260     }
 261     printDebug(DM_Alloc, "ChunkID:%d Node which might be free is %d",
 262                                                         chunkID_, i);
 263     //It comes here if the pageInfo hasFreeSpace
 264     //or there are no free data space in this page
 265     if (i == noOfDataNodes && *((InUse*)data) == 1)
 266     {
 267
 268         printDebug(DM_Alloc, "ChunkID:%d curPage does not have free nodes.", chunkID_);
 269         //there are no free data space in this page
 270         InUse oldVal = pageInfo->flags;
 271         InUse newVal = oldVal;
 272         CLEARBIT(newVal, HAS_SPACE);
 273         int ret = Mutex::CASGen(&pageInfo->flags, oldVal, newVal);
 274         if(ret !=0) {
 275             *status = ErrLockTimeOut;
 276             printDebug(DM_Warning, "Unable to set hasFreespace");
 277             releaseChunkMutex(db->procSlot);
 278             return NULL;
 279         }
 280        *status = OK;
 281        data = (char*) allocateFromFirstPage(db, noOfDataNodes, status);
 282        releaseChunkMutex(db->procSlot);
 283        if (NULL == data && *status != ErrLockTimeOut)
 284        {
 285           *status = OK;
 286           data = (char*) allocateFromNewPage(db, status);
 287           if (data == NULL && *status != ErrLockTimeOut)
 288           {
 289             printError(ErrNoMemory, "No memory in any of the pages:Increase db size");
 290             *status = ErrNoMemory;
 291           }
 292         }
 293         if (*status == OK) setPageDirty(db, data);
 294         return data;
 295     }
 296     int retVal = Mutex::CASGen(data, 0, 1);
 297     if(retVal !=0) {
 298         *status = ErrLockTimeOut;
 299         releaseChunkMutex(db->procSlot);
 300         printDebug(DM_Warning, "Unable to set isUsed : retry...");
 301         return NULL;
 302     }
 303     setPageDirty(db, data);
 304     releaseChunkMutex(db->procSlot);
 305     return data + sizeof(InUse);
 306 }
 307 void* Chunk::tryAllocate(Database *db, DbRetVal *status, int totalTries)
 308 {
 309     int tries=0;
 310     void *node = NULL;
 311     while (tries < totalTries)
 312     {
 313         *status = OK;
 314         node= allocate(db, status);
 315         if (NULL != node) break;
 316         if (*status != ErrLockTimeOut)
 317         {
 318             printError(*status, "Unable to allocate node");
 319             return NULL;
 320         }
 321         tries++;
 322         if (0 == tries %10 )
 323             printError(ErrWarning, "Unable to allocate after %d tries", tries);
 324     }
 325     return node;
 326 }
 327
 328 void Chunk::setPageDirty(Database *db, void *ptr)
 329 {
 330     if (chunkID_ < LastCatalogID) return;
 331     PageInfo *pageInfo = getPageInfo(db, ptr);
 332     if (NULL == pageInfo)
 333     {
 334         printError(ErrSysFatal,"Fatal: pageInfo is NULL %x", ptr );
 335         return;
 336     }
 337     SETBIT(pageInfo->flags, IS_DIRTY);
 338     return;
 339 }
 340 void Chunk::setPageDirty(PageInfo *pInfo)
 341 {
 342     if (chunkID_ < LastCatalogID) return;
 343     SETBIT(pInfo->flags, IS_DIRTY);
 344 }
 345
 346 void* Chunk::allocateForLargeDataSize(Database *db, size_t size)
 347 {
 348     //no need to take chunk mutexes for this, as we are taking alloc database mutex
 349     int multiple = (int) os::floor(size / PAGE_SIZE);
 350     int offset = ((multiple + 1) * PAGE_SIZE);
 351     PageInfo* pageInfo = ((PageInfo*)curPage_);
 352     if(0==allocSize_)
 353         pageInfo = (PageInfo*)db->getFreePage(size);
 354     else
 355         pageInfo = (PageInfo*)db->getFreePage(allocSize_);
 356     if (NULL == pageInfo)
 357     {
 358         printError(ErrNoMemory,"No more free pages in the database:Increase db size");
 359         return NULL;
 360     }
 361     printDebug(DM_VarAlloc,"Chunk::Large Data Item id:%d Size:%d curPage:%x ",
 362                                             chunkID_, size, curPage_);
 363     //TODO:: logic pending
 364     if(allocSize_!=0){
 365         //large size allocate for FixedSize data
 366         pageInfo->nextPageAfterMerge_ = ((char*)pageInfo + offset);
 367         ((PageInfo*)curPage_)->nextPage_ = (Page*) pageInfo;
 368         curPage_ = (Page*) pageInfo;
 369         char* data = ((char*)curPage_) + sizeof(PageInfo);
 370         int ret = Mutex::CASGen(data, 0, 1);
 371         if(ret !=0) {
 372              printError(ErrLockTimeOut, "Lock Timeout: retry...");
 373              return NULL;
 374         }
 375         pageInfo->isUsed_=1;
 376         InUse oldVal = pageInfo->flags;
 377         InUse newVal = oldVal;
 378         CLEARBIT(newVal, HAS_SPACE);
 379         setPageDirty(pageInfo);
 380         ret = Mutex::CASGen(&pageInfo->flags, oldVal, newVal);
 381         if (ret !=0) printError(ErrSysFatal, "Unable to set flags");
 382         return data + sizeof(InUse);
 383     }else{
 384         return allocateForVarLargeSize(pageInfo, size, offset);
 385     }
 386     //REDESIGN MAY BE REQUIRED:Lets us live with this for now.
 387     //what happens to the space lets say 10000 bytes is allocated
 388     //it needs 2 pages,= 16000 bytes, 6000 bytes should not be wasted
 389     //in this case.So need to take of this.
 390     //Will be coded at later stage as this is developed to support
 391     //undo logging and currently we shall assume that the logs generated
 392     //wont be greater than PAGE_SIZE.
 393     return NULL;
 394 }
 395
 396 void Chunk::freeForLargeAllocator(void *ptr, int pslot)
 397 {
 398     //There will be max only one data element in a page.
 399     //PageInfo is stored just before the data.
 400     int ret = getChunkMutex(pslot);
 401     if (ret != 0)
 402     {
 403         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 404         return;
 405     }
 406     PageInfo *pageInfo = (PageInfo*)(((char*)
 407                          ptr) - (sizeof(PageInfo) + sizeof(InUse)));
 408     PageInfo *pInfo = (PageInfo*)firstPage_, *prev = (PageInfo*)firstPage_;
 409     bool found = false;
 410     while(!found)
 411     {
 412         if (pInfo == pageInfo) {found = true;  break; }
 413         prev = pInfo;
 414         pInfo = (PageInfo*)pInfo->nextPage_;
 415     }
 416     if (!found)
 417     {
 418         printError(ErrSysFatal,"Page %x not found in page list:Logical error", pageInfo );
 419         releaseChunkMutex(pslot);
 420         return ;
 421     }
 422     os::memset(((char*)pageInfo+sizeof(PageInfo)),  0 , allocSize_);
 423     if(((PageInfo*)firstPage_)->nextPage_ != NULL){
 424       pageInfo->nextPageAfterMerge_ = NULL;
 425       //pageInfo->isUsed_ = 0;
 426       ret = Mutex::CASGen(&pageInfo->isUsed_, pageInfo->isUsed_, 0);
 427       if (ret != 0) printError(ErrSysFatal, "Unable to set isUsed flag");
 428       InUse oldVal = pageInfo->flags;
 429       InUse newVal = oldVal;
 430       SETBIT(newVal, HAS_SPACE);
 431       ret = Mutex::CASGen(&pageInfo->flags, oldVal, newVal);
 432       if (ret != 0) printError(ErrSysFatal, "Unable to set flags");
 433       if(pageInfo == firstPage_ && ((PageInfo*)firstPage_)->nextPage_ != NULL)
 434       {
 435         //firstPage_ = pageInfo->nextPage_ ;
 436         ret = Mutex::CASL((long*)&firstPage_, (long) firstPage_,
 437                                       (long)pageInfo->nextPage_);
 438         if (ret != 0) printError(ErrSysFatal, "Unable to set firstPage");
 439         setPageDirty(((PageInfo*)firstPage_));
 440       }
 441       else {
 442         //prev->nextPage_ = pageInfo->nextPage_;
 443         ret = Mutex::CASL((long*)&prev->nextPage_, (long) prev->nextPage_,
 444                                        (long)pageInfo->nextPage_);
 445         if (ret != 0) printError(ErrSysFatal, "Unable to set nextPage");
 446         setPageDirty(prev);
 447       }
 448     }
 449     setPageDirty(pageInfo);
 450     releaseChunkMutex(pslot);
 451     return;
 452 }
 453
 454 //Frees the memory pointed by ptr
 455 void Chunk::free(Database *db, void *ptr)
 456 {
 457     if (0 == allocSize_)
 458     {
 459         freeForVarSizeAllocator(db, ptr, db->procSlot);
 460         return;
 461     }
 462     int noOfDataNodes =(int) os::floor((PAGE_SIZE - sizeof(PageInfo)) / allocSize_);
 463
 464     if (0 == noOfDataNodes)
 465     {
 466         freeForLargeAllocator(ptr, db->procSlot);
 467         return;
 468     }
 469     int ret = getChunkMutex(db->procSlot);
 470     if (ret != 0)
 471     {
 472         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 473         return;
 474     }
 475     //unset the used flag
 476     //*((int*)ptr -1 ) = 0;
 477     InUse oldValue=1;
 478     if (*((InUse*)ptr -1 ) == 0) {
 479         printError(ErrSysFatal, "Fatal:Data node already freed %x Chunk:%d value:%d", ptr, chunkID_,
 480                                                                                     *((InUse*)ptr -1 ));
 481         oldValue=0;
 482         //return;
 483     }
 484     int retVal = Mutex::CASGen(((InUse*)ptr -1), oldValue, 0);
 485     if(retVal !=0) {
 486         printError(ErrSysFatal, "Unable to get lock to free for %x", ptr);
 487         releaseChunkMutex(db->procSlot);
 488         return;
 489     }
 490     PageInfo *pageInfo;
 491     pageInfo = getPageInfo(db, ptr);
 492     if (NULL == pageInfo)
 493     {
 494         printError(ErrSysFatal,"Fatal: pageInfo is NULL", pageInfo );
 495         releaseChunkMutex(db->procSlot);
 496         return;
 497     }
 498     //set the pageinfo where this ptr points
 499     InUse oldVal = pageInfo->flags;
 500     InUse newVal = oldVal;
 501     SETBIT(newVal, HAS_SPACE);
 502     retVal = Mutex::CASGen(&pageInfo->flags, oldVal, newVal);
 503     if(retVal !=0) {
 504         printError(ErrSysFatal, "Unable to get lock to set flags");
 505     }
 506     setPageDirty(pageInfo);
 507     releaseChunkMutex(db->procSlot);
 508     return;
 509 }
 510
 511 //returns the pageInfo of the page where this ptr points
 512 //This works only if the data size is less than PAGE_SIZE
 513 //If ptr points to data which is more than PAGE_SIZE,then
 514 //calling this might lead to memory corruption
 515 //Note:IMPORTANT::assumes db lock is taken before calling this
 516 PageInfo* Chunk::getPageInfo(Database *db, void *ptr)
 517 {
 518     if (allocSize_ < PAGE_SIZE - sizeof(PageInfo)) {
 519        int rem = (long) ptr % PAGE_SIZE;
 520        return (PageInfo*)(((char*)ptr) - rem);
 521     } else {
 522         //large size allocator
 523         char *inPtr = (char*)ptr;
 524         PageInfo* pageInfo = ((PageInfo*)firstPage_);
 525
 526         while( pageInfo != NULL )
 527         {
 528           if (inPtr > (char*) pageInfo && pageInfo->nextPageAfterMerge_ >inPtr)
 529             return pageInfo;
 530         pageInfo = (PageInfo*)pageInfo->nextPage_ ;
 531         }
 532     }
 533     return NULL;
 534 }
 535
 536 //If called on chunk used to store tuples, it returns the total number of rows
 537 //present in the table
 538 long Chunk::getTotalDataNodes()
 539 {
 540     long totalNodes =0;
 541     if (0 == allocSize_) //->variable size allocator
 542     {
 543         return getVarTotalDataNodes();
 544     }
 545
 546     //TODO::for large size allocator
 547     if (allocSize_ >PAGE_SIZE)//->each page has only one data node
 548     {
 549         Page *page = ((PageInfo*)firstPage_);
 550         while(NULL != page)
 551         {
 552                 //current it page wise  later this will done
 553                 if(1==*(int*)(((char*)page)+sizeof(PageInfo)))
 554                         totalNodes++;
 555                 page = ((PageInfo*) page)->nextPage_;
 556         }
 557         return totalNodes;
 558     }
 559
 560     int noOfDataNodes=(int) os::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
 561     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 562     char *data = ((char*)firstPage_) + sizeof(PageInfo);
 563     int i=0;
 564     while( pageInfo != NULL )
 565     {
 566         data = ((char*)pageInfo) + sizeof(PageInfo);
 567         for (i = 0; i< noOfDataNodes; i++)
 568         {
 569             if (*((InUse*)data) == 1) { totalNodes++;}
 570             data = data + allocSize_;
 571         }
 572         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 573     }
 574     return totalNodes;
 575 }
 576
 577 //TODO::for other type of allocators
 578 int Chunk::compact(int procSlot)
 579 {
 580     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 581     PageInfo* prevPage = pageInfo;
 582     if (NULL == pageInfo)
 583     {
 584         return 0;
 585     }
 586     int ret = getChunkMutex(procSlot);
 587     if (ret != 0)
 588     {
 589         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 590         return ret;
 591     }
 592     pageInfo = (PageInfo*)pageInfo->nextPage_;
 593     if (0 == allocSize_)
 594     {
 595         varCompact();
 596     }else if (allocSize_ < PAGE_SIZE)
 597     {
 598       while( pageInfo != NULL )
 599       {
 600         bool flag = false;
 601         int noOfDataNodes=(int) os::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
 602         char *data = ((char*)pageInfo) + sizeof(PageInfo);
 603         for (int i = 0; i< noOfDataNodes ; i++)
 604         {
 605             if (1 == *((InUse*)data)) { flag = true; break; }
 606             data = data +allocSize_;
 607         }
 608         if (!flag) {
 609             printDebug(DM_Alloc,"Freeing unused page in fixed allocator %x\n", pageInfo);
 610             prevPage->nextPage_ = pageInfo->nextPage_;
 611             pageInfo->isUsed_ = 0;
 612             pageInfo = (PageInfo*)(((PageInfo*)prevPage)->nextPage_) ;
 613         }else{
 614             prevPage = pageInfo;
 615             pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 616         }
 617         printDebug(DM_Alloc,"compact iter %x\n", pageInfo);
 618       }
 619     }
 620     releaseChunkMutex(procSlot);
 621     return 0;
 622 }
 623
 624 int Chunk::totalPages()
 625 {
 626     //logic is same for variable size and for large data node allocator.
 627     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 628     int totPages=0;
 629     while( pageInfo != NULL )
 630     {
 631         totPages++;
 632         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 633     }
 634     return totPages;
 635 }
 636
 637 int Chunk::totalDirtyPages()
 638 {
 639     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 640     int dirtyPages=0;
 641     while( pageInfo != NULL )
 642     {
 643         if(BITSET(pageInfo->flags, IS_DIRTY)) dirtyPages++;
 644         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 645     }
 646     return dirtyPages;
 647 }
 648
 649
 650 int Chunk::initMutex(int id)
 651 {
 652     char mutName[IDENTIFIER_LENGTH];
 653     if (-1 == id)
 654        sprintf(mutName, "Chunk");
 655     else
 656        sprintf(mutName, "Chunk:%d",id);
 657     return chunkMutex_.init(mutName);
 658 }
 659
 660 int Chunk::getChunkMutex(int procSlot)
 661 {
 662     return chunkMutex_.getLock(procSlot);
 663 }
 664
 665 int Chunk::releaseChunkMutex(int procSlot)
 666 {
 667     return chunkMutex_.releaseLock(procSlot);
 668 }
 669
 670 int Chunk::destroyMutex()
 671 {
 672     return chunkMutex_.destroy();
 673 }
 674
 675 void Chunk::setChunkNameForSystemDB(int id)
 676 {
 677     strcpy(chunkName,ChunkName[id]);
 678 }
 679
 680 void Chunk::print()
 681 {
 682     printf("        <Chunk Id> %d </Chunk Id> \n",chunkID_);
 683     printf("                <TotalPages> %d </TotalPages> \n",totalPages());
 684     if (Conf::config.useDurability())
 685         printf("                <DirtyPages> %d </DirtyPages> \n",totalDirtyPages());
 686     printf("                <ChunkName > %s </ChunkName> \n",getChunkName());
 687     printf("                <TotalDataNodes> %d </TotalDataNodes> \n",getTotalDataNodes());
 688     printf("                <SizeOfDataNodes> %d </SizeOfDataNodes> \n",getSize());
 689     printf("                <Allocation Type> ");
 690     if(allocType_==0)
 691     {
 692         printf("FixedSizeAllocator ");
 693     }else if(allocType_==1)
 694     {
 695         printf("VariableSizeAllocator ");
 696     }else
 697     {
 698         printf("UnknownAllocator ");
 699     }
 700     printf("</Allocation Type>\n");
 701 }