xref: /src/sys/dev/raidframe/rf_dagffwr.c

/*	$NetBSD: rf_dagffwr.c,v 1.1 1998/11/13 04:20:27 oster Exp $	*/
/*
 * Copyright (c) 1995 Carnegie-Mellon University.
 * All rights reserved.
 *
 * Author: Mark Holland, Daniel Stodolsky, William V. Courtright II
 *
 * Permission to use, copy, modify and distribute this software and
 * its documentation is hereby granted, provided that both the copyright
 * notice and this permission notice appear in all copies of the
 * software, derivative works or modified versions, and any portions
 * thereof, and that both notices appear in supporting documentation.
 *
 * CARNEGIE MELLON ALLOWS FREE USE OF THIS SOFTWARE IN ITS "AS IS"
 * CONDITION.  CARNEGIE MELLON DISCLAIMS ANY LIABILITY OF ANY KIND
 * FOR ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF THIS SOFTWARE.
 *
 * Carnegie Mellon requests users of this software to return to
 *
 *  Software Distribution Coordinator  or  Software.Distribution (at) CS.CMU.EDU
 *  School of Computer Science
 *  Carnegie Mellon University
 *  Pittsburgh PA 15213-3890
 *
 * any improvements or extensions that they make and grant Carnegie the
 * rights to redistribute these changes.
 */

/*
 * rf_dagff.c
 *
 * code for creating fault-free DAGs
 *
 * :
 * Log: rf_dagffwr.c,v
 * Revision 1.19  1996/07/31 15:35:24  jimz
 * evenodd changes; bugfixes for double-degraded archs, generalize
 * some formerly PQ-only functions
 *
 * Revision 1.18  1996/07/28  20:31:39  jimz
 * i386netbsd port
 * true/false fixup
 *
 * Revision 1.17  1996/07/27  18:40:24  jimz
 * cleanup sweep
 *
 * Revision 1.16  1996/07/22  19:52:16  jimz
 * switched node params to RF_DagParam_t, a union of
 * a 64-bit int and a void *, for better portability
 * attempted hpux port, but failed partway through for
 * lack of a single C compiler capable of compiling all
 * source files
 *
 * Revision 1.15  1996/06/11  01:27:50  jimz
 * Fixed bug where diskthread shutdown would crash or hang. This
 * turned out to be two distinct bugs:
 * (1) [crash] The thread shutdown code wasn't properly waiting for
 * all the diskthreads to complete. This caused diskthreads that were
 * exiting+cleaning up to unlock a destroyed mutex.
 * (2) [hang] TerminateDiskQueues wasn't locking, and DiskIODequeue
 * only checked for termination _after_ a wakeup if the queues were
 * empty. This was a race where the termination wakeup could be lost
 * by the dequeueing thread, and the system would hang waiting for the
 * thread to exit, while the thread waited for an I/O or a signal to
 * check the termination flag.
 *
 * Revision 1.14  1996/06/10  22:24:01  wvcii
 * added write dags which do not have a commit node and are
 * used in forward and backward error recovery experiments.
 *
 * Revision 1.13  1996/06/07  22:26:27  jimz
 * type-ify which_ru (RF_ReconUnitNum_t)
 *
 * Revision 1.12  1996/06/07  21:33:04  jimz
 * begin using consistent types for sector numbers,
 * stripe numbers, row+col numbers, recon unit numbers
 *
 * Revision 1.11  1996/05/31  22:26:54  jimz
 * fix a lot of mapping problems, memory allocation problems
 * found some weird lock issues, fixed 'em
 * more code cleanup
 *
 * Revision 1.10  1996/05/30  11:29:41  jimz
 * Numerous bug fixes. Stripe lock release code disagreed with the taking code
 * about when stripes should be locked (I made it consistent: no parity, no lock)
 * There was a lot of extra serialization of I/Os which I've removed- a lot of
 * it was to calculate values for the cache code, which is no longer with us.
 * More types, function, macro cleanup. Added code to properly quiesce the array
 * on shutdown. Made a lot of stuff array-specific which was (bogusly) general
 * before. Fixed memory allocation, freeing bugs.
 *
 * Revision 1.9  1996/05/27  18:56:37  jimz
 * more code cleanup
 * better typing
 * compiles in all 3 environments
 *
 * Revision 1.8  1996/05/24  22:17:04  jimz
 * continue code + namespace cleanup
 * typed a bunch of flags
 *
 * Revision 1.7  1996/05/24  04:28:55  jimz
 * release cleanup ckpt
 *
 * Revision 1.6  1996/05/23  21:46:35  jimz
 * checkpoint in code cleanup (release prep)
 * lots of types, function names have been fixed
 *
 * Revision 1.5  1996/05/23  00:33:23  jimz
 * code cleanup: move all debug decls to rf_options.c, all extern
 * debug decls to rf_options.h, all debug vars preceded by rf_
 *
 * Revision 1.4  1996/05/18  19:51:34  jimz
 * major code cleanup- fix syntax, make some types consistent,
 * add prototypes, clean out dead code, et cetera
 *
 * Revision 1.3  1996/05/15  23:23:12  wvcii
 * fixed bug in small write read old q node succedent initialization
 *
 * Revision 1.2  1996/05/08  21:01:24  jimz
 * fixed up enum type names that were conflicting with other
 * enums and function names (ie, "panic")
 * future naming trends will be towards RF_ and rf_ for
 * everything raidframe-related
 *
 * Revision 1.1  1996/05/03  19:20:45  wvcii
 * Initial revision
 *
 */

#include "rf_types.h"
#include "rf_raid.h"
#include "rf_dag.h"
#include "rf_dagutils.h"
#include "rf_dagfuncs.h"
#include "rf_threadid.h"
#include "rf_debugMem.h"
#include "rf_dagffrd.h"
#include "rf_memchunk.h"
#include "rf_general.h"
#include "rf_dagffwr.h"

/******************************************************************************
 *
 * General comments on DAG creation:
 *
 * All DAGs in this file use roll-away error recovery.  Each DAG has a single
 * commit node, usually called "Cmt."  If an error occurs before the Cmt node
 * is reached, the execution engine will halt forward execution and work
 * backward through the graph, executing the undo functions.  Assuming that
 * each node in the graph prior to the Cmt node are undoable and atomic - or -
 * does not make changes to permanent state, the graph will fail atomically.
 * If an error occurs after the Cmt node executes, the engine will roll-forward
 * through the graph, blindly executing nodes until it reaches the end.
 * If a graph reaches the end, it is assumed to have completed successfully.
 *
 * A graph has only 1 Cmt node.
 *
 */


/******************************************************************************
 *
 * The following wrappers map the standard DAG creation interface to the
 * DAG creation routines.  Additionally, these wrappers enable experimentation
 * with new DAG structures by providing an extra level of indirection, allowing
 * the DAG creation routines to be replaced at this single point.
 */


void rf_CreateNonRedundantWriteDAG(
  RF_Raid_t             *raidPtr,
  RF_AccessStripeMap_t  *asmap,
  RF_DagHeader_t        *dag_h,
  void                  *bp,
  RF_RaidAccessFlags_t   flags,
  RF_AllocListElem_t    *allocList,
  RF_IoType_t            type)
{
  rf_CreateNonredundantDAG(raidPtr, asmap, dag_h, bp, flags, allocList,
    RF_IO_TYPE_WRITE);
}

void rf_CreateRAID0WriteDAG(
  RF_Raid_t             *raidPtr,
  RF_AccessStripeMap_t  *asmap,
  RF_DagHeader_t        *dag_h,
  void                  *bp,
  RF_RaidAccessFlags_t   flags,
  RF_AllocListElem_t    *allocList,
  RF_IoType_t            type)
{
  rf_CreateNonredundantDAG(raidPtr, asmap, dag_h, bp, flags, allocList,
    RF_IO_TYPE_WRITE);
}

