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