dbtupsystemrestart.cpp

mysql-5.0.22.tar.gz源码包

      ljam();
      xlcIndicateNoOpActive(xlcStruct);
      break;
    case ZLCPR_TYPE_UPDATE_GCI:
      ljam();
      xlcUpdateGCI(xlcStruct);
      break;
    default:
      ndbrequire(false);
      break;
    }//switch
  } while (1);
}//Dbtup::xlcGetNextRecordLab()

/* ---------------------------------------------------------------- */
/* -----------------   XLC_GET_LOG_HEADER    ---------------------- */
/* ---------------------------------------------------------------- */
void Dbtup::xlcGetLogHeader(XlcStruct& xlcStruct) 
{
  Uint32 pIndex = xlcStruct.PageIndex;
  Uint32 fragId;
  Uint32 tableId;
  Uint32 prevId;
  if ((pIndex + 4) < ZWORDS_ON_PAGE) {
    UndoPage* const regUndoPagePtr = xlcStruct.UPPtr.p;
    ljam();
    xlcStruct.LogRecordType = regUndoPagePtr->undoPageWord[pIndex];
    prevId = regUndoPagePtr->undoPageWord[pIndex + 1];
    tableId = regUndoPagePtr->undoPageWord[pIndex + 2];
    fragId = regUndoPagePtr->undoPageWord[pIndex + 3];
    xlcStruct.PageIndex = pIndex + 4;
  } else {
    ljam();
    xlcStruct.LogRecordType = xlcGetLogWord(xlcStruct);
    prevId = xlcGetLogWord(xlcStruct);
    tableId = xlcGetLogWord(xlcStruct);
    fragId = xlcGetLogWord(xlcStruct);
  }//if
  xlcStruct.LliPtr.p->lliPrevRecordId = prevId;
  xlcStruct.FragId = fragId;
  xlcStruct.TabPtr.i = tableId;
  ptrCheckGuard(xlcStruct.TabPtr, cnoOfTablerec, tablerec);
}//Dbtup::xlcGetLogHeader()

/* ------------------------------------------------------------------- */
/* ---------------------- XLC_GET_LOG_WORD --------------------------- */
/* ------------------------------------------------------------------- */
Uint32 Dbtup::xlcGetLogWord(XlcStruct& xlcStruct) 
{
  Uint32 pIndex = xlcStruct.PageIndex;
  ndbrequire(xlcStruct.UPPtr.p != NULL);
  ndbrequire(pIndex < ZWORDS_ON_PAGE);
  Uint32 logWord = xlcStruct.UPPtr.p->undoPageWord[pIndex];
  pIndex++;
  xlcStruct.PageIndex = pIndex;
  if (pIndex == ZWORDS_ON_PAGE) {
    ljam();
    xlcStruct.PageIndex = ZUNDO_PAGE_HEADER_SIZE;
    xlcStruct.PageId++;
    if ((xlcStruct.PageId - xlcStruct.DbsiPtr.p->pdxFilePage) >= (2 * ZUB_SEGMENT_SIZE)) {
      ljam();
      xlcStruct.UPPtr.i = RNIL;
      ptrNull(xlcStruct.UPPtr);
    } else {
      ljam();
      Uint32 index = xlcStruct.PageId - xlcStruct.DbsiPtr.p->pdxFilePage;
      ndbrequire(index < 16);
      xlcStruct.UPPtr.i = xlcStruct.DbsiPtr.p->pdxDataPage[index];
      ptrCheckGuard(xlcStruct.UPPtr, cnoOfUndoPage, undoPage);
    }//if
  }//if
  return logWord;
}//Dbtup::xlcGetLogWord()

   /****************************************************/
   /* INSERT A TUPLE HEADER THE DATA IS THE TUPLE DATA */
   /****************************************************/
void Dbtup::xlcInsertTh(XlcStruct& xlcStruct) 
{
  PagePtr pagePtr;
  Fragrecord* const regFragPtr = xlcStruct.FragPtr.p;
  Tablerec* const regTabPtr = xlcStruct.TabPtr.p;

  Uint32 fragPageId = xlcGetLogWord(xlcStruct);
  Uint32 pageIndex = xlcGetLogWord(xlcStruct);
  ndbrequire((pageIndex & 1) == 0);
  pagePtr.i = getRealpid(regFragPtr, fragPageId);
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 pageOffset;
  getThAtPageSr(pagePtr.p, pageOffset);
  ndbrequire(pageOffset == (ZPAGE_HEADER_SIZE + (regTabPtr->tupheadsize * (pageIndex >> 1))));
  if (xlcStruct.LogRecordType == ZLCPR_TYPE_INSERT_TH) {
    ljam();
    xlcCopyData(xlcStruct, pageOffset, regTabPtr->tupheadsize, pagePtr);
  } else {
    ndbrequire(xlcStruct.LogRecordType == ZLCPR_TYPE_INSERT_TH_NO_DATA);
    ljam();
  }//if
/* ----------------------------------------*/
/* INDICATE THAT NO OPERATIONS ACTIVE      */
/* ----------------------------------------*/
  ndbrequire(pageOffset < ZWORDS_ON_PAGE);
  pagePtr.p->pageWord[pageOffset] = RNIL;
}//Dbtup::xlcInsertTh()

   /**********************************************/
   /* DELETE A TUPLE HEADER - NO ADDITIONAL DATA */
   /**********************************************/
