Commit b61bfd8c authored by unknown's avatar unknown
Browse files

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
parent 300346fa

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

+2 −1
Original line number Diff line number Diff line
@@ -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

+3 −3
Original line number Diff line number Diff line
@@ -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

+6 −9
Original line number Diff line number Diff line
@@ -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

+8 −12
Original line number Diff line number Diff line
@@ -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

+2 −1
Original line number Diff line number Diff line
@@ -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;