Merge whalegate.ndb.mysql.com:/home/tomas/mysql-5.0-ndb

into  whalegate.ndb.mysql.com:/home/tomas/mysql-5.0-maint

ndb/src/kernel/blocks/dblqh/Dblqh.hpp

@@ -1617,7 +1617,8 @@ public:
       ACTIVE_WRITE_LOG = 17,             ///< A write operation during 
                                         ///< writing of log
       READ_SR_INVALIDATE_PAGES = 18,
-      WRITE_SR_INVALIDATE_PAGES = 19
+      WRITE_SR_INVALIDATE_PAGES = 19,
+      WRITE_SR_INVALIDATE_PAGES_UPDATE_PAGE0 = 20
     };
     /**
      * We have to remember the log pages read. 
@@ -2436,7 +2437,7 @@ private:
   void errorReport(Signal* signal, int place);
   void warningReport(Signal* signal, int place);
   void invalidateLogAfterLastGCI(Signal *signal);
-  void readFileInInvalidate(Signal *signal);
+  void readFileInInvalidate(Signal *signal, bool stepNext);
   void exitFromInvalidate(Signal* signal);
   Uint32 calcPageCheckSum(LogPageRecordPtr logP);
 

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

@@ -109,6 +109,10 @@ operator<<(NdbOut& out, Dblqh::ScanRecord::ScanType state){
 //#define MARKER_TRACE 1
 //#define TRACE_SCAN_TAKEOVER 1
 
+#ifndef DEBUG_REDO
+#define DEBUG_REDO 0
+#endif
+
 const Uint32 NR_ScanNo = 0;
 
 void Dblqh::execACC_COM_BLOCK(Signal* signal)
@@ -11153,6 +11157,13 @@ void Dblqh::sendLCP_COMPLETE_REP(Signal* signal, Uint32 lcpId)
     jam();
     sendEMPTY_LCP_CONF(signal, true);
   }
+
+  if (getNodeState().getNodeRestartInProgress() && cstartRecReq != 3)
+  {
+    jam();
+    ndbrequire(cstartRecReq == 2);
+    cstartRecReq = 3;
+  }
   return;
 }//Dblqh::sendCOMP_LCP_ROUND()
 
@@ -11717,15 +11728,27 @@ void Dblqh::execGCP_SAVEREQ(Signal* signal)
     return;
   }
   
-  ccurrentGcprec = 0;
-  gcpPtr.i = ccurrentGcprec;
-  ptrCheckGuard(gcpPtr, cgcprecFileSize, gcpRecord);
-  
   cnewestCompletedGci = gci;
   if (gci > cnewestGci) {
     jam();
     cnewestGci = gci;
   }//if
+
+  if(getNodeState().getNodeRestartInProgress() && cstartRecReq < 3)
+  {
+    GCPSaveRef * const saveRef = (GCPSaveRef*)&signal->theData[0];
+    saveRef->dihPtr = dihPtr;
+    saveRef->nodeId = getOwnNodeId();
+    saveRef->gci    = gci;
+    saveRef->errorCode = GCPSaveRef::NodeRestartInProgress;
+    sendSignal(dihBlockRef, GSN_GCP_SAVEREF, signal, 
+	       GCPSaveRef::SignalLength, JBB);
+    return;
+  }
+
+  ccurrentGcprec = 0;
+  gcpPtr.i = ccurrentGcprec;
+  ptrCheckGuard(gcpPtr, cgcprecFileSize, gcpRecord);
   
   gcpPtr.p->gcpBlockref = dihBlockRef;
   gcpPtr.p->gcpUserptr = dihPtr;
@@ -11979,9 +12002,6 @@ void Dblqh::execFSCLOSECONF(Signal* signal)
   case LogFileRecord::CLOSE_SR_INVALIDATE_PAGES:
     jam();
     logFilePtr.p->logFileStatus = LogFileRecord::CLOSED;
-    // Set the prev file to check if we shall close it.
-    logFilePtr.i = logFilePtr.p->prevLogFile;
-    ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
 
     logPartPtr.i = logFilePtr.p->logPartRec;
     ptrCheckGuard(logPartPtr, clogPartFileSize, logPartRecord);
@@ -12031,7 +12051,7 @@ void Dblqh::execFSOPENCONF(Signal* signal)
   case LogFileRecord::OPEN_SR_INVALIDATE_PAGES:
     jam();
     logFilePtr.p->logFileStatus = LogFileRecord::OPEN;
-    readFileInInvalidate(signal);
+    readFileInInvalidate(signal, false);
     return;
   case LogFileRecord::OPENING_INIT:
     jam();
@@ -12211,6 +12231,7 @@ void Dblqh::execFSWRITECONF(Signal* signal)
   case LogFileOperationRecord::WRITE_SR_INVALIDATE_PAGES:
     jam();
     invalidateLogAfterLastGCI(signal);
+    CRASH_INSERTION(5047);
     return;
   case LogFileOperationRecord::WRITE_PAGE_ZERO:
     jam();
@@ -12248,6 +12269,14 @@ void Dblqh::execFSWRITECONF(Signal* signal)
     jam();
     firstPageWriteLab(signal);
     return;
+  case LogFileOperationRecord::WRITE_SR_INVALIDATE_PAGES_UPDATE_PAGE0:
+    jam();
+    // We are done...send completed signal and exit this phase.
+    releaseLfo(signal);
+    signal->theData[0] = ZSR_FOURTH_COMP;
+    signal->theData[1] = logPartPtr.i;
+    sendSignal(cownref, GSN_CONTINUEB, signal, 2, JBB);
+    return;
   default:
     jam();
     systemErrorLab(signal, __LINE__);
@@ -13318,6 +13347,12 @@ void Dblqh::writeSinglePage(Signal* signal, Uint32 pageNo,
   signal->theData[6] = logPagePtr.i;
   signal->theData[7] = pageNo;
   sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal, 8, JBA);
+
+  if (DEBUG_REDO)
+    ndbout_c("writeSingle 1 page at part: %u file: %u pos: %u",
+	     logPartPtr.i,
+	     logFilePtr.p->fileNo,
+	     pageNo);
 }//Dblqh::writeSinglePage()
 
 /* ##########################################################################
@@ -13383,6 +13418,12 @@ void Dblqh::openSrLastFileLab(Signal* signal)
 void Dblqh::readSrLastFileLab(Signal* signal) 
 {
   logPartPtr.p->logLap = logPagePtr.p->logPageWord[ZPOS_LOG_LAP];
+  if (DEBUG_REDO)
+    ndbout_c("readSrLastFileLab part: %u logExecState: %u logPartState: %u logLap: %u",
+             logPartPtr.i,
+ 	     logPartPtr.p->logExecState,
+ 	     logPartPtr.p->logPartState,
+ 	     logPartPtr.p->logLap);
   if (logPartPtr.p->noLogFiles > ZMAX_LOG_FILES_IN_PAGE_ZERO) {
     jam();
     initGciInLogFileRec(signal, ZMAX_LOG_FILES_IN_PAGE_ZERO);
@@ -14600,6 +14641,20 @@ void Dblqh::srLogLimits(Signal* signal)
       break;
     }//if
   }//while
+
+  if (DEBUG_REDO)
+  {
+    LogFileRecordPtr tmp;
+    tmp.i = logPartPtr.p->stopLogfile;
+    ptrCheckGuard(tmp, clogFileFileSize, logFileRecord);
+    ndbout_c("srLogLimits part: %u start file: %u mb: %u stop file: %u mb: %u",
+             logPartPtr.i,
+             tlastPrepRef >> 16,
+             tlastPrepRef & 65535,
+             tmp.p->fileNo,
+             logPartPtr.p->stopMbyte);           
+  }
+
   /* ------------------------------------------------------------------------
    *  WE HAVE NOW FOUND BOTH THE START AND THE STOP OF THE LOG. NOW START
    *  EXECUTING THE LOG. THE FIRST ACTION IS TO OPEN THE LOG FILE WHERE TO
@@ -15026,6 +15081,12 @@ void Dblqh::execSr(Signal* signal)
     case ZCOMPLETED_GCI_TYPE:
       jam();
       logWord = readLogword(signal);
+      if (DEBUG_REDO)
+	ndbout_c("found gci: %u part: %u file: %u page: %u",
+		 logWord,
+		 logPartPtr.i,
+		 logFilePtr.p->fileNo,
+		 logFilePtr.p->currentFilepage);
       if (logWord == logPartPtr.p->logLastGci) {
         jam();
 /*---------------------------------------------------------------------------*/
