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