rf_dagdegwr.c

RAIDFrame是个非常好的磁盘阵列RAID仿真工具

  i = 0;
  blockNode   = &nodes[i]; i += 1;
  commitNode    = &nodes[i]; i += 1;
  unblockNode = &nodes[i]; i += 1;
  termNode    = &nodes[i]; i += 1;
  xorNode     = &nodes[i]; i += 1;
  wnpNode     = &nodes[i]; i += 1;
  wndNodes    = &nodes[i]; i += nWndNodes;
  rrdNodes    = &nodes[i]; i += nRrdNodes;
  if (nfaults == 2) {
    wnqNode   = &nodes[i]; i += 1;
  }
  else {
    wnqNode = NULL;
  }
  RF_ASSERT(i == nNodes);

  /* this dag can not commit until all rrd and xor Nodes have completed */
  dag_h->numCommitNodes = 1;
  dag_h->numCommits = 0;
  dag_h->numSuccedents = 1;

  RF_ASSERT( nRrdNodes > 0 );
  rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
    NULL, nRrdNodes, 0, 0, 0, dag_h, "Nil", allocList);
  rf_InitNode(commitNode, rf_wait, RF_TRUE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
    NULL, nWndNodes + nfaults, 1, 0, 0, dag_h, "Cmt", allocList);
  rf_InitNode(unblockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
    NULL, 1, nWndNodes + nfaults, 0, 0, dag_h, "Nil", allocList);
  rf_InitNode(termNode, rf_wait, RF_FALSE, rf_TerminateFunc, rf_TerminateUndoFunc,
    NULL, 0, 1, 0, 0, dag_h, "Trm", allocList);
  rf_InitNode(xorNode, rf_wait, RF_FALSE, redFunc, rf_NullNodeUndoFunc, NULL, 1,
    nRrdNodes, 2*nXorBufs+2, nfaults, dag_h, "Xrc", allocList);

  /*
   * Fill in the Rrd nodes. If any of the rrd buffers are the same size as
   * the failed buffer, save a pointer to it so we can use it as the target
   * of the XOR. The pdas in the rrd nodes have been range-restricted, so if
   * a buffer is the same size as the failed buffer, it must also be at the
   * same alignment within the SU.
   */
  i = 0;
  if (new_asm_h[0]) {
    for (i=0, pda=new_asm_h[0]->stripeMap->physInfo;
        i<new_asm_h[0]->stripeMap->numStripeUnitsAccessed;
        i++, pda=pda->next)
    {
      rf_InitNode(&rrdNodes[i], rf_wait, RF_FALSE, rf_DiskReadFunc, rf_DiskReadUndoFunc,
        rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Rrd", allocList);
      RF_ASSERT(pda);
      rrdNodes[i].params[0].p = pda;
      rrdNodes[i].params[1].p = pda->bufPtr;
      rrdNodes[i].params[2].v = parityStripeID;
      rrdNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
    }
  }
  /* i now equals the number of stripe units accessed in new_asm_h[0] */
  if (new_asm_h[1]) {
    for (j=0,pda=new_asm_h[1]->stripeMap->physInfo;
        j<new_asm_h[1]->stripeMap->numStripeUnitsAccessed;
        j++, pda=pda->next)
    {
      rf_InitNode(&rrdNodes[i+j], rf_wait, RF_FALSE, rf_DiskReadFunc, rf_DiskReadUndoFunc,
        rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Rrd", allocList);
      RF_ASSERT(pda);
      rrdNodes[i+j].params[0].p = pda;
      rrdNodes[i+j].params[1].p = pda->bufPtr;
      rrdNodes[i+j].params[2].v = parityStripeID;
      rrdNodes[i+j].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
      if (allowBufferRecycle && (pda->numSector == failedPDA->numSector))
        xorTargetBuf = pda->bufPtr;
    }
  }
  if (rdnodesFaked) {
    /*
     * This is where we'll init that fake noop read node
     * (XXX should the wakeup func be different?)
     */
    rf_InitNode(&rrdNodes[0], rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
      NULL, 1, 1, 0, 0, dag_h, "RrN", allocList);
  }

  /*
   * Make a PDA for the parity unit.  The parity PDA should start at
   * the same offset into the SU as the failed PDA.
   */
  /* 
   * Danner comment:
   * I don't think this copy is really necessary.
   * We are in one of two cases here.
   *   (1) The entire failed unit is written. Then asmap->parityInfo will
   *       describe the entire parity.
   *   (2) We are only writing a subset of the failed unit and nothing
   *       else. Then the asmap->parityInfo describes the failed unit and
   *       the copy can also be avoided.
   */

  RF_MallocAndAdd(parityPDA, sizeof(RF_PhysDiskAddr_t), (RF_PhysDiskAddr_t *), allocList);
  parityPDA->row = asmap->parityInfo->row;
  parityPDA->col = asmap->parityInfo->col;
  parityPDA->startSector = ((asmap->parityInfo->startSector / sectorsPerSU)
    * sectorsPerSU) + (failedPDA->startSector % sectorsPerSU);
  parityPDA->numSector = failedPDA->numSector;

  if (!xorTargetBuf) {
    RF_CallocAndAdd(xorTargetBuf, 1,
      rf_RaidAddressToByte(raidPtr, failedPDA->numSector), (char *), allocList);
  }

  /* init the Wnp node */
  rf_InitNode(wnpNode, rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
    rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wnp", allocList);
  wnpNode->params[0].p = parityPDA;
  wnpNode->params[1].p = xorTargetBuf;
  wnpNode->params[2].v = parityStripeID;
  wnpNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);

  /* fill in the Wnq Node */
  if (nfaults == 2) {
    {
      RF_MallocAndAdd(parityPDA, sizeof(RF_PhysDiskAddr_t),
        (RF_PhysDiskAddr_t *), allocList);
      parityPDA->row = asmap->qInfo->row;
      parityPDA->col = asmap->qInfo->col;
      parityPDA->startSector = ((asmap->qInfo->startSector / sectorsPerSU)
        * sectorsPerSU) + (failedPDA->startSector % sectorsPerSU);
      parityPDA->numSector = failedPDA->numSector;

      rf_InitNode(wnqNode, rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
        rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wnq", allocList);
      wnqNode->params[0].p = parityPDA;
      RF_CallocAndAdd(xorNode->results[1], 1,
        rf_RaidAddressToByte(raidPtr, failedPDA->numSector), (char *), allocList); 
      wnqNode->params[1].p = xorNode->results[1];
      wnqNode->params[2].v = parityStripeID;
      wnqNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
    }
  }

  /* fill in the Wnd nodes */
  for (pda=asmap->physInfo, i=0; i<nWndNodes; i++, pda=pda->next) {
    if (pda == failedPDA) {
      i--;
      continue;
    }
    rf_InitNode(&wndNodes[i], rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
      rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wnd", allocList);
    RF_ASSERT(pda);
    wndNodes[i].params[0].p = pda;
    wndNodes[i].params[1].p = pda->bufPtr;
    wndNodes[i].params[2].v = parityStripeID;
    wndNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
  }

  /* fill in the results of the xor node */
  xorNode->results[0] = xorTargetBuf;

  /* fill in the params of the xor node */

  paramNum=0;
  if (rdnodesFaked == 0) {
    for (i=0; i<nRrdNodes; i++) {
      /* all the Rrd nodes need to be xored together */
      xorNode->params[paramNum++] = rrdNodes[i].params[0];
      xorNode->params[paramNum++] = rrdNodes[i].params[1];
    }
  }
  for (i=0; i < nWndNodes; i++) {
    /* any Wnd nodes that overlap the failed access need to be xored in */
    if (overlappingPDAs[i]) {
      RF_MallocAndAdd(pda, sizeof(RF_PhysDiskAddr_t), (RF_PhysDiskAddr_t *), allocList);
      bcopy((char *)wndNodes[i].params[0].p, (char *)pda, sizeof(RF_PhysDiskAddr_t));
      rf_RangeRestrictPDA(raidPtr, failedPDA, pda, RF_RESTRICT_DOBUFFER, 0);
      xorNode->params[paramNum++].p = pda;
      xorNode->params[paramNum++].p = pda->bufPtr;
    }
  }
  RF_Free(overlappingPDAs, asmap->numStripeUnitsAccessed * sizeof(char));

  /*
   * Install the failed PDA into the xor param list so that the
   * new data gets xor'd in.
   */
  xorNode->params[paramNum++].p = failedPDA;
  xorNode->params[paramNum++].p = failedPDA->bufPtr;

  /*
   * The last 2 params to the recovery xor node are always the failed
   * PDA and the raidPtr. install the failedPDA even though we have just
   * done so above. This allows us to use the same XOR function for both
   * degraded reads and degraded writes.
   */
  xorNode->params[paramNum++].p = failedPDA;
  xorNode->params[paramNum++].p = raidPtr;
  RF_ASSERT( paramNum == 2*nXorBufs+2 );

  /*
   * Code to link nodes begins here
   */

  /* link header to block node */
  RF_ASSERT(blockNode->numAntecedents == 0);
  dag_h->succedents[0] = blockNode;

  /* link block node to rd nodes */
  RF_ASSERT(blockNode->numSuccedents == nRrdNodes);
  for (i = 0; i < nRrdNodes; i++) {
    RF_ASSERT(rrdNodes[i].numAntecedents == 1);
    blockNode->succedents[i] = &rrdNodes[i];
    rrdNodes[i].antecedents[0] = blockNode;
    rrdNodes[i].antType[0] = rf_control;
  }

  /* link read nodes to xor node*/
  RF_ASSERT(xorNode->numAntecedents == nRrdNodes);
  for (i = 0; i < nRrdNodes; i++) {
    RF_ASSERT(rrdNodes[i].numSuccedents == 1);
    rrdNodes[i].succedents[0] = xorNode;
    xorNode->antecedents[i] = &rrdNodes[i];
    xorNode->antType[i] = rf_trueData;
  }

  /* link xor node to commit node */
  RF_ASSERT(xorNode->numSuccedents == 1);
  RF_ASSERT(commitNode->numAntecedents == 1);
  xorNode->succedents[0] = commitNode;
  commitNode->antecedents[0] = xorNode;
  commitNode->antType[0] = rf_control;

  /* link commit node to wnd nodes */
  RF_ASSERT(commitNode->numSuccedents == nfaults + nWndNodes);
  for (i = 0; i < nWndNodes; i++) {
    RF_ASSERT(wndNodes[i].numAntecedents == 1);
    commitNode->succedents[i] = &wndNodes[i];
    wndNodes[i].antecedents[0] = commitNode;
    wndNodes[i].antType[0] = rf_control;
  }

  /* link the commit node to wnp, wnq nodes */
  RF_ASSERT(wnpNode->numAntecedents == 1);
  commitNode->succedents[nWndNodes] = wnpNode;
  wnpNode->antecedents[0] = commitNode;
  wnpNode->antType[0] = rf_control;
  if (nfaults == 2) {
    RF_ASSERT(wnqNode->numAntecedents == 1);
    commitNode->succedents[nWndNodes + 1] = wnqNode;
    wnqNode->antecedents[0] = commitNode;
    wnqNode->antType[0] = rf_control;
  }

  /* link write new data nodes to unblock node */
  RF_ASSERT(unblockNode->numAntecedents == (nWndNodes + nfaults));
  for(i = 0; i < nWndNodes; i++) {
    RF_ASSERT(wndNodes[i].numSuccedents == 1);
    wndNodes[i].succedents[0] = unblockNode;
    unblockNode->antecedents[i] = &wndNodes[i];
    unblockNode->antType[i] = rf_control;
  }

  /* link write new parity node to unblock node */
  RF_ASSERT(wnpNode->numSuccedents == 1);
  wnpNode->succedents[0] = unblockNode;
  unblockNode->antecedents[nWndNodes] = wnpNode;
  unblockNode->antType[nWndNodes] = rf_control;

  /* link write new q node to unblock node */
  if (nfaults == 2) {
    RF_ASSERT(wnqNode->numSuccedents == 1);
    wnqNode->succedents[0] = unblockNode;
    unblockNode->antecedents[nWndNodes+1] = wnqNode;
    unblockNode->antType[nWndNodes+1] = rf_control;
  }

  /* link unblock node to term node */
  RF_ASSERT(unblockNode->numSuccedents == 1);
  RF_ASSERT(termNode->numAntecedents == 1);
  RF_ASSERT(termNode->numSuccedents == 0);
  unblockNode->succedents[0] = termNode;
  termNode->antecedents[0]  = unblockNode;
  termNode->antType[0] = rf_control;
}

#define CONS_PDA(if,start,num) \
  pda_p->row = asmap->if->row;    pda_p->col = asmap->if->col; \
  pda_p->startSector = ((asmap->if->startSector / secPerSU) * secPerSU) + start; \
  pda_p->numSector = num; \
  pda_p->next = NULL; \
  RF_MallocAndAdd(pda_p->bufPtr,rf_RaidAddressToByte(raidPtr,num),(char *), allocList)

void rf_WriteGenerateFailedAccessASMs(
  RF_Raid_t              *raidPtr,
  RF_AccessStripeMap_t   *asmap,
  RF_PhysDiskAddr_t     **pdap,
  int                    *nNodep,
  RF_PhysDiskAddr_t     **pqpdap,
  int                    *nPQNodep,
  RF_AllocListElem_t     *allocList)
{
  RF_RaidLayout_t *layoutPtr = &(raidPtr->Layout);
  int PDAPerDisk,i;
  RF_SectorCount_t secPerSU = layoutPtr->sectorsPerStripeUnit;
  int numDataCol = layoutPtr->numDataCol;
  int state;
  unsigned napdas;
  RF_SectorNum_t fone_start, fone_end, ftwo_start, ftwo_end;
  RF_PhysDiskAddr_t *fone = asmap->failedPDA, *ftwo = asmap->failedPDAtwo;
  RF_PhysDiskAddr_t *pda_p;
  RF_RaidAddr_t sosAddr;

  /* determine how many pda's we will have to generate per unaccess stripe.
     If there is only one failed data unit, it is one; if two, possibly two,
     depending wether they overlap. */

  fone_start = rf_StripeUnitOffset(layoutPtr,fone->startSector);
  fone_end = fone_start + fone->numSector;

  if (asmap->numDataFailed==1)
    {
      PDAPerDisk = 1;
      state = 1;
      RF_MallocAndAdd(*pqpdap,2*sizeof(RF_PhysDiskAddr_t),(RF_PhysDiskAddr_t *), allocList);
      pda_p = *pqpdap;