@@ -15042,6 +15103,10 @@ void Dblqh::execSr(Signal* signal)
           logPartPtr.p->headPageNo = logFilePtr.p->currentFilepage;
           logPartPtr.p->headPageIndex = 
                   logPagePtr.p->logPageWord[ZCURR_PAGE_INDEX];
+	  logPartPtr.p->logLap = logPagePtr.p->logPageWord[ZPOS_LOG_LAP];
+	  if (DEBUG_REDO)
+	    ndbout_c("execSr part: %u logLap: %u",
+		     logPartPtr.i, logPartPtr.p->logLap);
         }//if
 /*---------------------------------------------------------------------------*/
 /* THERE IS NO NEED OF EXECUTING PAST THIS LINE SINCE THERE WILL ONLY BE LOG */
@@ -15202,67 +15267,140 @@ void Dblqh::invalidateLogAfterLastGCI(Signal* signal) {
   }
 
   switch (lfoPtr.p->lfoState) {
-  case LogFileOperationRecord::WRITE_SR_INVALIDATE_PAGES:
-    jam();
-    releaseLfo(signal);
-    releaseLogpage(signal); 
-    if (logPartPtr.p->invalidatePageNo < (ZNO_MBYTES_IN_FILE * ZPAGES_IN_MBYTE - 1)) {
-      // We continue in this file.
-      logPartPtr.p->invalidatePageNo++;
-    } else {
-      // We continue in the next file.
-      logFilePtr.i = logFilePtr.p->nextLogFile;
-      ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
-      logPartPtr.p->invalidateFileNo = logFilePtr.p->fileNo;
-      // Page 0 is used for file descriptors.
-      logPartPtr.p->invalidatePageNo = 1; 
-      if (logFilePtr.p->logFileStatus != LogFileRecord::OPEN) {
-	jam();
-	logFilePtr.p->logFileStatus = LogFileRecord::OPEN_SR_INVALIDATE_PAGES;
-	openFileRw(signal, logFilePtr);
-	return;
-	break;
-      }
-    }
-    // Read a page from the log file. 
-    readFileInInvalidate(signal);
-    return;
-    break;
-
   case LogFileOperationRecord::READ_SR_INVALIDATE_PAGES:
     jam();
-    releaseLfo(signal);
     // Check if this page must be invalidated.
     // If the log lap number on a page after the head of the tail is the same 
     // as the actual log lap number we must invalidate this page. Otherwise it
     // could be impossible to find the end of the log in a later system/node 
     // restart.
-    if (logPagePtr.p->logPageWord[ZPOS_LOG_LAP] == logPartPtr.p->logLap) {
+    if (logPagePtr.p->logPageWord[ZPOS_LOG_LAP] == logPartPtr.p->logLap) 
+    {
       // This page must be invalidated.
-      logPagePtr.p->logPageWord[ZPOS_LOG_LAP] = 0;
-      // Contact NDBFS. Real time break.
-      writeSinglePage(signal, logPartPtr.p->invalidatePageNo,
-                      ZPAGE_SIZE - 1, __LINE__);
-      lfoPtr.p->lfoState = LogFileOperationRecord::WRITE_SR_INVALIDATE_PAGES;
-    } else {
-      // We are done with invalidating. Finish start phase 3.4. 
+      // We search for end
+      // read next
+      releaseLfo(signal);
+      releaseLogpage(signal); 
+      readFileInInvalidate(signal, true);
+      lfoPtr.p->lfoState = LogFileOperationRecord::READ_SR_INVALIDATE_PAGES;
+      return;
+    }
+
+    /**
+     * We found the "last" page to invalidate...
+     *   Invalidate backwards until head...
+     */
+
+    // Fall through...
+  case LogFileOperationRecord::WRITE_SR_INVALIDATE_PAGES:
+    jam();
+
+    releaseLfo(signal);
+    releaseLogpage(signal); 
+
+    // Step backwards...
+    logPartPtr.p->invalidatePageNo--;
+
+    if (logPartPtr.p->invalidatePageNo == 0)
+    {
+      jam();
+
+      if (logFilePtr.p->fileNo == 0)
+      {
+	/**
+	 * We're wrapping in the log...
+	 *   update logLap
+	 */
+	logPartPtr.p->logLap--;
+	ndbrequire(logPartPtr.p->logLap); // Should always be > 0
+	if (DEBUG_REDO)
+	  ndbout_c("invalidateLogAfterLastGCI part: %u wrap from file 0 -> logLap: %u",
+		   logPartPtr.i, logPartPtr.p->logLap);
+      }
+      
+      /**
+       * Move to prev file
+       */
+      logFilePtr.i = logFilePtr.p->prevLogFile;
+      ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
+      logPartPtr.p->invalidateFileNo = logFilePtr.p->fileNo;
+      logPartPtr.p->invalidatePageNo = ZNO_MBYTES_IN_FILE * ZPAGES_IN_MBYTE - 1;
+    }
+    
+    if (logPartPtr.p->invalidateFileNo == logPartPtr.p->headFileNo &&
+	logPartPtr.p->invalidatePageNo == logPartPtr.p->headPageNo)
+    {
+      /**
+       * Done...
+       */
+      logFilePtr.i = logPartPtr.p->currentLogfile;
+      ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
+      
+      logFilePtr.i = logFilePtr.p->nextLogFile;
+      ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
+
+      // Close files if necessary. Current file and the next file should be 
+      // left open.
       exitFromInvalidate(signal);
+      return;
     }
-    return;
-    break;
 
+    seizeLogpage(signal);
+
+    /**
+     * Make page really empty
+     */
+    bzero(logPagePtr.p, sizeof(LogPageRecord));
+    writeSinglePage(signal, logPartPtr.p->invalidatePageNo,
+		    ZPAGE_SIZE - 1, __LINE__);
+
+    lfoPtr.p->lfoState = LogFileOperationRecord::WRITE_SR_INVALIDATE_PAGES;
+    return;
   default:
     jam();
     systemError(signal, __LINE__);
     return;
     break;
   }
-
-  return;
 }//Dblqh::invalidateLogAfterLastGCI
 
