hammer - Fix core dump during remote termination of mirror-stream
[dragonfly.git] / sbin / hammer / cmd_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
35 #include "hammer.h"
36
37 #define SERIALBUF_SIZE  (512 * 1024)
38
39 typedef struct histogram {
40         hammer_tid_t    tid;
41         u_int64_t       bytes;
42 } *histogram_t;
43
44 static int read_mrecords(int fd, char *buf, u_int size,
45                          hammer_ioc_mrecord_head_t pickup);
46 static int generate_histogram(int fd, const char *filesystem,
47                          histogram_t *histogram_ary,
48                          struct hammer_ioc_mirror_rw *mirror_base,
49                          int *repeatp);
50 static hammer_ioc_mrecord_any_t read_mrecord(int fdin, int *errorp,
51                          hammer_ioc_mrecord_head_t pickup);
52 static void write_mrecord(int fdout, u_int32_t type,
53                          hammer_ioc_mrecord_any_t mrec, int bytes);
54 static void generate_mrec_header(int fd, int pfs_id,
55                          union hammer_ioc_mrecord_any *mrec_tmp);
56 static int validate_mrec_header(int fd, int fdin, int is_target, int pfs_id,
57                          struct hammer_ioc_mrecord_head *pickup,
58                          hammer_tid_t *tid_begp, hammer_tid_t *tid_endp);
59 static void update_pfs_snapshot(int fd, hammer_tid_t snapshot_tid, int pfs_id);
60 static ssize_t writebw(int fd, const void *buf, size_t nbytes,
61                         u_int64_t *bwcount, struct timeval *tv1);
62 static int getyn(void);
63 static void mirror_usage(int code);
64
65 /*
66  * Generate a mirroring data stream from the specific source over the
67  * entire key range, but restricted to the specified transaction range.
68  *
69  * The HAMMER VFS does most of the work, we add a few new mrecord
70  * types to negotiate the TID ranges and verify that the entire
71  * stream made it to the destination.
72  *
73  * streaming will be 0 for mirror-read, 1 for mirror-stream.  The code will
74  * set up a fake value of -1 when running the histogram for mirror-read.
75  */
76 void
77 hammer_cmd_mirror_read(char **av, int ac, int streaming)
78 {
79         struct hammer_ioc_mirror_rw mirror;
80         struct hammer_ioc_pseudofs_rw pfs;
81         union hammer_ioc_mrecord_any mrec_tmp;
82         struct hammer_ioc_mrecord_head pickup;
83         hammer_ioc_mrecord_any_t mrec;
84         hammer_tid_t sync_tid;
85         histogram_t histogram_ary;
86         const char *filesystem;
87         char *buf = malloc(SERIALBUF_SIZE);
88         int interrupted = 0;
89         int error;
90         int fd;
91         int n;
92         int didwork;
93         int histogram;
94         int histindex;
95         int histmax;
96         int repeat = 0;
97         int sameline;
98         int64_t total_bytes;
99         time_t base_t = time(NULL);
100         struct timeval bwtv;
101         u_int64_t bwcount;
102         u_int64_t estbytes;
103
104         if (ac == 0 || ac > 2)
105                 mirror_usage(1);
106         filesystem = av[0];
107
108         pickup.signature = 0;
109         pickup.type = 0;
110         histogram = 0;
111         histindex = 0;
112         histmax = 0;
113         histogram_ary = NULL;
114         sameline = 0;
115
116 again:
117         bzero(&mirror, sizeof(mirror));
118         hammer_key_beg_init(&mirror.key_beg);
119         hammer_key_end_init(&mirror.key_end);
120
121         fd = getpfs(&pfs, filesystem);
122
123         if (streaming >= 0 && VerboseOpt && VerboseOpt < 2) {
124                 fprintf(stderr, "%cRunning  \b\b", (sameline ? '\r' : '\n'));
125                 fflush(stderr);
126                 sameline = 1;
127         }
128         sameline = 1;
129         total_bytes = 0;
130         gettimeofday(&bwtv, NULL);
131         bwcount = 0;
132
133         /*
134          * Send initial header for the purpose of determining the
135          * shared-uuid.
136          */
137         generate_mrec_header(fd, pfs.pfs_id, &mrec_tmp);
138         write_mrecord(1, HAMMER_MREC_TYPE_PFSD,
139                       &mrec_tmp, sizeof(mrec_tmp.pfs));
140
141         /*
142          * In 2-way mode the target will send us a PFS info packet
143          * first.  Use the target's current snapshot TID as our default
144          * begin TID.
145          */
146         if (TwoWayPipeOpt) {
147                 mirror.tid_beg = 0;
148                 n = validate_mrec_header(fd, 0, 0, pfs.pfs_id, &pickup,
149                                          NULL, &mirror.tid_beg);
150                 if (n < 0) {    /* got TERM record */
151                         relpfs(fd, &pfs);
152                         return;
153                 }
154                 ++mirror.tid_beg;
155         } else if (streaming && histogram) {
156                 mirror.tid_beg = histogram_ary[histindex].tid + 1;
157         } else {
158                 mirror.tid_beg = 0;
159         }
160
161         /*
162          * Write out the PFS header, tid_beg will be updated if our PFS
163          * has a larger begin sync.  tid_end is set to the latest source
164          * TID whos flush cycle has completed.
165          */
166         generate_mrec_header(fd, pfs.pfs_id, &mrec_tmp);
167         if (mirror.tid_beg < mrec_tmp.pfs.pfsd.sync_beg_tid)
168                 mirror.tid_beg = mrec_tmp.pfs.pfsd.sync_beg_tid;
169         mirror.tid_end = mrec_tmp.pfs.pfsd.sync_end_tid;
170         mirror.ubuf = buf;
171         mirror.size = SERIALBUF_SIZE;
172         mirror.pfs_id = pfs.pfs_id;
173         mirror.shared_uuid = pfs.ondisk->shared_uuid;
174
175         /*
176          * XXX If the histogram is exhausted and the TID delta is large
177          *     the stream might have been offline for a while and is
178          *     now picking it up again.  Do another histogram.
179          */
180 #if 0
181         if (streaming && histogram && histindex == histend) {
182                 if (mirror.tid_end - mirror.tid_beg > BULK_MINIMUM)
183                         histogram = 0;
184         }
185 #endif
186
187         /*
188          * Initial bulk startup control, try to do some incremental
189          * mirroring in order to allow the stream to be killed and
190          * restarted without having to start over.
191          */
192         if (histogram == 0 && BulkOpt == 0) {
193                 if (VerboseOpt && repeat == 0) {
194                         fprintf(stderr, "\n");
195                         sameline = 0;
196                 }
197                 histmax = generate_histogram(fd, filesystem,
198                                              &histogram_ary, &mirror,
199                                              &repeat);
200                 histindex = 0;
201                 histogram = 1;
202
203                 /*
204                  * Just stream the histogram, then stop
205                  */
206                 if (streaming == 0)
207                         streaming = -1;
208         }
209
210         if (streaming && histogram) {
211                 ++histindex;
212                 mirror.tid_end = histogram_ary[histindex].tid;
213                 estbytes = histogram_ary[histindex-1].bytes;
214                 mrec_tmp.pfs.pfsd.sync_end_tid = mirror.tid_end;
215         } else {
216                 estbytes = 0;
217         }
218
219         write_mrecord(1, HAMMER_MREC_TYPE_PFSD,
220                       &mrec_tmp, sizeof(mrec_tmp.pfs));
221
222         /*
223          * A cycle file overrides the beginning TID only if we are
224          * not operating in two-way or histogram mode.
225          */
226         if (TwoWayPipeOpt == 0 && histogram == 0) {
227                 hammer_get_cycle(&mirror.key_beg, &mirror.tid_beg);
228         }
229
230         /*
231          * An additional argument overrides the beginning TID regardless
232          * of what mode we are in.  This is not recommending if operating
233          * in two-way mode.
234          */
235         if (ac == 2)
236                 mirror.tid_beg = strtoull(av[1], NULL, 0);
237
238         if (streaming == 0 || VerboseOpt >= 2) {
239                 fprintf(stderr,
240                         "Mirror-read: Mirror %016jx to %016jx",
241                         (uintmax_t)mirror.tid_beg, (uintmax_t)mirror.tid_end);
242                 if (histogram)
243                         fprintf(stderr, " (bulk= %ju)", (uintmax_t)estbytes);
244                 fprintf(stderr, "\n");
245                 fflush(stderr);
246         }
247         if (mirror.key_beg.obj_id != (int64_t)HAMMER_MIN_OBJID) {
248                 fprintf(stderr, "Mirror-read: Resuming at object %016jx\n",
249                         (uintmax_t)mirror.key_beg.obj_id);
250         }
251
252         /*
253          * Nothing to do if begin equals end.
254          */
255         if (mirror.tid_beg >= mirror.tid_end) {
256                 if (streaming == 0 || VerboseOpt >= 2)
257                         fprintf(stderr, "Mirror-read: No work to do\n");
258                 sleep(DelayOpt);
259                 didwork = 0;
260                 histogram = 0;
261                 goto done;
262         }
263         didwork = 1;
264
265         /*
266          * Write out bulk records
267          */
268         mirror.ubuf = buf;
269         mirror.size = SERIALBUF_SIZE;
270
271         do {
272                 mirror.count = 0;
273                 mirror.pfs_id = pfs.pfs_id;
274                 mirror.shared_uuid = pfs.ondisk->shared_uuid;
275                 if (ioctl(fd, HAMMERIOC_MIRROR_READ, &mirror) < 0) {
276                         fprintf(stderr, "Mirror-read %s failed: %s\n",
277                                 filesystem, strerror(errno));
278                         exit(1);
279                 }
280                 if (mirror.head.flags & HAMMER_IOC_HEAD_ERROR) {
281                         fprintf(stderr,
282                                 "Mirror-read %s fatal error %d\n",
283                                 filesystem, mirror.head.error);
284                         exit(1);
285                 }
286                 if (mirror.count) {
287                         if (BandwidthOpt) {
288                                 n = writebw(1, mirror.ubuf, mirror.count,
289                                             &bwcount, &bwtv);
290                         } else {
291                                 n = write(1, mirror.ubuf, mirror.count);
292                         }
293                         if (n != mirror.count) {
294                                 fprintf(stderr, "Mirror-read %s failed: "
295                                                 "short write\n",
296                                 filesystem);
297                                 exit(1);
298                         }
299                 }
300                 total_bytes += mirror.count;
301                 if (streaming && VerboseOpt) {
302                         fprintf(stderr,
303                                 "\rscan obj=%016jx tids=%016jx:%016jx %11jd",
304                                 (uintmax_t)mirror.key_cur.obj_id,
305                                 (uintmax_t)mirror.tid_beg,
306                                 (uintmax_t)mirror.tid_end,
307                                 (intmax_t)total_bytes);
308                         fflush(stderr);
309                         sameline = 0;
310                 }
311                 mirror.key_beg = mirror.key_cur;
312
313                 /*
314                  * Deal with time limit option
315                  */
316                 if (TimeoutOpt &&
317                     (unsigned)(time(NULL) - base_t) > (unsigned)TimeoutOpt) {
318                         fprintf(stderr,
319                                 "Mirror-read %s interrupted by timer at"
320                                 " %016jx\n",
321                                 filesystem,
322                                 (uintmax_t)mirror.key_cur.obj_id);
323                         interrupted = 1;
324                         break;
325                 }
326         } while (mirror.count != 0);
327
328 done:
329         if (streaming && VerboseOpt && sameline == 0) {
330                 fprintf(stderr, "\n");
331                 fflush(stderr);
332                 sameline = 1;
333         }
334
335         /*
336          * Write out the termination sync record - only if not interrupted
337          */
338         if (interrupted == 0) {
339                 if (didwork) {
340                         write_mrecord(1, HAMMER_MREC_TYPE_SYNC,
341                                       &mrec_tmp, sizeof(mrec_tmp.sync));
342                 } else {
343                         write_mrecord(1, HAMMER_MREC_TYPE_IDLE,
344                                       &mrec_tmp, sizeof(mrec_tmp.sync));
345                 }
346         }
347
348         /*
349          * If the -2 option was given (automatic when doing mirror-copy),
350          * a two-way pipe is assumed and we expect a response mrec from
351          * the target.
352          */
353         if (TwoWayPipeOpt) {
354                 mrec = read_mrecord(0, &error, &pickup);
355                 if (mrec == NULL ||
356                     mrec->head.type != HAMMER_MREC_TYPE_UPDATE ||
357                     mrec->head.rec_size != sizeof(mrec->update)) {
358                         fprintf(stderr, "mirror_read: Did not get final "
359                                         "acknowledgement packet from target\n");
360                         exit(1);
361                 }
362                 if (interrupted) {
363                         if (CyclePath) {
364                                 hammer_set_cycle(&mirror.key_cur,
365                                                  mirror.tid_beg);
366                                 fprintf(stderr, "Cyclefile %s updated for "
367                                         "continuation\n", CyclePath);
368                         }
369                 } else {
370                         sync_tid = mrec->update.tid;
371                         if (CyclePath) {
372                                 hammer_key_beg_init(&mirror.key_beg);
373                                 hammer_set_cycle(&mirror.key_beg, sync_tid);
374                                 fprintf(stderr,
375                                         "Cyclefile %s updated to 0x%016jx\n",
376                                         CyclePath, (uintmax_t)sync_tid);
377                         }
378                 }
379         } else if (CyclePath) {
380                 /* NOTE! mirror.tid_beg cannot be updated */
381                 fprintf(stderr, "Warning: cycle file (-c option) cannot be "
382                                 "fully updated unless you use mirror-copy\n");
383                 hammer_set_cycle(&mirror.key_beg, mirror.tid_beg);
384         }
385         if (streaming && interrupted == 0) {
386                 time_t t1 = time(NULL);
387                 time_t t2;
388
389                 /*
390                  * Try to break down large bulk transfers into smaller ones
391                  * so it can sync the transaction id on the slave.  This
392                  * way if we get interrupted a restart doesn't have to
393                  * start from scratch.
394                  */
395                 if (streaming && histogram) {
396                         if (histindex != histmax) {
397                                 if (VerboseOpt && VerboseOpt < 2 &&
398                                     streaming >= 0) {
399                                         fprintf(stderr, " (bulk incremental)");
400                                 }
401                                 relpfs(fd, &pfs);
402                                 goto again;
403                         }
404                 }
405
406                 if (VerboseOpt && streaming >= 0) {
407                         fprintf(stderr, " W");
408                         fflush(stderr);
409                 }
410                 pfs.ondisk->sync_end_tid = mirror.tid_end;
411                 if (streaming < 0) {
412                         /*
413                          * Fake streaming mode when using a histogram to
414                          * break up a mirror-read, do not wait on source.
415                          */
416                         streaming = 0;
417                 } else if (ioctl(fd, HAMMERIOC_WAI_PSEUDOFS, &pfs) < 0) {
418                         fprintf(stderr, "Mirror-read %s: cannot stream: %s\n",
419                                 filesystem, strerror(errno));
420                 } else {
421                         t2 = time(NULL) - t1;
422                         if (t2 >= 0 && t2 < DelayOpt) {
423                                 if (VerboseOpt) {
424                                         fprintf(stderr, "\bD");
425                                         fflush(stderr);
426                                 }
427                                 sleep(DelayOpt - t2);
428                         }
429                         if (VerboseOpt) {
430                                 fprintf(stderr, "\b ");
431                                 fflush(stderr);
432                         }
433                         relpfs(fd, &pfs);
434                         goto again;
435                 }
436         }
437         write_mrecord(1, HAMMER_MREC_TYPE_TERM,
438                       &mrec_tmp, sizeof(mrec_tmp.sync));
439         relpfs(fd, &pfs);
440         fprintf(stderr, "Mirror-read %s succeeded\n", filesystem);
441 }
442
443 /*
444  * What we are trying to do here is figure out how much data is
445  * going to be sent for the TID range and to break the TID range
446  * down into reasonably-sized slices (from the point of view of
447  * data sent) so a lost connection can restart at a reasonable
448  * place and not all the way back at the beginning.
449  *
450  * An entry's TID serves as the end_tid for the prior entry
451  * So we have to offset the calculation by 1 so that TID falls into
452  * the previous entry when populating entries.
453  *
454  * Because the transaction id space is bursty we need a relatively
455  * large number of buckets (like a million) to do a reasonable job
456  * for things like an initial bulk mirrors on a very large filesystem.
457  */
458 #define HIST_COUNT      (1024 * 1024)
459
460 static int
461 generate_histogram(int fd, const char *filesystem,
462                    histogram_t *histogram_ary,
463                    struct hammer_ioc_mirror_rw *mirror_base,
464                    int *repeatp)
465 {
466         struct hammer_ioc_mirror_rw mirror;
467         union hammer_ioc_mrecord_any *mrec;
468         hammer_tid_t tid_beg;
469         hammer_tid_t tid_end;
470         hammer_tid_t tid;
471         hammer_tid_t tidx;
472         u_int64_t *tid_bytes;
473         u_int64_t total;
474         u_int64_t accum;
475         int i;
476         int res;
477         int off;
478         int len;
479
480         mirror = *mirror_base;
481         tid_beg = mirror.tid_beg;
482         tid_end = mirror.tid_end;
483         mirror.head.flags |= HAMMER_IOC_MIRROR_NODATA;
484
485         if (*histogram_ary == NULL) {
486                 *histogram_ary = malloc(sizeof(struct histogram) *
487                                         (HIST_COUNT + 2));
488         }
489         if (tid_beg >= tid_end)
490                 return(0);
491
492         /* needs 2 extra */
493         tid_bytes = malloc(sizeof(*tid_bytes) * (HIST_COUNT + 2));
494         bzero(tid_bytes, sizeof(*tid_bytes) * (HIST_COUNT + 2));
495
496         if (*repeatp == 0) {
497                 fprintf(stderr, "Prescan to break up bulk transfer");
498                 if (VerboseOpt > 1)
499                         fprintf(stderr, " (%juMB chunks)",
500                                 (uintmax_t)(SplitupOpt / (1024 * 1024)));
501                 fprintf(stderr, "\n");
502         }
503
504         /*
505          * Note: (tid_beg,tid_end), range is inclusive of both beg & end.
506          *
507          * Note: Estimates can be off when the mirror is way behind due
508          *       to skips.
509          */
510         total = 0;
511         accum = 0;
512         for (;;) {
513                 mirror.count = 0;
514                 if (ioctl(fd, HAMMERIOC_MIRROR_READ, &mirror) < 0) {
515                         fprintf(stderr, "Mirror-read %s failed: %s\n",
516                                 filesystem, strerror(errno));
517                         exit(1);
518                 }
519                 if (mirror.head.flags & HAMMER_IOC_HEAD_ERROR) {
520                         fprintf(stderr,
521                                 "Mirror-read %s fatal error %d\n",
522                                 filesystem, mirror.head.error);
523                         exit(1);
524                 }
525                 for (off = 0;
526                      off < mirror.count;
527                      off += HAMMER_HEAD_DOALIGN(mrec->head.rec_size)
528                 ) {
529                         mrec = (void *)((char *)mirror.ubuf + off);
530
531                         /*
532                          * We only care about general RECs and PASS
533                          * records.  We ignore SKIPs.
534                          */
535                         switch (mrec->head.type & HAMMER_MRECF_TYPE_LOMASK) {
536                         case HAMMER_MREC_TYPE_REC:
537                         case HAMMER_MREC_TYPE_PASS:
538                                 break;
539                         default:
540                                 continue;
541                         }
542
543                         /*
544                          * Calculate for two indices, create_tid and
545                          * delete_tid.  Record data only applies to
546                          * the create_tid.
547                          *
548                          * When tid is exactly on the boundary it really
549                          * belongs to the previous entry because scans
550                          * are inclusive of the ending entry.
551                          */
552                         tid = mrec->rec.leaf.base.delete_tid;
553                         if (tid && tid >= tid_beg && tid <= tid_end) {
554                                 len = HAMMER_HEAD_DOALIGN(mrec->head.rec_size);
555                                 if (mrec->head.type ==
556                                     HAMMER_MREC_TYPE_REC) {
557                                         len -= HAMMER_HEAD_DOALIGN(
558                                                     mrec->rec.leaf.data_len);
559                                         assert(len > 0);
560                                 }
561                                 i = (tid - tid_beg) * HIST_COUNT /
562                                     (tid_end - tid_beg);
563                                 tidx = tid_beg + i * (tid_end - tid_beg) /
564                                                  HIST_COUNT;
565                                 if (tid == tidx && i)
566                                         --i;
567                                 assert(i >= 0 && i < HIST_COUNT);
568                                 tid_bytes[i] += len;
569                                 total += len;
570                                 accum += len;
571                         }
572
573                         tid = mrec->rec.leaf.base.create_tid;
574                         if (tid && tid >= tid_beg && tid <= tid_end) {
575                                 len = HAMMER_HEAD_DOALIGN(mrec->head.rec_size);
576                                 if (mrec->head.type ==
577                                     HAMMER_MREC_TYPE_REC_NODATA) {
578                                         len += HAMMER_HEAD_DOALIGN(
579                                                     mrec->rec.leaf.data_len);
580                                 }
581                                 i = (tid - tid_beg) * HIST_COUNT /
582                                     (tid_end - tid_beg);
583                                 tidx = tid_beg + i * (tid_end - tid_beg) /
584                                                  HIST_COUNT;
585                                 if (tid == tidx && i)
586                                         --i;
587                                 assert(i >= 0 && i < HIST_COUNT);
588                                 tid_bytes[i] += len;
589                                 total += len;
590                                 accum += len;
591                         }
592                 }
593                 if (VerboseOpt > 1) {
594                         if (*repeatp == 0 && accum > SplitupOpt) {
595                                 fprintf(stderr, ".");
596                                 fflush(stderr);
597                                 accum = 0;
598                         }
599                 }
600                 if (mirror.count == 0)
601                         break;
602                 mirror.key_beg = mirror.key_cur;
603         }
604
605         /*
606          * Reduce to SplitupOpt (default 4GB) chunks.  This code may
607          * use up to two additional elements.  Do the array in-place.
608          *
609          * Inefficient degenerate cases can occur if we do not accumulate
610          * at least the requested split amount, so error on the side of
611          * going over a bit.
612          */
613         res = 0;
614         (*histogram_ary)[res].tid = tid_beg;
615         (*histogram_ary)[res].bytes = tid_bytes[0];
616         for (i = 1; i < HIST_COUNT; ++i) {
617                 if ((*histogram_ary)[res].bytes >= SplitupOpt) {
618                         ++res;
619                         (*histogram_ary)[res].tid = tid_beg +
620                                         i * (tid_end - tid_beg) /
621                                         HIST_COUNT;
622                         (*histogram_ary)[res].bytes = 0;
623
624                 }
625                 (*histogram_ary)[res].bytes += tid_bytes[i];
626         }
627         ++res;
628         (*histogram_ary)[res].tid = tid_end;
629         (*histogram_ary)[res].bytes = -1;
630
631         if (*repeatp == 0) {
632                 if (VerboseOpt > 1)
633                         fprintf(stderr, "\n");  /* newline after ... */
634                 fprintf(stderr, "Prescan %d chunks, total %ju MBytes (",
635                         res, (uintmax_t)total / (1024 * 1024));
636                 for (i = 0; i < res && i < 3; ++i) {
637                         if (i)
638                                 fprintf(stderr, ", ");
639                         fprintf(stderr, "%ju",
640                                 (uintmax_t)(*histogram_ary)[i].bytes);
641                 }
642                 if (i < res)
643                         fprintf(stderr, ", ...");
644                 fprintf(stderr, ")\n");
645         }
646         assert(res <= HIST_COUNT);
647         *repeatp = 1;
648
649         free(tid_bytes);
650         return(res);
651 }
652
653 static void
654 create_pfs(const char *filesystem, uuid_t *s_uuid)
655 {
656         if (ForceYesOpt == 1) {
657                 fprintf(stderr, "PFS slave %s does not exist. "
658                         "Auto create new slave PFS!\n", filesystem);
659
660         } else {
661                 fprintf(stderr, "PFS slave %s does not exist.\n"
662                         "Do you want to create a new slave PFS? (yes|no) ",
663                         filesystem);
664                 fflush(stderr);
665                 if (getyn() != 1) {
666                         fprintf(stderr, "Aborting operation\n");
667                         exit(1);
668                 }
669         }
670
671         u_int32_t status;
672         char *shared_uuid = NULL;
673         uuid_to_string(s_uuid, &shared_uuid, &status);
674
675         char *cmd = NULL;
676         asprintf(&cmd, "/sbin/hammer pfs-slave '%s' shared-uuid=%s 1>&2",
677                  filesystem, shared_uuid);
678         free(shared_uuid);
679
680         if (cmd == NULL) {
681                 fprintf(stderr, "Failed to alloc memory\n");
682                 exit(1);
683         }
684         if (system(cmd) != 0) {
685                 fprintf(stderr, "Failed to create PFS\n");
686         }
687         free(cmd);
688 }
689
690 /*
691  * Pipe the mirroring data stream on stdin to the HAMMER VFS, adding
692  * some additional packet types to negotiate TID ranges and to verify
693  * completion.  The HAMMER VFS does most of the work.
694  *
695  * It is important to note that the mirror.key_{beg,end} range must
696  * match the ranged used by the original.  For now both sides use
697  * range the entire key space.
698  *
699  * It is even more important that the records in the stream conform
700  * to the TID range also supplied in the stream.  The HAMMER VFS will
701  * use the REC, PASS, and SKIP record types to track the portions of
702  * the B-Tree being scanned in order to be able to proactively delete
703  * records on the target within those active areas that are not mentioned
704  * by the source.
705  *
706  * The mirror.key_cur field is used by the VFS to do this tracking.  It
707  * must be initialized to key_beg but then is persistently updated by
708  * the HAMMER VFS on each successive ioctl() call.  If you blow up this
709  * field you will blow up the mirror target, possibly to the point of
710  * deleting everything.  As a safety measure the HAMMER VFS simply marks
711  * the records that the source has destroyed as deleted on the target,
712  * and normal pruning operations will deal with their final disposition
713  * at some later time.
714  */
715 void
716 hammer_cmd_mirror_write(char **av, int ac)
717 {
718         struct hammer_ioc_mirror_rw mirror;
719         const char *filesystem;
720         char *buf = malloc(SERIALBUF_SIZE);
721         struct hammer_ioc_pseudofs_rw pfs;
722         struct hammer_ioc_mrecord_head pickup;
723         struct hammer_ioc_synctid synctid;
724         union hammer_ioc_mrecord_any mrec_tmp;
725         hammer_ioc_mrecord_any_t mrec;
726         struct stat st;
727         int error;
728         int fd;
729         int n;
730
731         if (ac != 1)
732                 mirror_usage(1);
733         filesystem = av[0];
734
735         pickup.signature = 0;
736         pickup.type = 0;
737
738 again:
739         bzero(&mirror, sizeof(mirror));
740         hammer_key_beg_init(&mirror.key_beg);
741         hammer_key_end_init(&mirror.key_end);
742         mirror.key_end = mirror.key_beg;
743
744         /*
745          * Read initial packet
746          */
747         mrec = read_mrecord(0, &error, &pickup);
748         if (mrec == NULL) {
749                 if (error == 0)
750                         fprintf(stderr, "validate_mrec_header: short read\n");
751                 exit(1);
752         }
753         /*
754          * Validate packet
755          */
756         if (mrec->head.type == HAMMER_MREC_TYPE_TERM) {
757                 return;
758         }
759         if (mrec->head.type != HAMMER_MREC_TYPE_PFSD) {
760                 fprintf(stderr, "validate_mrec_header: did not get expected "
761                                 "PFSD record type\n");
762                 exit(1);
763         }
764         if (mrec->head.rec_size != sizeof(mrec->pfs)) {
765                 fprintf(stderr, "validate_mrec_header: unexpected payload "
766                                 "size\n");
767                 exit(1);
768         }
769
770         /*
771          * Create slave PFS if it doesn't yet exist
772          */
773         if (lstat(filesystem, &st) != 0) {
774                 create_pfs(filesystem, &mrec->pfs.pfsd.shared_uuid);
775         }
776         free(mrec);
777         mrec = NULL;
778
779         fd = getpfs(&pfs, filesystem);
780
781         /*
782          * In two-way mode the target writes out a PFS packet first.
783          * The source uses our tid_end as its tid_beg by default,
784          * picking up where it left off.
785          */
786         mirror.tid_beg = 0;
787         if (TwoWayPipeOpt) {
788                 generate_mrec_header(fd, pfs.pfs_id, &mrec_tmp);
789                 if (mirror.tid_beg < mrec_tmp.pfs.pfsd.sync_beg_tid)
790                         mirror.tid_beg = mrec_tmp.pfs.pfsd.sync_beg_tid;
791                 mirror.tid_end = mrec_tmp.pfs.pfsd.sync_end_tid;
792                 write_mrecord(1, HAMMER_MREC_TYPE_PFSD,
793                               &mrec_tmp, sizeof(mrec_tmp.pfs));
794         }
795
796         /*
797          * Read and process the PFS header.  The source informs us of
798          * the TID range the stream represents.
799          */
800         n = validate_mrec_header(fd, 0, 1, pfs.pfs_id, &pickup,
801                                  &mirror.tid_beg, &mirror.tid_end);
802         if (n < 0) {    /* got TERM record */
803                 relpfs(fd, &pfs);
804                 return;
805         }
806
807         mirror.ubuf = buf;
808         mirror.size = SERIALBUF_SIZE;
809
810         /*
811          * Read and process bulk records (REC, PASS, and SKIP types).
812          *
813          * On your life, do NOT mess with mirror.key_cur or your mirror
814          * target may become history.
815          */
816         for (;;) {
817                 mirror.count = 0;
818                 mirror.pfs_id = pfs.pfs_id;
819                 mirror.shared_uuid = pfs.ondisk->shared_uuid;
820                 mirror.size = read_mrecords(0, buf, SERIALBUF_SIZE, &pickup);
821                 if (mirror.size <= 0)
822                         break;
823                 if (ioctl(fd, HAMMERIOC_MIRROR_WRITE, &mirror) < 0) {
824                         fprintf(stderr, "Mirror-write %s failed: %s\n",
825                                 filesystem, strerror(errno));
826                         exit(1);
827                 }
828                 if (mirror.head.flags & HAMMER_IOC_HEAD_ERROR) {
829                         fprintf(stderr,
830                                 "Mirror-write %s fatal error %d\n",
831                                 filesystem, mirror.head.error);
832                         exit(1);
833                 }
834 #if 0
835                 if (mirror.head.flags & HAMMER_IOC_HEAD_INTR) {
836                         fprintf(stderr,
837                                 "Mirror-write %s interrupted by timer at"
838                                 " %016llx\n",
839                                 filesystem,
840                                 mirror.key_cur.obj_id);
841                         exit(0);
842                 }
843 #endif
844         }
845
846         /*
847          * Read and process the termination sync record.
848          */
849         mrec = read_mrecord(0, &error, &pickup);
850
851         if (mrec && mrec->head.type == HAMMER_MREC_TYPE_TERM) {
852                 fprintf(stderr, "Mirror-write: received termination request\n");
853                 free(mrec);
854                 return;
855         }
856
857         if (mrec == NULL ||
858             (mrec->head.type != HAMMER_MREC_TYPE_SYNC &&
859              mrec->head.type != HAMMER_MREC_TYPE_IDLE) ||
860             mrec->head.rec_size != sizeof(mrec->sync)) {
861                 fprintf(stderr, "Mirror-write %s: Did not get termination "
862                                 "sync record, or rec_size is wrong rt=%d\n",
863                                 filesystem,
864                                 (mrec ? mrec->head.type : -1));
865                 exit(1);
866         }
867
868         /*
869          * Update the PFS info on the target so the user has visibility
870          * into the new snapshot, and sync the target filesystem.
871          */
872         if (mrec->head.type == HAMMER_MREC_TYPE_SYNC) {
873                 update_pfs_snapshot(fd, mirror.tid_end, pfs.pfs_id);
874
875                 bzero(&synctid, sizeof(synctid));
876                 synctid.op = HAMMER_SYNCTID_SYNC2;
877                 ioctl(fd, HAMMERIOC_SYNCTID, &synctid);
878
879                 if (VerboseOpt >= 2) {
880                         fprintf(stderr, "Mirror-write %s: succeeded\n",
881                                 filesystem);
882                 }
883         }
884
885         free(mrec);
886         mrec = NULL;
887
888         /*
889          * Report back to the originator.
890          */
891         if (TwoWayPipeOpt) {
892                 mrec_tmp.update.tid = mirror.tid_end;
893                 write_mrecord(1, HAMMER_MREC_TYPE_UPDATE,
894                               &mrec_tmp, sizeof(mrec_tmp.update));
895         } else {
896                 printf("Source can update synctid to 0x%016jx\n",
897                        (uintmax_t)mirror.tid_end);
898         }
899         relpfs(fd, &pfs);
900         goto again;
901 }
902
903 void
904 hammer_cmd_mirror_dump(void)
905 {
906         char *buf = malloc(SERIALBUF_SIZE);
907         struct hammer_ioc_mrecord_head pickup;
908         hammer_ioc_mrecord_any_t mrec;
909         int error;
910         int size;
911         int offset;
912         int bytes;
913
914         /*
915          * Read and process the PFS header
916          */
917         pickup.signature = 0;
918         pickup.type = 0;
919
920         mrec = read_mrecord(0, &error, &pickup);
921
922 again:
923         /*
924          * Read and process bulk records
925          */
926         for (;;) {
927                 size = read_mrecords(0, buf, SERIALBUF_SIZE, &pickup);
928                 if (size <= 0)
929                         break;
930                 offset = 0;
931                 while (offset < size) {
932                         mrec = (void *)((char *)buf + offset);
933                         bytes = HAMMER_HEAD_DOALIGN(mrec->head.rec_size);
934                         if (offset + bytes > size) {
935                                 fprintf(stderr, "Misaligned record\n");
936                                 exit(1);
937                         }
938
939                         switch(mrec->head.type & HAMMER_MRECF_TYPE_MASK) {
940                         case HAMMER_MREC_TYPE_REC_BADCRC:
941                         case HAMMER_MREC_TYPE_REC:
942                                 printf("Record obj=%016jx key=%016jx "
943                                        "rt=%02x ot=%02x",
944                                         (uintmax_t)mrec->rec.leaf.base.obj_id,
945                                         (uintmax_t)mrec->rec.leaf.base.key,
946                                         mrec->rec.leaf.base.rec_type,
947                                         mrec->rec.leaf.base.obj_type);
948                                 if (mrec->head.type ==
949                                     HAMMER_MREC_TYPE_REC_BADCRC) {
950                                         printf(" (BAD CRC)");
951                                 }
952                                 printf("\n");
953                                 printf("       tids %016jx:%016jx data=%d\n",
954                                     (uintmax_t)mrec->rec.leaf.base.create_tid,
955                                     (uintmax_t)mrec->rec.leaf.base.delete_tid,
956                                     mrec->rec.leaf.data_len);
957                                 break;
958                         case HAMMER_MREC_TYPE_PASS:
959                                 printf("Pass   obj=%016jx key=%016jx "
960                                        "rt=%02x ot=%02x\n",
961                                         (uintmax_t)mrec->rec.leaf.base.obj_id,
962                                         (uintmax_t)mrec->rec.leaf.base.key,
963                                         mrec->rec.leaf.base.rec_type,
964                                         mrec->rec.leaf.base.obj_type);
965                                 printf("       tids %016jx:%016jx data=%d\n",
966                                     (uintmax_t)mrec->rec.leaf.base.create_tid,
967                                     (uintmax_t)mrec->rec.leaf.base.delete_tid,
968                                         mrec->rec.leaf.data_len);
969                                 break;
970                         case HAMMER_MREC_TYPE_SKIP:
971                                 printf("Skip   obj=%016jx key=%016jx rt=%02x to\n"
972                                        "       obj=%016jx key=%016jx rt=%02x\n",
973                                        (uintmax_t)mrec->skip.skip_beg.obj_id,
974                                        (uintmax_t)mrec->skip.skip_beg.key,
975                                        mrec->skip.skip_beg.rec_type,
976                                        (uintmax_t)mrec->skip.skip_end.obj_id,
977                                        (uintmax_t)mrec->skip.skip_end.key,
978                                        mrec->skip.skip_end.rec_type);
979                         default:
980                                 break;
981                         }
982                         offset += bytes;
983                 }
984         }
985
986         /*
987          * Read and process the termination sync record.
988          */
989         mrec = read_mrecord(0, &error, &pickup);
990         if (mrec == NULL ||
991             (mrec->head.type != HAMMER_MREC_TYPE_SYNC &&
992              mrec->head.type != HAMMER_MREC_TYPE_IDLE)
993          ) {
994                 fprintf(stderr, "Mirror-dump: Did not get termination "
995                                 "sync record\n");
996         }
997
998         /*
999          * Continue with more batches until EOF.
1000          */
1001         mrec = read_mrecord(0, &error, &pickup);
1002         if (mrec)
1003                 goto again;
1004 }
1005
1006 void
1007 hammer_cmd_mirror_copy(char **av, int ac, int streaming)
1008 {
1009         pid_t pid1;
1010         pid_t pid2;
1011         int fds[2];
1012         const char *xav[32];
1013         char tbuf[16];
1014         char *ptr;
1015         int xac;
1016
1017         if (ac != 2)
1018                 mirror_usage(1);
1019
1020         TwoWayPipeOpt = 1;
1021         signal(SIGPIPE, SIG_IGN);
1022
1023 again:
1024         if (pipe(fds) < 0) {
1025                 perror("pipe");
1026                 exit(1);
1027         }
1028
1029         /*
1030          * Source
1031          */
1032         if ((pid1 = fork()) == 0) {
1033                 signal(SIGPIPE, SIG_DFL);
1034                 dup2(fds[0], 0);
1035                 dup2(fds[0], 1);
1036                 close(fds[0]);
1037                 close(fds[1]);
1038                 if ((ptr = strchr(av[0], ':')) != NULL) {
1039                         *ptr++ = 0;
1040                         xac = 0;
1041                         xav[xac++] = "ssh";
1042                         if (CompressOpt)
1043                                 xav[xac++] = "-C";
1044                         if (SshPort) {
1045                                 xav[xac++] = "-p";
1046                                 xav[xac++] = SshPort;
1047                         }
1048                         xav[xac++] = av[0];
1049                         xav[xac++] = "hammer";
1050
1051                         switch(VerboseOpt) {
1052                         case 0:
1053                                 break;
1054                         case 1:
1055                                 xav[xac++] = "-v";
1056                                 break;
1057                         case 2:
1058                                 xav[xac++] = "-vv";
1059                                 break;
1060                         default:
1061                                 xav[xac++] = "-vvv";
1062                                 break;
1063                         }
1064                         if (ForceYesOpt) {
1065                                 xav[xac++] = "-y";
1066                         }
1067                         xav[xac++] = "-2";
1068                         if (TimeoutOpt) {
1069                                 snprintf(tbuf, sizeof(tbuf), "%d", TimeoutOpt);
1070                                 xav[xac++] = "-t";
1071                                 xav[xac++] = tbuf;
1072                         }
1073                         if (SplitupOptStr) {
1074                                 xav[xac++] = "-S";
1075                                 xav[xac++] = SplitupOptStr;
1076                         }
1077                         if (streaming)
1078                                 xav[xac++] = "mirror-read-stream";
1079                         else
1080                                 xav[xac++] = "mirror-read";
1081                         xav[xac++] = ptr;
1082                         xav[xac++] = NULL;
1083                         execv("/usr/bin/ssh", (void *)xav);
1084                 } else {
1085                         hammer_cmd_mirror_read(av, 1, streaming);
1086                         fflush(stdout);
1087                         fflush(stderr);
1088                 }
1089                 _exit(1);
1090         }
1091
1092         /*
1093          * Target
1094          */
1095         if ((pid2 = fork()) == 0) {
1096                 signal(SIGPIPE, SIG_DFL);
1097                 dup2(fds[1], 0);
1098                 dup2(fds[1], 1);
1099                 close(fds[0]);
1100                 close(fds[1]);
1101                 if ((ptr = strchr(av[1], ':')) != NULL) {
1102                         *ptr++ = 0;
1103                         xac = 0;
1104                         xav[xac++] = "ssh";
1105                         if (CompressOpt)
1106                                 xav[xac++] = "-C";
1107                         if (SshPort) {
1108                                 xav[xac++] = "-p";
1109                                 xav[xac++] = SshPort;
1110                         }
1111                         xav[xac++] = av[1];
1112                         xav[xac++] = "hammer";
1113
1114                         switch(VerboseOpt) {
1115                         case 0:
1116                                 break;
1117                         case 1:
1118                                 xav[xac++] = "-v";
1119                                 break;
1120                         case 2:
1121                                 xav[xac++] = "-vv";
1122                                 break;
1123                         default:
1124                                 xav[xac++] = "-vvv";
1125                                 break;
1126                         }
1127                         if (ForceYesOpt) {
1128                                 xav[xac++] = "-y";
1129                         }
1130                         xav[xac++] = "-2";
1131                         xav[xac++] = "mirror-write";
1132                         xav[xac++] = ptr;
1133                         xav[xac++] = NULL;
1134                         execv("/usr/bin/ssh", (void *)xav);
1135                 } else {
1136                         hammer_cmd_mirror_write(av + 1, 1);
1137                         fflush(stdout);
1138                         fflush(stderr);
1139                 }
1140                 _exit(1);
1141         }
1142         close(fds[0]);
1143         close(fds[1]);
1144
1145         while (waitpid(pid1, NULL, 0) <= 0)
1146                 ;
1147         while (waitpid(pid2, NULL, 0) <= 0)
1148                 ;
1149
1150         /*
1151          * If the link is lost restart
1152          */
1153         if (streaming) {
1154                 if (VerboseOpt) {
1155                         fprintf(stderr, "\nLost Link\n");
1156                         fflush(stderr);
1157                 }
1158                 sleep(15 + DelayOpt);
1159                 goto again;
1160         }
1161
1162 }
1163
1164 /*
1165  * Read and return multiple mrecords
1166  */
1167 static int
1168 read_mrecords(int fd, char *buf, u_int size, hammer_ioc_mrecord_head_t pickup)
1169 {
1170         hammer_ioc_mrecord_any_t mrec;
1171         u_int count;
1172         size_t n;
1173         size_t i;
1174         size_t bytes;
1175         int type;
1176
1177         count = 0;
1178         while (size - count >= HAMMER_MREC_HEADSIZE) {
1179                 /*
1180                  * Cached the record header in case we run out of buffer
1181                  * space.
1182                  */
1183                 fflush(stdout);
1184                 if (pickup->signature == 0) {
1185                         for (n = 0; n < HAMMER_MREC_HEADSIZE; n += i) {
1186                                 i = read(fd, (char *)pickup + n,
1187                                          HAMMER_MREC_HEADSIZE - n);
1188                                 if (i <= 0)
1189                                         break;
1190                         }
1191                         if (n == 0)
1192                                 break;
1193                         if (n != HAMMER_MREC_HEADSIZE) {
1194                                 fprintf(stderr, "read_mrecords: short read on pipe\n");
1195                                 exit(1);
1196                         }
1197                         if (pickup->signature != HAMMER_IOC_MIRROR_SIGNATURE) {
1198                                 fprintf(stderr, "read_mrecords: malformed record on pipe, "
1199                                         "bad signature\n");
1200                                 exit(1);
1201                         }
1202                 }
1203                 if (pickup->rec_size < HAMMER_MREC_HEADSIZE ||
1204                     pickup->rec_size > sizeof(*mrec) + HAMMER_XBUFSIZE) {
1205                         fprintf(stderr, "read_mrecords: malformed record on pipe, "
1206                                 "illegal rec_size\n");
1207                         exit(1);
1208                 }
1209
1210                 /*
1211                  * Stop if we have insufficient space for the record and data.
1212                  */
1213                 bytes = HAMMER_HEAD_DOALIGN(pickup->rec_size);
1214                 if (size - count < bytes)
1215                         break;
1216
1217                 /*
1218                  * Stop if the record type is not a REC, SKIP, or PASS,
1219                  * which are the only types the ioctl supports.  Other types
1220                  * are used only by the userland protocol.
1221                  *
1222                  * Ignore all flags.
1223                  */
1224                 type = pickup->type & HAMMER_MRECF_TYPE_LOMASK;
1225                 if (type != HAMMER_MREC_TYPE_PFSD &&
1226                     type != HAMMER_MREC_TYPE_REC &&
1227                     type != HAMMER_MREC_TYPE_SKIP &&
1228                     type != HAMMER_MREC_TYPE_PASS) {
1229                         break;
1230                 }
1231
1232                 /*
1233                  * Read the remainder and clear the pickup signature.
1234                  */
1235                 for (n = HAMMER_MREC_HEADSIZE; n < bytes; n += i) {
1236                         i = read(fd, buf + count + n, bytes - n);
1237                         if (i <= 0)
1238                                 break;
1239                 }
1240                 if (n != bytes) {
1241                         fprintf(stderr, "read_mrecords: short read on pipe\n");
1242                         exit(1);
1243                 }
1244
1245                 bcopy(pickup, buf + count, HAMMER_MREC_HEADSIZE);
1246                 pickup->signature = 0;
1247                 pickup->type = 0;
1248                 mrec = (void *)(buf + count);
1249
1250                 /*
1251                  * Validate the completed record
1252                  */
1253                 if (mrec->head.rec_crc !=
1254                     crc32((char *)mrec + HAMMER_MREC_CRCOFF,
1255                           mrec->head.rec_size - HAMMER_MREC_CRCOFF)) {
1256                         fprintf(stderr, "read_mrecords: malformed record "
1257                                         "on pipe, bad crc\n");
1258                         exit(1);
1259                 }
1260
1261                 /*
1262                  * If its a B-Tree record validate the data crc.
1263                  *
1264                  * NOTE: If the VFS passes us an explicitly errorde mrec
1265                  *       we just pass it through.
1266                  */
1267                 type = mrec->head.type & HAMMER_MRECF_TYPE_MASK;
1268
1269                 if (type == HAMMER_MREC_TYPE_REC) {
1270                         if (mrec->head.rec_size <
1271                             sizeof(mrec->rec) + mrec->rec.leaf.data_len) {
1272                                 fprintf(stderr,
1273                                         "read_mrecords: malformed record on "
1274                                         "pipe, illegal element data_len\n");
1275                                 exit(1);
1276                         }
1277                         if (mrec->rec.leaf.data_len &&
1278                             mrec->rec.leaf.data_offset &&
1279                             hammer_crc_test_leaf(&mrec->rec + 1, &mrec->rec.leaf) == 0) {
1280                                 fprintf(stderr,
1281                                         "read_mrecords: data_crc did not "
1282                                         "match data! obj=%016jx key=%016jx\n",
1283                                         (uintmax_t)mrec->rec.leaf.base.obj_id,
1284                                         (uintmax_t)mrec->rec.leaf.base.key);
1285                                 fprintf(stderr,
1286                                         "continuing, but there are problems\n");
1287                         }
1288                 }
1289                 count += bytes;
1290         }
1291         return(count);
1292 }
1293
1294 /*
1295  * Read and return a single mrecord.
1296  */
1297 static
1298 hammer_ioc_mrecord_any_t
1299 read_mrecord(int fdin, int *errorp, hammer_ioc_mrecord_head_t pickup)
1300 {
1301         hammer_ioc_mrecord_any_t mrec;
1302         struct hammer_ioc_mrecord_head mrechd;
1303         size_t bytes;
1304         size_t n;
1305         size_t i;
1306
1307         if (pickup && pickup->type != 0) {
1308                 mrechd = *pickup;
1309                 pickup->signature = 0;
1310                 pickup->type = 0;
1311                 n = HAMMER_MREC_HEADSIZE;
1312         } else {
1313                 /*
1314                  * Read in the PFSD header from the sender.
1315                  */
1316                 for (n = 0; n < HAMMER_MREC_HEADSIZE; n += i) {
1317                         i = read(fdin, (char *)&mrechd + n, HAMMER_MREC_HEADSIZE - n);
1318                         if (i <= 0)
1319                                 break;
1320                 }
1321                 if (n == 0) {
1322                         *errorp = 0;    /* EOF */
1323                         return(NULL);
1324                 }
1325                 if (n != HAMMER_MREC_HEADSIZE) {
1326                         fprintf(stderr, "short read of mrecord header\n");
1327                         *errorp = EPIPE;
1328                         return(NULL);
1329                 }
1330         }
1331         if (mrechd.signature != HAMMER_IOC_MIRROR_SIGNATURE) {
1332                 fprintf(stderr, "read_mrecord: bad signature\n");
1333                 *errorp = EINVAL;
1334                 return(NULL);
1335         }
1336         bytes = HAMMER_HEAD_DOALIGN(mrechd.rec_size);
1337         assert(bytes >= sizeof(mrechd));
1338         mrec = malloc(bytes);
1339         mrec->head = mrechd;
1340
1341         while (n < bytes) {
1342                 i = read(fdin, (char *)mrec + n, bytes - n);
1343                 if (i <= 0)
1344                         break;
1345                 n += i;
1346         }
1347         if (n != bytes) {
1348                 fprintf(stderr, "read_mrecord: short read on payload\n");
1349                 *errorp = EPIPE;
1350                 return(NULL);
1351         }
1352         if (mrec->head.rec_crc !=
1353             crc32((char *)mrec + HAMMER_MREC_CRCOFF,
1354                   mrec->head.rec_size - HAMMER_MREC_CRCOFF)) {
1355                 fprintf(stderr, "read_mrecord: bad CRC\n");
1356                 *errorp = EINVAL;
1357                 return(NULL);
1358         }
1359         *errorp = 0;
1360         return(mrec);
1361 }
1362
1363 static
1364 void
1365 write_mrecord(int fdout, u_int32_t type, hammer_ioc_mrecord_any_t mrec,
1366               int bytes)
1367 {
1368         char zbuf[HAMMER_HEAD_ALIGN];
1369         int pad;
1370
1371         pad = HAMMER_HEAD_DOALIGN(bytes) - bytes;
1372
1373         assert(bytes >= (int)sizeof(mrec->head));
1374         bzero(&mrec->head, sizeof(mrec->head));
1375         mrec->head.signature = HAMMER_IOC_MIRROR_SIGNATURE;
1376         mrec->head.type = type;
1377         mrec->head.rec_size = bytes;
1378         mrec->head.rec_crc = crc32((char *)mrec + HAMMER_MREC_CRCOFF,
1379                                    bytes - HAMMER_MREC_CRCOFF);
1380         if (write(fdout, mrec, bytes) != bytes) {
1381                 fprintf(stderr, "write_mrecord: error %d (%s)\n",
1382                         errno, strerror(errno));
1383                 exit(1);
1384         }
1385         if (pad) {
1386                 bzero(zbuf, pad);
1387                 if (write(fdout, zbuf, pad) != pad) {
1388                         fprintf(stderr, "write_mrecord: error %d (%s)\n",
1389                                 errno, strerror(errno));
1390                         exit(1);
1391                 }
1392         }
1393 }
1394
1395 /*
1396  * Generate a mirroring header with the pfs information of the
1397  * originating filesytem.
1398  */
1399 static void
1400 generate_mrec_header(int fd, int pfs_id,
1401                      union hammer_ioc_mrecord_any *mrec_tmp)
1402 {
1403         struct hammer_ioc_pseudofs_rw pfs;
1404
1405         bzero(&pfs, sizeof(pfs));
1406         bzero(mrec_tmp, sizeof(*mrec_tmp));
1407         pfs.pfs_id = pfs_id;
1408         pfs.ondisk = &mrec_tmp->pfs.pfsd;
1409         pfs.bytes = sizeof(mrec_tmp->pfs.pfsd);
1410         if (ioctl(fd, HAMMERIOC_GET_PSEUDOFS, &pfs) != 0) {
1411                 fprintf(stderr, "Mirror-read: not a HAMMER fs/pseudofs!\n");
1412                 exit(1);
1413         }
1414         if (pfs.version != HAMMER_IOC_PSEUDOFS_VERSION) {
1415                 fprintf(stderr, "Mirror-read: HAMMER pfs version mismatch!\n");
1416                 exit(1);
1417         }
1418         mrec_tmp->pfs.version = pfs.version;
1419 }
1420
1421 /*
1422  * Validate the pfs information from the originating filesystem
1423  * against the target filesystem.  shared_uuid must match.
1424  *
1425  * return -1 if we got a TERM record
1426  */
1427 static int
1428 validate_mrec_header(int fd, int fdin, int is_target, int pfs_id,
1429                      struct hammer_ioc_mrecord_head *pickup,
1430                      hammer_tid_t *tid_begp, hammer_tid_t *tid_endp)
1431 {
1432         struct hammer_ioc_pseudofs_rw pfs;
1433         struct hammer_pseudofs_data pfsd;
1434         hammer_ioc_mrecord_any_t mrec;
1435         int error;
1436
1437         /*
1438          * Get the PFSD info from the target filesystem.
1439          */
1440         bzero(&pfs, sizeof(pfs));
1441         bzero(&pfsd, sizeof(pfsd));
1442         pfs.pfs_id = pfs_id;
1443         pfs.ondisk = &pfsd;
1444         pfs.bytes = sizeof(pfsd);
1445         if (ioctl(fd, HAMMERIOC_GET_PSEUDOFS, &pfs) != 0) {
1446                 fprintf(stderr, "mirror-write: not a HAMMER fs/pseudofs!\n");
1447                 exit(1);
1448         }
1449         if (pfs.version != HAMMER_IOC_PSEUDOFS_VERSION) {
1450                 fprintf(stderr, "mirror-write: HAMMER pfs version mismatch!\n");
1451                 exit(1);
1452         }
1453
1454         mrec = read_mrecord(fdin, &error, pickup);
1455         if (mrec == NULL) {
1456                 if (error == 0)
1457                         fprintf(stderr, "validate_mrec_header: short read\n");
1458                 exit(1);
1459         }
1460         if (mrec->head.type == HAMMER_MREC_TYPE_TERM) {
1461                 free(mrec);
1462                 return(-1);
1463         }
1464
1465         if (mrec->head.type != HAMMER_MREC_TYPE_PFSD) {
1466                 fprintf(stderr, "validate_mrec_header: did not get expected "
1467                                 "PFSD record type\n");
1468                 exit(1);
1469         }
1470         if (mrec->head.rec_size != sizeof(mrec->pfs)) {
1471                 fprintf(stderr, "validate_mrec_header: unexpected payload "
1472                                 "size\n");
1473                 exit(1);
1474         }
1475         if (mrec->pfs.version != pfs.version) {
1476                 fprintf(stderr, "validate_mrec_header: Version mismatch\n");
1477                 exit(1);
1478         }
1479
1480         /*
1481          * Whew.  Ok, is the read PFS info compatible with the target?
1482          */
1483         if (bcmp(&mrec->pfs.pfsd.shared_uuid, &pfsd.shared_uuid,
1484                  sizeof(pfsd.shared_uuid)) != 0) {
1485                 fprintf(stderr,
1486                         "mirror-write: source and target have "
1487                         "different shared-uuid's!\n");
1488                 exit(1);
1489         }
1490         if (is_target &&
1491             (pfsd.mirror_flags & HAMMER_PFSD_SLAVE) == 0) {
1492                 fprintf(stderr, "mirror-write: target must be in slave mode\n");
1493                 exit(1);
1494         }
1495         if (tid_begp)
1496                 *tid_begp = mrec->pfs.pfsd.sync_beg_tid;
1497         if (tid_endp)
1498                 *tid_endp = mrec->pfs.pfsd.sync_end_tid;
1499         free(mrec);
1500         return(0);
1501 }
1502
1503 static void
1504 update_pfs_snapshot(int fd, hammer_tid_t snapshot_tid, int pfs_id)
1505 {
1506         struct hammer_ioc_pseudofs_rw pfs;
1507         struct hammer_pseudofs_data pfsd;
1508
1509         bzero(&pfs, sizeof(pfs));
1510         bzero(&pfsd, sizeof(pfsd));
1511         pfs.pfs_id = pfs_id;
1512         pfs.ondisk = &pfsd;
1513         pfs.bytes = sizeof(pfsd);
1514         if (ioctl(fd, HAMMERIOC_GET_PSEUDOFS, &pfs) != 0) {
1515                 perror("update_pfs_snapshot (read)");
1516                 exit(1);
1517         }
1518         if (pfsd.sync_end_tid != snapshot_tid) {
1519                 pfsd.sync_end_tid = snapshot_tid;
1520                 if (ioctl(fd, HAMMERIOC_SET_PSEUDOFS, &pfs) != 0) {
1521                         perror("update_pfs_snapshot (rewrite)");
1522                         exit(1);
1523                 }
1524                 if (VerboseOpt >= 2) {
1525                         fprintf(stderr,
1526                                 "Mirror-write: Completed, updated snapshot "
1527                                 "to %016jx\n",
1528                                 (uintmax_t)snapshot_tid);
1529                         fflush(stderr);
1530                 }
1531         }
1532 }
1533
1534 /*
1535  * Bandwidth-limited write in chunks
1536  */
1537 static
1538 ssize_t
1539 writebw(int fd, const void *buf, size_t nbytes,
1540         u_int64_t *bwcount, struct timeval *tv1)
1541 {
1542         struct timeval tv2;
1543         size_t n;
1544         ssize_t r;
1545         ssize_t a;
1546         int usec;
1547
1548         a = 0;
1549         r = 0;
1550         while (nbytes) {
1551                 if (*bwcount + nbytes > BandwidthOpt)
1552                         n = BandwidthOpt - *bwcount;
1553                 else
1554                         n = nbytes;
1555                 if (n)
1556                         r = write(fd, buf, n);
1557                 if (r >= 0) {
1558                         a += r;
1559                         nbytes -= r;
1560                         buf = (const char *)buf + r;
1561                 }
1562                 if ((size_t)r != n)
1563                         break;
1564                 *bwcount += n;
1565                 if (*bwcount >= BandwidthOpt) {
1566                         gettimeofday(&tv2, NULL);
1567                         usec = (int)(tv2.tv_sec - tv1->tv_sec) * 1000000 +
1568                                 (int)(tv2.tv_usec - tv1->tv_usec);
1569                         if (usec >= 0 && usec < 1000000)
1570                                 usleep(1000000 - usec);
1571                         gettimeofday(tv1, NULL);
1572                         *bwcount -= BandwidthOpt;
1573                 }
1574         }
1575         return(a ? a : r);
1576 }
1577
1578 /*
1579  * Get a yes or no answer from the terminal.  The program may be run as
1580  * part of a two-way pipe so we cannot use stdin for this operation.
1581  */
1582 static int
1583 getyn(void)
1584 {
1585         char buf[256];
1586         FILE *fp;
1587         int result;
1588
1589         fp = fopen("/dev/tty", "r");
1590         if (fp == NULL) {
1591                 fprintf(stderr, "No terminal for response\n");
1592                 return(-1);
1593         }
1594         result = -1;
1595         while (fgets(buf, sizeof(buf), fp) != NULL) {
1596                 if (buf[0] == 'y' || buf[0] == 'Y') {
1597                         result = 1;
1598                         break;
1599                 }
1600                 if (buf[0] == 'n' || buf[0] == 'N') {
1601                         result = 0;
1602                         break;
1603                 }
1604                 fprintf(stderr, "Response not understood\n");
1605                 break;
1606         }
1607         fclose(fp);
1608         return(result);
1609 }
1610
1611 static void
1612 mirror_usage(int code)
1613 {
1614         fprintf(stderr,
1615                 "hammer mirror-read <filesystem> [begin-tid]\n"
1616                 "hammer mirror-read-stream <filesystem> [begin-tid]\n"
1617                 "hammer mirror-write <filesystem>\n"
1618                 "hammer mirror-dump\n"
1619                 "hammer mirror-copy [[user@]host:]<filesystem>"
1620                                   " [[user@]host:]<filesystem>\n"
1621                 "hammer mirror-stream [[user@]host:]<filesystem>"
1622                                     " [[user@]host:]<filesystem>\n"
1623         );
1624         exit(code);
1625 }
1626