1 /*
2 * Copyright (c) 2008 The DragonFly Project. All rights reserved.
3 *
4 * This code is derived from software contributed to The DragonFly Project
5 * by Matthew Dillon <dillon@backplane.com>
6 *
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions
9 * are met:
10 *
11 * 1. Redistributions of source code must retain the above copyright
12 * notice, this list of conditions and the following disclaimer.
13 * 2. Redistributions in binary form must reproduce the above copyright
14 * notice, this list of conditions and the following disclaimer in
15 * the documentation and/or other materials provided with the
16 * distribution.
17 * 3. Neither the name of The DragonFly Project nor the names of its
18 * contributors may be used to endorse or promote products derived
19 * from this software without specific, prior written permission.
20 *
21 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
24 * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
25 * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
26 * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
27 * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
28 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
29 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
30 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
31 * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
32 * SUCH DAMAGE.
33 */
34 /*
35 * HAMMER mirroring ioctls - serialize and deserialize modifications made
36 * to a filesystem.
37 */
38
39 #include "hammer.h"
40
41 static int hammer_mirror_check(hammer_cursor_t cursor,
42 struct hammer_ioc_mrecord_rec *mrec);
43 static int hammer_mirror_update(hammer_cursor_t cursor,
44 struct hammer_ioc_mrecord_rec *mrec);
45 static int hammer_ioc_mirror_write_rec(hammer_cursor_t cursor,
46 struct hammer_ioc_mrecord_rec *mrec,
47 struct hammer_ioc_mirror_rw *mirror,
48 u_int32_t localization,
49 char *uptr);
50 static int hammer_ioc_mirror_write_pass(hammer_cursor_t cursor,
51 struct hammer_ioc_mrecord_rec *mrec,
52 struct hammer_ioc_mirror_rw *mirror,
53 u_int32_t localization);
54 static int hammer_ioc_mirror_write_skip(hammer_cursor_t cursor,
55 struct hammer_ioc_mrecord_skip *mrec,
56 struct hammer_ioc_mirror_rw *mirror,
57 u_int32_t localization);
58 static int hammer_mirror_delete_to(hammer_cursor_t cursor,
59 struct hammer_ioc_mirror_rw *mirror);
60 static int hammer_mirror_nomirror(struct hammer_base_elm *base);
61
62 /*
63 * All B-Tree records within the specified key range which also conform
64 * to the transaction id range are returned. Mirroring code keeps track
65 * of the last transaction id fully scanned and can efficiently pick up
66 * where it left off if interrupted.
67 *
68 * The PFS is identified in the mirror structure. The passed ip is just
69 * some directory in the overall HAMMER filesystem and has nothing to
70 * do with the PFS.
71 */
72 int
73 hammer_ioc_mirror_read(hammer_transaction_t trans, hammer_inode_t ip,
74 struct hammer_ioc_mirror_rw *mirror)
75 {
76 struct hammer_cmirror cmirror;
77 struct hammer_cursor cursor;
78 union hammer_ioc_mrecord_any mrec;
79 hammer_btree_leaf_elm_t elm;
80 const int crc_start = HAMMER_MREC_CRCOFF;
81 char *uptr;
82 int error;
83 int data_len;
84 int bytes;
85 int eatdisk;
86 int mrec_flags;
87 u_int32_t localization;
88 u_int32_t rec_crc;
89
90 localization = (u_int32_t)mirror->pfs_id << 16;
91
92 if ((mirror->key_beg.localization | mirror->key_end.localization) &
93 HAMMER_LOCALIZE_PSEUDOFS_MASK) {
94 return(EINVAL);
95 }
96 if (hammer_btree_cmp(&mirror->key_beg, &mirror->key_end) > 0)
97 return(EINVAL);
98
99 mirror->key_cur = mirror->key_beg;
100 mirror->key_cur.localization &= HAMMER_LOCALIZE_MASK;
101 mirror->key_cur.localization += localization;
102 bzero(&mrec, sizeof(mrec));
103 bzero(&cmirror, sizeof(cmirror));
104
105 /*
106 * Make CRC errors non-fatal (at least on data), causing an EDOM
107 * error instead of EIO.
108 */
109 trans->flags |= HAMMER_TRANSF_CRCDOM;
110
111 retry:
112 error = hammer_init_cursor(trans, &cursor, NULL, NULL);
113 if (error) {
114 hammer_done_cursor(&cursor);
115 goto failed;
116 }
117 cursor.key_beg = mirror->key_cur;
118 cursor.key_end = mirror->key_end;
119 cursor.key_end.localization &= HAMMER_LOCALIZE_MASK;
120 cursor.key_end.localization += localization;
121
122 cursor.flags |= HAMMER_CURSOR_END_INCLUSIVE;
123 cursor.flags |= HAMMER_CURSOR_BACKEND;
124
125 /*
126 * This flag filters the search to only return elements whos create
127 * or delete TID is >= mirror_tid. The B-Tree uses the mirror_tid
128 * field stored with internal and leaf nodes to shortcut the scan.
129 */
130 cursor.flags |= HAMMER_CURSOR_MIRROR_FILTERED;
131 cursor.cmirror = &cmirror;
132 cmirror.mirror_tid = mirror->tid_beg;
133
134 error = hammer_btree_first(&cursor);
135 while (error == 0) {
136 /*
137 * Yield to more important tasks
138 */
139 if (error == 0) {
140 error = hammer_signal_check(trans->hmp);
141 if (error)
142 break;
143 }
144
145 /*
146 * An internal node can be returned in mirror-filtered
147 * mode and indicates that the scan is returning a skip
148 * range in the cursor->cmirror structure.
149 */
150 uptr = (char *)mirror->ubuf + mirror->count;
151 if (cursor.node->ondisk->type == HAMMER_BTREE_TYPE_INTERNAL) {
152 /*
153 * Check space
154 */
155 mirror->key_cur = cmirror.skip_beg;
156 bytes = sizeof(mrec.skip);
157 if (mirror->count + HAMMER_HEAD_DOALIGN(bytes) >
158 mirror->size) {
159 break;
160 }
161
162 /*
163 * Fill mrec
164 */
165 mrec.head.signature = HAMMER_IOC_MIRROR_SIGNATURE;
166 mrec.head.type = HAMMER_MREC_TYPE_SKIP;
167 mrec.head.rec_size = bytes;
168 mrec.skip.skip_beg = cmirror.skip_beg;
169 mrec.skip.skip_end = cmirror.skip_end;
170 mrec.head.rec_crc = crc32(&mrec.head.rec_size,
171 bytes - crc_start);
172 error = copyout(&mrec, uptr, bytes);
173 eatdisk = 0;
174 goto didwrite;
175 }
176
177 /*
178 * Leaf node. In full-history mode we could filter out
179 * elements modified outside the user-requested TID range.
180 *
181 * However, such elements must be returned so the writer
182 * can compare them against the target to determine what
183 * needs to be deleted on the target, particular for
184 * no-history mirrors.
185 */
186 KKASSERT(cursor.node->ondisk->type == HAMMER_BTREE_TYPE_LEAF);
187 elm = &cursor.node->ondisk->elms[cursor.index].leaf;
188 mirror->key_cur = elm->base;
189
190 /*
191 * If the record was created after our end point we just
192 * ignore it.
193 */
194 if (elm->base.create_tid > mirror->tid_end) {
195 error = 0;
196 bytes = 0;
197 eatdisk = 1;
198 goto didwrite;
199 }
200
201 /*
202 * Determine if we should generate a PASS or a REC. PASS
203 * records are records without any data payload. Such
204 * records will be generated if the target is already expected
205 * to have the record, allowing it to delete the gaps.
206 *
207 * A PASS record is also used to perform deletions on the
208 * target.
209 *
210 * Such deletions are needed if the master or files on the
211 * master are no-history, or if the slave is so far behind
212 * the master has already been pruned.
213 */
214 if (elm->base.create_tid < mirror->tid_beg) {
215 bytes = sizeof(mrec.rec);
216 if (mirror->count + HAMMER_HEAD_DOALIGN(bytes) >
217 mirror->size) {
218 break;
219 }
220
221 /*
222 * Fill mrec.
223 */
224 mrec.head.signature = HAMMER_IOC_MIRROR_SIGNATURE;
225 mrec.head.type = HAMMER_MREC_TYPE_PASS;
226 mrec.head.rec_size = bytes;
227 mrec.rec.leaf = *elm;
228 mrec.head.rec_crc = crc32(&mrec.head.rec_size,
229 bytes - crc_start);
230 error = copyout(&mrec, uptr, bytes);
231 eatdisk = 1;
232 goto didwrite;
233
234 }
235
236 /*
237 * The core code exports the data to userland.
238 *
239 * CRC errors on data are reported but passed through,
240 * but the data must be washed by the user program.
241 *
242 * If userland just wants the btree records it can
243 * request that bulk data not be returned. This is
244 * use during mirror-stream histogram generation.
245 */
246 mrec_flags = 0;
247 data_len = (elm->data_offset) ? elm->data_len : 0;
248 if (data_len &&
249 (mirror->head.flags & HAMMER_IOC_MIRROR_NODATA)) {
250 data_len = 0;
251 mrec_flags |= HAMMER_MRECF_NODATA;
252 }
253 if (data_len) {
254 error = hammer_btree_extract(&cursor,
255 HAMMER_CURSOR_GET_DATA);
256 if (error) {
257 if (error != EDOM)
258 break;
259 mrec_flags |= HAMMER_MRECF_CRC_ERROR |
260 HAMMER_MRECF_DATA_CRC_BAD;
261 }
262 }
263
264 bytes = sizeof(mrec.rec) + data_len;
265 if (mirror->count + HAMMER_HEAD_DOALIGN(bytes) > mirror->size)
266 break;
267
268 /*
269 * Construct the record for userland and copyout.
270 *
271 * The user is asking for a snapshot, if the record was
272 * deleted beyond the user-requested ending tid, the record
273 * is not considered deleted from the point of view of
274 * userland and delete_tid is cleared.
275 */
276 mrec.head.signature = HAMMER_IOC_MIRROR_SIGNATURE;
277 mrec.head.type = HAMMER_MREC_TYPE_REC | mrec_flags;
278 mrec.head.rec_size = bytes;
279 mrec.rec.leaf = *elm;
280
281 if (elm->base.delete_tid > mirror->tid_end)
282 mrec.rec.leaf.base.delete_tid = 0;
283 rec_crc = crc32(&mrec.head.rec_size,
284 sizeof(mrec.rec) - crc_start);
285 if (data_len)
286 rec_crc = crc32_ext(cursor.data, data_len, rec_crc);
287 mrec.head.rec_crc = rec_crc;
288 error = copyout(&mrec, uptr, sizeof(mrec.rec));
289 if (data_len && error == 0) {
290 error = copyout(cursor.data, uptr + sizeof(mrec.rec),
291 data_len);
292 }
293 eatdisk = 1;
294
295 /*
296 * eatdisk controls whether we skip the current cursor
297 * position on the next scan or not. If doing a SKIP
298 * the cursor is already positioned properly for the next
299 * scan and eatdisk will be 0.
300 */
301 didwrite:
302 if (error == 0) {
303 mirror->count += HAMMER_HEAD_DOALIGN(bytes);
304 if (eatdisk)
305 cursor.flags |= HAMMER_CURSOR_ATEDISK;
306 else
307 cursor.flags &= ~HAMMER_CURSOR_ATEDISK;
308 error = hammer_btree_iterate(&cursor);
309 }
310 }
311 if (error == ENOENT) {
312 mirror->key_cur = mirror->key_end;
313 error = 0;
314 }
315 hammer_done_cursor(&cursor);
316 if (error == EDEADLK)
317 goto retry;
318 if (error == EINTR) {
319 mirror->head.flags |= HAMMER_IOC_HEAD_INTR;
320 error = 0;
321 }
322 failed:
323 mirror->key_cur.localization &= HAMMER_LOCALIZE_MASK;
324 return(error);
325 }
326
327 /*
328 * Copy records from userland to the target mirror.
329 *
330 * The PFS is identified in the mirror structure. The passed ip is just
331 * some directory in the overall HAMMER filesystem and has nothing to
332 * do with the PFS. In fact, there might not even be a root directory for
333 * the PFS yet!
334 */
335 int
336 hammer_ioc_mirror_write(hammer_transaction_t trans, hammer_inode_t ip,
337 struct hammer_ioc_mirror_rw *mirror)
338 {
339 union hammer_ioc_mrecord_any mrec;
340 struct hammer_cursor cursor;
341 u_int32_t localization;
342 int checkspace_count = 0;
343 int error;
344 int bytes;
345 char *uptr;
346 int seq;
347
348 localization = (u_int32_t)mirror->pfs_id << 16;
349 seq = trans->hmp->flusher.done;
350
351 /*
352 * Validate the mirror structure and relocalize the tracking keys.
353 */
354 if (mirror->size < 0 || mirror->size > 0x70000000)
355 return(EINVAL);
356 mirror->key_beg.localization &= HAMMER_LOCALIZE_MASK;
357 mirror->key_beg.localization += localization;
358 mirror->key_end.localization &= HAMMER_LOCALIZE_MASK;
359 mirror->key_end.localization += localization;
360 mirror->key_cur.localization &= HAMMER_LOCALIZE_MASK;
361 mirror->key_cur.localization += localization;
362
363 /*
364 * Set up our tracking cursor for the loop. The tracking cursor
365 * is used to delete records that are no longer present on the
366 * master. The last handled record at key_cur must be skipped.
367 */
368 error = hammer_init_cursor(trans, &cursor, NULL, NULL);
369
370 cursor.key_beg = mirror->key_cur;
371 cursor.key_end = mirror->key_end;
372 cursor.flags |= HAMMER_CURSOR_BACKEND;
373 error = hammer_btree_first(&cursor);
374 if (error == 0)
375 cursor.flags |= HAMMER_CURSOR_ATEDISK;
376 if (error == ENOENT)
377 error = 0;
378
379 /*
380 * Loop until our input buffer has been exhausted.
381 */
382 while (error == 0 &&
383 mirror->count + sizeof(mrec.head) <= mirror->size) {
384
385 /*
386 * Don't blow out the buffer cache. Leave room for frontend
387 * cache as well.
388 *
389 * WARNING: See warnings in hammer_unlock_cursor() function.
390 */
391 while (hammer_flusher_meta_halflimit(trans->hmp) ||
392 hammer_flusher_undo_exhausted(trans, 2)) {
393 hammer_unlock_cursor(&cursor);
394 hammer_flusher_wait(trans->hmp, seq);
395 hammer_lock_cursor(&cursor);
396 seq = hammer_flusher_async_one(trans->hmp);
397 }
398
399 /*
400 * If there is insufficient free space it may be due to
401 * reserved bigblocks, which flushing might fix.
402 */
403 if (hammer_checkspace(trans->hmp, HAMMER_CHKSPC_MIRROR)) {
404 if (++checkspace_count == 10) {
405 error = ENOSPC;
406 break;
407 }
408 hammer_unlock_cursor(&cursor);
409 hammer_flusher_wait(trans->hmp, seq);
410 hammer_lock_cursor(&cursor);
411 seq = hammer_flusher_async(trans->hmp, NULL);
412 }
413
414
415 /*
416 * Acquire and validate header
417 */
418 if ((bytes = mirror->size - mirror->count) > sizeof(mrec))
419 bytes = sizeof(mrec);
420 uptr = (char *)mirror->ubuf + mirror->count;
421 error = copyin(uptr, &mrec, bytes);
422 if (error)
423 break;
424 if (mrec.head.signature != HAMMER_IOC_MIRROR_SIGNATURE) {
425 error = EINVAL;
426 break;
427 }
428 if (mrec.head.rec_size < sizeof(mrec.head) ||
429 mrec.head.rec_size > sizeof(mrec) + HAMMER_XBUFSIZE ||
430 mirror->count + mrec.head.rec_size > mirror->size) {
431 error = EINVAL;
432 break;
433 }
434
435 switch(mrec.head.type & HAMMER_MRECF_TYPE_MASK) {
436 case HAMMER_MREC_TYPE_SKIP:
437 if (mrec.head.rec_size != sizeof(mrec.skip))
438 error = EINVAL;
439 if (error == 0)
440 error = hammer_ioc_mirror_write_skip(&cursor, &mrec.skip, mirror, localization);
441 break;
442 case HAMMER_MREC_TYPE_REC:
443 if (mrec.head.rec_size < sizeof(mrec.rec))
444 error = EINVAL;
445 if (error == 0)
446 error = hammer_ioc_mirror_write_rec(&cursor, &mrec.rec, mirror, localization, uptr + sizeof(mrec.rec));
447 break;
448 case HAMMER_MREC_TYPE_REC_NODATA:
449 case HAMMER_MREC_TYPE_REC_BADCRC:
450 /*
451 * Records with bad data payloads are ignored XXX.
452 * Records with no data payload have to be skipped
453 * (they shouldn't have been written in the first
454 * place).
455 */
456 if (mrec.head.rec_size < sizeof(mrec.rec))
457 error = EINVAL;
458 break;
459 case HAMMER_MREC_TYPE_PASS:
460 if (mrec.head.rec_size != sizeof(mrec.rec))
461 error = EINVAL;
462 if (error == 0)
463 error = hammer_ioc_mirror_write_pass(&cursor, &mrec.rec, mirror, localization);
464 break;
465 default:
466 error = EINVAL;
467 break;
468 }
469
470 /*
471 * Retry the current record on deadlock, otherwise setup
472 * for the next loop.
473 */
474 if (error == EDEADLK) {
475 while (error == EDEADLK) {
476 hammer_sync_lock_sh(trans);
477 hammer_recover_cursor(&cursor);
478 error = hammer_cursor_upgrade(&cursor);
479 hammer_sync_unlock(trans);
480 }
481 } else {
482 if (error == EALREADY)
483 error = 0;
484 if (error == 0) {
485 mirror->count +=
486 HAMMER_HEAD_DOALIGN(mrec.head.rec_size);
487 }
488 }
489 }
490 hammer_done_cursor(&cursor);
491
492 /*
493 * cumulative error
494 */
495 if (error) {
496 mirror->head.flags |= HAMMER_IOC_HEAD_ERROR;
497 mirror->head.error = error;
498 }
499
500 /*
501 * ioctls don't update the RW data structure if an error is returned,
502 * always return 0.
503 */
504 return(0);
505 }
506
507 /*
508 * Handle skip records.
509 *
510 * We must iterate from the last resolved record position at mirror->key_cur
511 * to skip_beg non-inclusive and delete any records encountered.
512 *
513 * mirror->key_cur must be carefully set when we succeed in processing
514 * this mrec.
515 */
516 static int
517 hammer_ioc_mirror_write_skip(hammer_cursor_t cursor,
518 struct hammer_ioc_mrecord_skip *mrec,
519 struct hammer_ioc_mirror_rw *mirror,
520 u_int32_t localization)
521 {
522 int error;
523
524 /*
525 * Relocalize the skip range
526 */
527 mrec->skip_beg.localization &= HAMMER_LOCALIZE_MASK;
528 mrec->skip_beg.localization += localization;
529 mrec->skip_end.localization &= HAMMER_LOCALIZE_MASK;
530 mrec->skip_end.localization += localization;
531
532 /*
533 * Iterate from current position to skip_beg, deleting any records
534 * we encounter. The record at skip_beg is not included (it is
535 * skipped).
536 */
537 cursor->key_end = mrec->skip_beg;
538 cursor->flags &= ~HAMMER_CURSOR_END_INCLUSIVE;
539 cursor->flags |= HAMMER_CURSOR_BACKEND;
540 error = hammer_mirror_delete_to(cursor, mirror);
541
542 /*
543 * Now skip past the skip (which is the whole point point of
544 * having a skip record). The sender has not sent us any records
545 * for the skip area so we wouldn't know what to keep and what
546 * to delete anyway.
547 *
548 * Clear ATEDISK because skip_end is non-inclusive, so we can't
549 * count an exact match if we happened to get one.
550 */
551 if (error == 0) {
552 mirror->key_cur = mrec->skip_end;
553 cursor->key_beg = mrec->skip_end;
554 error = hammer_btree_lookup(cursor);
555 cursor->flags &= ~HAMMER_CURSOR_ATEDISK;
556 if (error == ENOENT)
557 error = 0;
558 }
559 return(error);
560 }
561
562 /*
563 * Handle B-Tree records.
564 *
565 * We must iterate to mrec->base.key (non-inclusively), and then process
566 * the record. We are allowed to write a new record or delete an existing
567 * record, but cannot replace an existing record.
568 *
569 * mirror->key_cur must be carefully set when we succeed in processing
570 * this mrec.
571 */
572 static int
573 hammer_ioc_mirror_write_rec(hammer_cursor_t cursor,
574 struct hammer_ioc_mrecord_rec *mrec,
575 struct hammer_ioc_mirror_rw *mirror,
576 u_int32_t localization,
577 char *uptr)
578 {
579 int error;
580
581 if (mrec->leaf.data_len < 0 ||
582 mrec->leaf.data_len > HAMMER_XBUFSIZE ||
583 mrec->leaf.data_len + sizeof(*mrec) > mrec->head.rec_size) {
584 return(EINVAL);
585 }
586
587 /*
588 * Re-localize for target. relocalization of data is handled
589 * by hammer_mirror_write().
590 */
591 mrec->leaf.base.localization &= HAMMER_LOCALIZE_MASK;
592 mrec->leaf.base.localization += localization;
593
594 /*
595 * Delete records through until we reach (non-inclusively) the
596 * target record.
597 */
598 cursor->key_end = mrec->leaf.base;
599 cursor->flags &= ~HAMMER_CURSOR_END_INCLUSIVE;
600 cursor->flags |= HAMMER_CURSOR_BACKEND;
601 error = hammer_mirror_delete_to(cursor, mirror);
602
603 /*
604 * Certain records are not part of the mirroring operation
605 */
606 if (error == 0 && hammer_mirror_nomirror(&mrec->leaf.base))
607 return(0);
608
609 /*
610 * Locate the record.
611 *
612 * If the record exists only the delete_tid may be updated.
613 *
614 * If the record does not exist we can create it only if the
615 * create_tid is not too old. If the create_tid is too old
616 * it may have already been destroyed on the slave from pruning.
617 *
618 * Note that mirror operations are effectively as-of operations
619 * and delete_tid can be 0 for mirroring purposes even if it is
620 * not actually 0 at the originator.
621 *
622 * These functions can return EDEADLK
623 */
624 if (error == 0) {
625 cursor->key_beg = mrec->leaf.base;
626 cursor->flags |= HAMMER_CURSOR_BACKEND;
627 cursor->flags &= ~HAMMER_CURSOR_INSERT;
628 error = hammer_btree_lookup(cursor);
629 }
630
631 if (error == 0 && hammer_mirror_check(cursor, mrec)) {
632 error = hammer_mirror_update(cursor, mrec);
633 } else if (error == ENOENT) {
634 if (mrec->leaf.base.create_tid >= mirror->tid_beg) {
635 error = hammer_create_at_cursor(
636 cursor, &mrec->leaf,
637 uptr, HAMMER_CREATE_MODE_UMIRROR);
638 } else {
639 error = 0;
640 }
641 }
642 if (error == 0 || error == EALREADY)
643 mirror->key_cur = mrec->leaf.base;
644 return(error);
645 }
646
647 /*
648 * This works like write_rec but no write or update is necessary,
649 * and no data payload is included so we couldn't do a write even
650 * if we wanted to.
651 *
652 * We must still iterate for deletions, and we can validate the
653 * record header which is a good way to test for corrupted mirror
654 * targets XXX.
655 *
656 * mirror->key_cur must be carefully set when we succeed in processing
657 * this mrec.
658 */
659 static
660 int
661 hammer_ioc_mirror_write_pass(hammer_cursor_t cursor,
662 struct hammer_ioc_mrecord_rec *mrec,
663 struct hammer_ioc_mirror_rw *mirror,
664 u_int32_t localization)
665 {
666 int error;
667
668 /*
669 * Re-localize for target. Relocalization of data is handled
670 * by hammer_mirror_write().
671 */
672 mrec->leaf.base.localization &= HAMMER_LOCALIZE_MASK;
673 mrec->leaf.base.localization += localization;
674
675 /*
676 * Delete records through until we reach (non-inclusively) the
677 * target record.
678 */
679 cursor->key_end = mrec->leaf.base;
680 cursor->flags &= ~HAMMER_CURSOR_END_INCLUSIVE;
681 cursor->flags |= HAMMER_CURSOR_BACKEND;
682 error = hammer_mirror_delete_to(cursor, mirror);
683
684 /*
685 * Certain records are not part of the mirroring operation
686 */
687 if (hammer_mirror_nomirror(&mrec->leaf.base))
688 return(0);
689
690 /*
691 * Locate the record and get past it by setting ATEDISK. Perform
692 * any necessary deletions. We have no data payload and cannot
693 * create a new record.
694 */
695 if (error == 0) {
696 mirror->key_cur = mrec->leaf.base;
697 cursor->key_beg = mrec->leaf.base;
698 cursor->flags |= HAMMER_CURSOR_BACKEND;
699 cursor->flags &= ~HAMMER_CURSOR_INSERT;
700 error = hammer_btree_lookup(cursor);
701 if (error == 0) {
702 if (hammer_mirror_check(cursor, mrec))
703 error = hammer_mirror_update(cursor, mrec);
704 cursor->flags |= HAMMER_CURSOR_ATEDISK;
705 } else {
706 cursor->flags &= ~HAMMER_CURSOR_ATEDISK;
707 }
708 if (error == ENOENT)
709 error = 0;
710 }
711 return(error);
712 }
713
714 /*
715 * As part of the mirror write we iterate across swaths of records
716 * on the target which no longer exist on the source, and mark them
717 * deleted.
718 *
719 * The caller has indexed the cursor and set up key_end. We iterate
720 * through to key_end.
721 *
722 * There is an edge case where the master has deleted a record whos
723 * create_tid exactly matches our end_tid. We cannot delete this
724 * record on the slave yet because we cannot assign delete_tid == create_tid.
725 * The deletion should be picked up on the next sequence since in order
726 * to have been deleted on the master a transaction must have occured with
727 * a TID greater then the create_tid of the record.
728 *
729 * To support incremental re-mirroring, just for robustness, we do not
730 * touch any records created beyond (or equal to) mirror->tid_end.
731 */
732 static
733 int
734 hammer_mirror_delete_to(hammer_cursor_t cursor,
735 struct hammer_ioc_mirror_rw *mirror)
736 {
737 hammer_btree_leaf_elm_t elm;
738 int error;
739
740 error = hammer_btree_iterate(cursor);
741 while (error == 0) {
742 elm = &cursor->node->ondisk->elms[cursor->index].leaf;
743 KKASSERT(elm->base.btype == HAMMER_BTREE_TYPE_RECORD);
744 cursor->flags |= HAMMER_CURSOR_ATEDISK;
745
746 /*
747 * Certain records are not part of the mirroring operation
748 */
749 if (hammer_mirror_nomirror(&elm->base)) {
750 error = hammer_btree_iterate(cursor);
751 continue;
752 }
753
754 /*
755 * Note: Must still delete records with create_tid < tid_beg,
756 * as record may have been pruned-away on source.
757 */
758 if (elm->base.delete_tid == 0 &&
759 elm->base.create_tid < mirror->tid_end) {
760 error = hammer_delete_at_cursor(cursor,
761 HAMMER_DELETE_ADJUST,
762 mirror->tid_end,
763 time_second,
764 1, NULL);
765 }
766 if (error == 0)
767 error = hammer_btree_iterate(cursor);
768 }
769 if (error == ENOENT)
770 error = 0;
771 return(error);
772 }
773
774 /*
775 * Check whether an update is needed in the case where a match already
776 * exists on the target. The only type of update allowed in this case
777 * is an update of the delete_tid.
778 *
779 * Return non-zero if the update should proceed.
780 */
781 static
782 int
783 hammer_mirror_check(hammer_cursor_t cursor, struct hammer_ioc_mrecord_rec *mrec)
784 {
785 hammer_btree_leaf_elm_t leaf = cursor->leaf;
786
787 if (leaf->base.delete_tid != mrec->leaf.base.delete_tid) {
788 if (mrec->leaf.base.delete_tid != 0)
789 return(1);
790 }
791 return(0);
792 }
793
794 /*
795 * Filter out records which are never mirrored, such as configuration space
796 * records (for hammer cleanup).
797 *
798 * NOTE: We currently allow HAMMER_RECTYPE_SNAPSHOT records to be mirrored.
799 */
800 static
801 int
802 hammer_mirror_nomirror(struct hammer_base_elm *base)
803 {
804 /*
805 * Certain types of records are never updated when mirroring.
806 * Slaves have their own configuration space.
807 */
808 if (base->rec_type == HAMMER_RECTYPE_CONFIG)
809 return(1);
810 return(0);
811 }
812
813
814 /*
815 * Update a record in-place. Only the delete_tid can change, and
816 * only from zero to non-zero.
817 */
818 static
819 int
820 hammer_mirror_update(hammer_cursor_t cursor,
821 struct hammer_ioc_mrecord_rec *mrec)
822 {
823 int error;
824
825 /*
826 * This case shouldn't occur.
827 */
828 if (mrec->leaf.base.delete_tid == 0)
829 return(0);
830
831 /*
832 * Mark the record deleted on the mirror target.
833 */
834 error = hammer_delete_at_cursor(cursor, HAMMER_DELETE_ADJUST,
835 mrec->leaf.base.delete_tid,
836 mrec->leaf.delete_ts,
837 1, NULL);
838 cursor->flags |= HAMMER_CURSOR_ATEDISK;
839 return(error);
840 }
841
842 #if 0
843 /*
844 * MOVED TO HAMMER_OBJECT.C: hammer_create_at_cursor()
845 */
846
847 static int hammer_mirror_localize_data(hammer_data_ondisk_t data,
848 hammer_btree_leaf_elm_t leaf);
849
850 /*
851 * Write out a new record.
852 */
853 static
854 int
855 hammer_mirror_write(hammer_cursor_t cursor,
856 struct hammer_ioc_mrecord_rec *mrec,
857 char *udata)
858 {
859 hammer_transaction_t trans;
860 hammer_buffer_t data_buffer;
861 hammer_off_t ndata_offset;
862 hammer_tid_t high_tid;
863 void *ndata;
864 int error;
865 int doprop;
866
867 trans = cursor->trans;
868 data_buffer = NULL;
869
870 /*
871 * Get the sync lock so the whole mess is atomic
872 */
873 hammer_sync_lock_sh(trans);
874
875 /*
876 * Allocate and adjust data
877 */
878 if (mrec->leaf.data_len && mrec->leaf.data_offset) {
879 ndata = hammer_alloc_data(trans, mrec->leaf.data_len,
880 mrec->leaf.base.rec_type,
881 &ndata_offset, &data_buffer,
882 0, &error);
883 if (ndata == NULL)
884 return(error);
885 mrec->leaf.data_offset = ndata_offset;
886 hammer_modify_buffer(trans, data_buffer, NULL, 0);
887 error = copyin(udata, ndata, mrec->leaf.data_len);
888 if (error == 0) {
889 if (hammer_crc_test_leaf(ndata, &mrec->leaf) == 0) {
890 kprintf("data crc mismatch on pipe\n");
891 error = EINVAL;
892 } else {
893 error = hammer_mirror_localize_data(
894 ndata, &mrec->leaf);
895 }
896 }
897 hammer_modify_buffer_done(data_buffer);
898 } else {
899 mrec->leaf.data_offset = 0;
900 error = 0;
901 ndata = NULL;
902 }
903 if (error)
904 goto failed;
905
906 /*
907 * Do the insertion. This can fail with a EDEADLK or EALREADY
908 */
909 cursor->flags |= HAMMER_CURSOR_INSERT;
910 error = hammer_btree_lookup(cursor);
911 if (error != ENOENT) {
912 if (error == 0)
913 error = EALREADY;
914 goto failed;
915 }
916
917 error = hammer_btree_insert(cursor, &mrec->leaf, &doprop);
918
919 /*
920 * Cursor is left on the current element, we want to skip it now.
921 */
922 cursor->flags |= HAMMER_CURSOR_ATEDISK;
923 cursor->flags &= ~HAMMER_CURSOR_INSERT;
924
925 /*
926 * Track a count of active inodes.
927 */
928 if (error == 0 &&
929 mrec->leaf.base.rec_type == HAMMER_RECTYPE_INODE &&
930 mrec->leaf.base.delete_tid == 0) {
931 hammer_modify_volume_field(trans,
932 trans->rootvol,
933 vol0_stat_inodes);
934 ++trans->hmp->rootvol->ondisk->vol0_stat_inodes;
935 hammer_modify_volume_done(trans->rootvol);
936 }
937
938 /*
939 * vol0_next_tid must track the highest TID stored in the filesystem.
940 * We do not need to generate undo for this update.
941 */
942 high_tid = mrec->leaf.base.create_tid;
943 if (high_tid < mrec->leaf.base.delete_tid)
944 high_tid = mrec->leaf.base.delete_tid;
945 if (trans->rootvol->ondisk->vol0_next_tid < high_tid) {
946 hammer_modify_volume(trans, trans->rootvol, NULL, 0);
947 trans->rootvol->ondisk->vol0_next_tid = high_tid;
948 hammer_modify_volume_done(trans->rootvol);
949 }
950
951 /*
952 * WARNING! cursor's leaf pointer may have changed after
953 * do_propagation returns.
954 */
955 if (error == 0 && doprop)
956 hammer_btree_do_propagation(cursor, NULL, &mrec->leaf);
957
958 failed:
959 /*
960 * Cleanup
961 */
962 if (error && mrec->leaf.data_offset) {
963 hammer_blockmap_free(cursor->trans,
964 mrec->leaf.data_offset,
965 mrec->leaf.data_len);
966 }
967 hammer_sync_unlock(trans);
968 if (data_buffer)
969 hammer_rel_buffer(data_buffer, 0);
970 return(error);
971 }
972
973 /*
974 * Localize the data payload. Directory entries may need their
975 * localization adjusted.
976 */
977 static
978 int
979 hammer_mirror_localize_data(hammer_data_ondisk_t data,
980 hammer_btree_leaf_elm_t leaf)
981 {
982 u_int32_t localization;
983
984 if (leaf->base.rec_type == HAMMER_RECTYPE_DIRENTRY) {
985 localization = leaf->base.localization &
986 HAMMER_LOCALIZE_PSEUDOFS_MASK;
987 if (data->entry.localization != localization) {
988 data->entry.localization = localization;
989 hammer_crc_set_leaf(data, leaf);
990 }
991 }
992 return(0);
993 }
994
995 #endif
Cache object: 58067e295e47d3925524672df3c23f7f
|