2 * Copyright (c) 2008 The DragonFly Project. All rights reserved.
4 * This code is derived from software contributed to The DragonFly Project
5 * by Matthew Dillon <dillon@backplane.com>
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions
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
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.
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
34 * $DragonFly: src/sys/vfs/hammer/hammer_mirror.c,v 1.17 2008/07/31 22:30:33 dillon Exp $
37 * HAMMER mirroring ioctls - serialize and deserialize modifications made
43 static int hammer_mirror_check(hammer_cursor_t cursor,
44 struct hammer_ioc_mrecord_rec *mrec);
45 static int hammer_mirror_update(hammer_cursor_t cursor,
46 struct hammer_ioc_mrecord_rec *mrec);
47 static int hammer_mirror_write(hammer_cursor_t cursor,
48 struct hammer_ioc_mrecord_rec *mrec,
50 static int hammer_ioc_mirror_write_rec(hammer_cursor_t cursor,
51 struct hammer_ioc_mrecord_rec *mrec,
52 struct hammer_ioc_mirror_rw *mirror,
53 u_int32_t localization,
55 static int hammer_ioc_mirror_write_pass(hammer_cursor_t cursor,
56 struct hammer_ioc_mrecord_rec *mrec,
57 struct hammer_ioc_mirror_rw *mirror,
58 u_int32_t localization);
59 static int hammer_ioc_mirror_write_skip(hammer_cursor_t cursor,
60 struct hammer_ioc_mrecord_skip *mrec,
61 struct hammer_ioc_mirror_rw *mirror,
62 u_int32_t localization);
63 static int hammer_mirror_delete_to(hammer_cursor_t cursor,
64 struct hammer_ioc_mirror_rw *mirror);
65 static int hammer_mirror_localize_data(hammer_data_ondisk_t data,
66 hammer_btree_leaf_elm_t leaf);
69 * All B-Tree records within the specified key range which also conform
70 * to the transaction id range are returned. Mirroring code keeps track
71 * of the last transaction id fully scanned and can efficiently pick up
72 * where it left off if interrupted.
74 * The PFS is identified in the mirror structure. The passed ip is just
75 * some directory in the overall HAMMER filesystem and has nothing to
79 hammer_ioc_mirror_read(hammer_transaction_t trans, hammer_inode_t ip,
80 struct hammer_ioc_mirror_rw *mirror)
82 struct hammer_cmirror cmirror;
83 struct hammer_cursor cursor;
84 union hammer_ioc_mrecord_any mrec;
85 hammer_btree_leaf_elm_t elm;
86 const int crc_start = HAMMER_MREC_CRCOFF;
93 u_int32_t localization;
96 localization = (u_int32_t)mirror->pfs_id << 16;
98 if ((mirror->key_beg.localization | mirror->key_end.localization) &
99 HAMMER_LOCALIZE_PSEUDOFS_MASK) {
102 if (hammer_btree_cmp(&mirror->key_beg, &mirror->key_end) > 0)
105 mirror->key_cur = mirror->key_beg;
106 mirror->key_cur.localization &= HAMMER_LOCALIZE_MASK;
107 mirror->key_cur.localization += localization;
108 bzero(&mrec, sizeof(mrec));
109 bzero(&cmirror, sizeof(cmirror));
112 * Make CRC errors non-fatal (at least on data), causing an EDOM
113 * error instead of EIO.
115 trans->flags |= HAMMER_TRANSF_CRCDOM;
118 error = hammer_init_cursor(trans, &cursor, NULL, NULL);
120 hammer_done_cursor(&cursor);
123 cursor.key_beg = mirror->key_cur;
124 cursor.key_end = mirror->key_end;
125 cursor.key_end.localization &= HAMMER_LOCALIZE_MASK;
126 cursor.key_end.localization += localization;
128 cursor.flags |= HAMMER_CURSOR_END_INCLUSIVE;
129 cursor.flags |= HAMMER_CURSOR_BACKEND;
132 * This flag filters the search to only return elements whos create
133 * or delete TID is >= mirror_tid. The B-Tree uses the mirror_tid
134 * field stored with internal and leaf nodes to shortcut the scan.
136 cursor.flags |= HAMMER_CURSOR_MIRROR_FILTERED;
137 cursor.cmirror = &cmirror;
138 cmirror.mirror_tid = mirror->tid_beg;
140 error = hammer_btree_first(&cursor);
143 * Yield to more important tasks
146 error = hammer_signal_check(trans->hmp);
152 * An internal node can be returned in mirror-filtered
153 * mode and indicates that the scan is returning a skip
154 * range in the cursor->cmirror structure.
156 uptr = (char *)mirror->ubuf + mirror->count;
157 if (cursor.node->ondisk->type == HAMMER_BTREE_TYPE_INTERNAL) {
161 mirror->key_cur = cmirror.skip_beg;
162 bytes = sizeof(mrec.skip);
163 if (mirror->count + HAMMER_HEAD_DOALIGN(bytes) >
171 mrec.head.signature = HAMMER_IOC_MIRROR_SIGNATURE;
172 mrec.head.type = HAMMER_MREC_TYPE_SKIP;
173 mrec.head.rec_size = bytes;
174 mrec.skip.skip_beg = cmirror.skip_beg;
175 mrec.skip.skip_end = cmirror.skip_end;
176 mrec.head.rec_crc = crc32(&mrec.head.rec_size,
178 error = copyout(&mrec, uptr, bytes);
184 * Leaf node. In full-history mode we could filter out
185 * elements modified outside the user-requested TID range.
187 * However, such elements must be returned so the writer
188 * can compare them against the target to determine what
189 * needs to be deleted on the target, particular for
190 * no-history mirrors.
192 KKASSERT(cursor.node->ondisk->type == HAMMER_BTREE_TYPE_LEAF);
193 elm = &cursor.node->ondisk->elms[cursor.index].leaf;
194 mirror->key_cur = elm->base;
197 * Determine if we should generate a PASS or a REC. PASS
198 * records are records without any data payload. Such
199 * records will be generated if the target is already expected
200 * to have the record, allowing it to delete the gaps.
202 * A PASS record is also used to perform deletions on the
205 * Such deletions are needed if the master or files on the
206 * master are no-history, or if the slave is so far behind
207 * the master has already been pruned.
209 if (elm->base.create_tid < mirror->tid_beg ||
210 elm->base.create_tid > mirror->tid_end) {
211 bytes = sizeof(mrec.rec);
212 if (mirror->count + HAMMER_HEAD_DOALIGN(bytes) >
220 mrec.head.signature = HAMMER_IOC_MIRROR_SIGNATURE;
221 mrec.head.type = HAMMER_MREC_TYPE_PASS;
222 mrec.head.rec_size = bytes;
223 mrec.rec.leaf = *elm;
224 mrec.head.rec_crc = crc32(&mrec.head.rec_size,
226 error = copyout(&mrec, uptr, bytes);
233 * The core code exports the data to userland.
235 * CRC errors on data are reported but passed through,
236 * but the data must be washed by the user program.
239 data_len = (elm->data_offset) ? elm->data_len : 0;
241 error = hammer_btree_extract(&cursor,
242 HAMMER_CURSOR_GET_DATA);
246 mrec_flags |= HAMMER_MRECF_CRC_ERROR |
247 HAMMER_MRECF_DATA_CRC_BAD;
251 bytes = sizeof(mrec.rec) + data_len;
252 if (mirror->count + HAMMER_HEAD_DOALIGN(bytes) > mirror->size)
256 * Construct the record for userland and copyout.
258 * The user is asking for a snapshot, if the record was
259 * deleted beyond the user-requested ending tid, the record
260 * is not considered deleted from the point of view of
261 * userland and delete_tid is cleared.
263 mrec.head.signature = HAMMER_IOC_MIRROR_SIGNATURE;
264 mrec.head.type = HAMMER_MREC_TYPE_REC | mrec_flags;
265 mrec.head.rec_size = bytes;
266 mrec.rec.leaf = *elm;
268 if (elm->base.delete_tid > mirror->tid_end)
269 mrec.rec.leaf.base.delete_tid = 0;
270 rec_crc = crc32(&mrec.head.rec_size,
271 sizeof(mrec.rec) - crc_start);
273 rec_crc = crc32_ext(cursor.data, data_len, rec_crc);
274 mrec.head.rec_crc = rec_crc;
275 error = copyout(&mrec, uptr, sizeof(mrec.rec));
276 if (data_len && error == 0) {
277 error = copyout(cursor.data, uptr + sizeof(mrec.rec),
283 * eatdisk controls whether we skip the current cursor
284 * position on the next scan or not. If doing a SKIP
285 * the cursor is already positioned properly for the next
286 * scan and eatdisk will be 0.
290 mirror->count += HAMMER_HEAD_DOALIGN(bytes);
292 cursor.flags |= HAMMER_CURSOR_ATEDISK;
294 cursor.flags &= ~HAMMER_CURSOR_ATEDISK;
295 error = hammer_btree_iterate(&cursor);
298 if (error == ENOENT) {
299 mirror->key_cur = mirror->key_end;
302 hammer_done_cursor(&cursor);
303 if (error == EDEADLK)
305 if (error == EINTR) {
306 mirror->head.flags |= HAMMER_IOC_HEAD_INTR;
310 mirror->key_cur.localization &= HAMMER_LOCALIZE_MASK;
315 * Copy records from userland to the target mirror.
317 * The PFS is identified in the mirror structure. The passed ip is just
318 * some directory in the overall HAMMER filesystem and has nothing to
319 * do with the PFS. In fact, there might not even be a root directory for
323 hammer_ioc_mirror_write(hammer_transaction_t trans, hammer_inode_t ip,
324 struct hammer_ioc_mirror_rw *mirror)
326 union hammer_ioc_mrecord_any mrec;
327 struct hammer_cursor cursor;
328 u_int32_t localization;
329 int checkspace_count = 0;
335 localization = (u_int32_t)mirror->pfs_id << 16;
336 seq = trans->hmp->flusher.act;
339 * Validate the mirror structure and relocalize the tracking keys.
341 if (mirror->size < 0 || mirror->size > 0x70000000)
343 mirror->key_beg.localization &= HAMMER_LOCALIZE_MASK;
344 mirror->key_beg.localization += localization;
345 mirror->key_end.localization &= HAMMER_LOCALIZE_MASK;
346 mirror->key_end.localization += localization;
347 mirror->key_cur.localization &= HAMMER_LOCALIZE_MASK;
348 mirror->key_cur.localization += localization;
351 * Set up our tracking cursor for the loop. The tracking cursor
352 * is used to delete records that are no longer present on the
353 * master. The last handled record at key_cur must be skipped.
355 error = hammer_init_cursor(trans, &cursor, NULL, NULL);
357 cursor.key_beg = mirror->key_cur;
358 cursor.key_end = mirror->key_end;
359 cursor.flags |= HAMMER_CURSOR_BACKEND;
360 error = hammer_btree_first(&cursor);
362 cursor.flags |= HAMMER_CURSOR_ATEDISK;
367 * Loop until our input buffer has been exhausted.
370 mirror->count + sizeof(mrec.head) <= mirror->size) {
373 * Don't blow out the buffer cache. Leave room for frontend
376 while (hammer_flusher_meta_halflimit(trans->hmp) ||
377 hammer_flusher_undo_exhausted(trans, 2)) {
378 hammer_unlock_cursor(&cursor);
379 hammer_flusher_wait(trans->hmp, seq);
380 hammer_lock_cursor(&cursor);
381 seq = hammer_flusher_async_one(trans->hmp);
385 * If there is insufficient free space it may be due to
386 * reserved bigblocks, which flushing might fix.
388 if (hammer_checkspace(trans->hmp, HAMMER_CHKSPC_MIRROR)) {
389 if (++checkspace_count == 10) {
393 hammer_unlock_cursor(&cursor);
394 hammer_flusher_wait(trans->hmp, seq);
395 hammer_lock_cursor(&cursor);
396 seq = hammer_flusher_async(trans->hmp, NULL);
401 * Acquire and validate header
403 if ((bytes = mirror->size - mirror->count) > sizeof(mrec))
404 bytes = sizeof(mrec);
405 uptr = (char *)mirror->ubuf + mirror->count;
406 error = copyin(uptr, &mrec, bytes);
409 if (mrec.head.signature != HAMMER_IOC_MIRROR_SIGNATURE) {
413 if (mrec.head.rec_size < sizeof(mrec.head) ||
414 mrec.head.rec_size > sizeof(mrec) + HAMMER_XBUFSIZE ||
415 mirror->count + mrec.head.rec_size > mirror->size) {
420 switch(mrec.head.type & HAMMER_MRECF_TYPE_MASK) {
421 case HAMMER_MREC_TYPE_SKIP:
422 if (mrec.head.rec_size != sizeof(mrec.skip))
425 error = hammer_ioc_mirror_write_skip(&cursor, &mrec.skip, mirror, localization);
427 case HAMMER_MREC_TYPE_REC:
428 if (mrec.head.rec_size < sizeof(mrec.rec))
431 error = hammer_ioc_mirror_write_rec(&cursor, &mrec.rec, mirror, localization, uptr + sizeof(mrec.rec));
433 case HAMMER_MREC_TYPE_REC_BADCRC:
435 * Records with bad data payloads are ignored XXX.
437 if (mrec.head.rec_size < sizeof(mrec.rec))
440 case HAMMER_MREC_TYPE_PASS:
441 if (mrec.head.rec_size != sizeof(mrec.rec))
444 error = hammer_ioc_mirror_write_pass(&cursor, &mrec.rec, mirror, localization);
452 * Retry the current record on deadlock, otherwise setup
455 if (error == EDEADLK) {
456 while (error == EDEADLK) {
457 hammer_recover_cursor(&cursor);
458 error = hammer_cursor_upgrade(&cursor);
461 if (error == EALREADY)
465 HAMMER_HEAD_DOALIGN(mrec.head.rec_size);
469 hammer_done_cursor(&cursor);
475 mirror->head.flags |= HAMMER_IOC_HEAD_ERROR;
476 mirror->head.error = error;
480 * ioctls don't update the RW data structure if an error is returned,
487 * Handle skip records.
489 * We must iterate from the last resolved record position at mirror->key_cur
490 * to skip_beg and delete any records encountered.
492 * mirror->key_cur must be carefully set when we succeed in processing
496 hammer_ioc_mirror_write_skip(hammer_cursor_t cursor,
497 struct hammer_ioc_mrecord_skip *mrec,
498 struct hammer_ioc_mirror_rw *mirror,
499 u_int32_t localization)
504 * Relocalize the skip range
506 mrec->skip_beg.localization &= HAMMER_LOCALIZE_MASK;
507 mrec->skip_beg.localization += localization;
508 mrec->skip_end.localization &= HAMMER_LOCALIZE_MASK;
509 mrec->skip_end.localization += localization;
512 * Iterate from current position to skip_beg, deleting any records
515 cursor->key_end = mrec->skip_beg;
516 cursor->flags |= HAMMER_CURSOR_BACKEND;
517 error = hammer_mirror_delete_to(cursor, mirror);
520 * Now skip past the skip (which is the whole point point of
521 * having a skip record). The sender has not sent us any records
522 * for the skip area so we wouldn't know what to keep and what
525 * Clear ATEDISK because skip_end is non-inclusive, so we can't
526 * count an exact match if we happened to get one.
529 mirror->key_cur = mrec->skip_end;
530 cursor->key_beg = mrec->skip_end;
531 error = hammer_btree_lookup(cursor);
532 cursor->flags &= ~HAMMER_CURSOR_ATEDISK;
540 * Handle B-Tree records.
542 * We must iterate to mrec->base.key (non-inclusively), and then process
543 * the record. We are allowed to write a new record or delete an existing
544 * record, but cannot replace an existing record.
546 * mirror->key_cur must be carefully set when we succeed in processing
550 hammer_ioc_mirror_write_rec(hammer_cursor_t cursor,
551 struct hammer_ioc_mrecord_rec *mrec,
552 struct hammer_ioc_mirror_rw *mirror,
553 u_int32_t localization,
556 hammer_transaction_t trans;
560 trans = cursor->trans;
561 rec_crc = crc32(mrec, sizeof(*mrec));
563 if (mrec->leaf.data_len < 0 ||
564 mrec->leaf.data_len > HAMMER_XBUFSIZE ||
565 mrec->leaf.data_len + sizeof(*mrec) > mrec->head.rec_size) {
570 * Re-localize for target. relocalization of data is handled
571 * by hammer_mirror_write().
573 mrec->leaf.base.localization &= HAMMER_LOCALIZE_MASK;
574 mrec->leaf.base.localization += localization;
577 * Delete records through until we reach (non-inclusively) the
580 cursor->key_end = mrec->leaf.base;
581 cursor->flags &= ~HAMMER_CURSOR_END_INCLUSIVE;
582 cursor->flags |= HAMMER_CURSOR_BACKEND;
583 error = hammer_mirror_delete_to(cursor, mirror);
588 * If the record exists only the delete_tid may be updated.
590 * If the record does not exist we can create it only if the
591 * create_tid is not too old. If the create_tid is too old
592 * it may have already been destroyed on the slave from pruning.
594 * Note that mirror operations are effectively as-of operations
595 * and delete_tid can be 0 for mirroring purposes even if it is
596 * not actually 0 at the originator.
598 * These functions can return EDEADLK
600 cursor->key_beg = mrec->leaf.base;
601 cursor->flags |= HAMMER_CURSOR_BACKEND;
602 cursor->flags &= ~HAMMER_CURSOR_INSERT;
603 error = hammer_btree_lookup(cursor);
605 if (error == 0 && hammer_mirror_check(cursor, mrec)) {
606 error = hammer_mirror_update(cursor, mrec);
607 } else if (error == ENOENT) {
608 if (mrec->leaf.base.create_tid >= mirror->tid_beg)
609 error = hammer_mirror_write(cursor, mrec, uptr);
613 if (error == 0 || error == EALREADY)
614 mirror->key_cur = mrec->leaf.base;
619 * This works like write_rec but no write or update is necessary,
620 * and no data payload is included so we couldn't do a write even
623 * We must still iterate for deletions, and we can validate the
624 * record header which is a good way to test for corrupted mirror
627 * mirror->key_cur must be carefully set when we succeed in processing
632 hammer_ioc_mirror_write_pass(hammer_cursor_t cursor,
633 struct hammer_ioc_mrecord_rec *mrec,
634 struct hammer_ioc_mirror_rw *mirror,
635 u_int32_t localization)
637 hammer_transaction_t trans;
641 trans = cursor->trans;
642 rec_crc = crc32(mrec, sizeof(*mrec));
645 * Re-localize for target. Relocalization of data is handled
646 * by hammer_mirror_write().
648 mrec->leaf.base.localization &= HAMMER_LOCALIZE_MASK;
649 mrec->leaf.base.localization += localization;
652 * Delete records through until we reach (non-inclusively) the
655 cursor->key_end = mrec->leaf.base;
656 cursor->flags &= ~HAMMER_CURSOR_END_INCLUSIVE;
657 cursor->flags |= HAMMER_CURSOR_BACKEND;
659 error = hammer_mirror_delete_to(cursor, mirror);
662 * Locate the record and get past it by setting ATEDISK. Perform
663 * any necessary deletions. We have no data payload and cannot
664 * create a new record.
667 mirror->key_cur = mrec->leaf.base;
668 cursor->key_beg = mrec->leaf.base;
669 cursor->flags |= HAMMER_CURSOR_BACKEND;
670 cursor->flags &= ~HAMMER_CURSOR_INSERT;
671 error = hammer_btree_lookup(cursor);
673 if (hammer_mirror_check(cursor, mrec))
674 error = hammer_mirror_update(cursor, mrec);
675 cursor->flags |= HAMMER_CURSOR_ATEDISK;
677 cursor->flags &= ~HAMMER_CURSOR_ATEDISK;
686 * As part of the mirror write we iterate across swaths of records
687 * on the target which no longer exist on the source, and mark them
690 * The caller has indexed the cursor and set up key_end. We iterate
691 * through to key_end.
693 * There is an edge case where the master has deleted a record whos
694 * create_tid exactly matches our end_tid. We cannot delete this
695 * record on the slave yet because we cannot assign delete_tid == create_tid.
696 * The deletion should be picked up on the next sequence since in order
697 * to have been deleted on the master a transaction must have occured with
698 * a TID greater then the create_tid of the record.
702 hammer_mirror_delete_to(hammer_cursor_t cursor,
703 struct hammer_ioc_mirror_rw *mirror)
705 hammer_btree_leaf_elm_t elm;
708 error = hammer_btree_iterate(cursor);
710 elm = &cursor->node->ondisk->elms[cursor->index].leaf;
711 KKASSERT(elm->base.btype == HAMMER_BTREE_TYPE_RECORD);
712 cursor->flags |= HAMMER_CURSOR_ATEDISK;
713 if (elm->base.delete_tid == 0 &&
714 elm->base.create_tid != mirror->tid_end) {
715 error = hammer_delete_at_cursor(cursor,
716 HAMMER_DELETE_ADJUST,
722 error = hammer_btree_iterate(cursor);
730 * Check whether an update is needed in the case where a match already
731 * exists on the target. The only type of update allowed in this case
732 * is an update of the delete_tid.
734 * Return non-zero if the update should proceed.
738 hammer_mirror_check(hammer_cursor_t cursor, struct hammer_ioc_mrecord_rec *mrec)
740 hammer_btree_leaf_elm_t leaf = cursor->leaf;
742 if (leaf->base.delete_tid != mrec->leaf.base.delete_tid) {
743 if (mrec->leaf.base.delete_tid != 0)
750 * Update a record in-place. Only the delete_tid can change, and
751 * only from zero to non-zero.
755 hammer_mirror_update(hammer_cursor_t cursor,
756 struct hammer_ioc_mrecord_rec *mrec)
761 * This case shouldn't occur.
763 if (mrec->leaf.base.delete_tid == 0)
767 * Mark the record deleted on the mirror target.
769 error = hammer_delete_at_cursor(cursor, HAMMER_DELETE_ADJUST,
770 mrec->leaf.base.delete_tid,
771 mrec->leaf.delete_ts,
773 cursor->flags |= HAMMER_CURSOR_ATEDISK;
778 * Write out a new record.
782 hammer_mirror_write(hammer_cursor_t cursor,
783 struct hammer_ioc_mrecord_rec *mrec,
786 hammer_transaction_t trans;
787 hammer_buffer_t data_buffer;
788 hammer_off_t ndata_offset;
789 hammer_tid_t high_tid;
794 trans = cursor->trans;
798 * Get the sync lock so the whole mess is atomic
800 hammer_sync_lock_sh(trans);
803 * Allocate and adjust data
805 if (mrec->leaf.data_len && mrec->leaf.data_offset) {
806 ndata = hammer_alloc_data(trans, mrec->leaf.data_len,
807 mrec->leaf.base.rec_type,
808 &ndata_offset, &data_buffer,
812 mrec->leaf.data_offset = ndata_offset;
813 hammer_modify_buffer(trans, data_buffer, NULL, 0);
814 error = copyin(udata, ndata, mrec->leaf.data_len);
816 if (hammer_crc_test_leaf(ndata, &mrec->leaf) == 0) {
817 kprintf("data crc mismatch on pipe\n");
820 error = hammer_mirror_localize_data(
824 hammer_modify_buffer_done(data_buffer);
826 mrec->leaf.data_offset = 0;
834 * Do the insertion. This can fail with a EDEADLK or EALREADY
836 cursor->flags |= HAMMER_CURSOR_INSERT;
837 error = hammer_btree_lookup(cursor);
838 if (error != ENOENT) {
844 error = hammer_btree_insert(cursor, &mrec->leaf, &doprop);
847 * Cursor is left on the current element, we want to skip it now.
849 cursor->flags |= HAMMER_CURSOR_ATEDISK;
850 cursor->flags &= ~HAMMER_CURSOR_INSERT;
853 * Track a count of active inodes.
856 mrec->leaf.base.rec_type == HAMMER_RECTYPE_INODE &&
857 mrec->leaf.base.delete_tid == 0) {
858 hammer_modify_volume_field(trans,
861 ++trans->hmp->rootvol->ondisk->vol0_stat_inodes;
862 hammer_modify_volume_done(trans->rootvol);
866 * vol0_next_tid must track the highest TID stored in the filesystem.
867 * We do not need to generate undo for this update.
869 high_tid = mrec->leaf.base.create_tid;
870 if (high_tid < mrec->leaf.base.delete_tid)
871 high_tid = mrec->leaf.base.delete_tid;
872 if (trans->rootvol->ondisk->vol0_next_tid < high_tid) {
873 hammer_modify_volume(trans, trans->rootvol, NULL, 0);
874 trans->rootvol->ondisk->vol0_next_tid = high_tid;
875 hammer_modify_volume_done(trans->rootvol);
878 if (error == 0 && doprop)
879 hammer_btree_do_propagation(cursor, NULL, &mrec->leaf);
885 if (error && mrec->leaf.data_offset) {
886 hammer_blockmap_free(cursor->trans,
887 mrec->leaf.data_offset,
888 mrec->leaf.data_len);
890 hammer_sync_unlock(trans);
892 hammer_rel_buffer(data_buffer, 0);
897 * Localize the data payload. Directory entries may need their
898 * localization adjusted.
900 * PFS directory entries must be skipped entirely (return EALREADY).
904 hammer_mirror_localize_data(hammer_data_ondisk_t data,
905 hammer_btree_leaf_elm_t leaf)
907 u_int32_t localization;
909 if (leaf->base.rec_type == HAMMER_RECTYPE_DIRENTRY) {
910 if (data->entry.obj_id == HAMMER_OBJID_ROOT)
912 localization = leaf->base.localization &
913 HAMMER_LOCALIZE_PSEUDOFS_MASK;
914 if (data->entry.localization != localization) {
915 data->entry.localization = localization;
916 hammer_crc_set_leaf(data, leaf);