rf_dagdegwr.c

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

      /* build p */
      CONS_PDA(parityInfo,fone_start,fone->numSector);
      pda_p->type = RF_PDA_TYPE_PARITY;
      pda_p++;
      /* build q */
      CONS_PDA(qInfo,fone_start,fone->numSector);
      pda_p->type = RF_PDA_TYPE_Q;
    }
  else
    {
      ftwo_start = rf_StripeUnitOffset(layoutPtr,ftwo->startSector);
      ftwo_end = ftwo_start + ftwo->numSector;    
      if (fone->numSector + ftwo->numSector > secPerSU)
	{
	  PDAPerDisk = 1;
	  state = 2;
	  RF_MallocAndAdd(*pqpdap,2*sizeof(RF_PhysDiskAddr_t),(RF_PhysDiskAddr_t *), allocList);
	  pda_p = *pqpdap;
	  CONS_PDA(parityInfo,0,secPerSU);
	  pda_p->type = RF_PDA_TYPE_PARITY;
	  pda_p++;
	  CONS_PDA(qInfo,0,secPerSU);
	  pda_p->type = RF_PDA_TYPE_Q;
	}
      else
	{
	  PDAPerDisk = 2;
	  state = 3;
	  /* four of them, fone, then ftwo */
	  RF_MallocAndAdd(*pqpdap,4*sizeof(RF_PhysDiskAddr_t),(RF_PhysDiskAddr_t *), allocList);
	  pda_p = *pqpdap;
	  CONS_PDA(parityInfo,fone_start,fone->numSector);
	  pda_p->type = RF_PDA_TYPE_PARITY;
	  pda_p++;
	  CONS_PDA(qInfo,fone_start,fone->numSector);
	  pda_p->type = RF_PDA_TYPE_Q;
	  pda_p++;
	  CONS_PDA(parityInfo,ftwo_start,ftwo->numSector);
	  pda_p->type = RF_PDA_TYPE_PARITY;
	  pda_p++;
	  CONS_PDA(qInfo,ftwo_start,ftwo->numSector);
	  pda_p->type = RF_PDA_TYPE_Q;
	}
    }
  /* figure out number of nonaccessed pda */
  napdas = PDAPerDisk * (numDataCol - 2);
  *nPQNodep = PDAPerDisk;

  *nNodep = napdas;
  if (napdas == 0) return; /* short circuit */

  /* allocate up our list of pda's */
     
  RF_CallocAndAdd(pda_p, napdas, sizeof(RF_PhysDiskAddr_t), (RF_PhysDiskAddr_t *), allocList);
  *pdap = pda_p;
    
  /* linkem together */
  for (i=0; i < (napdas-1); i++)
    pda_p[i].next = pda_p+(i+1);

  sosAddr      = rf_RaidAddressOfPrevStripeBoundary(layoutPtr, asmap->raidAddress);
  for (i=0; i < numDataCol; i++)
    {
      if ((pda_p - (*pdap)) == napdas) 
	continue;
      pda_p->type = RF_PDA_TYPE_DATA;
      pda_p->raidAddress = sosAddr + (i * secPerSU);
      (raidPtr->Layout.map->MapSector)(raidPtr,pda_p->raidAddress, &(pda_p->row), &(pda_p->col), &(pda_p->startSector), 0);
      /* skip over dead disks */
      if (RF_DEAD_DISK(raidPtr->Disks[pda_p->row][pda_p->col].status)) 
	continue;
      switch (state)
	{
	case 1: /* fone */
	  pda_p->numSector = fone->numSector;
	  pda_p->raidAddress += fone_start;
	  pda_p->startSector += fone_start;
	  RF_MallocAndAdd(pda_p->bufPtr, rf_RaidAddressToByte(raidPtr,pda_p->numSector), (char *), allocList);
	  break;
	case 2: /* full stripe */
	  pda_p->numSector = secPerSU;
	  RF_MallocAndAdd(pda_p->bufPtr, rf_RaidAddressToByte(raidPtr,secPerSU), (char *), allocList);
	  break;
	case 3: /* two slabs */
	  pda_p->numSector = fone->numSector;
	  pda_p->raidAddress += fone_start;
	  pda_p->startSector += fone_start;
	  RF_MallocAndAdd(pda_p->bufPtr, rf_RaidAddressToByte(raidPtr,pda_p->numSector), (char *), allocList);
	  pda_p++;
          pda_p->type = RF_PDA_TYPE_DATA;
          pda_p->raidAddress = sosAddr + (i * secPerSU);
          (raidPtr->Layout.map->MapSector)(raidPtr,pda_p->raidAddress, &(pda_p->row), &(pda_p->col), &(pda_p->startSector), 0);
	  pda_p->numSector = ftwo->numSector;
	  pda_p->raidAddress += ftwo_start;
	  pda_p->startSector += ftwo_start;
	  RF_MallocAndAdd(pda_p->bufPtr, rf_RaidAddressToByte(raidPtr,pda_p->numSector), (char *), allocList);
	  break;
	default:
	  RF_PANIC();
	}
      pda_p++;
    }

  RF_ASSERT  (pda_p - *pdap == napdas);
  return;
}