void rf_CreateSmallWriteDAG(
  RF_Raid_t             *raidPtr,
  RF_AccessStripeMap_t  *asmap,
  RF_DagHeader_t        *dag_h,
  void                  *bp,
  RF_RaidAccessFlags_t   flags,
  RF_AllocListElem_t    *allocList)
{
#if RF_FORWARD > 0
  rf_CommonCreateSmallWriteDAGFwd(raidPtr, asmap, dag_h, bp, flags, allocList,
    &rf_xorFuncs, NULL);
#else /* RF_FORWARD > 0 */
#if RF_BACKWARD > 0
  rf_CommonCreateSmallWriteDAGFwd(raidPtr, asmap, dag_h, bp, flags, allocList,
    &rf_xorFuncs, NULL);
#else /* RF_BACKWARD > 0 */
  /* "normal" rollaway */
  rf_CommonCreateSmallWriteDAG(raidPtr, asmap, dag_h, bp, flags, allocList,
    &rf_xorFuncs, NULL);
#endif /* RF_BACKWARD > 0 */
#endif /* RF_FORWARD > 0 */
}

void rf_CreateLargeWriteDAG(
  RF_Raid_t             *raidPtr,
  RF_AccessStripeMap_t  *asmap,
  RF_DagHeader_t        *dag_h,
  void                  *bp,
  RF_RaidAccessFlags_t   flags,
  RF_AllocListElem_t    *allocList)
{
#if RF_FORWARD > 0
  rf_CommonCreateLargeWriteDAGFwd(raidPtr, asmap, dag_h, bp, flags, allocList,
    1, rf_RegularXorFunc, RF_TRUE);
#else /* RF_FORWARD > 0 */
#if RF_BACKWARD > 0
  rf_CommonCreateLargeWriteDAGFwd(raidPtr, asmap, dag_h, bp, flags, allocList,
    1, rf_RegularXorFunc, RF_TRUE);
#else /* RF_BACKWARD > 0 */
  /* "normal" rollaway */
  rf_CommonCreateLargeWriteDAG(raidPtr, asmap, dag_h, bp, flags, allocList,
    1, rf_RegularXorFunc, RF_TRUE);
#endif /* RF_BACKWARD > 0 */
#endif /* RF_FORWARD > 0 */
}


/******************************************************************************
 *
 * DAG creation code begins here
 */


/******************************************************************************
 *
 * creates a DAG to perform a large-write operation:
 *
 *           / Rod \           / Wnd \
 * H -- block- Rod - Xor - Cmt - Wnd --- T
 *           \ Rod /          \  Wnp /
 *                             \[Wnq]/
 *
 * The XOR node also does the Q calculation in the P+Q architecture.
 * All nodes are before the commit node (Cmt) are assumed to be atomic and
 * undoable - or - they make no changes to permanent state.
 *
 * Rod = read old data
 * Cmt = commit node
 * Wnp = write new parity
 * Wnd = write new data
 * Wnq = write new "q"
 * [] denotes optional segments in the graph
 *
 * Parameters:  raidPtr   - description of the physical array
 *              asmap     - logical & physical addresses for this access
 *              bp        - buffer ptr (holds write data)
 *              flags     - general flags (e.g. disk locking)
 *              allocList - list of memory allocated in DAG creation
 *              nfaults   - number of faults array can tolerate
 *                          (equal to # redundancy units in stripe)
 *              redfuncs  - list of redundancy generating functions
 *
 *****************************************************************************/

