wl1671 - bug fixes

ndb/include/ndbapi/NdbConnection.hpp:
  Made releaseCompletedOpertions private and introduced
  NdbConnection::restart instead. This methods basically check a bunch of 
  status flags before calling releaseCompletedOperations
ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  Introduced MARKRE_TRACE
  Fixed bug on not releasing marker at execABORT
ndb/src/ndbapi/NdbConnection.cpp:
  Made releaseCompletedOpertions private and introduced
  NdbConnection::restart instead. This methods basically check a bunch of 
  status flags before calling releaseCompletedOperations
ndb/src/ndbapi/Ndbif.cpp:
  Use MARKER_TRACE
ndb/test/src/HugoTransactions.cpp:
  Use restart instead of releaseCompletedOperations
ndb/test/src/UtilTransactions.cpp:
  Use restart instead of releaseCompletedOperations
ndb/tools/delete_all.cpp:
  Use restart instead of releaseCompletedOperations

ndb/include/ndbapi/NdbConnection.hpp

@@ -309,6 +309,16 @@ public:
    */
   void close();
 
+  /**
+   * Restart transaction
+   *
+   *   Once a transaction has been completed successfully
+   *     it can be started again wo/ calling closeTransaction/startTransaction
+   *
+   *   Note this method also releases completed operations
+   */
+  int restart();
+
   /** @} *********************************************************************/
 
   /** 
@@ -417,16 +427,14 @@ public:
    */
   const NdbOperation * getNextCompletedOperation(const NdbOperation * op)const;
 
+  /** @} *********************************************************************/
+
+private:						
   /**
    * Release completed operations
    */
   void releaseCompletedOperations();
 
-
-  /** @} *********************************************************************/
-
-private:						
-
   typedef Uint64 TimeMillis_t;
   /**************************************************************************
    *	These methods are service methods to other classes in the NDBAPI.   *

ndb/src/kernel/blocks/dblqh/DblqhMain.cpp

@@ -64,6 +64,8 @@
 #define DEBUG(x)
 #endif
 
+//#define MARKER_TRACE 1
+
 const Uint32 NR_ScanNo = 0;
 
 void Dblqh::execACC_COM_BLOCK(Signal* signal)
@@ -2404,6 +2406,9 @@ Dblqh::execREMOVE_MARKER_ORD(Signal* signal)
   CommitAckMarkerPtr removedPtr;
   m_commitAckMarkerHash.release(removedPtr, key);
   ndbrequire(removedPtr.i != RNIL);
+#ifdef MARKER_TRACE
+  ndbout_c("Rem marker[%.8x %.8x]", key.transid1, key.transid2);
+#endif
 }
 
 
@@ -3198,6 +3203,9 @@ void Dblqh::execLQHKEYREQ(Signal* signal)
     
     CommitAckMarkerPtr tmp;
 #ifdef VM_TRACE
+#ifdef MARKER_TRACE
+    ndbout_c("Add marker[%.8x %.8x]", markerPtr.p->transid1, markerPtr.p->transid2);
+#endif
     ndbrequire(!m_commitAckMarkerHash.find(tmp, * markerPtr.p));
 #endif
     m_commitAckMarkerHash.add(markerPtr);
@@ -5668,7 +5676,23 @@ void Dblqh::execABORT(Signal* signal)
   }//if
   regTcPtr->abortState = TcConnectionrec::ABORT_FROM_TC;
   regTcPtr->activeCreat = ZFALSE;
+
+  const Uint32 commitAckMarker = regTcPtr->commitAckMarker;
+  if(commitAckMarker != RNIL){
+    jam();
+#ifdef MARKER_TRACE
+    {
+      CommitAckMarkerPtr tmp;
+      m_commitAckMarkerHash.getPtr(tmp, commitAckMarker);
+      ndbout_c("Ab2 marker[%.8x %.8x]", tmp.p->transid1, tmp.p->transid2);
+    }
+#endif
+    m_commitAckMarkerHash.release(commitAckMarker);
+    regTcPtr->commitAckMarker = RNIL;
+  }
+
   abortStateHandlerLab(signal);
+
   return;
 }//Dblqh::execABORT()
 
@@ -6026,7 +6050,13 @@ void Dblqh::abortCommonLab(Signal* signal)
      * There is no NR ongoing and we have a marker
      */
     jam();
