src/storage/Transaction.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<Transaction.h>
  17 #include<Lock.h>
  18 #include<Database.h>
  19 #include<Allocator.h>
  20 #include<CatalogTables.h>
  21 #include<Debug.h>
  22
  23 //Code assumes that only one thread can work on a transaction
  24 DbRetVal Transaction::insertIntoHasList(Database *sysdb, LockHashNode *node)
  25 {
  26     //allocate lock node
  27     Chunk *chunk = sysdb->getSystemDatabaseChunk(TransHasTableId);
  28     DbRetVal rv = OK;
  29     TransHasNode *hasNode = NULL; //(TransHasNode*)chunk->allocate(sysdb, &rv);
  30     int tries=0;
  31     int totalTries = Conf::config.getMutexRetries();
  32     while (tries < totalTries)
  33     {
  34         rv = OK;
  35         hasNode= (TransHasNode*)chunk->allocate(sysdb, &rv);
  36         if (hasNode !=NULL) break;
  37         if (rv != ErrLockTimeOut)
  38         {
  39             printError(rv, "Unable to allocate trans has node");
  40             return rv;
  41         }
  42         tries++;
  43     }
  44     if (NULL == hasNode)
  45     {
  46         printError(rv, "Could not allocate Trans has node after %d retry", tries);
  47         return rv;
  48     }
  49     printDebug(DM_Transaction, "insertIntoHasList new TransHasNode created:%x",
  50                                                        hasNode);
  51     hasNode->node_ = node;
  52     hasNode->next_ = NULL;
  53     if (NULL == hasLockList_)
  54     {
  55         printDebug(DM_Transaction, "hasLockList is null:It is now %x",hasNode);
  56         hasLockList_ = hasNode;
  57         return OK;
  58     }
  59
  60     TransHasNode *it = hasLockList_;
  61     while (NULL != it->next_) { it = it->next_; }
  62     it->next_ = hasNode;
  63     printDebug(DM_Transaction, "Added to hasLockList at end:%x",it);
  64     logFinest(Conf::logger, "Added locknode:%x to hasLockList", hasNode->node_);
  65     return OK;
  66 }
  67
  68 DbRetVal Transaction::removeFromHasList(Database *sysdb, void *tuple)
  69 {
  70     Chunk *chunk = sysdb->getSystemDatabaseChunk(TransHasTableId);
  71     TransHasNode *iter = hasLockList_, *prev = hasLockList_;
  72     if (NULL == iter)
  73     {
  74         printError(ErrNotFound, "Fatal:HasList is empty");
  75         return ErrNotFound;
  76     }
  77     while (iter != NULL)
  78     {
  79         if (tuple == iter->node_->ptrToTuple_)
  80         {
  81             prev->next_ = iter->next_;
  82             chunk->free(sysdb, iter);
  83             if (iter == hasLockList_) hasLockList_ = NULL;
  84             logFinest(Conf::logger, "Removed locknode:%x from hasLockList",
  85                                                        iter->node_);
  86             return OK;
  87         }
  88         prev = iter;
  89         iter = iter->next_;
  90     }
  91     printStackTrace();
  92     printError(ErrNotFound, "Fatal:There is no tuple lock in has list for tuple:%x", tuple);
  93     //TEMP::for debugging.do not remove
  94     /*iter=hasLockList_;
  95     int cnt=0;
  96     while (iter != NULL)
  97     {
  98         printError(ErrWarning, "Element in hasList: %d %x: %x:%d\n", cnt, iter->node_, iter->node_->ptrToTuple_ , *(int*)iter->node_->ptrToTuple_);
  99         cnt++;
 100         iter = iter->next_;
 101     }*/
 102     return ErrNotFound;
 103 }
 104
 105
 106 DbRetVal Transaction::releaseAllLocks(LockManager *lockManager_)
 107 {
 108     Database *sysdb =lockManager_->systemDatabase_;
 109     Chunk *chunk = sysdb->getSystemDatabaseChunk(TransHasTableId);
 110     TransHasNode *iter  = hasLockList_, *prev;
 111     DbRetVal rv = OK;
 112     while (NULL != iter)
 113     {
 114         prev = iter;
 115         iter = iter->next_;
 116         printDebug(DM_Transaction, "Releasing lock %x",prev->node_->ptrToTuple_);
 117         logFinest(Conf::logger, "Releasing lock for tuple:%x",prev->node_->ptrToTuple_);
 118         rv = lockManager_->releaseLock(prev->node_->ptrToTuple_);
 119         chunk->free(sysdb, prev);
 120     }
 121     hasLockList_ = NULL;
 122     return OK;
 123 }
 124 bool Transaction::findInHasList(Database *sysdb, LockHashNode *node)
 125 {
 126     TransHasNode *iter  = hasLockList_;
 127     while (NULL != iter)
 128     {
 129         if (iter->node_ == node) return true;
 130         iter = iter->next_;
 131     }
 132     return false;
 133 }
 134
 135 DbRetVal Transaction::appendUndoLog(Database *sysdb, OperationType type,
 136                                                void *data, size_t size)
 137 {
 138     DbRetVal rv =OK;
 139     UndoLogInfo *logInfo = createUndoLog(sysdb, type, data, size, &rv);
 140     if (logInfo == NULL) return rv;
 141     if (size) os::memcpy((char*)logInfo + sizeof(UndoLogInfo), data, size);
 142     addAtBegin(logInfo);
 143     printDebug(DM_Transaction, "creating undo log and append %x optype:%d",
 144                                                logInfo, type);
 145     return OK;
 146 }
 147
 148
 149
 150 DbRetVal Transaction::appendLogicalUndoLog(Database *sysdb, OperationType type, void *data, size_t size, void* indexPtr)
 151 {
 152     DbRetVal rv = OK;
 153     UndoLogInfo *logInfo = createUndoLog(sysdb, type, data, size, &rv);
 154     if (logInfo == NULL) return rv;
 155     char **indPtr = (char**)((char*)logInfo + sizeof(UndoLogInfo));
 156     *indPtr = (char*)  indexPtr;
 157     addAtBegin(logInfo);
 158     printDebug(DM_Transaction, "creating logical undo log and append %x optype:%d", logInfo, type);
 159     return rv;
 160 }
 161
 162 DbRetVal Transaction::appendLogicalHashUndoLog(Database *sysdb, OperationType type, void *data, size_t size)
 163 {
 164     DbRetVal rv = OK;
 165     HashUndoLogInfo *hInfo = (HashUndoLogInfo *) data;
 166     UndoLogInfo *logInfo = createUndoLog(sysdb, type, hInfo->tuple_, size, &rv);
 167     if (logInfo == NULL) return rv;
 168     memcpy((char*)logInfo + sizeof(UndoLogInfo), data, sizeof(HashUndoLogInfo));
 169     addAtBegin(logInfo);
 170     printDebug(DM_Transaction, "creating logical undo log and append %x optype:%d", logInfo, type);
 171     return rv;
 172 }
 173 DbRetVal Transaction::appendLogicalTreeUndoLog(Database *sysdb, OperationType type, void *data, size_t size)
 174 {
 175     DbRetVal rv = OK;
 176     TreeUndoLogInfo *hInfo = (TreeUndoLogInfo *) data;
 177     UndoLogInfo *logInfo = createUndoLog(sysdb, type, hInfo->tuple_, size, &rv);
 178     if (logInfo == NULL) return rv;
 179     memcpy((char*)logInfo + sizeof(UndoLogInfo), data, sizeof(TreeUndoLogInfo));
 180     addAtBegin(logInfo);
 181     printDebug(DM_Transaction, "creating logical undo log and append %x optype:%d", logInfo, type);
 182     return rv;
 183 }
 184 DbRetVal Transaction::appendLogicalTrieUndoLog(Database *sysdb, OperationType type, void *data, size_t size)
 185 {
 186     DbRetVal rv = OK;
 187     TrieUndoLogInfo *hInfo = (TrieUndoLogInfo *) data;
 188     UndoLogInfo *logInfo = createUndoLog(sysdb, type, hInfo->tuple_, size, &rv);
 189     if (logInfo == NULL) return rv;
 190     memcpy((char*)logInfo + sizeof(UndoLogInfo), data, sizeof(TrieUndoLogInfo));
 191     addAtBegin(logInfo);
 192     printDebug(DM_Transaction, "creating logical undo log and append %x optype:%d", logInfo, type);
 193     return rv;
 194 }
 195
 196 UndoLogInfo* Transaction::createUndoLog(Database *sysdb, OperationType type, void *data,
 197                        size_t size, DbRetVal *rv)
 198 {
 199     Chunk *chunk = sysdb->getSystemDatabaseChunk(UndoLogTableID);
 200     int reqSize = size + sizeof(UndoLogInfo);
 201     UndoLogInfo *logInfo = NULL;
 202     int tries=0;
 203     int totalTries = Conf::config.getMutexRetries();
 204     while (tries < totalTries)
 205     {
 206         *rv = OK;
 207         logInfo= (UndoLogInfo*)chunk->allocate(sysdb, reqSize, rv);
 208         if (logInfo !=NULL) break;
 209         if (*rv != ErrLockTimeOut)
 210         {
 211             printError(*rv, "Unable to allocate undo log");
 212             return NULL;
 213         }
 214         tries++;
 215     }
 216
 217     if (logInfo == NULL) {
 218         printError(*rv, "Unable to allocate undo log record after %d retries", tries);
 219         return NULL;
 220     }
 221     logInfo->data_ = (char *)logInfo + sizeof(UndoLogInfo);
 222     logInfo->opType_ = type;
 223     logInfo->size_ = size;
 224     logInfo->next_ = NULL;
 225     return logInfo;
 226 }
 227
 228 void Transaction::addAtBegin(UndoLogInfo* logInfo)
 229 {
 230     //add it to the begin of the log list
 231     logInfo->next_ = firstUndoLog_;
 232     firstUndoLog_ = logInfo;
 233     return;
 234 }
 235
 236 UndoLogInfo* Transaction::popUndoLog()
 237 {
 238     UndoLogInfo *iter = firstUndoLog_, *prev = firstUndoLog_;
 239     if(NULL != iter)
 240     {
 241         prev = iter;
 242         iter = iter->next_;
 243     }
 244     firstUndoLog_ = iter;
 245     return prev;
 246
 247 }
 248
 249 int Transaction::noOfUndoLogs()
 250 {
 251     UndoLogInfo *iter = firstUndoLog_;
 252     int count =0;
 253     while(NULL != iter)
 254     {
 255         count++;
 256         iter = iter->next_;
 257     }
 258     return count;
 259 }
 260 void Transaction::printDebugInfo(Database *sysdb)
 261 {
 262    printf("<TransactionInfo>\n");
 263    if (waitLock_ != NULL)
 264    {
 265        printf("<WaitLock>");
 266        waitLock_->print();
 267        printf("</WaitLock>");
 268
 269    }
 270    printf("<UndoLogs>\n");
 271    Chunk *chunk = sysdb->getSystemDatabaseChunk(UndoLogTableID);
 272    printf("  <TotalPages> %d </TotalPages>\n", chunk->totalPages());
 273    UndoLogInfo *iter = firstUndoLog_;
 274    int count =0;
 275    while(NULL != iter)
 276    {
 277       iter->print();
 278       iter = iter->next_;
 279       count++;
 280    }
 281    printf("</TotalNodes> %d </TotalNodes>\n", count);
 282    printf("</UndoLogs>\n");
 283
 284    printf("<TransHasList>\n");
 285    chunk = sysdb->getSystemDatabaseChunk(TransHasTableId);
 286    printf("  <TotalPages> %d </TotalPages>\n", chunk->totalPages());
 287    TransHasNode *hasIter  = hasLockList_;
 288    count =0;
 289    while (NULL != hasIter)
 290    {
 291        hasIter->print();
 292        hasIter = hasIter->next_;
 293        count++;
 294    }
 295    printf("</TotalNodes> %d </TotalNodes>\n", count);
 296    printf("</TransHasList>\n");
 297
 298    printf("</TransactionInfo>\n");
 299    return ;
 300 }
 301 DbRetVal Transaction::removeUndoLogs(Database *sysdb)
 302 {
 303     Chunk *chunk = sysdb->getSystemDatabaseChunk(UndoLogTableID);
 304     UndoLogInfo *logInfo = NULL;
 305     while(NULL != (logInfo = popUndoLog()))
 306     {
 307         chunk->free(sysdb, logInfo);
 308     }
 309     return OK;
 310 }
 311
 312
 313 DbRetVal Transaction::applyUndoLogs(Database *sysdb)
 314 {
 315     Chunk *chunk = sysdb->getSystemDatabaseChunk(UndoLogTableID);
 316     UndoLogInfo *logInfo = NULL;
 317     while(NULL != (logInfo = popUndoLog()))
 318     {
 319         logFinest(Conf::logger, "Apply undo log type:%d", logInfo->opType_);
 320         switch(logInfo->opType_)
 321         {
 322             case InsertOperation:
 323              {
 324                 char *ptr = (char *)logInfo->data_;
 325                 void *ptrToTuple = (void *)*(long *)ptr;
 326                 ptr += sizeof(void *);
 327                 InUse *isUsed = ((InUse*)(ptrToTuple) - 1);
 328                 if (*isUsed == 0) {
 329                    printError(ErrSysFatal, "Fatal: Row is already not in use");
 330                 }
 331                 *isUsed = 0;
 332                 handleVarcharUndoInsert(sysdb, ptr);
 333                 break;
 334              }
 335              case DeleteOperation:
 336              {
 337                 char *ptr = (char *)logInfo->data_;
 338                 void *ptrToTuple = (void *)*(long *)ptr;
 339                 ptr += sizeof(void *);
 340                 InUse *isUsed = ((InUse*)(ptrToTuple) - 1);
 341                 if (*isUsed == 1) {
 342                     printError(ErrSysFatal, "Fatal: Row is already in use");
 343                 }
 344                 *isUsed = 1;
 345                 //data record will be intact as we have lock on that record
 346                 handleVarcharUndoDelete(sysdb, ptr);
 347                 break;
 348              }
 349              case UpdateOperation:
 350              {
 351                 char *ptr = (char *)logInfo->data_;
 352                 void *ptrToTuple = (void *)*(long *)ptr;
 353                 ptr += sizeof(void *);
 354                 InUse *isUsed = ((InUse*)(ptrToTuple) - 1);
 355                 if (*isUsed == 0) {
 356                     printError(ErrSysFatal, "Fatal: Row is not in use during update rollback");
 357                 }
 358                 handleVarcharUndoUpdate(sysdb, ptr, ptrToTuple);
 359                 break;
 360              }
 361              case InsertHashIndexOperation:
 362              {
 363                 HashIndex::deleteLogicalUndoLog(sysdb, (char *)logInfo
 364                                                  + sizeof(UndoLogInfo));
 365                 break;
 366              }
 367              case DeleteHashIndexOperation:
 368              {
 369                 HashIndex::insertLogicalUndoLog(sysdb, (char *)logInfo
 370                                                  + sizeof(UndoLogInfo));
 371                 break;
 372              }
 373              case InsertTreeIndexOperation:
 374              {
 375                 TreeIndex::deleteLogicalUndoLog(sysdb, (char *)logInfo
 376                                                  + sizeof(UndoLogInfo));
 377                 break;
 378              }
 379              case DeleteTreeIndexOperation:
 380              {
 381                 TreeIndex::insertLogicalUndoLog(sysdb, (char *)logInfo
 382                                                  + sizeof(UndoLogInfo));
 383                 break;
 384              }
 385              case InsertTrieIndexOperation:
 386              {
 387                 TrieIndex::deleteLogicalUndoLog(sysdb, (char *)logInfo
 388                                                  + sizeof(UndoLogInfo));
 389                 break;
 390              }
 391              case DeleteTrieIndexOperation:
 392              {
 393                 TrieIndex::insertLogicalUndoLog(sysdb, (char *)logInfo
 394                                                  + sizeof(UndoLogInfo));
 395                 break;
 396              }
 397              default:
 398              {
 399                  printError(ErrSysFatal, "Illegal undo log type");
 400                  break;
 401              }
 402         }
 403         chunk->free(sysdb, logInfo);
 404     }
 405     return OK;
 406 }
 407 DbRetVal Transaction::handleVarcharUndoInsert(Database *sysdb, char *ptr)
 408 {
 409     // ptr  will have following info encapsulated.
 410     // metadataPtr + nVarchars + varchar chunk ptr +
 411     // ptrs to varchars
 412
 413     Database db;
 414     void *metaData = (void *)*(long *)ptr;
 415     ptr += sizeof(void *);
 416     db.setMetaDataPtr((DatabaseMetaData *) metaData);
 417     db.setProcSlot(sysdb->procSlot);
 418
 419     int noOfVarchar = *(int *)ptr;
 420     ptr += sizeof(int);
 421     Chunk *vcchunk = (Chunk *) *(long *)ptr;
 422     ptr += sizeof(void *);
 423     void **ptrToVarchars = (void **) ptr;
 424     for (int i = 0; i < noOfVarchar; i++) {
 425         if (*(long *) ptrToVarchars[i] != 0L) {
 426             vcchunk->free(&db, (void *)*(long *)ptrToVarchars[i]);
 427             *(long *) ptrToVarchars[i] = 0L;
 428         }
 429     }
 430     return OK;
 431 }
 432
 433 DbRetVal Transaction::handleVarcharUndoDelete(Database *sysdb, char *ptr)
 434 {
 435     // ptr will have following info encapsulated.
 436     // metadataPtr + nVarchars + varchar chunk ptr + ptrs to varchars +
 437     // size and value pairs for varchars
 438     void *metaData = (void *)*(long *)ptr;
 439     ptr += sizeof(void *);
 440     Database db;
 441     db.setMetaDataPtr((DatabaseMetaData *) metaData);
 442     db.setProcSlot(sysdb->procSlot);
 443     DbRetVal rv = OK;
 444     int noOfVarchar = *(int *) ptr;
 445     ptr+= sizeof(int);
 446     Chunk *vcchunk = (Chunk *) *(long *)ptr;
 447     ptr += sizeof(void *);
 448     void **ptrToVarchars = (void **) ptr;
 449     ptr += noOfVarchar * sizeof (void *);
 450     char *lenValPtr = (char *) ptr;
 451     for (int i = 0; i < noOfVarchar; i++) {
 452         int len = *(int *) lenValPtr;
 453         lenValPtr += sizeof(int);
 454         if (len != 0) {
 455             void *ptr = vcchunk->allocate(&db, len, &rv);
 456             strcpy((char *)ptr, lenValPtr);
 457             lenValPtr += len;
 458             *(long *) ptrToVarchars[i] =  (long)ptr;
 459         } else {
 460             *(long *) ptrToVarchars[i] = 0L;
 461         }
 462     }
 463     return rv;
 464 }
 465 DbRetVal Transaction::handleVarcharUndoUpdate(Database *sysdb, char *ptr, void *ptrToTuple)
 466 {
 467     // logInfo->data_ will have following info encapsulated.
 468     // tupleptr + tuple length + actual tuple + metadataPtr +
 469     // nVarchars + varchar chunk ptr + ptrs to varchars +
 470     // size and value pairs for varchars
 471
 472     int tupleLen = *(int *) ptr;
 473     ptr += sizeof(int);
 474     void *tuple = ptr;
 475     ptr += tupleLen;
 476     void *metaData = (void *)*(long *)ptr;
 477     ptr += sizeof(void *);
 478     Database db;
 479     db.setMetaDataPtr((DatabaseMetaData *) metaData);
 480     db.setProcSlot(sysdb->procSlot);
 481     DbRetVal rv = OK;
 482     int noOfVarchar = *(int *) ptr;
 483     ptr+= sizeof(int);
 484     Chunk *vcchunk = (Chunk *) *(long *)ptr;
 485     ptr += sizeof(void *);
 486     void **ptrToVarchars = (void **) ptr;
 487     ptr += noOfVarchar * sizeof (void *);
 488     char *lenValPtr = (char *) ptr;
 489     for (int i = 0; i < noOfVarchar; i++) {
 490         if (*(long *) ptrToVarchars[i] != 0L) {
 491             vcchunk->free(&db, (void *)*(long *) ptrToVarchars[i]);
 492             *(long *) ptrToVarchars[i] = 0L;
 493         }
 494     }
 495     os::memcpy(ptrToTuple, tuple, tupleLen);
 496     for (int i = 0; i < noOfVarchar; i++) {
 497         int len = *(int *) lenValPtr; lenValPtr += sizeof(int);
 498         if (len != 0) {
 499             void *ptr = vcchunk->allocate(&db, len, &rv);
 500             strcpy((char *)ptr, lenValPtr);
 501             lenValPtr += len;
 502             *(long *) ptrToVarchars[i] = (long) ptr;
 503         } else {
 504             *(long *) ptrToVarchars[i] = 0L;
 505         }
 506     }
 507     return rv;
 508 }