Merge mysql.com:/home/jonas/src/bug-4586

into mysql.com:/home/jonas/src/mysql-4.1-ndb

ndb/src/kernel/blocks/dbacc/Dbacc.hpp

@@ -1192,6 +1192,7 @@ private:
 
   void reportMemoryUsage(Signal* signal, int gth);
   void lcp_write_op_to_undolog(Signal* signal);
+  void reenable_expand_after_redo_log_exection_complete(Signal*);
 
 
   // Initialisation

ndb/src/kernel/blocks/dbacc/DbaccMain.cpp

@@ -2339,13 +2339,14 @@ void Dbacc::execACC_COMMITREQ(Signal* signal)
       fragrecptr.p->slack += operationRecPtr.p->insertDeleteLen;
       if (fragrecptr.p->slack > fragrecptr.p->slackCheck) {           /* TIME FOR JOIN BUCKETS PROCESS */
         if (fragrecptr.p->expandCounter > 0) {
-          if (fragrecptr.p->expandFlag == 0) {
+	  if (fragrecptr.p->expandFlag < 2) {
             jam();
-            fragrecptr.p->expandFlag = 1;
             signal->theData[0] = fragrecptr.i;
             signal->theData[1] = fragrecptr.p->p;
             signal->theData[2] = fragrecptr.p->maxp;
-            sendSignal(cownBlockref, GSN_SHRINKCHECK2, signal, 3, JBB);
+	    signal->theData[3] = fragrecptr.p->expandFlag;
+            fragrecptr.p->expandFlag = 2;
+            sendSignal(cownBlockref, GSN_SHRINKCHECK2, signal, 4, JBB);
           }//if
         }//if
       }//if
@@ -2359,7 +2360,7 @@ void Dbacc::execACC_COMMITREQ(Signal* signal)
     if (fragrecptr.p->slack >= (Uint32)(1 << 31)) { /* IT MEANS THAT IF SLACK < ZERO */
       if (fragrecptr.p->expandFlag == 0) {
         jam();
-        fragrecptr.p->expandFlag = 1;
+        fragrecptr.p->expandFlag = 2;
         signal->theData[0] = fragrecptr.i;
         signal->theData[1] = fragrecptr.p->p;
         signal->theData[2] = fragrecptr.p->maxp;
@@ -6331,9 +6332,16 @@ Uint32 Dbacc::checkScanExpand(Signal* signal)
 
 void Dbacc::execEXPANDCHECK2(Signal* signal) 
 {
+  jamEntry();
+
+  if(refToBlock(signal->getSendersBlockRef()) == DBLQH){
+    jam();
+    reenable_expand_after_redo_log_exection_complete(signal);
+    return;
+  }
+
   DirectoryarrayPtr newDirptr;
 
-  jamEntry();
   fragrecptr.i = signal->theData[0];
   tresult = 0;	/* 0= FALSE,1= TRUE,> ZLIMIT_OF_ERROR =ERRORCODE */
   Uint32 tmp = 1;
@@ -6390,6 +6398,7 @@ void Dbacc::execEXPANDCHECK2(Signal* signal)
       return;
     }//if
   }//if
+
   /*--------------------------------------------------------------------------*/
   /*       WE START BY FINDING THE PAGE, THE PAGE INDEX AND THE PAGE DIRECTORY*/
   /*       OF THE NEW BUCKET WHICH SHALL RECEIVE THE ELEMENT WHICH HAVE A 1 IN*/
@@ -6443,6 +6452,7 @@ void Dbacc::execEXPANDCHECK2(Signal* signal)
   } else {
     ptrCheckGuard(expPageptr, cpagesize, page8);
   }//if
+
   fragrecptr.p->expReceivePageptr = expPageptr.i;
   fragrecptr.p->expReceiveIndex = texpReceivedBucket & ((1 << fragrecptr.p->k) - 1);
   /*--------------------------------------------------------------------------*/
@@ -6493,7 +6503,7 @@ void Dbacc::endofexpLab(Signal* signal)
     /*       IT IS STILL NECESSARY TO EXPAND THE FRAGMENT EVEN MORE. START IT FROM HERE  */
     /*       WITHOUT WAITING FOR NEXT COMMIT ON THE FRAGMENT.                            */
     /* --------------------------------------------------------------------------------- */
-    fragrecptr.p->expandFlag = 1;
+    fragrecptr.p->expandFlag = 2;
     signal->theData[0] = fragrecptr.i;
     signal->theData[1] = fragrecptr.p->p;
     signal->theData[2] = fragrecptr.p->maxp;
@@ -6502,6 +6512,47 @@ void Dbacc::endofexpLab(Signal* signal)
   return;
 }//Dbacc::endofexpLab()
 
