src/server/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
  22 // sets the size of the Chunk allocator for fixed size
  23 // allocator
  24 // we need one integer to store book keeping information
  25 // whether the storage allocation unit is used of not
  26 // when it is deleted this flag is only set to unused
  27 void Chunk::setSize(size_t size)
  28 {
  29
  30     size_t needSize = size + sizeof(int);
  31     size_t multiple = os::floor(needSize / sizeof(size_t));
  32     size_t rem = needSize % sizeof(size_t);
  33     if (0 == rem)
  34         allocSize_ = needSize;
  35     else
  36         allocSize_  = (multiple + 1) * sizeof(size_t);
  37 }
  38
  39 //Allocates memory to store data
  40 //TODO::check whether it is locked before allocating.
  41 //delete tuple will set the usedflag to true, but locks will be held
  42 //till commit and it shall be rolledback.So make sure that it does not
  43 //allocate deleted tuple which is yet to be commited.
  44
  45 void* Chunk::allocate(Database *db)
  46 {
  47     int ret = getChunkMutex();
  48     if (ret != 0)
  49     {
  50         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
  51         return NULL;
  52     }
  53     PageInfo* pageInfo = ((PageInfo*)curPage_);
  54
  55     int noOfDataNodes=os::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
  56     char *data = ((char*)curPage_) + sizeof(PageInfo);
  57     printDebug(DM_Alloc, "Chunk::allocate id:%d curPage:%x noOfDataNodes:%d",
  58                          chunkID_, curPage_, noOfDataNodes);
  59
  60     //1.scan through data list and find if any is free to use in current page
  61     //2.If there is none then
  62     //    a) get new free page from db. set the prev->next to point
  63     //       to this new page
  64     //4.  b) initialize the free page to zero and get first data ptr.
  65     //5.If there is one, return that
  66
  67     //For allocation more than PAGE_SIZE
  68     if (0 == noOfDataNodes)
  69     {
  70         ret = db->getAllocDatabaseMutex();
  71         if (ret != 0)
  72         {
  73             printError(ErrLockTimeOut,"Unable to acquire alloc database Mutex");
  74             return NULL;
  75         }
  76         pageInfo = (PageInfo*)db->getFreePage(allocSize_);
  77         if (NULL == pageInfo)
  78         {
  79             releaseChunkMutex();
  80             db->releaseAllocDatabaseMutex();
  81             printError(ErrNoMemory,"No more free pages in the database");
  82             return NULL;
  83         }
  84         printDebug(DM_Alloc, "Chunk ID:%d Large Data Item newPage:%x",
  85                                            chunkID_, pageInfo);
  86         int multiple = os::floor(allocSize_ / PAGE_SIZE);
  87         int offset = ((multiple + 1) * PAGE_SIZE);
  88
  89         pageInfo->setPageAsUsed(offset);
  90
  91         //create the link
  92         ((PageInfo*)curPage_)->nextPage_ = (Page*) pageInfo;
  93         //Make this as current page
  94         curPage_ = (Page*) pageInfo;
  95         data = ((char*)curPage_) + sizeof(PageInfo);
  96         //TODO::check whether it is locked
  97         *((int*)data) = 1;
  98         releaseChunkMutex();
  99         db->releaseAllocDatabaseMutex();
 100         return data + sizeof(int);
 101
 102     }
 103
 104     int i = noOfDataNodes;
 105     if (pageInfo->hasFreeSpace_ == 1)
 106     {
 107         for (i = 1; i< noOfDataNodes; i++)
 108         {
 109             if (*((int*)data) == 1) data = data + allocSize_;
 110             else break;
 111         }
 112
 113     }
 114     printDebug(DM_Alloc, "ChunkID:%d Node which might be free is %d",
 115                                                         chunkID_, i);
 116     //It comes here if the pageInfo->hasFreeSpace ==0
 117     //or there are no free data space in this page
 118     if (i == noOfDataNodes && *((int*)data) == 1)
 119     {
 120         ret = db->getAllocDatabaseMutex();
 121         if (ret != 0)
 122         {
 123             printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 124             return NULL;
 125         }
 126
 127        //there are no free data space in this page
 128         pageInfo->hasFreeSpace_ = 0;
 129         //get a new page from db
 130         printDebug(DM_Alloc, "ChunkID:%d curPage does not have free nodes.", chunkID_);
 131         Page *page = db->getFreePage();
 132         if (page == NULL)
 133         {
 134             PageInfo *pageIter = ((PageInfo*)firstPage_);
 135             printDebug(DM_Alloc, "Chunk ID:%d. No free page in database",
 136                                                                chunkID_);
 137             printDebug(DM_Alloc, "Scan from firstPage:%x for free nodes",
 138                                                              firstPage_);
 139             //scan from first page to locate a free node available
 140             while(NULL != pageIter)
 141             {
 142                 data = ((char*)pageIter) + sizeof(PageInfo);
 143                 if (pageIter->hasFreeSpace_ == 1)
 144                 {
 145                     for (i = 0; i< noOfDataNodes -1; i++)
 146                     {
 147                         if (1 == *((int*)data))
 148                             data = data + allocSize_;
 149                         else break;
 150                     }
 151                     if (i != noOfDataNodes -1) break;
 152                 }
 153                 printDebug(DM_Alloc, "Chunk ID: %d Page :%x does not have free nodes",
 154                                                            chunkID_, pageIter);
 155                 pageIter = (PageInfo*)((PageInfo*) pageIter)->nextPage_;
 156             }
 157             if (NULL == pageIter)
 158             {
 159                 releaseChunkMutex();
 160                 db->releaseAllocDatabaseMutex();
 161                 printError(ErrNoMemory,"No more free pages in the database:Increase db size");
 162                 return NULL;
 163             }
 164             printDebug(DM_Alloc,"ChunkID:%d Scan for free node End:Page :%x",
 165                                                           chunkID_, pageIter);
 166             *((int*)data) = 1;
 167             releaseChunkMutex();
 168             db->releaseAllocDatabaseMutex();
 169             return data + sizeof(int);
 170         }
 171         db->releaseAllocDatabaseMutex();
 172         //os::memset(page, 0, PAGE_SIZE);
 173         printDebug(DM_Alloc, "ChunkID:%d Normal Data Item newPage:%x",
 174                                                       chunkID_, page);
 175         //Initialize pageInfo for this new page
 176         PageInfo *pInfo = (PageInfo*)page;
 177         pInfo->setPageAsUsed(0);
 178
 179         //create the link between old page and the newly created page
 180         data = ((char*)page) + sizeof(PageInfo);
 181         pageInfo->nextPage_ = page;
 182
 183         //make this new page as the current page
 184         curPage_ = page;
 185
 186     }
 187     *((int*)data) = 1;
 188     releaseChunkMutex();
 189     return data + sizeof(int);
 190 }
 191
 192 //Allocates memory to store data
 193 void* Chunk::allocate(Database *db, size_t size)
 194 {
 195
 196     if (0 == size) return NULL;
 197     //align the size first
 198     //check if the size is more than PAGE_SIZE
 199     //if it is more than the PAGE_SIZE, then allocate new
 200     //page using database and then link the curPage to the
 201     //newly allocated page
 202     //if it is less than PAGE_SIZE, then check the curpage for
 203     //free memory of specified size
 204     //if not available, then scan from the firstPage for the free
 205     //space
 206
 207     //TODO::During the scan, merge nearby nodes if both are free
 208     //if not available then allocate new page
 209     int ret = getChunkMutex();
 210     if (ret != 0)
 211     {
 212         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 213         return NULL;
 214     }
 215     size_t alignedSize = os::align(size);
 216
 217     if (size > PAGE_SIZE)
 218     {
 219         int multiple = os::floor(alignedSize / PAGE_SIZE);
 220         int offset = ((multiple + 1) * PAGE_SIZE);
 221         PageInfo* pageInfo = ((PageInfo*)curPage_);
 222         ret = db->getAllocDatabaseMutex();
 223         if (ret != 0)
 224         {
 225             printError(ErrLockTimeOut,"Unable to acquire alloc database Mutex");
 226             return NULL;
 227         }
 228         pageInfo = (PageInfo*)db->getFreePage(allocSize_);
 229         if (NULL == pageInfo)
 230         {
 231             releaseChunkMutex();
 232             db->releaseAllocDatabaseMutex();
 233             printError(ErrNoMemory,"No more free pages in the database:Increase db size");
 234             return NULL;
 235         }
 236         printDebug(DM_VarAlloc,"Chunk::allocate Large Data Item id:%d Size:%d curPage:%x ",
 237                                             chunkID_, alignedSize, curPage_);
 238         //TODO:: logic pending
 239         //what happens to the space lets say 10000 bytes is allocated
 240         //it needs 2 pages,= 16000 bytes, 6000 bytes should not be wasted
 241         //in this case.So need to take of this.
 242         //Will be coded at later stage as this is developed to support
 243         //undo logging and currently we shall assume that the logs generated
 244         //wont be greater than PAGE_SIZE.
 245         printf("data size greater than page size\n");
 246         releaseChunkMutex();
 247         db->releaseAllocDatabaseMutex();
 248         return NULL;
 249     }
 250     else
 251     {
 252         Page *page = ((PageInfo*)curPage_);
 253         printDebug(DM_VarAlloc, "Chunk::allocate Normal Data Item id:%d Size:%d curPage:%x ",
 254                                                chunkID_, alignedSize, curPage_);
 255         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) +
 256                                                     sizeof(PageInfo));
 257         while ((char*) varInfo < ((char*)page + PAGE_SIZE))
 258         {
 259             if (0 == varInfo->isUsed_)
 260             {
 261                 if( alignedSize + sizeof(VarSizeInfo) < varInfo->size_)
 262                 {
 263                     splitDataBucket(varInfo, alignedSize);
 264                     releaseChunkMutex();
 265                     printDebug(DM_VarAlloc, "Chunkid:%d splitDataBucket: Size: %d Item:%x ",
 266                                                          chunkID_, alignedSize, varInfo);
 267                     return (char*)varInfo + sizeof(VarSizeInfo);
 268                 }
 269                 else if (alignedSize == varInfo->size_) {
 270                     varInfo->isUsed_ = 1;
 271                     releaseChunkMutex();
 272                     return (char *) varInfo + sizeof(VarSizeInfo);
 273                 }
 274
 275             }
 276             varInfo = (VarSizeInfo*)((char*)varInfo + sizeof(VarSizeInfo)
 277                                     +varInfo->size_);
 278         }
 279         //No available spaces in the current page.
 280         //allocate new page
 281         ret = db->getAllocDatabaseMutex();
 282         if (ret != 0)
 283         {
 284             printError(ErrLockTimeOut,"Unable to acquire alloc database Mutex");
 285             return NULL;
 286         }
 287         Page *newPage = db->getFreePage();
 288         if (NULL == newPage)
 289         {
 290             void *data = varSizeFirstFitAllocate(size);
 291             releaseChunkMutex();
 292             db->releaseAllocDatabaseMutex();
 293             if (NULL == data)
 294                 printError(ErrNoMemory,"No more free space in the database:Increase db size");
 295             return data;
 296         }
 297         db->releaseAllocDatabaseMutex();
 298
 299         printDebug(DM_VarAlloc, "ChunkID:%d New Page: %x ", chunkID_, newPage);
 300         PageInfo *pInfo = (PageInfo*) newPage;
 301         pInfo->setPageAsUsed(0);
 302         createDataBucket(newPage, PAGE_SIZE, alignedSize);
 303
 304         ((PageInfo*)curPage_)->nextPage_ = newPage;
 305         curPage_ = newPage;
 306         releaseChunkMutex();
 307         char *data= ((char*)newPage) + sizeof(PageInfo) + sizeof(VarSizeInfo);
 308         return data;
 309     }
 310 }
 311
 312 //Assumes chunk mutex is already taken, before calling this
 313 void* Chunk::varSizeFirstFitAllocate(size_t size)
 314 {
 315     Page *page = ((PageInfo*)firstPage_);
 316     size_t alignedSize = os::align(size);
 317     while(NULL != page)
 318     {
 319         VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) + sizeof(PageInfo));
 320         while ((char*) varInfo < ((char*)page + PAGE_SIZE))
 321         {
 322             if (0 == varInfo->isUsed_)
 323             {
 324                 if( alignedSize < varInfo->size_)
 325                 {
 326                     splitDataBucket(varInfo, alignedSize);
 327                     return (char*)varInfo + sizeof(VarSizeInfo);
 328                 }
 329             }
 330             varInfo = (VarSizeInfo*)((char*)varInfo + sizeof(VarSizeInfo)
 331                                     +varInfo->size_);
 332         }
 333         page = ((PageInfo*) page)->nextPage_;
 334     }
 335     return NULL;
 336 }
 337
 338
 339
 340 //Frees the memory pointed by ptr
 341 void Chunk::free(Database *db, void *ptr)
 342 {
 343     int ret = getChunkMutex();
 344     if (ret != 0)
 345     {
 346         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 347         return;
 348     }
 349     if (0 == allocSize_)
 350     {
 351         VarSizeInfo *varInfo = (VarSizeInfo*)((char*)ptr- sizeof(VarSizeInfo));
 352         varInfo->isUsed_ = 0;
 353         printDebug(DM_VarAlloc,"chunkID:%d Unset isUsed for %x", chunkID_, varInfo);
 354         //return from here for variable size allocator
 355         releaseChunkMutex();
 356         return;
 357     }
 358      //unset the used flag
 359     *((int*)ptr -1 ) = 0;
 360     int noOfDataNodes =os::floor((PAGE_SIZE - sizeof(PageInfo)) / allocSize_);
 361
 362     if (0 == noOfDataNodes)
 363     {
 364         //means tuple larger than PAGE_SIZE
 365         PageInfo *pageInfo = (PageInfo*)(((char*)
 366                              ptr) - (sizeof(PageInfo) + sizeof(int)));
 367         PageInfo *pInfo = (PageInfo*)firstPage_, *prev = (PageInfo*)firstPage_;
 368         bool found = false;
 369         while(!found)
 370         {
 371             if (pInfo == pageInfo) {found = true;  break; }
 372             prev = pInfo;
 373             pInfo = (PageInfo*)pInfo->nextPage_;
 374         }
 375         if (!found)
 376         {
 377             printError(ErrSysFatal,"Page %x not found in page list:Logic error", pageInfo );
 378             releaseChunkMutex();
 379             return ;
 380         }
 381         prev->nextPage_ = pageInfo->nextPage_;
 382         pageInfo->nextPageAfterMerge_ = NULL;
 383         pageInfo->isUsed_ = 0;
 384         os::memset(pageInfo,  0 , allocSize_);
 385         pageInfo->hasFreeSpace_ = 1;
 386         releaseChunkMutex();
 387         return;
 388     }
 389     PageInfo *pageInfo;
 390     pageInfo = getPageInfo(db, ptr);
 391     if (NULL == pageInfo)
 392     {
 393         printError(ErrSysFatal,"Probable Data corruption: pageInfo is NULL", pageInfo );
 394         releaseChunkMutex();
 395         return;
 396     }
 397     //set the pageinfo where this ptr points
 398     pageInfo->hasFreeSpace_ = 1;
 399     releaseChunkMutex();
 400     return;
 401 }
 402
 403 //returns the pageInfo of the page where this ptr points
 404 //This works only if the data size is less than PAGE_SIZE
 405 //If ptr points to data which is more than PAGE_SIZE,then
 406 //calling this might lead to memory corruption
 407 //Note:IMPORTANT::assumes db lock is taken before calling this
 408 PageInfo* Chunk::getPageInfo(Database *db, void *ptr)
 409 {
 410     char *inPtr = (char*)ptr;
 411     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 412
 413     while( pageInfo != NULL )
 414     {
 415         if (inPtr > (char*) pageInfo && ((char*)pageInfo + PAGE_SIZE) >inPtr)
 416             return pageInfo;
 417         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 418     }
 419     return NULL;
 420 }
 421
 422 long Chunk::getTotalDataNodes()
 423 {
 424     int ret = getChunkMutex();
 425     if (ret != 0)
 426     {
 427         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 428         return NULL;
 429     }
 430
 431     //TODO:: for variable size allocator
 432     if (0 == allocSize_) //->variable size allocator
 433         return 0;
 434
 435     //TODO::for large size allocator
 436     if (allocSize_ >PAGE_SIZE)//->each page has only one data node
 437         return 0;
 438
 439     int noOfDataNodes=os::floor((PAGE_SIZE - sizeof(PageInfo))/allocSize_);
 440     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 441     char *data = ((char*)firstPage_) + sizeof(PageInfo);
 442     long totalNodes = 0;
 443     int i=0;
 444     while( pageInfo != NULL )
 445     {
 446         data = ((char*)pageInfo) + sizeof(PageInfo);
 447         for (i = 0; i< noOfDataNodes; i++)
 448         {
 449             if (*((int*)data) == 1) { totalNodes++;}
 450             data = data + allocSize_;
 451         }
 452         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 453     }
 454     releaseChunkMutex();
 455     return totalNodes;
 456 }
 457
 458 int Chunk::totalPages()
 459 {
 460     int ret = getChunkMutex();
 461     if (ret != 0)
 462     {
 463         printError(ErrLockTimeOut,"Unable to acquire chunk Mutex");
 464         return NULL;
 465     }
 466     //logic is same for variable size and for large data node allocator.
 467     PageInfo* pageInfo = ((PageInfo*)firstPage_);
 468     int totPages=0;
 469     while( pageInfo != NULL )
 470     {
 471         totPages++;
 472         pageInfo = (PageInfo*)(((PageInfo*)pageInfo)->nextPage_) ;
 473     }
 474     releaseChunkMutex();
 475     return totPages;
 476 }
 477
 478 int Chunk::initMutex()
 479 {
 480     return chunkMutex_.init();
 481 }
 482 int Chunk::getChunkMutex()
 483 {
 484     return chunkMutex_.tryLock();
 485 }
 486 int Chunk::releaseChunkMutex()
 487 {
 488     return chunkMutex_.releaseLock();
 489 }
 490 int Chunk::destroyMutex()
 491 {
 492     return chunkMutex_.destroy();
 493 }
 494 void Chunk::splitDataBucket(VarSizeInfo *varInfo, size_t needSize)
 495 {
 496     int remSpace = varInfo->size_ - sizeof(VarSizeInfo) - needSize;
 497     varInfo->isUsed_ = 1;
 498     varInfo->size_ = needSize;
 499     varInfo = (VarSizeInfo*)((char*)varInfo +
 500                sizeof(VarSizeInfo) +  varInfo->size_);
 501     varInfo->isUsed_ = 0;
 502     varInfo->size_  = remSpace;
 503     return;
 504 }
 505
 506
 507 void Chunk::createDataBucket(Page *page, size_t totalSize, size_t needSize)
 508 {
 509     VarSizeInfo *varInfo = (VarSizeInfo*)(((char*)page) + sizeof(PageInfo));
 510     varInfo->isUsed_ = 0;
 511     varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
 512     splitDataBucket(varInfo, needSize);
 513
 514     return;
 515 }
 516
 517
 518