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