+void Dbacc::reenable_expand_after_redo_log_exection_complete(Signal* signal){
+
+  tabptr.i = signal->theData[0];
+  Uint32 fragId = signal->theData[1];
+
+  ptrCheckGuard(tabptr, ctablesize, tabrec);
+  ndbrequire(getrootfragmentrec(signal, rootfragrecptr, fragId));
+#if 0
+  ndbout_c("reenable expand check for table %d fragment: %d", 
+	   tabptr.i, fragId);
+#endif
+
+  for (Uint32 i = 0; i < 2; i++) {
+    fragrecptr.i = rootfragrecptr.p->fragmentptr[i];
+    ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
+    switch(fragrecptr.p->expandFlag){
+    case 0:
+      /**
+       * Hmm... this means that it's alreay has been reenabled...
+       */
+      ndbassert(false);
+      continue;
+    case 1:
+      /**
+       * Nothing is going on start expand check
+       */
+    case 2:
+      /**
+       * A shrink is running, do expand check anyway
+       *  (to reset expandFlag)
+       */
+      fragrecptr.p->expandFlag = 2; 
+      signal->theData[0] = fragrecptr.i;
+      signal->theData[1] = fragrecptr.p->p;
+      signal->theData[2] = fragrecptr.p->maxp;
+      sendSignal(cownBlockref, GSN_EXPANDCHECK2, signal, 3, JBB);
+      break;
+    }
+  }
+}
+
 void Dbacc::execDEBUG_SIG(Signal* signal) 
 {
   jamEntry();
@@ -6897,9 +6948,10 @@ void Dbacc::execSHRINKCHECK2(Signal* signal)
 
   jamEntry();
   fragrecptr.i = signal->theData[0];
-  tresult = 0;	/* 0= FALSE,1= TRUE,> ZLIMIT_OF_ERROR =ERRORCODE */
+  Uint32 oldFlag = signal->theData[3];
   ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
-  fragrecptr.p->expandFlag = 0;
+  fragrecptr.p->expandFlag = oldFlag;
+  tresult = 0;	/* 0= FALSE,1= TRUE,> ZLIMIT_OF_ERROR =ERRORCODE */
   if (fragrecptr.p->slack <= fragrecptr.p->slackCheck) {
     jam();
     /* TIME FOR JOIN BUCKETS PROCESS */
@@ -6980,6 +7032,7 @@ void Dbacc::execSHRINKCHECK2(Signal* signal)
     jam();
     fragrecptr.p->p--;
   }//if
+
   /*--------------------------------------------------------------------------*/
   /*       WE START BY FINDING THE NECESSARY INFORMATION OF THE BUCKET TO BE  */
   /*       REMOVED WHICH WILL SEND ITS ELEMENTS TO THE RECEIVING BUCKET.      */
@@ -7172,11 +7225,13 @@ void Dbacc::endofshrinkbucketLab(Signal* signal)
 	/*       SHRINKING BELOW 2^K - 1 (NOW 63). THIS WAS A BUG THAT  */
 	/*       WAS REMOVED 2000-05-12.                                */
 	/*--------------------------------------------------------------*/
-        fragrecptr.p->expandFlag = 1;
         signal->theData[0] = fragrecptr.i;
         signal->theData[1] = fragrecptr.p->p;
         signal->theData[2] = fragrecptr.p->maxp;
-        sendSignal(cownBlockref, GSN_SHRINKCHECK2, signal, 3, JBB);
+        signal->theData[3] = fragrecptr.p->expandFlag;
+	ndbrequire(fragrecptr.p->expandFlag < 2);
+        fragrecptr.p->expandFlag = 2;
+        sendSignal(cownBlockref, GSN_SHRINKCHECK2, signal, 4, JBB);
       }//if
     }//if
   }//if
