rf_dagffwr.c

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

  rf_InitNode(termNode, rf_wait, RF_FALSE, rf_TerminateFunc, rf_TerminateUndoFunc,
    NULL, 0, nNodes, 0, 0, dag_h, "Trm", allocList);

  /* initialize nodes which read old data (Rod) */
  for (i = 0; i < numDataNodes; i++) {
    rf_InitNode(&readDataNodes[i], rf_wait, RF_FALSE, rf_DiskReadFunc, rf_DiskReadUndoFunc,
      rf_GenericWakeupFunc, (nfaults * numParityNodes), 1, 4, 0, dag_h,
      "Rod", allocList);
    RF_ASSERT(pda != NULL);
    /* physical disk addr desc */
    readDataNodes[i].params[0].p = pda;
    /* buffer to hold old data */
    readDataNodes[i].params[1].p = rf_AllocBuffer(raidPtr,
      dag_h, pda, allocList);
    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;
    for (j = 0; j < readDataNodes[i].numSuccedents; j++) {
      readDataNodes[i].propList[j] = 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, numParityNodes, 1, 4,
      0, dag_h, "Rop", allocList);
    readParityNodes[i].params[0].p = pda;
    /* buffer to hold old parity */
    readParityNodes[i].params[1].p = rf_AllocBuffer(raidPtr,
      dag_h, pda, allocList);
    readParityNodes[i].params[2].v = parityStripeID;
    readParityNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
      lu_flag, 0, which_ru);
    pda = pda->next;
    for (j = 0; j < readParityNodes[i].numSuccedents; j++) {
      readParityNodes[i].propList[0] = NULL;
    }
  }

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

  /* 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_FALSE, rf_DiskWriteFunc,
      rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h,
      "Wnd", allocList);
    /* physical disk addr desc */
    writeDataNodes[i].params[0].p = pda;
    /* buffer holding new data to be written */
    writeDataNodes[i].params[1].p = pda->bufPtr;
    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);
      /* physical disk addr desc */
      unlockDataNodes[i].params[0].p = pda;
      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 and Q.
   */
  /*
   * 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 {
      qfunc = NULL;
      qname = NULL;
    }
  }
  else {
    func = pfuncs->regular;
    undoFunc = rf_NullNodeUndoFunc;
    name = pfuncs->RegularName;
    if (qfuncs) {
      qfunc = qfuncs->regular;
      qname = qfuncs->RegularName;
    }
    else {
      qfunc = NULL;
      qname = NULL;
    }
  }
  /*
   * 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++) {
      /* note: no wakeup func for xor */
      rf_InitNode(&xorNodes[i], rf_wait, RF_FALSE, func, undoFunc, NULL,
        1, (numDataNodes + numParityNodes), 7, 1, dag_h, name, allocList);
      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;
      /* use old parity buf as target buf */
      xorNodes[i].results[0] = readParityNodes[i].params[1].p;
      if (nfaults == 2) {
        /* note: no wakeup func for qor */
        rf_InitNode(&qNodes[i], rf_wait, RF_FALSE, qfunc, undoFunc, NULL, 1,
          (numDataNodes + numParityNodes), 7, 1, dag_h, qname, allocList);
        qNodes[i].params[0]   = readDataNodes[i].params[0];
        qNodes[i].params[1]   = readDataNodes[i].params[1];
        qNodes[i].params[2]   = readQNodes[i].params[0];
        qNodes[i].params[3]   = readQNodes[i].params[1];
        qNodes[i].params[4]   = writeDataNodes[i].params[0];
        qNodes[i].params[5]   = writeDataNodes[i].params[1];
        qNodes[i].params[6].p = raidPtr;
        /* use old Q buf as target buf */
        qNodes[i].results[0] = readQNodes[i].params[1].p;
      }
    }
  }
  else {
    /* there is only one xor node in this case */
    rf_InitNode(&xorNodes[0], rf_wait, RF_FALSE, func, undoFunc, NULL, 1,
      (numDataNodes + numParityNodes),
      (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 ptr */
    }
    for (i=0; i < numDataNodes; i++) {
      /* set up params related to Wnd and Wnp nodes */
      xorNodes[0].params[2*(numDataNodes+1+i)+0] = /* pda */
        writeDataNodes[i].params[0];
      xorNodes[0].params[2*(numDataNodes+1+i)+1] = /* buffer ptr */
       writeDataNodes[i].params[1];
    }
    /* xor node needs to get at RAID information */
    xorNodes[0].params[2*(numDataNodes+numDataNodes+1)].p = raidPtr;
    xorNodes[0].results[0] = readParityNodes[0].params[1].p;
    if (nfaults == 2)  {
      rf_InitNode(&qNodes[0], rf_wait, RF_FALSE, qfunc, undoFunc, NULL, 1,
        (numDataNodes + numParityNodes),
        (2 * (numDataNodes + numDataNodes + 1) + 1), 1, dag_h,
        qname, allocList);
      for (i=0; i<numDataNodes; i++) {
        /* set up params related to Rod */
        qNodes[0].params[2*i+0] = readDataNodes[i].params[0]; /* pda */
        qNodes[0].params[2*i+1] = readDataNodes[i].params[1]; /* buffer ptr */
      }
      /* and read old q */
      qNodes[0].params[2*numDataNodes + 0] = /* pda */
        readQNodes[0].params[0];
      qNodes[0].params[2*numDataNodes + 1] = /* buffer ptr */
        readQNodes[0].params[1];
      for (i=0; i < numDataNodes; i++) {
        /* set up params related to Wnd nodes */
        qNodes[0].params[2*(numDataNodes+1+i)+0] = /* pda */
          writeDataNodes[i].params[0];
        qNodes[0].params[2*(numDataNodes+1+i)+1] = /* buffer ptr */
          writeDataNodes[i].params[1];
      }
      /* xor node needs to get at RAID information */
      qNodes[0].params[2*(numDataNodes+numDataNodes+1)].p = raidPtr;
      qNodes[0].results[0] = readQNodes[0].params[1].p;
    }
  }

  /* initialize nodes which write new parity (Wnp) */
  pda = asmap->parityInfo;
  for (i=0;  i < numParityNodes; i++) {
    rf_InitNode(&writeParityNodes[i], rf_wait, RF_FALSE, rf_DiskWriteFunc,
      rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h,
      "Wnp", allocList);
    RF_ASSERT(pda != NULL);
    writeParityNodes[i].params[0].p = pda; /* param 1 (bufPtr) filled in by xor node */
    writeParityNodes[i].params[1].p = xorNodes[i].results[0]; /* buffer pointer for parity write operation */
    writeParityNodes[i].params[2].v = parityStripeID;
    writeParityNodes[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(&unlockParityNodes[i], rf_wait, RF_FALSE, rf_DiskUnlockFunc,
        rf_DiskUnlockUndoFunc, rf_GenericWakeupFunc, 1, 1, 2, 0, dag_h,
        "Unp", allocList);
      unlockParityNodes[i].params[0].p = pda; /* physical disk addr desc */
      unlockParityNodes[i].params[1].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
        0, lu_flag, which_ru);
    }
    pda = pda->next;
  }

  /* initialize nodes which write new Q (Wnq) */
  if (nfaults == 2) {
    pda = asmap->qInfo;
    for (i=0;  i < numParityNodes; i++) {
      rf_InitNode(&writeQNodes[i], rf_wait, RF_FALSE, rf_DiskWriteFunc,
        rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h,
        "Wnq", allocList);
      RF_ASSERT(pda != NULL);
      writeQNodes[i].params[0].p = pda; /* param 1 (bufPtr) filled in by xor node */
      writeQNodes[i].params[1].p = qNodes[i].results[0]; /* buffer pointer for parity write operation */
      writeQNodes[i].params[2].v = parityStripeID;
      writeQNodes[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(&unlockQNodes[i], rf_wait, RF_FALSE, rf_DiskUnlockFunc,
          rf_DiskUnlockUndoFunc, rf_GenericWakeupFunc, 1, 1, 2, 0, dag_h,
          "Unq", allocList);
        unlockQNodes[i].params[0].p = pda; /* physical disk addr desc */
        unlockQNodes[i].params[1].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
          0, lu_flag, which_ru);
      }
      pda = pda->next;
    }
  }

  /*
   * Step 4. connect the nodes.
   */

  /* connect header to block node */
  dag_h->succedents[0] = blockNode;

  /* connect block node to read old data nodes */
  RF_ASSERT(blockNode->numSuccedents == (numDataNodes + (numParityNodes * nfaults)));
  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 block node to read old Q nodes */
  if (nfaults == 2) {
    for (i = 0; i < numParityNodes; i++) {
      blockNode->succedents[numDataNodes + numParityNodes + i] = &readQNodes[i];
      RF_ASSERT(readQNodes[i].numAntecedents == 1);
      readQNodes[i].antecedents[0] = blockNode;
      readQNodes[i].antType[0] = rf_control;
    }
  }

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

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

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

  /* connect read old q nodes to q nodes */
  if (nfaults == 2) {
    for (i = 0; i < numParityNodes; i++) {
      RF_ASSERT(readParityNodes[i].numSuccedents == numParityNodes);
      for (j = 0; j < numParityNodes; j++) {
        readQNodes[i].succedents[j] = &qNodes[j];
        qNodes[j].antecedents[numDataNodes + i] = &readQNodes[i];
        qNodes[j].antType[numDataNodes + i] = rf_trueData;
      }
    }
  }

  /* connect xor nodes to commit node */
  RF_ASSERT(commitNode->numAntecedents == (nfaults * numParityNodes));
  for (i = 0; i < numParityNodes; i++) {
    RF_ASSERT(xorNodes[i].numSuccedents == 1);