rf_engine.c

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

    /* XXX need to fix this case */
    /* for now, assume that we're done */
    return(RF_TRUE);
    break;
  default :
    /* illegal node status */
    RF_PANIC();
    break;
  }
}

#ifdef SIMULATE
/* this is only ifdef SIMULATE because nothing else needs it */
/* recursively determine if a DAG has completed execution */
static int DAGDone(RF_DagHeader_t *dag)
{
  int i;

  for (i = 0; i < dag->numSuccedents; i++)
    if (!BranchDone(dag->succedents[i]))
      return RF_FALSE;
  return RF_TRUE;
}
#endif /* SIMULATE */

static int NodeReady(RF_DagNode_t *node)
{
  int ready;

  switch (node->dagHdr->status) {
  case rf_enable :
  case rf_rollForward :
    if ((node->status == rf_wait) && (node->numAntecedents == node->numAntDone))
      ready = RF_TRUE;
    else
      ready = RF_FALSE;
    break;
  case rf_rollBackward :
    RF_ASSERT(node->numSuccDone <= node->numSuccedents);
    RF_ASSERT(node->numSuccFired <= node->numSuccedents);
    RF_ASSERT(node->numSuccFired <= node->numSuccDone);
    if ((node->status == rf_good) && (node->numSuccDone == node->numSuccedents))
      ready = RF_TRUE;
    else
      ready = RF_FALSE;
    break;
  default :
    printf("Execution engine found illegal DAG status in NodeReady\n");
    RF_PANIC();
    break;
  }  

  return(ready);
}



/* user context and dag-exec-thread context:
 * Fire a node.  The node's status field determines which function, do or undo,
 * to be fired.
 * This routine assumes that the node's status field has alread been set to
 * "fired" or "recover" to indicate the direction of execution.
 */
static void FireNode(RF_DagNode_t *node)
{
  int tid;

  switch (node->status) {
  case rf_fired :
    /* fire the do function of a node */
    if (rf_engineDebug) {
      rf_get_threadid(tid);
      printf("[%d] Firing node 0x%lx (%s)\n",tid,(unsigned long) node, node->name);
    }
#ifdef KERNEL
    if (node->flags & RF_DAGNODE_FLAG_YIELD)
      thread_block();
#endif /* KERNEL */
    (*(node->doFunc)) (node);
    break;
  case rf_recover :
    /* fire the undo function of a node */
      if (rf_engineDebug || 1) {
        rf_get_threadid(tid);
        printf("[%d] Firing (undo) node 0x%lx (%s)\n",tid,(unsigned long) node, node->name);
      }
#ifdef KERNEL
      if (node->flags & RF_DAGNODE_FLAG_YIELD)
        thread_block();
#endif /* KERNEL */
      (*(node->undoFunc)) (node);
    break;
  default :
    RF_PANIC();
    break;
  }
}



/* user context:
 * Attempt to fire each node in a linear array.
 * The entire list is fired atomically.
 */
static void FireNodeArray(
  int             numNodes,
  RF_DagNode_t  **nodeList)
{
  RF_DagStatus_t dstat;
  RF_DagNode_t *node;
  int i, j;

  /* first, mark all nodes which are ready to be fired */
  for (i = 0; i < numNodes; i++) {
    node = nodeList[i];
    dstat = node->dagHdr->status;
    RF_ASSERT((node->status == rf_wait) || (node->status == rf_good));
    if (NodeReady(node))
      if ((dstat == rf_enable) || (dstat == rf_rollForward)) {
	RF_ASSERT(node->status == rf_wait);
	if (node->commitNode)
	  node->dagHdr->numCommits++;
	node->status = rf_fired;
	for (j = 0; j < node->numAntecedents; j++)
	  node->antecedents[j]->numSuccFired++;
      }
      else {
	RF_ASSERT(dstat == rf_rollBackward);
	RF_ASSERT(node->status == rf_good);
	RF_ASSERT(node->commitNode == RF_FALSE); /* only one commit node per graph */
	node->status = rf_recover;
      }
  }
  /* now, fire the nodes */
  for (i = 0; i < numNodes; i++) {
    if ((nodeList[i]->status == rf_fired) || (nodeList[i]->status == rf_recover))
      FireNode(nodeList[i]);
  }
}


#ifndef SIMULATE
/* user context:
 * Attempt to fire each node in a linked list.
 * The entire list is fired atomically.
 */
static void FireNodeList(RF_DagNode_t *nodeList)
{
  RF_DagNode_t *node, *next;
  RF_DagStatus_t dstat;
  int j;

  if (nodeList) {
    /* first, mark all nodes which are ready to be fired */
    for (node = nodeList; node; node = next) {
      next = node->next;
      dstat = node->dagHdr->status;
      RF_ASSERT((node->status == rf_wait) || (node->status == rf_good));
      if (NodeReady(node))
        if ((dstat == rf_enable) || (dstat == rf_rollForward)) {
          RF_ASSERT(node->status == rf_wait);
          if (node->commitNode)
            node->dagHdr->numCommits++;
          node->status = rf_fired;
          for (j = 0; j < node->numAntecedents; j++)
            node->antecedents[j]->numSuccFired++;
        }
        else {
          RF_ASSERT(dstat == rf_rollBackward);
          RF_ASSERT(node->status == rf_good);
          RF_ASSERT(node->commitNode == RF_FALSE); /* only one commit node per graph */
          node->status = rf_recover;
        }
    }
    /* now, fire the nodes */
    for (node = nodeList; node; node = next) {
      next = node->next;
      if ((node->status == rf_fired) || (node->status == rf_recover))
        FireNode(node);
    }
  }
}
#endif /* !SIMULATE */



