===== storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp 1.45 vs edited =====
--- 1.45/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp	2008-05-05 11:53:56 +02:00
+++ storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp	2008-05-05 11:32:40 +02:00
@@ -2943,6 +2943,11 @@
   }
 
   DLHashTable<ScanRecord> c_scanTakeOverHash;
+
+
+  Uint32 c_last_force_lcp_time;
+  void force_lcp(Signal*);
+
 };
 
 inline
New file: storage/ndb/src/kernel/blocks/dblqh/DblqhInit.o
===== storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp 1.100 vs edited =====
--- 1.100/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp	2008-05-05 11:53:56 +02:00
+++ storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp	2008-05-05 11:34:20 +02:00
@@ -922,6 +922,7 @@
   initRecords();
   initialiseRecordsLab(signal, 0, ref, senderData);
   
+  c_last_force_lcp_time = 0;
   return;
 }//Dblqh::execSIZEALT_REP()
 
@@ -10233,7 +10234,8 @@
 {
   jamEntry();
   CRASH_INSERTION(5010);
-  LcpFragOrd * const lcpFragOrd = (LcpFragOrd *)&signal->theData[0];
+  LcpFragOrd copy = *(LcpFragOrd *)&signal->theData[0];
+  LcpFragOrd * lcpFragOrd = &copy;
   Uint32 lcpId = lcpFragOrd->lcpId;
 
   lcpPtr.i = 0;
@@ -11484,9 +11486,12 @@
   UintR tsltStartMbyte;
   UintR tsltIndex;
   UintR tsltFlag;
+  UintR TchangeMbyte = 0;
 
   for (sltLogPartPtr.i = 0; sltLogPartPtr.i < 4; sltLogPartPtr.i++) {
     jam();
+    TchangeMbyte = 0;
+retry:
     ptrAss(sltLogPartPtr, logPartRecord);
     findLogfile(signal, sltLogPartPtr.p->logTailFileNo,
                 sltLogPartPtr, &sltLogFilePtr);
@@ -11592,7 +11597,8 @@
       sltLogPartPtr.p->logTailMbyte = 
         sltLogFilePtr.p->logLastPrepRef[tsltMbyte] & 65535;
       if ((ToldTailFileNo != sltLogPartPtr.p->logTailFileNo) ||
-          (ToldTailMByte != sltLogPartPtr.p->logTailMbyte)) {
+          (ToldTailMByte != sltLogPartPtr.p->logTailMbyte)) 
+      {
         jam();
         if (sltLogPartPtr.p->logPartState == LogPartRecord::TAIL_PROBLEM) {
           if (sltLogPartPtr.p->firstLogQueue == RNIL) {
@@ -11603,14 +11609,81 @@
             sltLogPartPtr.p->logPartState = LogPartRecord::ACTIVE;
           }//if
         }//if
-      }//if
+      }
+      else if (sltLogPartPtr.p->logPartState == LogPartRecord::TAIL_PROBLEM)
+      {
+        jam();
+
+        /**
+         * tail didnt move and we have tail problem 
+         * (i.e less than ZLOG_FREE_MB_LIMIT)
+         */
+        force_lcp(signal);
+
+        /**
+         * We already changed mbyte...skip to next logpart
+         */
+        if (TchangeMbyte)
+        {
+          jam();
+          goto next;
+        }
+        
+        /**
+         * head is not place where we would like to cut log
+         * skip to next
+         */
+        if (sltLogFilePtr.i != sltLogPartPtr.p->currentLogfile)
+        {
+          jam();
+          goto next;
+        }
+
+        TcConnectionrecPtr tmp;
+        tmp.i = sltLogPartPtr.p->firstLogTcrec;
+        if (tmp.i != RNIL) 
+        {
+          jam();
+          ptrCheckGuard(tmp, ctcConnectrecFileSize, tcConnectionrec);
+          Uint32 fileNo = tmp.p->logStartFileNo;
+          Uint32 mbyte = tmp.p->logStartPageNo >> ZTWOLOG_NO_PAGES_IN_MBYTE;
+          
+          if (fileNo == sltLogPartPtr.p->logTailFileNo &&
+              mbyte == sltLogPartPtr.p->logTailMbyte)
+          {
+            jam();
+            /**
+             * An uncommitted operation...still pending...
+             * skip to next
+             */
+            goto next;
+          }
+        }
+        
+        {
+          /**
+           * Try forcing a changeMbyte
+           */
+          jam();
+          logPartPtr = sltLogPartPtr;
+          logFilePtr = sltLogFilePtr;
+          logPagePtr.i = logFilePtr.p->currentLogpage;
+          ptrCheckGuard(logPagePtr, clogPageFileSize, logPageRecord);
+          
+          changeMbyte(signal);
+          TchangeMbyte = 1;
+          goto retry;
+        }
+      }
+  next:
+      (void)1;
     }
 #if 0
     infoEvent("setLogTail: Available log file %d size = %d[mbytes]+%d[words]", sltLogPartPtr.i,
 	      remainingLogSize(sltCurrLogFilePtr, sltLogPartPtr), sltCurrLogFilePtr.p->remainingWordsInMbyte);
 #endif
   }//for
