sys/vfs/hammer/hammer_mirror.c

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