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