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         //large size allocate for varSize data
 385         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)pageInfo) + sizeof(PageInfo));
 386         pageInfo->nextPageAfterMerge_ = ((char*)pageInfo + offset);
 387         ((PageInfo*)curPage_)->nextPage_ = (Page*) pageInfo;
 388         curPage_ = (Page*) pageInfo;
 389         varInfo->size_= size;
 390         int ret = Mutex::CASGen(&varInfo->isUsed_ , varInfo->isUsed_, 1);
 391         if(ret !=0) {
 392              printError(ErrLockTimeOut, "Unable to get lock for var alloc. Retry...");
 393              return NULL;
 394         }
 395         pageInfo->isUsed_=1;
 396         setPageDirty(pageInfo);
 397         return (char *) varInfo + sizeof(VarSizeInfo);
 398     }
 399     //REDESIGN MAY BE REQUIRED:Lets us live with this for now.
 400     //what happens to the space lets say 10000 bytes is allocated
 401     //it needs 2 pages,= 16000 bytes, 6000 bytes should not be wasted
 402     //in this case.So need to take of this.
 403     //Will be coded at later stage as this is developed to support
 404     //undo logging and currently we shall assume that the logs generated
 405     //wont be greater than PAGE_SIZE.
 406     return NULL;
 407 }
 408
 409
 410 //Will check from first page to current page and do first fit allocate
 411 //if no space available, allocates new page
 412 void* Chunk::allocFromNewPageForVarSize(Database *db, size_t size, int pslot, DbRetVal *rv)
 413 {
 414     //Should be called only for data items <PAGE_SIZE
 415     void *vnode = varSizeFirstFitAllocate(size, pslot, rv);
 416     if (vnode != NULL)
 417     {
 418        return vnode;
 419     }
 420     printDebug(DM_Warning, "No free space in any of the pages already being used");
 421     *rv =OK;
 422     DbRetVal ret = db->getAllocDatabaseMutex();
 423     if (ret != 0)
 424     {
 425         printError(ErrLockTimeOut,"Unable to acquire alloc database Mutex");
 426         return NULL;
 427     }
 428     Page *newPage = db->getFreePage();
 429     if (NULL == newPage)
 430     {
 431         db->releaseAllocDatabaseMutex();
 432         return NULL;
 433     }
 434
 435     printDebug(DM_VarAlloc, "ChunkID:%d New Page: %x ", chunkID_, newPage);
 436     PageInfo *pInfo = (PageInfo*) newPage;
 437     pInfo->setPageAsUsed(0);
 438     setPageDirty(pInfo);
 439     SETBIT(pInfo->flags, HAS_SPACE);
 440     if (1 == createDataBucket(newPage, PAGE_SIZE, size, pslot))
 441     {
 442         printError(ErrSysFatal, "Split failed in new page...Should never happen");
 443         *rv = ErrSysFatal;
 444         db->releaseAllocDatabaseMutex();
 445         return NULL;
 446     }
 447     long oldPage = (long)((PageInfo*)curPage_)->nextPage_;
 448     //((PageInfo*)curPage_)->nextPage_ = newPage;
 449     int retVal = Mutex::CASL((long*) &(((PageInfo*)curPage_)->nextPage_),
 450                (long)(((PageInfo*)curPage_)->nextPage_), (long)newPage);
 451     if(retVal !=0) {
 452         printError(ErrSysFatal, "Unable to get lock to set chunk next page");
 453         pInfo->setPageAsFree();
 454         db->releaseAllocDatabaseMutex();
 455         *rv = ErrSysFatal;
 456         return NULL;
 457     }
 458     //curPage_ = newPage;
 459     retVal = Mutex::CASL((long*) &curPage_, (long)curPage_, (long)newPage);
 460     if(retVal !=0) {
 461         printError(ErrSysFatal, "Unable to get lock to set curPage");
 462         retVal = Mutex::CASL((long*) &(((PageInfo*)curPage_)->nextPage_),
 463                (long)(((PageInfo*)curPage_)->nextPage_), (long)oldPage);
 464         if (retVal !=0 ) printError(ErrSysFatal, "Unable to reset curPage");
 465         pInfo->setPageAsFree();
 466         db->releaseAllocDatabaseMutex();
 467         *rv = ErrSysFatal;
 468         return NULL;
 469     }
 470     db->releaseAllocDatabaseMutex();
 471     char *data= ((char*)newPage) + sizeof(PageInfo) + sizeof(VarSizeInfo);
 472     return data;
 473 }
 474
 475 //Allocates from the current page of the chunk.
 476 //Scans through the VarSizeInfo objects in the page and gets the free slot
 477 void* Chunk::allocateFromCurPageForVarSize(size_t size, int pslot, DbRetVal *rv)
 478 {
 479     //Should be called only for data items <PAGE_SIZE
 480     Page *page = ((PageInfo*)curPage_);
 481     printDebug(DM_VarAlloc, "Chunk::allocate Normal Data Item id:%d Size:%d curPage:%x ",
 482                                                chunkID_, size, curPage_);
 483     VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) +
 484                                                 sizeof(PageInfo));
 485     //if ( 0 != getChunkMutex(pslot)) { *rv = ErrLockTimeOut; return NULL; }
 486     int pageSize = PAGE_SIZE;
 487     bool hasSpace= false;
 488     while ((char*) varInfo < ((char*)page + pageSize))
 489     {
 490         if (0 == varInfo->isUsed_)
 491         {
 492             hasSpace= true;
 493             if( size + sizeof(VarSizeInfo) < varInfo->size_)
 494             {
 495                 if (1 == splitDataBucket(varInfo, size, pslot, rv))
 496                 {
 497                     printDebug(DM_Warning, "Unable to split the data bucket");
 498                     //releaseChunkMutex(pslot);
 499                     return NULL;
 500                 }
 501                 printDebug(DM_VarAlloc, "Chunkid:%d splitDataBucket: Size: %d Item:%x ",
 502                                                          chunkID_, size, varInfo);
 503                 //releaseChunkMutex(pslot);
 504                 return (char*)varInfo + sizeof(VarSizeInfo);
 505             }
 506             else if (size == varInfo->size_) {
 507                 //varInfo->isUsed_ = 1;
 508                 int ret = Mutex::CASGen(&varInfo->isUsed_ , 0, 1);
 509                 if(ret !=0) {
 510                      printDebug(DM_Warning, "Unable to get lock for var alloc size:%d ", size);
 511                      *rv = ErrLockTimeOut;
 512                      //releaseChunkMutex(pslot);
 513                      return NULL;
 514                 }
 515                 //releaseChunkMutex(pslot);
 516                 return (char *) varInfo + sizeof(VarSizeInfo);
 517             }
 518
 519         }
 520         varInfo = (VarSizeInfo*)((char*)varInfo + sizeof(VarSizeInfo)
 521                                     +varInfo->size_);
 522     }
 523     if (!hasSpace) CLEARBIT(((PageInfo*)curPage_)->flags, HAS_SPACE);
 524     if (hasSpace && size < MIN_VARCHAR_ALLOC_SIZE)
 525                    CLEARBIT(((PageInfo*)curPage_)->flags, HAS_SPACE);
 526     //releaseChunkMutex(pslot);
 527     return NULL;
 528 }
 529
 530 //Allocates memory to store data of variable size
 531 void* Chunk::allocate(Database *db, size_t size, DbRetVal *status)
 532 {
 533     if (0 == size) return NULL;
 534     //check if the size is more than PAGE_SIZE
 535     //if it is more than the PAGE_SIZE, then allocate new
 536     //page using database and then link the curPage to the
 537     //newly allocated page
 538     //if it is less than PAGE_SIZE, then check the curpage for
 539     //free memory of specified size
 540     //if not available, then scan from the firstPage for the free
 541     //space
 542
 543     //TODO::During the scan, merge nearby nodes if both are free
 544     //if not available then allocate new page
 545
 546     size_t alignedSize = os::alignLong(size);
 547     void *data = NULL;
 548     int ret = getChunkMutex(db->procSlot);
 549     if (ret != 0)
 550     {
 551         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 552         *status = ErrLockTimeOut;
 553         return NULL;
 554     }
 555     if (alignedSize > PAGE_SIZE )
 556     {
 557         data =  allocateForLargeDataSize(db, alignedSize);
 558     }
 559     else
 560     {
 561         data = allocateFromCurPageForVarSize(alignedSize, db->procSlot, status);
 562         if (NULL == data) {
 563             *status = OK;
 564             //No available spaces in the current page.
 565             //allocate new page
 566             data= allocFromNewPageForVarSize(db, alignedSize, db->procSlot, status);
 567             if (NULL == data && *status !=ErrLockTimeOut) {
 568                 printError(ErrNoMemory, "No memory in any of the pages:Increase db size");
 569                 *status = ErrNoMemory;
 570             }
 571         }
 572     }
 573     releaseChunkMutex(db->procSlot);
 574     return data;
 575 }
 576
 577 //Assumes chunk mutex is already taken, before calling this
 578 void* Chunk::varSizeFirstFitAllocate(size_t size, int pslot, DbRetVal *rv)
 579 {
 580     printDebug(DM_VarAlloc, "Chunk::varSizeFirstFitAllocate size:%d firstPage:%x",
 581                                                size, firstPage_);
 582
 583     Page *page = ((PageInfo*)firstPage_);
 584     size_t alignedSize = os::alignLong(size);
 585     //if ( 0 != getChunkMutex(pslot)) { *rv = ErrLockTimeOut; return NULL; }
 586     int pageSize = PAGE_SIZE;
 587     bool hasSpace=false;
 588     while(NULL != page)
 589     {
 590       VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) + sizeof(PageInfo));
 591       hasSpace=false;
 592       if (BITSET(((PageInfo*)page)->flags, HAS_SPACE)){
 593         while ((char*) varInfo < ((char*)page + pageSize))
 594         {
 595             if (0 == varInfo->isUsed_)
 596             {
 597                 hasSpace=true;
 598                 if( alignedSize +sizeof(VarSizeInfo) < varInfo->size_)
 599                 {
 600                     if( 1 == splitDataBucket(varInfo, alignedSize, pslot, rv))
 601                     {
 602                         printDebug(DM_Warning, "Unable to split the data bucket");
 603                         //releaseChunkMutex(pslot);
 604                         return NULL;
 605                     }
 606                     //releaseChunkMutex(pslot);
 607                     return ((char*)varInfo) + sizeof(VarSizeInfo);
 608                 }
 609                 else if (alignedSize == varInfo->size_) {
 610                     //varInfo->isUsed_ = 1;
 611                     int ret = Mutex::CASGen(&varInfo->isUsed_, 0, 1);
 612                     if(ret !=0) {
 613                        printDebug(DM_Warning,"Unable to get lock to set isUsed flag.");
 614                        *rv = ErrLockTimeOut;
 615                        //releaseChunkMutex(pslot);
 616                        return NULL;
 617                     }
 618                     printDebug(DM_VarAlloc, "VarSizeFirstFitAllocate returning %x", ((char*)varInfo) +sizeof(VarSizeInfo));
 619                     //releaseChunkMutex(pslot);
 620                     return ((char *) varInfo) + sizeof(VarSizeInfo);
 621                 }
 622             }
 623             varInfo = (VarSizeInfo*)((char*)varInfo + sizeof(VarSizeInfo)
 624                                     +varInfo->size_);
 625         }
 626         if (!hasSpace) CLEARBIT(((PageInfo*)page)->flags, HAS_SPACE);
 627         if (hasSpace && size < MIN_VARCHAR_ALLOC_SIZE)
 628                                   CLEARBIT(((PageInfo*)page)->flags, HAS_SPACE);
 629       }
 630       printDebug(DM_VarAlloc, "Chunk:This page does not have free data nodes page:%x", page);
 631       page = ((PageInfo*) page)->nextPage_;
 632     }
 633     //releaseChunkMutex(pslot);
 634     return NULL;
 635 }
 636
 637 void Chunk::freeForVarSizeAllocator(Database *db, void *ptr, int pslot)
 638 {
 639     int ret = getChunkMutex(pslot);
 640     if (ret != 0)
 641     {
 642         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 643         return;
 644     }
 645     VarSizeInfo *varInfo = (VarSizeInfo*)((char*)ptr- sizeof(VarSizeInfo));
 646     //varInfo->isUsed_ = 0;
 647     if(varInfo->size_ > (PAGE_SIZE - (sizeof(VarSizeInfo)+sizeof(PageInfo)))) {
 648         PageInfo *pageInfo = (PageInfo*)((char*)varInfo - sizeof(PageInfo));
 649         PageInfo *pInfo = (PageInfo*)firstPage_, *prev = (PageInfo*)firstPage_;
 650         bool found = false;
 651         while(!found)
 652         {
 653             if(NULL==pInfo) break;
 654             if (pInfo == pageInfo) {found = true;  break; }
 655             prev = pInfo;
 656             pInfo = (PageInfo*)pInfo->nextPage_;
 657         }
 658         if (!found)
 659         {
 660              printError(ErrSysFatal,"Page %x not found in page list:Logical error", pageInfo );
 661              releaseChunkMutex(pslot);
 662              return ;
 663         }
 664          if(curPage_== pageInfo) {curPage_ = prev ; }
 665          pageInfo->isUsed_ = 0;
 666          pageInfo->nextPageAfterMerge_ = NULL;
 667          setPageDirty(pageInfo);
 668          SETBIT(pageInfo->flags, HAS_SPACE);
 669          prev->nextPage_ = pageInfo->nextPage_;
 670     }
 671     int retVal = Mutex::CASGen(&varInfo->isUsed_, 1, 0);
 672     if(retVal !=0) {
 673         printError(ErrAlready, "Fatal: Varsize double free for %x", ptr);
 674     }
 675     PageInfo *pageInfo = getPageInfo(db, ptr);
 676     if (NULL == pageInfo)
 677     {
 678         printError(ErrSysFatal,"Fatal: pageInfo is NULL", pageInfo );
 679         releaseChunkMutex(db->procSlot);
 680         return;
 681     }
 682     SETBIT(pageInfo->flags, HAS_SPACE);
 683     SETBIT(pageInfo->flags, IS_DIRTY);
 684     printDebug(DM_VarAlloc,"chunkID:%d Unset isUsed for %x", chunkID_, varInfo);
 685     releaseChunkMutex(pslot);
 686     return;
 687
 688 }
 689
 690 void Chunk::freeForLargeAllocator(void *ptr, int pslot)
 691 {
 692     //There will be max only one data element in a page.
 693     //PageInfo is stored just before the data.
 694     int ret = getChunkMutex(pslot);
 695     if (ret != 0)
 696     {
 697         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 698         return;
 699     }
 700     PageInfo *pageInfo = (PageInfo*)(((char*)
 701                          ptr) - (sizeof(PageInfo) + sizeof(InUse)));
 702     PageInfo *pInfo = (PageInfo*)firstPage_, *prev = (PageInfo*)firstPage_;
 703     bool found = false;
 704     while(!found)
 705     {
 706         if (pInfo == pageInfo) {found = true;  break; }
 707         prev = pInfo;
 708         pInfo = (PageInfo*)pInfo->nextPage_;
 709     }
 710     if (!found)
 711     {
 712         printError(ErrSysFatal,"Page %x not found in page list:Logical error", pageInfo );
 713         releaseChunkMutex(pslot);
 714         return ;
 715     }
 716     os::memset(((char*)pageInfo+sizeof(PageInfo)),  0 , allocSize_);
 717     if(((PageInfo*)firstPage_)->nextPage_ != NULL){
 718       pageInfo->nextPageAfterMerge_ = NULL;
 719       //pageInfo->isUsed_ = 0;
 720       ret = Mutex::CASGen(&pageInfo->isUsed_, pageInfo->isUsed_, 0);
 721       if (ret != 0) printError(ErrSysFatal, "Unable to set isUsed flag");
 722       InUse oldVal = pageInfo->flags;
 723       InUse newVal = oldVal;
 724       SETBIT(newVal, HAS_SPACE);
 725       ret = Mutex::CASGen(&pageInfo->flags, oldVal, newVal);
 726       if (ret != 0) printError(ErrSysFatal, "Unable to set flags");
 727       if(pageInfo == firstPage_ && ((PageInfo*)firstPage_)->nextPage_ != NULL)
 728       {
 729         //firstPage_ = pageInfo->nextPage_ ;
 730         ret = Mutex::CASL((long*)&firstPage_, (long) firstPage_,
 731                                       (long)pageInfo->nextPage_);
 732         if (ret != 0) printError(ErrSysFatal, "Unable to set firstPage");
 733         setPageDirty(((PageInfo*)firstPage_));
 734       }
 735       else {
 736         //prev->nextPage_ = pageInfo->nextPage_;
 737         ret = Mutex::CASL((long*)&prev->nextPage_, (long) prev->nextPage_,
 738                                        (long)pageInfo->nextPage_);
 739         if (ret != 0) printError(ErrSysFatal, "Unable to set nextPage");
 740         setPageDirty(prev);
 741       }
 742     }
 743     setPageDirty(pageInfo);
 744     releaseChunkMutex(pslot);
 745     return;
 746 }
 747
 748 //Frees the memory pointed by ptr
 749 void Chunk::free(Database *db, void *ptr)
 750 {
 751     if (0 == allocSize_)
 752     {
 753         freeForVarSizeAllocator(db, ptr, db->procSlot);
 754         return;
 755     }
 756     int noOfDataNodes =(int) os::floor((PAGE_SIZE - sizeof(PageInfo)) / allocSize_);
 757
 758     if (0 == noOfDataNodes)
 759     {
 760         freeForLargeAllocator(ptr, db->procSlot);
 761         return;
 762     }
 763     int ret = getChunkMutex(db->procSlot);
 764     if (ret != 0)
 765     {
 766         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 767         return;
 768     }
 769     //unset the used flag
 770     //*((int*)ptr -1 ) = 0;
 771     InUse oldValue=1;
 772     if (*((InUse*)ptr -1 ) == 0) {
 773         printError(ErrSysFatal, "Fatal:Data node already freed %x Chunk:%d value:%d", ptr, chunkID_,
 774                                                                                     *((InUse*)ptr -1 ));
 775         oldValue=0;
 776         //return;
 777     }
 778     int retVal = Mutex::CASGen(((InUse*)ptr -1), oldValue, 0);
 779     if(retVal !=0) {
 780         printError(ErrSysFatal, "Unable to get lock to free for %x", ptr);
 781         releaseChunkMutex(db->procSlot);
 782         return;
 783     }
 784     PageInfo *pageInfo;
 785     pageInfo = getPageInfo(db, ptr);
 786     if (NULL == pageInfo)
 787     {
 788         printError(ErrSysFatal,"Fatal: pageInfo is NULL", pageInfo );
 789         releaseChunkMutex(db->procSlot);
 790         return;
 791     }
 792     //set the pageinfo where this ptr points
 793     InUse oldVal = pageInfo->flags;
 794     InUse newVal = oldVal;
 795     SETBIT(newVal, HAS_SPACE);
 796     retVal = Mutex::CASGen(&pageInfo->flags, oldVal, newVal);
 797     if(retVal !=0) {
 798         printError(ErrSysFatal, "Unable to get lock to set flags");
 799     }
 800     setPageDirty(pageInfo);
 801     releaseChunkMutex(db->procSlot);
 802     return;
 803 }
 804
 805 //returns the pageInfo of the page where this ptr points
 806 //This works only if the data size is less than PAGE_SIZE
 807 //If ptr points to data which is more than PAGE_SIZE,then
 808 //calling this might lead to memory corruption
 809 //Note:IMPORTANT::assumes db lock is taken before calling this
 810 PageInfo* Chunk::getPageInfo(Database *db, void *ptr)
 811 {
 812     if (allocSize_ < PAGE_SIZE - sizeof(PageInfo)) {
 813        int rem = (long) ptr % PAGE_SIZE;
 814        return (PageInfo*)(((char*)ptr) - rem);
 815     } else {
 816         //large size allocator
 817         char *inPtr = (char*)ptr;
 818         PageInfo* pageInfo = ((PageInfo*)firstPage_);
 819
 820         while( pageInfo != NULL )
 821         {
 822           if (inPtr > (char*) pageInfo && pageInfo->nextPageAfterMerge_ >inPtr)
 823             return pageInfo;
 824         pageInfo = (PageInfo*)pageInfo->nextPage_ ;
 825         }
 826     }
 827     return NULL;
 828 }
 829
 830 //If called on chunk used to store tuples, it returns the total number of rows
 831 //present in the table
 832 long Chunk::getTotalDataNodes()
 833 {
 834     long totalNodes =0;
 835     if (0 == allocSize_) //->variable size allocator
 836     {
 837         Page *page = ((PageInfo*)firstPage_);
 838         while(NULL != page)
 839         {
 840             VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) + sizeof(PageInfo));
 841             while ((char*) varInfo < ((char*)page + PAGE_SIZE))
 842             {
 843                 if (1 == varInfo->isUsed_) totalNodes++;
 844                 varInfo = (VarSizeInfo*)((char*)varInfo + sizeof(VarSizeInfo)
 845                                     +varInfo->size_);
 846             }
 847             page = ((PageInfo*) page)->nextPage_;
 848         }
 849         return totalNodes;
 850     }
 851
 852     //TODO::for large size allocator
 853     if (allocSize_ >PAGE_SIZE)//->each page has only one data node
 854     {
 855         Page *page = ((PageInfo*)firstPage_);
 856         while(NULL != page)
 857         {
 858                 //current it page wise  later this will done
 859                 if(1==*(int*)(((char*)page)+sizeof(PageInfo)))
 860                         totalNodes++;
 861                 page = ((PageInfo*) page)->nextPage_;
 862         }
 863         return totalNodes;
 864     }
 865
 866     int noOfDataNodes=(int) os::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
 867     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 868     char *data = ((char*)firstPage_) + sizeof(PageInfo);
 869     int i=0;
 870     while( pageInfo != NULL )
 871     {
 872         data = ((char*)pageInfo) + sizeof(PageInfo);
 873         for (i = 0; i< noOfDataNodes; i++)
 874         {
 875             if (*((InUse*)data) == 1) { totalNodes++;}
 876             data = data + allocSize_;
 877         }
 878         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 879     }
 880     return totalNodes;
 881 }
 882
 883 //TODO::for other type of allocators
 884 int Chunk::compact(int procSlot)
 885 {
 886     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 887     PageInfo* prevPage = pageInfo;
 888     if (NULL == pageInfo)
 889     {
 890         return 0;
 891     }
 892     int ret = getChunkMutex(procSlot);
 893     if (ret != 0)
 894     {
 895         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 896         return ret;
 897     }
 898     pageInfo = (PageInfo*)pageInfo->nextPage_;
 899     if (0 == allocSize_)
 900     {
 901       while( pageInfo != NULL )
 902       {
 903         bool flag = false;
 904         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)pageInfo) +
 905                                             sizeof(PageInfo));
 906         while ((char*) varInfo < ((char*)pageInfo + PAGE_SIZE))
 907         {
 908             if (1 == varInfo->isUsed_) {flag=true; break;}
 909             varInfo = (VarSizeInfo*)((char*)varInfo + sizeof(VarSizeInfo)
 910                                    +varInfo->size_);
 911         }
 912         if (!flag) {
 913             printDebug(DM_VarAlloc,"Freeing unused page in varsize allocator %x\n", pageInfo);
 914             prevPage->nextPage_ = pageInfo->nextPage_;
 915             pageInfo->isUsed_ = 0;
 916         }
 917         prevPage = pageInfo;
 918         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 919         printDebug(DM_VarAlloc,"compact iter %x\n", pageInfo);
 920       }
 921     }else if (allocSize_ < PAGE_SIZE)
 922     {
 923       while( pageInfo != NULL )
 924       {
 925         bool flag = false;
 926         int noOfDataNodes=(int) os::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
 927         char *data = ((char*)pageInfo) + sizeof(PageInfo);
 928         for (int i = 0; i< noOfDataNodes ; i++)
 929         {
 930             if (1 == *((InUse*)data)) { flag = true; break; }
 931             data = data +allocSize_;
 932         }
 933         if (!flag) {
 934             printDebug(DM_Alloc,"Freeing unused page in fixed allocator %x\n", pageInfo);
 935             prevPage->nextPage_ = pageInfo->nextPage_;
 936             pageInfo->isUsed_ = 0;
 937             pageInfo = (PageInfo*)(((PageInfo*)prevPage)->nextPage_) ;
 938         }else{
 939             prevPage = pageInfo;
 940             pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 941         }
 942         printDebug(DM_Alloc,"compact iter %x\n", pageInfo);
 943       }
 944     }
 945     releaseChunkMutex(procSlot);
 946     return 0;
 947 }
 948
 949 int Chunk::totalPages()
 950 {
 951     //logic is same for variable size and for large data node allocator.
 952     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 953     int totPages=0;
 954     while( pageInfo != NULL )
 955     {
 956         totPages++;
 957         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 958     }
 959     return totPages;
 960 }
 961 int Chunk::totalDirtyPages()
 962 {
 963     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 964     int dirtyPages=0;
 965     while( pageInfo != NULL )
 966     {
 967         if(BITSET(pageInfo->flags, IS_DIRTY)) dirtyPages++;
 968         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 969     }
 970     return dirtyPages;
 971 }
 972
 973
 974 int Chunk::initMutex(int id)
 975 {
 976     char mutName[IDENTIFIER_LENGTH];
 977     if (-1 == id)
 978        sprintf(mutName, "Chunk");
 979     else
 980        sprintf(mutName, "Chunk:%d",id);
 981     return chunkMutex_.init(mutName);
 982 }
 983 int Chunk::getChunkMutex(int procSlot)
 984 {
 985     return chunkMutex_.getLock(procSlot);
 986 }
 987 int Chunk::releaseChunkMutex(int procSlot)
 988 {
 989     return chunkMutex_.releaseLock(procSlot);
 990 }
 991 int Chunk::destroyMutex()
 992 {
 993     return chunkMutex_.destroy();
 994 }
 995 int Chunk::splitDataBucket(VarSizeInfo *varInfo, size_t needSize, int pSlot, DbRetVal *rv)
 996 {
 997     InUse remSpace = varInfo->size_ - sizeof(VarSizeInfo) - needSize;
 998     //varInfo->isUsed_ = 1;
 999     int ret = Mutex::CASGen(&varInfo->isUsed_ , 0, 1);
1000     if(ret !=0) {
1001         printDebug(DM_Warning, "Unable to set I isUsed flag");
1002         *rv = ErrLockTimeOut;
1003         return 1;
1004     }
1005     //varInfo->size_ = needSize;
1006     ret = Mutex::CASGen(&varInfo->size_, varInfo->size_ , needSize);
1007     if(ret !=0) {
1008         printError(ErrSysFatal, "Unable to set I size flag");
1009         ret = Mutex::CASGen(&varInfo->isUsed_ , varInfo->isUsed_, 0);
1010         if (ret !=0) printError(ErrSysFatal, "Unable to reset isUsed flag");
1011         *rv = ErrSysFatal;
1012         return 1;
1013     }
1014     VarSizeInfo *varInfo2 = (VarSizeInfo*)((char*)varInfo +
1015                sizeof(VarSizeInfo) +  varInfo->size_);
1016     //varInfo2->isUsed_ = 0;
1017     ret = Mutex::CASGen(&varInfo2->isUsed_ , varInfo2->isUsed_, 0);
1018     if(ret !=0) {
1019         printError(ErrSysFatal, "Unable to set II isUsed flag");
1020         ret = Mutex::CASGen(&varInfo->isUsed_ , varInfo->isUsed_, 0);
1021         if (ret !=0) printError(ErrSysFatal, "Unable to reset isUsed flag");
1022         *rv = ErrSysFatal;
1023         return 1;
1024     }
1025     //varInfo2->size_  = remSpace;
1026     ret = Mutex::CASGen(&varInfo2->size_, varInfo2->size_ , remSpace);
1027     if(ret !=0) {
1028         printError(ErrSysFatal, "Unable to set II size flag");
1029         ret = Mutex::CASGen((int*)&varInfo->isUsed_ , varInfo->isUsed_, 0);
1030         if (ret !=0) printError(ErrSysFatal, "Unable to reset isUsed flag");
1031         *rv = ErrSysFatal;
1032         return 1;
1033     }
1034     printDebug(DM_VarAlloc, "Remaining space is %d\n", remSpace);
1035     return 0;
1036 }
1037
1038
1039 int Chunk::createDataBucket(Page *page, size_t totalSize, size_t needSize, int pslot)
1040 {
1041     //already db alloc mutex is taken
1042     VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) + sizeof(PageInfo));
1043     //varInfo->isUsed_ = 0;
1044     int ret = Mutex::CASGen(&varInfo->isUsed_ , varInfo->isUsed_, 0);
1045     if(ret !=0) {
1046         printError(ErrSysFatal, "Fatal:Unable to get lock to set isUsed flag");
1047     }
1048     //varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
1049     ret = Mutex::CASGen(&varInfo->size_, varInfo->size_ ,
1050                        PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo));
1051     if(ret !=0) {
1052         printError(ErrSysFatal, "Unable to get lock to set size");
1053     }
1054     DbRetVal rv =OK;
1055     return splitDataBucket(varInfo, needSize, pslot, &rv);
1056 }
1057 void Chunk::setChunkNameForSystemDB(int id)
1058 {
1059     strcpy(chunkName,ChunkName[id]);
1060 }
1061
1062 void Chunk::print()
1063 {
1064     printf("        <Chunk Id> %d </Chunk Id> \n",chunkID_);
1065     printf("                <TotalPages> %d </TotalPages> \n",totalPages());
1066     if (Conf::config.useDurability())
1067         printf("                <DirtyPages> %d </DirtyPages> \n",totalDirtyPages());
1068     printf("                <ChunkName > %s </ChunkName> \n",getChunkName());
1069     printf("                <TotalDataNodes> %d </TotalDataNodes> \n",getTotalDataNodes());
1070     printf("                <SizeOfDataNodes> %d </SizeOfDataNodes> \n",getSize());
1071     printf("                <Allocation Type> ");
1072     if(allocType_==0)
1073     {
1074         printf("FixedSizeAllocator ");
1075     }else if(allocType_==1)
1076     {
1077         printf("VariableSizeAllocator ");
1078     }else
1079     {
1080         printf("UnknownAllocator ");
1081     }
1082     printf("</Allocation Type>\n");
1083 }