-void Dblqh::readFileInInvalidate(Signal* signal) {
+void Dblqh::readFileInInvalidate(Signal* signal, bool stepNext) 
+{
   jam();
+
+  if (stepNext)
+  {
+    logPartPtr.p->invalidatePageNo++;
+    if (logPartPtr.p->invalidatePageNo == (ZNO_MBYTES_IN_FILE * ZPAGES_IN_MBYTE)) 
+    {
+      // We continue in the next file.
+      logFilePtr.i = logFilePtr.p->nextLogFile;
+      ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
+      logPartPtr.p->invalidateFileNo = logFilePtr.p->fileNo;
+      // Page 0 is used for file descriptors.
+      logPartPtr.p->invalidatePageNo = 1; 
+
+      if (logFilePtr.p->fileNo == 0)
+      {
+	/**
+	 * We're wrapping in the log...
+	 *   update logLap
+	 */
+	logPartPtr.p->logLap++;
+	if (DEBUG_REDO)
+	  ndbout_c("readFileInInvalidate part: %u wrap to file 0 -> logLap: %u",
+		   logPartPtr.i, logPartPtr.p->logLap);
+      }
+      if (logFilePtr.p->logFileStatus != LogFileRecord::OPEN) 
+      {
+	jam();
+	logFilePtr.p->logFileStatus = LogFileRecord::OPEN_SR_INVALIDATE_PAGES;
+	openFileRw(signal, logFilePtr);
+	return;
+      }
+    }
+  }
+  
   // Contact NDBFS. Real time break.
   readSinglePage(signal, logPartPtr.p->invalidatePageNo); 
   lfoPtr.p->lfoState = LogFileOperationRecord::READ_SR_INVALIDATE_PAGES;
@@ -15270,34 +15408,57 @@ void Dblqh::readFileInInvalidate(Signal* signal) {
 
 void Dblqh::exitFromInvalidate(Signal* signal) {
   jam();
-  // Close files if necessary. Current file and the next file should be 
-  // left open.
-  if (logFilePtr.i != logPartPtr.p->currentLogfile) {
-    LogFileRecordPtr currentLogFilePtr;
-    LogFileRecordPtr nextAfterCurrentLogFilePtr;
-
-    currentLogFilePtr.i = logPartPtr.p->currentLogfile;
-    ptrCheckGuard(currentLogFilePtr, clogFileFileSize, logFileRecord);
-
-    nextAfterCurrentLogFilePtr.i = currentLogFilePtr.p->nextLogFile;
-
-    if (logFilePtr.i != nextAfterCurrentLogFilePtr.i) {
-      // This file should be closed.
-      logFilePtr.p->logFileStatus = LogFileRecord::CLOSE_SR_INVALIDATE_PAGES;
-      closeFile(signal, logFilePtr);  
-      // Return from this function and wait for close confirm. Then come back 
-      // and test the previous file for closing.
-      return;
-    }
-  }    
 
-  // We are done with closing files, send completed signal and exit this phase.
-  signal->theData[0] = ZSR_FOURTH_COMP;
-  signal->theData[1] = logPartPtr.i;
-  sendSignal(cownref, GSN_CONTINUEB, signal, 2, JBB);
+loop:  
+  logFilePtr.i = logFilePtr.p->nextLogFile;
+  ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
+
+  if (logFilePtr.i == logPartPtr.p->currentLogfile)
+  {
+    jam();
+    goto done;
+  }
+  
+  if (logFilePtr.p->fileNo == 0)
+  {
+    jam();
+    /**
+     * Logfile 0 shoult *not* be closed
+     */
+    goto loop;
+  }
+  
+  if (logFilePtr.p->logFileStatus == LogFileRecord::CLOSED)
+  {
+    jam();
+    goto done;
+  }
+
+  jam();
+  ndbrequire(logFilePtr.p->logFileStatus == LogFileRecord::OPEN);
+  logFilePtr.p->logFileStatus = LogFileRecord::CLOSE_SR_INVALIDATE_PAGES;
+  closeFile(signal, logFilePtr);
   return;
-}
 
+done:
+  if (DEBUG_REDO)
+    ndbout_c("exitFromInvalidate part: %u head file: %u page: %u", 
+	     logPartPtr.i,
+	     logPartPtr.p->headFileNo,
+	     logPartPtr.p->headPageNo);
+  
+  logFilePtr.i = logPartPtr.p->firstLogfile;
+  ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
+  logPagePtr.i = logFilePtr.p->logPageZero;
+  ptrCheckGuard(logPagePtr, clogPageFileSize, logPageRecord);
+  logPagePtr.p->logPageWord[ZPAGE_HEADER_SIZE + ZPOS_FILE_NO] = 
+    logPartPtr.p->headFileNo;
+  writeSinglePage(signal, 0, ZPAGE_SIZE - 1, __LINE__);
+
+  lfoPtr.p->logFileRec = logFilePtr.i;
+  lfoPtr.p->lfoState = LogFileOperationRecord::WRITE_SR_INVALIDATE_PAGES_UPDATE_PAGE0;
+  return;
+}
 
 /*---------------------------------------------------------------------------*/
 /* THE EXECUTION OF A LOG RECORD IS COMPLETED. RELEASE PAGES IF THEY WERE    */
@@ -15684,20 +15845,10 @@ void Dblqh::readSrFourthZeroLab(Signal* signal)
   ptrCheckGuard(logFilePtr, clogFileFileSize, logFileRecord);
   logPartPtr.p->invalidateFileNo = logPartPtr.p->headFileNo;
   logPartPtr.p->invalidatePageNo = logPartPtr.p->headPageNo;
-   
   logPartPtr.p->logExecState = LogPartRecord::LES_EXEC_LOG_INVALIDATE;
-  seizeLfo(signal);
-  initLfo(signal);
-  // The state here is a little confusing, but simulates that we return
-  // to invalidateLogAfterLastGCI() from an invalidate write and are ready
-  // 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);
+   
+  readFileInInvalidate(signal, true);
+  lfoPtr.p->lfoState = LogFileOperationRecord::READ_SR_INVALIDATE_PAGES;
   return;
 }//Dblqh::readSrFourthZeroLab()
 
@@ -16269,6 +16420,14 @@ void Dblqh::completedLogPage(Signal* signal, Uint32 clpType, Uint32 place)
   signal->theData[4] = ZVAR_NO_LOG_PAGE_WORD;
   signal->theData[5] = twlpNoPages;
   sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal, 15, JBA);