@@ -9192,7 +9247,14 @@ void Dbacc::initFragAdd(Signal* signal,
   ndbrequire(req->kValue == 6);
   regFragPtr.p->k = req->kValue;	/* TK_SIZE = 6 IN THIS VERSION */
   regFragPtr.p->expandCounter = 0;
-  regFragPtr.p->expandFlag = 0;
+
+  /**
+   * Only allow shrink during SR
+   *   - to make sure we don't run out of pages during REDO log execution
+   *
+   * Is later restored to 0 by LQH at end of REDO log execution
+   */
+  regFragPtr.p->expandFlag = (getNodeState().getSystemRestartInProgress()?1:0);
   regFragPtr.p->p = 0;
   regFragPtr.p->maxp = (1 << req->kValue) - 1;
   regFragPtr.p->minloadfactor = minLoadFactor;

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

@@ -2764,6 +2764,11 @@ private:
 /* ------------------------------------------------------------------------- */
   UintR cfirstCompletedFragSr;
 
+  /**
+   * List of fragment that the log execution is completed for
+   */
+  Uint32 c_redo_log_complete_frags;
+  
 /* ------------------------------------------------------------------------- */
 /*USED DURING SYSTEM RESTART, INDICATES THE OLDEST GCI THAT CAN BE RESTARTED */
 /*FROM AFTER THIS SYSTEM RESTART. USED TO FIND THE LOG TAIL.                 */

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

@@ -65,6 +65,7 @@ void Dblqh::initData()
   cLqhTimeOutCount = 0;
   cLqhTimeOutCheckCount = 0;
   cbookedAccOps = 0;
+  c_redo_log_complete_frags = RNIL;
 }//Dblqh::initData()
 
 void Dblqh::initRecords() 

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

@@ -13642,14 +13642,22 @@ void Dblqh::execSTART_EXEC_SR(Signal* signal)
        *  ALSO SEND START_FRAGCONF TO DIH AND SET THE STATE TO ACTIVE ON THE
        *  FRAGMENT.
        * ------------------------------------------------------------------- */
+      Uint32 next = fragptr.p->nextFrag;
       if (prevFragptr.i != RNIL) {
         jam();
         ptrCheckGuard(prevFragptr, cfragrecFileSize, fragrecord);
-        prevFragptr.p->nextFrag = fragptr.p->nextFrag;
+        prevFragptr.p->nextFrag = next;
       } else {
         jam();
-        cfirstCompletedFragSr = fragptr.p->nextFrag;
+        cfirstCompletedFragSr = next;
       }//if
+
+      /**
+       * Put fragment on list which has completed REDO log
+       */
+      fragptr.p->nextFrag = c_redo_log_complete_frags;
+      c_redo_log_complete_frags = fragptr.i;
+      
       fragptr.p->fragStatus = Fragrecord::FSACTIVE;
       fragptr.p->logFlag = Fragrecord::STATE_TRUE;
       signal->theData[0] = fragptr.p->srUserptr;
@@ -13661,7 +13669,7 @@ void Dblqh::execSTART_EXEC_SR(Signal* signal)
        *  THIS IS PERFORMED BY KEEPING PREV_FRAGPTR AS PREV_FRAGPTR BUT MOVING
        *  FRAGPTR TO THE NEXT FRAGMENT IN THE LIST.
        * ------------------------------------------------------------------- */
-      fragptr.i = fragptr.p->nextFrag;
+      fragptr.i = next;
     }//if
     signal->theData[0] = fragptr.i;
     signal->theData[1] = prevFragptr.i;
@@ -15267,6 +15275,17 @@ void Dblqh::srFourthComp(Signal* signal)
     conf->startingNodeId = getOwnNodeId();
     sendSignal(cmasterDihBlockref, GSN_START_RECCONF, signal, 
 	       StartRecConf::SignalLength, JBB);
+
+    if(cstartType == NodeState::ST_SYSTEM_RESTART){
+      fragptr.i = c_redo_log_complete_frags;
+      while(fragptr.i != RNIL){
+	ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
+	signal->theData[0] = fragptr.p->tabRef;
+	signal->theData[1] = fragptr.p->fragId;
+	sendSignal(DBACC_REF, GSN_EXPANDCHECK2, signal, 2, JBB);
+	fragptr.i = fragptr.p->nextFrag;
+      }
+    }
   } else {
     ndbrequire(false);
   }//if