rf_dagffwr.c revision 1.11 1 /* $NetBSD: rf_dagffwr.c,v 1.11 2003/07/01 22:43:59 oster Exp $ */
2 /*
3 * Copyright (c) 1995 Carnegie-Mellon University.
4 * All rights reserved.
5 *
6 * Author: Mark Holland, Daniel Stodolsky, William V. Courtright II
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 * rf_dagff.c
31 *
32 * code for creating fault-free DAGs
33 *
34 */
35
36 #include <sys/cdefs.h>
37 __KERNEL_RCSID(0, "$NetBSD: rf_dagffwr.c,v 1.11 2003/07/01 22:43:59 oster Exp $");
38
39 #include <dev/raidframe/raidframevar.h>
40
41 #include "rf_raid.h"
42 #include "rf_dag.h"
43 #include "rf_dagutils.h"
44 #include "rf_dagfuncs.h"
45 #include "rf_debugMem.h"
46 #include "rf_dagffrd.h"
47 #include "rf_general.h"
48 #include "rf_dagffwr.h"
49
50 /******************************************************************************
51 *
52 * General comments on DAG creation:
53 *
54 * All DAGs in this file use roll-away error recovery. Each DAG has a single
55 * commit node, usually called "Cmt." If an error occurs before the Cmt node
56 * is reached, the execution engine will halt forward execution and work
57 * backward through the graph, executing the undo functions. Assuming that
58 * each node in the graph prior to the Cmt node are undoable and atomic - or -
59 * does not make changes to permanent state, the graph will fail atomically.
60 * If an error occurs after the Cmt node executes, the engine will roll-forward
61 * through the graph, blindly executing nodes until it reaches the end.
62 * If a graph reaches the end, it is assumed to have completed successfully.
63 *
64 * A graph has only 1 Cmt node.
65 *
66 */
67
68
69 /******************************************************************************
70 *
71 * The following wrappers map the standard DAG creation interface to the
72 * DAG creation routines. Additionally, these wrappers enable experimentation
73 * with new DAG structures by providing an extra level of indirection, allowing
74 * the DAG creation routines to be replaced at this single point.
75 */
76
77
78 void
79 rf_CreateNonRedundantWriteDAG(
80 RF_Raid_t * raidPtr,
81 RF_AccessStripeMap_t * asmap,
82 RF_DagHeader_t * dag_h,
83 void *bp,
84 RF_RaidAccessFlags_t flags,
85 RF_AllocListElem_t * allocList,
86 RF_IoType_t type)
87 {
88 rf_CreateNonredundantDAG(raidPtr, asmap, dag_h, bp, flags, allocList,
89 RF_IO_TYPE_WRITE);
90 }
91
92 void
93 rf_CreateRAID0WriteDAG(
94 RF_Raid_t * raidPtr,
95 RF_AccessStripeMap_t * asmap,
96 RF_DagHeader_t * dag_h,
97 void *bp,
98 RF_RaidAccessFlags_t flags,
99 RF_AllocListElem_t * allocList,
100 RF_IoType_t type)
101 {
102 rf_CreateNonredundantDAG(raidPtr, asmap, dag_h, bp, flags, allocList,
103 RF_IO_TYPE_WRITE);
104 }
105
106 void
107 rf_CreateSmallWriteDAG(
108 RF_Raid_t * raidPtr,
109 RF_AccessStripeMap_t * asmap,
110 RF_DagHeader_t * dag_h,
111 void *bp,
112 RF_RaidAccessFlags_t flags,
113 RF_AllocListElem_t * allocList)
114 {
115 /* "normal" rollaway */
116 rf_CommonCreateSmallWriteDAG(raidPtr, asmap, dag_h, bp, flags, allocList,
117 &rf_xorFuncs, NULL);
118 }
119
120 void
121 rf_CreateLargeWriteDAG(
122 RF_Raid_t * raidPtr,
123 RF_AccessStripeMap_t * asmap,
124 RF_DagHeader_t * dag_h,
125 void *bp,
126 RF_RaidAccessFlags_t flags,
127 RF_AllocListElem_t * allocList)
128 {
129 /* "normal" rollaway */
130 rf_CommonCreateLargeWriteDAG(raidPtr, asmap, dag_h, bp, flags, allocList,
131 1, rf_RegularXorFunc, RF_TRUE);
132 }
133
134
135 /******************************************************************************
136 *
137 * DAG creation code begins here
138 */
139
140
141 /******************************************************************************
142 *
143 * creates a DAG to perform a large-write operation:
144 *
145 * / Rod \ / Wnd \
146 * H -- block- Rod - Xor - Cmt - Wnd --- T
147 * \ Rod / \ Wnp /
148 * \[Wnq]/
149 *
150 * The XOR node also does the Q calculation in the P+Q architecture.
151 * All nodes are before the commit node (Cmt) are assumed to be atomic and
152 * undoable - or - they make no changes to permanent state.
153 *
154 * Rod = read old data
155 * Cmt = commit node
156 * Wnp = write new parity
157 * Wnd = write new data
158 * Wnq = write new "q"
159 * [] denotes optional segments in the graph
160 *
161 * Parameters: raidPtr - description of the physical array
162 * asmap - logical & physical addresses for this access
163 * bp - buffer ptr (holds write data)
164 * flags - general flags (e.g. disk locking)
165 * allocList - list of memory allocated in DAG creation
166 * nfaults - number of faults array can tolerate
167 * (equal to # redundancy units in stripe)
168 * redfuncs - list of redundancy generating functions
169 *
170 *****************************************************************************/
171
172 void
173 rf_CommonCreateLargeWriteDAG(
174 RF_Raid_t * raidPtr,
175 RF_AccessStripeMap_t * asmap,
176 RF_DagHeader_t * dag_h,
177 void *bp,
178 RF_RaidAccessFlags_t flags,
179 RF_AllocListElem_t * allocList,
180 int nfaults,
181 int (*redFunc) (RF_DagNode_t *),
182 int allowBufferRecycle)
183 {
184 RF_DagNode_t *nodes, *wndNodes, *rodNodes, *xorNode, *wnpNode;
185 RF_DagNode_t *wnqNode, *blockNode, *commitNode, *termNode;
186 int nWndNodes, nRodNodes, i, nodeNum, asmNum;
187 RF_AccessStripeMapHeader_t *new_asm_h[2];
188 RF_StripeNum_t parityStripeID;
189 char *sosBuffer, *eosBuffer;
190 RF_ReconUnitNum_t which_ru;
191 RF_RaidLayout_t *layoutPtr;
192 RF_PhysDiskAddr_t *pda;
193
194 layoutPtr = &(raidPtr->Layout);
195 parityStripeID = rf_RaidAddressToParityStripeID(layoutPtr, asmap->raidAddress,
196 &which_ru);
197
198 if (rf_dagDebug) {
199 printf("[Creating large-write DAG]\n");
200 }
201 dag_h->creator = "LargeWriteDAG";
202
203 dag_h->numCommitNodes = 1;
204 dag_h->numCommits = 0;
205 dag_h->numSuccedents = 1;
206
207 /* alloc the nodes: Wnd, xor, commit, block, term, and Wnp */
208 nWndNodes = asmap->numStripeUnitsAccessed;
209 RF_CallocAndAdd(nodes, nWndNodes + 4 + nfaults, sizeof(RF_DagNode_t),
210 (RF_DagNode_t *), allocList);
211 i = 0;
212 wndNodes = &nodes[i];
213 i += nWndNodes;
214 xorNode = &nodes[i];
215 i += 1;
216 wnpNode = &nodes[i];
217 i += 1;
218 blockNode = &nodes[i];
219 i += 1;
220 commitNode = &nodes[i];
221 i += 1;
222 termNode = &nodes[i];
223 i += 1;
224 if (nfaults == 2) {
225 wnqNode = &nodes[i];
226 i += 1;
227 } else {
228 wnqNode = NULL;
229 }
230 rf_MapUnaccessedPortionOfStripe(raidPtr, layoutPtr, asmap, dag_h, new_asm_h,
231 &nRodNodes, &sosBuffer, &eosBuffer, allocList);
232 if (nRodNodes > 0) {
233 RF_CallocAndAdd(rodNodes, nRodNodes, sizeof(RF_DagNode_t),
234 (RF_DagNode_t *), allocList);
235 } else {
236 rodNodes = NULL;
237 }
238
239 /* begin node initialization */
240 if (nRodNodes > 0) {
241 rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
242 NULL, nRodNodes, 0, 0, 0, dag_h, "Nil", allocList);
243 } else {
244 rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
245 NULL, 1, 0, 0, 0, dag_h, "Nil", allocList);
246 }
247
248 rf_InitNode(commitNode, rf_wait, RF_TRUE, rf_NullNodeFunc, rf_NullNodeUndoFunc, NULL,
249 nWndNodes + nfaults, 1, 0, 0, dag_h, "Cmt", allocList);
250 rf_InitNode(termNode, rf_wait, RF_FALSE, rf_TerminateFunc, rf_TerminateUndoFunc, NULL,
251 0, nWndNodes + nfaults, 0, 0, dag_h, "Trm", allocList);
252
253 /* initialize the Rod nodes */
254 for (nodeNum = asmNum = 0; asmNum < 2; asmNum++) {
255 if (new_asm_h[asmNum]) {
256 pda = new_asm_h[asmNum]->stripeMap->physInfo;
257 while (pda) {
258 rf_InitNode(&rodNodes[nodeNum], rf_wait, RF_FALSE, rf_DiskReadFunc,
259 rf_DiskReadUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h,
260 "Rod", allocList);
261 rodNodes[nodeNum].params[0].p = pda;
262 rodNodes[nodeNum].params[1].p = pda->bufPtr;
263 rodNodes[nodeNum].params[2].v = parityStripeID;
264 rodNodes[nodeNum].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
265 0, 0, which_ru);
266 nodeNum++;
267 pda = pda->next;
268 }
269 }
270 }
271 RF_ASSERT(nodeNum == nRodNodes);
272
273 /* initialize the wnd nodes */
274 pda = asmap->physInfo;
275 for (i = 0; i < nWndNodes; i++) {
276 rf_InitNode(&wndNodes[i], rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
277 rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wnd", allocList);
278 RF_ASSERT(pda != NULL);
279 wndNodes[i].params[0].p = pda;
280 wndNodes[i].params[1].p = pda->bufPtr;
281 wndNodes[i].params[2].v = parityStripeID;
282 wndNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
283 pda = pda->next;
284 }
285
286 /* initialize the redundancy node */
287 if (nRodNodes > 0) {
288 rf_InitNode(xorNode, rf_wait, RF_FALSE, redFunc, rf_NullNodeUndoFunc, NULL, 1,
289 nRodNodes, 2 * (nWndNodes + nRodNodes) + 1, nfaults, dag_h,
290 "Xr ", allocList);
291 } else {
292 rf_InitNode(xorNode, rf_wait, RF_FALSE, redFunc, rf_NullNodeUndoFunc, NULL, 1,
293 1, 2 * (nWndNodes + nRodNodes) + 1, nfaults, dag_h, "Xr ", allocList);
294 }
295 xorNode->flags |= RF_DAGNODE_FLAG_YIELD;
296 for (i = 0; i < nWndNodes; i++) {
297 xorNode->params[2 * i + 0] = wndNodes[i].params[0]; /* pda */
298 xorNode->params[2 * i + 1] = wndNodes[i].params[1]; /* buf ptr */
299 }
300 for (i = 0; i < nRodNodes; i++) {
301 xorNode->params[2 * (nWndNodes + i) + 0] = rodNodes[i].params[0]; /* pda */
302 xorNode->params[2 * (nWndNodes + i) + 1] = rodNodes[i].params[1]; /* buf ptr */
303 }
304 /* xor node needs to get at RAID information */
305 xorNode->params[2 * (nWndNodes + nRodNodes)].p = raidPtr;
306
307 /*
308 * Look for an Rod node that reads a complete SU. If none, alloc a buffer
309 * to receive the parity info. Note that we can't use a new data buffer
310 * because it will not have gotten written when the xor occurs.
311 */
312 if (allowBufferRecycle) {
313 for (i = 0; i < nRodNodes; i++) {
314 if (((RF_PhysDiskAddr_t *) rodNodes[i].params[0].p)->numSector == raidPtr->Layout.sectorsPerStripeUnit)
315 break;
316 }
317 }
318 if ((!allowBufferRecycle) || (i == nRodNodes)) {
319 RF_CallocAndAdd(xorNode->results[0], 1,
320 rf_RaidAddressToByte(raidPtr, raidPtr->Layout.sectorsPerStripeUnit),
321 (void *), allocList);
322 } else {
323 xorNode->results[0] = rodNodes[i].params[1].p;
324 }
325
326 /* initialize the Wnp node */
327 rf_InitNode(wnpNode, rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
328 rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wnp", allocList);
329 wnpNode->params[0].p = asmap->parityInfo;
330 wnpNode->params[1].p = xorNode->results[0];
331 wnpNode->params[2].v = parityStripeID;
332 wnpNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
333 /* parityInfo must describe entire parity unit */
334 RF_ASSERT(asmap->parityInfo->next == NULL);
335
336 if (nfaults == 2) {
337 /*
338 * We never try to recycle a buffer for the Q calcuation
339 * in addition to the parity. This would cause two buffers
340 * to get smashed during the P and Q calculation, guaranteeing
341 * one would be wrong.
342 */
343 RF_CallocAndAdd(xorNode->results[1], 1,
344 rf_RaidAddressToByte(raidPtr, raidPtr->Layout.sectorsPerStripeUnit),
345 (void *), allocList);
346 rf_InitNode(wnqNode, rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
347 rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wnq", allocList);
348 wnqNode->params[0].p = asmap->qInfo;
349 wnqNode->params[1].p = xorNode->results[1];
350 wnqNode->params[2].v = parityStripeID;
351 wnqNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
352 /* parityInfo must describe entire parity unit */
353 RF_ASSERT(asmap->parityInfo->next == NULL);
354 }
355 /*
356 * Connect nodes to form graph.
357 */
358
359 /* connect dag header to block node */
360 RF_ASSERT(blockNode->numAntecedents == 0);
361 dag_h->succedents[0] = blockNode;
362
363 if (nRodNodes > 0) {
364 /* connect the block node to the Rod nodes */
365 RF_ASSERT(blockNode->numSuccedents == nRodNodes);
366 RF_ASSERT(xorNode->numAntecedents == nRodNodes);
367 for (i = 0; i < nRodNodes; i++) {
368 RF_ASSERT(rodNodes[i].numAntecedents == 1);
369 blockNode->succedents[i] = &rodNodes[i];
370 rodNodes[i].antecedents[0] = blockNode;
371 rodNodes[i].antType[0] = rf_control;
372
373 /* connect the Rod nodes to the Xor node */
374 RF_ASSERT(rodNodes[i].numSuccedents == 1);
375 rodNodes[i].succedents[0] = xorNode;
376 xorNode->antecedents[i] = &rodNodes[i];
377 xorNode->antType[i] = rf_trueData;
378 }
379 } else {
380 /* connect the block node to the Xor node */
381 RF_ASSERT(blockNode->numSuccedents == 1);
382 RF_ASSERT(xorNode->numAntecedents == 1);
383 blockNode->succedents[0] = xorNode;
384 xorNode->antecedents[0] = blockNode;
385 xorNode->antType[0] = rf_control;
386 }
387
388 /* connect the xor node to the commit node */
389 RF_ASSERT(xorNode->numSuccedents == 1);
390 RF_ASSERT(commitNode->numAntecedents == 1);
391 xorNode->succedents[0] = commitNode;
392 commitNode->antecedents[0] = xorNode;
393 commitNode->antType[0] = rf_control;
394
395 /* connect the commit node to the write nodes */
396 RF_ASSERT(commitNode->numSuccedents == nWndNodes + nfaults);
397 for (i = 0; i < nWndNodes; i++) {
398 RF_ASSERT(wndNodes->numAntecedents == 1);
399 commitNode->succedents[i] = &wndNodes[i];
400 wndNodes[i].antecedents[0] = commitNode;
401 wndNodes[i].antType[0] = rf_control;
402 }
403 RF_ASSERT(wnpNode->numAntecedents == 1);
404 commitNode->succedents[nWndNodes] = wnpNode;
405 wnpNode->antecedents[0] = commitNode;
406 wnpNode->antType[0] = rf_trueData;
407 if (nfaults == 2) {
408 RF_ASSERT(wnqNode->numAntecedents == 1);
409 commitNode->succedents[nWndNodes + 1] = wnqNode;
410 wnqNode->antecedents[0] = commitNode;
411 wnqNode->antType[0] = rf_trueData;
412 }
413 /* connect the write nodes to the term node */
414 RF_ASSERT(termNode->numAntecedents == nWndNodes + nfaults);
415 RF_ASSERT(termNode->numSuccedents == 0);
416 for (i = 0; i < nWndNodes; i++) {
417 RF_ASSERT(wndNodes->numSuccedents == 1);
418 wndNodes[i].succedents[0] = termNode;
419 termNode->antecedents[i] = &wndNodes[i];
420 termNode->antType[i] = rf_control;
421 }
422 RF_ASSERT(wnpNode->numSuccedents == 1);
423 wnpNode->succedents[0] = termNode;
424 termNode->antecedents[nWndNodes] = wnpNode;
425 termNode->antType[nWndNodes] = rf_control;
426 if (nfaults == 2) {
427 RF_ASSERT(wnqNode->numSuccedents == 1);
428 wnqNode->succedents[0] = termNode;
429 termNode->antecedents[nWndNodes + 1] = wnqNode;
430 termNode->antType[nWndNodes + 1] = rf_control;
431 }
432 }
433 /******************************************************************************
434 *
435 * creates a DAG to perform a small-write operation (either raid 5 or pq),
436 * which is as follows:
437 *
438 * Hdr -> Nil -> Rop -> Xor -> Cmt ----> Wnp [Unp] --> Trm
439 * \- Rod X / \----> Wnd [Und]-/
440 * [\- Rod X / \---> Wnd [Und]-/]
441 * [\- Roq -> Q / \--> Wnq [Unq]-/]
442 *
443 * Rop = read old parity
444 * Rod = read old data
445 * Roq = read old "q"
446 * Cmt = commit node
447 * Und = unlock data disk
448 * Unp = unlock parity disk
449 * Unq = unlock q disk
450 * Wnp = write new parity
451 * Wnd = write new data
452 * Wnq = write new "q"
453 * [ ] denotes optional segments in the graph
454 *
455 * Parameters: raidPtr - description of the physical array
456 * asmap - logical & physical addresses for this access
457 * bp - buffer ptr (holds write data)
458 * flags - general flags (e.g. disk locking)
459 * allocList - list of memory allocated in DAG creation
460 * pfuncs - list of parity generating functions
461 * qfuncs - list of q generating functions
462 *
463 * A null qfuncs indicates single fault tolerant
464 *****************************************************************************/
465
466 void
467 rf_CommonCreateSmallWriteDAG(
468 RF_Raid_t * raidPtr,
469 RF_AccessStripeMap_t * asmap,
470 RF_DagHeader_t * dag_h,
471 void *bp,
472 RF_RaidAccessFlags_t flags,
473 RF_AllocListElem_t * allocList,
474 const RF_RedFuncs_t * pfuncs,
475 const RF_RedFuncs_t * qfuncs)
476 {
477 RF_DagNode_t *readDataNodes, *readParityNodes, *readQNodes, *termNode;
478 RF_DagNode_t *unlockDataNodes, *unlockParityNodes, *unlockQNodes;
479 RF_DagNode_t *xorNodes, *qNodes, *blockNode, *commitNode, *nodes;
480 RF_DagNode_t *writeDataNodes, *writeParityNodes, *writeQNodes;
481 int i, j, nNodes, totalNumNodes, lu_flag;
482 RF_ReconUnitNum_t which_ru;
483 int (*func) (RF_DagNode_t *), (*undoFunc) (RF_DagNode_t *);
484 int (*qfunc) (RF_DagNode_t *);
485 int numDataNodes, numParityNodes;
486 RF_StripeNum_t parityStripeID;
487 RF_PhysDiskAddr_t *pda;
488 char *name, *qname;
489 long nfaults;
490
491 nfaults = qfuncs ? 2 : 1;
492 lu_flag = (rf_enableAtomicRMW) ? 1 : 0; /* lock/unlock flag */
493
494 parityStripeID = rf_RaidAddressToParityStripeID(&(raidPtr->Layout),
495 asmap->raidAddress, &which_ru);
496 pda = asmap->physInfo;
497 numDataNodes = asmap->numStripeUnitsAccessed;
498 numParityNodes = (asmap->parityInfo->next) ? 2 : 1;
499
500 if (rf_dagDebug) {
501 printf("[Creating small-write DAG]\n");
502 }
503 RF_ASSERT(numDataNodes > 0);
504 dag_h->creator = "SmallWriteDAG";
505
506 dag_h->numCommitNodes = 1;
507 dag_h->numCommits = 0;
508 dag_h->numSuccedents = 1;
509
510 /*
511 * DAG creation occurs in four steps:
512 * 1. count the number of nodes in the DAG
513 * 2. create the nodes
514 * 3. initialize the nodes
515 * 4. connect the nodes
516 */
517
518 /*
519 * Step 1. compute number of nodes in the graph
520 */
521
522 /* number of nodes: a read and write for each data unit a redundancy
523 * computation node for each parity node (nfaults * nparity) a read
524 * and write for each parity unit a block and commit node (2) a
525 * terminate node if atomic RMW an unlock node for each data unit,
526 * redundancy unit */
527 totalNumNodes = (2 * numDataNodes) + (nfaults * numParityNodes)
528 + (nfaults * 2 * numParityNodes) + 3;
529 if (lu_flag) {
530 totalNumNodes += (numDataNodes + (nfaults * numParityNodes));
531 }
532 /*
533 * Step 2. create the nodes
534 */
535 RF_CallocAndAdd(nodes, totalNumNodes, sizeof(RF_DagNode_t),
536 (RF_DagNode_t *), allocList);
537 i = 0;
538 blockNode = &nodes[i];
539 i += 1;
540 commitNode = &nodes[i];
541 i += 1;
542 readDataNodes = &nodes[i];
543 i += numDataNodes;
544 readParityNodes = &nodes[i];
545 i += numParityNodes;
546 writeDataNodes = &nodes[i];
547 i += numDataNodes;
548 writeParityNodes = &nodes[i];
549 i += numParityNodes;
550 xorNodes = &nodes[i];
551 i += numParityNodes;
552 termNode = &nodes[i];
553 i += 1;
554 if (lu_flag) {
555 unlockDataNodes = &nodes[i];
556 i += numDataNodes;
557 unlockParityNodes = &nodes[i];
558 i += numParityNodes;
559 } else {
560 unlockDataNodes = unlockParityNodes = NULL;
561 }
562 if (nfaults == 2) {
563 readQNodes = &nodes[i];
564 i += numParityNodes;
565 writeQNodes = &nodes[i];
566 i += numParityNodes;
567 qNodes = &nodes[i];
568 i += numParityNodes;
569 if (lu_flag) {
570 unlockQNodes = &nodes[i];
571 i += numParityNodes;
572 } else {
573 unlockQNodes = NULL;
574 }
575 } else {
576 readQNodes = writeQNodes = qNodes = unlockQNodes = NULL;
577 }
578 RF_ASSERT(i == totalNumNodes);
579
580 /*
581 * Step 3. initialize the nodes
582 */
583 /* initialize block node (Nil) */
584 nNodes = numDataNodes + (nfaults * numParityNodes);
585 rf_InitNode(blockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
586 NULL, nNodes, 0, 0, 0, dag_h, "Nil", allocList);
587
588 /* initialize commit node (Cmt) */
589 rf_InitNode(commitNode, rf_wait, RF_TRUE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
590 NULL, nNodes, (nfaults * numParityNodes), 0, 0, dag_h, "Cmt", allocList);
591
592 /* initialize terminate node (Trm) */
593 rf_InitNode(termNode, rf_wait, RF_FALSE, rf_TerminateFunc, rf_TerminateUndoFunc,
594 NULL, 0, nNodes, 0, 0, dag_h, "Trm", allocList);
595
596 /* initialize nodes which read old data (Rod) */
597 for (i = 0; i < numDataNodes; i++) {
598 rf_InitNode(&readDataNodes[i], rf_wait, RF_FALSE, rf_DiskReadFunc, rf_DiskReadUndoFunc,
599 rf_GenericWakeupFunc, (nfaults * numParityNodes), 1, 4, 0, dag_h,
600 "Rod", allocList);
601 RF_ASSERT(pda != NULL);
602 /* physical disk addr desc */
603 readDataNodes[i].params[0].p = pda;
604 /* buffer to hold old data */
605 readDataNodes[i].params[1].p = rf_AllocBuffer(raidPtr,
606 dag_h, pda, allocList);
607 readDataNodes[i].params[2].v = parityStripeID;
608 readDataNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
609 lu_flag, 0, which_ru);
610 pda = pda->next;
611 for (j = 0; j < readDataNodes[i].numSuccedents; j++) {
612 readDataNodes[i].propList[j] = NULL;
613 }
614 }
615
616 /* initialize nodes which read old parity (Rop) */
617 pda = asmap->parityInfo;
618 i = 0;
619 for (i = 0; i < numParityNodes; i++) {
620 RF_ASSERT(pda != NULL);
621 rf_InitNode(&readParityNodes[i], rf_wait, RF_FALSE, rf_DiskReadFunc,
622 rf_DiskReadUndoFunc, rf_GenericWakeupFunc, numParityNodes, 1, 4,
623 0, dag_h, "Rop", allocList);
624 readParityNodes[i].params[0].p = pda;
625 /* buffer to hold old parity */
626 readParityNodes[i].params[1].p = rf_AllocBuffer(raidPtr,
627 dag_h, pda, allocList);
628 readParityNodes[i].params[2].v = parityStripeID;
629 readParityNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
630 lu_flag, 0, which_ru);
631 pda = pda->next;
632 for (j = 0; j < readParityNodes[i].numSuccedents; j++) {
633 readParityNodes[i].propList[0] = NULL;
634 }
635 }
636
637 /* initialize nodes which read old Q (Roq) */
638 if (nfaults == 2) {
639 pda = asmap->qInfo;
640 for (i = 0; i < numParityNodes; i++) {
641 RF_ASSERT(pda != NULL);
642 rf_InitNode(&readQNodes[i], rf_wait, RF_FALSE, rf_DiskReadFunc, rf_DiskReadUndoFunc,
643 rf_GenericWakeupFunc, numParityNodes, 1, 4, 0, dag_h, "Roq", allocList);
644 readQNodes[i].params[0].p = pda;
645 /* buffer to hold old Q */
646 readQNodes[i].params[1].p = rf_AllocBuffer(raidPtr, dag_h, pda,
647 allocList);
648 readQNodes[i].params[2].v = parityStripeID;
649 readQNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
650 lu_flag, 0, which_ru);
651 pda = pda->next;
652 for (j = 0; j < readQNodes[i].numSuccedents; j++) {
653 readQNodes[i].propList[0] = NULL;
654 }
655 }
656 }
657 /* initialize nodes which write new data (Wnd) */
658 pda = asmap->physInfo;
659 for (i = 0; i < numDataNodes; i++) {
660 RF_ASSERT(pda != NULL);
661 rf_InitNode(&writeDataNodes[i], rf_wait, RF_FALSE, rf_DiskWriteFunc,
662 rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h,
663 "Wnd", allocList);
664 /* physical disk addr desc */
665 writeDataNodes[i].params[0].p = pda;
666 /* buffer holding new data to be written */
667 writeDataNodes[i].params[1].p = pda->bufPtr;
668 writeDataNodes[i].params[2].v = parityStripeID;
669 writeDataNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
670 0, 0, which_ru);
671 if (lu_flag) {
672 /* initialize node to unlock the disk queue */
673 rf_InitNode(&unlockDataNodes[i], rf_wait, RF_FALSE, rf_DiskUnlockFunc,
674 rf_DiskUnlockUndoFunc, rf_GenericWakeupFunc, 1, 1, 2, 0, dag_h,
675 "Und", allocList);
676 /* physical disk addr desc */
677 unlockDataNodes[i].params[0].p = pda;
678 unlockDataNodes[i].params[1].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
679 0, lu_flag, which_ru);
680 }
681 pda = pda->next;
682 }
683
684 /*
685 * Initialize nodes which compute new parity and Q.
686 */
687 /*
688 * We use the simple XOR func in the double-XOR case, and when
689 * we're accessing only a portion of one stripe unit. The distinction
690 * between the two is that the regular XOR func assumes that the targbuf
691 * is a full SU in size, and examines the pda associated with the buffer
692 * to decide where within the buffer to XOR the data, whereas
693 * the simple XOR func just XORs the data into the start of the buffer.
694 */
695 if ((numParityNodes == 2) || ((numDataNodes == 1)
696 && (asmap->totalSectorsAccessed < raidPtr->Layout.sectorsPerStripeUnit))) {
697 func = pfuncs->simple;
698 undoFunc = rf_NullNodeUndoFunc;
699 name = pfuncs->SimpleName;
700 if (qfuncs) {
701 qfunc = qfuncs->simple;
702 qname = qfuncs->SimpleName;
703 } else {
704 qfunc = NULL;
705 qname = NULL;
706 }
707 } else {
708 func = pfuncs->regular;
709 undoFunc = rf_NullNodeUndoFunc;
710 name = pfuncs->RegularName;
711 if (qfuncs) {
712 qfunc = qfuncs->regular;
713 qname = qfuncs->RegularName;
714 } else {
715 qfunc = NULL;
716 qname = NULL;
717 }
718 }
719 /*
720 * Initialize the xor nodes: params are {pda,buf}
721 * from {Rod,Wnd,Rop} nodes, and raidPtr
722 */
723 if (numParityNodes == 2) {
724 /* double-xor case */
725 for (i = 0; i < numParityNodes; i++) {
726 /* note: no wakeup func for xor */
727 rf_InitNode(&xorNodes[i], rf_wait, RF_FALSE, func, undoFunc, NULL,
728 1, (numDataNodes + numParityNodes), 7, 1, dag_h, name, allocList);
729 xorNodes[i].flags |= RF_DAGNODE_FLAG_YIELD;
730 xorNodes[i].params[0] = readDataNodes[i].params[0];
731 xorNodes[i].params[1] = readDataNodes[i].params[1];
732 xorNodes[i].params[2] = readParityNodes[i].params[0];
733 xorNodes[i].params[3] = readParityNodes[i].params[1];
734 xorNodes[i].params[4] = writeDataNodes[i].params[0];
735 xorNodes[i].params[5] = writeDataNodes[i].params[1];
736 xorNodes[i].params[6].p = raidPtr;
737 /* use old parity buf as target buf */
738 xorNodes[i].results[0] = readParityNodes[i].params[1].p;
739 if (nfaults == 2) {
740 /* note: no wakeup func for qor */
741 rf_InitNode(&qNodes[i], rf_wait, RF_FALSE, qfunc, undoFunc, NULL, 1,
742 (numDataNodes + numParityNodes), 7, 1, dag_h, qname, allocList);
743 qNodes[i].params[0] = readDataNodes[i].params[0];
744 qNodes[i].params[1] = readDataNodes[i].params[1];
745 qNodes[i].params[2] = readQNodes[i].params[0];
746 qNodes[i].params[3] = readQNodes[i].params[1];
747 qNodes[i].params[4] = writeDataNodes[i].params[0];
748 qNodes[i].params[5] = writeDataNodes[i].params[1];
749 qNodes[i].params[6].p = raidPtr;
750 /* use old Q buf as target buf */
751 qNodes[i].results[0] = readQNodes[i].params[1].p;
752 }
753 }
754 } else {
755 /* there is only one xor node in this case */
756 rf_InitNode(&xorNodes[0], rf_wait, RF_FALSE, func, undoFunc, NULL, 1,
757 (numDataNodes + numParityNodes),
758 (2 * (numDataNodes + numDataNodes + 1) + 1), 1, dag_h, name, allocList);
759 xorNodes[0].flags |= RF_DAGNODE_FLAG_YIELD;
760 for (i = 0; i < numDataNodes + 1; i++) {
761 /* set up params related to Rod and Rop nodes */
762 xorNodes[0].params[2 * i + 0] = readDataNodes[i].params[0]; /* pda */
763 xorNodes[0].params[2 * i + 1] = readDataNodes[i].params[1]; /* buffer ptr */
764 }
765 for (i = 0; i < numDataNodes; i++) {
766 /* set up params related to Wnd and Wnp nodes */
767 xorNodes[0].params[2 * (numDataNodes + 1 + i) + 0] = /* pda */
768 writeDataNodes[i].params[0];
769 xorNodes[0].params[2 * (numDataNodes + 1 + i) + 1] = /* buffer ptr */
770 writeDataNodes[i].params[1];
771 }
772 /* xor node needs to get at RAID information */
773 xorNodes[0].params[2 * (numDataNodes + numDataNodes + 1)].p = raidPtr;
774 xorNodes[0].results[0] = readParityNodes[0].params[1].p;
775 if (nfaults == 2) {
776 rf_InitNode(&qNodes[0], rf_wait, RF_FALSE, qfunc, undoFunc, NULL, 1,
777 (numDataNodes + numParityNodes),
778 (2 * (numDataNodes + numDataNodes + 1) + 1), 1, dag_h,
779 qname, allocList);
780 for (i = 0; i < numDataNodes; i++) {
781 /* set up params related to Rod */
782 qNodes[0].params[2 * i + 0] = readDataNodes[i].params[0]; /* pda */
783 qNodes[0].params[2 * i + 1] = readDataNodes[i].params[1]; /* buffer ptr */
784 }
785 /* and read old q */
786 qNodes[0].params[2 * numDataNodes + 0] = /* pda */
787 readQNodes[0].params[0];
788 qNodes[0].params[2 * numDataNodes + 1] = /* buffer ptr */
789 readQNodes[0].params[1];
790 for (i = 0; i < numDataNodes; i++) {
791 /* set up params related to Wnd nodes */
792 qNodes[0].params[2 * (numDataNodes + 1 + i) + 0] = /* pda */
793 writeDataNodes[i].params[0];
794 qNodes[0].params[2 * (numDataNodes + 1 + i) + 1] = /* buffer ptr */
795 writeDataNodes[i].params[1];
796 }
797 /* xor node needs to get at RAID information */
798 qNodes[0].params[2 * (numDataNodes + numDataNodes + 1)].p = raidPtr;
799 qNodes[0].results[0] = readQNodes[0].params[1].p;
800 }
801 }
802
803 /* initialize nodes which write new parity (Wnp) */
804 pda = asmap->parityInfo;
805 for (i = 0; i < numParityNodes; i++) {
806 rf_InitNode(&writeParityNodes[i], rf_wait, RF_FALSE, rf_DiskWriteFunc,
807 rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h,
808 "Wnp", allocList);
809 RF_ASSERT(pda != NULL);
810 writeParityNodes[i].params[0].p = pda; /* param 1 (bufPtr)
811 * filled in by xor node */
812 writeParityNodes[i].params[1].p = xorNodes[i].results[0]; /* buffer pointer for
813 * parity write
814 * operation */
815 writeParityNodes[i].params[2].v = parityStripeID;
816 writeParityNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
817 0, 0, which_ru);
818 if (lu_flag) {
819 /* initialize node to unlock the disk queue */
820 rf_InitNode(&unlockParityNodes[i], rf_wait, RF_FALSE, rf_DiskUnlockFunc,
821 rf_DiskUnlockUndoFunc, rf_GenericWakeupFunc, 1, 1, 2, 0, dag_h,
822 "Unp", allocList);
823 unlockParityNodes[i].params[0].p = pda; /* physical disk addr
824 * desc */
825 unlockParityNodes[i].params[1].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
826 0, lu_flag, which_ru);
827 }
828 pda = pda->next;
829 }
830
831 /* initialize nodes which write new Q (Wnq) */
832 if (nfaults == 2) {
833 pda = asmap->qInfo;
834 for (i = 0; i < numParityNodes; i++) {
835 rf_InitNode(&writeQNodes[i], rf_wait, RF_FALSE, rf_DiskWriteFunc,
836 rf_DiskWriteUndoFunc, rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h,
837 "Wnq", allocList);
838 RF_ASSERT(pda != NULL);
839 writeQNodes[i].params[0].p = pda; /* param 1 (bufPtr)
840 * filled in by xor node */
841 writeQNodes[i].params[1].p = qNodes[i].results[0]; /* buffer pointer for
842 * parity write
843 * operation */
844 writeQNodes[i].params[2].v = parityStripeID;
845 writeQNodes[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
846 0, 0, which_ru);
847 if (lu_flag) {
848 /* initialize node to unlock the disk queue */
849 rf_InitNode(&unlockQNodes[i], rf_wait, RF_FALSE, rf_DiskUnlockFunc,
850 rf_DiskUnlockUndoFunc, rf_GenericWakeupFunc, 1, 1, 2, 0, dag_h,
851 "Unq", allocList);
852 unlockQNodes[i].params[0].p = pda; /* physical disk addr
853 * desc */
854 unlockQNodes[i].params[1].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY,
855 0, lu_flag, which_ru);
856 }
857 pda = pda->next;
858 }
859 }
860 /*
861 * Step 4. connect the nodes.
862 */
863
864 /* connect header to block node */
865 dag_h->succedents[0] = blockNode;
866
867 /* connect block node to read old data nodes */
868 RF_ASSERT(blockNode->numSuccedents == (numDataNodes + (numParityNodes * nfaults)));
869 for (i = 0; i < numDataNodes; i++) {
870 blockNode->succedents[i] = &readDataNodes[i];
871 RF_ASSERT(readDataNodes[i].numAntecedents == 1);
872 readDataNodes[i].antecedents[0] = blockNode;
873 readDataNodes[i].antType[0] = rf_control;
874 }
875
876 /* connect block node to read old parity nodes */
877 for (i = 0; i < numParityNodes; i++) {
878 blockNode->succedents[numDataNodes + i] = &readParityNodes[i];
879 RF_ASSERT(readParityNodes[i].numAntecedents == 1);
880 readParityNodes[i].antecedents[0] = blockNode;
881 readParityNodes[i].antType[0] = rf_control;
882 }
883
884 /* connect block node to read old Q nodes */
885 if (nfaults == 2) {
886 for (i = 0; i < numParityNodes; i++) {
887 blockNode->succedents[numDataNodes + numParityNodes + i] = &readQNodes[i];
888 RF_ASSERT(readQNodes[i].numAntecedents == 1);
889 readQNodes[i].antecedents[0] = blockNode;
890 readQNodes[i].antType[0] = rf_control;
891 }
892 }
893 /* connect read old data nodes to xor nodes */
894 for (i = 0; i < numDataNodes; i++) {
895 RF_ASSERT(readDataNodes[i].numSuccedents == (nfaults * numParityNodes));
896 for (j = 0; j < numParityNodes; j++) {
897 RF_ASSERT(xorNodes[j].numAntecedents == numDataNodes + numParityNodes);
898 readDataNodes[i].succedents[j] = &xorNodes[j];
899 xorNodes[j].antecedents[i] = &readDataNodes[i];
900 xorNodes[j].antType[i] = rf_trueData;
901 }
902 }
903
904 /* connect read old data nodes to q nodes */
905 if (nfaults == 2) {
906 for (i = 0; i < numDataNodes; i++) {
907 for (j = 0; j < numParityNodes; j++) {
908 RF_ASSERT(qNodes[j].numAntecedents == numDataNodes + numParityNodes);
909 readDataNodes[i].succedents[numParityNodes + j] = &qNodes[j];
910 qNodes[j].antecedents[i] = &readDataNodes[i];
911 qNodes[j].antType[i] = rf_trueData;
912 }
913 }
914 }
915 /* connect read old parity nodes to xor nodes */
916 for (i = 0; i < numParityNodes; i++) {
917 RF_ASSERT(readParityNodes[i].numSuccedents == numParityNodes);
918 for (j = 0; j < numParityNodes; j++) {
919 readParityNodes[i].succedents[j] = &xorNodes[j];
920 xorNodes[j].antecedents[numDataNodes + i] = &readParityNodes[i];
921 xorNodes[j].antType[numDataNodes + i] = rf_trueData;
922 }
923 }
924
925 /* connect read old q nodes to q nodes */
926 if (nfaults == 2) {
927 for (i = 0; i < numParityNodes; i++) {
928 RF_ASSERT(readParityNodes[i].numSuccedents == numParityNodes);
929 for (j = 0; j < numParityNodes; j++) {
930 readQNodes[i].succedents[j] = &qNodes[j];
931 qNodes[j].antecedents[numDataNodes + i] = &readQNodes[i];
932 qNodes[j].antType[numDataNodes + i] = rf_trueData;
933 }
934 }
935 }
936 /* connect xor nodes to commit node */
937 RF_ASSERT(commitNode->numAntecedents == (nfaults * numParityNodes));
938 for (i = 0; i < numParityNodes; i++) {
939 RF_ASSERT(xorNodes[i].numSuccedents == 1);
940 xorNodes[i].succedents[0] = commitNode;
941 commitNode->antecedents[i] = &xorNodes[i];
942 commitNode->antType[i] = rf_control;
943 }
944
945 /* connect q nodes to commit node */
946 if (nfaults == 2) {
947 for (i = 0; i < numParityNodes; i++) {
948 RF_ASSERT(qNodes[i].numSuccedents == 1);
949 qNodes[i].succedents[0] = commitNode;
950 commitNode->antecedents[i + numParityNodes] = &qNodes[i];
951 commitNode->antType[i + numParityNodes] = rf_control;
952 }
953 }
954 /* connect commit node to write nodes */
955 RF_ASSERT(commitNode->numSuccedents == (numDataNodes + (nfaults * numParityNodes)));
956 for (i = 0; i < numDataNodes; i++) {
957 RF_ASSERT(writeDataNodes[i].numAntecedents == 1);
958 commitNode->succedents[i] = &writeDataNodes[i];
959 writeDataNodes[i].antecedents[0] = commitNode;
960 writeDataNodes[i].antType[0] = rf_trueData;
961 }
962 for (i = 0; i < numParityNodes; i++) {
963 RF_ASSERT(writeParityNodes[i].numAntecedents == 1);
964 commitNode->succedents[i + numDataNodes] = &writeParityNodes[i];
965 writeParityNodes[i].antecedents[0] = commitNode;
966 writeParityNodes[i].antType[0] = rf_trueData;
967 }
968 if (nfaults == 2) {
969 for (i = 0; i < numParityNodes; i++) {
970 RF_ASSERT(writeQNodes[i].numAntecedents == 1);
971 commitNode->succedents[i + numDataNodes + numParityNodes] = &writeQNodes[i];
972 writeQNodes[i].antecedents[0] = commitNode;
973 writeQNodes[i].antType[0] = rf_trueData;
974 }
975 }
976 RF_ASSERT(termNode->numAntecedents == (numDataNodes + (nfaults * numParityNodes)));
977 RF_ASSERT(termNode->numSuccedents == 0);
978 for (i = 0; i < numDataNodes; i++) {
979 if (lu_flag) {
980 /* connect write new data nodes to unlock nodes */
981 RF_ASSERT(writeDataNodes[i].numSuccedents == 1);
982 RF_ASSERT(unlockDataNodes[i].numAntecedents == 1);
983 writeDataNodes[i].succedents[0] = &unlockDataNodes[i];
984 unlockDataNodes[i].antecedents[0] = &writeDataNodes[i];
985 unlockDataNodes[i].antType[0] = rf_control;
986
987 /* connect unlock nodes to term node */
988 RF_ASSERT(unlockDataNodes[i].numSuccedents == 1);
989 unlockDataNodes[i].succedents[0] = termNode;
990 termNode->antecedents[i] = &unlockDataNodes[i];
991 termNode->antType[i] = rf_control;
992 } else {
993 /* connect write new data nodes to term node */
994 RF_ASSERT(writeDataNodes[i].numSuccedents == 1);
995 RF_ASSERT(termNode->numAntecedents == (numDataNodes + (nfaults * numParityNodes)));
996 writeDataNodes[i].succedents[0] = termNode;
997 termNode->antecedents[i] = &writeDataNodes[i];
998 termNode->antType[i] = rf_control;
999 }
1000 }
1001
1002 for (i = 0; i < numParityNodes; i++) {
1003 if (lu_flag) {
1004 /* connect write new parity nodes to unlock nodes */
1005 RF_ASSERT(writeParityNodes[i].numSuccedents == 1);
1006 RF_ASSERT(unlockParityNodes[i].numAntecedents == 1);
1007 writeParityNodes[i].succedents[0] = &unlockParityNodes[i];
1008 unlockParityNodes[i].antecedents[0] = &writeParityNodes[i];
1009 unlockParityNodes[i].antType[0] = rf_control;
1010
1011 /* connect unlock nodes to term node */
1012 RF_ASSERT(unlockParityNodes[i].numSuccedents == 1);
1013 unlockParityNodes[i].succedents[0] = termNode;
1014 termNode->antecedents[numDataNodes + i] = &unlockParityNodes[i];
1015 termNode->antType[numDataNodes + i] = rf_control;
1016 } else {
1017 RF_ASSERT(writeParityNodes[i].numSuccedents == 1);
1018 writeParityNodes[i].succedents[0] = termNode;
1019 termNode->antecedents[numDataNodes + i] = &writeParityNodes[i];
1020 termNode->antType[numDataNodes + i] = rf_control;
1021 }
1022 }
1023
1024 if (nfaults == 2) {
1025 for (i = 0; i < numParityNodes; i++) {
1026 if (lu_flag) {
1027 /* connect write new Q nodes to unlock nodes */
1028 RF_ASSERT(writeQNodes[i].numSuccedents == 1);
1029 RF_ASSERT(unlockQNodes[i].numAntecedents == 1);
1030 writeQNodes[i].succedents[0] = &unlockQNodes[i];
1031 unlockQNodes[i].antecedents[0] = &writeQNodes[i];
1032 unlockQNodes[i].antType[0] = rf_control;
1033
1034 /* connect unlock nodes to unblock node */
1035 RF_ASSERT(unlockQNodes[i].numSuccedents == 1);
1036 unlockQNodes[i].succedents[0] = termNode;
1037 termNode->antecedents[numDataNodes + numParityNodes + i] = &unlockQNodes[i];
1038 termNode->antType[numDataNodes + numParityNodes + i] = rf_control;
1039 } else {
1040 RF_ASSERT(writeQNodes[i].numSuccedents == 1);
1041 writeQNodes[i].succedents[0] = termNode;
1042 termNode->antecedents[numDataNodes + numParityNodes + i] = &writeQNodes[i];
1043 termNode->antType[numDataNodes + numParityNodes + i] = rf_control;
1044 }
1045 }
1046 }
1047 }
1048
1049
1050 /******************************************************************************
1051 * create a write graph (fault-free or degraded) for RAID level 1
1052 *
1053 * Hdr -> Commit -> Wpd -> Nil -> Trm
1054 * -> Wsd ->
1055 *
1056 * The "Wpd" node writes data to the primary copy in the mirror pair
1057 * The "Wsd" node writes data to the secondary copy in the mirror pair
1058 *
1059 * Parameters: raidPtr - description of the physical array
1060 * asmap - logical & physical addresses for this access
1061 * bp - buffer ptr (holds write data)
1062 * flags - general flags (e.g. disk locking)
1063 * allocList - list of memory allocated in DAG creation
1064 *****************************************************************************/
1065
1066 void
1067 rf_CreateRaidOneWriteDAG(
1068 RF_Raid_t * raidPtr,
1069 RF_AccessStripeMap_t * asmap,
1070 RF_DagHeader_t * dag_h,
1071 void *bp,
1072 RF_RaidAccessFlags_t flags,
1073 RF_AllocListElem_t * allocList)
1074 {
1075 RF_DagNode_t *unblockNode, *termNode, *commitNode;
1076 RF_DagNode_t *nodes, *wndNode, *wmirNode;
1077 int nWndNodes, nWmirNodes, i;
1078 RF_ReconUnitNum_t which_ru;
1079 RF_PhysDiskAddr_t *pda, *pdaP;
1080 RF_StripeNum_t parityStripeID;
1081
1082 parityStripeID = rf_RaidAddressToParityStripeID(&(raidPtr->Layout),
1083 asmap->raidAddress, &which_ru);
1084 if (rf_dagDebug) {
1085 printf("[Creating RAID level 1 write DAG]\n");
1086 }
1087 dag_h->creator = "RaidOneWriteDAG";
1088
1089 /* 2 implies access not SU aligned */
1090 nWmirNodes = (asmap->parityInfo->next) ? 2 : 1;
1091 nWndNodes = (asmap->physInfo->next) ? 2 : 1;
1092
1093 /* alloc the Wnd nodes and the Wmir node */
1094 if (asmap->numDataFailed == 1)
1095 nWndNodes--;
1096 if (asmap->numParityFailed == 1)
1097 nWmirNodes--;
1098
1099 /* total number of nodes = nWndNodes + nWmirNodes + (commit + unblock
1100 * + terminator) */
1101 RF_CallocAndAdd(nodes, nWndNodes + nWmirNodes + 3, sizeof(RF_DagNode_t),
1102 (RF_DagNode_t *), allocList);
1103 i = 0;
1104 wndNode = &nodes[i];
1105 i += nWndNodes;
1106 wmirNode = &nodes[i];
1107 i += nWmirNodes;
1108 commitNode = &nodes[i];
1109 i += 1;
1110 unblockNode = &nodes[i];
1111 i += 1;
1112 termNode = &nodes[i];
1113 i += 1;
1114 RF_ASSERT(i == (nWndNodes + nWmirNodes + 3));
1115
1116 /* this dag can commit immediately */
1117 dag_h->numCommitNodes = 1;
1118 dag_h->numCommits = 0;
1119 dag_h->numSuccedents = 1;
1120
1121 /* initialize the commit, unblock, and term nodes */
1122 rf_InitNode(commitNode, rf_wait, RF_TRUE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
1123 NULL, (nWndNodes + nWmirNodes), 0, 0, 0, dag_h, "Cmt", allocList);
1124 rf_InitNode(unblockNode, rf_wait, RF_FALSE, rf_NullNodeFunc, rf_NullNodeUndoFunc,
1125 NULL, 1, (nWndNodes + nWmirNodes), 0, 0, dag_h, "Nil", allocList);
1126 rf_InitNode(termNode, rf_wait, RF_FALSE, rf_TerminateFunc, rf_TerminateUndoFunc,
1127 NULL, 0, 1, 0, 0, dag_h, "Trm", allocList);
1128
1129 /* initialize the wnd nodes */
1130 if (nWndNodes > 0) {
1131 pda = asmap->physInfo;
1132 for (i = 0; i < nWndNodes; i++) {
1133 rf_InitNode(&wndNode[i], rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
1134 rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wpd", allocList);
1135 RF_ASSERT(pda != NULL);
1136 wndNode[i].params[0].p = pda;
1137 wndNode[i].params[1].p = pda->bufPtr;
1138 wndNode[i].params[2].v = parityStripeID;
1139 wndNode[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
1140 pda = pda->next;
1141 }
1142 RF_ASSERT(pda == NULL);
1143 }
1144 /* initialize the mirror nodes */
1145 if (nWmirNodes > 0) {
1146 pda = asmap->physInfo;
1147 pdaP = asmap->parityInfo;
1148 for (i = 0; i < nWmirNodes; i++) {
1149 rf_InitNode(&wmirNode[i], rf_wait, RF_FALSE, rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
1150 rf_GenericWakeupFunc, 1, 1, 4, 0, dag_h, "Wsd", allocList);
1151 RF_ASSERT(pda != NULL);
1152 wmirNode[i].params[0].p = pdaP;
1153 wmirNode[i].params[1].p = pda->bufPtr;
1154 wmirNode[i].params[2].v = parityStripeID;
1155 wmirNode[i].params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0, 0, which_ru);
1156 pda = pda->next;
1157 pdaP = pdaP->next;
1158 }
1159 RF_ASSERT(pda == NULL);
1160 RF_ASSERT(pdaP == NULL);
1161 }
1162 /* link the header node to the commit node */
1163 RF_ASSERT(dag_h->numSuccedents == 1);
1164 RF_ASSERT(commitNode->numAntecedents == 0);
1165 dag_h->succedents[0] = commitNode;
1166
1167 /* link the commit node to the write nodes */
1168 RF_ASSERT(commitNode->numSuccedents == (nWndNodes + nWmirNodes));
1169 for (i = 0; i < nWndNodes; i++) {
1170 RF_ASSERT(wndNode[i].numAntecedents == 1);
1171 commitNode->succedents[i] = &wndNode[i];
1172 wndNode[i].antecedents[0] = commitNode;
1173 wndNode[i].antType[0] = rf_control;
1174 }
1175 for (i = 0; i < nWmirNodes; i++) {
1176 RF_ASSERT(wmirNode[i].numAntecedents == 1);
1177 commitNode->succedents[i + nWndNodes] = &wmirNode[i];
1178 wmirNode[i].antecedents[0] = commitNode;
1179 wmirNode[i].antType[0] = rf_control;
1180 }
1181
1182 /* link the write nodes to the unblock node */
1183 RF_ASSERT(unblockNode->numAntecedents == (nWndNodes + nWmirNodes));
1184 for (i = 0; i < nWndNodes; i++) {
1185 RF_ASSERT(wndNode[i].numSuccedents == 1);
1186 wndNode[i].succedents[0] = unblockNode;
1187 unblockNode->antecedents[i] = &wndNode[i];
1188 unblockNode->antType[i] = rf_control;
1189 }
1190 for (i = 0; i < nWmirNodes; i++) {
1191 RF_ASSERT(wmirNode[i].numSuccedents == 1);
1192 wmirNode[i].succedents[0] = unblockNode;
1193 unblockNode->antecedents[i + nWndNodes] = &wmirNode[i];
1194 unblockNode->antType[i + nWndNodes] = rf_control;
1195 }
1196
1197 /* link the unblock node to the term node */
1198 RF_ASSERT(unblockNode->numSuccedents == 1);
1199 RF_ASSERT(termNode->numAntecedents == 1);
1200 RF_ASSERT(termNode->numSuccedents == 0);
1201 unblockNode->succedents[0] = termNode;
1202 termNode->antecedents[0] = unblockNode;
1203 termNode->antType[0] = rf_control;
1204 }
1205