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:
  Send fragrecord ptr to initOpConnection so that is does not
  have to look it up every time (if needed)
  
  Make operation list on fragment fifo so that uncommitted operations
    are undo-logged in correct order
ndb/src/kernel/blocks/dbtup/DbtupAbort.cpp:
  Send fragptr to initOpConnection
ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp:
  dllist -> dlfifolist
ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp:
  dllist -> dlfifolist
ndb/src/kernel/blocks/dbtup/DbtupGen.cpp:
  Make operation list on fragment fifo so that uncommitted operations
    are undo-logged in correct order
ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp:
  Fix uninitialized variable c_start.m_startTime
    which made startPartial and startPartitioned fail
ndb/test/ndbapi/testSystemRestart.cpp:
  Added testSystemRestart -n SR9 which
    demonstrates that two prepared operation on same
    records _used to_ be undo-logged in wrong order makeing
    system restart fail
ndb/test/run-test/daily-devel-tests.txt:
  Add
  * testSystemRestart -n SR6
    Restart while some nodes have fs and other not
  * testSystemRestart -n SR7
    Restart in partition win
  * testSystemRestart -n SR8
    Restart in partition win, others starting during restart
  * testSystemRestart -n SR9
    Multiple ops on same record prepared before

6 files changed, 27 insertions, 28 deletions

diff --git a/ndb/src/kernel/blocks/dbtup/Dbtup.hpp b/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
index ce81c1c9bc8..0e8dd5fbbe8 100644
--- a/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
+++ b/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*);
 
 //------------------------------------------------------------------
 //------------------------------------------------------------------
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupAbort.cpp b/ndb/src/kernel/blocks/dbtup/DbtupAbort.cpp
index 1ffc5f06754..e9043a8b52d 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupAbort.cpp
+++ b/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();  
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp b/ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp
index fa3667b221e..cbd56c3281f 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp
+++ b/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
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp b/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
index dfd1e37d4f5..0061ebe812d 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
+++ b/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;
-  regFragPtr->firstusedOprec = regOperPtr.i;
-  regOperPtr.p->nextOprecInList = sopTmpOperPtr.i;
-  if (sopTmpOperPtr.i == RNIL) {
-    return;
+  regOperPtr.p->prevOprecInList = tail;
+  regOperPtr.p->nextOprecInList = RNIL;
+  sopTmpOperPtr.i = tail;
+  if (tail == RNIL) {
+    regFragPtr->firstusedOprec = regOperPtr.i;
   } else {
     jam();
     ptrCheckGuard(sopTmpOperPtr, cnoOfOprec, operationrec);
-    sopTmpOperPtr.p->prevOprecInList = regOperPtr.i;
+    sopTmpOperPtr.p->nextOprecInList = regOperPtr.i;
   }//if
+  regFragPtr->lastusedOprec = regOperPtr.i;
 }//Dbtup::linkOpIntoFragList()
 
 /*
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp b/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
index f3391ff7b59..d33adcd08e1 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
+++ b/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;
diff --git a/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp b/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp
index ff4876b1506..568ed6c6566 100644
--- a/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp
+++ b/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,13 +446,17 @@ 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);
-
+  
   UpgradeStartup::sendCmAppChg(* this, signal, 0); // ADD
   
   sendCntrStartReq(signal);
+
+  signal->theData[0] = ZSTARTUP;
+  sendSignalWithDelay(reference(), GSN_CONTINUEB, signal, 1000, 1);
   
   return;
 }