void rf_CommonCreateLargeWriteDAG(
  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)(RF_DagNode_t *),
  int                    allowBufferRecycle)
{
  RF_DagNode_t *nodes, *wndNodes, *rodNodes, *xorNode, *wnpNode;
  RF_DagNode_t *wnqNode, *blockNode, *commitNode, *termNode;
  int nWndNodes, nRodNodes, i, nodeNum, asmNum;
  RF_AccessStripeMapHeader_t *new_asm_h[2];
  RF_StripeNum_t parityStripeID;
  char *sosBuffer, *eosBuffer;
  RF_ReconUnitNum_t which_ru;
  RF_RaidLayout_t *layoutPtr;
  RF_PhysDiskAddr_t *pda;

  layoutPtr = &(raidPtr->Layout);
  parityStripeID = rf_RaidAddressToParityStripeID(layoutPtr, asmap->raidAddress,
    &which_ru);

  if (rf_dagDebug) {
    printf("[Creating large-write DAG]\n");
  }
  dag_h->creator = "LargeWriteDAG";

  dag_h->numCommitNodes = 1;
  dag_h->numCommits = 0;
  dag_h->numSuccedents = 1;

  /* alloc the nodes: Wnd, xor, commit, block, term, and  Wnp */
  nWndNodes = asmap->numStripeUnitsAccessed;
  RF_CallocAndAdd(nodes, nWndNodes + 4 + nfaults, sizeof(RF_DagNode_t),
    (RF_DagNode_t *), allocList);
  i = 0;
  wndNodes    = &nodes[i]; i += nWndNodes;
  xorNode     = &nodes[i]; i += 1;
  wnpNode     = &nodes[i]; i += 1;
  blockNode   = &nodes[i]; i += 1;
  commitNode  = &nodes[i]; i += 1;
  termNode    = &nodes[i]; i += 1;
  if (nfaults == 2) {
    wnqNode   = &nodes[i]; i += 1;
  }
  else {
    wnqNode = NULL;
  }
  rf_MapUnaccessedPortionOfStripe(raidPtr, layoutPtr, asmap, dag_h, new_asm_h,
    &nRodNodes, &sosBuffer, &eosBuffer, allocList);
  if (nRodNodes > 0) {
    RF_CallocAndAdd(rodNodes, nRodNodes, sizeof(RF_DagNode_t),
      (RF_DagNode_t *), allocList);
  }
  else {
    rodNodes = NULL;
  }

  /* begin node initialization */
  if (nRodNodes > 0) {
    rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
      NULL, nRodNodes, 0, 0, 0, dag_h, "Nil", allocList);
  }
  else {
    rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
      NULL, 1, 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(termNode, rf_wait, RF_FALSE, rf_TerminateFunc, rf_TerminateUndoFunc, NULL,
    0, nWndNodes + nfaults, 0, 0, dag_h, "Trm", allocList);

  /* initialize the Rod nodes */
  for (nodeNum = asmNum = 0; asmNum < 2; asmNum++) {
    if (new_asm_h[asmNum]) {
      pda = new_asm_h[asmNum]->stripeMap->physInfo;
      while (pda) {
        rf_InitNode(&rodNodes[nodeNum], rf_wait, RF_FALSE, rf_DiskReadFunc,
          rf_DiskReadUndoFunc,rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h,
          "Rod", allocList);
        rodNodes[nodeNum].params[0].p = pda;
        rodNodes[nodeNum].params[1].p = pda->bufPtr;
        rodNodes[nodeNum].params[2].v = parityStripeID;
        rodNodes[nodeNum].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
          0, 0, which_ru);
        nodeNum++;
        pda = pda->next;
      }
    }
  }
  RF_ASSERT(nodeNum == nRodNodes);

  /* initialize the wnd nodes */
  pda = asmap->physInfo;
  for (i=0; i < nWndNodes; i++) {
    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 != NULL);
    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);
    pda = pda->next;
  }

  /* initialize the redundancy node */
  if (nRodNodes > 0) {
    rf_InitNode(xorNode, rf_wait, RF_FALSE, redFunc, rf_NullNodeUndoFunc, NULL, 1,
      nRodNodes, 2 * (nWndNodes+nRodNodes) + 1, nfaults, dag_h,
      "Xr ", allocList);
  }
  else {
    rf_InitNode(xorNode, rf_wait, RF_FALSE, redFunc, rf_NullNodeUndoFunc, NULL, 1,
      1, 2 * (nWndNodes+nRodNodes) + 1, nfaults, dag_h, "Xr ", allocList);
  }
  xorNode->flags |= RF_DAGNODE_FLAG_YIELD;
  for (i=0; i < nWndNodes; i++) {
    xorNode->params[2*i+0] = wndNodes[i].params[0];         /* pda */
    xorNode->params[2*i+1] = wndNodes[i].params[1];         /* buf ptr */
  }
  for (i=0; i < nRodNodes; i++) {
    xorNode->params[2*(nWndNodes+i)+0] = rodNodes[i].params[0];  /* pda */
    xorNode->params[2*(nWndNodes+i)+1] = rodNodes[i].params[1];  /* buf ptr */
  }
  /* xor node needs to get at RAID information */
  xorNode->params[2*(nWndNodes+nRodNodes)].p = raidPtr;

  /*
   * Look for an Rod node that reads a complete SU. If none, alloc a buffer
   * to receive the parity info. Note that we can't use a new data buffer
   * because it will not have gotten written when the xor occurs.
   */
  if (allowBufferRecycle) {
    for (i = 0; i < nRodNodes; i++) {
      if (((RF_PhysDiskAddr_t *)rodNodes[i].params[0].p)->numSector == raidPtr->Layout.sectorsPerStripeUnit)
        break;
    }
  }
  if ((!allowBufferRecycle) || (i == nRodNodes)) {
    RF_CallocAndAdd(xorNode->results[0], 1,
      rf_RaidAddressToByte(raidPtr, raidPtr->Layout.sectorsPerStripeUnit),
      (void *), allocList);
  }
  else {
    xorNode->results[0] = rodNodes[i].params[1].p;
  }

  /* initialize 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 = asmap->parityInfo;
  wnpNode->params[1].p = xorNode->results[0];
  wnpNode->params[2].v = parityStripeID;
  wnpNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
  /* parityInfo must describe entire parity unit */
  RF_ASSERT(asmap->parityInfo->next == NULL);

  if (nfaults == 2) {
      /*
       * We never try to recycle a buffer for the Q calcuation
       * in addition to the parity. This would cause two buffers
       * to get smashed during the P and Q calculation, guaranteeing
       * one would be wrong.
       */
      RF_CallocAndAdd(xorNode->results[1], 1,
        rf_RaidAddressToByte(raidPtr, raidPtr->Layout.sectorsPerStripeUnit),
        (void *),allocList);
      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 = asmap->qInfo;
      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);
      /* parityInfo must describe entire parity unit */
      RF_ASSERT(asmap->parityInfo->next == NULL);
  }

  /*
   * Connect nodes to form graph.
   */

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

  if (nRodNodes > 0) {
    /* connect the block node to the Rod nodes */
    RF_ASSERT(blockNode->numSuccedents == nRodNodes);
    RF_ASSERT(xorNode->numAntecedents == nRodNodes);
    for (i = 0; i < nRodNodes; i++) {
      RF_ASSERT(rodNodes[i].numAntecedents == 1);
      blockNode->succedents[i] = &rodNodes[i];
      rodNodes[i].antecedents[0] = blockNode;
      rodNodes[i].antType[0] = rf_control;

      /* connect the Rod nodes to the Xor node */
      RF_ASSERT(rodNodes[i].numSuccedents == 1);
      rodNodes[i].succedents[0] = xorNode;
      xorNode->antecedents[i] = &rodNodes[i];
      xorNode->antType[i] = rf_trueData;
    }
  }
  else {
    /* connect the block node to the Xor node */
    RF_ASSERT(blockNode->numSuccedents == 1);
    RF_ASSERT(xorNode->numAntecedents == 1);
    blockNode->succedents[0] = xorNode;
    xorNode->antecedents[0] = blockNode;
    xorNode->antType[0] = rf_control;
  }

  /* connect the xor node to the 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;

  /* connect the commit node to the write nodes */
  RF_ASSERT(commitNode->numSuccedents == nWndNodes + nfaults);
  for (i = 0; i < nWndNodes; i++) {
    RF_ASSERT(wndNodes->numAntecedents == 1);
    commitNode->succedents[i] = &wndNodes[i];
    wndNodes[i].antecedents[0] = commitNode;
    wndNodes[i].antType[0] = rf_control;
  }
  RF_ASSERT(wnpNode->numAntecedents == 1);
  commitNode->succedents[nWndNodes] = wnpNode;
  wnpNode->antecedents[0]= commitNode;
  wnpNode->antType[0] = rf_trueData;
  if (nfaults == 2) {
    RF_ASSERT(wnqNode->numAntecedents == 1);
    commitNode->succedents[nWndNodes + 1] = wnqNode;
    wnqNode->antecedents[0] = commitNode;
    wnqNode->antType[0] = rf_trueData;
  }

  /* connect the write nodes to the term node */
  RF_ASSERT(termNode->numAntecedents == nWndNodes + nfaults);
  RF_ASSERT(termNode->numSuccedents == 0);
  for (i = 0; i < nWndNodes; i++) {
    RF_ASSERT(wndNodes->numSuccedents == 1);
    wndNodes[i].succedents[0] = termNode;
    termNode->antecedents[i] = &wndNodes[i];
    termNode->antType[i] = rf_control;
  }
  RF_ASSERT(wnpNode->numSuccedents == 1);
  wnpNode->succedents[0] = termNode;
  termNode->antecedents[nWndNodes] = wnpNode;
  termNode->antType[nWndNodes] = rf_control;
  if (nfaults == 2) {
    RF_ASSERT(wnqNode->numSuccedents == 1);
    wnqNode->succedents[0] = termNode;
    termNode->antecedents[nWndNodes + 1] = wnqNode;
    termNode->antType[nWndNodes + 1] = rf_control;
  }
}

/******************************************************************************
 *
 * creates a DAG to perform a small-write operation (either raid 5 or pq),
 * which is as follows:
 *
 * Hdr -> Nil -> Rop -> Xor -> Cmt ----> Wnp [Unp] --> Trm
 *            \- Rod X      /     \----> Wnd [Und]-/
 *           [\- Rod X     /       \---> Wnd [Und]-/]
 *           [\- Roq -> Q /         \--> Wnq [Unq]-/]
 *
 * Rop = read old parity
 * Rod = read old data
 * Roq = read old "q"
 * Cmt = commit node
 * Und = unlock data disk
 * Unp = unlock parity disk
 * Unq = unlock q disk
 * Wnp = write new parity
 * Wnd = write new data
 * Wnq = write new "q"
 * [ ] denotes optional segments in the graph
 *
 * Parameters:  raidPtr   - description of the physical array
 *              asmap     - logical & physical addresses for this access
 *              bp        - buffer ptr (holds write data)
 *              flags     - general flags (e.g. disk locking)
 *              allocList - list of memory allocated in DAG creation
 *              pfuncs    - list of parity generating functions
 *              qfuncs    - list of q generating functions
 *
 * A null qfuncs indicates single fault tolerant
 *****************************************************************************/