/* interrupt context:
 * for each succedent
 *    propagate required results from node to succedent
 *    increment succedent's numAntDone
 *    place newly-enable nodes on node queue for firing
 *
 * To save context switches, we don't place NIL nodes on the node queue,
 * but rather just process them as if they had fired.  Note that NIL nodes
 * that are the direct successors of the header will actually get fired by
 * DispatchDAG, which is fine because no context switches are involved.
 *
 * Important:  when running at user level, this can be called by any
 * disk thread, and so the increment and check of the antecedent count
 * must be locked.  I used the node queue mutex and locked down the
 * entire function, but this is certainly overkill.
 */
static void PropagateResults(
  RF_DagNode_t  *node,
  int            context)
{
  RF_DagNode_t *s, *a;
  RF_Raid_t *raidPtr;
  int tid, i, ks;
#ifdef SIMULATE
  RF_PropHeader_t *p;         /* prop list for succedent i */
#else /* SIMULATE */
  RF_DagNode_t *finishlist = NULL; /* a list of NIL nodes to be finished */
  RF_DagNode_t *skiplist = NULL;   /* list of nodes with failed truedata antecedents */
  RF_DagNode_t *firelist = NULL;   /* a list of nodes to be fired */
  RF_DagNode_t *q = NULL, *qh = NULL, *next;
  int j, skipNode;
#endif /* SIMULATE */

  rf_get_threadid(tid);

  raidPtr = node->dagHdr->raidPtr;

  DO_LOCK(raidPtr);

  /* debug - validate fire counts */
  for (i = 0; i < node->numAntecedents; i++) {
    a = *(node->antecedents + i);
    RF_ASSERT(a->numSuccFired >= a->numSuccDone);
    RF_ASSERT(a->numSuccFired <= a->numSuccedents);
    a->numSuccDone++;
  }

  switch (node->dagHdr->status) {
  case rf_enable :
  case rf_rollForward :
#ifdef SIMULATE
      /* currently we never propagate results unless in simulation */
    for (i = 0; i < node->numSuccedents; i++) {
      s = *(node->succedents + i);      
      RF_ASSERT(s->status == rf_wait);
      (s->numAntDone)++;
      if (node->propList == NULL)
	/* null propList implies no results to be propagated */
	p = NULL;
      else
	/* p=head of prop list for succedent i */
	p = *(node->propList + i);
      while (p != NULL) {
	/* bind node results to succedent's parameters */
#if 0
	*(s->params + p->paramNum) = *(node->results + p->resultNum);
#else
	s->params[p->paramNum].p = node->results[p->resultNum];
#endif
	p = p->next;
      }
    }
#else  /* SIMULATE */
    for (i = 0; i < node->numSuccedents; i++) {
      s = *(node->succedents + i);      
      RF_ASSERT(s->status == rf_wait);
      (s->numAntDone)++;
      if (s->numAntDone == s->numAntecedents) {
	/* look for NIL nodes */
	if (s->doFunc == rf_NullNodeFunc) {
	  /* don't fire NIL nodes, just process them */
	  s->next = finishlist;
	  finishlist = s;
	}
	else {
	  /* look to see if the node is to be skipped */
	  skipNode = RF_FALSE;
	  for (j = 0; j < s->numAntecedents; j++)
	    if ((s->antType[j] == rf_trueData) && (s->antecedents[j]->status == rf_bad))
	      skipNode = RF_TRUE;
	  if (skipNode) {
	    /* this node has one or more failed true data dependencies, so skip it */
	    s->next = skiplist;
	    skiplist = s;
	  }
	  else
	    /* add s to list of nodes (q) to execute */
	    if (context != RF_INTR_CONTEXT) {
	      /* we only have to enqueue if we're at intr context */
	      s->next = firelist;  /* put node on a list to be fired after we unlock */
	      firelist = s;
	    } else {               /* enqueue the node for the dag exec thread to fire */
	      RF_ASSERT(NodeReady(s));
	      if (q) {
	        q->next = s;
	        q = s;
	      }
	      else {
	        qh = q = s;
	        qh->next = NULL;
	      }
	    }
	}
      }
    }

    if (q) {
      /* xfer our local list of nodes to the node queue */
      q->next = raidPtr->node_queue; raidPtr->node_queue = qh;
      DO_SIGNAL(raidPtr);
    }
    DO_UNLOCK(raidPtr);

    for (; skiplist; skiplist = next) {
      next = skiplist->next;
      skiplist->status = rf_skipped;
      for (i = 0; i < skiplist->numAntecedents; i++) {
	skiplist->antecedents[i]->numSuccFired++;
      }
      if (skiplist->commitNode) {
	skiplist->dagHdr->numCommits++;
      }
      rf_FinishNode(skiplist, context);
    }
    for (; finishlist; finishlist = next) {
      /* NIL nodes: no need to fire them */
      next = finishlist->next;
      finishlist->status = rf_good;
      for (i = 0; i < finishlist->numAntecedents; i++) {
	finishlist->antecedents[i]->numSuccFired++;
      }
      if (finishlist->commitNode)
	finishlist->dagHdr->numCommits++;
      /*
       * Okay, here we're calling rf_FinishNode() on nodes that
       * have the null function as their work proc. Such a node
       * could be the terminal node in a DAG. If so, it will
       * cause the DAG to complete, which will in turn free
       * memory used by the DAG, which includes the node in
       * question. Thus, we must avoid referencing the node
       * at all after calling rf_FinishNode() on it.
       */
      rf_FinishNode(finishlist, context);   /* recursive call */
    }
    /* fire all nodes in firelist */
    FireNodeList(firelist);
#endif   /* SIMULATE */
    break;

  case rf_rollBackward :
#ifdef SIMULATE