src/storage/LockManager.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<Lock.h>
  17 #include<Allocator.h>
  18 #include<Database.h>
  19 #include<CatalogTables.h>
  20 #include<Transaction.h>
  21 #include<Debug.h>
  22 #include<Config.h>
  23 #include<Process.h>
  24 Bucket* LockManager::getLockBucket(void *tuple)
  25 {
  26    int noOfBuckets = LOCK_BUCKET_SIZE;
  27    Bucket* buckets = systemDatabase_->getLockHashBuckets();
  28    unsigned long key =(unsigned long)tuple ;
  29    int bucketNo = HashIndex::computeHashBucket(typeULong, &key, noOfBuckets);
  30
  31    Bucket *bucket = &(buckets[bucketNo]);
  32    printDebug(DM_Lock, "getLockBucket bucketno:%d bucket:%x",bucketNo, bucket);
  33    return bucket;
  34 }
  35
  36 void LockManager::printUsageStatistics()
  37 {
  38    Bucket* buckets = systemDatabase_->getLockHashBuckets();
  39    Bucket* bucket;
  40    LockHashNode *lockNode;
  41    int nodeCount =0, bucketCount =0;
  42    for (int i =0; i< LOCK_BUCKET_SIZE; i++)
  43    {
  44        bucket = &(buckets[i]);
  45        lockNode = (LockHashNode*) bucket->bucketList_;
  46        if (lockNode) bucketCount++; else continue;
  47        while (NULL != lockNode) { nodeCount++; lockNode = lockNode->next_; }
  48    }
  49    printf("<LockTable>\n");
  50    printf("  <TotalBuckets> %d  </TotalBuckets>\n", LOCK_BUCKET_SIZE);
  51    printf("  <UsedBuckets> %d  </UsedBuckets>\n", bucketCount);
  52    printf("  <TotalLockNodes> %d  </TotalLockNodes>\n", nodeCount);
  53    printf("</LockTable>\n");
  54
  55 }
  56
  57 void LockManager::printDebugInfo()
  58 {
  59    Bucket* buckets = systemDatabase_->getLockHashBuckets();
  60    Bucket* bucket;
  61    LockHashNode *lockNode;
  62    int nodeCount =0, bucketCount =0;
  63    printf("<LockTable>\n");
  64    for (int i =0; i< LOCK_BUCKET_SIZE; i++)
  65    {
  66        nodeCount =0;
  67        bucket = &(buckets[i]);
  68        //if (bucket) bucketCount++; else continue;
  69        lockNode = (LockHashNode*) bucket->bucketList_;
  70
  71        while (NULL != lockNode)
  72        {
  73            nodeCount++;
  74            lockNode->print();
  75            lockNode = lockNode->next_;
  76        }
  77        if (nodeCount) {
  78            bucketCount++;
  79            printf("  <LockBucket> \n");
  80            printf("    <BucketNo> %d </BucketNo> \n", i);
  81            printf("    <TotalNodes> %d </TotalNodes>\n", nodeCount);
  82            printf("  <LockBucket>\n");
  83        }
  84    }
  85
  86    printf("  <TotalUsedBuckets> %d </TotalUsedBuckets>\n", bucketCount);
  87    Chunk *chunk = systemDatabase_->getSystemDatabaseChunk(LockTableId);
  88    printf("  <TotalPages> %d </TotalPages>\n", chunk->totalPages());
  89    printf("</LockTable>\n");
  90
  91 }
  92
  93 DbRetVal LockManager::getSharedLock(void *tuple, Transaction **trans)
  94 {
  95    //get the bucket list
  96    //take the bucket mutex for read
  97    //go the the next level bucket list
  98    //get the bucket iterator
  99    //go the node where the lock info resides
 100    //check which mode the lock is taken
 101    // if shared then
 102    //      upgrade the bucket mutex to write
 103    //      take it and increment the readers count
 104    //      release bucket mutex and exit
 105    // if exclusive then
 106    //       go into the loop
 107    //       upgrade the bucket mutex to write
 108    //       increment waitReaders count
 109    //       release the bucket mutex
 110    //       wait for timeout period or (takes shared lock and release it ) till it becomes free.
 111    //       if times out
 112    //            take bucket mutex for write
 113    //            decrement waitReaders count
 114    //            releaese bucket mutex
 115
 116    //            return
 117    //       if it becomes free
 118    //            take bucket mutex for write
 119    //            increment readers
 120    //            releaese bucket mutex
 121
 122    //            return
 123    LockInfo linfo;
 124    linfo.noOfReaders_ = 1;
 125    //keeping it ready for the allocation, because when
 126    //lock node is not present in the list, then it means we are the first
 127    //to acquire lock so for sure we will get it.
 128    printDebug(DM_Lock, "LockManager::getSharedLock Begin");
 129    Bucket *bucket = getLockBucket(tuple);
 130    int lockRet = bucket->mutex_.getLock(systemDatabase_->procSlot);
 131    if (lockRet != 0)
 132    {
 133        printDebug(DM_Lock, "LockManager::getSharedLock:End-Unable to get mutex");
 134        printError(ErrLockTimeOut,"Unable to acquire bucket mutex");
 135        return ErrLockTimeOut;
 136    }
 137    LockHashNode *lockNode = (LockHashNode*) bucket->bucketList_;
 138    if (NULL == lockNode)
 139    {
 140         DbRetVal rv = OK;
 141         LockHashNode *node = allocLockNode(linfo, tuple, &rv);
 142         if (NULL == node)
 143         {
 144             printError(rv, "Could not allocate Lock node");
 145             return rv;
 146             }
 147         printDebug(DM_Lock, "Bucket list is null: Allocating new LockHashNode %x", node);
 148         bucket->bucketList_ = (void*)node; //make it as head
 149         bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 150         rv = OK;
 151         if (trans != NULL)
 152             rv = (*trans)->insertIntoHasList(systemDatabase_, node);
 153         if (rv !=OK) linfo.noOfReaders_--;
 154         printDebug(DM_Lock, "LockManager::getSharedLock End");
 155         return rv;
 156    }
 157    LockHashNode *cachedLockNode = NULL;
 158
 159    LockHashNode *iter = lockNode;
 160    //Iterate though the list and find the element's lock info
 161    while(iter != NULL)
 162    {
 163        if(iter->ptrToTuple_ == tuple)
 164        {
 165            if (iter->lInfo_.noOfReaders_ == -1)
 166            {
 167
 168                iter->lInfo_.waitReaders_++;
 169                        cachedLockNode = iter;
 170                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 171                if (trans != NULL) (*trans)->updateWaitLock(iter);
 172                printDebug(DM_Lock, "lock node:%x exclusive locked",iter);
 173                break;
 174            }
 175            else if (iter->lInfo_.noOfReaders_ == 0)
 176            {
 177                if(iter->lInfo_.waitWriters_ >0)
 178                {
 179                    iter->lInfo_.waitReaders_++;
 180                    cachedLockNode = iter;
 181                    bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 182                    if (trans != NULL) (*trans)->updateWaitLock(iter);
 183                    printDebug(DM_Lock, "lock node:%x Writers waiting.",iter);
 184                    break;
 185                }
 186                else
 187                {
 188                    iter->lInfo_.noOfReaders_++;
 189                    bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 190                    DbRetVal rv = OK;
 191                    if (trans != NULL)
 192                      rv = (*trans)->insertIntoHasList(systemDatabase_, iter);
 193                    if (rv != OK) iter->lInfo_.noOfReaders_--;
 194                    printDebug(DM_Lock, "lock node:%x First to take shared lock",
 195                                                                           iter);
 196                    printDebug(DM_Lock, "LockManager::getSharedLock End");
 197                    return rv;
 198                 }
 199            }else
 200            {
 201                iter->lInfo_.noOfReaders_++;
 202                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 203                DbRetVal rv = OK;
 204                if (trans != NULL)
 205                  rv = (*trans)->insertIntoHasList(systemDatabase_, iter);
 206                if (rv != OK) iter->lInfo_.noOfReaders_--;
 207                printDebug(DM_Lock, "lock node:%x incr readers",iter);
 208                printDebug(DM_Lock, "LockManager::getSharedLock End");
 209                return rv;
 210            }
 211        }
 212        printDebug(DM_Lock, "Finding the lock node. iter:%x",iter);
 213        iter = iter->next_;
 214    }
 215    if (NULL == cachedLockNode)
 216    {
 217         DbRetVal rv =OK;
 218         LockHashNode *node = allocLockNode(linfo, tuple, &rv);
 219         if (NULL == node)
 220         {
 221             bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 222             printError(rv, "Could not allocate Lock node");
 223             if (trans != NULL) (*trans)->removeWaitLock();
 224             return rv;
 225         }
 226         printDebug(DM_Lock,"Not Found.Created new lock node:%x",node);
 227         LockHashNode *it = lockNode;
 228         while (NULL != it->next_) it = it->next_;
 229         it->next_ = node;
 230         bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 231         rv = OK;
 232         if (trans != NULL)
 233           rv = (*trans)->insertIntoHasList(systemDatabase_, node);
 234         if (rv != OK) linfo.noOfReaders_--;
 235         if (trans != NULL) (*trans)->removeWaitLock();
 236         printDebug(DM_Lock, "LockManager::getSharedLock End");
 237         return rv;
 238    }
 239    //bucket->mutex_.releaseLock();
 240    int tries = 0;
 241    int ret = 0;
 242    struct timeval timeout;
 243    timeout.tv_sec = Conf::config.getLockSecs();
 244    timeout.tv_usec = Conf::config.getLockUSecs();
 245
 246    //printDebug(DM_Lock, "Trying to get mutex: for bucket %x\n", bucket);
 247    while (tries < Conf::config.getLockRetries())
 248    {
 249        lockRet = bucket->mutex_.getLock(systemDatabase_->procSlot);
 250        if (lockRet != 0)
 251        {
 252            printDebug(DM_Lock, "Mutex is waiting for long time:May be deadlock");
 253            printDebug(DM_Lock, "LockManager::getSharedLock End");
 254            printError(ErrLockTimeOut, "Unable to get bucket mutex");
 255            if (trans != NULL) (*trans)->removeWaitLock();
 256            return ErrLockTimeOut;
 257        }
 258        if (cachedLockNode->lInfo_.noOfReaders_ == 0)
 259        {
 260            //if there are waiters allow then to take the lock
 261            if (cachedLockNode->lInfo_.waitWriters_ <0)
 262            {
 263                cachedLockNode->lInfo_.noOfReaders_++;
 264                cachedLockNode->lInfo_.waitReaders_--;
 265                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 266                DbRetVal rv = OK;
 267                if (trans != NULL)
 268                  rv = (*trans)->insertIntoHasList(systemDatabase_, cachedLockNode);
 269                if (rv !=OK) {
 270                    cachedLockNode->lInfo_.noOfReaders_--;
 271                    cachedLockNode->lInfo_.waitReaders_++;
 272                    if (trans != NULL) (*trans)->removeWaitLock();
 273                    return rv;
 274                }
 275                if (trans != NULL) (*trans)->removeWaitLock();
 276                printDebug(DM_Lock, "LockManager::getSharedLock End");
 277                return OK;
 278            }
 279        } else if (cachedLockNode->lInfo_.noOfReaders_ == -1)
 280        {
 281            if (trans !=NULL && (*trans)->findInHasList(systemDatabase_, cachedLockNode))
 282            {
 283                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 284                if (trans != NULL) (*trans)->removeWaitLock();
 285                printDebug(DM_Lock, "LockManager::getSharedLock End");
 286                return OK;
 287            }
 288        } else
 289        {
 290            cachedLockNode->lInfo_.noOfReaders_++;
 291            cachedLockNode->lInfo_.waitReaders_--;
 292            bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 293            DbRetVal rv =OK;
 294            if (trans != NULL)
 295               rv = (*trans)->insertIntoHasList(systemDatabase_, cachedLockNode);
 296            if (rv !=OK) {
 297                cachedLockNode->lInfo_.noOfReaders_--;
 298                cachedLockNode->lInfo_.waitReaders_++;
 299            }
 300            if (trans != NULL) (*trans)->removeWaitLock();
 301            printDebug(DM_Lock, "LockManager::getSharedLock End");
 302            return rv;
 303        }
 304
 305        bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 306        os::select(0, 0, 0, 0, &timeout);
 307        tries++;
 308        printDebug(DM_Lock, "Trying to lock the lock node:%x iteration:%d",cachedLockNode, tries);
 309    }
 310    printDebug(DM_Lock, "Mutex is waiting for long time:May be deadlock");
 311    printDebug(DM_Lock, "LockManager::getSharedLock End");
 312    printError(ErrLockTimeOut, "Unable to acquire lock for long time.Timed out");
 313    if (trans != NULL) (*trans)->removeWaitLock();
 314    return ErrLockTimeOut;
 315
 316 }
 317
 318 DbRetVal LockManager::getExclusiveLock(void *tuple, Transaction **trans)
 319 {
 320    LockInfo linfo;
 321    linfo.noOfReaders_ = -1;
 322    printDebug(DM_Lock, "LockManager::getExclusiveLock Begin");
 323    //keeping it ready for the allocation, because when
 324    //lock node is not present in the list, then it means we are the first
 325    //to acquire lock so for sure we will get it.
 326
 327    Bucket *bucket = getLockBucket(tuple);
 328    int lockRet = bucket->mutex_.getLock(systemDatabase_->procSlot);
 329    if (lockRet != 0)
 330    {
 331        printDebug(DM_Lock, "Unable to acquire bucket mutex:May be deadlock");
 332        printError(ErrLockTimeOut, "Unable to acquire bucket mutex");
 333        return ErrLockTimeOut;
 334    }
 335    LockHashNode *lockNode = (LockHashNode*) bucket->bucketList_;
 336    if (NULL == lockNode)
 337    {
 338         DbRetVal rv = OK;
 339         LockHashNode *node = allocLockNode(linfo, tuple, &rv);
 340         if (NULL == node)
 341         {
 342             printError(rv, "Could not allocate Lock node");
 343             return rv;
 344         }
 345         printDebug(DM_Lock, "No head. So new lock node allocated:%x",node);
 346         bucket->bucketList_ = (void*)node; //make it as head
 347         bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 348         rv =OK;
 349         if (trans != NULL)
 350            rv = (*trans)->insertIntoHasList(systemDatabase_, node);
 351         if (rv !=OK) linfo.noOfReaders_ = 0;
 352         printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 353         return rv;
 354
 355    }
 356    LockHashNode *cachedLockNode = NULL;
 357
 358    LockHashNode *iter = lockNode;
 359    //Iterate though the list and find the element's lock info
 360    while(iter != NULL)
 361    {
 362        if(iter->ptrToTuple_ == tuple)
 363        {
 364            if (iter->lInfo_.noOfReaders_ != 0)
 365            {
 366                iter->lInfo_.waitWriters_++;
 367                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 368                if (trans != NULL) (*trans)->updateWaitLock(iter);
 369                cachedLockNode = iter;
 370                printDebug(DM_Lock, "Either some one has exclusive or shared lock:%x",iter);
 371                break;
 372            }
 373            else
 374            {
 375                iter->lInfo_.noOfReaders_ = -1;
 376                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 377                DbRetVal rv =OK;
 378                if (trans != NULL)
 379                  rv = (*trans)->insertIntoHasList(systemDatabase_, iter);
 380                if (rv != OK) iter->lInfo_.noOfReaders_ = 0;
 381                printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 382                return rv;
 383            }
 384        }
 385        printDebug(DM_Lock, "Finding the lock node. iter:%x",iter);
 386        iter = iter->next_;
 387    }
 388    if (NULL == cachedLockNode)
 389    {
 390         DbRetVal rv =OK;
 391         LockHashNode *node = allocLockNode(linfo, tuple, &rv);
 392         if (NULL == node)
 393         {
 394             bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 395             if (trans != NULL) (*trans)->updateWaitLock(NULL);
 396             printError(rv, "Could not allocate Lock node");
 397             return rv;
 398         }
 399         printDebug(DM_Lock, "Not Found:Creating new lock node:%x",node);
 400         LockHashNode *it = lockNode;
 401         while (NULL != it->next_) it = it->next_;
 402         it->next_ = node;
 403         bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 404         rv = OK;
 405         if (trans != NULL)
 406           rv = (*trans)->insertIntoHasList(systemDatabase_, node);
 407         if (rv != OK) linfo.noOfReaders_ = 0;
 408          if (trans != NULL) (*trans)->removeWaitLock();
 409         printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 410         return rv;
 411    }
 412    //bucket->mutex_.releaseLock();
 413    int tries = 0;
 414    int ret = 0;
 415    struct timeval timeout;
 416    timeout.tv_sec = Conf::config.getLockSecs();
 417    timeout.tv_usec = Conf::config.getLockUSecs();
 418
 419    while (tries < Conf::config.getLockRetries())
 420    {
 421        lockRet = bucket->mutex_.getLock(systemDatabase_->procSlot);
 422        if (lockRet != 0)
 423        {
 424            printError(ErrLockTimeOut, "Unable to get bucket mutex");
 425            return ErrLockTimeOut;
 426        }
 427        if (cachedLockNode->lInfo_.noOfReaders_ == 0)
 428        {
 429            cachedLockNode->lInfo_.noOfReaders_ = -1;
 430            cachedLockNode->lInfo_.waitWriters_--;
 431            bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 432            DbRetVal rv ;
 433            if (trans != NULL)
 434                rv = (*trans)->insertIntoHasList(systemDatabase_, cachedLockNode);
 435            if (trans != NULL) (*trans)->removeWaitLock();
 436            printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 437            return rv;
 438        }else if ( cachedLockNode->lInfo_.noOfReaders_ == 1)
 439        {
 440            if (trans !=NULL && (*trans)->findInHasList(systemDatabase_, cachedLockNode))
 441            {
 442                printDebug(DM_Lock, "upgrading shared to exclusive lock:%x",
 443                                                            cachedLockNode);
 444                //upgrade it to exclusive lock
 445                cachedLockNode->lInfo_.noOfReaders_ = -1;
 446                cachedLockNode->lInfo_.waitWriters_--;
 447                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 448                if (trans != NULL) (*trans)->removeWaitLock();
 449                printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 450                return OK;
 451            }
 452            if (trans ==NULL && ProcessManager::hasLockList.exists(cachedLockNode->ptrToTuple_))
 453            {
 454                printDebug(DM_Lock, "upgrading shared to exclusive lock:%x",
 455                                                            cachedLockNode);
 456                //upgrade it to exclusive lock
 457                cachedLockNode->lInfo_.noOfReaders_ = -1;
 458                cachedLockNode->lInfo_.waitWriters_--;
 459                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 460                if (trans != NULL) (*trans)->removeWaitLock();
 461                printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 462                return OK;
 463            }
 464        }else if ( cachedLockNode->lInfo_.noOfReaders_ == -1)
 465        {
 466            if (trans !=NULL && (*trans)->findInHasList(systemDatabase_, cachedLockNode))
 467            {
 468                printDebug(DM_Lock, "You already have exclusive lock:%x",
 469                                                         cachedLockNode);
 470                cachedLockNode->lInfo_.waitWriters_--;
 471                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 472                if (trans != NULL) (*trans)->removeWaitLock();
 473                printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 474                return OK;
 475            }
 476            if (trans ==NULL && ProcessManager::hasLockList.exists(cachedLockNode->ptrToTuple_))
 477            {
 478                printDebug(DM_Lock, "You already have exclusive lock:%x",
 479                                                            cachedLockNode);
 480                cachedLockNode->lInfo_.waitWriters_--;
 481                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 482                if (trans != NULL) (*trans)->removeWaitLock();
 483                printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 484                return OK;
 485            }
 486        }
 487        bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 488        os::select(0, 0, 0, 0, &timeout);
 489        tries++;
 490        printDebug(DM_Lock, "Trying to lock the lock node:%x iteration:%d",cachedLockNode, tries);
 491    }
 492    printDebug(DM_Lock, "LockManager::getExclusiveLock End");
 493    if (trans != NULL) (*trans)->removeWaitLock();
 494    printError(ErrLockTimeOut, "Unable to acquire lock for long time.Timed out");
 495    return ErrLockTimeOut;
 496 }
 497
 498 DbRetVal LockManager::releaseLock(void *tuple)
 499 {
 500    LockInfo linfo;
 501    linfo.noOfReaders_ = 0;
 502    //keeping it ready for the allocation, because when
 503    //lock node is not present in the list, then it means we are the first
 504    //to acquire lock so for sure we will get it.
 505    printDebug(DM_Lock, "LockManager:releaseLock Start");
 506    Bucket *bucket = getLockBucket(tuple);
 507    printDebug(DM_Lock,"Bucket is %x", bucket);
 508    int lockRet = bucket->mutex_.getLock(systemDatabase_->procSlot);
 509    if (lockRet != 0)
 510    {
 511        printDebug(DM_Lock, "Mutex is waiting for long time:May be deadlock");
 512        printDebug(DM_Lock, "LockManager:releaseLock End");
 513        printError(ErrLockTimeOut, "Unable to get bucket mutex");
 514        return ErrLockTimeOut;
 515    }
 516    LockHashNode *lockNode = (LockHashNode*) bucket->bucketList_;
 517    if (NULL == lockNode)
 518    {
 519        bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 520        printDebug(DM_Lock, "LockManager:releaseLock End");
 521        printError(ErrSysFatal, "Lock Element Not found: Probable Data Corruption.\n");
 522        return ErrSysFatal;
 523    }
 524
 525    LockHashNode *iter = lockNode;
 526    //Iterate though the list and find the element's lock info
 527    while(iter != NULL)
 528    {
 529        if(iter->ptrToTuple_ == tuple)
 530        {
 531
 532            if (iter->lInfo_.noOfReaders_ == -1)
 533            {
 534                iter->lInfo_.noOfReaders_ = 0;
 535                if (iter->lInfo_.waitWriters_ == 0 || iter->lInfo_.waitReaders_ ==0)
 536                {
 537                    deallocLockNode(iter, bucket);
 538                    bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 539                    printDebug(DM_Lock, "Releasing exclusive lock and dealloc node:%x",
 540                                                                                 iter);
 541                    printDebug(DM_Lock, "LockManager:releaseLock End");
 542                    return OK;
 543                }
 544                else
 545                {
 546                    bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 547                    printDebug(DM_Lock, "Releasing exclusive lock");
 548                    printDebug(DM_Lock, "LockManager:releaseLock End");
 549                    return OK;
 550                }
 551            }
 552            else if (iter->lInfo_.noOfReaders_ == 1)
 553            {
 554                iter->lInfo_.noOfReaders_ = 0;
 555                if (iter->lInfo_.waitWriters_ == 0 || iter->lInfo_.waitReaders_ ==0)
 556                {
 557                    deallocLockNode(iter, bucket);
 558                    bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 559                    printDebug(DM_Lock, "Releasing read lock and dealloc node:%x",iter);
 560                    printDebug(DM_Lock, "LockManager:releaseLock End");
 561                    return OK;
 562                }
 563            }
 564            else
 565            {
 566                iter->lInfo_.noOfReaders_--;
 567                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 568                printDebug(DM_Lock, "Decrementing read lock:%x",iter);
 569                printDebug(DM_Lock, "LockManager:releaseLock End");
 570                return OK;
 571            }
 572
 573        }
 574        printDebug(DM_Lock, "Finding the lock node. iter:%x",iter);
 575        iter = iter->next_;
 576    }
 577    bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 578    printError(ErrSysFatal, "Lock Element Not found: Probable Data Corruption");
 579    return ErrSysFatal;
 580 }
 581
 582 DbRetVal LockManager::isExclusiveLocked(void *tuple, Transaction **trans, bool &status)
 583 {
 584    Bucket *bucket = getLockBucket(tuple);
 585    printDebug(DM_Lock,"Bucket is %x", bucket);
 586    int lockRet = bucket->mutex_.getLock(systemDatabase_->procSlot);
 587    if (lockRet != 0)
 588    {
 589        printDebug(DM_Lock, "Mutex is waiting for long time:May be deadlock");
 590        printDebug(DM_Lock, "LockManager:releaseLock End");
 591        printError(ErrLockTimeOut, "Unable to get bucket mutex");
 592        return ErrLockTimeOut;
 593    }
 594    LockHashNode *lockNode = (LockHashNode*) bucket->bucketList_;
 595    if (NULL == lockNode)
 596    {
 597        bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 598        printDebug(DM_Lock, "bucketList is empty. so data element not locked");
 599        status = false;
 600        return OK;
 601    }
 602
 603    LockHashNode *iter = lockNode;
 604    //Iterate though the list and find the element's lock info
 605    //Only exclusive locks are checked. shared locks are not considered for this
 606    while(iter != NULL)
 607    {
 608        if(iter->ptrToTuple_ == tuple)
 609        {
 610            if (iter->lInfo_.noOfReaders_ == -1)
 611            {
 612                if (trans != NULL && (*trans)->findInHasList(systemDatabase_, iter))
 613                {
 614                    printDebug(DM_Lock, "You already have exclusive Lock: %x", iter);
 615                    status = false;
 616                }
 617                else
 618                    status = true;
 619                bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 620                return OK;
 621            }
 622        }
 623        printDebug(DM_Lock, "Finding the lock node. iter:%x",iter);
 624        iter = iter->next_;
 625    }
 626    bucket->mutex_.releaseLock(systemDatabase_->procSlot);
 627    status = false;
 628    return OK;
 629 }
 630
 631 LockHashNode* LockManager::allocLockNode(LockInfo &info, void *tuple, DbRetVal *rv)
 632 {
 633     //allocate lock node
 634     Chunk *chunk = systemDatabase_->getSystemDatabaseChunk(LockTableId);
 635     LockHashNode *node = (LockHashNode*)chunk->allocate(systemDatabase_, rv);
 636     if (NULL == node)
 637     {
 638         printError(*rv, "Could not allocate Lock node");
 639         return NULL;
 640     }
 641     node->ptrToTuple_ = tuple;
 642     node->lInfo_ = info;
 643     node->next_ = NULL;
 644     return node;
 645 }
 646
 647 void LockManager::deallocLockNode(LockHashNode *node, Bucket *bucket)
 648 {
 649     Chunk *chunk = systemDatabase_->getSystemDatabaseChunk(LockTableId);
 650     LockHashNode *nodeList = (LockHashNode*) bucket->bucketList_;
 651     LockHashNode *iter = nodeList, *prev = nodeList;
 652     if (NULL == nodeList)
 653     {
 654         printError(ErrSysFatal, "Lock Bucket corrupted");
 655         return;
 656     }
 657     //If it is the first node, then make the bucket point to the next node
 658     //in the list
 659     if (nodeList == node)
 660     {
 661        bucket->bucketList_ = node->next_;
 662        chunk->free(systemDatabase_, node);
 663        return;
 664     }
 665     while(iter != node)
 666     {
 667         prev = iter;
 668         iter = iter->next_;
 669     }
 670     //delete the node by making previous element point to the next element
 671     //of the deleted element in the list
 672     prev->next_ = iter->next_;
 673     chunk->free(systemDatabase_, node);
 674     return ;
 675 }