void Dbtup::xlcDeleteTh(XlcStruct& xlcStruct) 
{
  PagePtr pagePtr;
  Fragrecord* const regFragPtr = xlcStruct.FragPtr.p;
  Tablerec* const regTabPtr = xlcStruct.TabPtr.p;

  Uint32 fragPageId = xlcGetLogWord(xlcStruct);
  Uint32 pageIndex = xlcGetLogWord(xlcStruct);
  ndbrequire((pageIndex & 1) == 0);
  pagePtr.i = getRealpid(regFragPtr, fragPageId);
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 pageOffset = ZPAGE_HEADER_SIZE + (regTabPtr->tupheadsize * (pageIndex >> 1));
  freeThSr(regTabPtr, pagePtr.p, pageOffset);
}//Dbtup::xlcDeleteTh()

   /*****************************************************/
   /* UPDATE A TUPLE HEADER, THE DATA IS THE TUPLE DATA */
   /*****************************************************/
void Dbtup::xlcUpdateTh(XlcStruct& xlcStruct) 
{
  PagePtr pagePtr;
  Fragrecord* const regFragPtr = xlcStruct.FragPtr.p;
  Tablerec* const regTabPtr = xlcStruct.TabPtr.p;

  Uint32 fragPageId = xlcGetLogWord(xlcStruct);
  Uint32 pageIndex = xlcGetLogWord(xlcStruct);
  ndbrequire((pageIndex & 1) == 0);
  pagePtr.i = getRealpid(regFragPtr, fragPageId);
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 pageOffset = ZPAGE_HEADER_SIZE + (regTabPtr->tupheadsize * (pageIndex >> 1));
  xlcCopyData(xlcStruct, pageOffset, regTabPtr->tupheadsize, pagePtr);
/* ----------------------------------------*/
/* INDICATE THAT NO OPERATIONS ACTIVE      */
/* ----------------------------------------*/
  ndbrequire(pageOffset < ZWORDS_ON_PAGE);
  pagePtr.p->pageWord[pageOffset] = RNIL;
}//Dbtup::xlcUpdateTh()

   /**************************************************/
   /* ABORT AN INSERT OPERATION - NO ADDITIONAL DATA */
   /**************************************************/
void Dbtup::xlcAbortInsert(Signal* signal, XlcStruct& xlcStruct) 
{
  PagePtr pagePtr;
  Fragrecord* const regFragPtr = xlcStruct.FragPtr.p;
  Tablerec* const regTabPtr = xlcStruct.TabPtr.p;

  Uint32 fragPageId = xlcGetLogWord(xlcStruct);
  Uint32 pageIndex = xlcGetLogWord(xlcStruct);
  ndbrequire((pageIndex & 1) == 0);
  pagePtr.i = getRealpid(regFragPtr, fragPageId);
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 pageOffset = ZPAGE_HEADER_SIZE + (regTabPtr->tupheadsize * (pageIndex >> 1));
  freeTh(regFragPtr, regTabPtr, signal, pagePtr.p, pageOffset);
}//Dbtup::xlcAbortInsert()

   /*****************************************************/
   /* COPY DATA FROM COPY TUPLE TO ORIGINAL TUPLE       */
   /*****************************************************/