#define DISK_NODE_PDA(node)  ((node)->params[0].p)

#define DISK_NODE_PARAMS(_node_,_p_) \
  (_node_).params[0].p = _p_ ; \
  (_node_).params[1].p = (_p_)->bufPtr; \
  (_node_).params[2].v = parityStripeID; \
  (_node_).params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru)

void rf_DoubleDegSmallWrite(
  RF_Raid_t              *raidPtr,
  RF_AccessStripeMap_t   *asmap,
  RF_DagHeader_t         *dag_h,
  void                   *bp,
  RF_RaidAccessFlags_t    flags,
  RF_AllocListElem_t     *allocList,
  char                   *redundantReadNodeName,
  char                   *redundantWriteNodeName,
  char                   *recoveryNodeName,
  int                   (*recovFunc)())
{
  RF_RaidLayout_t *layoutPtr = &(raidPtr->Layout);
  RF_DagNode_t *nodes, *wudNodes, *rrdNodes, *recoveryNode, *blockNode, *unblockNode, *rpNodes,*rqNodes, *wpNodes, *wqNodes, *termNode;
  RF_PhysDiskAddr_t *pda, *pqPDAs;
  RF_PhysDiskAddr_t *npdas;
  int nWriteNodes, nNodes, nReadNodes, nRrdNodes, nWudNodes, i;
  RF_ReconUnitNum_t which_ru;
  int nPQNodes;
  RF_StripeNum_t parityStripeID = rf_RaidAddressToParityStripeID(layoutPtr, asmap->raidAddress, &which_ru);

  /* simple small write case -
     First part looks like a reconstruct-read of the failed data units.
     Then a write of all data units not failed. */


  /* 
             Hdr
	      |
       ------Block- 
      /  /         \   
    Rrd  Rrd ...  Rrd  Rp Rq
      \  \          /  
      -------PQ-----
            /   \   \
          Wud   Wp  WQ	     
           \    |   /
           --Unblock-
                |
                T

   Rrd = read recovery data  (potentially none)
   Wud = write user data (not incl. failed disks)
   Wp = Write P (could be two)
   Wq = Write Q (could be two)

   */
  
  rf_WriteGenerateFailedAccessASMs(raidPtr, asmap, &npdas, &nRrdNodes, &pqPDAs, &nPQNodes,allocList);

  RF_ASSERT(asmap->numDataFailed == 1);
  
  nWudNodes = asmap->numStripeUnitsAccessed - (asmap->numDataFailed);
  nReadNodes = nRrdNodes + 2*nPQNodes;
  nWriteNodes = nWudNodes+ 2*nPQNodes;
  nNodes = 4 + nReadNodes + nWriteNodes;

  RF_CallocAndAdd(nodes, nNodes, sizeof(RF_DagNode_t), (RF_DagNode_t *), allocList);
  blockNode = nodes;
  unblockNode = blockNode+1;
  termNode = unblockNode+1;
  recoveryNode = termNode+1;
  rrdNodes = recoveryNode+1;
  rpNodes = rrdNodes + nRrdNodes;
  rqNodes = rpNodes + nPQNodes;
  wudNodes = rqNodes + nPQNodes;
  wpNodes = wudNodes + nWudNodes;
  wqNodes = wpNodes + nPQNodes;
  
  dag_h->creator = "PQ_DDSimpleSmallWrite";
  dag_h->numSuccedents = 1;
  dag_h->succedents[0] = blockNode;
  rf_InitNode(termNode, rf_wait, RF_FALSE, rf_TerminateFunc, rf_TerminateUndoFunc, NULL, 0, 1, 0, 0, dag_h, "Trm", allocList);
  termNode->antecedents[0]  = unblockNode;
  termNode->antType[0] = rf_control;

  /* init the block and unblock nodes */
  /* The block node has all the read nodes as successors */
  rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, nReadNodes, 0, 0, 0, dag_h, "Nil", allocList);
  for (i=0; i < nReadNodes; i++)
    blockNode->succedents[i] = rrdNodes+i;

  /* The unblock node has all the writes as successors */
  rf_InitNode(unblockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL, 1, nWriteNodes, 0, 0, dag_h, "Nil", allocList);
  for (i=0; i < nWriteNodes; i++) {
    unblockNode->antecedents[i] = wudNodes+i;
    unblockNode->antType[i] = rf_control;
  }
  unblockNode->succedents[0] = termNode;