-    
+#ifdef MARKER_TRACE
+    {
+      CommitAckMarkerPtr tmp;
+      m_commitAckMarkerHash.getPtr(tmp, commitAckMarker);
+      ndbout_c("Abo marker[%.8x %.8x]", tmp.p->transid1, tmp.p->transid2);
+    }
+#endif
     m_commitAckMarkerHash.release(commitAckMarker);
     regTcPtr->commitAckMarker = RNIL;
   }

ndb/src/ndbapi/NdbConnection.cpp

@@ -193,6 +193,23 @@ NdbConnection::setErrorCode(int anErrorCode)
     theError.code = anErrorCode;
 }//NdbConnection::setErrorCode()
 
+int
+NdbConnection::restart(){
+  if(theCompletionStatus == CompletedSuccess){
+    releaseCompletedOperations();
+    Uint64 tTransid = theNdb->theFirstTransId;
+    theTransactionId = tTransid;
+    if((Uint32)tTransid == ((Uint32)~0)){
+      theNdb->theFirstTransId = (tTransid >> 32) << 32;
+    } else {
+      theNdb->theFirstTransId = tTransid + 1;
+    }
+    theCompletionStatus = NotCompleted;
+    return 0;
+  }
+  return -1;
+}
+
 /*****************************************************************************
 void handleExecuteCompletion(void);
 
@@ -1307,7 +1324,6 @@ from other transactions.
     if (tCommitFlag == 1) {
       theCommitStatus = Committed;
       theGlobalCheckpointId = tGCI;
-      theTransactionId++;
     } else if ((tNoComp >= tNoSent) &&
                (theLastExecOpInList->theCommitIndicator == 1)){
 /**********************************************************************/

ndb/src/ndbapi/Ndbif.cpp

@@ -1308,8 +1308,8 @@ void
 NdbConnection::sendTC_COMMIT_ACK(NdbApiSignal * aSignal,
 				 Uint32 transId1, Uint32 transId2, 
 				 Uint32 aTCRef){
-#if 0
-  ndbout_c("Sending TC_COMMIT_ACK(0x%x, 0x%x) to -> %d",
+#ifdef MARKER_TRACE
+  ndbout_c("Sending TC_COMMIT_ACK(0x%.8x, 0x%.8x) to -> %d",
 	   transId1,
 	   transId2,
 	   refToNode(aTCRef));

ndb/test/src/HugoTransactions.cpp

@@ -654,7 +654,7 @@ HugoTransactions::scanUpdateRecords3(Ndb* pNdb,
 
       if(check != -1){
 	check = pTrans->execute(Commit);   
-	pTrans->releaseCompletedOperations();
+	pTrans->restart();
       }
 
       const NdbError err = pTrans->getNdbError();    

ndb/test/src/UtilTransactions.cpp

@@ -412,7 +412,7 @@ UtilTransactions::clearTable3(Ndb* pNdb,
       
       if(check != -1){
 	check = pTrans->execute(Commit);   
-	pTrans->releaseCompletedOperations();
+	pTrans->restart();
       }
       
       err = pTrans->getNdbError();    
@@ -536,7 +536,7 @@ UtilTransactions::copyTableData(Ndb* pNdb,
       } while((eof = rs->nextResult(false)) == 0);
       
       check = pTrans->execute(Commit);   
-      pTrans->releaseCompletedOperations();
+      pTrans->restart();
       if( check == -1 ) {
 	const NdbError err = pTrans->getNdbError();    
 	ERR(err);

ndb/tools/delete_all.cpp

@@ -143,7 +143,7 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab, int parallelism)
       
       if(check != -1){
 	check = pTrans->execute(Commit);   
-	pTrans->releaseCompletedOperations();
+	pTrans->restart();
       }
       
       err = pTrans->getNdbError();