void rf_CommonCreateSmallWriteDAG(
  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)
{
  RF_DagNode_t *readDataNodes, *readParityNodes, *readQNodes, *termNode;
  RF_DagNode_t *unlockDataNodes, *unlockParityNodes, *unlockQNodes;
  RF_DagNode_t *xorNodes, *qNodes, *blockNode, *commitNode, *nodes;
  RF_DagNode_t *writeDataNodes, *writeParityNodes, *writeQNodes;
  int i, j, nNodes, totalNumNodes, lu_flag;
  RF_ReconUnitNum_t which_ru;
  int (*func)(RF_DagNode_t *), (*undoFunc)(RF_DagNode_t *);
  int (*qfunc)(RF_DagNode_t *);
  int numDataNodes, numParityNodes;
  RF_StripeNum_t parityStripeID;
  RF_PhysDiskAddr_t *pda;
  char *name, *qname;
  long nfaults;

  nfaults = qfuncs ? 2 : 1;
  lu_flag = (rf_enableAtomicRMW) ? 1 : 0; /* lock/unlock flag */

  parityStripeID = rf_RaidAddressToParityStripeID(&(raidPtr->Layout),
    asmap->raidAddress, &which_ru);
  pda = asmap->physInfo;
  numDataNodes = asmap->numStripeUnitsAccessed;
  numParityNodes = (asmap->parityInfo->next) ? 2 : 1;

  if (rf_dagDebug) {
    printf("[Creating small-write DAG]\n");
  }
  RF_ASSERT(numDataNodes > 0);
  dag_h->creator = "SmallWriteDAG";

  dag_h->numCommitNodes = 1;
  dag_h->numCommits = 0;
  dag_h->numSuccedents = 1;

  /*
   * DAG creation occurs in four steps:
   * 1. count the number of nodes in the DAG
   * 2. create the nodes
   * 3. initialize the nodes
   * 4. connect the nodes
   */

  /*
   * Step 1. compute number of nodes in the graph
   */

  /* number of nodes:
   *  a read and write for each data unit
   *  a redundancy computation node for each parity node (nfaults * nparity)
   *  a read and write for each parity unit
   *  a block and commit node (2)
   *  a terminate node
   *  if atomic RMW
   *    an unlock node for each data unit, redundancy unit
   */
  totalNumNodes = (2 * numDataNodes) + (nfaults * numParityNodes)
    + (nfaults * 2 * numParityNodes) + 3;
  if (lu_flag) {
    totalNumNodes += (numDataNodes + (nfaults * numParityNodes));
  }

  /*
   * Step 2. create the nodes
   */
  RF_CallocAndAdd(nodes, totalNumNodes, sizeof(RF_DagNode_t),
    (RF_DagNode_t *), allocList);
  i = 0;
  blockNode        = &nodes[i]; i += 1;
  commitNode       = &nodes[i]; i += 1;
  readDataNodes    = &nodes[i]; i += numDataNodes;
  readParityNodes  = &nodes[i]; i += numParityNodes;
  writeDataNodes   = &nodes[i]; i += numDataNodes;
  writeParityNodes = &nodes[i]; i += numParityNodes;
  xorNodes         = &nodes[i]; i += numParityNodes;
  termNode         = &nodes[i]; i += 1;
  if (lu_flag) {
    unlockDataNodes   = &nodes[i]; i += numDataNodes;
    unlockParityNodes = &nodes[i]; i += numParityNodes;
  }
  else {
    unlockDataNodes = unlockParityNodes = NULL;
  }
  if (nfaults == 2) {
    readQNodes     = &nodes[i]; i += numParityNodes;
    writeQNodes    = &nodes[i]; i += numParityNodes;
    qNodes         = &nodes[i]; i += numParityNodes;
    if (lu_flag) {
      unlockQNodes    = &nodes[i]; i += numParityNodes;
    }
    else {
      unlockQNodes = NULL;
    }
  }
  else {
    readQNodes = writeQNodes = qNodes = unlockQNodes = NULL;
  }
  RF_ASSERT(i == totalNumNodes);

  /*
   * Step 3. initialize the nodes
   */
  /* initialize block node (Nil) */
  nNodes     = numDataNodes + (nfaults * numParityNodes);
  rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
    NULL, nNodes, 0, 0, 0, dag_h, "Nil", allocList);

  /* initialize commit node (Cmt) */
  rf_InitNode(commitNode, rf_wait, RF_TRUE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
    NULL, nNodes, (nfaults * numParityNodes), 0, 0, dag_h, "Cmt", allocList);

  /* initialize terminate node (Trm) */
  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);
    xorNodes[i].succedents[0] = commitNode;
    commitNode->antecedents[i] = &xorNodes[i];
    commitNode->antType[i] = rf_control;
  }

  /* connect q nodes to commit node */
  if (nfaults == 2) {
    for (i = 0; i < numParityNodes; i++) {
      RF_ASSERT(qNodes[i].numSuccedents == 1);
      qNodes[i].succedents[0] = commitNode;
      commitNode->antecedents[i + numParityNodes] = &qNodes[i];
      commitNode->antType[i + numParityNodes] = rf_control;
    }
  }

  /* connect commit node to write nodes */
  RF_ASSERT(commitNode->numSuccedents == (numDataNodes + (nfaults * numParityNodes)));
  for (i = 0; i < numDataNodes; i++) {
    RF_ASSERT(writeDataNodes[i].numAntecedents == 1);
    commitNode->succedents[i] = &writeDataNodes[i];
    writeDataNodes[i].antecedents[0] = commitNode;
    writeDataNodes[i].antType[0] = rf_trueData;
  }
  for (i = 0; i < numParityNodes; i++) {
    RF_ASSERT(writeParityNodes[i].numAntecedents == 1);
    commitNode->succedents[i + numDataNodes] = &writeParityNodes[i];
    writeParityNodes[i].antecedents[0] = commitNode;
    writeParityNodes[i].antType[0] = rf_trueData;
  }
  if (nfaults == 2) {
    for (i = 0; i < numParityNodes; i++) {
      RF_ASSERT(writeQNodes[i].numAntecedents == 1);
      commitNode->succedents[i + numDataNodes + numParityNodes] = &writeQNodes[i];
      writeQNodes[i].antecedents[0] = commitNode;
      writeQNodes[i].antType[0] = rf_trueData;
    }
  }

  RF_ASSERT(termNode->numAntecedents == (numDataNodes + (nfaults * numParityNodes)));
  RF_ASSERT(termNode->numSuccedents == 0);
  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 term node */
      RF_ASSERT(unlockDataNodes[i].numSuccedents == 1);
      unlockDataNodes[i].succedents[0] = termNode;
      termNode->antecedents[i] = &unlockDataNodes[i];
      termNode->antType[i] = rf_control;
    }
    else {
      /* connect write new data nodes to term node */
      RF_ASSERT(writeDataNodes[i].numSuccedents == 1);
      RF_ASSERT(termNode->numAntecedents == (numDataNodes + (nfaults * numParityNodes)));
      writeDataNodes[i].succedents[0] = termNode;
      termNode->antecedents[i] = &writeDataNodes[i];
      termNode->antType[i] = rf_control;
    }
  }

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

      /* connect unlock nodes to term node */
      RF_ASSERT(unlockParityNodes[i].numSuccedents == 1);
      unlockParityNodes[i].succedents[0] = termNode;
      termNode->antecedents[numDataNodes + i] = &unlockParityNodes[i];
      termNode->antType[numDataNodes + i] = rf_control;
    }
    else {
      RF_ASSERT(writeParityNodes[i].numSuccedents == 1);
      writeParityNodes[i].succedents[0] = termNode;
      termNode->antecedents[numDataNodes + i] = &writeParityNodes[i];
      termNode->antType[numDataNodes + i] = rf_control;
    }
  }

  if (nfaults == 2) {
    for (i = 0; i < numParityNodes; i++) {
      if (lu_flag) {
        /* connect write new Q nodes to unlock nodes */
        RF_ASSERT(writeQNodes[i].numSuccedents == 1);
        RF_ASSERT(unlockQNodes[i].numAntecedents == 1);
        writeQNodes[i].succedents[0] = &unlockQNodes[i];
        unlockQNodes[i].antecedents[0] = &writeQNodes[i];
        unlockQNodes[i].antType[0] = rf_control;

        /* connect unlock nodes to unblock node */
        RF_ASSERT(unlockQNodes[i].numSuccedents == 1);
        unlockQNodes[i].succedents[0] = termNode;
        termNode->antecedents[numDataNodes + numParityNodes + i] = &unlockQNodes[i];
        termNode->antType[numDataNodes + numParityNodes + i] = rf_control;
      }
      else {
        RF_ASSERT(writeQNodes[i].numSuccedents == 1);
        writeQNodes[i].succedents[0] = termNode;
        termNode->antecedents[numDataNodes + numParityNodes + i] = &writeQNodes[i];
        termNode->antType[numDataNodes + numParityNodes + i] = rf_control;
      }
    }
  }
}


