---
 storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp |   33 +
 storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp       |   22 -
 storage/ndb/src/kernel/blocks/dbtc/DbtcInit.cpp   |    3 
 storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp   |  446 ++++++++++++++--------
 4 files changed, 343 insertions(+), 161 deletions(-)

Index: telco-6.2/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
===================================================================
--- telco-6.2.orig/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp	2008-07-02 12:37:23.000000000 +0200
+++ telco-6.2/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp	2008-07-04 09:03:48.000000000 +0200
@@ -86,6 +86,8 @@
   
 #define INTERNAL_TRIGGER_TCKEYREQ_JBA 0
 
+int level = 1;
+
 #ifdef VM_TRACE
 NdbOut &
 operator<<(NdbOut& out, Dbtc::ConnectionState state){
@@ -4397,22 +4399,21 @@ void Dbtc::commitGciHandling(Signal* sig
 
   UintR TgcpFilesize = cgcpFilesize;
   UintR Tfirstgcp = cfirstgcp;
-  ApiConnectRecord * const regApiPtr = apiConnectptr.p;
+  Ptr<ApiConnectRecord> regApiPtr = apiConnectptr;
   GcpRecord *localGcpRecord = gcpRecord;
 
-  regApiPtr->globalcheckpointid = Tgci;
+  regApiPtr.p->globalcheckpointid = Tgci;
   if (Tfirstgcp != RNIL) {
     /* IF THIS GLOBAL CHECKPOINT ALREADY EXISTS */
     localGcpPointer.i = Tfirstgcp;
     ptrCheckGuard(localGcpPointer, TgcpFilesize, localGcpRecord);
     do {
-      if (regApiPtr->globalcheckpointid == localGcpPointer.p->gcpId) {
+      if (regApiPtr.p->globalcheckpointid == localGcpPointer.p->gcpId) {
         jam();
-        gcpPtr.i = localGcpPointer.i;
-        gcpPtr.p = localGcpPointer.p;
-        linkApiToGcp(signal);
+        linkApiToGcp(localGcpPointer, regApiPtr);
         return;
       } else {
+        ndbrequire(regApiPtr.p->globalcheckpointid > localGcpPointer.p->gcpId);
         localGcpPointer.i = localGcpPointer.p->nextGcp;
         jam();
         if (localGcpPointer.i != RNIL) {
@@ -4421,14 +4422,14 @@ void Dbtc::commitGciHandling(Signal* sig
           continue;
         }//if
       }//if
-      seizeGcp(signal);
-      linkApiToGcp(signal);
+      seizeGcp(localGcpPointer, Tgci);
+      linkApiToGcp(localGcpPointer, regApiPtr);
       return;
     } while (1);
   } else {
     jam();
-    seizeGcp(signal);
-    linkApiToGcp(signal);
+    seizeGcp(localGcpPointer, Tgci);
+    linkApiToGcp(localGcpPointer, regApiPtr);
   }//if
 }//Dbtc::commitGciHandling()
 
@@ -4437,46 +4438,43 @@ void Dbtc::commitGciHandling(Signal* sig
 /* CHECKPOINTS. WHEN THE TRANSACTION I COMPLETED THE API CONNECT RECORD IS   */
 /* LINKED OUT OF THE LIST.                                                   */
 /*---------------------------------------------------------------------------*/
-void Dbtc::linkApiToGcp(Signal* signal) 
+void Dbtc::linkApiToGcp(Ptr<GcpRecord> regGcpPtr,
+                        Ptr<ApiConnectRecord> regApiPtr)
 {
   ApiConnectRecordPtr localApiConnectptr;
-  ApiConnectRecord * const regApiPtr = apiConnectptr.p;
-  GcpRecord * const regGcpPtr = gcpPtr.p;
-  UintR TapiConnectptrIndex = apiConnectptr.i;
   ApiConnectRecord *localApiConnectRecord = apiConnectRecord;
 
-  regApiPtr->nextGcpConnect = RNIL;
-  if (regGcpPtr->firstApiConnect == RNIL) {
-    regGcpPtr->firstApiConnect = TapiConnectptrIndex;
+  regApiPtr.p->nextGcpConnect = RNIL;
+  if (regGcpPtr.p->firstApiConnect == RNIL) {
+    regGcpPtr.p->firstApiConnect = regApiPtr.i;
     jam();
   } else {
     UintR TapiConnectFilesize = capiConnectFilesize;
-    localApiConnectptr.i = regGcpPtr->lastApiConnect;
+    localApiConnectptr.i = regGcpPtr.p->lastApiConnect;
     jam();
     ptrCheckGuard(localApiConnectptr,
                   TapiConnectFilesize, localApiConnectRecord);
-    localApiConnectptr.p->nextGcpConnect = TapiConnectptrIndex;
+    localApiConnectptr.p->nextGcpConnect = regApiPtr.i;
   }//if
-  UintR TlastApiConnect = regGcpPtr->lastApiConnect;
-  regApiPtr->gcpPointer = gcpPtr.i;
-  regApiPtr->prevGcpConnect = TlastApiConnect;
-  regGcpPtr->lastApiConnect = TapiConnectptrIndex;
+  UintR TlastApiConnect = regGcpPtr.p->lastApiConnect;
+  regApiPtr.p->gcpPointer = regGcpPtr.i;
+  regApiPtr.p->prevGcpConnect = TlastApiConnect;
+  regGcpPtr.p->lastApiConnect = regApiPtr.i;
 }//Dbtc::linkApiToGcp()
 
-void Dbtc::seizeGcp(Signal* signal) 
+void Dbtc::seizeGcp(Ptr<GcpRecord> & dst, Uint64 Tgci)
 {
   GcpRecordPtr tmpGcpPointer;
   GcpRecordPtr localGcpPointer;
 
   UintR Tfirstgcp = cfirstgcp;
-  Uint64 Tglobalcheckpointid = apiConnectptr.p->globalcheckpointid;
   UintR TgcpFilesize = cgcpFilesize;
   GcpRecord *localGcpRecord = gcpRecord;
 
   localGcpPointer.i = cfirstfreeGcp;
   ptrCheckGuard(localGcpPointer, TgcpFilesize, localGcpRecord);
   UintR TfirstfreeGcp = localGcpPointer.p->nextGcp;
-  localGcpPointer.p->gcpId = Tglobalcheckpointid;
+  localGcpPointer.p->gcpId = Tgci;
   localGcpPointer.p->nextGcp = RNIL;
   localGcpPointer.p->firstApiConnect = RNIL;
   localGcpPointer.p->lastApiConnect = RNIL;
@@ -4493,7 +4491,11 @@ void Dbtc::seizeGcp(Signal* signal) 
     tmpGcpPointer.p->nextGcp = localGcpPointer.i;
   }//if
   clastgcp = localGcpPointer.i;
-  gcpPtr = localGcpPointer;
+  dst = localGcpPointer;
+
+  ndbout_c("seizeGcp(%u/%u)",
+           Uint32(Tgci >> 32),
+           Uint32(Tgci));
 }//Dbtc::seizeGcp()
 
 /*---------------------------------------------------------------------------*/
@@ -4621,6 +4623,7 @@ void Dbtc::execCOMMITTED(Signal* signal)
 {
   TcConnectRecordPtr localTcConnectptr;
   ApiConnectRecordPtr localApiConnectptr;
+  ApiConnectRecordPtr localCopyPtr;
 
   UintR TtcConnectFilesize = ctcConnectFilesize;
   UintR TapiConnectFilesize = capiConnectFilesize;
@@ -4696,14 +4699,14 @@ void Dbtc::execCOMMITTED(Signal* signal)
   /*-------------------------------------------------------*/
 
   apiConnectptr = localApiConnectptr;
-  sendApiCommit(signal);
+  localCopyPtr = sendApiCommit(signal);
 
-  ApiConnectRecord * const regApiPtr = apiConnectptr.p;
-  localTcConnectptr.i = regApiPtr->firstTcConnect;
-  UintR Tlqhkeyconfrec = regApiPtr->lqhkeyconfrec;
+  localTcConnectptr.i = localCopyPtr.p->firstTcConnect;
+  UintR Tlqhkeyconfrec = localCopyPtr.p->lqhkeyconfrec;
   ptrCheckGuard(localTcConnectptr, TtcConnectFilesize, localTcConnectRecord);
-  regApiPtr->counter = Tlqhkeyconfrec;
+  localCopyPtr.p->counter = Tlqhkeyconfrec;
 
+  apiConnectptr = localCopyPtr;
   tcConnectptr = localTcConnectptr;
   complete010Lab(signal);
   return;
@@ -4714,9 +4717,10 @@ void Dbtc::execCOMMITTED(Signal* signal)
 /*                       SEND_API_COMMIT                 */
 /*       SEND COMMIT DECISION TO THE API.                */
 /*-------------------------------------------------------*/
-void Dbtc::sendApiCommit(Signal* signal) 
+Ptr<Dbtc::ApiConnectRecord>
+Dbtc::sendApiCommit(Signal* signal)
 {
-  ApiConnectRecord * const regApiPtr = apiConnectptr.p;
+  ApiConnectRecordPtr regApiPtr = apiConnectptr;
 
   if (ERROR_INSERTED(8055))
   {
@@ -4728,58 +4732,59 @@ void Dbtc::sendApiCommit(Signal* signal)
     signal->theData[0] = 9999;
     sendSignalWithDelay(CMVMI_REF, GSN_NDB_TAMPER, signal, 1000, 1);
     
-    Uint32 node = refToNode(regApiPtr->ndbapiBlockref);
+    Uint32 node = refToNode(regApiPtr.p->ndbapiBlockref);
     signal->theData[0] = node;
     sendSignal(QMGR_REF, GSN_API_FAILREQ, signal, 1, JBB);
 
     SET_ERROR_INSERT_VALUE(8056);
-    return;
+
+    Ptr<ApiConnectRecord> copyPtr;
+    copyPtr.i = regApiPtr.p->apiCopyRecord;
+    ptrCheckGuard(copyPtr, capiConnectFilesize, apiConnectRecord);
+    return copyPtr;
   }
   
-  if (regApiPtr->returnsignal == RS_TCKEYCONF) {
+  if (regApiPtr.p->returnsignal == RS_TCKEYCONF) {
     sendtckeyconf(signal, 1);
-  } else if (regApiPtr->returnsignal == RS_TC_COMMITCONF) {
+  } else if (regApiPtr.p->returnsignal == RS_TC_COMMITCONF) {
     jam();
     TcCommitConf * const commitConf = (TcCommitConf *)&signal->theData[0];
-    if(regApiPtr->commitAckMarker == RNIL){
+    if(regApiPtr.p->commitAckMarker == RNIL){
       jam();
-      commitConf->apiConnectPtr = regApiPtr->ndbapiConnect;
+      commitConf->apiConnectPtr = regApiPtr.p->ndbapiConnect;
     } else {
       jam();
-      commitConf->apiConnectPtr = regApiPtr->ndbapiConnect | 1;
+      commitConf->apiConnectPtr = regApiPtr.p->ndbapiConnect | 1;
     }
-    commitConf->transId1 = regApiPtr->transid[0];
-    commitConf->transId2 = regApiPtr->transid[1];
-    commitConf->gci_hi = Uint32(regApiPtr->globalcheckpointid >> 32);
-    commitConf->gci_lo = Uint32(regApiPtr->globalcheckpointid);
+    commitConf->transId1 = regApiPtr.p->transid[0];
+    commitConf->transId2 = regApiPtr.p->transid[1];
+    commitConf->gci_hi = Uint32(regApiPtr.p->globalcheckpointid >> 32);
+    commitConf->gci_lo = Uint32(regApiPtr.p->globalcheckpointid);
 
-    sendSignal(regApiPtr->ndbapiBlockref, GSN_TC_COMMITCONF, signal,
+    sendSignal(regApiPtr.p->ndbapiBlockref, GSN_TC_COMMITCONF, signal,
 	       TcCommitConf::SignalLength, JBB);
-  } else if (regApiPtr->returnsignal == RS_NO_RETURN) {
+  } else if (regApiPtr.p->returnsignal == RS_NO_RETURN) {
     jam();
   } else {
     TCKEY_abort(signal, 37);
-    return;
+    return regApiPtr;
   }//if
+  Ptr<ApiConnectRecord> copyPtr;
   UintR TapiConnectFilesize = capiConnectFilesize;
   UintR TcommitCount = c_counters.ccommitCount;
-  UintR TapiIndex = apiConnectptr.i;
-  UintR TnewApiIndex = regApiPtr->apiCopyRecord;
-  UintR TapiFailState = regApiPtr->apiFailState;
+  copyPtr.i = regApiPtr.p->apiCopyRecord;
+  UintR TapiFailState = regApiPtr.p->apiFailState;
   ApiConnectRecord *localApiConnectRecord = apiConnectRecord;
 
-  tmpApiConnectptr.p = apiConnectptr.p;
-  tmpApiConnectptr.i = TapiIndex;
   c_counters.ccommitCount = TcommitCount + 1;
-  apiConnectptr.i = TnewApiIndex;
-  ptrCheckGuard(apiConnectptr, TapiConnectFilesize, localApiConnectRecord);
-  copyApi(signal);
+  ptrCheckGuard(copyPtr, TapiConnectFilesize, localApiConnectRecord);
+  copyApi(copyPtr, regApiPtr);
   if (TapiFailState != ZTRUE) {
-    return;
+    return copyPtr;
   } else {
     jam();
-    handleApiFailState(signal, tmpApiConnectptr.i);
-    return;
+    handleApiFailState(signal, regApiPtr.i);
+    return copyPtr;
   }//if
 }//Dbtc::sendApiCommit()
 
@@ -4788,54 +4793,52 @@ void Dbtc::sendApiCommit(Signal* signal)
 /*   COPY API RECORD ALSO RESET THE OLD API RECORD SO THAT IT                */
 /*   IS PREPARED TO RECEIVE A NEW TRANSACTION.                               */
 /*===========================================================================*/
-void Dbtc::copyApi(Signal* signal) 
+void Dbtc::copyApi(ApiConnectRecordPtr copyPtr, ApiConnectRecordPtr regApiPtr)
 {
-  ApiConnectRecord * const regApiPtr = apiConnectptr.p;
-  ApiConnectRecord * const regTmpApiPtr = tmpApiConnectptr.p;
-
-  UintR TndbapiConnect = regTmpApiPtr->ndbapiConnect;
-  UintR TfirstTcConnect = regTmpApiPtr->firstTcConnect;
-  UintR Ttransid1 = regTmpApiPtr->transid[0];
-  UintR Ttransid2 = regTmpApiPtr->transid[1];
-  UintR Tlqhkeyconfrec = regTmpApiPtr->lqhkeyconfrec;
-  UintR TgcpPointer = regTmpApiPtr->gcpPointer;
+  UintR TndbapiConnect = regApiPtr.p->ndbapiConnect;
+  UintR TfirstTcConnect = regApiPtr.p->firstTcConnect;
+  UintR Ttransid1 = regApiPtr.p->transid[0];
+  UintR Ttransid2 = regApiPtr.p->transid[1];
+  UintR Tlqhkeyconfrec = regApiPtr.p->lqhkeyconfrec;
+  UintR TgcpPointer = regApiPtr.p->gcpPointer;
   UintR TgcpFilesize = cgcpFilesize;
-  UintR TcommitAckMarker = regTmpApiPtr->commitAckMarker;
-  NdbNodeBitmask Tnodes = regTmpApiPtr->m_transaction_nodes;
+  UintR TcommitAckMarker = regApiPtr.p->commitAckMarker;
+  NdbNodeBitmask Tnodes = regApiPtr.p->m_transaction_nodes;
   GcpRecord *localGcpRecord = gcpRecord;
 
-  regApiPtr->ndbapiBlockref = regTmpApiPtr->ndbapiBlockref;
-  regApiPtr->ndbapiConnect = TndbapiConnect;
-  regApiPtr->firstTcConnect = TfirstTcConnect;
-  regApiPtr->apiConnectstate = CS_COMPLETING;
-  regApiPtr->transid[0] = Ttransid1;
-  regApiPtr->transid[1] = Ttransid2;
-  regApiPtr->lqhkeyconfrec = Tlqhkeyconfrec;
-  regApiPtr->commitAckMarker = TcommitAckMarker;
-  regApiPtr->m_transaction_nodes = Tnodes;
-  regApiPtr->singleUserMode = 0;
+  copyPtr.p->ndbapiBlockref = regApiPtr.p->ndbapiBlockref;
+  copyPtr.p->ndbapiConnect = TndbapiConnect;
+  copyPtr.p->firstTcConnect = TfirstTcConnect;
+  copyPtr.p->apiConnectstate = CS_COMPLETING;
+  copyPtr.p->transid[0] = Ttransid1;
+  copyPtr.p->transid[1] = Ttransid2;
+  copyPtr.p->lqhkeyconfrec = Tlqhkeyconfrec;
+  copyPtr.p->commitAckMarker = TcommitAckMarker;
+  copyPtr.p->m_transaction_nodes = Tnodes;
+  copyPtr.p->singleUserMode = 0;
 
+  Ptr<GcpRecord> gcpPtr;
   gcpPtr.i = TgcpPointer;
   ptrCheckGuard(gcpPtr, TgcpFilesize, localGcpRecord);
-  unlinkApiConnect(signal);
-  linkApiToGcp(signal);
-  setApiConTimer(tmpApiConnectptr.i, 0, __LINE__);
-  regTmpApiPtr->apiConnectstate = CS_CONNECTED;
-  regTmpApiPtr->commitAckMarker = RNIL;
-  regTmpApiPtr->firstTcConnect = RNIL;
-  regTmpApiPtr->lastTcConnect = RNIL;
-  regTmpApiPtr->m_transaction_nodes.clear();
-  regTmpApiPtr->singleUserMode = 0;
-  releaseAllSeizedIndexOperations(regTmpApiPtr);
+  unlinkApiConnect(gcpPtr, regApiPtr);
+  linkApiToGcp(gcpPtr, copyPtr);
+  setApiConTimer(regApiPtr.i, 0, __LINE__);
+  regApiPtr.p->apiConnectstate = CS_CONNECTED;
+  regApiPtr.p->commitAckMarker = RNIL;
+  regApiPtr.p->firstTcConnect = RNIL;
+  regApiPtr.p->lastTcConnect = RNIL;
+  regApiPtr.p->m_transaction_nodes.clear();
+  regApiPtr.p->singleUserMode = 0;
+  releaseAllSeizedIndexOperations(regApiPtr.p);
 }//Dbtc::copyApi()
 
-void Dbtc::unlinkApiConnect(Signal* signal) 
+void Dbtc::unlinkApiConnect(Ptr<GcpRecord> gcpPtr,
+                            Ptr<ApiConnectRecord> regApiPtr)
 {
   ApiConnectRecordPtr localApiConnectptr;
-  ApiConnectRecord * const regTmpApiPtr = tmpApiConnectptr.p;
   UintR TapiConnectFilesize = capiConnectFilesize;
-  UintR TprevGcpConnect = regTmpApiPtr->prevGcpConnect;
-  UintR TnextGcpConnect = regTmpApiPtr->nextGcpConnect;
+  UintR TprevGcpConnect = regApiPtr.p->prevGcpConnect;
+  UintR TnextGcpConnect = regApiPtr.p->nextGcpConnect;
   ApiConnectRecord *localApiConnectRecord = apiConnectRecord;
 
   if (TprevGcpConnect == RNIL) {
@@ -5096,7 +5099,7 @@ void Dbtc::releaseTransResources(Signal*
     localTcConnectptr.i = rtrTcConnectptrIndex;
     releaseTcCon();
   } while (localTcConnectptr.i != RNIL);
-  handleGcp(signal);
+  handleGcp(signal, apiConnectptr);
   releaseFiredTriggerData(&apiConnectptr.p->theFiredTriggers);
   releaseAllSeizedIndexOperations(apiConnectptr.p);
   releaseApiConCopy(signal);
@@ -5109,27 +5112,30 @@ void Dbtc::releaseTransResources(Signal*
 /*       SENDS GCP_TCFINISHED WHEN ALL TRANSACTIONS BELONGING TO A CERTAIN */
 /*       GLOBAL CHECKPOINT HAVE COMPLETED.                                 */
 /* *********************************************************************>> */
-void Dbtc::handleGcp(Signal* signal) 
+void Dbtc::handleGcp(Signal* signal, Ptr<ApiConnectRecord> regApiPtr)
 {
   GcpRecord *localGcpRecord = gcpRecord;
   GcpRecordPtr localGcpPtr;
-  UintR TapiConnectptrIndex = apiConnectptr.i;
   UintR TgcpFilesize = cgcpFilesize;
   localGcpPtr.i = apiConnectptr.p->gcpPointer;
-  tmpApiConnectptr.i = TapiConnectptrIndex;
-  tmpApiConnectptr.p = apiConnectptr.p;
   ptrCheckGuard(localGcpPtr, TgcpFilesize, localGcpRecord);
-  gcpPtr.i = localGcpPtr.i;
-  gcpPtr.p = localGcpPtr.p;
-  unlinkApiConnect(signal);
+  unlinkApiConnect(localGcpPtr, regApiPtr);
   if (localGcpPtr.p->firstApiConnect == RNIL) {
     if (localGcpPtr.p->gcpNomoretransRec == ZTRUE) {
-      jam();
-      tcheckGcpId = localGcpPtr.p->gcpId;
-      gcpTcfinished(signal);
-      unlinkGcp(signal);
+      if (c_ongoing_take_over_cnt == 0)
+      {
+        jam();
+        gcpTcfinished(signal, localGcpPtr.p->gcpId);
+        unlinkGcp(localGcpPtr);
+      }
+      else
+      {
+        ndbout_c("handleGcp NOT returning gcpTcfinished nfhandling %u/%u",
+                 Uint32(localGcpPtr.p->gcpId>> 32),
+                 Uint32(localGcpPtr.p->gcpId));
+      }
     }//if
-  }//if
+  }
 }//Dbtc::handleGcp()
 
 void Dbtc::releaseApiConCopy(Signal* signal) 
@@ -7124,35 +7130,149 @@ void Dbtc::execGCP_NOMORETRANS(Signal* s
   Uint32 gci_lo = req->gci_lo;
   Uint32 gci_hi = req->gci_hi;
   tcheckGcpId = gci_lo | (Uint64(gci_hi) << 32);
+
+  if (level >= 2)
+    ndbout_c("GCP_NOMORETRANS(%u/%u)", gci_hi, gci_lo);
+
+  Ptr<GcpRecord> gcpPtr;
   if (cfirstgcp != RNIL) {
     jam();
                                       /* A GLOBAL CHECKPOINT IS GOING ON */
     gcpPtr.i = cfirstgcp;             /* SET POINTER TO FIRST GCP IN QUEUE*/
     ptrCheckGuard(gcpPtr, cgcpFilesize, gcpRecord);
-    if (gcpPtr.p->gcpId == tcheckGcpId) {
+    if (gcpPtr.p->gcpId == tcheckGcpId)
+    {
       jam();
-      if (gcpPtr.p->firstApiConnect != RNIL) {
+      bool empty = gcpPtr.p->firstApiConnect == RNIL;
+      bool nfhandling = c_ongoing_take_over_cnt;
+
+      if (empty && nfhandling)
+      {
+        ndbout_c("NOT returning gcpTcfinished due to nfhandling %u/%u",
+                 gci_hi, gci_lo);
+      }
+
+      if (!empty || c_ongoing_take_over_cnt)
+      {
         jam();
         gcpPtr.p->gcpNomoretransRec = ZTRUE;
       } else {
         jam();
-        gcpTcfinished(signal);
-        unlinkGcp(signal);
+        gcpTcfinished(signal, tcheckGcpId);
+        unlinkGcp(gcpPtr);
       }//if
-    } else {
+    }
+    else if (c_ongoing_take_over_cnt == 0)
+    {
       jam();
       /*------------------------------------------------------------*/
       /*       IF IT IS NOT THE FIRST THEN THERE SHOULD BE NO       */
       /*       RECORD FOR THIS GLOBAL CHECKPOINT. WE ALWAYS REMOVE  */
       /*       THE GLOBAL CHECKPOINTS IN ORDER.                     */
       /*------------------------------------------------------------*/
-      gcpTcfinished(signal);
-    }//if
-  } else {
+
+      int save = level;
+      level = 100;
+      printf("execGCP_NOMORETRANS %u/%u != %u/%u -> ",
+             Uint32(gcpPtr.p->gcpId >> 32),
+             Uint32(gcpPtr.p->gcpId),
+             gci_hi, gci_lo);
+      gcpTcfinished(signal, tcheckGcpId);
+      level = save;
+    }
+    else
+    {
+      jam();
+      goto outoforder;
+    }
+  }
+  else if (c_ongoing_take_over_cnt == 0)
+  {
     jam();
-    gcpTcfinished(signal);
-  }//if
+    int save = level;
+    level = 100;
+    printf("cfirstgcp == RNIL ");
+    gcpTcfinished(signal, tcheckGcpId);
+    level = save;
+  }
+  else
+  {
+seize:
+    jam();
+    printf("execGCP_NOMORETRANS c_ongoing_take_over_cnt-> ");
+    seizeGcp(gcpPtr, tcheckGcpId);
+    gcpPtr.p->gcpNomoretransRec = ZTRUE;
+  }
   return;
+
+outoforder:
+  printf("ooo: execGCP_NOMORETRANS tcheckGcpId: %u/%u cfirstgcp: %u/%u",
+         gci_hi, gci_lo,
+         Uint32(gcpPtr.p->gcpId >> 32), Uint32(gcpPtr.p->gcpId));
+
+  if (tcheckGcpId < gcpPtr.p->gcpId)
+  {
+    jam();
+
+    Ptr<GcpRecord> tmp;
+    tmp.i = cfirstfreeGcp;
+    ptrCheckGuard(tmp, cgcpFilesize, gcpRecord);
+    cfirstfreeGcp = tmp.p->nextGcp;
+
+    tmp.p->gcpId = tcheckGcpId;
+    tmp.p->nextGcp = cfirstgcp;
+    tmp.p->firstApiConnect = RNIL;
+    tmp.p->lastApiConnect = RNIL;
+    tmp.p->gcpNomoretransRec = ZTRUE;
+    cfirstgcp = tmp.i;
+    ndbout_c("LINK FIRST");
+    return;
+  }
+  else
+  {
+    Ptr<GcpRecord> prev = gcpPtr;
+    while (tcheckGcpId > gcpPtr.p->gcpId)
+    {
+      jam();
+      if (gcpPtr.p->nextGcp == RNIL)
+      {
+        printf("nextGcp == RNIL -> ");
+        goto seize;
+      }
+
+      prev = gcpPtr;
+      gcpPtr.i = gcpPtr.p->nextGcp;
+      ptrCheckGuard(gcpPtr, cgcpFilesize, gcpRecord);
+    }
+
+    if (tcheckGcpId == gcpPtr.p->gcpId)
+    {
+      jam();
+      gcpPtr.p->gcpNomoretransRec = ZTRUE;
+      ndbout_c("found");
+      return;
+    }
+    ndbrequire(prev.i != gcpPtr.i); // checked earlier with initial "<"
+    ndbrequire(prev.p->gcpId < tcheckGcpId);
+    ndbrequire(gcpPtr.p->gcpId > tcheckGcpId);
+
+    Ptr<GcpRecord> tmp;
+    tmp.i = cfirstfreeGcp;
+    ptrCheckGuard(tmp, cgcpFilesize, gcpRecord);
+    cfirstfreeGcp = tmp.p->nextGcp;
+
+    tmp.p->gcpId = tcheckGcpId;
+    tmp.p->nextGcp = gcpPtr.i;
+    tmp.p->firstApiConnect = RNIL;
+    tmp.p->lastApiConnect = RNIL;
+    tmp.p->gcpNomoretransRec = ZTRUE;
+    prev.p->nextGcp = tmp.i;
+    ndbout_c("link middle %u/%u < %u/%u < %u/%u",
+             Uint32(prev.p->gcpId >> 32), Uint32(prev.p->gcpId),
+             gci_hi, gci_lo,
+             Uint32(gcpPtr.p->gcpId >> 32), Uint32(gcpPtr.p->gcpId));
+    return;
+  }
 }//Dbtc::execGCP_NOMORETRANS()
 
 /*****************************************************************************/
@@ -7210,6 +7330,7 @@ void Dbtc::execNODE_FAILREP(Signal* sign
     /*------------------------------------------------------------*/
     hostptr.p->hostStatus = HS_DEAD;
     hostptr.p->m_nf_bits = HostRecord::NF_NODE_FAIL_BITS;
+    c_ongoing_take_over_cnt++;
     c_alive_nodes.clear(hostptr.i);
 
     if (tcNodeFailptr.p->failStatus == FS_LISTENING) 
@@ -7413,8 +7534,31 @@ void Dbtc::execTAKE_OVERTCCONF(Signal* s
     tcNodeFailptr.p->queueList[i] = tcNodeFailptr.p->queueList[end-1];
     tcNodeFailptr.p->queueIndex = end - 1;
   }
-  
+
+  Uint32 cnt = c_ongoing_take_over_cnt;
+  ndbrequire(cnt);
+  c_ongoing_take_over_cnt = cnt - 1;
   checkNodeFailComplete(signal, hostptr.i, HostRecord::NF_TAKEOVER);
+
+  if (cnt == 1 && cfirstgcp != RNIL)
+  {
+    /**
+     * Check if there are any hanging GCP_NOMORETRANS
+     */
+    GcpRecordPtr tmpGcpPointer;
+    tmpGcpPointer.i = cfirstgcp;
+    ptrCheckGuard(tmpGcpPointer, cgcpFilesize, gcpRecord);
+    if (tmpGcpPointer.p->gcpNomoretransRec &&
+        tmpGcpPointer.p->firstApiConnect == RNIL)
+    {
+      jam();
+      ndbout_c("completing gcp %u/%u in execTAKE_OVERTCCONF",
+               Uint32(tmpGcpPointer.p->gcpId >> 32),
+               Uint32(tmpGcpPointer.p->gcpId));
+      gcpTcfinished(signal, tmpGcpPointer.p->gcpId);
+      unlinkGcp(tmpGcpPointer);
+    }
+  }
 }//Dbtc::execTAKE_OVERTCCONF()
 
 void Dbtc::execTAKE_OVERTCREQ(Signal* signal) 
@@ -7729,6 +7873,11 @@ void Dbtc::completeTransAtTakeOverDoOne(
     apiConnectptr.p->currentTcConnect = tcConnectptr.i;
     apiConnectptr.p->currentReplicaNo = tcConnectptr.p->lastReplicaNo;
     tcurrentReplicaNo = tcConnectptr.p->lastReplicaNo;
+    ndbout_c("FAIL_COMMITTED adding [ 0x%.8x 0x%.8x ] to GCI %u/%u",
+             apiConnectptr.p->transid[0], apiConnectptr.p->transid[1],
+             Uint32(apiConnectptr.p->globalcheckpointid >> 32),
+             Uint32(apiConnectptr.p->globalcheckpointid));
+    commitGciHandling(signal, apiConnectptr.p->globalcheckpointid);
     toCompleteHandlingLab(signal);
     return;
   case CS_FAIL_COMMITTING:
@@ -7743,6 +7892,11 @@ void Dbtc::completeTransAtTakeOverDoOne(
     apiConnectptr.p->currentTcConnect = tcConnectptr.i;
     apiConnectptr.p->currentReplicaNo = tcConnectptr.p->lastReplicaNo;
     tcurrentReplicaNo = tcConnectptr.p->lastReplicaNo;
+    ndbout_c("FAIL_COMMITTING adding [ 0x%.8x 0x%.8x ] to GCI %u/%u",
+             apiConnectptr.p->transid[0], apiConnectptr.p->transid[1],
+             Uint32(apiConnectptr.p->globalcheckpointid >> 32),
+             Uint32(apiConnectptr.p->globalcheckpointid));
+    commitGciHandling(signal, apiConnectptr.p->globalcheckpointid);
     toCommitHandlingLab(signal);
     return;
   case CS_FAIL_ABORTING:
@@ -8208,7 +8362,7 @@ void Dbtc::toCommitHandlingLab(Signal* s
 	  sendTCKEY_FAILCONF(signal, apiConnectptr.p);
 	} else {
           jam();
-          sendApiCommit(signal);
+          apiConnectptr = sendApiCommit(signal);
         }//if
         apiConnectptr.p->currentTcConnect = apiConnectptr.p->firstTcConnect;
         tcConnectptr.i = apiConnectptr.p->firstTcConnect;
@@ -8344,6 +8498,12 @@ void Dbtc::toCompleteHandlingLab(Signal*
           signal->theData[1] = (UintR)apiConnectptr.p->takeOverRec;
           signal->theData[2] = apiConnectptr.p->takeOverInd;
           sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);
+          ndbout_c("handleGcp [ 0x%.8x 0x%.8x ] %u/%u",
+                   apiConnectptr.p->transid[0],
+                   apiConnectptr.p->transid[1],
+                   Uint32(apiConnectptr.p->globalcheckpointid >> 32),
+                   Uint32(apiConnectptr.p->globalcheckpointid));
+          handleGcp(signal, apiConnectptr);
           releaseTakeOver(signal);
         } else {
           jam();
@@ -10344,12 +10504,17 @@ void Dbtc::sendScanTabConf(Signal* signa
 }//Dbtc::sendScanTabConf()
 
 
-void Dbtc::gcpTcfinished(Signal* signal) 
+void Dbtc::gcpTcfinished(Signal* signal, Uint64 gci)
 {
+  if (level >= 2)
+    ndbout_c("GSN_GCP_TCFINISHED(%u/%u)",
+             Uint32(gci >> 32),
+             Uint32(gci));
+
   GCPTCFinished* conf = (GCPTCFinished*)signal->getDataPtrSend();
   conf->senderData = c_gcp_ref;
-  conf->gci_hi = Uint32(tcheckGcpId >> 32);
-  conf->gci_lo = Uint32(tcheckGcpId);
+  conf->gci_hi = Uint32(gci >> 32);
+  conf->gci_lo = Uint32(gci);
   sendSignal(cdihblockref, GSN_GCP_TCFINISHED, signal, 
              GCPTCFinished::SignalLength, JBB);
 }//Dbtc::gcpTcfinished()
@@ -10495,6 +10660,7 @@ void Dbtc::initdatabuf(Signal* signal) 
 
 void Dbtc::initgcp(Signal* signal) 
 {
+  Ptr<GcpRecord> gcpPtr;
   ndbrequire(cgcpFilesize > 0);
   for (gcpPtr.i = 0; gcpPtr.i < cgcpFilesize; gcpPtr.i++) {
     ptrAss(gcpPtr, gcpRecord);
@@ -10696,7 +10862,7 @@ void Dbtc::initialiseTcConnect(Signal* s
 /* ----     LINK A GLOBAL CHECKPOINT RECORD INTO THE LIST WITH TRANSACTIONS  */
 /*          WAITING FOR COMPLETION.                                          */
 /* ------------------------------------------------------------------------- */
-void Dbtc::linkGciInGcilist(Signal* signal) 
+void Dbtc::linkGciInGcilist(Ptr<GcpRecord> gcpPtr)
 {
   GcpRecordPtr tmpGcpPointer;
   if (cfirstgcp == RNIL) {
@@ -10869,13 +11035,6 @@ void Dbtc::releaseApiConnectFail(Signal*
   cfirstfreeApiConnectFail = apiConnectptr.i;
 }//Dbtc::releaseApiConnectFail()
 
-void Dbtc::releaseGcp(Signal* signal) 
-{
-  ptrGuard(gcpPtr);
-  gcpPtr.p->nextGcp = cfirstfreeGcp;
-  cfirstfreeGcp = gcpPtr.i;
-}//Dbtc::releaseGcp()
-
 void Dbtc::releaseKeys() 
 {
   UintR Tmp;
@@ -11068,25 +11227,18 @@ void Dbtc::sendSystemError(Signal* signa
 /* ========================================================================= */
 /* -------             LINK ACTUAL GCP OUT OF LIST                   ------- */
 /* ------------------------------------------------------------------------- */
-void Dbtc::unlinkGcp(Signal* signal) 
+void Dbtc::unlinkGcp(Ptr<GcpRecord> tmpGcpPtr)
 {
-  if (cfirstgcp == gcpPtr.i) {
-    jam();
-    cfirstgcp = gcpPtr.p->nextGcp;
-    if (gcpPtr.i == clastgcp) {
-      jam();
-      clastgcp = RNIL;
-    }//if
-  } else {
+  ndbrequire(cfirstgcp == tmpGcpPtr.i);
+
+  cfirstgcp = tmpGcpPtr.p->nextGcp;
+  if (tmpGcpPtr.i == clastgcp) {
     jam();
-    /* -------------------------------------------------------------------- 
-     * WE ARE TRYING TO REMOVE A GLOBAL CHECKPOINT WHICH WAS NOT THE OLDEST. 
-     * THIS IS A SYSTEM ERROR.                                              
-     * ------------------------------------------------------------------- */
-    sendSystemError(signal, __LINE__);
+    clastgcp = RNIL;
   }//if
-  gcpPtr.p->nextGcp = cfirstfreeGcp;
-  cfirstfreeGcp = gcpPtr.i;
+
+  tmpGcpPtr.p->nextGcp = cfirstfreeGcp;
+  cfirstfreeGcp = tmpGcpPtr.i;
 }//Dbtc::unlinkGcp()
 
 void
Index: telco-6.2/storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp
===================================================================
--- telco-6.2.orig/storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp	2008-07-02 13:06:01.000000000 +0200
+++ telco-6.2/storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp	2008-07-02 15:15:31.000000000 +0200
@@ -1431,10 +1431,10 @@ private:
   
   void checkGcp(Signal* signal);
   void commitGciHandling(Signal* signal, Uint64 Tgci);
-  void copyApi(Signal* signal);
+  void copyApi(Ptr<ApiConnectRecord> dst, Ptr<ApiConnectRecord> src);
   void DIVER_node_fail_handling(Signal* signal, Uint64 Tgci);
-  void gcpTcfinished(Signal* signal);
-  void handleGcp(Signal* signal);
+  void gcpTcfinished(Signal* signal, Uint64 gci);
+  void handleGcp(Signal* signal, Ptr<ApiConnectRecord>);
   void hash(Signal* signal);
   bool handle_special_hash(Uint32 dstHash[4], 
 			     Uint32* src, Uint32 srcLen, 
@@ -1453,8 +1453,8 @@ private:
   void initialiseScanOprec(Signal* signal);
   void initTable(Signal* signal);
   void initialiseTcConnect(Signal* signal);
-  void linkApiToGcp(Signal* signal);
-  void linkGciInGcilist(Signal* signal);
+  void linkApiToGcp(Ptr<GcpRecord>, Ptr<ApiConnectRecord>);
+  void linkGciInGcilist(Ptr<GcpRecord>);
   void linkKeybuf(Signal* signal);
   void linkTcInConnectionlist(Signal* signal);
   void releaseAbortResources(Signal* signal);
@@ -1462,7 +1462,6 @@ private:
   void releaseApiConCopy(Signal* signal);
   void releaseApiConnectFail(Signal* signal);
   void releaseAttrinfo();
-  void releaseGcp(Signal* signal);
   void releaseKeys();
   void releaseDirtyRead(Signal*, ApiConnectRecordPtr, TcConnectRecord*);
   void releaseDirtyWrite(Signal* signal);
@@ -1474,10 +1473,10 @@ private:
   void seizeApiConnectCopy(Signal* signal);
   void seizeApiConnectFail(Signal* signal);
   void seizeDatabuf(Signal* signal);
-  void seizeGcp(Signal* signal);
+  void seizeGcp(Ptr<GcpRecord> & dst, Uint64 gci);
   void seizeTcConnect(Signal* signal);
   void seizeTcConnectFail(Signal* signal);
-  void sendApiCommit(Signal* signal);
+  Ptr<ApiConnectRecord> sendApiCommit(Signal* signal);
   void sendAttrinfo(Signal* signal,
                     UintR TattrinfoPtr,
                     AttrbufRecord * const regAttrPtr,
@@ -1488,8 +1487,8 @@ private:
   void sendSystemError(Signal* signal, int line);
   void sendtckeyconf(Signal* signal, UintR TcommitFlag);
   void sendTcIndxConf(Signal* signal, UintR TcommitFlag);
-  void unlinkApiConnect(Signal* signal);
-  void unlinkGcp(Signal* signal);
+  void unlinkApiConnect(Ptr<GcpRecord>, Ptr<ApiConnectRecord>);
+  void unlinkGcp(Ptr<GcpRecord>);
   void unlinkReadyTcCon(Signal* signal);
   void handleFailedOperation(Signal* signal,
 			     const LqhKeyRef * const lqhKeyRef, 
@@ -1661,8 +1660,8 @@ private:
   UintR chostFilesize;
   NdbNodeBitmask c_alive_nodes;
 
+  Uint32 c_ongoing_take_over_cnt;
   GcpRecord *gcpRecord;
-  GcpRecordPtr gcpPtr;
   UintR cgcpFilesize;
 
   TableRecord *tableRecord;
@@ -1677,7 +1676,6 @@ private:
   UintR ctcTimer;
   UintR cDbHbInterval;
 
-  ApiConnectRecordPtr tmpApiConnectptr;
   Uint64 tcheckGcpId;
 
   struct TransCounters {
Index: telco-6.2/storage/ndb/src/kernel/blocks/dbtc/DbtcInit.cpp
===================================================================
--- telco-6.2.orig/storage/ndb/src/kernel/blocks/dbtc/DbtcInit.cpp	2008-06-02 07:38:39.000000000 +0200
+++ telco-6.2/storage/ndb/src/kernel/blocks/dbtc/DbtcInit.cpp	2008-07-02 14:30:46.000000000 +0200
@@ -292,6 +292,7 @@ Dbtc::Dbtc(Block_context& ctx):
   tcFailRecord = 0;
   c_apiConTimer = 0;
   c_apiConTimer_line = 0;
+  c_ongoing_take_over_cnt = 0;
 
 #ifdef VM_TRACE
   {
@@ -300,8 +301,6 @@ Dbtc::Dbtc(Block_context& ctx):
 		    &cachePtr,
 		    &attrbufptr,
 		    &hostptr,
-		    &gcpPtr,
-		    &tmpApiConnectptr,
 		    &timeOutptr,
 		    &scanFragptr,
 		    &databufptr,
Index: telco-6.2/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
===================================================================
--- telco-6.2.orig/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp	2008-07-02 12:38:03.000000000 +0200
+++ telco-6.2/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp	2008-07-03 10:11:46.000000000 +0200
@@ -5456,18 +5456,43 @@ void Dbdih::execMASTER_GCPREQ(Signal* si
   case MicroGcp::M_GCP_IDLE:
     jam();
     gcpState = MasterGCPConf::GCP_READY;
+    ndbout_c("execMASTER_GCPREQ M_GCP_IDLE old: %u/%u",
+             Uint32(m_micro_gcp.m_old_gci >> 32),
+             Uint32(m_micro_gcp.m_old_gci));
     break;
   case MicroGcp::M_GCP_PREPARE:
     jam();
     gcpState = MasterGCPConf::GCP_PREPARE_RECEIVED;
+    ndbout_c("execMASTER_GCPREQ M_GCP_PREPARE old: %u/%u",
+             Uint32(m_micro_gcp.m_old_gci >> 32),
+             Uint32(m_micro_gcp.m_old_gci));
     break;
   case MicroGcp::M_GCP_COMMIT:
     jam();
     gcpState = MasterGCPConf::GCP_COMMIT_RECEIVED;
+    ndbout_c("execMASTER_GCPREQ M_GCP_COMMIT old: %u/%u",
+             Uint32(m_micro_gcp.m_old_gci >> 32),
+             Uint32(m_micro_gcp.m_old_gci));
     break;
   case MicroGcp::M_GCP_COMMITTED:
     jam();
     gcpState = MasterGCPConf::GCP_COMMITTED;
+    ndbout_c("execMASTER_GCPREQ M_GCP_COMMITTED old: %u/%u",
+             Uint32(m_micro_gcp.m_old_gci >> 32),
+             Uint32(m_micro_gcp.m_old_gci));
+    ndbout_c("changing state to M_GCP_COMMIT and running GSN_GCP_NOMORETRANS");
+
+    gcpState = MasterGCPConf::GCP_COMMIT_RECEIVED;
+    m_micro_gcp.m_state = MicroGcp::M_GCP_COMMIT;
+
+    {
+      GCPNoMoreTrans* req2 = (GCPNoMoreTrans*)signal->getDataPtrSend();
+      req2->senderData = m_micro_gcp.m_master_ref;
+      req2->gci_hi = m_micro_gcp.m_old_gci >> 32;
+      req2->gci_lo = m_micro_gcp.m_old_gci & 0xFFFFFFFF;
+      sendSignal(clocaltcblockref, GSN_GCP_NOMORETRANS, signal,
+                 GCPNoMoreTrans::SignalLength, JBB);
+    }
     break;
   };
 
@@ -8736,6 +8761,14 @@ void Dbdih::execGCP_COMMIT(Signal* signa
                GSN_NDB_TAMPER, signal, 1, JBB);
     signal->theData[0] = save;
     CLEAR_ERROR_INSERT_VALUE;
+
+    signal->theData[0] = 9999;
+    sendSignal(numberToRef(CMVMI, c_error_insert_extra),
+               GSN_DUMP_STATE_ORD, signal, 1, JBB);
+
+    signal->theData[0] = save;
+    CLEAR_ERROR_INSERT_VALUE;
+
     return;
   }
 #endif
