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