Make operation list on fragment fifo so that uncommitted operations

  are undo-logged in correct order

Add bunch of testcases to autotest

ndb/src/kernel/blocks/dbtup/Dbtup.hpp

@@ -515,6 +515,7 @@ struct Fragrecord {
   Uint32 emptyPrimPage;
 
   Uint32 firstusedOprec;
+  Uint32 lastusedOprec;
 
   Uint32 thFreeFirst;
   Uint32 thFreeCopyFirst;
@@ -1653,7 +1654,7 @@ private:
 
 //------------------------------------------------------------------
 //------------------------------------------------------------------
-  void initOpConnection(Operationrec* const regOperPtr);
+  void initOpConnection(Operationrec* regOperPtr, Fragrecord*);
 
 //------------------------------------------------------------------
 //------------------------------------------------------------------

ndb/src/kernel/blocks/dbtup/DbtupAbort.cpp

@@ -77,7 +77,7 @@ void Dbtup::execTUP_ABORTREQ(Signal* signal)
   if (regOperPtr.p->optype == ZREAD) {
     ljam();
     freeAllAttrBuffers(regOperPtr.p);
-    initOpConnection(regOperPtr.p);
+    initOpConnection(regOperPtr.p, 0);
     return;
   }//if
 
@@ -134,7 +134,7 @@ void Dbtup::execTUP_ABORTREQ(Signal* signal)
     ndbrequire(regOperPtr.p->tupleState == ALREADY_ABORTED);
     commitUpdate(signal, regOperPtr.p, regFragPtr.p, regTabPtr.p);
   }//if
-  initOpConnection(regOperPtr.p);
+  initOpConnection(regOperPtr.p, regFragPtr.p);
 }//execTUP_ABORTREQ()
 
 void Dbtup::setTupleStateOnPreviousOps(Uint32 prevOpIndex)
@@ -459,7 +459,7 @@ void Dbtup::tupkeyErrorLab(Signal* signal)
   freeAllAttrBuffers(regOperPtr);
   abortUpdate(signal, regOperPtr, fragptr.p, tabptr.p);
   removeActiveOpList(regOperPtr);
-  initOpConnection(regOperPtr);
+  initOpConnection(regOperPtr, fragptr.p);
   regOperPtr->transstate = IDLE;
   regOperPtr->tupleState = NO_OTHER_OP;
   TupKeyRef * const tupKeyRef = (TupKeyRef *)signal->getDataPtrSend();  

ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp

@@ -224,7 +224,8 @@ void Dbtup::removeActiveOpList(Operationrec*  const regOperPtr)
 /* ---------------------------------------------------------------- */
 /* INITIALIZATION OF ONE CONNECTION RECORD TO PREPARE FOR NEXT OP.  */
 /* ---------------------------------------------------------------- */
-void Dbtup::initOpConnection(Operationrec* const regOperPtr)
+void Dbtup::initOpConnection(Operationrec* regOperPtr,
+			     Fragrecord * fragPtrP)
 {
   Uint32 RinFragList = regOperPtr->inFragList;
   regOperPtr->transstate = IDLE;
@@ -244,22 +245,18 @@ void Dbtup::initOpConnection(Operationrec* const regOperPtr)
     regOperPtr->inFragList = ZFALSE;
     if (tropPrevLinkPtr.i == RNIL) {
       ljam();
-      FragrecordPtr regFragPtr;
-      regFragPtr.i = regOperPtr->fragmentPtr;
-      ptrCheckGuard(regFragPtr, cnoOfFragrec, fragrecord);
-      regFragPtr.p->firstusedOprec = tropNextLinkPtr.i;
+      fragPtrP->firstusedOprec = tropNextLinkPtr.i;
     } else {
       ljam();
       ptrCheckGuard(tropPrevLinkPtr, cnoOfOprec, operationrec);
       tropPrevLinkPtr.p->nextOprecInList = tropNextLinkPtr.i;
     }//if
     if (tropNextLinkPtr.i == RNIL) {
-      ;
+      fragPtrP->lastusedOprec = tropPrevLinkPtr.i;
     } else {
-      ljam();
       ptrCheckGuard(tropNextLinkPtr, cnoOfOprec, operationrec);
       tropNextLinkPtr.p->prevOprecInList = tropPrevLinkPtr.i;
-    }//if
+    }
     regOperPtr->prevOprecInList = RNIL;
     regOperPtr->nextOprecInList = RNIL;
   }//if
