1096 lines
32 KiB
C
1096 lines
32 KiB
C
/* $NetBSD: rf_engine.c,v 1.2 1998/11/13 11:48:26 simonb Exp $ */
|
|
/*
|
|
* Copyright (c) 1995 Carnegie-Mellon University.
|
|
* All rights reserved.
|
|
*
|
|
* Author: William V. Courtright II, Mark Holland, Rachad Youssef
|
|
*
|
|
* Permission to use, copy, modify and distribute this software and
|
|
* its documentation is hereby granted, provided that both the copyright
|
|
* notice and this permission notice appear in all copies of the
|
|
* software, derivative works or modified versions, and any portions
|
|
* thereof, and that both notices appear in supporting documentation.
|
|
*
|
|
* CARNEGIE MELLON ALLOWS FREE USE OF THIS SOFTWARE IN ITS "AS IS"
|
|
* CONDITION. CARNEGIE MELLON DISCLAIMS ANY LIABILITY OF ANY KIND
|
|
* FOR ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF THIS SOFTWARE.
|
|
*
|
|
* Carnegie Mellon requests users of this software to return to
|
|
*
|
|
* Software Distribution Coordinator or Software.Distribution@CS.CMU.EDU
|
|
* School of Computer Science
|
|
* Carnegie Mellon University
|
|
* Pittsburgh PA 15213-3890
|
|
*
|
|
* any improvements or extensions that they make and grant Carnegie the
|
|
* rights to redistribute these changes.
|
|
*/
|
|
|
|
/****************************************************************************
|
|
* *
|
|
* engine.c -- code for DAG execution engine *
|
|
* *
|
|
* Modified to work as follows (holland): *
|
|
* A user-thread calls into DispatchDAG, which fires off the nodes that *
|
|
* are direct successors to the header node. DispatchDAG then returns, *
|
|
* and the rest of the I/O continues asynchronously. As each node *
|
|
* completes, the node execution function calls FinishNode(). FinishNode *
|
|
* scans the list of successors to the node and increments the antecedent *
|
|
* counts. Each node that becomes enabled is placed on a central node *
|
|
* queue. A dedicated dag-execution thread grabs nodes off of this *
|
|
* queue and fires them. *
|
|
* *
|
|
* NULL nodes are never fired. *
|
|
* *
|
|
* Terminator nodes are never fired, but rather cause the callback *
|
|
* associated with the DAG to be invoked. *
|
|
* *
|
|
* If a node fails, the dag either rolls forward to the completion or *
|
|
* rolls back, undoing previously-completed nodes and fails atomically. *
|
|
* The direction of recovery is determined by the location of the failed *
|
|
* node in the graph. If the failure occured before the commit node in *
|
|
* the graph, backward recovery is used. Otherwise, forward recovery is *
|
|
* used. *
|
|
* *
|
|
****************************************************************************/
|
|
|
|
/*
|
|
* :
|
|
*
|
|
* Log: rf_engine.c,v
|
|
* Revision 1.56 1996/07/28 20:31:39 jimz
|
|
* i386netbsd port
|
|
* true/false fixup
|
|
*
|
|
* Revision 1.55 1996/07/22 19:52:16 jimz
|
|
* switched node params to RF_DagParam_t, a union of
|
|
* a 64-bit int and a void *, for better portability
|
|
* attempted hpux port, but failed partway through for
|
|
* lack of a single C compiler capable of compiling all
|
|
* source files
|
|
*
|
|
* Revision 1.54 1996/07/17 21:00:58 jimz
|
|
* clean up timer interface, tracing
|
|
*
|
|
* Revision 1.53 1996/07/15 17:22:18 jimz
|
|
* nit-pick code cleanup
|
|
* resolve stdlib problems on DEC OSF
|
|
*
|
|
* Revision 1.52 1996/06/17 03:17:08 jimz
|
|
* correctly shut down engine thread in kernel
|
|
*
|
|
* Revision 1.51 1996/06/14 15:02:10 jimz
|
|
* make new engine code happy in simulator
|
|
*
|
|
* Revision 1.50 1996/06/14 14:19:48 jimz
|
|
* use diskgroup to control engine thread, make all engine-thread-related
|
|
* stuff per-array
|
|
*
|
|
* Revision 1.49 1996/06/10 11:55:47 jimz
|
|
* Straightened out some per-array/not-per-array distinctions, fixed
|
|
* a couple bugs related to confusion. Added shutdown lists. Removed
|
|
* layout shutdown function (now subsumed by shutdown lists).
|
|
*
|
|
* Revision 1.48 1996/06/09 02:36:46 jimz
|
|
* lots of little crufty cleanup- fixup whitespace
|
|
* issues, comment #ifdefs, improve typing in some
|
|
* places (esp size-related)
|
|
*
|
|
* Revision 1.47 1996/06/06 01:23:23 jimz
|
|
* fix bug in node traversal when firing multiple nodes simultaneously
|
|
*
|
|
* Revision 1.46 1996/06/05 18:06:02 jimz
|
|
* Major code cleanup. The Great Renaming is now done.
|
|
* Better modularity. Better typing. Fixed a bunch of
|
|
* synchronization bugs. Made a lot of global stuff
|
|
* per-desc or per-array. Removed dead code.
|
|
*
|
|
* Revision 1.45 1996/05/30 12:59:18 jimz
|
|
* make etimer happier, more portable
|
|
*
|
|
* Revision 1.44 1996/05/30 11:29:41 jimz
|
|
* Numerous bug fixes. Stripe lock release code disagreed with the taking code
|
|
* about when stripes should be locked (I made it consistent: no parity, no lock)
|
|
* There was a lot of extra serialization of I/Os which I've removed- a lot of
|
|
* it was to calculate values for the cache code, which is no longer with us.
|
|
* More types, function, macro cleanup. Added code to properly quiesce the array
|
|
* on shutdown. Made a lot of stuff array-specific which was (bogusly) general
|
|
* before. Fixed memory allocation, freeing bugs.
|
|
*
|
|
* Revision 1.43 1996/05/27 18:56:37 jimz
|
|
* more code cleanup
|
|
* better typing
|
|
* compiles in all 3 environments
|
|
*
|
|
* Revision 1.42 1996/05/24 22:17:04 jimz
|
|
* continue code + namespace cleanup
|
|
* typed a bunch of flags
|
|
*
|
|
* Revision 1.41 1996/05/24 04:28:55 jimz
|
|
* release cleanup ckpt
|
|
*
|
|
* Revision 1.40 1996/05/23 00:33:23 jimz
|
|
* code cleanup: move all debug decls to rf_options.c, all extern
|
|
* debug decls to rf_options.h, all debug vars preceded by rf_
|
|
*
|
|
* Revision 1.39 1996/05/20 16:15:17 jimz
|
|
* switch to rf_{mutex,cond}_{init,destroy}
|
|
*
|
|
* Revision 1.38 1996/05/18 20:09:54 jimz
|
|
* bit of cleanup to compile cleanly in kernel, once again
|
|
*
|
|
* Revision 1.37 1996/05/18 19:51:34 jimz
|
|
* major code cleanup- fix syntax, make some types consistent,
|
|
* add prototypes, clean out dead code, et cetera
|
|
*
|
|
* Revision 1.36 1996/05/15 20:24:19 wvcii
|
|
* fixed syntax bug in SIMULATE clause above ProcessNode
|
|
*
|
|
* Revision 1.35 1996/05/08 21:01:24 jimz
|
|
* fixed up enum type names that were conflicting with other
|
|
* enums and function names (ie, "panic")
|
|
* future naming trends will be towards RF_ and rf_ for
|
|
* everything raidframe-related
|
|
*
|
|
* Revision 1.34 1996/05/08 15:25:28 wvcii
|
|
* eliminated dead code
|
|
* merged common cases (sim/user/kernel)
|
|
* entire node lists (arrays) now fired atomically
|
|
* reordered source code for readability
|
|
* beefed-up & corrected comments
|
|
*
|
|
* Revision 1.33 1996/05/07 19:39:40 jimz
|
|
* 1. fixed problems in PropogateResults() with nodes being referenced
|
|
* after they were no longer valid
|
|
* 2. fixed problems in PropogateResults() with the node list being
|
|
* incorrectly threaded
|
|
*
|
|
* Revision 1.32 1996/05/07 19:03:56 wvcii
|
|
* in PropagateResults, fixed a bug in the rollBackward case:
|
|
* node data is copied before the call to FinishNode which
|
|
* frees the node and destroys its data.
|
|
*
|
|
* Revision 1.31 1996/05/07 17:45:17 jimz
|
|
* remove old #if 0 code from PropogateResults() (was kept in
|
|
* previous version for archival purposes (rcsdiff))
|
|
*
|
|
* Revision 1.30 1996/05/07 17:44:19 jimz
|
|
* fix threading of nodes to be fired in PropagateResults()
|
|
* fix iteration through skiplist in PropagateResults()
|
|
* fix incorrect accesses to freed memory (dereferencing a
|
|
* node that was freed by the action of calling FinishNode()
|
|
* on it, which in turn completed its DAG) in PropagateResults()
|
|
*
|
|
* Revision 1.29 1996/05/02 15:04:15 wvcii
|
|
* fixed bad array index in PropagateResults
|
|
*
|
|
* Revision 1.28 1995/12/12 18:10:06 jimz
|
|
* MIN -> RF_MIN, MAX -> RF_MAX, ASSERT -> RF_ASSERT
|
|
* fix 80-column brain damage in comments
|
|
*
|
|
* Revision 1.27 1995/12/08 15:07:03 arw
|
|
* cache code cleanup
|
|
*
|
|
* Revision 1.26 1995/11/07 16:18:01 wvcii
|
|
* numerous changes associated with roll-away error recovery
|
|
* when a node fails, dag enters rollForward or rollBackward state
|
|
*
|
|
* Revision 1.25 1995/09/06 19:27:17 wvcii
|
|
* added debug vars enableRollAway and debugRecovery
|
|
*
|
|
*/
|
|
|
|
#ifdef _KERNEL
|
|
#define KERNEL
|
|
#endif
|
|
|
|
#include "rf_threadstuff.h"
|
|
|
|
#ifndef KERNEL
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#endif /* !KERNEL */
|
|
|
|
#include <sys/errno.h>
|
|
|
|
#include "rf_dag.h"
|
|
#include "rf_engine.h"
|
|
#include "rf_threadid.h"
|
|
#include "rf_etimer.h"
|
|
#include "rf_general.h"
|
|
#include "rf_dagutils.h"
|
|
#include "rf_shutdown.h"
|
|
#include "rf_raid.h"
|
|
|
|
#ifndef SIMULATE
|
|
static void DAGExecutionThread(RF_ThreadArg_t arg);
|
|
#endif /* !SIMULATE */
|
|
|
|
#define DO_INIT(_l_,_r_) { \
|
|
int _rc; \
|
|
_rc = rf_create_managed_mutex(_l_,&(_r_)->node_queue_mutex); \
|
|
if (_rc) { \
|
|
return(_rc); \
|
|
} \
|
|
_rc = rf_create_managed_cond(_l_,&(_r_)->node_queue_cond); \
|
|
if (_rc) { \
|
|
return(_rc); \
|
|
} \
|
|
}
|
|
|
|
/* synchronization primitives for this file. DO_WAIT should be enclosed in a while loop. */
|
|
#ifndef KERNEL
|
|
|
|
#define DO_LOCK(_r_) RF_LOCK_MUTEX((_r_)->node_queue_mutex)
|
|
#define DO_UNLOCK(_r_) RF_UNLOCK_MUTEX((_r_)->node_queue_mutex)
|
|
#define DO_WAIT(_r_) RF_WAIT_COND((_r_)->node_queue_cond, (_r_)->node_queue_mutex)
|
|
#define DO_SIGNAL(_r_) RF_SIGNAL_COND((_r_)->node_queue_cond)
|
|
|
|
#else /* !KERNEL */
|
|
|
|
/*
|
|
* XXX Is this spl-ing really necessary?
|
|
*/
|
|
#define DO_LOCK(_r_) { ks = splbio(); RF_LOCK_MUTEX((_r_)->node_queue_mutex); }
|
|
#define DO_UNLOCK(_r_) { RF_UNLOCK_MUTEX((_r_)->node_queue_mutex); splx(ks); }
|
|
#ifndef __NetBSD__
|
|
#define DO_WAIT(_r_) mpsleep(&(_r_)->node_queue, PZERO, "raidframe nq", 0, (void *) simple_lock_addr((_r_)->node_queue_mutex), MS_LOCK_SIMPLE)
|
|
#else
|
|
#define DO_WAIT(_r_) tsleep(&(_r_)->node_queue, PRIBIO | PCATCH, "raidframe nq",0)
|
|
#endif
|
|
#define DO_SIGNAL(_r_) wakeup(&(_r_)->node_queue)
|
|
|
|
#endif /* !KERNEL */
|
|
|
|
static void rf_ShutdownEngine(void *);
|
|
|
|
static void rf_ShutdownEngine(arg)
|
|
void *arg;
|
|
{
|
|
RF_Raid_t *raidPtr;
|
|
|
|
raidPtr = (RF_Raid_t *)arg;
|
|
#ifndef SIMULATE
|
|
raidPtr->shutdown_engine = 1;
|
|
DO_SIGNAL(raidPtr);
|
|
/* XXX something is missing here... */
|
|
#ifdef DEBUG
|
|
printf("IGNORING WAIT_STOP\n");
|
|
#endif
|
|
#if 0
|
|
RF_THREADGROUP_WAIT_STOP(&raidPtr->engine_tg);
|
|
#endif
|
|
#endif /* !SIMULATE */
|
|
}
|
|
|
|
int rf_ConfigureEngine(
|
|
RF_ShutdownList_t **listp,
|
|
RF_Raid_t *raidPtr,
|
|
RF_Config_t *cfgPtr)
|
|
{
|
|
int rc, tid=0;
|
|
|
|
if (rf_engineDebug) {
|
|
rf_get_threadid(tid);
|
|
}
|
|
|
|
DO_INIT(listp,raidPtr);
|
|
|
|
raidPtr->node_queue = NULL;
|
|
raidPtr->dags_in_flight = 0;
|
|
|
|
#ifndef SIMULATE
|
|
rc = rf_init_managed_threadgroup(listp, &raidPtr->engine_tg);
|
|
if (rc)
|
|
return(rc);
|
|
|
|
/* we create the execution thread only once per system boot.
|
|
* no need to check return code b/c the kernel panics if it can't create the thread.
|
|
*/
|
|
if (rf_engineDebug) {
|
|
printf("[%d] Creating engine thread\n", tid);
|
|
}
|
|
|
|
if (RF_CREATE_THREAD(raidPtr->engine_thread, DAGExecutionThread, raidPtr)) {
|
|
RF_ERRORMSG("RAIDFRAME: Unable to create engine thread\n");
|
|
return(ENOMEM);
|
|
}
|
|
if (rf_engineDebug) {
|
|
printf("[%d] Created engine thread\n", tid);
|
|
}
|
|
RF_THREADGROUP_STARTED(&raidPtr->engine_tg);
|
|
/* XXX something is missing here... */
|
|
#ifdef debug
|
|
printf("Skipping the WAIT_START!!\n");
|
|
#endif
|
|
#if 0
|
|
RF_THREADGROUP_WAIT_START(&raidPtr->engine_tg);
|
|
#endif
|
|
/* engine thread is now running and waiting for work */
|
|
if (rf_engineDebug) {
|
|
printf("[%d] Engine thread running and waiting for events\n", tid);
|
|
}
|
|
#endif /* !SIMULATE */
|
|
|
|
rc = rf_ShutdownCreate(listp, rf_ShutdownEngine, raidPtr);
|
|
if (rc) {
|
|
RF_ERRORMSG3("Unable to add to shutdown list file %s line %d rc=%d\n", __FILE__,
|
|
__LINE__, rc);
|
|
rf_ShutdownEngine(NULL);
|
|
}
|
|
|
|
return(rc);
|
|
}
|
|
|
|
static int BranchDone(RF_DagNode_t *node)
|
|
{
|
|
int i;
|
|
|
|
/* return true if forward execution is completed for a node and it's succedents */
|
|
switch (node->status) {
|
|
case rf_wait :
|
|
/* should never be called in this state */
|
|
RF_PANIC();
|
|
break;
|
|
case rf_fired :
|
|
/* node is currently executing, so we're not done */
|
|
return(RF_FALSE);
|
|
case rf_good :
|
|
for (i = 0; i < node->numSuccedents; i++) /* for each succedent */
|
|
if (!BranchDone(node->succedents[i])) /* recursively check branch */
|
|
return RF_FALSE;
|
|
return RF_TRUE; /* node and all succedent branches aren't in fired state */
|
|
break;
|
|
case rf_bad :
|
|
/* succedents can't fire */
|
|
return(RF_TRUE);
|
|
case rf_recover :
|
|
/* should never be called in this state */
|
|
RF_PANIC();
|
|
break;
|
|
case rf_undone :
|
|
case rf_panic :
|
|
/* XXX need to fix this case */
|
|
/* for now, assume that we're done */
|
|
return(RF_TRUE);
|
|
break;
|
|
default :
|
|
/* illegal node status */
|
|
RF_PANIC();
|
|
break;
|
|
}
|
|
}
|
|
|
|
#ifdef SIMULATE
|
|
/* this is only ifdef SIMULATE because nothing else needs it */
|
|
/* recursively determine if a DAG has completed execution */
|
|
static int DAGDone(RF_DagHeader_t *dag)
|
|
{
|
|
int i;
|
|
|
|
for (i = 0; i < dag->numSuccedents; i++)
|
|
if (!BranchDone(dag->succedents[i]))
|
|
return RF_FALSE;
|
|
return RF_TRUE;
|
|
}
|
|
#endif /* SIMULATE */
|
|
|
|
static int NodeReady(RF_DagNode_t *node)
|
|
{
|
|
int ready;
|
|
|
|
switch (node->dagHdr->status) {
|
|
case rf_enable :
|
|
case rf_rollForward :
|
|
if ((node->status == rf_wait) && (node->numAntecedents == node->numAntDone))
|
|
ready = RF_TRUE;
|
|
else
|
|
ready = RF_FALSE;
|
|
break;
|
|
case rf_rollBackward :
|
|
RF_ASSERT(node->numSuccDone <= node->numSuccedents);
|
|
RF_ASSERT(node->numSuccFired <= node->numSuccedents);
|
|
RF_ASSERT(node->numSuccFired <= node->numSuccDone);
|
|
if ((node->status == rf_good) && (node->numSuccDone == node->numSuccedents))
|
|
ready = RF_TRUE;
|
|
else
|
|
ready = RF_FALSE;
|
|
break;
|
|
default :
|
|
printf("Execution engine found illegal DAG status in NodeReady\n");
|
|
RF_PANIC();
|
|
break;
|
|
}
|
|
|
|
return(ready);
|
|
}
|
|
|
|
|
|
|
|
/* user context and dag-exec-thread context:
|
|
* Fire a node. The node's status field determines which function, do or undo,
|
|
* to be fired.
|
|
* This routine assumes that the node's status field has alread been set to
|
|
* "fired" or "recover" to indicate the direction of execution.
|
|
*/
|
|
static void FireNode(RF_DagNode_t *node)
|
|
{
|
|
int tid;
|
|
|
|
switch (node->status) {
|
|
case rf_fired :
|
|
/* fire the do function of a node */
|
|
if (rf_engineDebug) {
|
|
rf_get_threadid(tid);
|
|
printf("[%d] Firing node 0x%lx (%s)\n",tid,(unsigned long) node, node->name);
|
|
}
|
|
#ifdef KERNEL
|
|
if (node->flags & RF_DAGNODE_FLAG_YIELD) {
|
|
#if defined(__NetBSD__) && defined(_KERNEL)
|
|
/* thread_block(); */
|
|
/* printf("Need to block the thread here...\n"); */
|
|
/* XXX thread_block is actually mentioned in
|
|
/usr/include/vm/vm_extern.h */
|
|
#else
|
|
thread_block();
|
|
#endif
|
|
}
|
|
#endif /* KERNEL */
|
|
(*(node->doFunc)) (node);
|
|
break;
|
|
case rf_recover :
|
|
/* fire the undo function of a node */
|
|
if (rf_engineDebug || 1) {
|
|
rf_get_threadid(tid);
|
|
printf("[%d] Firing (undo) node 0x%lx (%s)\n",tid,(unsigned long) node, node->name);
|
|
}
|
|
#ifdef KERNEL
|
|
if (node->flags & RF_DAGNODE_FLAG_YIELD)
|
|
#if defined(__NetBSD__) && defined(_KERNEL)
|
|
/* thread_block(); */
|
|
/* printf("Need to block the thread here...\n"); */
|
|
/* XXX thread_block is actually mentioned in
|
|
/usr/include/vm/vm_extern.h */
|
|
#else
|
|
thread_block();
|
|
#endif
|
|
#endif /* KERNEL */
|
|
(*(node->undoFunc)) (node);
|
|
break;
|
|
default :
|
|
RF_PANIC();
|
|
break;
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/* user context:
|
|
* Attempt to fire each node in a linear array.
|
|
* The entire list is fired atomically.
|
|
*/
|
|
static void FireNodeArray(
|
|
int numNodes,
|
|
RF_DagNode_t **nodeList)
|
|
{
|
|
RF_DagStatus_t dstat;
|
|
RF_DagNode_t *node;
|
|
int i, j;
|
|
|
|
/* first, mark all nodes which are ready to be fired */
|
|
for (i = 0; i < numNodes; i++) {
|
|
node = nodeList[i];
|
|
dstat = node->dagHdr->status;
|
|
RF_ASSERT((node->status == rf_wait) || (node->status == rf_good));
|
|
if (NodeReady(node)) {
|
|
if ((dstat == rf_enable) || (dstat == rf_rollForward)) {
|
|
RF_ASSERT(node->status == rf_wait);
|
|
if (node->commitNode)
|
|
node->dagHdr->numCommits++;
|
|
node->status = rf_fired;
|
|
for (j = 0; j < node->numAntecedents; j++)
|
|
node->antecedents[j]->numSuccFired++;
|
|
}
|
|
else {
|
|
RF_ASSERT(dstat == rf_rollBackward);
|
|
RF_ASSERT(node->status == rf_good);
|
|
RF_ASSERT(node->commitNode == RF_FALSE); /* only one commit node per graph */
|
|
node->status = rf_recover;
|
|
}
|
|
}
|
|
}
|
|
/* now, fire the nodes */
|
|
for (i = 0; i < numNodes; i++) {
|
|
if ((nodeList[i]->status == rf_fired) || (nodeList[i]->status == rf_recover))
|
|
FireNode(nodeList[i]);
|
|
}
|
|
}
|
|
|
|
|
|
#ifndef SIMULATE
|
|
/* user context:
|
|
* Attempt to fire each node in a linked list.
|
|
* The entire list is fired atomically.
|
|
*/
|
|
static void FireNodeList(RF_DagNode_t *nodeList)
|
|
{
|
|
RF_DagNode_t *node, *next;
|
|
RF_DagStatus_t dstat;
|
|
int j;
|
|
|
|
if (nodeList) {
|
|
/* first, mark all nodes which are ready to be fired */
|
|
for (node = nodeList; node; node = next) {
|
|
next = node->next;
|
|
dstat = node->dagHdr->status;
|
|
RF_ASSERT((node->status == rf_wait) || (node->status == rf_good));
|
|
if (NodeReady(node)) {
|
|
if ((dstat == rf_enable) || (dstat == rf_rollForward)) {
|
|
RF_ASSERT(node->status == rf_wait);
|
|
if (node->commitNode)
|
|
node->dagHdr->numCommits++;
|
|
node->status = rf_fired;
|
|
for (j = 0; j < node->numAntecedents; j++)
|
|
node->antecedents[j]->numSuccFired++;
|
|
}
|
|
else {
|
|
RF_ASSERT(dstat == rf_rollBackward);
|
|
RF_ASSERT(node->status == rf_good);
|
|
RF_ASSERT(node->commitNode == RF_FALSE); /* only one commit node per graph */
|
|
node->status = rf_recover;
|
|
}
|
|
}
|
|
}
|
|
/* now, fire the nodes */
|
|
for (node = nodeList; node; node = next) {
|
|
next = node->next;
|
|
if ((node->status == rf_fired) || (node->status == rf_recover))
|
|
FireNode(node);
|
|
}
|
|
}
|
|
}
|
|
#endif /* !SIMULATE */
|
|
|
|
|
|
|
|
/* interrupt context:
|
|
* for each succedent
|
|
* propagate required results from node to succedent
|
|
* increment succedent's numAntDone
|
|
* place newly-enable nodes on node queue for firing
|
|
*
|
|
* To save context switches, we don't place NIL nodes on the node queue,
|
|
* but rather just process them as if they had fired. Note that NIL nodes
|
|
* that are the direct successors of the header will actually get fired by
|
|
* DispatchDAG, which is fine because no context switches are involved.
|
|
*
|
|
* Important: when running at user level, this can be called by any
|
|
* disk thread, and so the increment and check of the antecedent count
|
|
* must be locked. I used the node queue mutex and locked down the
|
|
* entire function, but this is certainly overkill.
|
|
*/
|
|
static void PropagateResults(
|
|
RF_DagNode_t *node,
|
|
int context)
|
|
{
|
|
RF_DagNode_t *s, *a;
|
|
RF_Raid_t *raidPtr;
|
|
int tid, i, ks;
|
|
#ifdef SIMULATE
|
|
RF_PropHeader_t *p; /* prop list for succedent i */
|
|
#else /* SIMULATE */
|
|
RF_DagNode_t *finishlist = NULL; /* a list of NIL nodes to be finished */
|
|
RF_DagNode_t *skiplist = NULL; /* list of nodes with failed truedata antecedents */
|
|
RF_DagNode_t *firelist = NULL; /* a list of nodes to be fired */
|
|
RF_DagNode_t *q = NULL, *qh = NULL, *next;
|
|
int j, skipNode;
|
|
#endif /* SIMULATE */
|
|
|
|
rf_get_threadid(tid);
|
|
|
|
raidPtr = node->dagHdr->raidPtr;
|
|
|
|
DO_LOCK(raidPtr);
|
|
|
|
/* debug - validate fire counts */
|
|
for (i = 0; i < node->numAntecedents; i++) {
|
|
a = *(node->antecedents + i);
|
|
RF_ASSERT(a->numSuccFired >= a->numSuccDone);
|
|
RF_ASSERT(a->numSuccFired <= a->numSuccedents);
|
|
a->numSuccDone++;
|
|
}
|
|
|
|
switch (node->dagHdr->status) {
|
|
case rf_enable :
|
|
case rf_rollForward :
|
|
#ifdef SIMULATE
|
|
/* currently we never propagate results unless in simulation */
|
|
for (i = 0; i < node->numSuccedents; i++) {
|
|
s = *(node->succedents + i);
|
|
RF_ASSERT(s->status == rf_wait);
|
|
(s->numAntDone)++;
|
|
if (node->propList == NULL)
|
|
/* null propList implies no results to be propagated */
|
|
p = NULL;
|
|
else
|
|
/* p=head of prop list for succedent i */
|
|
p = *(node->propList + i);
|
|
while (p != NULL) {
|
|
/* bind node results to succedent's parameters */
|
|
#if 0
|
|
*(s->params + p->paramNum) = *(node->results + p->resultNum);
|
|
#else
|
|
s->params[p->paramNum].p = node->results[p->resultNum];
|
|
#endif
|
|
p = p->next;
|
|
}
|
|
}
|
|
#else /* SIMULATE */
|
|
for (i = 0; i < node->numSuccedents; i++) {
|
|
s = *(node->succedents + i);
|
|
RF_ASSERT(s->status == rf_wait);
|
|
(s->numAntDone)++;
|
|
if (s->numAntDone == s->numAntecedents) {
|
|
/* look for NIL nodes */
|
|
if (s->doFunc == rf_NullNodeFunc) {
|
|
/* don't fire NIL nodes, just process them */
|
|
s->next = finishlist;
|
|
finishlist = s;
|
|
}
|
|
else {
|
|
/* look to see if the node is to be skipped */
|
|
skipNode = RF_FALSE;
|
|
for (j = 0; j < s->numAntecedents; j++)
|
|
if ((s->antType[j] == rf_trueData) && (s->antecedents[j]->status == rf_bad))
|
|
skipNode = RF_TRUE;
|
|
if (skipNode) {
|
|
/* this node has one or more failed true data dependencies, so skip it */
|
|
s->next = skiplist;
|
|
skiplist = s;
|
|
}
|
|
else
|
|
/* add s to list of nodes (q) to execute */
|
|
if (context != RF_INTR_CONTEXT) {
|
|
/* we only have to enqueue if we're at intr context */
|
|
s->next = firelist; /* put node on a list to be fired after we unlock */
|
|
firelist = s;
|
|
} else { /* enqueue the node for the dag exec thread to fire */
|
|
RF_ASSERT(NodeReady(s));
|
|
if (q) {
|
|
q->next = s;
|
|
q = s;
|
|
}
|
|
else {
|
|
qh = q = s;
|
|
qh->next = NULL;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if (q) {
|
|
/* xfer our local list of nodes to the node queue */
|
|
q->next = raidPtr->node_queue; raidPtr->node_queue = qh;
|
|
DO_SIGNAL(raidPtr);
|
|
}
|
|
DO_UNLOCK(raidPtr);
|
|
|
|
for (; skiplist; skiplist = next) {
|
|
next = skiplist->next;
|
|
skiplist->status = rf_skipped;
|
|
for (i = 0; i < skiplist->numAntecedents; i++) {
|
|
skiplist->antecedents[i]->numSuccFired++;
|
|
}
|
|
if (skiplist->commitNode) {
|
|
skiplist->dagHdr->numCommits++;
|
|
}
|
|
rf_FinishNode(skiplist, context);
|
|
}
|
|
for (; finishlist; finishlist = next) {
|
|
/* NIL nodes: no need to fire them */
|
|
next = finishlist->next;
|
|
finishlist->status = rf_good;
|
|
for (i = 0; i < finishlist->numAntecedents; i++) {
|
|
finishlist->antecedents[i]->numSuccFired++;
|
|
}
|
|
if (finishlist->commitNode)
|
|
finishlist->dagHdr->numCommits++;
|
|
/*
|
|
* Okay, here we're calling rf_FinishNode() on nodes that
|
|
* have the null function as their work proc. Such a node
|
|
* could be the terminal node in a DAG. If so, it will
|
|
* cause the DAG to complete, which will in turn free
|
|
* memory used by the DAG, which includes the node in
|
|
* question. Thus, we must avoid referencing the node
|
|
* at all after calling rf_FinishNode() on it.
|
|
*/
|
|
rf_FinishNode(finishlist, context); /* recursive call */
|
|
}
|
|
/* fire all nodes in firelist */
|
|
FireNodeList(firelist);
|
|
#endif /* SIMULATE */
|
|
break;
|
|
|
|
case rf_rollBackward :
|
|
#ifdef SIMULATE
|
|
#else /* SIMULATE */
|
|
for (i = 0; i < node->numAntecedents; i++) {
|
|
a = *(node->antecedents + i);
|
|
RF_ASSERT(a->status == rf_good);
|
|
RF_ASSERT(a->numSuccDone <= a->numSuccedents);
|
|
RF_ASSERT(a->numSuccDone <= a->numSuccFired);
|
|
|
|
if (a->numSuccDone == a->numSuccFired) {
|
|
if (a->undoFunc == rf_NullNodeFunc) {
|
|
/* don't fire NIL nodes, just process them */
|
|
a->next = finishlist;
|
|
finishlist = a;
|
|
} else {
|
|
if (context != RF_INTR_CONTEXT) {
|
|
/* we only have to enqueue if we're at intr context */
|
|
a->next = firelist; /* put node on a list to be fired after we unlock */
|
|
firelist = a;
|
|
} else { /* enqueue the node for the dag exec thread to fire */
|
|
RF_ASSERT(NodeReady(a));
|
|
if (q) {
|
|
q->next = a;
|
|
q = a;
|
|
}
|
|
else {
|
|
qh = q = a;
|
|
qh->next = NULL;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
}
|
|
if (q) {
|
|
/* xfer our local list of nodes to the node queue */
|
|
q->next = raidPtr->node_queue; raidPtr->node_queue = qh;
|
|
DO_SIGNAL(raidPtr);
|
|
}
|
|
DO_UNLOCK(raidPtr);
|
|
for (; finishlist; finishlist = next) { /* NIL nodes: no need to fire them */
|
|
next = finishlist->next;
|
|
finishlist->status = rf_good;
|
|
/*
|
|
* Okay, here we're calling rf_FinishNode() on nodes that
|
|
* have the null function as their work proc. Such a node
|
|
* could be the first node in a DAG. If so, it will
|
|
* cause the DAG to complete, which will in turn free
|
|
* memory used by the DAG, which includes the node in
|
|
* question. Thus, we must avoid referencing the node
|
|
* at all after calling rf_FinishNode() on it.
|
|
*/
|
|
rf_FinishNode(finishlist, context); /* recursive call */
|
|
}
|
|
/* fire all nodes in firelist */
|
|
FireNodeList(firelist);
|
|
#endif /* SIMULATE */
|
|
|
|
break;
|
|
default :
|
|
printf("Engine found illegal DAG status in PropagateResults()\n");
|
|
RF_PANIC();
|
|
break;
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
* Process a fired node which has completed
|
|
*/
|
|
static void ProcessNode(
|
|
RF_DagNode_t *node,
|
|
int context)
|
|
{
|
|
RF_Raid_t *raidPtr;
|
|
int tid;
|
|
|
|
raidPtr = node->dagHdr->raidPtr;
|
|
|
|
switch (node->status) {
|
|
case rf_good :
|
|
/* normal case, don't need to do anything */
|
|
break;
|
|
case rf_bad :
|
|
if ((node->dagHdr->numCommits > 0) || (node->dagHdr->numCommitNodes == 0)) {
|
|
node->dagHdr->status = rf_rollForward; /* crossed commit barrier */
|
|
if (rf_engineDebug || 1) {
|
|
rf_get_threadid(tid);
|
|
printf("[%d] node (%s) returned fail, rolling forward\n", tid, node->name);
|
|
}
|
|
}
|
|
else {
|
|
node->dagHdr->status = rf_rollBackward; /* never reached commit barrier */
|
|
if (rf_engineDebug || 1) {
|
|
rf_get_threadid(tid);
|
|
printf("[%d] node (%s) returned fail, rolling backward\n", tid, node->name);
|
|
}
|
|
}
|
|
break;
|
|
case rf_undone :
|
|
/* normal rollBackward case, don't need to do anything */
|
|
break;
|
|
case rf_panic :
|
|
/* an undo node failed!!! */
|
|
printf("UNDO of a node failed!!!/n");
|
|
break;
|
|
default :
|
|
printf("node finished execution with an illegal status!!!\n");
|
|
RF_PANIC();
|
|
break;
|
|
}
|
|
|
|
#ifdef SIMULATE
|
|
/* simulator fires nodes here.
|
|
* user/kernel rely upon PropagateResults to do this.
|
|
* XXX seems like this code should be merged so that the same thing happens for
|
|
* both sim, user, and kernel. -wvcii
|
|
*/
|
|
switch (node->dagHdr->status) {
|
|
case rf_enable :
|
|
case rf_rollForward :
|
|
if (node->numSuccedents == 0) {
|
|
/* process terminal node */
|
|
if (rf_engineDebug) if (!DAGDone(node->dagHdr)) {
|
|
rf_get_threadid(tid);
|
|
printf("[%d] ProcessNode: !!!done but dag still in flight\n",tid);
|
|
RF_PANIC();
|
|
}
|
|
if (rf_engineDebug) printf("[%d] ProcessNode: !!!done will return true\n",tid);
|
|
/* Mark dag as done */
|
|
(node->dagHdr)->done=RF_TRUE;
|
|
raidPtr->dags_in_flight--;
|
|
}
|
|
else {
|
|
PropagateResults(node, context);
|
|
FireNodeArray(node->numSuccedents, node->succedents);
|
|
}
|
|
break;
|
|
case rf_rollBackward :
|
|
if (node->numAntecedents == 0) {
|
|
/* reached head of dag, we're done */
|
|
if (rf_engineDebug) if (!DAGDone(node->dagHdr)) {
|
|
rf_get_threadid(tid);
|
|
printf("[%d] ProcessNode: !!!done but dag still in flight\n",tid);
|
|
RF_PANIC();
|
|
}
|
|
if (rf_engineDebug) printf("[%d] ProcessNode: !!!done will return true\n",tid);
|
|
/* Mark dag as done */
|
|
(node->dagHdr)->done=RF_TRUE;
|
|
raidPtr->dags_in_flight--;
|
|
}
|
|
else {
|
|
PropagateResults(node, context);
|
|
FireNodeArray(node->numAntecedents, node->antecedents);
|
|
}
|
|
break;
|
|
default :
|
|
RF_PANIC();
|
|
break;
|
|
}
|
|
|
|
|
|
#else /* SIMULATE */
|
|
/* enqueue node's succedents (antecedents if rollBackward) for execution */
|
|
PropagateResults(node, context);
|
|
#endif /* SIMULATE */
|
|
}
|
|
|
|
|
|
|
|
/* user context or dag-exec-thread context:
|
|
* This is the first step in post-processing a newly-completed node.
|
|
* This routine is called by each node execution function to mark the node
|
|
* as complete and fire off any successors that have been enabled.
|
|
*/
|
|
int rf_FinishNode(
|
|
RF_DagNode_t *node,
|
|
int context)
|
|
{
|
|
/* as far as I can tell, retcode is not used -wvcii */
|
|
int retcode = RF_FALSE;
|
|
node->dagHdr->numNodesCompleted++;
|
|
ProcessNode(node, context);
|
|
|
|
#ifdef SIMULATE
|
|
if ((node->dagHdr)->done == RF_TRUE)
|
|
retcode = RF_TRUE;
|
|
#endif /* SIMULATE */
|
|
|
|
return(retcode);
|
|
}
|
|
|
|
|
|
/* user context:
|
|
* submit dag for execution, return non-zero if we have to wait for completion.
|
|
* if and only if we return non-zero, we'll cause cbFunc to get invoked with
|
|
* cbArg when the DAG has completed.
|
|
*
|
|
* for now we always return 1. If the DAG does not cause any I/O, then the callback
|
|
* may get invoked before DispatchDAG returns. There's code in state 5 of ContinueRaidAccess
|
|
* to handle this.
|
|
*
|
|
* All we do here is fire the direct successors of the header node. The
|
|
* DAG execution thread does the rest of the dag processing.
|
|
*/
|
|
int rf_DispatchDAG(
|
|
RF_DagHeader_t *dag,
|
|
void (*cbFunc)(void *),
|
|
void *cbArg)
|
|
{
|
|
RF_Raid_t *raidPtr;
|
|
int tid;
|
|
|
|
raidPtr = dag->raidPtr;
|
|
if (dag->tracerec) {
|
|
RF_ETIMER_START(dag->tracerec->timer);
|
|
}
|
|
|
|
if (rf_engineDebug || rf_validateDAGDebug) {
|
|
if (rf_ValidateDAG(dag))
|
|
RF_PANIC();
|
|
}
|
|
if (rf_engineDebug) {
|
|
rf_get_threadid(tid);
|
|
printf("[%d] Entering DispatchDAG\n",tid);
|
|
}
|
|
|
|
raidPtr->dags_in_flight++; /* debug only: blow off proper locking */
|
|
dag->cbFunc = cbFunc;
|
|
dag->cbArg = cbArg;
|
|
dag->numNodesCompleted = 0;
|
|
dag->status = rf_enable;
|
|
FireNodeArray(dag->numSuccedents, dag->succedents);
|
|
return(1);
|
|
}
|
|
|
|
/* dedicated kernel thread:
|
|
* the thread that handles all DAG node firing.
|
|
* To minimize locking and unlocking, we grab a copy of the entire node queue and then set the
|
|
* node queue to NULL before doing any firing of nodes. This way we only have to release the
|
|
* lock once. Of course, it's probably rare that there's more than one node in the queue at
|
|
* any one time, but it sometimes happens.
|
|
*
|
|
* In the kernel, this thread runs at spl0 and is not swappable. I copied these
|
|
* characteristics from the aio_completion_thread.
|
|
*/
|
|
|
|
#ifndef SIMULATE
|
|
static void DAGExecutionThread(RF_ThreadArg_t arg)
|
|
{
|
|
RF_DagNode_t *nd, *local_nq, *term_nq, *fire_nq;
|
|
RF_Raid_t *raidPtr;
|
|
int ks, tid;
|
|
int s;
|
|
#ifndef __NetBSD__
|
|
RF_Thread_t thread;
|
|
#endif
|
|
|
|
raidPtr = (RF_Raid_t *)arg;
|
|
|
|
rf_assign_threadid();
|
|
if (rf_engineDebug) {
|
|
rf_get_threadid(tid);
|
|
printf("[%d] Engine thread is running\n", tid);
|
|
}
|
|
|
|
#ifdef KERNEL
|
|
#ifndef __NetBSD__
|
|
thread = current_thread();
|
|
thread_swappable(thread, RF_FALSE);
|
|
thread->priority = thread->sched_pri = BASEPRI_SYSTEM;
|
|
s = spl0();
|
|
#endif
|
|
/* XXX what to put here XXX */
|
|
|
|
s=splbio();
|
|
|
|
#endif /* KERNEL */
|
|
|
|
RF_THREADGROUP_RUNNING(&raidPtr->engine_tg);
|
|
|
|
DO_LOCK(raidPtr);
|
|
while (!raidPtr->shutdown_engine) {
|
|
|
|
while (raidPtr->node_queue != NULL) {
|
|
local_nq = raidPtr->node_queue;
|
|
fire_nq = NULL;
|
|
term_nq = NULL;
|
|
raidPtr->node_queue = NULL;
|
|
DO_UNLOCK(raidPtr);
|
|
|
|
/* first, strip out the terminal nodes */
|
|
while (local_nq) {
|
|
nd = local_nq;
|
|
local_nq = local_nq->next;
|
|
switch(nd->dagHdr->status) {
|
|
case rf_enable :
|
|
case rf_rollForward :
|
|
if (nd->numSuccedents == 0) {
|
|
/* end of the dag, add to callback list */
|
|
nd->next = term_nq;
|
|
term_nq = nd;
|
|
}
|
|
else {
|
|
/* not the end, add to the fire queue */
|
|
nd->next = fire_nq;
|
|
fire_nq = nd;
|
|
}
|
|
break;
|
|
case rf_rollBackward :
|
|
if (nd->numAntecedents == 0) {
|
|
/* end of the dag, add to the callback list */
|
|
nd->next = term_nq;
|
|
term_nq = nd;
|
|
}
|
|
else {
|
|
/* not the end, add to the fire queue */
|
|
nd->next = fire_nq;
|
|
fire_nq = nd;
|
|
}
|
|
break;
|
|
default :
|
|
RF_PANIC();
|
|
break;
|
|
}
|
|
}
|
|
|
|
/* execute callback of dags which have reached the terminal node */
|
|
while (term_nq) {
|
|
nd = term_nq;
|
|
term_nq = term_nq->next;
|
|
nd->next = NULL;
|
|
(nd->dagHdr->cbFunc)(nd->dagHdr->cbArg);
|
|
raidPtr->dags_in_flight--; /* debug only */
|
|
}
|
|
|
|
/* fire remaining nodes */
|
|
FireNodeList(fire_nq);
|
|
|
|
DO_LOCK(raidPtr);
|
|
}
|
|
while (!raidPtr->shutdown_engine && raidPtr->node_queue == NULL)
|
|
DO_WAIT(raidPtr);
|
|
}
|
|
DO_UNLOCK(raidPtr);
|
|
|
|
RF_THREADGROUP_DONE(&raidPtr->engine_tg);
|
|
#ifdef KERNEL
|
|
#ifdef __NetBSD__
|
|
splx(s);
|
|
kthread_exit(0);
|
|
#else
|
|
splx(s);
|
|
thread_terminate(thread);
|
|
thread_halt_self();
|
|
#endif
|
|
#endif /* KERNEL */
|
|
}
|
|
|
|
#endif /* !SIMULATE */
|