void Dbtup::xlcAbortUpdate(Signal* signal, XlcStruct& xlcStruct) 
{
  PagePtr pagePtr;
  Fragrecord* const regFragPtr = xlcStruct.FragPtr.p;
  Tablerec* const regTabPtr = xlcStruct.TabPtr.p;
  Uint32 tuple_size = regTabPtr->tupheadsize;

  Uint32 fragPageIdC = xlcGetLogWord(xlcStruct);
  Uint32 pageIndexC = xlcGetLogWord(xlcStruct);
  ndbrequire((pageIndexC & 1) == 0);
  Uint32 TdestPageId = getRealpid(regFragPtr, fragPageIdC);
  Uint32 TcmDestIndex = ZPAGE_HEADER_SIZE +
                       (tuple_size * (pageIndexC >> 1));

  Uint32 fragPageId = xlcGetLogWord(xlcStruct);
  Uint32 pageIndex = xlcGetLogWord(xlcStruct);
  ndbrequire((pageIndex & 1) == 0);
  Uint32 TsourcePageId = getRealpid(regFragPtr, fragPageId);
  Uint32 TcmSourceIndex = ZPAGE_HEADER_SIZE +
                         (tuple_size * (pageIndex >> 1));
  Uint32 end_source = tuple_size + TcmSourceIndex;
  Uint32 end_dest = tuple_size + TcmDestIndex;

  void* Tdestination = (void*)&page[TdestPageId].pageWord[TcmDestIndex + 1]; 
  const void* Tsource = 
    (void*)&page[TsourcePageId].pageWord[TcmSourceIndex + 1];

  ndbrequire(TsourcePageId < cnoOfPage &&
             TdestPageId < cnoOfPage &&
             end_source <= ZWORDS_ON_PAGE &&
             end_dest <= ZWORDS_ON_PAGE);
  MEMCOPY_NO_WORDS(Tdestination, Tsource, (tuple_size - 1));

  pagePtr.i = TsourcePageId;
  ptrAss(pagePtr, page);
  freeTh(regFragPtr, regTabPtr, signal, pagePtr.p, TcmSourceIndex);

  pagePtr.i = TdestPageId;
  ptrAss(pagePtr, page);
  pagePtr.p->pageWord[TcmDestIndex] = RNIL;
}//Dbtup::xlcAbortUpdate()

   /*****************************/
   /* RESTORE UPDATED GCI VALUE */
   /*****************************/
void Dbtup::xlcUpdateGCI(XlcStruct& xlcStruct) 
{
  PagePtr pagePtr;
  Fragrecord* const regFragPtr = xlcStruct.FragPtr.p;
  Tablerec* const regTabPtr = xlcStruct.TabPtr.p;

  Uint32 fragPageId = xlcGetLogWord(xlcStruct);
  Uint32 pageIndex = xlcGetLogWord(xlcStruct);
  Uint32 restoredGCI = xlcGetLogWord(xlcStruct);

  ndbrequire((pageIndex & 1) == 0);
  pagePtr.i = getRealpid(regFragPtr, fragPageId);
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 pageOffset = ZPAGE_HEADER_SIZE + (regTabPtr->tupheadsize * (pageIndex >> 1));
  Uint32 gciOffset = pageOffset + regTabPtr->tupGCPIndex;
  ndbrequire((gciOffset < ZWORDS_ON_PAGE) &&
             (regTabPtr->tupGCPIndex < regTabPtr->tupheadsize));
  pagePtr.p->pageWord[gciOffset] = restoredGCI;
}//Dbtup::xlcUpdateGCI()

   /*****************************************************/
   /* READ TABLE DESCRIPTOR FROM UNDO LOG               */
   /*****************************************************/
void Dbtup::xlcTableDescriptor(XlcStruct& xlcStruct) 
{
  xlcStruct.LliPtr.p->lliNumFragments--;
  xlcStruct.FragPtr.p->fragStatus = ACTIVE;
}//Dbtup::xlcTableDescriptor()

   /********************************************************/
   /* UPDATE PAGE STATE AND NEXT POINTER IN PAGE           */
   /********************************************************/
void Dbtup::xlcUndoLogPageHeader(XlcStruct& xlcStruct) 
{
  Fragrecord* const regFragPtr = xlcStruct.FragPtr.p;
  PagePtr xlcPagep;

  Uint32 fragPageId = xlcGetLogWord(xlcStruct);
  xlcPagep.i = getRealpid(regFragPtr, fragPageId);
  ptrCheckGuard(xlcPagep, cnoOfPage, page);
  Uint32 logWord = xlcGetLogWord(xlcStruct);
  ndbrequire(logWord != 0);
  ndbrequire(logWord <= ZAC_MM_FREE_COPY);

  xlcPagep.p->pageWord[ZPAGE_STATE_POS] = logWord;
  xlcPagep.p->pageWord[ZPAGE_NEXT_POS] = xlcGetLogWord(xlcStruct);
}//Dbtup::xlcUndoLogPageHeader()

   /********************************************************/
   /* INDICATE THAT NO OPERATIONS ACTIVE                   */
   /********************************************************/