@@ -336,7 +333,7 @@ void Dbtup::execTUP_COMMITREQ(Signal* signal)
     commitUpdate(signal, regOperPtr.p, regFragPtr.p, regTabPtr.p);
     removeActiveOpList(regOperPtr.p);
   }//if
-  initOpConnection(regOperPtr.p);
+  initOpConnection(regOperPtr.p, regFragPtr.p);
 }//execTUP_COMMITREQ()
 
 void

ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp

@@ -319,24 +319,20 @@ void Dbtup::linkOpIntoFragList(OperationrecPtr regOperPtr,
                                Fragrecord* const regFragPtr) 
 {
   OperationrecPtr sopTmpOperPtr;
-/* ----------------------------------------------------------------- */
-/*       LINK THE OPERATION INTO A DOUBLY LINKED LIST ON THE FRAGMENT*/
-/*       PUT IT FIRST IN THIS LIST SINCE IT DOESN'T MATTER WHERE IT  */
-/*       IS PUT.                                                     */
-/* ----------------------------------------------------------------- */
+  Uint32 tail = regFragPtr->lastusedOprec;
   ndbrequire(regOperPtr.p->inFragList == ZFALSE);
   regOperPtr.p->inFragList = ZTRUE;
-  regOperPtr.p->prevOprecInList = RNIL;
-  sopTmpOperPtr.i = regFragPtr->firstusedOprec;
+  regOperPtr.p->prevOprecInList = tail;
+  regOperPtr.p->nextOprecInList = RNIL;
+  sopTmpOperPtr.i = tail;
+  if (tail == RNIL) {
     regFragPtr->firstusedOprec = regOperPtr.i;
-  regOperPtr.p->nextOprecInList = sopTmpOperPtr.i;
-  if (sopTmpOperPtr.i == RNIL) {
-    return;
   } else {
     jam();
     ptrCheckGuard(sopTmpOperPtr, cnoOfOprec, operationrec);
-    sopTmpOperPtr.p->prevOprecInList = regOperPtr.i;
+    sopTmpOperPtr.p->nextOprecInList = regOperPtr.i;
   }//if
+  regFragPtr->lastusedOprec = regOperPtr.i;
 }//Dbtup::linkOpIntoFragList()
 
 /*

ndb/src/kernel/blocks/dbtup/DbtupGen.cpp

@@ -963,6 +963,7 @@ void Dbtup::initializeFragrecord()
     regFragPtr.p->nextfreefrag = regFragPtr.i + 1;
     regFragPtr.p->checkpointVersion = RNIL;
     regFragPtr.p->firstusedOprec = RNIL;
+    regFragPtr.p->lastusedOprec = RNIL;
     regFragPtr.p->fragStatus = IDLE;
   }//for
   regFragPtr.i = cnoOfFragrec - 1;
@@ -1164,7 +1165,7 @@ void Dbtup::execTUPSEIZEREQ(Signal* signal)
     return;
   }//if
   regOperPtr.p->optype = ZREAD;
-  initOpConnection(regOperPtr.p);
+  initOpConnection(regOperPtr.p, 0);
   regOperPtr.p->userpointer = userPtr;
   regOperPtr.p->userblockref = userRef;
   signal->theData[0] = regOperPtr.p->userpointer;

ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp

@@ -103,7 +103,7 @@ void Ndbcntr::execCONTINUEB(Signal* signal)
     }
     
     Uint64 now = NdbTick_CurrentMillisecond();
-    if(c_start.m_startFailureTimeout > now){
+    if(now > c_start.m_startFailureTimeout){
       ndbrequire(false);
     }
     
@@ -446,6 +446,7 @@ void Ndbcntr::execREAD_NODESCONF(Signal* signal)
   ndb_mgm_get_int_parameter(p, CFG_DB_START_PARTITION_TIMEOUT, &to_2);
   ndb_mgm_get_int_parameter(p, CFG_DB_START_FAILURE_TIMEOUT, &to_3);
   
+  c_start.m_startTime = NdbTick_CurrentMillisecond();
   c_start.m_startPartialTimeout = setTimeout(c_start.m_startTime, to_1);
   c_start.m_startPartitionedTimeout = setTimeout(c_start.m_startTime, to_2);
   c_start.m_startFailureTimeout = setTimeout(c_start.m_startTime, to_3);
@@ -454,6 +455,9 @@ void Ndbcntr::execREAD_NODESCONF(Signal* signal)
   
   sendCntrStartReq(signal);
 
+  signal->theData[0] = ZSTARTUP;
+  sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 1000, 1);
+  
   return;
 }
 

ndb/test/ndbapi/testSystemRestart.cpp

@@ -872,7 +872,7 @@ int runSystemRestart7(NDBT_Context* ctx, NDBT_Step* step){
 
   const Uint32 nodeCount = restarter.getNumDbNodes();
   if(nodeCount < 2){
-    g_info << "SR8 - Needs atleast 2 nodes to test" << endl;
+    g_info << "SR7 - Needs atleast 2 nodes to test" << endl;
     return NDBT_OK;
   }
 
@@ -1001,7 +1001,52 @@ int runSystemRestart8(NDBT_Context* ctx, NDBT_Step* step){
     i++;
   }
   
-  g_info << "runSystemRestart7 finished" << endl;  
+  g_info << "runSystemRestart8 finished" << endl;  
+
+  return result;
+}
+
+int runSystemRestart9(NDBT_Context* ctx, NDBT_Step* step){
+  Ndb* pNdb = GETNDB(step);
+  int result = NDBT_OK;
+  int timeout = 300;
+  Uint32 loops = ctx->getNumLoops();
+  int records = ctx->getNumRecords();
+  NdbRestarter restarter;
+  Uint32 i = 1;
+  
+  Uint32 currentRestartNodeIndex = 1;
+  UtilTransactions utilTrans(*ctx->getTab());
+  HugoTransactions hugoTrans(*ctx->getTab());
+
+  int args[] = { DumpStateOrd::DihMaxTimeBetweenLCP };
+  int dump[] = { DumpStateOrd::DihStartLcpImmediately };
+  
+  do {
+    CHECK(restarter.dumpStateAllNodes(args, 1) == 0);
+    
+    HugoOperations ops(* ctx->getTab());
+    CHECK(ops.startTransaction(pNdb) == 0);
+    for(i = 0; i<10; i++){
+      CHECK(ops.pkInsertRecord(pNdb, i, 1, 1) == 0);
+      CHECK(ops.execute_NoCommit(pNdb) == 0);
+    }
+    for(i = 0; i<10; i++){
+      CHECK(ops.pkUpdateRecord(pNdb, i, 1) == 0);
+      CHECK(ops.execute_NoCommit(pNdb) == 0);
+    }
+    NdbSleep_SecSleep(10);
+    CHECK(restarter.dumpStateAllNodes(dump, 1) == 0);
+    NdbSleep_SecSleep(10);
+    CHECK(ops.execute_Commit(pNdb) == 0);  
+    
+    CHECK(restarter.restartAll() == 0);
+    CHECK(restarter.waitClusterStarted(timeout) == 0);
+    CHECK(pNdb->waitUntilReady(timeout) == 0);
+    ops.closeTransaction(pNdb);
+  } while(0);
+  
+  g_info << "runSystemRestart9 finished" << endl;  
 
   return result;
 }
@@ -1176,6 +1221,19 @@ TESTCASE("SR8",
   STEP(runSystemRestart8);
   FINALIZER(runClearTable);
 }
+TESTCASE("SR9", 
+	 "Perform partition win system restart with other nodes delayed\n"
+	 "* 1. Start transaction\n"
+	 "* 2. insert (1,1)\n"
+	 "* 3. update (1,2)\n"
+	 "* 4. start lcp\n"
+	 "* 5. commit\n"
+	 "* 6. restart\n"){
+  INITIALIZER(runWaitStarted);
+  INITIALIZER(runClearTable);
+  STEP(runSystemRestart9);
+  FINALIZER(runClearTable);
+}
 NDBT_TESTSUITE_END(testSystemRestart);
 
 int main(int argc, const char** argv){

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

@@ -190,3 +190,19 @@ max-time: 2500
 cmd: test_event
 args: -n BasicEventOperation T1 T6 
 
+#
+max-time: 1500
+cmd: testSystemRestart
+args: -n SR6 -l 1 T1 
+
+max-time: 1500
+cmd: testSystemRestart
+args: -n SR7 -l 1 T1 
+
+max-time: 1500
+cmd: testSystemRestart
+args: -n S8 -l 1 T1 
+
+max-time: 1500
+cmd: testSystemRestart
+args: -n S9 -l 1 T1