-
+  
 }//Dblqh::setLogTail()
 
 /* ######################################################################### */
@@ -18647,8 +18720,18 @@
     twnlNextMbyte = logFilePtr.p->currentMbyte + 1;
     twnlNextFileNo = logFilePtr.p->fileNo;
   }//if
-  if (twnlNextFileNo == logPartPtr.p->logTailFileNo) {
-    if (logPartPtr.p->logTailMbyte == twnlNextMbyte) {
+
+
+  /**
+   * Leave 16Mb free in REDO
+   */
+  LogFileRecordPtr TfilePtr = logFilePtr;
+  Uint32 Tmbyte = logFilePtr.p->currentMbyte;
+  for (Uint32 i = 0; i<16; i++)
+  {
+    if (TfilePtr.p->fileNo == logPartPtr.p->logTailFileNo &&
+        Tmbyte == logPartPtr.p->logTailMbyte)
+    {
       jam();
 /* -------------------------------------------------- */
 /*       THE NEXT MBYTE WILL BE THE TAIL. WE MUST     */
@@ -18660,9 +18743,67 @@
 /*       _ROUND SIGNAL ARRIVING FROM DBDIH.           */
 /* -------------------------------------------------- */
       logPartPtr.p->logPartState = LogPartRecord::TAIL_PROBLEM;
-    }//if
-  }//if
+      force_lcp(signal); // Start a LCP
+      return;
+    }
+    
+    Tmbyte++;
+    if (Tmbyte == ZNO_MBYTES_IN_FILE)
+    {
+      jam();
+      Tmbyte = 0;
+      TfilePtr.i = TfilePtr.p->nextLogFile;
+      ptrCheckGuard(TfilePtr, clogFileFileSize, logFileRecord);
+    }
+  }
+
+  /**
+   * Force a LCP if less than 32Mb free
+   */
+  for (Uint32 i = 0; i<16; i++)
+  {
+    if (TfilePtr.p->fileNo == logPartPtr.p->logTailFileNo &&
+        Tmbyte == logPartPtr.p->logTailMbyte)
+    {
+      jam();
+      force_lcp(signal); // Start a LCP
+      return;
+    }
+    
+    Tmbyte++;
+    if (Tmbyte == ZNO_MBYTES_IN_FILE)
+    {
+      jam();
+      Tmbyte = 0;
+      TfilePtr.i = TfilePtr.p->nextLogFile;
+      ptrCheckGuard(TfilePtr, clogFileFileSize, logFileRecord);
+    }
+  }
 }//Dblqh::writeNextLog()
+
+void
+Dblqh::force_lcp(Signal* signal)
+{
+  if (cLqhTimeOutCount == c_last_force_lcp_time)
+  {
+    jam();
+    return;
+  }
+  
+  c_last_force_lcp_time = cLqhTimeOutCount;
+  signal->theData[0] = 7099;
+  for (Uint32 i = 0; i < cnoOfNodes; i++) 
+  {
+    jam();
+    Uint32 nodeId = cnodeData[i];
+    if(cnodeStatus[i] == ZNODE_UP)
+    {
+      jam();
+      BlockReference Tblockref = calcDihBlockRef(nodeId);
+      sendSignal(Tblockref, GSN_DUMP_STATE_ORD, signal, 1, JBB);
+    }
+  }
+}
 
 bool
 Dblqh::validate_filter(Signal* signal)
New file: storage/ndb/src/kernel/blocks/dblqh/DblqhMain.o