/******************************************************************************
 * create a write graph (fault-free or degraded) for RAID level 1
 *
 * Hdr -> Commit -> Wpd -> Nil -> Trm
 *               -> Wsd ->
 *
 * The "Wpd" node writes data to the primary copy in the mirror pair
 * The "Wsd" node writes data to the secondary copy in the mirror pair
 *
 * Parameters:  raidPtr   - description of the physical array
 *              asmap     - logical & physical addresses for this access
 *              bp        - buffer ptr (holds write data)
 *              flags     - general flags (e.g. disk locking)
 *              allocList - list of memory allocated in DAG creation
 *****************************************************************************/

void rf_CreateRaidOneWriteDAG(
  RF_Raid_t             *raidPtr,
  RF_AccessStripeMap_t  *asmap,
  RF_DagHeader_t        *dag_h,
  void                  *bp,
  RF_RaidAccessFlags_t   flags,
  RF_AllocListElem_t    *allocList)
{
  RF_DagNode_t *unblockNode, *termNode, *commitNode;
  RF_DagNode_t *nodes, *wndNode, *wmirNode;
  int nWndNodes, nWmirNodes, i;
  RF_ReconUnitNum_t which_ru;
  RF_PhysDiskAddr_t *pda, *pdaP;
  RF_StripeNum_t parityStripeID;

  parityStripeID = rf_RaidAddressToParityStripeID(&(raidPtr->Layout),
    asmap->raidAddress, &which_ru);
  if (rf_dagDebug) {
    printf("[Creating RAID level 1 write DAG]\n");
  }
  dag_h->creator = "RaidOneWriteDAG";

  /* 2 implies access not SU aligned */
  nWmirNodes = (asmap->parityInfo->next) ? 2 : 1;
  nWndNodes =  (asmap->physInfo->next) ? 2 : 1;

  /* alloc the Wnd nodes and the Wmir node */
  if (asmap->numDataFailed == 1)
    nWndNodes--;
  if (asmap->numParityFailed == 1)
    nWmirNodes--;

  /* total number of nodes = nWndNodes + nWmirNodes + (commit + unblock + terminator) */
  RF_CallocAndAdd(nodes, nWndNodes + nWmirNodes + 3, sizeof(RF_DagNode_t),
    (RF_DagNode_t *), allocList);
  i = 0;
  wndNode     = &nodes[i]; i += nWndNodes;
  wmirNode    = &nodes[i]; i += nWmirNodes;
  commitNode   = &nodes[i]; i += 1;
  unblockNode = &nodes[i]; i += 1;
  termNode = &nodes[i]; i += 1;
  RF_ASSERT(i == (nWndNodes + nWmirNodes + 3));

  /* this dag can commit immediately */
  dag_h->numCommitNodes = 1;
  dag_h->numCommits = 0;
  dag_h->numSuccedents = 1;

  /* initialize the commit, unblock, and term nodes */
  rf_InitNode(commitNode, rf_wait, RF_TRUE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
    NULL, (nWndNodes + nWmirNodes), 0, 0, 0, dag_h, "Cmt", allocList);
  rf_InitNode(unblockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
    NULL, 1, (nWndNodes + nWmirNodes), 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);

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

  /* initialize the mirror nodes */
  if (nWmirNodes > 0) {
    pda = asmap->physInfo;
    pdaP = asmap->parityInfo;
    for (i = 0; i < nWmirNodes; i++) {
      rf_InitNode(&wmirNode[i], rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
        rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wsd", allocList);
      RF_ASSERT(pda != NULL);
      wmirNode[i].params[0].p = pdaP;
      wmirNode[i].params[1].p = pda->bufPtr;
      wmirNode[i].params[2].v = parityStripeID;
      wmirNode[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
      pda = pda->next;
      pdaP = pdaP->next;
    }
    RF_ASSERT(pda == NULL);
    RF_ASSERT(pdaP == NULL);
  }

  /* link the header node to the commit node */
  RF_ASSERT(dag_h->numSuccedents == 1);
  RF_ASSERT(commitNode->numAntecedents == 0);
  dag_h->succedents[0] = commitNode;

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

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

  /* link the unblock node to the 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;
}



/* DAGs which have no commit points.
 *
 * The following DAGs are used in forward and backward error recovery experiments.
 * They are identical to the DAGs above this comment with the exception that the
 * the commit points have been removed.
 */



void rf_CommonCreateLargeWriteDAGFwd(
  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)(RF_DagNode_t *),
  int                    allowBufferRecycle)
{
  RF_DagNode_t *nodes, *wndNodes, *rodNodes, *xorNode, *wnpNode;
  RF_DagNode_t *wnqNode, *blockNode, *syncNode, *termNode;
  int nWndNodes, nRodNodes, i, nodeNum, asmNum;
  RF_AccessStripeMapHeader_t *new_asm_h[2];
  RF_StripeNum_t parityStripeID;
  char *sosBuffer, *eosBuffer;
  RF_ReconUnitNum_t which_ru;
  RF_RaidLayout_t *layoutPtr;
  RF_PhysDiskAddr_t *pda;

  layoutPtr = &(raidPtr->Layout);
  parityStripeID = rf_RaidAddressToParityStripeID(&(raidPtr->Layout), asmap->raidAddress, &which_ru);

  if (rf_dagDebug)
    printf("[Creating large-write DAG]\n");
  dag_h->creator = "LargeWriteDAGFwd";

  dag_h->numCommitNodes = 0;
  dag_h->numCommits = 0;
  dag_h->numSuccedents = 1;

  /* alloc the nodes: Wnd, xor, commit, block, term, and  Wnp */
  nWndNodes = asmap->numStripeUnitsAccessed;
  RF_CallocAndAdd(nodes, nWndNodes + 4 + nfaults, sizeof(RF_DagNode_t), (RF_DagNode_t *), allocList);
  i = 0;
  wndNodes    = &nodes[i]; i += nWndNodes;
  xorNode     = &nodes[i]; i += 1;
  wnpNode     = &nodes[i]; i += 1;
  blockNode   = &nodes[i]; i += 1;
  syncNode  = &nodes[i]; i += 1;
  termNode    = &nodes[i]; i += 1;
  if (nfaults == 2) {
    wnqNode   = &nodes[i]; i += 1;
  }
  else {
    wnqNode = NULL;
  }
  rf_MapUnaccessedPortionOfStripe(raidPtr, layoutPtr, asmap, dag_h, new_asm_h, &nRodNodes, &sosBuffer, &eosBuffer, allocList);
  if (nRodNodes > 0) {
    RF_CallocAndAdd(rodNodes, nRodNodes, sizeof(RF_DagNode_t), (RF_DagNode_t *), allocList);
  }
  else {
    rodNodes = NULL;
  }

  /* begin node initialization */
  if (nRodNodes > 0) {
    rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, nRodNodes, 0, 0, 0, dag_h, "Nil", allocList);
    rf_InitNode(syncNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, nWndNodes + 1, nRodNodes, 0, 0, dag_h, "Nil", allocList);
  }
  else {
    rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, 1, 0, 0, 0, dag_h, "Nil", allocList);
    rf_InitNode(syncNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, nWndNodes + 1, 1, 0, 0, dag_h, "Nil", allocList);
  }

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

  /* initialize the Rod nodes */
  for (nodeNum = asmNum = 0; asmNum < 2; asmNum++) {
    if (new_asm_h[asmNum]) {
      pda = new_asm_h[asmNum]->stripeMap->physInfo;
      while (pda) {
	rf_InitNode(&rodNodes[nodeNum], rf_wait, RF_FALSE, rf_DiskReadFunc, rf_DiskReadUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Rod", allocList);
	rodNodes[nodeNum].params[0].p = pda;
	rodNodes[nodeNum].params[1].p = pda->bufPtr;
	rodNodes[nodeNum].params[2].v = parityStripeID;
	rodNodes[nodeNum].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
	nodeNum++;
	pda=pda->next;
      }
    }
  }
  RF_ASSERT(nodeNum == nRodNodes);

  /* initialize the wnd nodes */
  pda = asmap->physInfo;
  for (i=0; i < nWndNodes; i++) {
    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 != NULL);
    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);
    pda = pda->next;
  }

  /* initialize the redundancy node */
  rf_InitNode(xorNode, rf_wait, RF_FALSE, redFunc, rf_NullNodeUndoFunc, NULL, 1, nfaults, 2 * (nWndNodes + nRodNodes) + 1, nfaults, dag_h, "Xr ", allocList);
  xorNode->flags |= RF_DAGNODE_FLAG_YIELD;
  for (i=0; i < nWndNodes; i++) {
    xorNode->params[2*i+0] = wndNodes[i].params[0];         /* pda */
    xorNode->params[2*i+1] = wndNodes[i].params[1];         /* buf ptr */
  }
  for (i=0; i < nRodNodes; i++) {
    xorNode->params[2*(nWndNodes+i)+0] = rodNodes[i].params[0];         /* pda */
    xorNode->params[2*(nWndNodes+i)+1] = rodNodes[i].params[1];         /* buf ptr */
  }
  xorNode->params[2*(nWndNodes+nRodNodes)].p = raidPtr; /* xor node needs to get at RAID information */

  /* look for an Rod node that reads a complete SU.  If none, alloc a buffer to receive the parity info.
   * Note that we can't use a new data buffer because it will not have gotten written when the xor occurs.
   */
  if (allowBufferRecycle) {
    for (i = 0; i < nRodNodes; i++)
      if (((RF_PhysDiskAddr_t *) rodNodes[i].params[0].p)->numSector == raidPtr->Layout.sectorsPerStripeUnit)
        break;
  }
  if ((!allowBufferRecycle) || (i == nRodNodes)) {
    RF_CallocAndAdd(xorNode->results[0], 1, rf_RaidAddressToByte(raidPtr, raidPtr->Layout.sectorsPerStripeUnit), (void *), allocList);
  }
  else
    xorNode->results[0] = rodNodes[i].params[1].p;

  /* initialize 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 = asmap->parityInfo;
  wnpNode->params[1].p = xorNode->results[0];
  wnpNode->params[2].v = parityStripeID;
  wnpNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
  RF_ASSERT(asmap->parityInfo->next == NULL);        /* parityInfo must describe entire parity unit */

  if (nfaults == 2)
    {
      /* we never try to recycle a buffer for the Q calcuation in addition to the parity.
	 This would cause two buffers to get smashed during the P and Q calculation,
	 guaranteeing one would be wrong.
      */
      RF_CallocAndAdd(xorNode->results[1], 1, rf_RaidAddressToByte(raidPtr, raidPtr->Layout.sectorsPerStripeUnit), (void *), allocList);
      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 = asmap->qInfo;
      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);
      RF_ASSERT(asmap->parityInfo->next == NULL);        /* parityInfo must describe entire parity unit */
    }


  /* connect nodes to form graph */

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

  if (nRodNodes > 0) {
    /* connect the block node to the Rod nodes */
    RF_ASSERT(blockNode->numSuccedents == nRodNodes);
    RF_ASSERT(syncNode->numAntecedents == nRodNodes);
    for (i = 0; i < nRodNodes; i++) {
      RF_ASSERT(rodNodes[i].numAntecedents == 1);
      blockNode->succedents[i] = &rodNodes[i];
      rodNodes[i].antecedents[0] = blockNode;
      rodNodes[i].antType[0] = rf_control;

      /* connect the Rod nodes to the Nil node */
      RF_ASSERT(rodNodes[i].numSuccedents == 1);
      rodNodes[i].succedents[0] = syncNode;
      syncNode->antecedents[i] = &rodNodes[i];
      syncNode->antType[i] = rf_trueData;
    }
  }
  else {
    /* connect the block node to the Nil node */
    RF_ASSERT(blockNode->numSuccedents == 1);
    RF_ASSERT(syncNode->numAntecedents == 1);
    blockNode->succedents[0] = syncNode;
    syncNode->antecedents[0] = blockNode;
    syncNode->antType[0] = rf_control;
  }

  /* connect the sync node to the Wnd nodes */
  RF_ASSERT(syncNode->numSuccedents == (1 + nWndNodes));
  for (i = 0; i < nWndNodes; i++) {
    RF_ASSERT(wndNodes->numAntecedents == 1);
    syncNode->succedents[i] = &wndNodes[i];
    wndNodes[i].antecedents[0] = syncNode;
    wndNodes[i].antType[0] = rf_control;
  }

  /* connect the sync node to the Xor node */
  RF_ASSERT(xorNode->numAntecedents == 1);
  syncNode->succedents[nWndNodes] = xorNode;
  xorNode->antecedents[0] = syncNode;
  xorNode->antType[0] = rf_control;

  /* connect the xor node to the write parity node */
  RF_ASSERT(xorNode->numSuccedents == nfaults);
  RF_ASSERT(wnpNode->numAntecedents == 1);
  xorNode->succedents[0] = wnpNode;
  wnpNode->antecedents[0]= xorNode;
  wnpNode->antType[0] = rf_trueData;
  if (nfaults == 2) {
    RF_ASSERT(wnqNode->numAntecedents == 1);
    xorNode->succedents[1] = wnqNode;
    wnqNode->antecedents[0] = xorNode;
    wnqNode->antType[0] = rf_trueData;
  }

  /* connect the write nodes to the term node */
  RF_ASSERT(termNode->numAntecedents == nWndNodes + nfaults);
  RF_ASSERT(termNode->numSuccedents == 0);
  for (i = 0; i < nWndNodes; i++) {
    RF_ASSERT(wndNodes->numSuccedents == 1);
    wndNodes[i].succedents[0] = termNode;
    termNode->antecedents[i] = &wndNodes[i];
    termNode->antType[i] = rf_control;
  }
  RF_ASSERT(wnpNode->numSuccedents == 1);
  wnpNode->succedents[0] = termNode;
  termNode->antecedents[nWndNodes] = wnpNode;
  termNode->antType[nWndNodes] = rf_control;
  if (nfaults == 2) {
    RF_ASSERT(wnqNode->numSuccedents == 1);
    wnqNode->succedents[0] = termNode;
    termNode->antecedents[nWndNodes + 1] = wnqNode;
    termNode->antType[nWndNodes + 1] = rf_control;
  }
}


