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
 185 UndoLogInfo* Transaction::createUndoLog(Database *sysdb, OperationType type, void *data,
 186                        size_t size, DbRetVal *rv)
 187 {
 188     Chunk *chunk = sysdb->getSystemDatabaseChunk(UndoLogTableID);
 189     int reqSize = size + sizeof(UndoLogInfo);
 190     UndoLogInfo *logInfo = NULL;
 191     int tries=0;
 192     int totalTries = Conf::config.getMutexRetries();
 193     while (tries < totalTries)
 194     {
 195         *rv = OK;
 196         logInfo= (UndoLogInfo*)chunk->allocate(sysdb, reqSize, rv);
 197         if (logInfo !=NULL) break;
 198         if (*rv != ErrLockTimeOut)
 199         {
 200             printError(*rv, "Unable to allocate undo log");
 201             return NULL;
 202         }
 203         tries++;
 204     }
 205
 206     if (logInfo == NULL) {
 207         printError(*rv, "Unable to allocate undo log record after %d retries", tries);
 208         return NULL;
 209     }
 210     logInfo->data_ = (char *)logInfo + sizeof(UndoLogInfo);
 211     logInfo->opType_ = type;
 212     logInfo->size_ = size;
 213     logInfo->next_ = NULL;
 214     return logInfo;
 215 }
 216
 217 void Transaction::addAtBegin(UndoLogInfo* logInfo)
 218 {
 219     //add it to the begin of the log list
 220     logInfo->next_ = firstUndoLog_;
 221     firstUndoLog_ = logInfo;
 222     return;
 223 }
 224
 225 UndoLogInfo* Transaction::popUndoLog()
 226 {
 227     UndoLogInfo *iter = firstUndoLog_, *prev = firstUndoLog_;
 228     if(NULL != iter)
 229     {
 230         prev = iter;
 231         iter = iter->next_;
 232     }
 233     firstUndoLog_ = iter;
 234     return prev;
 235
 236 }
 237
 238 int Transaction::noOfUndoLogs()
 239 {
 240     UndoLogInfo *iter = firstUndoLog_;
 241     int count =0;
 242     while(NULL != iter)
 243     {
 244         count++;
 245         iter = iter->next_;
 246     }
 247     return count;
 248 }
 249 void Transaction::printDebugInfo(Database *sysdb)
 250 {
 251    printf("<TransactionInfo>\n");
 252    if (waitLock_ != NULL)
 253    {
 254        printf("<WaitLock>");
 255        waitLock_->print();
 256        printf("</WaitLock>");
 257
 258    }
 259    printf("<UndoLogs>\n");
 260    Chunk *chunk = sysdb->getSystemDatabaseChunk(UndoLogTableID);
 261    printf("  <TotalPages> %d </TotalPages>\n", chunk->totalPages());
 262    UndoLogInfo *iter = firstUndoLog_;
 263    int count =0;
 264    while(NULL != iter)
 265    {
 266       iter->print();
 267       iter = iter->next_;
 268       count++;
 269    }
 270    printf("</TotalNodes> %d </TotalNodes>\n", count);
 271    printf("</UndoLogs>\n");
 272
 273    printf("<TransHasList>\n");
 274    chunk = sysdb->getSystemDatabaseChunk(TransHasTableId);
 275    printf("  <TotalPages> %d </TotalPages>\n", chunk->totalPages());
 276    TransHasNode *hasIter  = hasLockList_;
 277    count =0;
 278    while (NULL != hasIter)
 279    {
 280        hasIter->print();
 281        hasIter = hasIter->next_;
 282        count++;
 283    }
 284    printf("</TotalNodes> %d </TotalNodes>\n", count);
 285    printf("</TransHasList>\n");
 286
 287    printf("</TransactionInfo>\n");
 288    return ;
 289 }
 290 DbRetVal Transaction::removeUndoLogs(Database *sysdb)
 291 {
 292     Chunk *chunk = sysdb->getSystemDatabaseChunk(UndoLogTableID);
 293     UndoLogInfo *logInfo = NULL;
 294     while(NULL != (logInfo = popUndoLog()))
 295     {
 296         chunk->free(sysdb, logInfo);
 297     }
 298     return OK;
 299 }
 300
 301
 302 DbRetVal Transaction::applyUndoLogs(Database *sysdb)
 303 {
 304     Chunk *chunk = sysdb->getSystemDatabaseChunk(UndoLogTableID);
 305     UndoLogInfo *logInfo = NULL;
 306     while(NULL != (logInfo = popUndoLog()))
 307     {
 308         logFinest(Conf::logger, "Apply undo log type:%d", logInfo->opType_);
 309         switch(logInfo->opType_)
 310         {
 311             case InsertOperation:
 312              {
 313                 // logInfo->data_ will have following info encapsulated.
 314                 // tupleptr + metadataPtr + nVarchars + varchar chunk ptr +
 315                 // ptrs to varchars
 316                 char *ptr = (char *)logInfo->data_;
 317                 void *ptrToTuple = (void *)*(long *)ptr; ptr += sizeof(void *);
 318                 InUse *isUsed = ((InUse*)(ptrToTuple) - 1);
 319                 if (*isUsed == 0) {
 320                    printError(ErrSysFatal, "Fatal: Row is already not in use");
 321                 }
 322                 *isUsed = 0;
 323
 324                 Database db;
 325                 void *metaData = (void *)*(long *)ptr; ptr += sizeof(void *);
 326                 db.setMetaDataPtr((DatabaseMetaData *) metaData);
 327                 db.setProcSlot(sysdb->procSlot);
 328
 329                 int noOfVarchar = *(int *)ptr; ptr += sizeof(int);
 330                 Chunk *vcchunk = (Chunk *) *(long *)ptr; ptr += sizeof(void *);
 331                 void **ptrToVarchars = (void **) ptr;
 332                 for (int i = 0; i < noOfVarchar; i++) {
 333                     if (*(long *) ptrToVarchars[i] != 0L) {
 334                         vcchunk->free(&db, (void *)*(long *)ptrToVarchars[i]);
 335                         *(long *) ptrToVarchars[i] = 0L;
 336                     }
 337                 }
 338                 break;
 339              }
 340              case DeleteOperation:
 341              {
 342                 // logInfo->data_ will have following info encapsulated.
 343                 // tupleptr + tuple length + actual tuple + metadataPtr +
 344                 // nVarchars + varchar chunk ptr + ptrs to varchars +
 345                 // size and value pairs for varchars
 346                 char *ptr = (char *)logInfo->data_;
 347                 void *ptrToTuple = (void *)*(long *)ptr; ptr += sizeof(void *);
 348                 InUse *isUsed = ((InUse*)(ptrToTuple) - 1);
 349                 if (*isUsed == 1) {
 350                     printError(ErrSysFatal, "Fatal: Row is already in use");
 351                 }
 352                 *isUsed = 1;
 353                 //data record will be intact as we have lock on that record
 354
 355                 void *metaData = (void *)*(long *)ptr; ptr += sizeof(void *);
 356                 Database db;
 357                 db.setMetaDataPtr((DatabaseMetaData *) metaData);
 358                 db.setProcSlot(sysdb->procSlot);
 359                 DbRetVal rv = OK;
 360                 int noOfVarchar = *(int *) ptr; ptr+= sizeof(int);
 361                 Chunk *vcchunk = (Chunk *) *(long *)ptr; ptr += sizeof(void *);
 362                 void **ptrToVarchars = (void **) ptr;
 363                 ptr += noOfVarchar * sizeof (void *);
 364                 char *lenValPtr = (char *) ptr;
 365                 for (int i = 0; i < noOfVarchar; i++) {
 366                     int len = *(int *) lenValPtr; lenValPtr += sizeof(int);
 367                     if (len != 0) {
 368                         void *ptr = vcchunk->allocate(&db, len, &rv);
 369                         strcpy((char *)ptr, lenValPtr); lenValPtr += len;
 370                         *(long *) ptrToVarchars[i] =  (long)ptr;
 371                     } else *(long *) ptrToVarchars[i] = 0L;
 372                 }
 373                 break;
 374              }
 375              case UpdateOperation:
 376              {
 377                 // logInfo->data_ will have following info encapsulated.
 378                 // tupleptr + tuple length + actual tuple + metadataPtr +
 379                 // nVarchars + varchar chunk ptr + ptrs to varchars +
 380                 // size and value pairs for varchars
 381                 char *ptr = (char *)logInfo->data_;
 382                 void *ptrToTuple = (void *)*(long *)ptr; ptr += sizeof(void *);                 InUse *isUsed = ((InUse*)(ptrToTuple) - 1);
 383                 if (*isUsed == 0) {
 384                     printError(ErrSysFatal, "Fatal: Row is not in use during update rollback");
 385                 }
 386                 int tupleLen = *(int *) ptr; ptr += sizeof(int);
 387                 void *tuple = ptr; ptr += tupleLen;
 388                 void *metaData = (void *)*(long *)ptr; ptr += sizeof(void *);
 389                 Database db;
 390                 db.setMetaDataPtr((DatabaseMetaData *) metaData);
 391                 db.setProcSlot(sysdb->procSlot);
 392                 DbRetVal rv = OK;
 393                 int noOfVarchar = *(int *) ptr; ptr+= sizeof(int);
 394                 Chunk *vcchunk = (Chunk *) *(long *)ptr; ptr += sizeof(void *);
 395                 void **ptrToVarchars = (void **) ptr;
 396                 ptr += noOfVarchar * sizeof (void *);
 397                 char *lenValPtr = (char *) ptr;
 398                 for (int i = 0; i < noOfVarchar; i++) {
 399                     if (*(long *) ptrToVarchars[i] != 0L) {
 400                         vcchunk->free(&db, (void *)*(long *) ptrToVarchars[i]);
 401                         *(long *) ptrToVarchars[i] = 0L;
 402                     }
 403                 }
 404                 os::memcpy(ptrToTuple, tuple, tupleLen);
 405                 for (int i = 0; i < noOfVarchar; i++) {
 406                     int len = *(int *) lenValPtr; lenValPtr += sizeof(int);
 407                     if (len != 0) {
 408                         void *ptr = vcchunk->allocate(&db, len, &rv);
 409                         strcpy((char *)ptr, lenValPtr); lenValPtr += len;
 410                         *(long *) ptrToVarchars[i] = (long) ptr;
 411                     } else *(long *) ptrToVarchars[i] = 0L;
 412                 }
 413                 break;
 414              }
 415              case InsertHashIndexOperation:
 416              {
 417                 HashIndex::deleteLogicalUndoLog(sysdb, (char *)logInfo
 418                                                  + sizeof(UndoLogInfo));
 419                 break;
 420              }
 421              case DeleteHashIndexOperation:
 422              {
 423                 HashIndex::insertLogicalUndoLog(sysdb, (char *)logInfo
 424                                                  + sizeof(UndoLogInfo));
 425                 break;
 426              }
 427              case InsertTreeIndexOperation:
 428              {
 429                 TreeIndex::deleteLogicalUndoLog(sysdb, (char *)logInfo
 430                                                  + sizeof(UndoLogInfo));
 431                 break;
 432              }
 433              case DeleteTreeIndexOperation:
 434              {
 435                 TreeIndex::insertLogicalUndoLog(sysdb, (char *)logInfo
 436                                                  + sizeof(UndoLogInfo));
 437                 break;
 438              }
 439              default:
 440              {
 441                  printError(ErrSysFatal, "Illegal undo log type");
 442                  break;
 443              }
 444         }
 445         chunk->free(sysdb, logInfo);
 446     }
 447     return OK;
 448 }