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

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

ndb/src/kernel/blocks/dbdih/Dbdih.hpp

@@ -774,7 +774,7 @@ private:
 //------------------------------------
 // Methods for LCP functionality
 //------------------------------------
-  void checkKeepGci(Uint32 replicaStartIndex);
+  void checkKeepGci(TabRecordPtr, Uint32, Fragmentstore*, Uint32);
   void checkLcpStart(Signal *, Uint32 lineNo);
   void checkStartMoreLcp(Signal *, Uint32 nodeId);
   bool reportLcpCompletion(const class LcpFragRep *);
@@ -1292,7 +1292,7 @@ private:
     }
 
     Uint32 lcpStart;
-    Uint32 lcpStartGcp; 
+    Uint32 lcpStopGcp; 
     Uint32 keepGci;      /* USED TO CALCULATE THE GCI TO KEEP AFTER A LCP  */
     Uint32 oldestRestorableGci;
     
@@ -1361,7 +1361,8 @@ private:
   Uint32 cstarttype;
   Uint32 csystemnodes;
   Uint32 currentgcp;
-  
+  Uint32 c_newest_restorable_gci;
+
   enum GcpMasterTakeOverState {
     GMTOS_IDLE = 0,
     GMTOS_INITIAL = 1,

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

@@ -674,6 +674,7 @@ void Dbdih::execCOPY_GCIREQ(Signal* signal)
     jam();
     coldgcp = SYSFILE->newestRestorableGCI;
     crestartGci = SYSFILE->newestRestorableGCI;
+    c_newest_restorable_gci = SYSFILE->newestRestorableGCI;
     Sysfile::setRestartOngoing(SYSFILE->systemRestartBits);
     currentgcp = coldgcp + 1;
     cnewgcp = coldgcp + 1;
@@ -692,6 +693,7 @@ void Dbdih::execCOPY_GCIREQ(Signal* signal)
     ok = true;
     jam();
     cgcpParticipantState = GCP_PARTICIPANT_COPY_GCI_RECEIVED;
+    c_newest_restorable_gci = SYSFILE->newestRestorableGCI;
     setNodeInfo(signal);
     break;
   }//if
@@ -7790,6 +7792,8 @@ void Dbdih::execCOPY_GCICONF(Signal* signal)
     signal->theData[1] = coldgcp;
     sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);    
 
+    c_newest_restorable_gci = coldgcp;
+
     CRASH_INSERTION(7004);
     emptyWaitGCPMasterQueue(signal);    
     cgcpStatus = GCP_READY;
