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