/******************************************************************************
 *
 * creates a DAG to perform a small-write operation (either raid 5 or pq),
 * which is as follows:
 *
 * Hdr -> Nil -> Rop - Xor - Wnp [Unp] -- Trm
 *            \- Rod X- Wnd [Und] -------/
 *           [\- Rod X- Wnd [Und] ------/]
 *           [\- Roq - Q --> Wnq [Unq]-/]
 *
 * Rop = read old parity
 * Rod = read old data
 * Roq = read old "q"
 * Cmt = commit node
 * Und = unlock data disk
 * Unp = unlock parity disk
 * Unq = unlock q disk
 * Wnp = write new parity
 * Wnd = write new data
 * Wnq = write new "q"
 * [ ] denotes optional segments in the graph
 *
 * Parameters:  raidPtr   - description of the physical array
 *              asmap     - logical & physical addresses for this access
 *              bp        - buffer ptr (holds write data)
 *              flags     - general flags (e.g. disk locking)
 *              allocList - list of memory allocated in DAG creation
 *              pfuncs    - list of parity generating functions
 *              qfuncs    - list of q generating functions
 *
 * A null qfuncs indicates single fault tolerant
 *****************************************************************************/

void rf_CommonCreateSmallWriteDAGFwd(
  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)
{
  RF_DagNode_t *readDataNodes, *readParityNodes, *readQNodes, *termNode;
  RF_DagNode_t *unlockDataNodes, *unlockParityNodes, *unlockQNodes;
  RF_DagNode_t *xorNodes, *qNodes, *blockNode, *nodes;
  RF_DagNode_t *writeDataNodes, *writeParityNodes, *writeQNodes;
  int i, j, nNodes, totalNumNodes, lu_flag;
  RF_ReconUnitNum_t which_ru;
  int (*func)(RF_DagNode_t *), (*undoFunc)(RF_DagNode_t *);
  int (*qfunc)(RF_DagNode_t *);
  int numDataNodes, numParityNodes;
  RF_StripeNum_t parityStripeID;
  RF_PhysDiskAddr_t *pda;
  char *name, *qname;
  long nfaults;

  nfaults = qfuncs ? 2 : 1;
  lu_flag = (rf_enableAtomicRMW) ? 1 : 0;          /* lock/unlock flag */

  parityStripeID = rf_RaidAddressToParityStripeID(&(raidPtr->Layout), asmap->raidAddress, &which_ru);
  pda = asmap->physInfo;
  numDataNodes = asmap->numStripeUnitsAccessed;
  numParityNodes = (asmap->parityInfo->next) ? 2 : 1;

  if (rf_dagDebug) printf("[Creating small-write DAG]\n");
  RF_ASSERT(numDataNodes > 0);
  dag_h->creator = "SmallWriteDAGFwd";

  dag_h->numCommitNodes = 0;
  dag_h->numCommits = 0;
  dag_h->numSuccedents = 1;

  qfunc = NULL;
  qname = NULL;

  /* DAG creation occurs in four steps:
     1. count the number of nodes in the DAG
     2. create the nodes
     3. initialize the nodes
     4. connect the nodes
   */

  /* Step 1. compute number of nodes in the graph */

  /* number of nodes:
      a read and write for each data unit
      a redundancy computation node for each parity node (nfaults * nparity)
      a read and write for each parity unit
      a block node
      a terminate node
      if atomic RMW
        an unlock node for each data unit, redundancy unit
  */
  totalNumNodes = (2 * numDataNodes) + (nfaults * numParityNodes) + (nfaults * 2 * numParityNodes) + 2;
  if (lu_flag)
    totalNumNodes += (numDataNodes + (nfaults * numParityNodes));


  /* Step 2. create the nodes */
  RF_CallocAndAdd(nodes, totalNumNodes, sizeof(RF_DagNode_t), (RF_DagNode_t *), allocList);
  i = 0;
  blockNode        = &nodes[i]; i += 1;
  readDataNodes    = &nodes[i]; i += numDataNodes;
  readParityNodes  = &nodes[i]; i += numParityNodes;
  writeDataNodes   = &nodes[i]; i += numDataNodes;
  writeParityNodes = &nodes[i]; i += numParityNodes;
  xorNodes         = &nodes[i]; i += numParityNodes;
  termNode         = &nodes[i]; i += 1;
  if (lu_flag) {
    unlockDataNodes   = &nodes[i]; i += numDataNodes;
    unlockParityNodes = &nodes[i]; i += numParityNodes;
  }
  else {
    unlockDataNodes = unlockParityNodes = NULL;
  }
  if (nfaults == 2) {
    readQNodes     = &nodes[i]; i += numParityNodes;
    writeQNodes    = &nodes[i]; i += numParityNodes;
    qNodes         = &nodes[i]; i += numParityNodes;
    if (lu_flag) {
      unlockQNodes    = &nodes[i]; i += numParityNodes;
    }
    else {
      unlockQNodes = NULL;
    }
  }
  else {
    readQNodes = writeQNodes = qNodes = unlockQNodes = NULL;
  }
  RF_ASSERT(i == totalNumNodes);

  /* Step 3. initialize the nodes */
  /* initialize block node (Nil) */
  nNodes     = numDataNodes + (nfaults * numParityNodes);
  rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, nNodes, 0, 0, 0, dag_h, "Nil", allocList);

  /* initialize terminate node (Trm) */
  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, (numParityNodes * nfaults) + 1, 1, 4, 0, dag_h, "Rod", allocList);
    RF_ASSERT(pda != NULL);
    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;
    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;
    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, lu_flag, 0, which_ru);
    for (j = 0; j < readParityNodes[i].numSuccedents; j++)
      readParityNodes[i].propList[0] = NULL;
    pda=pda->next;
  }

  /* 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;
	readQNodes[i].params[1].p = rf_AllocBuffer(raidPtr, dag_h, pda, allocList); /* buffer to hold old Q */
	readQNodes[i].params[2].v = parityStripeID;
	readQNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, lu_flag, 0, which_ru);
	for (j = 0; j < readQNodes[i].numSuccedents; j++)
	  readQNodes[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_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 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 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 {
    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_FALSE, func, undoFunc, NULL, numParityNodes, numParityNodes + numDataNodes, 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 */
      if (nfaults==2)
	{
	  rf_InitNode(&qNodes[i], rf_wait, RF_FALSE, qfunc, undoFunc, NULL, numParityNodes, numParityNodes + numDataNodes, 7, 1, dag_h, qname, allocList);  /* no wakeup func for xor */
	  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;
	  qNodes[i].results[0] = readQNodes[i].params[1].p;   /* use old Q buf as target buf */
	}
    }
  }
  else {
    /* there is only one xor node in this case */
    rf_InitNode(&xorNodes[0], rf_wait, RF_FALSE, func, undoFunc, NULL, numParityNodes, numParityNodes + numDataNodes, (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;
    if (nfaults==2)
      {
	rf_InitNode(&qNodes[0], rf_wait, RF_FALSE, qfunc, undoFunc, NULL, numParityNodes, numParityNodes + numDataNodes, (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 pointer */
	}
	/* and read old q */
	qNodes[0].params[2*numDataNodes + 0] = readQNodes[0].params[0];    /* pda */
	qNodes[0].params[2*numDataNodes + 1] = readQNodes[0].params[1];    /* buffer pointer */
	for (i=0; i < numDataNodes; i++) {
	  /* set up params related to Wnd nodes */
	  qNodes[0].params[2*(numDataNodes+1+i)+0] = writeDataNodes[i].params[0]; /* pda */
	  qNodes[0].params[2*(numDataNodes+1+i)+1] = writeDataNodes[i].params[1]; /* buffer pointer */
	}
	qNodes[0].params[2*(numDataNodes+numDataNodes+1)].p = raidPtr;  /* xor node needs to get at RAID information */
	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, numParityNodes, 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, numParityNodes, 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 write new data nodes */
  for (i = 0; i < numDataNodes; i++) {
    RF_ASSERT(readDataNodes[i].numSuccedents == ((nfaults * numParityNodes) + 1));
    RF_ASSERT(writeDataNodes[i].numAntecedents == 1);
    readDataNodes[i].succedents[0] = &writeDataNodes[i];
    writeDataNodes[i].antecedents[0] = &readDataNodes[i];
    writeDataNodes[i].antType[0] = rf_antiData;
  }

  /* 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[1 + 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[1 + 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++) {
    for (j = 0; j < numParityNodes; j++) {
      RF_ASSERT(readParityNodes[i].numSuccedents == numParityNodes);
      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++) {
      for (j = 0; j < numParityNodes; j++) {
	RF_ASSERT(readQNodes[i].numSuccedents == numParityNodes);
	readQNodes[i].succedents[j] = &qNodes[j];
	qNodes[j].antecedents[numDataNodes + i] = &readQNodes[i];
	qNodes[j].antType[numDataNodes + i] = rf_trueData;
      }
    }

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

  /* connect q nodes to the write new q nodes */
  if (nfaults == 2)
    for (i = 0; i < numParityNodes; i++) {
      RF_ASSERT(writeQNodes[i].numAntecedents == numParityNodes);
      for (j = 0; j < numParityNodes; j++) {
	RF_ASSERT(qNodes[j].numSuccedents == 1);
	qNodes[i].succedents[j] = &writeQNodes[j];
	writeQNodes[j].antecedents[i] = &qNodes[i];
	writeQNodes[j].antType[i] = rf_trueData;
      }
    }

  RF_ASSERT(termNode->numAntecedents == (numDataNodes + (nfaults * numParityNodes)));
  RF_ASSERT(termNode->numSuccedents == 0);
  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 term node */
      RF_ASSERT(unlockDataNodes[i].numSuccedents == 1);
      unlockDataNodes[i].succedents[0] = termNode;
      termNode->antecedents[i] = &unlockDataNodes[i];
      termNode->antType[i] = rf_control;
    }
    else {
      /* connect write new data nodes to term node */
      RF_ASSERT(writeDataNodes[i].numSuccedents == 1);
      RF_ASSERT(termNode->numAntecedents == (numDataNodes + (nfaults * numParityNodes)));
      writeDataNodes[i].succedents[0] = termNode;
      termNode->antecedents[i] = &writeDataNodes[i];
      termNode->antType[i] = rf_control;
    }
  }

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

      /* connect unlock nodes to term node */
      RF_ASSERT(unlockParityNodes[i].numSuccedents == 1);
      unlockParityNodes[i].succedents[0] = termNode;
      termNode->antecedents[numDataNodes + i] = &unlockParityNodes[i];
      termNode->antType[numDataNodes + i] = rf_control;
    }
    else {
      RF_ASSERT(writeParityNodes[i].numSuccedents == 1);
      writeParityNodes[i].succedents[0] = termNode;
      termNode->antecedents[numDataNodes + i] = &writeParityNodes[i];
      termNode->antType[numDataNodes + i] = rf_control;
    }
  }

  if (nfaults == 2)
    for (i = 0; i < numParityNodes; i++) {
      if (lu_flag) {
	/* connect write new Q nodes to unlock nodes */
	RF_ASSERT(writeQNodes[i].numSuccedents == 1);
	RF_ASSERT(unlockQNodes[i].numAntecedents == 1);
	writeQNodes[i].succedents[0] = &unlockQNodes[i];
	unlockQNodes[i].antecedents[0] = &writeQNodes[i];
	unlockQNodes[i].antType[0] = rf_control;

	/* connect unlock nodes to unblock node */
	RF_ASSERT(unlockQNodes[i].numSuccedents == 1);
	unlockQNodes[i].succedents[0] = termNode;
	termNode->antecedents[numDataNodes + numParityNodes + i] = &unlockQNodes[i];
	termNode->antType[numDataNodes + numParityNodes + i] = rf_control;
      }
      else {
	RF_ASSERT(writeQNodes[i].numSuccedents == 1);
	writeQNodes[i].succedents[0] = termNode;
	termNode->antecedents[numDataNodes + numParityNodes + i] = &writeQNodes[i];
	termNode->antType[numDataNodes + numParityNodes + i] = rf_control;
      }
    }
}



