Merge perch.ndb.mysql.com:/home/jonas/src/51-new

into  perch.ndb.mysql.com:/home/jonas/src/mysql-5.1-new

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

@@ -775,7 +775,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 *);
@@ -1300,7 +1300,7 @@ private:
     }
 
     Uint32 lcpStart;
-    Uint32 lcpStartGcp; 
+    Uint32 lcpStopGcp; 
     Uint32 keepGci;      /* USED TO CALCULATE THE GCI TO KEEP AFTER A LCP  */
     Uint32 oldestRestorableGci;
     
@@ -1369,7 +1369,8 @@ private:
   Uint32 cstarttype;
   Uint32 csystemnodes;
   Uint32 currentgcp;
-  
+  Uint32 c_newest_restorable_gci;
+
   enum GcpMasterTakeOverState {
     GMTOS_IDLE = 0,
     GMTOS_INITIAL = 1,

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

@@ -685,6 +685,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;
@@ -703,6 +704,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
@@ -8039,6 +8041,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;
@@ -9522,7 +9526,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
@@ -9605,7 +9609,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.                        */
   /* ----------------------------------------------------------------------- */
@@ -9677,7 +9680,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();
@@ -10537,6 +10540,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
@@ -10971,7 +10975,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;
@@ -10993,7 +10998,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
@@ -11287,7 +11291,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];
@@ -11409,7 +11413,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;
@@ -11446,6 +11450,7 @@ void Dbdih::initCommonData()
   csystemnodes = 0;
   c_updateToLock = RNIL;
   currentgcp = 0;
+  c_newest_restorable_gci = 0;
   cverifyQueueCounter = 0;
   cwaitLcpSr = false;
   c_nextLogPart = 0;
@@ -11522,6 +11527,7 @@ void Dbdih::initRestartInfo()
   currentgcp = 2;
   cnewgcp = 2;
   crestartGci = 1;
+  c_newest_restorable_gci = 1;
 
   SYSFILE->keepGCI             = 1;
   SYSFILE->oldestRestorableGCI = 1;
@@ -13494,9 +13500,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  
@@ -13677,8 +13683,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

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

@@ -14738,7 +14738,9 @@ void Dblqh::execSr(Signal* signal)
           signal->theData[4] = logFilePtr.p->currentFilepage;
           signal->theData[5] = logFilePtr.p->currentMbyte;
           signal->theData[6] = logPagePtr.p->logPageWord[ZCURR_PAGE_INDEX];
-          sendSignal(cownref, GSN_DEBUG_SIG, signal, 7, JBA);
+	  signal->theData[7] = ~0;
+	  signal->theData[8] = __LINE__;
+          sendSignal(cownref, GSN_DEBUG_SIG, signal, 9, JBA);
           return;
         }//if
       }//if
@@ -14804,7 +14806,8 @@ void Dblqh::execSr(Signal* signal)
       signal->theData[5] = logFilePtr.p->currentFilepage;
       signal->theData[6] = logPagePtr.p->logPageWord[ZCURR_PAGE_INDEX];
       signal->theData[7] = logWord;
-      sendSignal(cownref, GSN_DEBUG_SIG, signal, 8, JBA);
+      signal->theData[8] = __LINE__;
+      sendSignal(cownref, GSN_DEBUG_SIG, signal, 9, JBA);
       return;
       break;
     }//switch
@@ -14833,8 +14836,9 @@ void Dblqh::execDEBUG_SIG(Signal* signal)
 
   char buf[100];
   BaseString::snprintf(buf, 100, 
-	   "Error while reading REDO log.\n"
+	   "Error while reading REDO log. from %d\n"
 	   "D=%d, F=%d Mb=%d FP=%d W1=%d W2=%d",
+	   signal->theData[8],
 	   signal->theData[2], signal->theData[3], signal->theData[4],
 	   signal->theData[5], signal->theData[6], signal->theData[7]);
 
@@ -15417,6 +15421,10 @@ void Dblqh::readSrFourthZeroLab(Signal* signal)
   // to read a page from file. 
   lfoPtr.p->lfoState = LogFileOperationRecord::WRITE_SR_INVALIDATE_PAGES;
 
+  /**
+   * Make sure we dont release zero page
+   */
+  seizeLogpage(signal);
   invalidateLogAfterLastGCI(signal);
   return;
 }//Dblqh::readSrFourthZeroLab()