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