Merge perch.ndb.mysql.com:/home/jonas/src/50-work

into  perch.ndb.mysql.com:/home/jonas/src/51-work

storage/ndb/include/ndbapi/NdbTransaction.hpp

@@ -658,7 +658,10 @@ private:
   // Release all cursor operations in connection
   void releaseOps(NdbOperation*);	
   void releaseScanOperations(NdbIndexScanOperation*);	
-  void releaseScanOperation(NdbIndexScanOperation*);
+  bool releaseScanOperation(NdbIndexScanOperation** listhead,
+			    NdbIndexScanOperation** listtail,
+			    NdbIndexScanOperation* op);
+  void releaseExecutedScanOperation(NdbIndexScanOperation*);
   
   // Set the transaction identity of the transaction
   void		setTransactionId(Uint64 aTransactionId);

storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp

@@ -10015,15 +10015,19 @@ void Dbdih::startNextChkpt(Signal* signal)
       nodePtr.i = replicaPtr.p->procNode;
       ptrCheckGuard(nodePtr, MAX_NDB_NODES, nodeRecord);
       
+      if (c_lcpState.m_participatingLQH.get(nodePtr.i))
+      {
 	if (replicaPtr.p->lcpOngoingFlag &&
-          replicaPtr.p->lcpIdStarted < lcpId) {
+	    replicaPtr.p->lcpIdStarted < lcpId) 
+	{
 	  jam();
 	  //-------------------------------------------------------------------
 	  // We have found a replica on a node that performs local checkpoint
 	  // that is alive and that have not yet been started.
 	  //-------------------------------------------------------------------
 	  
-        if (nodePtr.p->noOfStartedChkpt < 2) {
+	  if (nodePtr.p->noOfStartedChkpt < 2) 
+	  {
 	    jam();
 	    /**
 	     * Send LCP_FRAG_ORD to LQH
@@ -10041,7 +10045,9 @@ void Dbdih::startNextChkpt(Signal* signal)
 	    nodePtr.p->noOfStartedChkpt = i + 1;
 	    
 	    sendLCP_FRAG_ORD(signal, nodePtr.p->startedChkpt[i]);
-        } else if (nodePtr.p->noOfQueuedChkpt < 2) {
+	  } 
+	  else if (nodePtr.p->noOfQueuedChkpt < 2) 
+	  {
 	    jam();
 	    /**
 	     * Put LCP_FRAG_ORD "in queue"
@@ -10057,10 +10063,13 @@ void Dbdih::startNextChkpt(Signal* signal)
 	    nodePtr.p->queuedChkpt[i].fragId = curr.fragmentId;
 	    nodePtr.p->queuedChkpt[i].replicaPtr = replicaPtr.i;
 	    nodePtr.p->noOfQueuedChkpt = i + 1;
-        } else {
+	  } 
+	  else 
+	  {
 	    jam();
 	    
-	  if(save){
+	    if(save)
+	    {
 	      /**
 	       * Stop increasing value on first that was "full"
 	       */
@@ -10069,7 +10078,8 @@ void Dbdih::startNextChkpt(Signal* signal)
 	    }
 	    
 	    busyNodes.set(nodePtr.i);
-	  if(busyNodes.count() == lcpNodes){
+	    if(busyNodes.count() == lcpNodes)
+	    {
 	      /**
 	       * There were no possibility to start the local checkpoint 
 	       * and it was not possible to queue it up. In this case we 
@@ -10082,6 +10092,7 @@ void Dbdih::startNextChkpt(Signal* signal)
 	  }//if
 	}
       }//while
+    }
     curr.fragmentId++;
     if (curr.fragmentId >= tabPtr.p->totalfragments) {
       jam();

storage/ndb/src/ndbapi/NdbScanOperation.cpp

@@ -694,9 +694,27 @@ void NdbScanOperation::close(bool forceSend, bool releaseOp)
   theNdbCon = NULL;
   m_transConnection = NULL;
 
-  if (releaseOp && tTransCon) {
+  if (tTransCon) 
+  {
     NdbIndexScanOperation* tOp = (NdbIndexScanOperation*)this;
-    tTransCon->releaseScanOperation(tOp);
+
+    bool ret = true;
+    if (theStatus != WaitResponse)
+    {
+      /**
+       * Not executed yet
+       */
+      ret = 
+	tTransCon->releaseScanOperation(&tTransCon->m_theFirstScanOperation,
+					&tTransCon->m_theLastScanOperation,
+					tOp);
+    }
+    else if (releaseOp)
+    {
+      ret = tTransCon->releaseScanOperation(&tTransCon->m_firstExecutedScanOp,
+					    0, tOp);
+    }
+    assert(ret);
   }
   
   tCon->theScanningOp = 0;

storage/ndb/src/ndbapi/NdbTransaction.cpp

@@ -979,50 +979,59 @@ Remark:         Release scan op when hupp'ed trans closed (save memory)
 void 
 NdbTransaction::releaseScanOperation(NdbIndexScanOperation* cursorOp)
 {
-  DBUG_ENTER("NdbTransaction::releaseScanOperation");
+  DBUG_ENTER("NdbTransaction::releaseExecutedScanOperation");
   DBUG_PRINT("enter", ("this=0x%x op=0x%x", (UintPtr)this, (UintPtr)cursorOp));
   
-  // here is one reason to make op lists doubly linked
-  if (cursorOp->m_executed)
+  releaseScanOperation(&m_firstExecutedScanOp, 0, cursorOp);
+  
+  DBUG_VOID_RETURN;
+}//NdbTransaction::releaseExecutedScanOperation()
+
+bool
+NdbTransaction::releaseScanOperation(NdbIndexScanOperation** listhead,
+				     NdbIndexScanOperation** listtail,
+				     NdbIndexScanOperation* op)
+{
+  if (* listhead == op)
   {
-    if (m_firstExecutedScanOp == cursorOp) {
-      m_firstExecutedScanOp = (NdbIndexScanOperation*)cursorOp->theNext;
-      cursorOp->release();
-      theNdb->releaseScanOperation(cursorOp);
-    } else if (m_firstExecutedScanOp != NULL) {
-      NdbIndexScanOperation* tOp = m_firstExecutedScanOp;
-      while (tOp->theNext != NULL) {
-        if (tOp->theNext == cursorOp) {
-          tOp->theNext = cursorOp->theNext;
-          cursorOp->release();
-          theNdb->releaseScanOperation(cursorOp);
-          break;
-        }
-        tOp = (NdbIndexScanOperation*)tOp->theNext;
-      }
+    * listhead = (NdbIndexScanOperation*)op->theNext;
+    if (listtail && *listtail == op)
+    {
+      assert(* listhead == 0);
+      * listtail = 0;
     }
+      
   }
   else
   {
-    if (m_theFirstScanOperation == cursorOp) {
-      m_theFirstScanOperation = (NdbIndexScanOperation*)cursorOp->theNext;
-      cursorOp->release();
-      theNdb->releaseScanOperation(cursorOp);
-    } else if (m_theFirstScanOperation != NULL) {
-      NdbIndexScanOperation* tOp = m_theFirstScanOperation;
-      while (tOp->theNext != NULL) {
-        if (tOp->theNext == cursorOp) {
-          tOp->theNext = cursorOp->theNext;
-          cursorOp->release();
-          theNdb->releaseScanOperation(cursorOp);
+    NdbIndexScanOperation* tmp = * listhead;
+    while (tmp != NULL)
+    {
+      if (tmp->theNext == op)
+      {
+	tmp->theNext = (NdbIndexScanOperation*)op->theNext;
+	if (listtail && *listtail == op)
+	{
+	  assert(op->theNext == 0);
+	  *listtail = tmp;
+	}
 	break;
       }
-        tOp = (NdbIndexScanOperation*)tOp->theNext;
+      tmp = (NdbIndexScanOperation*)tmp->theNext;
     }
+    if (tmp == NULL)
+      op = NULL;
   }
+  
+  if (op != NULL)
+  {
+    op->release();
+    theNdb->releaseScanOperation(op);
+    return true;
   }
-  DBUG_VOID_RETURN;
-}//NdbTransaction::releaseScanOperation()
+  
+  return false;
+}
 
 /*****************************************************************************
 NdbOperation* getNdbOperation(const char* aTableName);