rf_parityloggingdags.c

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

    readDataNodes[i].params[0].p = pda; /* physical disk addr desc */
    readDataNodes[i].params[1].p = rf_AllocBuffer(raidPtr, dag_h, pda, allocList);  /* buffer to hold old data */
    readDataNodes[i].params[2].v = parityStripeID;
    readDataNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, lu_flag, 0, which_ru);
    pda=pda->next;
    readDataNodes[i].propList[0] = NULL;
    readDataNodes[i].propList[1] = NULL;
  }

  /* initialize nodes which read old parity (Rop) */
  pda = asmap->parityInfo; i = 0;
  for (i = 0; i < numParityNodes; i++) {
    RF_ASSERT(pda != NULL);
    rf_InitNode(&readParityNodes[i], rf_wait, RF_FALSE, rf_DiskReadFunc, rf_DiskReadUndoFunc, rf_GenericWakeupFunc, nNodes, 1, 4, 0, dag_h, "Rop", allocList);
    readParityNodes[i].params[0].p = pda;
    readParityNodes[i].params[1].p = rf_AllocBuffer(raidPtr, dag_h, pda, allocList);    /* buffer to hold old parity */
    readParityNodes[i].params[2].v = parityStripeID;
    readParityNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
    readParityNodes[i].propList[0] = NULL;
    pda=pda->next;
  }

  /* initialize nodes which write new data (Wnd) */
  pda = asmap->physInfo;
  for (i=0; i < numDataNodes; i++) {
    RF_ASSERT(pda != NULL);
    rf_InitNode(&writeDataNodes[i], rf_wait, RF_TRUE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, nNodes, 4, 0, dag_h, "Wnd", allocList);
    writeDataNodes[i].params[0].p = pda;                    /* physical disk addr desc */
    writeDataNodes[i].params[1].p = pda->bufPtr;   /* buffer holding new data to be written */
    writeDataNodes[i].params[2].v = parityStripeID;
    writeDataNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);

    if (lu_flag) {
      /* initialize node to unlock the disk queue */
      rf_InitNode(&unlockDataNodes[i], rf_wait, RF_FALSE, rf_DiskUnlockFunc, rf_DiskUnlockUndoFunc, rf_GenericWakeupFunc, 1, 1, 2, 0, dag_h, "Und", allocList);
      unlockDataNodes[i].params[0].p = pda; /* physical disk addr desc */
      unlockDataNodes[i].params[1].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, lu_flag, which_ru);
    }
    pda = pda->next;
  }


  /* initialize nodes which compute new parity */
  /* we use the simple XOR func in the double-XOR case, and when we're accessing only a portion of one stripe unit.
   * the distinction between the two is that the regular XOR func assumes that the targbuf is a full SU in size,
   * and examines the pda associated with the buffer to decide where within the buffer to XOR the data, whereas
   * the simple XOR func just XORs the data into the start of the buffer.
   */
  if ((numParityNodes==2) || ((numDataNodes == 1) && (asmap->totalSectorsAccessed < raidPtr->Layout.sectorsPerStripeUnit))) {
    func = pfuncs->simple; undoFunc = rf_NullNodeUndoFunc; name = pfuncs->SimpleName;
    if (qfuncs)
      { qfunc = qfuncs->simple; qname = qfuncs->SimpleName;}
  } else {
    func = pfuncs->regular; undoFunc = rf_NullNodeUndoFunc; name = pfuncs->RegularName;
    if (qfuncs) { qfunc = qfuncs->regular; qname = qfuncs->RegularName;}
  }
  /* initialize the xor nodes: params are {pda,buf} from {Rod,Wnd,Rop} nodes, and raidPtr  */
  if (numParityNodes==2) {        /* double-xor case */
    for (i=0; i < numParityNodes; i++) {
      rf_InitNode(&xorNodes[i], rf_wait, RF_TRUE, func, undoFunc, NULL, 1, nNodes, 7, 1, dag_h, name, allocList);  /* no wakeup func for xor */
      xorNodes[i].flags |= RF_DAGNODE_FLAG_YIELD;
      xorNodes[i].params[0] = readDataNodes[i].params[0];
      xorNodes[i].params[1] = readDataNodes[i].params[1];
      xorNodes[i].params[2] = readParityNodes[i].params[0];
      xorNodes[i].params[3] = readParityNodes[i].params[1];
      xorNodes[i].params[4] = writeDataNodes[i].params[0];
      xorNodes[i].params[5] = writeDataNodes[i].params[1];
      xorNodes[i].params[6].p = raidPtr;
      xorNodes[i].results[0] = readParityNodes[i].params[1].p;   /* use old parity buf as target buf */
    }
  }
  else {
    /* there is only one xor node in this case */
    rf_InitNode(&xorNodes[0], rf_wait, RF_TRUE, func, undoFunc, NULL, 1, nNodes, (2 * (numDataNodes + numDataNodes + 1) + 1), 1, dag_h, name, allocList);
    xorNodes[0].flags |= RF_DAGNODE_FLAG_YIELD;
    for (i=0; i < numDataNodes + 1; i++) {
      /* set up params related to Rod and Rop nodes */
      xorNodes[0].params[2*i+0] = readDataNodes[i].params[0];    /* pda */
      xorNodes[0].params[2*i+1] = readDataNodes[i].params[1];    /* buffer pointer */
    }
    for (i=0; i < numDataNodes; i++) {
      /* set up params related to Wnd and Wnp nodes */
      xorNodes[0].params[2*(numDataNodes+1+i)+0] = writeDataNodes[i].params[0]; /* pda */
      xorNodes[0].params[2*(numDataNodes+1+i)+1] = writeDataNodes[i].params[1]; /* buffer pointer */
    }
    xorNodes[0].params[2*(numDataNodes+numDataNodes+1)].p = raidPtr;  /* xor node needs to get at RAID information */
    xorNodes[0].results[0] = readParityNodes[0].params[1].p;
  }

  /* initialize the log node(s) */
  pda = asmap->parityInfo;
  for (i = 0;  i < numParityNodes; i++) {
    RF_ASSERT(pda);
    rf_InitNode(&lpuNodes[i], rf_wait, RF_FALSE, rf_ParityLogUpdateFunc, rf_ParityLogUpdateUndoFunc, rf_GenericWakeupFunc, 1, 1, 2, 0, dag_h, "Lpu", allocList);
    lpuNodes[i].params[0].p = pda;                    /* PhysDiskAddr of parity */
    lpuNodes[i].params[1].p = xorNodes[i].results[0]; /* buffer pointer to parity */
    pda = pda->next;
  }
  

  /* Step 4. connect the nodes */

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

  /* connect block node to read old data nodes */
  RF_ASSERT(blockNode->numSuccedents == (numDataNodes + numParityNodes));
  for (i = 0; i < numDataNodes; i++) {
    blockNode->succedents[i] = &readDataNodes[i];
    RF_ASSERT(readDataNodes[i].numAntecedents == 1);
    readDataNodes[i].antecedents[0]= blockNode;
    readDataNodes[i].antType[0] = rf_control;
  }
  
  /* connect block node to read old parity nodes */
  for (i = 0; i < numParityNodes; i++) {
    blockNode->succedents[numDataNodes + i] = &readParityNodes[i];
    RF_ASSERT(readParityNodes[i].numAntecedents == 1);
    readParityNodes[i].antecedents[0] = blockNode;
    readParityNodes[i].antType[0] = rf_control;
  }

  /* connect read old data nodes to write new data nodes */
  for (i = 0; i < numDataNodes; i++) {
    RF_ASSERT(readDataNodes[i].numSuccedents == numDataNodes + numParityNodes);
    for (j = 0; j < numDataNodes; j++) {
      RF_ASSERT(writeDataNodes[j].numAntecedents == numDataNodes + numParityNodes);
      readDataNodes[i].succedents[j] = &writeDataNodes[j];
      writeDataNodes[j].antecedents[i] = &readDataNodes[i];
      if (i == j)
	writeDataNodes[j].antType[i] = rf_antiData;
      else
	writeDataNodes[j].antType[i] = rf_control;
    }
  }

  /* connect read old data nodes to xor nodes */
  for (i = 0; i < numDataNodes; i++)
    for (j = 0; j < numParityNodes; j++){
      RF_ASSERT(xorNodes[j].numAntecedents == numDataNodes + numParityNodes);
      readDataNodes[i].succedents[numDataNodes + j] = &xorNodes[j];
      xorNodes[j].antecedents[i] = &readDataNodes[i];
      xorNodes[j].antType[i] = rf_trueData;
    }

  /* connect read old parity nodes to write new data nodes */
  for (i = 0; i < numParityNodes; i++) {
    RF_ASSERT(readParityNodes[i].numSuccedents == numDataNodes + numParityNodes);
    for (j = 0; j < numDataNodes; j++) {
      readParityNodes[i].succedents[j] = &writeDataNodes[j];
      writeDataNodes[j].antecedents[numDataNodes + i] = &readParityNodes[i];
      writeDataNodes[j].antType[numDataNodes + i] = rf_control;
    }
  }

  /* connect read old parity nodes to xor nodes */
  for (i = 0; i < numParityNodes; i++)
    for (j = 0; j < numParityNodes; j++) {
      readParityNodes[i].succedents[numDataNodes + j] = &xorNodes[j];
      xorNodes[j].antecedents[numDataNodes + i] = &readParityNodes[i];
      xorNodes[j].antType[numDataNodes + i] = rf_trueData;
    }

  /* connect xor nodes to write new parity nodes */
  for (i = 0; i < numParityNodes; i++) {
    RF_ASSERT(xorNodes[i].numSuccedents == 1);
    RF_ASSERT(lpuNodes[i].numAntecedents == 1);
    xorNodes[i].succedents[0] = &lpuNodes[i];
    lpuNodes[i].antecedents[0] = &xorNodes[i];
    lpuNodes[i].antType[0] = rf_trueData;
  }

  for (i = 0; i < numDataNodes; i++) {
    if (lu_flag) {
      /* connect write new data nodes to unlock nodes */
      RF_ASSERT(writeDataNodes[i].numSuccedents == 1);
      RF_ASSERT(unlockDataNodes[i].numAntecedents == 1);
      writeDataNodes[i].succedents[0] = &unlockDataNodes[i];
      unlockDataNodes[i].antecedents[0] = &writeDataNodes[i];
      unlockDataNodes[i].antType[0] = rf_control;

      /* connect unlock nodes to unblock node */
      RF_ASSERT(unlockDataNodes[i].numSuccedents == 1);
      RF_ASSERT(unblockNode->numAntecedents == (numDataNodes + (nfaults * numParityNodes)));
      unlockDataNodes[i].succedents[0] = unblockNode;
      unblockNode->antecedents[i] = &unlockDataNodes[i];
      unblockNode->antType[i] = rf_control;
    }
    else {
      /* connect write new data nodes to unblock node */
      RF_ASSERT(writeDataNodes[i].numSuccedents == 1);
      RF_ASSERT(unblockNode->numAntecedents == (numDataNodes + (nfaults * numParityNodes)));
      writeDataNodes[i].succedents[0] = unblockNode;
      unblockNode->antecedents[i] = &writeDataNodes[i];
      unblockNode->antType[i] = rf_control;
    }
  }

  /* connect write new parity nodes to unblock node */
  for (i = 0; i < numParityNodes; i++) {
    RF_ASSERT(lpuNodes[i].numSuccedents == 1);
    lpuNodes[i].succedents[0] = unblockNode;
    unblockNode->antecedents[numDataNodes + i] = &lpuNodes[i];
    unblockNode->antType[numDataNodes + i] = rf_control;
  }

  /* connect unblock node to terminator */
  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;
}


