Home | History | Annotate | Line # | Download | only in raidframe
rf_engine.c revision 1.17
      1  1.17    oster /*	$NetBSD: rf_engine.c,v 1.17 2002/09/15 23:40:40 oster Exp $	*/
      2   1.1    oster /*
      3   1.1    oster  * Copyright (c) 1995 Carnegie-Mellon University.
      4   1.1    oster  * All rights reserved.
      5   1.1    oster  *
      6   1.1    oster  * Author: William V. Courtright II, Mark Holland, Rachad Youssef
      7   1.1    oster  *
      8   1.1    oster  * Permission to use, copy, modify and distribute this software and
      9   1.1    oster  * its documentation is hereby granted, provided that both the copyright
     10   1.1    oster  * notice and this permission notice appear in all copies of the
     11   1.1    oster  * software, derivative works or modified versions, and any portions
     12   1.1    oster  * thereof, and that both notices appear in supporting documentation.
     13   1.1    oster  *
     14   1.1    oster  * CARNEGIE MELLON ALLOWS FREE USE OF THIS SOFTWARE IN ITS "AS IS"
     15   1.1    oster  * CONDITION.  CARNEGIE MELLON DISCLAIMS ANY LIABILITY OF ANY KIND
     16   1.1    oster  * FOR ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF THIS SOFTWARE.
     17   1.1    oster  *
     18   1.1    oster  * Carnegie Mellon requests users of this software to return to
     19   1.1    oster  *
     20   1.1    oster  *  Software Distribution Coordinator  or  Software.Distribution (at) CS.CMU.EDU
     21   1.1    oster  *  School of Computer Science
     22   1.1    oster  *  Carnegie Mellon University
     23   1.1    oster  *  Pittsburgh PA 15213-3890
     24   1.1    oster  *
     25   1.1    oster  * any improvements or extensions that they make and grant Carnegie the
     26   1.1    oster  * rights to redistribute these changes.
     27   1.1    oster  */
     28   1.1    oster 
     29   1.1    oster /****************************************************************************
     30   1.1    oster  *                                                                          *
     31   1.1    oster  * engine.c -- code for DAG execution engine                                *
     32   1.1    oster  *                                                                          *
     33   1.1    oster  * Modified to work as follows (holland):                                   *
     34   1.1    oster  *   A user-thread calls into DispatchDAG, which fires off the nodes that   *
     35   1.1    oster  *   are direct successors to the header node.  DispatchDAG then returns,   *
     36   1.1    oster  *   and the rest of the I/O continues asynchronously.  As each node        *
     37   1.1    oster  *   completes, the node execution function calls FinishNode().  FinishNode *
     38   1.1    oster  *   scans the list of successors to the node and increments the antecedent *
     39   1.1    oster  *   counts.  Each node that becomes enabled is placed on a central node    *
     40   1.1    oster  *   queue.  A dedicated dag-execution thread grabs nodes off of this       *
     41   1.1    oster  *   queue and fires them.                                                  *
     42   1.1    oster  *                                                                          *
     43   1.1    oster  *   NULL nodes are never fired.                                            *
     44   1.1    oster  *                                                                          *
     45   1.1    oster  *   Terminator nodes are never fired, but rather cause the callback        *
     46   1.1    oster  *   associated with the DAG to be invoked.                                 *
     47   1.1    oster  *                                                                          *
     48   1.1    oster  *   If a node fails, the dag either rolls forward to the completion or     *
     49   1.1    oster  *   rolls back, undoing previously-completed nodes and fails atomically.   *
     50   1.1    oster  *   The direction of recovery is determined by the location of the failed  *
     51  1.11      wiz  *   node in the graph.  If the failure occurred before the commit node in   *
     52   1.1    oster  *   the graph, backward recovery is used.  Otherwise, forward recovery is  *
     53   1.1    oster  *   used.                                                                  *
     54   1.1    oster  *                                                                          *
     55   1.1    oster  ****************************************************************************/
     56  1.12    lukem 
     57  1.12    lukem #include <sys/cdefs.h>
     58  1.17    oster __KERNEL_RCSID(0, "$NetBSD: rf_engine.c,v 1.17 2002/09/15 23:40:40 oster Exp $");
     59   1.1    oster 
     60   1.1    oster #include "rf_threadstuff.h"
     61   1.1    oster 
     62   1.1    oster #include <sys/errno.h>
     63   1.1    oster 
     64   1.1    oster #include "rf_dag.h"
     65   1.1    oster #include "rf_engine.h"
     66   1.1    oster #include "rf_etimer.h"
     67   1.1    oster #include "rf_general.h"
     68   1.1    oster #include "rf_dagutils.h"
     69   1.1    oster #include "rf_shutdown.h"
     70   1.1    oster #include "rf_raid.h"
     71   1.1    oster 
     72   1.1    oster static void DAGExecutionThread(RF_ThreadArg_t arg);
     73   1.1    oster 
     74   1.1    oster #define DO_INIT(_l_,_r_) { \
     75   1.1    oster   int _rc; \
     76   1.1    oster   _rc = rf_create_managed_mutex(_l_,&(_r_)->node_queue_mutex); \
     77   1.1    oster   if (_rc) { \
     78   1.1    oster     return(_rc); \
     79   1.1    oster   } \
     80   1.1    oster   _rc = rf_create_managed_cond(_l_,&(_r_)->node_queue_cond); \
     81   1.1    oster   if (_rc) { \
     82   1.1    oster     return(_rc); \
     83   1.1    oster   } \
     84   1.1    oster }
     85   1.1    oster 
     86   1.1    oster /* synchronization primitives for this file.  DO_WAIT should be enclosed in a while loop. */
     87   1.1    oster 
     88   1.1    oster /*
     89   1.1    oster  * XXX Is this spl-ing really necessary?
     90   1.1    oster  */
     91  1.10  thorpej #define DO_LOCK(_r_) \
     92  1.10  thorpej do { \
     93  1.10  thorpej 	ks = splbio(); \
     94  1.10  thorpej 	RF_LOCK_MUTEX((_r_)->node_queue_mutex); \
     95  1.10  thorpej } while (0)
     96  1.10  thorpej 
     97  1.10  thorpej #define DO_UNLOCK(_r_) \
     98  1.10  thorpej do { \
     99  1.10  thorpej 	RF_UNLOCK_MUTEX((_r_)->node_queue_mutex); \
    100  1.10  thorpej 	splx(ks); \
    101  1.10  thorpej } while (0)
    102  1.10  thorpej 
    103  1.10  thorpej #define	DO_WAIT(_r_) \
    104  1.10  thorpej 	RF_WAIT_COND((_r_)->node_queue, (_r_)->node_queue_mutex)
    105  1.10  thorpej 
    106  1.10  thorpej #define	DO_SIGNAL(_r_) \
    107  1.10  thorpej 	RF_BROADCAST_COND((_r_)->node_queue)	/* XXX RF_SIGNAL_COND? */
    108   1.1    oster 
    109   1.1    oster static void rf_ShutdownEngine(void *);
    110   1.1    oster 
    111   1.4    oster static void
    112   1.4    oster rf_ShutdownEngine(arg)
    113   1.4    oster 	void   *arg;
    114   1.4    oster {
    115   1.4    oster 	RF_Raid_t *raidPtr;
    116   1.4    oster 
    117   1.4    oster 	raidPtr = (RF_Raid_t *) arg;
    118   1.4    oster 	raidPtr->shutdown_engine = 1;
    119   1.4    oster 	DO_SIGNAL(raidPtr);
    120   1.1    oster }
    121   1.1    oster 
    122   1.4    oster int
    123   1.4    oster rf_ConfigureEngine(
    124   1.4    oster     RF_ShutdownList_t ** listp,
    125   1.4    oster     RF_Raid_t * raidPtr,
    126   1.4    oster     RF_Config_t * cfgPtr)
    127   1.4    oster {
    128   1.9    oster 	int     rc;
    129   1.4    oster 
    130   1.4    oster 	DO_INIT(listp, raidPtr);
    131   1.4    oster 
    132   1.4    oster 	raidPtr->node_queue = NULL;
    133   1.4    oster 	raidPtr->dags_in_flight = 0;
    134   1.4    oster 
    135   1.4    oster 	rc = rf_init_managed_threadgroup(listp, &raidPtr->engine_tg);
    136   1.4    oster 	if (rc)
    137   1.4    oster 		return (rc);
    138   1.4    oster 
    139   1.4    oster 	/* we create the execution thread only once per system boot. no need
    140   1.4    oster 	 * to check return code b/c the kernel panics if it can't create the
    141   1.4    oster 	 * thread. */
    142   1.4    oster 	if (rf_engineDebug) {
    143   1.9    oster 		printf("raid%d: Creating engine thread\n", raidPtr->raidid);
    144   1.4    oster 	}
    145   1.6    oster 	if (RF_CREATE_THREAD(raidPtr->engine_thread, DAGExecutionThread, raidPtr,"raid")) {
    146   1.4    oster 		RF_ERRORMSG("RAIDFRAME: Unable to create engine thread\n");
    147   1.4    oster 		return (ENOMEM);
    148   1.4    oster 	}
    149   1.4    oster 	if (rf_engineDebug) {
    150   1.9    oster 		printf("raid%d: Created engine thread\n", raidPtr->raidid);
    151   1.4    oster 	}
    152   1.4    oster 	RF_THREADGROUP_STARTED(&raidPtr->engine_tg);
    153   1.4    oster 	/* XXX something is missing here... */
    154   1.1    oster #ifdef debug
    155   1.4    oster 	printf("Skipping the WAIT_START!!\n");
    156   1.1    oster #endif
    157   1.1    oster #if 0
    158   1.4    oster 	RF_THREADGROUP_WAIT_START(&raidPtr->engine_tg);
    159   1.1    oster #endif
    160   1.4    oster 	/* engine thread is now running and waiting for work */
    161   1.4    oster 	if (rf_engineDebug) {
    162   1.9    oster 		printf("raid%d: Engine thread running and waiting for events\n", raidPtr->raidid);
    163   1.4    oster 	}
    164   1.4    oster 	rc = rf_ShutdownCreate(listp, rf_ShutdownEngine, raidPtr);
    165   1.4    oster 	if (rc) {
    166  1.16    oster 		rf_print_unable_to_add_shutdown(__FILE__, __LINE__, rc);
    167   1.4    oster 		rf_ShutdownEngine(NULL);
    168   1.4    oster 	}
    169   1.4    oster 	return (rc);
    170   1.4    oster }
    171   1.4    oster 
    172   1.4    oster static int
    173   1.4    oster BranchDone(RF_DagNode_t * node)
    174   1.4    oster {
    175   1.4    oster 	int     i;
    176   1.4    oster 
    177   1.4    oster 	/* return true if forward execution is completed for a node and it's
    178   1.4    oster 	 * succedents */
    179   1.4    oster 	switch (node->status) {
    180   1.4    oster 	case rf_wait:
    181   1.4    oster 		/* should never be called in this state */
    182   1.4    oster 		RF_PANIC();
    183   1.4    oster 		break;
    184   1.4    oster 	case rf_fired:
    185   1.4    oster 		/* node is currently executing, so we're not done */
    186   1.4    oster 		return (RF_FALSE);
    187   1.4    oster 	case rf_good:
    188   1.4    oster 		for (i = 0; i < node->numSuccedents; i++)	/* for each succedent */
    189   1.4    oster 			if (!BranchDone(node->succedents[i]))	/* recursively check
    190   1.4    oster 								 * branch */
    191   1.4    oster 				return RF_FALSE;
    192   1.4    oster 		return RF_TRUE;	/* node and all succedent branches aren't in
    193   1.4    oster 				 * fired state */
    194   1.4    oster 		break;
    195   1.4    oster 	case rf_bad:
    196   1.4    oster 		/* succedents can't fire */
    197   1.4    oster 		return (RF_TRUE);
    198   1.4    oster 	case rf_recover:
    199   1.4    oster 		/* should never be called in this state */
    200   1.4    oster 		RF_PANIC();
    201   1.4    oster 		break;
    202   1.4    oster 	case rf_undone:
    203   1.4    oster 	case rf_panic:
    204   1.4    oster 		/* XXX need to fix this case */
    205   1.4    oster 		/* for now, assume that we're done */
    206   1.4    oster 		return (RF_TRUE);
    207   1.4    oster 		break;
    208   1.4    oster 	default:
    209   1.4    oster 		/* illegal node status */
    210   1.4    oster 		RF_PANIC();
    211   1.4    oster 		break;
    212   1.4    oster 	}
    213   1.4    oster }
    214   1.4    oster 
    215   1.4    oster static int
    216   1.4    oster NodeReady(RF_DagNode_t * node)
    217   1.4    oster {
    218   1.4    oster 	int     ready;
    219   1.4    oster 
    220   1.4    oster 	switch (node->dagHdr->status) {
    221   1.4    oster 	case rf_enable:
    222   1.4    oster 	case rf_rollForward:
    223   1.4    oster 		if ((node->status == rf_wait) && (node->numAntecedents == node->numAntDone))
    224   1.4    oster 			ready = RF_TRUE;
    225   1.4    oster 		else
    226   1.4    oster 			ready = RF_FALSE;
    227   1.4    oster 		break;
    228   1.4    oster 	case rf_rollBackward:
    229   1.4    oster 		RF_ASSERT(node->numSuccDone <= node->numSuccedents);
    230   1.4    oster 		RF_ASSERT(node->numSuccFired <= node->numSuccedents);
    231   1.4    oster 		RF_ASSERT(node->numSuccFired <= node->numSuccDone);
    232   1.4    oster 		if ((node->status == rf_good) && (node->numSuccDone == node->numSuccedents))
    233   1.4    oster 			ready = RF_TRUE;
    234   1.4    oster 		else
    235   1.4    oster 			ready = RF_FALSE;
    236   1.4    oster 		break;
    237   1.4    oster 	default:
    238   1.4    oster 		printf("Execution engine found illegal DAG status in NodeReady\n");
    239   1.4    oster 		RF_PANIC();
    240   1.4    oster 		break;
    241   1.4    oster 	}
    242   1.1    oster 
    243   1.4    oster 	return (ready);
    244   1.1    oster }
    245   1.1    oster 
    246   1.1    oster 
    247   1.1    oster 
    248   1.1    oster /* user context and dag-exec-thread context:
    249   1.1    oster  * Fire a node.  The node's status field determines which function, do or undo,
    250   1.1    oster  * to be fired.
    251   1.1    oster  * This routine assumes that the node's status field has alread been set to
    252   1.1    oster  * "fired" or "recover" to indicate the direction of execution.
    253   1.1    oster  */
    254   1.4    oster static void
    255   1.4    oster FireNode(RF_DagNode_t * node)
    256   1.1    oster {
    257   1.4    oster 	switch (node->status) {
    258   1.4    oster 	case rf_fired:
    259   1.4    oster 		/* fire the do function of a node */
    260   1.4    oster 		if (rf_engineDebug) {
    261   1.9    oster 			printf("raid%d: Firing node 0x%lx (%s)\n",
    262   1.9    oster 			       node->dagHdr->raidPtr->raidid,
    263   1.9    oster 			       (unsigned long) node, node->name);
    264   1.4    oster 		}
    265   1.4    oster 		if (node->flags & RF_DAGNODE_FLAG_YIELD) {
    266   1.1    oster #if defined(__NetBSD__) && defined(_KERNEL)
    267   1.4    oster 			/* thread_block(); */
    268   1.4    oster 			/* printf("Need to block the thread here...\n");  */
    269   1.4    oster 			/* XXX thread_block is actually mentioned in
    270   1.4    oster 			 * /usr/include/vm/vm_extern.h */
    271   1.1    oster #else
    272   1.4    oster 			thread_block();
    273   1.1    oster #endif
    274   1.4    oster 		}
    275   1.4    oster 		(*(node->doFunc)) (node);
    276   1.4    oster 		break;
    277   1.4    oster 	case rf_recover:
    278   1.4    oster 		/* fire the undo function of a node */
    279   1.9    oster 		if (rf_engineDebug) {
    280   1.9    oster 			printf("raid%d: Firing (undo) node 0x%lx (%s)\n",
    281   1.9    oster 			       node->dagHdr->raidPtr->raidid,
    282   1.9    oster 			       (unsigned long) node, node->name);
    283   1.4    oster 		}
    284   1.4    oster 		if (node->flags & RF_DAGNODE_FLAG_YIELD)
    285   1.1    oster #if defined(__NetBSD__) && defined(_KERNEL)
    286   1.4    oster 			/* thread_block(); */
    287   1.4    oster 			/* printf("Need to block the thread here...\n"); */
    288   1.4    oster 			/* XXX thread_block is actually mentioned in
    289   1.4    oster 			 * /usr/include/vm/vm_extern.h */
    290   1.1    oster #else
    291   1.4    oster 			thread_block();
    292   1.1    oster #endif
    293   1.4    oster 		(*(node->undoFunc)) (node);
    294   1.4    oster 		break;
    295   1.4    oster 	default:
    296   1.4    oster 		RF_PANIC();
    297   1.4    oster 		break;
    298   1.4    oster 	}
    299   1.1    oster }
    300   1.1    oster 
    301   1.1    oster 
    302   1.1    oster 
    303   1.1    oster /* user context:
    304   1.1    oster  * Attempt to fire each node in a linear array.
    305   1.1    oster  * The entire list is fired atomically.
    306   1.1    oster  */
    307   1.4    oster static void
    308   1.4    oster FireNodeArray(
    309   1.4    oster     int numNodes,
    310   1.4    oster     RF_DagNode_t ** nodeList)
    311   1.4    oster {
    312   1.4    oster 	RF_DagStatus_t dstat;
    313   1.4    oster 	RF_DagNode_t *node;
    314   1.4    oster 	int     i, j;
    315   1.4    oster 
    316   1.4    oster 	/* first, mark all nodes which are ready to be fired */
    317   1.4    oster 	for (i = 0; i < numNodes; i++) {
    318   1.4    oster 		node = nodeList[i];
    319   1.4    oster 		dstat = node->dagHdr->status;
    320   1.4    oster 		RF_ASSERT((node->status == rf_wait) || (node->status == rf_good));
    321   1.4    oster 		if (NodeReady(node)) {
    322   1.4    oster 			if ((dstat == rf_enable) || (dstat == rf_rollForward)) {
    323   1.4    oster 				RF_ASSERT(node->status == rf_wait);
    324   1.4    oster 				if (node->commitNode)
    325   1.4    oster 					node->dagHdr->numCommits++;
    326   1.4    oster 				node->status = rf_fired;
    327   1.4    oster 				for (j = 0; j < node->numAntecedents; j++)
    328   1.4    oster 					node->antecedents[j]->numSuccFired++;
    329   1.4    oster 			} else {
    330   1.4    oster 				RF_ASSERT(dstat == rf_rollBackward);
    331   1.4    oster 				RF_ASSERT(node->status == rf_good);
    332   1.4    oster 				RF_ASSERT(node->commitNode == RF_FALSE);	/* only one commit node
    333   1.4    oster 										 * per graph */
    334   1.4    oster 				node->status = rf_recover;
    335   1.4    oster 			}
    336   1.4    oster 		}
    337   1.4    oster 	}
    338   1.4    oster 	/* now, fire the nodes */
    339   1.4    oster 	for (i = 0; i < numNodes; i++) {
    340   1.4    oster 		if ((nodeList[i]->status == rf_fired) || (nodeList[i]->status == rf_recover))
    341   1.4    oster 			FireNode(nodeList[i]);
    342   1.4    oster 	}
    343   1.1    oster }
    344   1.1    oster 
    345   1.1    oster 
    346   1.1    oster /* user context:
    347   1.1    oster  * Attempt to fire each node in a linked list.
    348   1.1    oster  * The entire list is fired atomically.
    349   1.1    oster  */
    350   1.4    oster static void
    351   1.4    oster FireNodeList(RF_DagNode_t * nodeList)
    352   1.1    oster {
    353   1.4    oster 	RF_DagNode_t *node, *next;
    354   1.4    oster 	RF_DagStatus_t dstat;
    355   1.4    oster 	int     j;
    356   1.4    oster 
    357   1.4    oster 	if (nodeList) {
    358   1.4    oster 		/* first, mark all nodes which are ready to be fired */
    359   1.4    oster 		for (node = nodeList; node; node = next) {
    360   1.4    oster 			next = node->next;
    361   1.4    oster 			dstat = node->dagHdr->status;
    362   1.4    oster 			RF_ASSERT((node->status == rf_wait) || (node->status == rf_good));
    363   1.4    oster 			if (NodeReady(node)) {
    364   1.4    oster 				if ((dstat == rf_enable) || (dstat == rf_rollForward)) {
    365   1.4    oster 					RF_ASSERT(node->status == rf_wait);
    366   1.4    oster 					if (node->commitNode)
    367   1.4    oster 						node->dagHdr->numCommits++;
    368   1.4    oster 					node->status = rf_fired;
    369   1.4    oster 					for (j = 0; j < node->numAntecedents; j++)
    370   1.4    oster 						node->antecedents[j]->numSuccFired++;
    371   1.4    oster 				} else {
    372   1.4    oster 					RF_ASSERT(dstat == rf_rollBackward);
    373   1.4    oster 					RF_ASSERT(node->status == rf_good);
    374   1.4    oster 					RF_ASSERT(node->commitNode == RF_FALSE);	/* only one commit node
    375   1.4    oster 											 * per graph */
    376   1.4    oster 					node->status = rf_recover;
    377   1.4    oster 				}
    378   1.4    oster 			}
    379   1.4    oster 		}
    380   1.4    oster 		/* now, fire the nodes */
    381   1.4    oster 		for (node = nodeList; node; node = next) {
    382   1.4    oster 			next = node->next;
    383   1.4    oster 			if ((node->status == rf_fired) || (node->status == rf_recover))
    384   1.4    oster 				FireNode(node);
    385   1.4    oster 		}
    386   1.4    oster 	}
    387   1.1    oster }
    388   1.1    oster /* interrupt context:
    389   1.1    oster  * for each succedent
    390   1.1    oster  *    propagate required results from node to succedent
    391   1.1    oster  *    increment succedent's numAntDone
    392   1.1    oster  *    place newly-enable nodes on node queue for firing
    393   1.1    oster  *
    394   1.1    oster  * To save context switches, we don't place NIL nodes on the node queue,
    395   1.1    oster  * but rather just process them as if they had fired.  Note that NIL nodes
    396   1.1    oster  * that are the direct successors of the header will actually get fired by
    397   1.1    oster  * DispatchDAG, which is fine because no context switches are involved.
    398   1.1    oster  *
    399   1.1    oster  * Important:  when running at user level, this can be called by any
    400   1.1    oster  * disk thread, and so the increment and check of the antecedent count
    401   1.1    oster  * must be locked.  I used the node queue mutex and locked down the
    402   1.1    oster  * entire function, but this is certainly overkill.
    403   1.1    oster  */
    404   1.4    oster static void
    405   1.4    oster PropagateResults(
    406   1.4    oster     RF_DagNode_t * node,
    407   1.4    oster     int context)
    408   1.4    oster {
    409   1.4    oster 	RF_DagNode_t *s, *a;
    410   1.4    oster 	RF_Raid_t *raidPtr;
    411   1.9    oster 	int     i, ks;
    412   1.4    oster 	RF_DagNode_t *finishlist = NULL;	/* a list of NIL nodes to be
    413   1.4    oster 						 * finished */
    414   1.4    oster 	RF_DagNode_t *skiplist = NULL;	/* list of nodes with failed truedata
    415   1.4    oster 					 * antecedents */
    416   1.4    oster 	RF_DagNode_t *firelist = NULL;	/* a list of nodes to be fired */
    417   1.4    oster 	RF_DagNode_t *q = NULL, *qh = NULL, *next;
    418   1.4    oster 	int     j, skipNode;
    419   1.4    oster 
    420   1.4    oster 	raidPtr = node->dagHdr->raidPtr;
    421   1.4    oster 
    422   1.4    oster 	DO_LOCK(raidPtr);
    423   1.4    oster 
    424   1.4    oster 	/* debug - validate fire counts */
    425   1.4    oster 	for (i = 0; i < node->numAntecedents; i++) {
    426   1.4    oster 		a = *(node->antecedents + i);
    427   1.4    oster 		RF_ASSERT(a->numSuccFired >= a->numSuccDone);
    428   1.4    oster 		RF_ASSERT(a->numSuccFired <= a->numSuccedents);
    429   1.4    oster 		a->numSuccDone++;
    430   1.4    oster 	}
    431   1.4    oster 
    432   1.4    oster 	switch (node->dagHdr->status) {
    433   1.4    oster 	case rf_enable:
    434   1.4    oster 	case rf_rollForward:
    435   1.4    oster 		for (i = 0; i < node->numSuccedents; i++) {
    436   1.4    oster 			s = *(node->succedents + i);
    437   1.4    oster 			RF_ASSERT(s->status == rf_wait);
    438   1.4    oster 			(s->numAntDone)++;
    439   1.4    oster 			if (s->numAntDone == s->numAntecedents) {
    440   1.4    oster 				/* look for NIL nodes */
    441   1.4    oster 				if (s->doFunc == rf_NullNodeFunc) {
    442   1.4    oster 					/* don't fire NIL nodes, just process
    443   1.4    oster 					 * them */
    444   1.4    oster 					s->next = finishlist;
    445   1.4    oster 					finishlist = s;
    446   1.4    oster 				} else {
    447   1.4    oster 					/* look to see if the node is to be
    448   1.4    oster 					 * skipped */
    449   1.4    oster 					skipNode = RF_FALSE;
    450   1.4    oster 					for (j = 0; j < s->numAntecedents; j++)
    451   1.4    oster 						if ((s->antType[j] == rf_trueData) && (s->antecedents[j]->status == rf_bad))
    452   1.4    oster 							skipNode = RF_TRUE;
    453   1.4    oster 					if (skipNode) {
    454   1.4    oster 						/* this node has one or more
    455   1.4    oster 						 * failed true data
    456   1.4    oster 						 * dependencies, so skip it */
    457   1.4    oster 						s->next = skiplist;
    458   1.4    oster 						skiplist = s;
    459   1.4    oster 					} else
    460   1.4    oster 						/* add s to list of nodes (q)
    461   1.4    oster 						 * to execute */
    462   1.4    oster 						if (context != RF_INTR_CONTEXT) {
    463   1.4    oster 							/* we only have to
    464   1.4    oster 							 * enqueue if we're at
    465   1.4    oster 							 * intr context */
    466   1.4    oster 							s->next = firelist;	/* put node on a list to
    467   1.4    oster 										 * be fired after we
    468   1.4    oster 										 * unlock */
    469   1.4    oster 							firelist = s;
    470   1.4    oster 						} else {	/* enqueue the node for
    471   1.4    oster 								 * the dag exec thread
    472   1.4    oster 								 * to fire */
    473   1.4    oster 							RF_ASSERT(NodeReady(s));
    474   1.4    oster 							if (q) {
    475   1.4    oster 								q->next = s;
    476   1.4    oster 								q = s;
    477   1.4    oster 							} else {
    478   1.4    oster 								qh = q = s;
    479   1.4    oster 								qh->next = NULL;
    480   1.4    oster 							}
    481   1.4    oster 						}
    482   1.4    oster 				}
    483   1.4    oster 			}
    484   1.4    oster 		}
    485   1.4    oster 
    486   1.4    oster 		if (q) {
    487   1.4    oster 			/* xfer our local list of nodes to the node queue */
    488   1.4    oster 			q->next = raidPtr->node_queue;
    489   1.4    oster 			raidPtr->node_queue = qh;
    490   1.4    oster 			DO_SIGNAL(raidPtr);
    491   1.4    oster 		}
    492   1.4    oster 		DO_UNLOCK(raidPtr);
    493   1.4    oster 
    494   1.4    oster 		for (; skiplist; skiplist = next) {
    495   1.4    oster 			next = skiplist->next;
    496   1.4    oster 			skiplist->status = rf_skipped;
    497   1.4    oster 			for (i = 0; i < skiplist->numAntecedents; i++) {
    498   1.4    oster 				skiplist->antecedents[i]->numSuccFired++;
    499   1.4    oster 			}
    500   1.4    oster 			if (skiplist->commitNode) {
    501   1.4    oster 				skiplist->dagHdr->numCommits++;
    502   1.4    oster 			}
    503   1.4    oster 			rf_FinishNode(skiplist, context);
    504   1.4    oster 		}
    505   1.4    oster 		for (; finishlist; finishlist = next) {
    506   1.4    oster 			/* NIL nodes: no need to fire them */
    507   1.4    oster 			next = finishlist->next;
    508   1.4    oster 			finishlist->status = rf_good;
    509   1.4    oster 			for (i = 0; i < finishlist->numAntecedents; i++) {
    510   1.4    oster 				finishlist->antecedents[i]->numSuccFired++;
    511   1.4    oster 			}
    512   1.4    oster 			if (finishlist->commitNode)
    513   1.4    oster 				finishlist->dagHdr->numCommits++;
    514   1.4    oster 			/*
    515   1.4    oster 		         * Okay, here we're calling rf_FinishNode() on nodes that
    516   1.4    oster 		         * have the null function as their work proc. Such a node
    517   1.4    oster 		         * could be the terminal node in a DAG. If so, it will
    518   1.4    oster 		         * cause the DAG to complete, which will in turn free
    519   1.4    oster 		         * memory used by the DAG, which includes the node in
    520   1.4    oster 		         * question. Thus, we must avoid referencing the node
    521   1.4    oster 		         * at all after calling rf_FinishNode() on it.
    522   1.4    oster 		         */
    523   1.4    oster 			rf_FinishNode(finishlist, context);	/* recursive call */
    524   1.4    oster 		}
    525   1.4    oster 		/* fire all nodes in firelist */
    526   1.4    oster 		FireNodeList(firelist);
    527   1.4    oster 		break;
    528   1.4    oster 
    529   1.4    oster 	case rf_rollBackward:
    530   1.4    oster 		for (i = 0; i < node->numAntecedents; i++) {
    531   1.4    oster 			a = *(node->antecedents + i);
    532   1.4    oster 			RF_ASSERT(a->status == rf_good);
    533   1.4    oster 			RF_ASSERT(a->numSuccDone <= a->numSuccedents);
    534   1.4    oster 			RF_ASSERT(a->numSuccDone <= a->numSuccFired);
    535   1.4    oster 
    536   1.4    oster 			if (a->numSuccDone == a->numSuccFired) {
    537   1.4    oster 				if (a->undoFunc == rf_NullNodeFunc) {
    538   1.4    oster 					/* don't fire NIL nodes, just process
    539   1.4    oster 					 * them */
    540   1.4    oster 					a->next = finishlist;
    541   1.4    oster 					finishlist = a;
    542   1.4    oster 				} else {
    543   1.4    oster 					if (context != RF_INTR_CONTEXT) {
    544   1.4    oster 						/* we only have to enqueue if
    545   1.4    oster 						 * we're at intr context */
    546   1.4    oster 						a->next = firelist;	/* put node on a list to
    547   1.4    oster 									 * be fired after we
    548   1.4    oster 									 * unlock */
    549   1.4    oster 						firelist = a;
    550   1.4    oster 					} else {	/* enqueue the node for
    551   1.4    oster 							 * the dag exec thread
    552   1.4    oster 							 * to fire */
    553   1.4    oster 						RF_ASSERT(NodeReady(a));
    554   1.4    oster 						if (q) {
    555   1.4    oster 							q->next = a;
    556   1.4    oster 							q = a;
    557   1.4    oster 						} else {
    558   1.4    oster 							qh = q = a;
    559   1.4    oster 							qh->next = NULL;
    560   1.4    oster 						}
    561   1.4    oster 					}
    562   1.4    oster 				}
    563   1.4    oster 			}
    564   1.4    oster 		}
    565   1.4    oster 		if (q) {
    566   1.4    oster 			/* xfer our local list of nodes to the node queue */
    567   1.4    oster 			q->next = raidPtr->node_queue;
    568   1.4    oster 			raidPtr->node_queue = qh;
    569   1.4    oster 			DO_SIGNAL(raidPtr);
    570   1.4    oster 		}
    571   1.4    oster 		DO_UNLOCK(raidPtr);
    572   1.4    oster 		for (; finishlist; finishlist = next) {	/* NIL nodes: no need to
    573   1.4    oster 							 * fire them */
    574   1.4    oster 			next = finishlist->next;
    575   1.4    oster 			finishlist->status = rf_good;
    576   1.4    oster 			/*
    577   1.4    oster 		         * Okay, here we're calling rf_FinishNode() on nodes that
    578   1.4    oster 		         * have the null function as their work proc. Such a node
    579   1.4    oster 		         * could be the first node in a DAG. If so, it will
    580   1.4    oster 		         * cause the DAG to complete, which will in turn free
    581   1.4    oster 		         * memory used by the DAG, which includes the node in
    582   1.4    oster 		         * question. Thus, we must avoid referencing the node
    583   1.4    oster 		         * at all after calling rf_FinishNode() on it.
    584   1.4    oster 		         */
    585   1.4    oster 			rf_FinishNode(finishlist, context);	/* recursive call */
    586   1.4    oster 		}
    587   1.4    oster 		/* fire all nodes in firelist */
    588   1.4    oster 		FireNodeList(firelist);
    589   1.4    oster 
    590   1.4    oster 		break;
    591   1.4    oster 	default:
    592   1.4    oster 		printf("Engine found illegal DAG status in PropagateResults()\n");
    593   1.4    oster 		RF_PANIC();
    594   1.4    oster 		break;
    595   1.4    oster 	}
    596   1.1    oster }
    597   1.1    oster 
    598   1.1    oster 
    599   1.1    oster 
    600   1.1    oster /*
    601   1.1    oster  * Process a fired node which has completed
    602   1.1    oster  */
    603   1.4    oster static void
    604   1.4    oster ProcessNode(
    605   1.4    oster     RF_DagNode_t * node,
    606   1.4    oster     int context)
    607   1.4    oster {
    608   1.4    oster 	RF_Raid_t *raidPtr;
    609   1.4    oster 
    610   1.4    oster 	raidPtr = node->dagHdr->raidPtr;
    611   1.4    oster 
    612   1.4    oster 	switch (node->status) {
    613   1.4    oster 	case rf_good:
    614   1.4    oster 		/* normal case, don't need to do anything */
    615   1.4    oster 		break;
    616   1.4    oster 	case rf_bad:
    617   1.4    oster 		if ((node->dagHdr->numCommits > 0) || (node->dagHdr->numCommitNodes == 0)) {
    618   1.4    oster 			node->dagHdr->status = rf_rollForward;	/* crossed commit
    619   1.4    oster 								 * barrier */
    620   1.4    oster 			if (rf_engineDebug || 1) {
    621   1.9    oster 				printf("raid%d: node (%s) returned fail, rolling forward\n", raidPtr->raidid, node->name);
    622   1.4    oster 			}
    623   1.4    oster 		} else {
    624   1.4    oster 			node->dagHdr->status = rf_rollBackward;	/* never reached commit
    625   1.4    oster 								 * barrier */
    626   1.4    oster 			if (rf_engineDebug || 1) {
    627   1.9    oster 				printf("raid%d: node (%s) returned fail, rolling backward\n", raidPtr->raidid, node->name);
    628   1.4    oster 			}
    629   1.4    oster 		}
    630   1.4    oster 		break;
    631   1.4    oster 	case rf_undone:
    632   1.4    oster 		/* normal rollBackward case, don't need to do anything */
    633   1.4    oster 		break;
    634   1.4    oster 	case rf_panic:
    635   1.4    oster 		/* an undo node failed!!! */
    636   1.4    oster 		printf("UNDO of a node failed!!!/n");
    637   1.4    oster 		break;
    638   1.4    oster 	default:
    639   1.4    oster 		printf("node finished execution with an illegal status!!!\n");
    640   1.4    oster 		RF_PANIC();
    641   1.4    oster 		break;
    642   1.4    oster 	}
    643   1.1    oster 
    644   1.4    oster 	/* enqueue node's succedents (antecedents if rollBackward) for
    645   1.4    oster 	 * execution */
    646   1.4    oster 	PropagateResults(node, context);
    647   1.1    oster }
    648   1.1    oster 
    649   1.1    oster 
    650   1.1    oster 
    651   1.1    oster /* user context or dag-exec-thread context:
    652   1.1    oster  * This is the first step in post-processing a newly-completed node.
    653   1.1    oster  * This routine is called by each node execution function to mark the node
    654   1.1    oster  * as complete and fire off any successors that have been enabled.
    655   1.1    oster  */
    656   1.4    oster int
    657   1.4    oster rf_FinishNode(
    658   1.4    oster     RF_DagNode_t * node,
    659   1.4    oster     int context)
    660   1.4    oster {
    661   1.4    oster 	int     retcode = RF_FALSE;
    662   1.4    oster 	node->dagHdr->numNodesCompleted++;
    663   1.4    oster 	ProcessNode(node, context);
    664   1.1    oster 
    665   1.4    oster 	return (retcode);
    666   1.1    oster }
    667   1.1    oster 
    668   1.1    oster 
    669   1.1    oster /* user context:
    670   1.1    oster  * submit dag for execution, return non-zero if we have to wait for completion.
    671   1.1    oster  * if and only if we return non-zero, we'll cause cbFunc to get invoked with
    672   1.1    oster  * cbArg when the DAG has completed.
    673   1.1    oster  *
    674   1.1    oster  * for now we always return 1.  If the DAG does not cause any I/O, then the callback
    675   1.1    oster  * may get invoked before DispatchDAG returns.  There's code in state 5 of ContinueRaidAccess
    676   1.1    oster  * to handle this.
    677   1.1    oster  *
    678   1.1    oster  * All we do here is fire the direct successors of the header node.  The
    679   1.1    oster  * DAG execution thread does the rest of the dag processing.
    680   1.1    oster  */
    681   1.4    oster int
    682   1.4    oster rf_DispatchDAG(
    683   1.4    oster     RF_DagHeader_t * dag,
    684   1.4    oster     void (*cbFunc) (void *),
    685   1.4    oster     void *cbArg)
    686   1.4    oster {
    687   1.4    oster 	RF_Raid_t *raidPtr;
    688   1.4    oster 
    689   1.4    oster 	raidPtr = dag->raidPtr;
    690   1.4    oster 	if (dag->tracerec) {
    691   1.4    oster 		RF_ETIMER_START(dag->tracerec->timer);
    692   1.4    oster 	}
    693  1.13    oster #if DEBUG
    694  1.14    oster #if RF_DEBUG_VALIDATE_DAG
    695   1.4    oster 	if (rf_engineDebug || rf_validateDAGDebug) {
    696   1.4    oster 		if (rf_ValidateDAG(dag))
    697   1.4    oster 			RF_PANIC();
    698   1.4    oster 	}
    699  1.14    oster #endif
    700  1.13    oster #endif
    701   1.4    oster 	if (rf_engineDebug) {
    702   1.9    oster 		printf("raid%d: Entering DispatchDAG\n", raidPtr->raidid);
    703   1.4    oster 	}
    704   1.4    oster 	raidPtr->dags_in_flight++;	/* debug only:  blow off proper
    705   1.4    oster 					 * locking */
    706   1.4    oster 	dag->cbFunc = cbFunc;
    707   1.4    oster 	dag->cbArg = cbArg;
    708   1.4    oster 	dag->numNodesCompleted = 0;
    709   1.4    oster 	dag->status = rf_enable;
    710   1.4    oster 	FireNodeArray(dag->numSuccedents, dag->succedents);
    711   1.4    oster 	return (1);
    712   1.1    oster }
    713   1.1    oster /* dedicated kernel thread:
    714   1.1    oster  * the thread that handles all DAG node firing.
    715   1.1    oster  * To minimize locking and unlocking, we grab a copy of the entire node queue and then set the
    716   1.1    oster  * node queue to NULL before doing any firing of nodes.  This way we only have to release the
    717   1.1    oster  * lock once.  Of course, it's probably rare that there's more than one node in the queue at
    718   1.1    oster  * any one time, but it sometimes happens.
    719   1.1    oster  *
    720   1.1    oster  * In the kernel, this thread runs at spl0 and is not swappable.  I copied these
    721   1.1    oster  * characteristics from the aio_completion_thread.
    722   1.1    oster  */
    723   1.1    oster 
    724   1.4    oster static void
    725   1.4    oster DAGExecutionThread(RF_ThreadArg_t arg)
    726   1.1    oster {
    727   1.4    oster 	RF_DagNode_t *nd, *local_nq, *term_nq, *fire_nq;
    728   1.4    oster 	RF_Raid_t *raidPtr;
    729   1.9    oster 	int     ks;
    730   1.4    oster 	int     s;
    731   1.4    oster 
    732   1.4    oster 	raidPtr = (RF_Raid_t *) arg;
    733   1.4    oster 
    734   1.4    oster 	if (rf_engineDebug) {
    735   1.9    oster 		printf("raid%d: Engine thread is running\n", raidPtr->raidid);
    736   1.4    oster 	}
    737   1.1    oster 
    738   1.4    oster 	s = splbio();
    739   1.1    oster 
    740   1.4    oster 	RF_THREADGROUP_RUNNING(&raidPtr->engine_tg);
    741   1.1    oster 
    742   1.4    oster 	DO_LOCK(raidPtr);
    743   1.4    oster 	while (!raidPtr->shutdown_engine) {
    744   1.1    oster 
    745   1.4    oster 		while (raidPtr->node_queue != NULL) {
    746   1.4    oster 			local_nq = raidPtr->node_queue;
    747   1.4    oster 			fire_nq = NULL;
    748   1.4    oster 			term_nq = NULL;
    749   1.4    oster 			raidPtr->node_queue = NULL;
    750   1.4    oster 			DO_UNLOCK(raidPtr);
    751   1.4    oster 
    752   1.4    oster 			/* first, strip out the terminal nodes */
    753   1.4    oster 			while (local_nq) {
    754   1.4    oster 				nd = local_nq;
    755   1.4    oster 				local_nq = local_nq->next;
    756   1.4    oster 				switch (nd->dagHdr->status) {
    757   1.4    oster 				case rf_enable:
    758   1.4    oster 				case rf_rollForward:
    759   1.4    oster 					if (nd->numSuccedents == 0) {
    760   1.4    oster 						/* end of the dag, add to
    761   1.4    oster 						 * callback list */
    762   1.4    oster 						nd->next = term_nq;
    763   1.4    oster 						term_nq = nd;
    764   1.4    oster 					} else {
    765   1.4    oster 						/* not the end, add to the
    766   1.4    oster 						 * fire queue */
    767   1.4    oster 						nd->next = fire_nq;
    768   1.4    oster 						fire_nq = nd;
    769   1.4    oster 					}
    770   1.4    oster 					break;
    771   1.4    oster 				case rf_rollBackward:
    772   1.4    oster 					if (nd->numAntecedents == 0) {
    773   1.4    oster 						/* end of the dag, add to the
    774   1.4    oster 						 * callback list */
    775   1.4    oster 						nd->next = term_nq;
    776   1.4    oster 						term_nq = nd;
    777   1.4    oster 					} else {
    778   1.4    oster 						/* not the end, add to the
    779   1.4    oster 						 * fire queue */
    780   1.4    oster 						nd->next = fire_nq;
    781   1.4    oster 						fire_nq = nd;
    782   1.4    oster 					}
    783   1.4    oster 					break;
    784   1.4    oster 				default:
    785   1.4    oster 					RF_PANIC();
    786   1.4    oster 					break;
    787   1.4    oster 				}
    788   1.4    oster 			}
    789   1.4    oster 
    790   1.4    oster 			/* execute callback of dags which have reached the
    791   1.4    oster 			 * terminal node */
    792   1.4    oster 			while (term_nq) {
    793   1.4    oster 				nd = term_nq;
    794   1.4    oster 				term_nq = term_nq->next;
    795   1.4    oster 				nd->next = NULL;
    796   1.4    oster 				(nd->dagHdr->cbFunc) (nd->dagHdr->cbArg);
    797   1.4    oster 				raidPtr->dags_in_flight--;	/* debug only */
    798   1.4    oster 			}
    799   1.4    oster 
    800   1.4    oster 			/* fire remaining nodes */
    801   1.4    oster 			FireNodeList(fire_nq);
    802   1.4    oster 
    803   1.4    oster 			DO_LOCK(raidPtr);
    804   1.4    oster 		}
    805  1.17    oster 		while (!raidPtr->shutdown_engine &&
    806  1.17    oster 		       raidPtr->node_queue == NULL) {
    807  1.17    oster 			DO_UNLOCK(raidPtr);
    808   1.4    oster 			DO_WAIT(raidPtr);
    809  1.17    oster 			DO_LOCK(raidPtr);
    810  1.17    oster 		}
    811   1.4    oster 	}
    812   1.4    oster 	DO_UNLOCK(raidPtr);
    813   1.1    oster 
    814   1.4    oster 	RF_THREADGROUP_DONE(&raidPtr->engine_tg);
    815   1.8    oster 
    816   1.4    oster 	splx(s);
    817   1.4    oster 	kthread_exit(0);
    818   1.1    oster }
    819