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