void rf_CreateParityLoggingSmallWriteDAG(
  RF_Raid_t             *raidPtr,
  RF_AccessStripeMap_t  *asmap,
  RF_DagHeader_t        *dag_h,
  void                  *bp,
  RF_RaidAccessFlags_t   flags,
  RF_AllocListElem_t    *allocList,
  RF_RedFuncs_t         *pfuncs,
  RF_RedFuncs_t         *qfuncs)
{
  dag_h->creator = "ParityLoggingSmallWriteDAG";
  rf_CommonCreateParityLoggingSmallWriteDAG(raidPtr, asmap, dag_h, bp, flags, allocList, &rf_xorFuncs, NULL);
}


void rf_CreateParityLoggingLargeWriteDAG(
  RF_Raid_t              *raidPtr,
  RF_AccessStripeMap_t   *asmap,
  RF_DagHeader_t         *dag_h,
  void                   *bp,
  RF_RaidAccessFlags_t    flags,
  RF_AllocListElem_t     *allocList,
  int                     nfaults,
  int                   (*redFunc)())
{
  dag_h->creator = "ParityLoggingSmallWriteDAG";
  rf_CommonCreateParityLoggingLargeWriteDAG(raidPtr, asmap, dag_h, bp, flags, allocList, 1, rf_RegularXorFunc);
}

#endif /* RF_INCLUDE_PARITYLOGGING > 0 */