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