/******************************************************************************
 * create a write graph (fault-free or degraded) for RAID level 1
 *
 * Hdr  Nil -> Wpd -> Nil -> Trm
 *      Nil -> Wsd ->
 *
 * The "Wpd" node writes data to the primary copy in the mirror pair
 * The "Wsd" node writes data to the secondary copy in the mirror pair
 *
 * Parameters:  raidPtr   - description of the physical array
 *              asmap     - logical & physical addresses for this access
 *              bp        - buffer ptr (holds write data)
 *              flags     - general flags (e.g. disk locking)
 *              allocList - list of memory allocated in DAG creation
 *****************************************************************************/

void rf_CreateRaidOneWriteDAGFwd(
  RF_Raid_t             *raidPtr,
  RF_AccessStripeMap_t  *asmap,
  RF_DagHeader_t        *dag_h,
  void                  *bp,
  RF_RaidAccessFlags_t   flags,
  RF_AllocListElem_t    *allocList)
{
  RF_DagNode_t *blockNode, *unblockNode, *termNode;
  RF_DagNode_t *nodes, *wndNode, *wmirNode;
  int nWndNodes, nWmirNodes, i;
  RF_ReconUnitNum_t which_ru;
  RF_PhysDiskAddr_t *pda, *pdaP;
  RF_StripeNum_t parityStripeID;

  parityStripeID = rf_RaidAddressToParityStripeID(&(raidPtr->Layout),
    asmap->raidAddress, &which_ru);
  if (rf_dagDebug) {
    printf("[Creating RAID level 1 write DAG]\n");
  }

  nWmirNodes = (asmap->parityInfo->next) ? 2 : 1;  /* 2 implies access not SU aligned */
  nWndNodes =  (asmap->physInfo->next) ? 2 : 1;

  /* alloc the Wnd nodes and the Wmir node */
  if (asmap->numDataFailed == 1)
    nWndNodes--;
  if (asmap->numParityFailed == 1)
    nWmirNodes--;

  /* total number of nodes = nWndNodes + nWmirNodes + (block + unblock + terminator) */
  RF_CallocAndAdd(nodes, nWndNodes + nWmirNodes + 3, sizeof(RF_DagNode_t), (RF_DagNode_t *), allocList);
  i = 0;
  wndNode     = &nodes[i]; i += nWndNodes;
  wmirNode    = &nodes[i]; i += nWmirNodes;
  blockNode   = &nodes[i]; i += 1;
  unblockNode = &nodes[i]; i += 1;
  termNode = &nodes[i]; i += 1;
  RF_ASSERT(i == (nWndNodes + nWmirNodes + 3));

  /* this dag can commit immediately */
  dag_h->numCommitNodes = 0;
  dag_h->numCommits = 0;
  dag_h->numSuccedents = 1;

  /* initialize the unblock and term nodes */
  rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, (nWndNodes + nWmirNodes), 0, 0, 0, dag_h, "Nil", allocList);
  rf_InitNode(unblockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, 1, (nWndNodes + nWmirNodes), 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);

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

  /* initialize the mirror nodes */
  if (nWmirNodes > 0) {
    pda = asmap->physInfo;
    pdaP = asmap->parityInfo;
    for (i = 0; i < nWmirNodes; i++) {
      rf_InitNode(&wmirNode[i], rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wsd", allocList);
      RF_ASSERT(pda != NULL);
      wmirNode[i].params[0].p = pdaP;
      wmirNode[i].params[1].p = pda->bufPtr;
      wmirNode[i].params[2].v = parityStripeID;
      wmirNode[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
      pda = pda->next;
      pdaP = pdaP->next;
    }
    RF_ASSERT(pda == NULL);
    RF_ASSERT(pdaP == NULL);
  }

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

  /* link the block node to the write nodes */
  RF_ASSERT(blockNode->numSuccedents == (nWndNodes + nWmirNodes));
  for (i = 0; i < nWndNodes; i++) {
    RF_ASSERT(wndNode[i].numAntecedents == 1);
    blockNode->succedents[i] = &wndNode[i];
    wndNode[i].antecedents[0] = blockNode;
    wndNode[i].antType[0] = rf_control;
  }
  for (i = 0; i < nWmirNodes; i++) {
    RF_ASSERT(wmirNode[i].numAntecedents == 1);
    blockNode->succedents[i + nWndNodes] = &wmirNode[i];
    wmirNode[i].antecedents[0] = blockNode;
    wmirNode[i].antType[0] = rf_control;
  }

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

  /* link the unblock node to the 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;

  return;
}