Bug#34098384 Dbacc lock handling has too much overhead when WITH_DEBUG compiled

Backport:

WITH_DEBUG compiled code had some self check code which required the
lockOwnersList to be fully iterated. For large transactions, holding many
locks, the overhead of this becomes prohobitive. The value of these self
check loops were considered to be questionable.

This patch replace the lockOwnersList with a simple lockCount which enables
us to verify that all locks had been released before removing a fragment.
Other self-check code checking the lockOwnersList for lock not already
held or release is just removed.

Related methods for inserting / removing a lock into the lockOwnersList
are removed as well.

Change-Id: I3182c154f1afb5d1bac847432d3ec1d1b5c0d3ba

Author: Ole John Aske

storage/ndb/src/kernel/blocks/dbacc/Dbacc.hpp

@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2003, 2021, Oracle and/or its affiliates.
+   Copyright (c) 2003, 2022, Oracle and/or its affiliates.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License, version 2.0,
@@ -420,9 +420,9 @@ struct Fragmentrec {
   Uint32 expSenderPageptr;
 
 //-----------------------------------------------------------------------------
-// List of lock owners currently used only for self-check
+// Number of locks held on fragment, only for self-check
 //-----------------------------------------------------------------------------
-  Uint32 lockOwnersList;
+  Uint32 lockCount;
 
 //-----------------------------------------------------------------------------
 // References to Directory Ranges (which in turn references directories, which
@@ -695,15 +695,13 @@ struct Operationrec {
   Uint32 fid;
   Uint32 fragptr;
   LHBits32 hashValue;
-  Uint32 nextLockOwnerOp;
   Uint32 nextOp;
   Uint32 nextParallelQue;
   union {
     Uint32 nextSerialQue;      
     Uint32 m_lock_owner_ptr_i; // if nextParallelQue = RNIL, else undefined
   };
   Uint32 prevOp;
-  Uint32 prevLockOwnerOp;
   union {
     Uint32 prevParallelQue;
     Uint32 m_lo_last_parallel_op_ptr_i;
@@ -890,7 +888,6 @@ struct Tabrec {
   void initFragAdd(Signal*, FragmentrecPtr) const;
   void initFragPageZero(FragmentrecPtr, Page8Ptr) const;
   void initFragGeneral(FragmentrecPtr) const;
-  void verifyFragCorrect(FragmentrecPtr regFragPtr) const;
   void releaseFragResources(Signal* signal, Uint32 fragIndex);
   void releaseRootFragRecord(Signal* signal, RootfragmentrecPtr rootPtr) const;
   void releaseRootFragResources(Signal* signal, Uint32 tableId);
@@ -1062,8 +1059,6 @@ struct Tabrec {
 			  OperationrecPtr release_op) const;
   void allocOverflowPage();
   bool getfragmentrec(FragmentrecPtr&, Uint32 fragId);
-  void insertLockOwnersList(const OperationrecPtr&) const;
-  void takeOutLockOwnersList(const OperationrecPtr&) const;
 
   void initFsOpRec(Signal* signal) const;
   void initOverpage(Page8Ptr);

storage/ndb/src/kernel/blocks/dbacc/DbaccMain.cpp

@@ -646,7 +646,7 @@ void Dbacc::releaseFragResources(Signal* signal, Uint32 fragIndex)
   FragmentrecPtr regFragPtr;
   regFragPtr.i = fragIndex;
   ptrCheckGuard(regFragPtr, cfragmentsize, fragmentrec);
-  verifyFragCorrect(regFragPtr);
+  ndbrequire(regFragPtr.p->lockCount == 0);
 
   if (regFragPtr.p->expandOrShrinkQueued)
   {
@@ -705,11 +705,6 @@ void Dbacc::releaseFragResources(Signal* signal, Uint32 fragIndex)
   ndbassert(validatePageCount());
 }//Dbacc::releaseFragResources()
 
-void Dbacc::verifyFragCorrect(FragmentrecPtr regFragPtr)const
-{
-  ndbrequire(regFragPtr.p->lockOwnersList == RNIL);
-}//Dbacc::verifyFragCorrect()
-
 void Dbacc::releaseDirResources(Signal* signal)
 {
   jam();
@@ -721,7 +716,7 @@ void Dbacc::releaseDirResources(Signal* signal)
   FragmentrecPtr regFragPtr;
   regFragPtr.i = fragIndex;
   ptrCheckGuard(regFragPtr, cfragmentsize, fragmentrec);
-  verifyFragCorrect(regFragPtr);
+  ndbrequire(regFragPtr.p->lockCount == 0);
 
   DynArr256::Head* directory;
   ndbrequire(signal->theData[0] == ZREL_DIR);
@@ -1079,8 +1074,8 @@ void Dbacc::execACCKEYREQ(Signal* signal)
           dbgWord32(elemPageptr, elemptr, eh);
           elemPageptr.p->word32[elemptr] = eh;
 
-	  opbits |= Operationrec::OP_LOCK_OWNER;
-	  insertLockOwnersList(operationRecPtr);
+          opbits |= Operationrec::OP_LOCK_OWNER;
+          fragrecptr.p->lockCount++;
 
           fragrecptr.p->
             m_lockStats.req_start_imm_ok((opbits & 
@@ -1598,7 +1593,8 @@ void Dbacc::insertelementLab(Signal* signal,
   ndbrequire(operationRecPtr.p->tupkeylen <= fragrecptr.p->keyLength);
   ndbassert(!(operationRecPtr.p->m_op_bits & Operationrec::OP_LOCK_REQ));
 
-  insertLockOwnersList(operationRecPtr);
+  fragrecptr.p->lockCount++;
+  operationRecPtr.p->m_op_bits |= Operationrec::OP_LOCK_OWNER;
 
   operationRecPtr.p->reducedHashValue = fragrecptr.p->level.reduce(operationRecPtr.p->hashValue);
   const Uint32 tidrElemhead = ElementHeader::setLocked(operationRecPtr.i);
@@ -4834,7 +4830,7 @@ void Dbacc::abortOperation(Signal* signal)
 
   if (opbits & Operationrec::OP_LOCK_OWNER) 
   {
-    takeOutLockOwnersList(operationRecPtr);
+    fragrecptr.p->lockCount--;
     opbits &= ~(Uint32)Operationrec::OP_LOCK_OWNER;
     if (opbits & Operationrec::OP_INSERT_IS_DONE)
     { 
@@ -5013,7 +5009,7 @@ void Dbacc::commitOperation(Signal* signal)
 
   if (opbits & Operationrec::OP_LOCK_OWNER) 
   {
-    takeOutLockOwnersList(operationRecPtr);
+    fragrecptr.p->lockCount--;
     opbits &= ~(Uint32)Operationrec::OP_LOCK_OWNER;
     operationRecPtr.p->m_op_bits = opbits;
 
@@ -5340,7 +5336,8 @@ Dbacc::release_lockowner(Signal* signal, OperationrecPtr opPtr, bool commit)
     action = START_NEW;
   }
 
-  insertLockOwnersList(newOwner);
+  fragrecptr.p->lockCount++;
+  newOwner.p->m_op_bits |= Operationrec::OP_LOCK_OWNER;
 
   /**
    * Copy op info, and store op in element
@@ -5497,99 +5494,6 @@ Dbacc::startNew(Signal* signal, OperationrecPtr newOwner)
   return;
 }
 
-/**
- * takeOutLockOwnersList
- *
- * Description: Take out an operation from the doubly linked 
- * lock owners list on the fragment.
- *
- */
-void Dbacc::takeOutLockOwnersList(const OperationrecPtr& outOperPtr) const
-{
-  const Uint32 Tprev = outOperPtr.p->prevLockOwnerOp;
-  const Uint32 Tnext = outOperPtr.p->nextLockOwnerOp;
-#ifdef VM_TRACE
-  // Check that operation is already in the list
-  OperationrecPtr tmpOperPtr;
-  bool inList = false;
-  tmpOperPtr.i = fragrecptr.p->lockOwnersList;
-  while (tmpOperPtr.i != RNIL){
-    ptrCheckGuard(tmpOperPtr, coprecsize, operationrec);
-    if (tmpOperPtr.i == outOperPtr.i)
-      inList = true;
-    tmpOperPtr.i = tmpOperPtr.p->nextLockOwnerOp;
-  }
-  ndbrequire(inList == true);
-#endif
-  
-  ndbassert(outOperPtr.p->m_op_bits & Operationrec::OP_LOCK_OWNER);
-  
-  // Fast path through the code for the common case.
-  if ((Tprev == RNIL) && (Tnext == RNIL)) {
-    ndbrequire(fragrecptr.p->lockOwnersList == outOperPtr.i);
-    fragrecptr.p->lockOwnersList = RNIL;
-    return;
-  } 
-
-  // Check previous operation 
-  if (Tprev != RNIL) {    
-    jam();
-    arrGuard(Tprev, coprecsize);
-    operationrec[Tprev].nextLockOwnerOp = Tnext;
-  } else {
-    fragrecptr.p->lockOwnersList = Tnext;
-  }//if
-
-  // Check next operation
-  if (Tnext == RNIL) {
-    return;
-  } else {
-    jam();
-    arrGuard(Tnext, coprecsize);
-    operationrec[Tnext].prevLockOwnerOp = Tprev;
-  }//if
-
-  return;
-}//Dbacc::takeOutLockOwnersList()
-
-/**
- * insertLockOwnersList
- *
- * Description: Insert an operation first in the dubly linked lock owners 
- * list on the fragment.
- *
- */
-void Dbacc::insertLockOwnersList(const OperationrecPtr& insOperPtr) const
-{
-  OperationrecPtr tmpOperPtr;
-#ifdef VM_TRACE
-  // Check that operation is not already in list
-  tmpOperPtr.i = fragrecptr.p->lockOwnersList;
-  while(tmpOperPtr.i != RNIL){
-    ptrCheckGuard(tmpOperPtr, coprecsize, operationrec);
-    ndbrequire(tmpOperPtr.i != insOperPtr.i);
-    tmpOperPtr.i = tmpOperPtr.p->nextLockOwnerOp;    
-  }
-#endif
-  tmpOperPtr.i = fragrecptr.p->lockOwnersList;
-  
-  ndbrequire(! (insOperPtr.p->m_op_bits & Operationrec::OP_LOCK_OWNER));
-
-  insOperPtr.p->m_op_bits |= Operationrec::OP_LOCK_OWNER;
-  insOperPtr.p->prevLockOwnerOp = RNIL;
-  insOperPtr.p->nextLockOwnerOp = tmpOperPtr.i;
-  
-  fragrecptr.p->lockOwnersList = insOperPtr.i;
-  if (tmpOperPtr.i == RNIL) {
-    return;
-  } else {
-    jam();
-    ptrCheckGuard(tmpOperPtr, coprecsize, operationrec);
-    tmpOperPtr.p->prevLockOwnerOp = insOperPtr.i;
-  }//if
-}//Dbacc::insertLockOwnersList()
-
-
 /* --------------------------------------------------------------------------------- */
 /* --------------------------------------------------------------------------------- */
 /* --------------------------------------------------------------------------------- */
@@ -7043,8 +6947,7 @@ void Dbacc::initFragGeneral(FragmentrecPtr regFragPtr)const
 {
   new (&regFragPtr.p->directory) DynArr256::Head();
 
-  regFragPtr.p->lockOwnersList = RNIL;
-
+  regFragPtr.p->lockCount = 0;
   regFragPtr.p->hasCharAttr = ZFALSE;
   regFragPtr.p->dirRangeFull = ZFALSE;
   regFragPtr.p->fragState = FREEFRAG;
@@ -7326,9 +7229,10 @@ void Dbacc::checkNextBucketLab(Signal* signal)
                          getHighResTimer());
 
       setlock(nsPageptr, tnsElementptr);
-      insertLockOwnersList(operationRecPtr);
-      operationRecPtr.p->m_op_bits |= 
-	Operationrec::OP_STATE_RUNNING | Operationrec::OP_RUN_QUEUE;
+      fragrecptr.p->lockCount++;
+      operationRecPtr.p->m_op_bits |=
+        Operationrec::OP_LOCK_OWNER |
+        Operationrec::OP_STATE_RUNNING | Operationrec::OP_RUN_QUEUE;
     }//if
   } else {
     arrGuard(tnsElementptr, 2048);
@@ -8027,26 +7931,6 @@ void Dbacc::putActiveScanOp() const
  */
 void Dbacc::putOpScanLockQue() const
 {
-
-#ifdef VM_TRACE
-  // DEBUG CODE
-  // Check that there are as many operations in the lockqueue as 
-  // scanLockHeld indicates
-  OperationrecPtr tmpOp;
-  int numLockedOpsBefore = 0;
-  tmpOp.i = scanPtr.p->scanFirstLockedOp;
-  while(tmpOp.i != RNIL){
-    numLockedOpsBefore++;
-    ptrCheckGuard(tmpOp, coprecsize, operationrec);
-    if (tmpOp.p->nextOp == RNIL)
-    {
-      ndbrequire(tmpOp.i == scanPtr.p->scanLastLockedOp);
-    }
-    tmpOp.i = tmpOp.p->nextOp;
-  } 
-  ndbrequire(numLockedOpsBefore==scanPtr.p->scanLockHeld);
-#endif
-
   OperationrecPtr pslOperationRecPtr;
   ScanRec theScanRec;
   theScanRec = *scanPtr.p;
@@ -8414,25 +8298,6 @@ void Dbacc::takeOutScanLockQueue(Uint32 scanRecIndex) const
     TscanPtr.p->scanLastLockedOp = operationRecPtr.p->prevOp;
   }//if
   TscanPtr.p->scanLockHeld--;
-
-#ifdef VM_TRACE
-  // DEBUG CODE
-  // Check that there are as many operations in the lockqueue as 
-  // scanLockHeld indicates
-  OperationrecPtr tmpOp;
-  int numLockedOps = 0;
-  tmpOp.i = TscanPtr.p->scanFirstLockedOp;
-  while(tmpOp.i != RNIL){
-    numLockedOps++;
-    ptrCheckGuard(tmpOp, coprecsize, operationrec);
-    if (tmpOp.p->nextOp == RNIL)
-    {
-      ndbrequire(tmpOp.i == TscanPtr.p->scanLastLockedOp);
-    }
-    tmpOp.i = tmpOp.p->nextOp;
-  } 
-  ndbrequire(numLockedOps==TscanPtr.p->scanLockHeld);
-#endif
 }//Dbacc::takeOutScanLockQueue()
 
 /* --------------------------------------------------------------------------------- */
@@ -9215,15 +9080,13 @@ Dbacc::execDUMP_STATE_ORD(Signal* signal)
     infoEvent("fid=%d, fragptr=%d ",
               tmpOpPtr.p->fid, tmpOpPtr.p->fragptr);
     infoEvent("hashValue=%d", tmpOpPtr.p->hashValue.pack());
-    infoEvent("nextLockOwnerOp=%d, nextOp=%d, nextParallelQue=%d ",
-	      tmpOpPtr.p->nextLockOwnerOp, tmpOpPtr.p->nextOp, 
-	      tmpOpPtr.p->nextParallelQue);
+    infoEvent("nextOp=%d, nextParallelQue=%d ",
+	      tmpOpPtr.p->nextOp, tmpOpPtr.p->nextParallelQue);
     infoEvent("nextSerialQue=%d, prevOp=%d ",
 	      tmpOpPtr.p->nextSerialQue, 
 	      tmpOpPtr.p->prevOp);
-    infoEvent("prevLockOwnerOp=%d, prevParallelQue=%d",
-	      tmpOpPtr.p->prevLockOwnerOp, tmpOpPtr.p->nextParallelQue);
-    infoEvent("prevSerialQue=%d, scanRecPtr=%d",
+    infoEvent("prevParallelQue=%d, prevSerialQue=%d, scanRecPtr=%d",
+	      tmpOpPtr.p->prevParallelQue,
 	      tmpOpPtr.p->prevSerialQue, tmpOpPtr.p->scanRecPtr);
     infoEvent("m_op_bits=0x%x, reducedHashValue=%x ",
               tmpOpPtr.p->m_op_bits, tmpOpPtr.p->reducedHashValue.pack());