@@ -9192,7 +9196,7 @@ void Dbdih::checkTcCounterLab(Signal* signal)
   }//if
   c_lcpState.ctimer += 32;
   if ((c_nodeStartMaster.blockLcp == true) ||
-      ((c_lcpState.lcpStartGcp + 1) > currentgcp)) {
+      (c_lcpState.lcpStopGcp >= c_newest_restorable_gci)) {
     jam();
     /* --------------------------------------------------------------------- */
     // No reason to start juggling the states and checking for start of LCP if
@@ -9275,7 +9279,6 @@ void Dbdih::execTCGETOPSIZECONF(Signal* signal)
   /* ----------------------------------------------------------------------- */
   c_lcpState.ctimer = 0;
   c_lcpState.keepGci = coldgcp;
-  c_lcpState.lcpStartGcp = currentgcp;
   /* ----------------------------------------------------------------------- */
   /*       UPDATE THE NEW LATEST LOCAL CHECKPOINT ID.                        */
   /* ----------------------------------------------------------------------- */
@@ -9347,7 +9350,7 @@ void Dbdih::calculateKeepGciLab(Signal* signal, Uint32 tableId, Uint32 fragId)
   cnoOfActiveTables++;
   FragmentstorePtr fragPtr;
   getFragstore(tabPtr.p, fragId, fragPtr);
-  checkKeepGci(fragPtr.p->storedReplicas);
+  checkKeepGci(tabPtr, fragId, fragPtr.p, fragPtr.p->storedReplicas);
   fragId++;
   if (fragId >= tabPtr.p->totalfragments) {
     jam();
@@ -10205,6 +10208,7 @@ void Dbdih::allNodesLcpCompletedLab(Signal* signal)
   signal->theData[0] = NDB_LE_LocalCheckpointCompleted; //Event type
   signal->theData[1] = SYSFILE->latestLCP_ID;
   sendSignal(CMVMI_REF, GSN_EVENT_REP, signal, 2, JBB);
+  c_lcpState.lcpStopGcp = c_newest_restorable_gci;
   
   /**
    * Start checking for next LCP
@@ -10639,7 +10643,8 @@ void Dbdih::checkEscalation()
 /*       DESCRIPTION: CHECK FOR MINIMUM GCI RESTORABLE WITH NEW LOCAL    */
 /*                    CHECKPOINT.                                        */
 /*************************************************************************/
-void Dbdih::checkKeepGci(Uint32 replicaStartIndex) 
+void Dbdih::checkKeepGci(TabRecordPtr tabPtr, Uint32 fragId, Fragmentstore*, 
+			 Uint32 replicaStartIndex) 
 {
   ReplicaRecordPtr ckgReplicaPtr;
   ckgReplicaPtr.i = replicaStartIndex;
@@ -10661,7 +10666,6 @@ void Dbdih::checkKeepGci(Uint32 replicaStartIndex)
     if (oldestRestorableGci > c_lcpState.oldestRestorableGci) {
       jam();
       c_lcpState.oldestRestorableGci = oldestRestorableGci;
-      ndbrequire(((int)c_lcpState.oldestRestorableGci) >= 0);
     }//if
     ckgReplicaPtr.i = ckgReplicaPtr.p->nextReplica;
   }//while
@@ -10955,7 +10959,7 @@ void Dbdih::findMinGci(ReplicaRecordPtr fmgReplicaPtr,
   do {
     ndbrequire(lcpNo < MAX_LCP_STORED);
     if (fmgReplicaPtr.p->lcpStatus[lcpNo] == ZVALID &&
-	fmgReplicaPtr.p->maxGciStarted[lcpNo] <= coldgcp)
+	fmgReplicaPtr.p->maxGciStarted[lcpNo] < c_newest_restorable_gci)
     {
       jam();
       keepGci = fmgReplicaPtr.p->maxGciCompleted[lcpNo];
@@ -11077,7 +11081,7 @@ void Dbdih::initCommonData()
 
   c_lcpState.clcpDelay = 0;
   c_lcpState.lcpStart = ZIDLE;
-  c_lcpState.lcpStartGcp = 0;
+  c_lcpState.lcpStopGcp = 0;
   c_lcpState.setLcpStatus(LCP_STATUS_IDLE, __LINE__);
   c_lcpState.currentFragment.tableId = 0;
   c_lcpState.currentFragment.fragmentId = 0;
@@ -11113,6 +11117,7 @@ void Dbdih::initCommonData()
   csystemnodes = 0;
   c_updateToLock = RNIL;
   currentgcp = 0;
+  c_newest_restorable_gci = 0;
   cverifyQueueCounter = 0;
   cwaitLcpSr = false;
 
@@ -11184,6 +11189,7 @@ void Dbdih::initRestartInfo()
   currentgcp = 2;
   cnewgcp = 2;
   crestartGci = 1;
+  c_newest_restorable_gci = 1;
 
   SYSFILE->keepGCI             = 1;
   SYSFILE->oldestRestorableGCI = 1;
@@ -13151,9 +13157,9 @@ Dbdih::execDUMP_STATE_ORD(Signal* signal)
   if (signal->theData[0] == 7001) {
     infoEvent("c_lcpState.keepGci = %d",
               c_lcpState.keepGci);
-    infoEvent("c_lcpState.lcpStatus = %d, clcpStartGcp = %d",
+    infoEvent("c_lcpState.lcpStatus = %d, clcpStopGcp = %d",
               c_lcpState.lcpStatus, 
-	      c_lcpState.lcpStartGcp);
+	      c_lcpState.lcpStopGcp);
     infoEvent("cgcpStartCounter = %d, cimmediateLcpStart = %d",
               cgcpStartCounter, c_lcpState.immediateLcpStart);
   }//if  
@@ -13334,8 +13340,8 @@ Dbdih::execDUMP_STATE_ORD(Signal* signal)
     infoEvent("lcpStatus = %d (update place = %d) ",
 	      c_lcpState.lcpStatus, c_lcpState.lcpStatusUpdatedPlace);
     infoEvent
-      ("lcpStart = %d lcpStartGcp = %d keepGci = %d oldestRestorable = %d",
-       c_lcpState.lcpStart, c_lcpState.lcpStartGcp, 
+      ("lcpStart = %d lcpStopGcp = %d keepGci = %d oldestRestorable = %d",
+       c_lcpState.lcpStart, c_lcpState.lcpStopGcp, 
        c_lcpState.keepGci, c_lcpState.oldestRestorableGci);
     
     infoEvent