+
+  if (DEBUG_REDO)
+    ndbout_c("writing %d pages at part: %u file: %u pos: %u",
+	     twlpNoPages,
+	     logPartPtr.i,
+	     logFilePtr.p->fileNo,
+	     logFilePtr.p->filePosition);
+
   if (twlpType == ZNORMAL) {
     jam();
     lfoPtr.p->lfoState = LogFileOperationRecord::ACTIVE_WRITE_LOG;
@@ -17508,6 +17667,14 @@ void Dblqh::readExecLog(Signal* signal)
   signal->theData[14] = lfoPtr.p->logPageArray[8];
   signal->theData[15] = lfoPtr.p->logPageArray[9];
   sendSignal(NDBFS_REF, GSN_FSREADREQ, signal, 16, JBA);
+
+  if (DEBUG_REDO)
+    ndbout_c("readExecLog %u page at part: %u file: %u pos: %u",
+	     lfoPtr.p->noPagesRw,
+	     logPartPtr.i,
+	     logFilePtr.p->fileNo,
+	     logPartPtr.p->execSrStartPageNo);
+
 }//Dblqh::readExecLog()
 
 /* ------------------------------------------------------------------------- */
@@ -17570,6 +17737,14 @@ void Dblqh::readExecSr(Signal* signal)
   signal->theData[13] = lfoPtr.p->logPageArray[7];
   signal->theData[14] = tresPageid;
   sendSignal(NDBFS_REF, GSN_FSREADREQ, signal, 15, JBA);