void Dbtup::xlcIndicateNoOpActive(XlcStruct& xlcStruct) 
{
  PagePtr pagePtr;
  Fragrecord* const regFragPtr = xlcStruct.FragPtr.p;
  Tablerec* const regTabPtr = xlcStruct.TabPtr.p;

  Uint32 fragPageId = xlcGetLogWord(xlcStruct);
  Uint32 pageIndex = xlcGetLogWord(xlcStruct);
  ndbrequire((pageIndex & 1) == 0);
  pagePtr.i = getRealpid(regFragPtr, fragPageId);
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 pageOffset = ZPAGE_HEADER_SIZE + (regTabPtr->tupheadsize * (pageIndex >> 1));
/* ----------------------------------------*/
/* INDICATE THAT NO OPERATIONS ACTIVE      */
/* ----------------------------------------*/
  ndbrequire(pageOffset < ZWORDS_ON_PAGE);
  pagePtr.p->pageWord[pageOffset] = RNIL;
}//Dbtup::xlcIndicateNoOpActive()

   /********************************************************/
   /* THIS IS THE COMMON ROUTINE TO COPY DATA FROM THE     */
   /* UNDO BUFFER TO THE DATA PAGES. IT USES THE           */
   /* XLC_REQUEST_SEGMENT SUB TO GET MORE DATA WHEN NEEDED */
   /********************************************************/
void Dbtup::xlcCopyData(XlcStruct& xlcStruct, Uint32 pageOffset, Uint32 noOfWords, PagePtr pagePtr) 
{
  ndbrequire((pageOffset + noOfWords - 1) < ZWORDS_ON_PAGE);
  for (Uint32 i = 1; i < noOfWords; i++) {
    ljam();
    pagePtr.p->pageWord[pageOffset + i] = xlcGetLogWord(xlcStruct);
  }//for
}//Dbtup::xlcCopyData()

void Dbtup::allocRestartUndoBufferSegment(Signal* signal, DiskBufferSegmentInfoPtr& dbsiPtr, LocalLogInfoPtr lliPtr) 
{
  UndoPagePtr undoPagePtr;

  ndbrequire(cfirstfreeUndoSeg != RNIL);
  if (cnoFreeUndoSeg == ZMIN_PAGE_LIMIT_TUP_COMMITREQ) {
    EXECUTE_DIRECT(DBLQH, GSN_TUP_COM_BLOCK, signal, 1);
    ljamEntry();
  }//if
  cnoFreeUndoSeg--;
  ndbrequire(cnoFreeUndoSeg >= 0);
  undoPagePtr.i = cfirstfreeUndoSeg;
  ptrCheckGuard(undoPagePtr, cnoOfUndoPage, undoPage);
  cfirstfreeUndoSeg = undoPagePtr.p->undoPageWord[ZPAGE_NEXT_POS];
  undoPagePtr.p->undoPageWord[ZPAGE_NEXT_POS] = RNIL;
  seizeDiskBufferSegmentRecord(dbsiPtr);
  dbsiPtr.p->pdxBuffertype = UNDO_RESTART_PAGES;
  dbsiPtr.p->pdxUndoBufferSet[0] = undoPagePtr.i;
  Uint32 i;
  for (i = 0; i < ZUB_SEGMENT_SIZE; i++) {
    dbsiPtr.p->pdxDataPage[i] = undoPagePtr.i + i;
  }//for

  ndbrequire(cfirstfreeUndoSeg != RNIL);
  if (cnoFreeUndoSeg == ZMIN_PAGE_LIMIT_TUP_COMMITREQ) {
    EXECUTE_DIRECT(DBLQH, GSN_TUP_COM_BLOCK, signal, 1);
    ljamEntry();
  }//if
  cnoFreeUndoSeg--;
  ndbrequire(cnoFreeUndoSeg >= 0);
  undoPagePtr.i = cfirstfreeUndoSeg;
  ptrCheckGuard(undoPagePtr, cnoOfUndoPage, undoPage);
  cfirstfreeUndoSeg = undoPagePtr.p->undoPageWord[ZPAGE_NEXT_POS];
  undoPagePtr.p->undoPageWord[ZPAGE_NEXT_POS] = RNIL;
  dbsiPtr.p->pdxUndoBufferSet[1] = undoPagePtr.i;
//  lliPtr.p->lliUndoPage = undoPagePtr.i;
  for (i = ZUB_SEGMENT_SIZE; i < (2 * ZUB_SEGMENT_SIZE); i++) {
    dbsiPtr.p->pdxDataPage[i] = undoPagePtr.i + (i - ZUB_SEGMENT_SIZE);
  }//for
  return;
}//Dbtup::allocRestartUndoBufferSegment()