#define INIT_READ_NODE(node,name) \
  rf_InitNode(node, rf_wait, RF_FALSE, rf_DiskReadFunc, rf_DiskReadUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, name, allocList); \
  (node)->succedents[0] = recoveryNode; \
  (node)->antecedents[0] = blockNode; \
  (node)->antType[0] = rf_control;
  
  /* build the read nodes */
  pda = npdas;
  for (i=0; i < nRrdNodes; i++, pda = pda->next) {
    INIT_READ_NODE(rrdNodes+i,"rrd");
    DISK_NODE_PARAMS(rrdNodes[i],pda);
  }

  /* read redundancy pdas */
  pda = pqPDAs;
  INIT_READ_NODE(rpNodes,"Rp");
  RF_ASSERT(pda);
  DISK_NODE_PARAMS(rpNodes[0],pda);
  pda++;
  INIT_READ_NODE(rqNodes, redundantReadNodeName );
  RF_ASSERT(pda);
  DISK_NODE_PARAMS(rqNodes[0],pda);
  if (nPQNodes==2)
    {
      pda++;
      INIT_READ_NODE(rpNodes+1,"Rp");
      RF_ASSERT(pda);
      DISK_NODE_PARAMS(rpNodes[1],pda);
      pda++;
      INIT_READ_NODE(rqNodes+1,redundantReadNodeName );
      RF_ASSERT(pda);
      DISK_NODE_PARAMS(rqNodes[1],pda);
    }

  /* the recovery node has all reads as precedessors and all writes as successors.
     It generates a result for every write P or write Q node. 
     As parameters, it takes a pda per read and a pda per stripe of user data written. 
     It also takes as the last params the raidPtr and asm.
     For results, it takes PDA for P & Q. */


    rf_InitNode(recoveryNode, rf_wait, RF_FALSE, recovFunc, rf_NullNodeUndoFunc, NULL,
           nWriteNodes, /* succesors */
           nReadNodes, /* preds */
           nReadNodes + nWudNodes + 3, /* params */
           2 * nPQNodes, /* results */
           dag_h, recoveryNodeName, allocList);
 


  for (i=0; i < nReadNodes; i++ )
    {
      recoveryNode->antecedents[i] = rrdNodes+i;
      recoveryNode->antType[i] = rf_control;
      recoveryNode->params[i].p = DISK_NODE_PDA(rrdNodes+i);
    }
  for (i=0; i < nWudNodes; i++)
    {
      recoveryNode->succedents[i] = wudNodes+i;
    }
  recoveryNode->params[nReadNodes+nWudNodes].p = asmap->failedPDA;
  recoveryNode->params[nReadNodes+nWudNodes+1].p = raidPtr;
  recoveryNode->params[nReadNodes+nWudNodes+2].p = asmap;
  
  for ( ; i < nWriteNodes; i++)
      recoveryNode->succedents[i] = wudNodes+i;

  pda = pqPDAs;
  recoveryNode->results[0] = pda;
  pda++;
  recoveryNode->results[1] = pda;
  if ( nPQNodes == 2)
    {
      pda++;
      recoveryNode->results[2] = pda;
      pda++;
      recoveryNode->results[3] = pda;
    }

  /* fill writes */
#define INIT_WRITE_NODE(node,name) \
  rf_InitNode(node, rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, name, allocList); \
    (node)->succedents[0] = unblockNode; \
    (node)->antecedents[0] = recoveryNode; \
    (node)->antType[0] = rf_control;

  pda = asmap->physInfo;
  for (i=0; i < nWudNodes; i++)
    {
      INIT_WRITE_NODE(wudNodes+i,"Wd");
      DISK_NODE_PARAMS(wudNodes[i],pda);
      recoveryNode->params[nReadNodes+i].p = DISK_NODE_PDA(wudNodes+i);
      pda = pda->next;
    }
  /* write redundancy pdas */
  pda = pqPDAs;
  INIT_WRITE_NODE(wpNodes,"Wp");
  RF_ASSERT(pda);
  DISK_NODE_PARAMS(wpNodes[0],pda);
  pda++;
  INIT_WRITE_NODE(wqNodes,"Wq");
  RF_ASSERT(pda);
  DISK_NODE_PARAMS(wqNodes[0],pda);
  if (nPQNodes==2)
    {
      pda++;
      INIT_WRITE_NODE(wpNodes+1,"Wp");
      RF_ASSERT(pda);
      DISK_NODE_PARAMS(wpNodes[1],pda);
      pda++;
      INIT_WRITE_NODE(wqNodes+1,"Wq");
      RF_ASSERT(pda);
      DISK_NODE_PARAMS(wqNodes[1],pda);
    }
}