+
+  if (DEBUG_REDO)
+    ndbout_c("readExecSr %u page at part: %u file: %u pos: %u",
+	     8,
+	     logPartPtr.i,
+	     logFilePtr.p->fileNo,
+	     tresPageid);
+
 }//Dblqh::readExecSr()
 
 /* ------------------------------------------------------------------------- */
@@ -17719,6 +17894,13 @@ void Dblqh::readSinglePage(Signal* signal, Uint32 pageNo)
   signal->theData[6] = logPagePtr.i;
   signal->theData[7] = pageNo;
   sendSignal(NDBFS_REF, GSN_FSREADREQ, signal, 8, JBA);
+
+  if (DEBUG_REDO)
+    ndbout_c("readSinglePage 1 page at part: %u file: %u pos: %u",
+	     logPartPtr.i,
+	     logFilePtr.p->fileNo,
+	     pageNo);
+
 }//Dblqh::readSinglePage()
 
 /* --------------------------------------------------------------------------
@@ -18301,8 +18483,17 @@ void Dblqh::writeCompletedGciLog(Signal* signal)
     jam();
     changeMbyte(signal);
   }//if
+
   logFilePtr.p->remainingWordsInMbyte = 
     logFilePtr.p->remainingWordsInMbyte - ZCOMPLETED_GCI_LOG_SIZE;
+
+  if (DEBUG_REDO)
+    ndbout_c("writeCompletedGciLog gci: %u part: %u file: %u page: %u",
+	     cnewestCompletedGci,
+	     logPartPtr.i,
+	     logFilePtr.p->fileNo,
+	     logFilePtr.p->currentFilepage);
+
   writeLogWord(signal, ZCOMPLETED_GCI_TYPE);
   writeLogWord(signal, cnewestCompletedGci);
   logPartPtr.p->logPartNewestCompletedGCI = cnewestCompletedGci;
@@ -18339,6 +18530,13 @@ void Dblqh::writeDirty(Signal* signal, Uint32 place)
   signal->theData[6] = logPagePtr.i;
   signal->theData[7] = logPartPtr.p->prevFilepage;
   sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal, 8, JBA);
+
+  if (DEBUG_REDO)
+    ndbout_c("writeDirty 1 page at part: %u file: %u pos: %u",
+	     logPartPtr.i,
+	     logFilePtr.p->fileNo,
+	     logPartPtr.p->prevFilepage);
+
 }//Dblqh::writeDirty()
 
 /* --------------------------------------------------------------------------

ndb/src/kernel/blocks/dbtc/DbtcMain.cpp

@@ -883,6 +883,12 @@ void Dbtc::execREAD_NODESCONF(Signal* signal)
         hostptr.p->hostStatus = HS_ALIVE;
 	c_alive_nodes.set(i);
       }//if
+
+      if (NodeBitmask::get(readNodes->startedNodes, i))
+      {
+        jam();
+        hostptr.p->m_nf_bits = HostRecord::NF_STARTED;
+      }
     }//if
   }//for
   ndbsttorry010Lab(signal);
@@ -10223,6 +10229,7 @@ void Dbtc::inithost(Signal* signal)
     hostptr.p->noOfWordsTCINDXCONF = 0;
     hostptr.p->noOfPackedWordsLqh = 0;
     hostptr.p->hostLqhBlockRef = calcLqhBlockRef(hostptr.i);
+    hostptr.p->m_nf_bits = 0;
   }//for
   c_alive_nodes.clear();
 }//Dbtc::inithost()

ndb/test/ndbapi/testNodeRestart.cpp

@@ -943,12 +943,62 @@ int runBug24717(NDBT_Context* ctx, NDBT_Step* step){
     
     restarter.startNodes(&nodeId, 1);
     
-    for (Uint32 i = 0; i < 100; i++)
-    {
-      hugoTrans.pkReadRecords(pNdb, 100, 1, NdbOperation::LM_CommittedRead);
-    }
-    
+    do {
+      for (Uint32 i = 0; i < 100; i++)
+      {
+        hugoTrans.pkReadRecords(pNdb, 100, 1, NdbOperation::LM_CommittedRead);
+      }
+    } while (restarter.waitClusterStarted(5) != 0);
+  }
+  
+  return NDBT_OK;
+}
+
+int 
+runBug29364(NDBT_Context* ctx, NDBT_Step* step){
+  int result = NDBT_OK;
+  int loops = ctx->getNumLoops();
+  int records = ctx->getNumRecords();
+  NdbRestarter restarter;
+  Ndb* pNdb = GETNDB(step);
+  
+  HugoTransactions hugoTrans(*ctx->getTab());
+
+  if (restarter.getNumDbNodes() < 4)
+    return NDBT_OK;
+
+  int dump0[] = { 9000, 0 } ;
+  int dump1[] = { 9001, 0 } ;
+  Uint32 ownNode = refToNode(pNdb->getReference());
+  dump0[1] = ownNode;
+
+  for (; loops; loops --)
+  {
+    int node0 = restarter.getDbNodeId(rand() % restarter.getNumDbNodes());
+    int node1 = restarter.getRandomNodeOtherNodeGroup(node0, rand());
+
+    restarter.restartOneDbNode(node0, false, true, true);
+    restarter.waitNodesNoStart(&node0, 1);
+    restarter.startNodes(&node0, 1);
     restarter.waitClusterStarted();
+
+    restarter.restartOneDbNode(node1, false, true, true);    
+    restarter.waitNodesNoStart(&node1, 1);
+    if (restarter.dumpStateOneNode(node1, dump0, 2))
+      return NDBT_FAILED;
+
+    restarter.startNodes(&node1, 1);    
+    
+    do {
+      
+      for (Uint32 i = 0; i < 100; i++)
+      {
+        hugoTrans.pkReadRecords(pNdb, 100, 1, NdbOperation::LM_CommittedRead);
+      }
+    } while (restarter.waitClusterStarted(5) != 0);
+    
+    if (restarter.dumpStateOneNode(node1, dump1, 1))
+      return NDBT_FAILED;
   }
   
   return NDBT_OK;
@@ -1633,6 +1683,9 @@ TESTCASE("Bug27283", ""){
 TESTCASE("Bug28717", ""){
   INITIALIZER(runBug28717);
 }
+TESTCASE("Bug29364", ""){
+  INITIALIZER(runBug29364);
+}
 NDBT_TESTSUITE_END(testNodeRestart);
 
 int main(int argc, const char** argv){

ndb/test/run-test/daily-basic-tests.txt

@@ -477,6 +477,10 @@ max-time: 1000
 cmd: testNodeRestart
 args: -n Bug24717 T1
 
+max-time: 1000
+cmd: testNodeRestart
+args: -n Bug29364 T1
+
 max-time: 1000
 cmd: testNodeRestart
 args: -n Bug25364 T1
@@ -637,6 +641,10 @@ max-time: 1000
 cmd: testNdbApi
 args: -n BugBug28443
 
+max-time: 1000
+cmd: testNdbApi
+args: -n Bug28443
+
 #max-time: 500
 #cmd: testInterpreter
 #args: T1 

sql/ha_ndbcluster_cond.h

@@ -259,8 +259,20 @@ class Ndb_cond : public Sql_alloc
   ~Ndb_cond() 
   { 
     if (ndb_item) delete ndb_item; 
-    ndb_item= NULL; 
-    if (next) delete next;
+    ndb_item= NULL;
+    /*
+      First item in the linked list deletes all in a loop
+      Note - doing it recursively causes stack issues for
+      big IN clauses
+    */
+    Ndb_cond *n= next;
+    while (n)
+    {
+      Ndb_cond *tmp= n;
+      n= n->next;
+      tmp->next= NULL;
+      delete tmp;
+    }
     next= prev= NULL; 
   };
   Ndb_item *ndb_item;