rf_engine.c revision 1.17
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 (at) 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
Indexes created Fri Sep 26 20:09:58 GMT 2025