sys/cddl/contrib/opensolaris/uts/common/fs/zfs/zil.c

   1 /*
   2  * CDDL HEADER START
   3  *
   4  * The contents of this file are subject to the terms of the
   5  * Common Development and Distribution License (the "License").
   6  * You may not use this file except in compliance with the License.
   7  *
   8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
   9  * or http://www.opensolaris.org/os/licensing.
  10  * See the License for the specific language governing permissions
  11  * and limitations under the License.
  12  *
  13  * When distributing Covered Code, include this CDDL HEADER in each
  14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
  15  * If applicable, add the following below this CDDL HEADER, with the
  16  * fields enclosed by brackets "[]" replaced with your own identifying
  17  * information: Portions Copyright [yyyy] [name of copyright owner]
  18  *
  19  * CDDL HEADER END
  20  */
  21 /*
  22  * Copyright 2007 Sun Microsystems, Inc.  All rights reserved.
  23  * Use is subject to license terms.
  24  */
  25
  26 #pragma ident   "%Z%%M% %I%     %E% SMI"
  27
  28 #include <sys/zfs_context.h>
  29 #include <sys/spa.h>
  30 #include <sys/dmu.h>
  31 #include <sys/zap.h>
  32 #include <sys/arc.h>
  33 #include <sys/stat.h>
  34 #include <sys/resource.h>
  35 #include <sys/zil.h>
  36 #include <sys/zil_impl.h>
  37 #include <sys/dsl_dataset.h>
  38 #include <sys/vdev.h>
  39 #include <sys/dmu_tx.h>
  40
  41 /*
  42  * The zfs intent log (ZIL) saves transaction records of system calls
  43  * that change the file system in memory with enough information
  44  * to be able to replay them. These are stored in memory until
  45  * either the DMU transaction group (txg) commits them to the stable pool
  46  * and they can be discarded, or they are flushed to the stable log
  47  * (also in the pool) due to a fsync, O_DSYNC or other synchronous
  48  * requirement. In the event of a panic or power fail then those log
  49  * records (transactions) are replayed.
  50  *
  51  * There is one ZIL per file system. Its on-disk (pool) format consists
  52  * of 3 parts:
  53  *
  54  *      - ZIL header
  55  *      - ZIL blocks
  56  *      - ZIL records
  57  *
  58  * A log record holds a system call transaction. Log blocks can
  59  * hold many log records and the blocks are chained together.
  60  * Each ZIL block contains a block pointer (blkptr_t) to the next
  61  * ZIL block in the chain. The ZIL header points to the first
  62  * block in the chain. Note there is not a fixed place in the pool
  63  * to hold blocks. They are dynamically allocated and freed as
  64  * needed from the blocks available. Figure X shows the ZIL structure:
  65  */
  66
  67 /*
  68  * This global ZIL switch affects all pools
  69  */
  70 int zil_disable = 0;    /* disable intent logging */
  71 SYSCTL_DECL(_vfs_zfs);
  72 TUNABLE_INT("vfs.zfs.zil_disable", &zil_disable);
  73 SYSCTL_INT(_vfs_zfs, OID_AUTO, zil_disable, CTLFLAG_RW, &zil_disable, 0,
  74     "Disable ZFS Intent Log (ZIL)");
  75
  76 /*
  77  * Tunable parameter for debugging or performance analysis.  Setting
  78  * zfs_nocacheflush will cause corruption on power loss if a volatile
  79  * out-of-order write cache is enabled.
  80  */
  81 boolean_t zfs_nocacheflush = B_FALSE;
  82 TUNABLE_INT("vfs.zfs.cache_flush_disable", &zfs_nocacheflush);
  83 SYSCTL_INT(_vfs_zfs, OID_AUTO, cache_flush_disable, CTLFLAG_RDTUN,
  84     &zfs_nocacheflush, 0, "Disable cache flush");
  85
  86 static kmem_cache_t *zil_lwb_cache;
  87
  88 static int
  89 zil_dva_compare(const void *x1, const void *x2)
  90 {
  91         const dva_t *dva1 = x1;
  92         const dva_t *dva2 = x2;
  93
  94         if (DVA_GET_VDEV(dva1) < DVA_GET_VDEV(dva2))
  95                 return (-1);
  96         if (DVA_GET_VDEV(dva1) > DVA_GET_VDEV(dva2))
  97                 return (1);
  98
  99         if (DVA_GET_OFFSET(dva1) < DVA_GET_OFFSET(dva2))
 100                 return (-1);
 101         if (DVA_GET_OFFSET(dva1) > DVA_GET_OFFSET(dva2))
 102                 return (1);
 103
 104         return (0);
 105 }
 106
 107 static void
 108 zil_dva_tree_init(avl_tree_t *t)
 109 {
 110         avl_create(t, zil_dva_compare, sizeof (zil_dva_node_t),
 111             offsetof(zil_dva_node_t, zn_node));
 112 }
 113
 114 static void
 115 zil_dva_tree_fini(avl_tree_t *t)
 116 {
 117         zil_dva_node_t *zn;
 118         void *cookie = NULL;
 119
 120         while ((zn = avl_destroy_nodes(t, &cookie)) != NULL)
 121                 kmem_free(zn, sizeof (zil_dva_node_t));
 122
 123         avl_destroy(t);
 124 }
 125
 126 static int
 127 zil_dva_tree_add(avl_tree_t *t, dva_t *dva)
 128 {
 129         zil_dva_node_t *zn;
 130         avl_index_t where;
 131
 132         if (avl_find(t, dva, &where) != NULL)
 133                 return (EEXIST);
 134
 135         zn = kmem_alloc(sizeof (zil_dva_node_t), KM_SLEEP);
 136         zn->zn_dva = *dva;
 137         avl_insert(t, zn, where);
 138
 139         return (0);
 140 }
 141
 142 static zil_header_t *
 143 zil_header_in_syncing_context(zilog_t *zilog)
 144 {
 145         return ((zil_header_t *)zilog->zl_header);
 146 }
 147
 148 static void
 149 zil_init_log_chain(zilog_t *zilog, blkptr_t *bp)
 150 {
 151         zio_cksum_t *zc = &bp->blk_cksum;
 152
 153         zc->zc_word[ZIL_ZC_GUID_0] = spa_get_random(-1ULL);
 154         zc->zc_word[ZIL_ZC_GUID_1] = spa_get_random(-1ULL);
 155         zc->zc_word[ZIL_ZC_OBJSET] = dmu_objset_id(zilog->zl_os);
 156         zc->zc_word[ZIL_ZC_SEQ] = 1ULL;
 157 }
 158
 159 /*
 160  * Read a log block, make sure it's valid, and byteswap it if necessary.
 161  */
 162 static int
 163 zil_read_log_block(zilog_t *zilog, const blkptr_t *bp, arc_buf_t **abufpp)
 164 {
 165         blkptr_t blk = *bp;
 166         zbookmark_t zb;
 167         uint32_t aflags = ARC_WAIT;
 168         int error;
 169
 170         zb.zb_objset = bp->blk_cksum.zc_word[ZIL_ZC_OBJSET];
 171         zb.zb_object = 0;
 172         zb.zb_level = -1;
 173         zb.zb_blkid = bp->blk_cksum.zc_word[ZIL_ZC_SEQ];
 174
 175         *abufpp = NULL;
 176
 177         error = arc_read(NULL, zilog->zl_spa, &blk, byteswap_uint64_array,
 178             arc_getbuf_func, abufpp, ZIO_PRIORITY_SYNC_READ, ZIO_FLAG_CANFAIL |
 179             ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB, &aflags, &zb);
 180
 181         if (error == 0) {
 182                 char *data = (*abufpp)->b_data;
 183                 uint64_t blksz = BP_GET_LSIZE(bp);
 184                 zil_trailer_t *ztp = (zil_trailer_t *)(data + blksz) - 1;
 185                 zio_cksum_t cksum = bp->blk_cksum;
 186
 187                 /*
 188                  * Sequence numbers should be... sequential.  The checksum
 189                  * verifier for the next block should be bp's checksum plus 1.
 190                  */
 191                 cksum.zc_word[ZIL_ZC_SEQ]++;
 192
 193                 if (bcmp(&cksum, &ztp->zit_next_blk.blk_cksum, sizeof (cksum)))
 194                         error = ESTALE;
 195                 else if (BP_IS_HOLE(&ztp->zit_next_blk))
 196                         error = ENOENT;
 197                 else if (ztp->zit_nused > (blksz - sizeof (zil_trailer_t)))
 198                         error = EOVERFLOW;
 199
 200                 if (error) {
 201                         VERIFY(arc_buf_remove_ref(*abufpp, abufpp) == 1);
 202                         *abufpp = NULL;
 203                 }
 204         }
 205
 206         dprintf("error %d on %llu:%llu\n", error, zb.zb_objset, zb.zb_blkid);
 207
 208         return (error);
 209 }
 210
 211 /*
 212  * Parse the intent log, and call parse_func for each valid record within.
 213  * Return the highest sequence number.
 214  */
 215 uint64_t
 216 zil_parse(zilog_t *zilog, zil_parse_blk_func_t *parse_blk_func,
 217     zil_parse_lr_func_t *parse_lr_func, void *arg, uint64_t txg)
 218 {
 219         const zil_header_t *zh = zilog->zl_header;
 220         uint64_t claim_seq = zh->zh_claim_seq;
 221         uint64_t seq = 0;
 222         uint64_t max_seq = 0;
 223         blkptr_t blk = zh->zh_log;
 224         arc_buf_t *abuf;
 225         char *lrbuf, *lrp;
 226         zil_trailer_t *ztp;
 227         int reclen, error;
 228
 229         if (BP_IS_HOLE(&blk))
 230                 return (max_seq);
 231
 232         /*
 233          * Starting at the block pointed to by zh_log we read the log chain.
 234          * For each block in the chain we strongly check that block to
 235          * ensure its validity.  We stop when an invalid block is found.
 236          * For each block pointer in the chain we call parse_blk_func().
 237          * For each record in each valid block we call parse_lr_func().
 238          * If the log has been claimed, stop if we encounter a sequence
 239          * number greater than the highest claimed sequence number.
 240          */
 241         zil_dva_tree_init(&zilog->zl_dva_tree);
 242         for (;;) {
 243                 seq = blk.blk_cksum.zc_word[ZIL_ZC_SEQ];
 244
 245                 if (claim_seq != 0 && seq > claim_seq)
 246                         break;
 247
 248                 ASSERT(max_seq < seq);
 249                 max_seq = seq;
 250
 251                 error = zil_read_log_block(zilog, &blk, &abuf);
 252
 253                 if (parse_blk_func != NULL)
 254                         parse_blk_func(zilog, &blk, arg, txg);
 255
 256                 if (error)
 257                         break;
 258
 259                 lrbuf = abuf->b_data;
 260                 ztp = (zil_trailer_t *)(lrbuf + BP_GET_LSIZE(&blk)) - 1;
 261                 blk = ztp->zit_next_blk;
 262
 263                 if (parse_lr_func == NULL) {
 264                         VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);
 265                         continue;
 266                 }
 267
 268                 for (lrp = lrbuf; lrp < lrbuf + ztp->zit_nused; lrp += reclen) {
 269                         lr_t *lr = (lr_t *)lrp;
 270                         reclen = lr->lrc_reclen;
 271                         ASSERT3U(reclen, >=, sizeof (lr_t));
 272                         parse_lr_func(zilog, lr, arg, txg);
 273                 }
 274                 VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);
 275         }
 276         zil_dva_tree_fini(&zilog->zl_dva_tree);
 277
 278         return (max_seq);
 279 }
 280
 281 /* ARGSUSED */
 282 static void
 283 zil_claim_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t first_txg)
 284 {
 285         spa_t *spa = zilog->zl_spa;
 286         int err;
 287
 288         /*
 289          * Claim log block if not already committed and not already claimed.
 290          */
 291         if (bp->blk_birth >= first_txg &&
 292             zil_dva_tree_add(&zilog->zl_dva_tree, BP_IDENTITY(bp)) == 0) {
 293                 err = zio_wait(zio_claim(NULL, spa, first_txg, bp, NULL, NULL));
 294                 ASSERT(err == 0);
 295         }
 296 }
 297
 298 static void
 299 zil_claim_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t first_txg)
 300 {
 301         if (lrc->lrc_txtype == TX_WRITE) {
 302                 lr_write_t *lr = (lr_write_t *)lrc;
 303                 zil_claim_log_block(zilog, &lr->lr_blkptr, tx, first_txg);
 304         }
 305 }
 306
 307 /* ARGSUSED */
 308 static void
 309 zil_free_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t claim_txg)
 310 {
 311         zio_free_blk(zilog->zl_spa, bp, dmu_tx_get_txg(tx));
 312 }
 313
 314 static void
 315 zil_free_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t claim_txg)
 316 {
 317         /*
 318          * If we previously claimed it, we need to free it.
 319          */
 320         if (claim_txg != 0 && lrc->lrc_txtype == TX_WRITE) {
 321                 lr_write_t *lr = (lr_write_t *)lrc;
 322                 blkptr_t *bp = &lr->lr_blkptr;
 323                 if (bp->blk_birth >= claim_txg &&
 324                     !zil_dva_tree_add(&zilog->zl_dva_tree, BP_IDENTITY(bp))) {
 325                         (void) arc_free(NULL, zilog->zl_spa,
 326                             dmu_tx_get_txg(tx), bp, NULL, NULL, ARC_WAIT);
 327                 }
 328         }
 329 }
 330
 331 /*
 332  * Create an on-disk intent log.
 333  */
 334 static void
 335 zil_create(zilog_t *zilog)
 336 {
 337         const zil_header_t *zh = zilog->zl_header;
 338         lwb_t *lwb;
 339         uint64_t txg = 0;
 340         dmu_tx_t *tx = NULL;
 341         blkptr_t blk;
 342         int error = 0;
 343
 344         /*
 345          * Wait for any previous destroy to complete.
 346          */
 347         txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
 348
 349         ASSERT(zh->zh_claim_txg == 0);
 350         ASSERT(zh->zh_replay_seq == 0);
 351
 352         blk = zh->zh_log;
 353
 354         /*
 355          * If we don't already have an initial log block, allocate one now.
 356          */
 357         if (BP_IS_HOLE(&blk)) {
 358                 tx = dmu_tx_create(zilog->zl_os);
 359                 (void) dmu_tx_assign(tx, TXG_WAIT);
 360                 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
 361                 txg = dmu_tx_get_txg(tx);
 362
 363                 error = zio_alloc_blk(zilog->zl_spa, ZIL_MIN_BLKSZ, &blk,
 364                     NULL, txg);
 365
 366                 if (error == 0)
 367                         zil_init_log_chain(zilog, &blk);
 368         }
 369
 370         /*
 371          * Allocate a log write buffer (lwb) for the first log block.
 372          */
 373         if (error == 0) {
 374                 lwb = kmem_cache_alloc(zil_lwb_cache, KM_SLEEP);
 375                 lwb->lwb_zilog = zilog;
 376                 lwb->lwb_blk = blk;
 377                 lwb->lwb_nused = 0;
 378                 lwb->lwb_sz = BP_GET_LSIZE(&lwb->lwb_blk);
 379                 lwb->lwb_buf = zio_buf_alloc(lwb->lwb_sz);
 380                 lwb->lwb_max_txg = txg;
 381                 lwb->lwb_zio = NULL;
 382
 383                 mutex_enter(&zilog->zl_lock);
 384                 list_insert_tail(&zilog->zl_lwb_list, lwb);
 385                 mutex_exit(&zilog->zl_lock);
 386         }
 387
 388         /*
 389          * If we just allocated the first log block, commit our transaction
 390          * and wait for zil_sync() to stuff the block poiner into zh_log.
 391          * (zh is part of the MOS, so we cannot modify it in open context.)
 392          */
 393         if (tx != NULL) {
 394                 dmu_tx_commit(tx);
 395                 txg_wait_synced(zilog->zl_dmu_pool, txg);
 396         }
 397
 398         ASSERT(bcmp(&blk, &zh->zh_log, sizeof (blk)) == 0);
 399 }
 400
 401 /*
 402  * In one tx, free all log blocks and clear the log header.
 403  * If keep_first is set, then we're replaying a log with no content.
 404  * We want to keep the first block, however, so that the first
 405  * synchronous transaction doesn't require a txg_wait_synced()
 406  * in zil_create().  We don't need to txg_wait_synced() here either
 407  * when keep_first is set, because both zil_create() and zil_destroy()
 408  * will wait for any in-progress destroys to complete.
 409  */
 410 void
 411 zil_destroy(zilog_t *zilog, boolean_t keep_first)
 412 {
 413         const zil_header_t *zh = zilog->zl_header;
 414         lwb_t *lwb;
 415         dmu_tx_t *tx;
 416         uint64_t txg;
 417
 418         /*
 419          * Wait for any previous destroy to complete.
 420          */
 421         txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
 422
 423         if (BP_IS_HOLE(&zh->zh_log))
 424                 return;
 425
 426         tx = dmu_tx_create(zilog->zl_os);
 427         (void) dmu_tx_assign(tx, TXG_WAIT);
 428         dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
 429         txg = dmu_tx_get_txg(tx);
 430
 431         mutex_enter(&zilog->zl_lock);
 432
 433         ASSERT3U(zilog->zl_destroy_txg, <, txg);
 434         zilog->zl_destroy_txg = txg;
 435         zilog->zl_keep_first = keep_first;
 436
 437         if (!list_is_empty(&zilog->zl_lwb_list)) {
 438                 ASSERT(zh->zh_claim_txg == 0);
 439                 ASSERT(!keep_first);
 440                 while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
 441                         list_remove(&zilog->zl_lwb_list, lwb);
 442                         if (lwb->lwb_buf != NULL)
 443                                 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
 444                         zio_free_blk(zilog->zl_spa, &lwb->lwb_blk, txg);
 445                         kmem_cache_free(zil_lwb_cache, lwb);
 446                 }
 447         } else {
 448                 if (!keep_first) {
 449                         (void) zil_parse(zilog, zil_free_log_block,
 450                             zil_free_log_record, tx, zh->zh_claim_txg);
 451                 }
 452         }
 453         mutex_exit(&zilog->zl_lock);
 454
 455         dmu_tx_commit(tx);
 456
 457         if (keep_first)                 /* no need to wait in this case */
 458                 return;
 459
 460         txg_wait_synced(zilog->zl_dmu_pool, txg);
 461         ASSERT(BP_IS_HOLE(&zh->zh_log));
 462 }
 463
 464 int
 465 zil_claim(char *osname, void *txarg)
 466 {
 467         dmu_tx_t *tx = txarg;
 468         uint64_t first_txg = dmu_tx_get_txg(tx);
 469         zilog_t *zilog;
 470         zil_header_t *zh;
 471         objset_t *os;
 472         int error;
 473
 474         error = dmu_objset_open(osname, DMU_OST_ANY, DS_MODE_STANDARD, &os);
 475         if (error) {
 476                 cmn_err(CE_WARN, "can't process intent log for %s", osname);
 477                 return (0);
 478         }
 479
 480         zilog = dmu_objset_zil(os);
 481         zh = zil_header_in_syncing_context(zilog);
 482
 483         /*
 484          * Claim all log blocks if we haven't already done so, and remember
 485          * the highest claimed sequence number.  This ensures that if we can
 486          * read only part of the log now (e.g. due to a missing device),
 487          * but we can read the entire log later, we will not try to replay
 488          * or destroy beyond the last block we successfully claimed.
 489          */
 490         ASSERT3U(zh->zh_claim_txg, <=, first_txg);
 491         if (zh->zh_claim_txg == 0 && !BP_IS_HOLE(&zh->zh_log)) {
 492                 zh->zh_claim_txg = first_txg;
 493                 zh->zh_claim_seq = zil_parse(zilog, zil_claim_log_block,
 494                     zil_claim_log_record, tx, first_txg);
 495                 dsl_dataset_dirty(dmu_objset_ds(os), tx);
 496         }
 497
 498         ASSERT3U(first_txg, ==, (spa_last_synced_txg(zilog->zl_spa) + 1));
 499         dmu_objset_close(os);
 500         return (0);
 501 }
 502
 503 void
 504 zil_add_vdev(zilog_t *zilog, uint64_t vdev)
 505 {
 506         zil_vdev_t *zv, *new;
 507         uint64_t bmap_sz = sizeof (zilog->zl_vdev_bmap) << 3;
 508         uchar_t *cp;
 509
 510         if (zfs_nocacheflush)
 511                 return;
 512
 513         if (vdev < bmap_sz) {
 514                 cp = zilog->zl_vdev_bmap + (vdev / 8);
 515                 atomic_or_8(cp, 1 << (vdev % 8));
 516         } else  {
 517                 /*
 518                  * insert into ordered list
 519                  */
 520                 mutex_enter(&zilog->zl_lock);
 521                 for (zv = list_head(&zilog->zl_vdev_list); zv != NULL;
 522                     zv = list_next(&zilog->zl_vdev_list, zv)) {
 523                         if (zv->vdev == vdev) {
 524                                 /* duplicate found - just return */
 525                                 mutex_exit(&zilog->zl_lock);
 526                                 return;
 527                         }
 528                         if (zv->vdev > vdev) {
 529                                 /* insert before this entry */
 530                                 new = kmem_alloc(sizeof (zil_vdev_t),
 531                                     KM_SLEEP);
 532                                 new->vdev = vdev;
 533                                 list_insert_before(&zilog->zl_vdev_list,
 534                                     zv, new);
 535                                 mutex_exit(&zilog->zl_lock);
 536                                 return;
 537                         }
 538                 }
 539                 /* ran off end of list, insert at the end */
 540                 ASSERT(zv == NULL);
 541                 new = kmem_alloc(sizeof (zil_vdev_t), KM_SLEEP);
 542                 new->vdev = vdev;
 543                 list_insert_tail(&zilog->zl_vdev_list, new);
 544                 mutex_exit(&zilog->zl_lock);
 545         }
 546 }
 547
 548 /* start an async flush of the write cache for this vdev */
 549 void
 550 zil_flush_vdev(spa_t *spa, uint64_t vdev, zio_t **zio)
 551 {
 552         vdev_t *vd;
 553
 554         if (*zio == NULL)
 555                 *zio = zio_root(spa, NULL, NULL, ZIO_FLAG_CANFAIL);
 556
 557         vd = vdev_lookup_top(spa, vdev);
 558         ASSERT(vd);
 559
 560         (void) zio_nowait(zio_ioctl(*zio, spa, vd, DKIOCFLUSHWRITECACHE,
 561             NULL, NULL, ZIO_PRIORITY_NOW,
 562             ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_RETRY));
 563 }
 564
 565 void
 566 zil_flush_vdevs(zilog_t *zilog)
 567 {
 568         zil_vdev_t *zv;
 569         zio_t *zio = NULL;
 570         spa_t *spa = zilog->zl_spa;
 571         uint64_t vdev;
 572         uint8_t b;
 573         int i, j;
 574
 575         ASSERT(zilog->zl_writer);
 576
 577         for (i = 0; i < sizeof (zilog->zl_vdev_bmap); i++) {
 578                 b = zilog->zl_vdev_bmap[i];
 579                 if (b == 0)
 580                         continue;
 581                 for (j = 0; j < 8; j++) {
 582                         if (b & (1 << j)) {
 583                                 vdev = (i << 3) + j;
 584                                 zil_flush_vdev(spa, vdev, &zio);
 585                         }
 586                 }
 587                 zilog->zl_vdev_bmap[i] = 0;
 588         }
 589
 590         while ((zv = list_head(&zilog->zl_vdev_list)) != NULL) {
 591                 zil_flush_vdev(spa, zv->vdev, &zio);
 592                 list_remove(&zilog->zl_vdev_list, zv);
 593                 kmem_free(zv, sizeof (zil_vdev_t));
 594         }
 595         /*
 596          * Wait for all the flushes to complete.  Not all devices actually
 597          * support the DKIOCFLUSHWRITECACHE ioctl, so it's OK if it fails.
 598          */
 599         if (zio)
 600                 (void) zio_wait(zio);
 601 }
 602
 603 /*
 604  * Function called when a log block write completes
 605  */
 606 static void
 607 zil_lwb_write_done(zio_t *zio)
 608 {
 609         lwb_t *lwb = zio->io_private;
 610         zilog_t *zilog = lwb->lwb_zilog;
 611
 612         /*
 613          * Now that we've written this log block, we have a stable pointer
 614          * to the next block in the chain, so it's OK to let the txg in
 615          * which we allocated the next block sync.
 616          */
 617         txg_rele_to_sync(&lwb->lwb_txgh);
 618
 619         zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
 620         mutex_enter(&zilog->zl_lock);
 621         lwb->lwb_buf = NULL;
 622         if (zio->io_error) {
 623                 zilog->zl_log_error = B_TRUE;
 624                 mutex_exit(&zilog->zl_lock);
 625                 return;
 626         }
 627         mutex_exit(&zilog->zl_lock);
 628 }
 629
 630 /*
 631  * Initialize the io for a log block.
 632  *
 633  * Note, we should not initialize the IO until we are about
 634  * to use it, since zio_rewrite() does a spa_config_enter().
 635  */
 636 static void
 637 zil_lwb_write_init(zilog_t *zilog, lwb_t *lwb)
 638 {
 639         zbookmark_t zb;
 640
 641         zb.zb_objset = lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_OBJSET];
 642         zb.zb_object = 0;
 643         zb.zb_level = -1;
 644         zb.zb_blkid = lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_SEQ];
 645
 646         if (zilog->zl_root_zio == NULL) {
 647                 zilog->zl_root_zio = zio_root(zilog->zl_spa, NULL, NULL,
 648                     ZIO_FLAG_CANFAIL);
 649         }
 650         if (lwb->lwb_zio == NULL) {
 651                 lwb->lwb_zio = zio_rewrite(zilog->zl_root_zio, zilog->zl_spa,
 652                     ZIO_CHECKSUM_ZILOG, 0, &lwb->lwb_blk, lwb->lwb_buf,
 653                     lwb->lwb_sz, zil_lwb_write_done, lwb,
 654                     ZIO_PRIORITY_LOG_WRITE, ZIO_FLAG_MUSTSUCCEED, &zb);
 655         }
 656 }
 657
 658 /*
 659  * Start a log block write and advance to the next log block.
 660  * Calls are serialized.
 661  */
 662 static lwb_t *
 663 zil_lwb_write_start(zilog_t *zilog, lwb_t *lwb)
 664 {
 665         lwb_t *nlwb;
 666         zil_trailer_t *ztp = (zil_trailer_t *)(lwb->lwb_buf + lwb->lwb_sz) - 1;
 667         spa_t *spa = zilog->zl_spa;
 668         blkptr_t *bp = &ztp->zit_next_blk;
 669         uint64_t txg;
 670         uint64_t zil_blksz;
 671         int error;
 672
 673         ASSERT(lwb->lwb_nused <= ZIL_BLK_DATA_SZ(lwb));
 674
 675         /*
 676          * Allocate the next block and save its address in this block
 677          * before writing it in order to establish the log chain.
 678          * Note that if the allocation of nlwb synced before we wrote
 679          * the block that points at it (lwb), we'd leak it if we crashed.
 680          * Therefore, we don't do txg_rele_to_sync() until zil_lwb_write_done().
 681          */
 682         txg = txg_hold_open(zilog->zl_dmu_pool, &lwb->lwb_txgh);
 683         txg_rele_to_quiesce(&lwb->lwb_txgh);
 684
 685         /*
 686          * Pick a ZIL blocksize. We request a size that is the
 687          * maximum of the previous used size, the current used size and
 688          * the amount waiting in the queue.
 689          */
 690         zil_blksz = MAX(zilog->zl_prev_used,
 691             zilog->zl_cur_used + sizeof (*ztp));
 692         zil_blksz = MAX(zil_blksz, zilog->zl_itx_list_sz + sizeof (*ztp));
 693         zil_blksz = P2ROUNDUP_TYPED(zil_blksz, ZIL_MIN_BLKSZ, uint64_t);
 694         if (zil_blksz > ZIL_MAX_BLKSZ)
 695                 zil_blksz = ZIL_MAX_BLKSZ;
 696
 697         BP_ZERO(bp);
 698         /* pass the old blkptr in order to spread log blocks across devs */
 699         error = zio_alloc_blk(spa, zil_blksz, bp, &lwb->lwb_blk, txg);
 700         if (error) {
 701                 dmu_tx_t *tx = dmu_tx_create_assigned(zilog->zl_dmu_pool, txg);
 702
 703                 /*
 704                  * We dirty the dataset to ensure that zil_sync() will
 705                  * be called to remove this lwb from our zl_lwb_list.
 706                  * Failing to do so, may leave an lwb with a NULL lwb_buf
 707                  * hanging around on the zl_lwb_list.
 708                  */
 709                 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
 710                 dmu_tx_commit(tx);
 711
 712                 /*
 713                  * Since we've just experienced an allocation failure so we
 714                  * terminate the current lwb and send it on its way.
 715                  */
 716                 ztp->zit_pad = 0;
 717                 ztp->zit_nused = lwb->lwb_nused;
 718                 ztp->zit_bt.zbt_cksum = lwb->lwb_blk.blk_cksum;
 719                 zio_nowait(lwb->lwb_zio);
 720
 721                 /*
 722                  * By returning NULL the caller will call tx_wait_synced()
 723                  */
 724                 return (NULL);
 725         }
 726
 727         ASSERT3U(bp->blk_birth, ==, txg);
 728         ztp->zit_pad = 0;
 729         ztp->zit_nused = lwb->lwb_nused;
 730         ztp->zit_bt.zbt_cksum = lwb->lwb_blk.blk_cksum;
 731         bp->blk_cksum = lwb->lwb_blk.blk_cksum;
 732         bp->blk_cksum.zc_word[ZIL_ZC_SEQ]++;
 733
 734         /*
 735          * Allocate a new log write buffer (lwb).
 736          */
 737         nlwb = kmem_cache_alloc(zil_lwb_cache, KM_SLEEP);
 738
 739         nlwb->lwb_zilog = zilog;
 740         nlwb->lwb_blk = *bp;
 741         nlwb->lwb_nused = 0;
 742         nlwb->lwb_sz = BP_GET_LSIZE(&nlwb->lwb_blk);
 743         nlwb->lwb_buf = zio_buf_alloc(nlwb->lwb_sz);
 744         nlwb->lwb_max_txg = txg;
 745         nlwb->lwb_zio = NULL;
 746
 747         /*
 748          * Put new lwb at the end of the log chain
 749          */
 750         mutex_enter(&zilog->zl_lock);
 751         list_insert_tail(&zilog->zl_lwb_list, nlwb);
 752         mutex_exit(&zilog->zl_lock);
 753
 754         /* Record the vdev for later flushing */
 755         zil_add_vdev(zilog, DVA_GET_VDEV(BP_IDENTITY(&(lwb->lwb_blk))));
 756
 757         /*
 758          * kick off the write for the old log block
 759          */
 760         dprintf_bp(&lwb->lwb_blk, "lwb %p txg %llu: ", lwb, txg);
 761         ASSERT(lwb->lwb_zio);
 762         zio_nowait(lwb->lwb_zio);
 763
 764         return (nlwb);
 765 }
 766
 767 static lwb_t *
 768 zil_lwb_commit(zilog_t *zilog, itx_t *itx, lwb_t *lwb)
 769 {
 770         lr_t *lrc = &itx->itx_lr; /* common log record */
 771         lr_write_t *lr = (lr_write_t *)lrc;
 772         uint64_t txg = lrc->lrc_txg;
 773         uint64_t reclen = lrc->lrc_reclen;
 774         uint64_t dlen;
 775
 776         if (lwb == NULL)
 777                 return (NULL);
 778         ASSERT(lwb->lwb_buf != NULL);
 779
 780         if (lrc->lrc_txtype == TX_WRITE && itx->itx_wr_state == WR_NEED_COPY)
 781                 dlen = P2ROUNDUP_TYPED(
 782                     lr->lr_length, sizeof (uint64_t), uint64_t);
 783         else
 784                 dlen = 0;
 785
 786         zilog->zl_cur_used += (reclen + dlen);
 787
 788         zil_lwb_write_init(zilog, lwb);
 789
 790         /*
 791          * If this record won't fit in the current log block, start a new one.
 792          */
 793         if (lwb->lwb_nused + reclen + dlen > ZIL_BLK_DATA_SZ(lwb)) {
 794                 lwb = zil_lwb_write_start(zilog, lwb);
 795                 if (lwb == NULL)
 796                         return (NULL);
 797                 zil_lwb_write_init(zilog, lwb);
 798                 ASSERT(lwb->lwb_nused == 0);
 799                 if (reclen + dlen > ZIL_BLK_DATA_SZ(lwb)) {
 800                         txg_wait_synced(zilog->zl_dmu_pool, txg);
 801                         return (lwb);
 802                 }
 803         }
 804
 805         /*
 806          * Update the lrc_seq, to be log record sequence number. See zil.h
 807          * Then copy the record to the log buffer.
 808          */
 809         lrc->lrc_seq = ++zilog->zl_lr_seq; /* we are single threaded */
 810         bcopy(lrc, lwb->lwb_buf + lwb->lwb_nused, reclen);
 811
 812         /*
 813          * If it's a write, fetch the data or get its blkptr as appropriate.
 814          */
 815         if (lrc->lrc_txtype == TX_WRITE) {
 816                 if (txg > spa_freeze_txg(zilog->zl_spa))
 817                         txg_wait_synced(zilog->zl_dmu_pool, txg);
 818                 if (itx->itx_wr_state != WR_COPIED) {
 819                         char *dbuf;
 820                         int error;
 821
 822                         /* alignment is guaranteed */
 823                         lr = (lr_write_t *)(lwb->lwb_buf + lwb->lwb_nused);
 824                         if (dlen) {
 825                                 ASSERT(itx->itx_wr_state == WR_NEED_COPY);
 826                                 dbuf = lwb->lwb_buf + lwb->lwb_nused + reclen;
 827                                 lr->lr_common.lrc_reclen += dlen;
 828                         } else {
 829                                 ASSERT(itx->itx_wr_state == WR_INDIRECT);
 830                                 dbuf = NULL;
 831                         }
 832                         error = zilog->zl_get_data(
 833                             itx->itx_private, lr, dbuf, lwb->lwb_zio);
 834                         if (error) {
 835                                 ASSERT(error == ENOENT || error == EEXIST ||
 836                                     error == EALREADY);
 837                                 return (lwb);
 838                         }
 839                 }
 840         }
 841
 842         lwb->lwb_nused += reclen + dlen;
 843         lwb->lwb_max_txg = MAX(lwb->lwb_max_txg, txg);
 844         ASSERT3U(lwb->lwb_nused, <=, ZIL_BLK_DATA_SZ(lwb));
 845         ASSERT3U(P2PHASE(lwb->lwb_nused, sizeof (uint64_t)), ==, 0);
 846
 847         return (lwb);
 848 }
 849
 850 itx_t *
 851 zil_itx_create(int txtype, size_t lrsize)
 852 {
 853         itx_t *itx;
 854
 855         lrsize = P2ROUNDUP_TYPED(lrsize, sizeof (uint64_t), size_t);
 856
 857         itx = kmem_alloc(offsetof(itx_t, itx_lr) + lrsize, KM_SLEEP);
 858         itx->itx_lr.lrc_txtype = txtype;
 859         itx->itx_lr.lrc_reclen = lrsize;
 860         itx->itx_lr.lrc_seq = 0;        /* defensive */
 861
 862         return (itx);
 863 }
 864
 865 uint64_t
 866 zil_itx_assign(zilog_t *zilog, itx_t *itx, dmu_tx_t *tx)
 867 {
 868         uint64_t seq;
 869
 870         ASSERT(itx->itx_lr.lrc_seq == 0);
 871
 872         mutex_enter(&zilog->zl_lock);
 873         list_insert_tail(&zilog->zl_itx_list, itx);
 874         zilog->zl_itx_list_sz += itx->itx_lr.lrc_reclen;
 875         itx->itx_lr.lrc_txg = dmu_tx_get_txg(tx);
 876         itx->itx_lr.lrc_seq = seq = ++zilog->zl_itx_seq;
 877         mutex_exit(&zilog->zl_lock);
 878
 879         return (seq);
 880 }
 881
 882 /*
 883  * Free up all in-memory intent log transactions that have now been synced.
 884  */
 885 static void
 886 zil_itx_clean(zilog_t *zilog)
 887 {
 888         uint64_t synced_txg = spa_last_synced_txg(zilog->zl_spa);
 889         uint64_t freeze_txg = spa_freeze_txg(zilog->zl_spa);
 890         list_t clean_list;
 891         itx_t *itx;
 892
 893         list_create(&clean_list, sizeof (itx_t), offsetof(itx_t, itx_node));
 894
 895         mutex_enter(&zilog->zl_lock);
 896         /* wait for a log writer to finish walking list */
 897         while (zilog->zl_writer) {
 898                 cv_wait(&zilog->zl_cv_writer, &zilog->zl_lock);
 899         }
 900
 901         /*
 902          * Move the sync'd log transactions to a separate list so we can call
 903          * kmem_free without holding the zl_lock.
 904          *
 905          * There is no need to set zl_writer as we don't drop zl_lock here
 906          */
 907         while ((itx = list_head(&zilog->zl_itx_list)) != NULL &&
 908             itx->itx_lr.lrc_txg <= MIN(synced_txg, freeze_txg)) {
 909                 list_remove(&zilog->zl_itx_list, itx);
 910                 zilog->zl_itx_list_sz -= itx->itx_lr.lrc_reclen;
 911                 list_insert_tail(&clean_list, itx);
 912         }
 913         cv_broadcast(&zilog->zl_cv_writer);
 914         mutex_exit(&zilog->zl_lock);
 915
 916         /* destroy sync'd log transactions */
 917         while ((itx = list_head(&clean_list)) != NULL) {
 918                 list_remove(&clean_list, itx);
 919                 kmem_free(itx, offsetof(itx_t, itx_lr)
 920                     + itx->itx_lr.lrc_reclen);
 921         }
 922         list_destroy(&clean_list);
 923 }
 924
 925 /*
 926  * If there are any in-memory intent log transactions which have now been
 927  * synced then start up a taskq to free them.
 928  */
 929 void
 930 zil_clean(zilog_t *zilog)
 931 {
 932         itx_t *itx;
 933
 934         mutex_enter(&zilog->zl_lock);
 935         itx = list_head(&zilog->zl_itx_list);
 936         if ((itx != NULL) &&
 937             (itx->itx_lr.lrc_txg <= spa_last_synced_txg(zilog->zl_spa))) {
 938                 (void) taskq_dispatch(zilog->zl_clean_taskq,
 939                     (void (*)(void *))zil_itx_clean, zilog, TQ_NOSLEEP);
 940         }
 941         mutex_exit(&zilog->zl_lock);
 942 }
 943
 944 void
 945 zil_commit_writer(zilog_t *zilog, uint64_t seq, uint64_t foid)
 946 {
 947         uint64_t txg;
 948         uint64_t reclen;
 949         uint64_t commit_seq = 0;
 950         itx_t *itx, *itx_next = (itx_t *)-1;
 951         lwb_t *lwb;
 952         spa_t *spa;
 953
 954         zilog->zl_writer = B_TRUE;
 955         zilog->zl_root_zio = NULL;
 956         spa = zilog->zl_spa;
 957
 958         if (zilog->zl_suspend) {
 959                 lwb = NULL;
 960         } else {
 961                 lwb = list_tail(&zilog->zl_lwb_list);
 962                 if (lwb == NULL) {
 963                         /*
 964                          * Return if there's nothing to flush before we
 965                          * dirty the fs by calling zil_create()
 966                          */
 967                         if (list_is_empty(&zilog->zl_itx_list)) {
 968                                 zilog->zl_writer = B_FALSE;
 969                                 return;
 970                         }
 971                         mutex_exit(&zilog->zl_lock);
 972                         zil_create(zilog);
 973                         mutex_enter(&zilog->zl_lock);
 974                         lwb = list_tail(&zilog->zl_lwb_list);
 975                 }
 976         }
 977
 978         /* Loop through in-memory log transactions filling log blocks. */
 979         DTRACE_PROBE1(zil__cw1, zilog_t *, zilog);
 980         for (;;) {
 981                 /*
 982                  * Find the next itx to push:
 983                  * Push all transactions related to specified foid and all
 984                  * other transactions except TX_WRITE, TX_TRUNCATE,
 985                  * TX_SETATTR and TX_ACL for all other files.
 986                  */
 987                 if (itx_next != (itx_t *)-1)
 988                         itx = itx_next;
 989                 else
 990                         itx = list_head(&zilog->zl_itx_list);
 991                 for (; itx != NULL; itx = list_next(&zilog->zl_itx_list, itx)) {
 992                         if (foid == 0) /* push all foids? */
 993                                 break;
 994                         if (itx->itx_sync) /* push all O_[D]SYNC */
 995                                 break;
 996                         switch (itx->itx_lr.lrc_txtype) {
 997                         case TX_SETATTR:
 998                         case TX_WRITE:
 999                         case TX_TRUNCATE:
1000                         case TX_ACL:
1001                                 /* lr_foid is same offset for these records */
1002                                 if (((lr_write_t *)&itx->itx_lr)->lr_foid
1003                                     != foid) {
1004                                         continue; /* skip this record */
1005                                 }
1006                         }
1007                         break;
1008                 }
1009                 if (itx == NULL)
1010                         break;
1011
1012                 reclen = itx->itx_lr.lrc_reclen;
1013                 if ((itx->itx_lr.lrc_seq > seq) &&
1014                     ((lwb == NULL) || (lwb->lwb_nused == 0) ||
1015                     (lwb->lwb_nused + reclen > ZIL_BLK_DATA_SZ(lwb)))) {
1016                         break;
1017                 }
1018
1019                 /*
1020                  * Save the next pointer.  Even though we soon drop
1021                  * zl_lock all threads that may change the list
1022                  * (another writer or zil_itx_clean) can't do so until
1023                  * they have zl_writer.
1024                  */
1025                 itx_next = list_next(&zilog->zl_itx_list, itx);
1026                 list_remove(&zilog->zl_itx_list, itx);
1027                 mutex_exit(&zilog->zl_lock);
1028                 txg = itx->itx_lr.lrc_txg;
1029                 ASSERT(txg);
1030
1031                 if (txg > spa_last_synced_txg(spa) ||
1032                     txg > spa_freeze_txg(spa))
1033                         lwb = zil_lwb_commit(zilog, itx, lwb);
1034                 kmem_free(itx, offsetof(itx_t, itx_lr)
1035                     + itx->itx_lr.lrc_reclen);
1036                 mutex_enter(&zilog->zl_lock);
1037                 zilog->zl_itx_list_sz -= reclen;
1038         }
1039         DTRACE_PROBE1(zil__cw2, zilog_t *, zilog);
1040         /* determine commit sequence number */
1041         itx = list_head(&zilog->zl_itx_list);
1042         if (itx)
1043                 commit_seq = itx->itx_lr.lrc_seq;
1044         else
1045                 commit_seq = zilog->zl_itx_seq;
1046         mutex_exit(&zilog->zl_lock);
1047
1048         /* write the last block out */
1049         if (lwb != NULL && lwb->lwb_zio != NULL)
1050                 lwb = zil_lwb_write_start(zilog, lwb);
1051
1052         zilog->zl_prev_used = zilog->zl_cur_used;
1053         zilog->zl_cur_used = 0;
1054
1055         /*
1056          * Wait if necessary for the log blocks to be on stable storage.
1057          */
1058         if (zilog->zl_root_zio) {
1059                 DTRACE_PROBE1(zil__cw3, zilog_t *, zilog);
1060                 (void) zio_wait(zilog->zl_root_zio);
1061                 DTRACE_PROBE1(zil__cw4, zilog_t *, zilog);
1062                 if (!zfs_nocacheflush)
1063                         zil_flush_vdevs(zilog);
1064         }
1065
1066         if (zilog->zl_log_error || lwb == NULL) {
1067                 zilog->zl_log_error = 0;
1068                 txg_wait_synced(zilog->zl_dmu_pool, 0);
1069         }
1070
1071         mutex_enter(&zilog->zl_lock);
1072         zilog->zl_writer = B_FALSE;
1073
1074         ASSERT3U(commit_seq, >=, zilog->zl_commit_seq);
1075         zilog->zl_commit_seq = commit_seq;
1076 }
1077
1078 /*
1079  * Push zfs transactions to stable storage up to the supplied sequence number.
1080  * If foid is 0 push out all transactions, otherwise push only those
1081  * for that file or might have been used to create that file.
1082  */
1083 void
1084 zil_commit(zilog_t *zilog, uint64_t seq, uint64_t foid)
1085 {
1086         if (zilog == NULL || seq == 0)
1087                 return;
1088
1089         mutex_enter(&zilog->zl_lock);
1090
1091         seq = MIN(seq, zilog->zl_itx_seq);      /* cap seq at largest itx seq */
1092
1093         while (zilog->zl_writer) {
1094                 cv_wait(&zilog->zl_cv_writer, &zilog->zl_lock);
1095                 if (seq < zilog->zl_commit_seq) {
1096                         mutex_exit(&zilog->zl_lock);
1097                         return;
1098                 }
1099         }
1100         zil_commit_writer(zilog, seq, foid); /* drops zl_lock */
1101         /* wake up others waiting on the commit */
1102         cv_broadcast(&zilog->zl_cv_writer);
1103         mutex_exit(&zilog->zl_lock);
1104 }
1105
1106 /*
1107  * Called in syncing context to free committed log blocks and update log header.
1108  */
1109 void
1110 zil_sync(zilog_t *zilog, dmu_tx_t *tx)
1111 {
1112         zil_header_t *zh = zil_header_in_syncing_context(zilog);
1113         uint64_t txg = dmu_tx_get_txg(tx);
1114         spa_t *spa = zilog->zl_spa;
1115         lwb_t *lwb;
1116
1117         mutex_enter(&zilog->zl_lock);
1118
1119         ASSERT(zilog->zl_stop_sync == 0);
1120
1121         zh->zh_replay_seq = zilog->zl_replay_seq[txg & TXG_MASK];
1122
1123         if (zilog->zl_destroy_txg == txg) {
1124                 blkptr_t blk = zh->zh_log;
1125
1126                 ASSERT(list_head(&zilog->zl_lwb_list) == NULL);
1127                 ASSERT(spa_sync_pass(spa) == 1);
1128
1129                 bzero(zh, sizeof (zil_header_t));
1130                 bzero(zilog->zl_replay_seq, sizeof (zilog->zl_replay_seq));
1131
1132                 if (zilog->zl_keep_first) {
1133                         /*
1134                          * If this block was part of log chain that couldn't
1135                          * be claimed because a device was missing during
1136                          * zil_claim(), but that device later returns,
1137                          * then this block could erroneously appear valid.
1138                          * To guard against this, assign a new GUID to the new
1139                          * log chain so it doesn't matter what blk points to.
1140                          */
1141                         zil_init_log_chain(zilog, &blk);
1142                         zh->zh_log = blk;
1143                 }
1144         }
1145
1146         for (;;) {
1147                 lwb = list_head(&zilog->zl_lwb_list);
1148                 if (lwb == NULL) {
1149                         mutex_exit(&zilog->zl_lock);
1150                         return;
1151                 }
1152                 zh->zh_log = lwb->lwb_blk;
1153                 if (lwb->lwb_buf != NULL || lwb->lwb_max_txg > txg)
1154                         break;
1155                 list_remove(&zilog->zl_lwb_list, lwb);
1156                 zio_free_blk(spa, &lwb->lwb_blk, txg);
1157                 kmem_cache_free(zil_lwb_cache, lwb);
1158
1159                 /*
1160                  * If we don't have anything left in the lwb list then
1161                  * we've had an allocation failure and we need to zero
1162                  * out the zil_header blkptr so that we don't end
1163                  * up freeing the same block twice.
1164                  */
1165                 if (list_head(&zilog->zl_lwb_list) == NULL)
1166                         BP_ZERO(&zh->zh_log);
1167         }
1168         mutex_exit(&zilog->zl_lock);
1169 }
1170
1171 void
1172 zil_init(void)
1173 {
1174         zil_lwb_cache = kmem_cache_create("zil_lwb_cache",
1175             sizeof (struct lwb), 0, NULL, NULL, NULL, NULL, NULL, 0);
1176 }
1177
1178 void
1179 zil_fini(void)
1180 {
1181         kmem_cache_destroy(zil_lwb_cache);
1182 }
1183
1184 zilog_t *
1185 zil_alloc(objset_t *os, zil_header_t *zh_phys)
1186 {
1187         zilog_t *zilog;
1188
1189         zilog = kmem_zalloc(sizeof (zilog_t), KM_SLEEP);
1190
1191         zilog->zl_header = zh_phys;
1192         zilog->zl_os = os;
1193         zilog->zl_spa = dmu_objset_spa(os);
1194         zilog->zl_dmu_pool = dmu_objset_pool(os);
1195         zilog->zl_destroy_txg = TXG_INITIAL - 1;
1196
1197         mutex_init(&zilog->zl_lock, NULL, MUTEX_DEFAULT, NULL);
1198         cv_init(&zilog->zl_cv_writer, NULL, CV_DEFAULT, NULL);
1199         cv_init(&zilog->zl_cv_suspend, NULL, CV_DEFAULT, NULL);
1200
1201         list_create(&zilog->zl_itx_list, sizeof (itx_t),
1202             offsetof(itx_t, itx_node));
1203
1204         list_create(&zilog->zl_lwb_list, sizeof (lwb_t),
1205             offsetof(lwb_t, lwb_node));
1206
1207         list_create(&zilog->zl_vdev_list, sizeof (zil_vdev_t),
1208             offsetof(zil_vdev_t, vdev_seq_node));
1209
1210         return (zilog);
1211 }
1212
1213 void
1214 zil_free(zilog_t *zilog)
1215 {
1216         lwb_t *lwb;
1217         zil_vdev_t *zv;
1218
1219         zilog->zl_stop_sync = 1;
1220
1221         while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
1222                 list_remove(&zilog->zl_lwb_list, lwb);
1223                 if (lwb->lwb_buf != NULL)
1224                         zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
1225                 kmem_cache_free(zil_lwb_cache, lwb);
1226         }
1227         list_destroy(&zilog->zl_lwb_list);
1228
1229         while ((zv = list_head(&zilog->zl_vdev_list)) != NULL) {
1230                 list_remove(&zilog->zl_vdev_list, zv);
1231                 kmem_free(zv, sizeof (zil_vdev_t));
1232         }
1233         list_destroy(&zilog->zl_vdev_list);
1234
1235         ASSERT(list_head(&zilog->zl_itx_list) == NULL);
1236         list_destroy(&zilog->zl_itx_list);
1237         cv_destroy(&zilog->zl_cv_suspend);
1238         cv_destroy(&zilog->zl_cv_writer);
1239         mutex_destroy(&zilog->zl_lock);
1240
1241         kmem_free(zilog, sizeof (zilog_t));
1242 }
1243
1244 /*
1245  * return true if the initial log block is not valid
1246  */
1247 static int
1248 zil_empty(zilog_t *zilog)
1249 {
1250         const zil_header_t *zh = zilog->zl_header;
1251         arc_buf_t *abuf = NULL;
1252
1253         if (BP_IS_HOLE(&zh->zh_log))
1254                 return (1);
1255
1256         if (zil_read_log_block(zilog, &zh->zh_log, &abuf) != 0)
1257                 return (1);
1258
1259         VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);
1260         return (0);
1261 }
1262
1263 /*
1264  * Open an intent log.
1265  */
1266 zilog_t *
1267 zil_open(objset_t *os, zil_get_data_t *get_data)
1268 {
1269         zilog_t *zilog = dmu_objset_zil(os);
1270
1271         zilog->zl_get_data = get_data;
1272         zilog->zl_clean_taskq = taskq_create("zil_clean", 1, minclsyspri,
1273             2, 2, TASKQ_PREPOPULATE);
1274
1275         return (zilog);
1276 }
1277
1278 /*
1279  * Close an intent log.
1280  */
1281 void
1282 zil_close(zilog_t *zilog)
1283 {
1284         /*
1285          * If the log isn't already committed, mark the objset dirty
1286          * (so zil_sync() will be called) and wait for that txg to sync.
1287          */
1288         if (!zil_is_committed(zilog)) {
1289                 uint64_t txg;
1290                 dmu_tx_t *tx = dmu_tx_create(zilog->zl_os);
1291                 (void) dmu_tx_assign(tx, TXG_WAIT);
1292                 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
1293                 txg = dmu_tx_get_txg(tx);
1294                 dmu_tx_commit(tx);
1295                 txg_wait_synced(zilog->zl_dmu_pool, txg);
1296         }
1297
1298         taskq_destroy(zilog->zl_clean_taskq);
1299         zilog->zl_clean_taskq = NULL;
1300         zilog->zl_get_data = NULL;
1301
1302         zil_itx_clean(zilog);
1303         ASSERT(list_head(&zilog->zl_itx_list) == NULL);
1304 }
1305
1306 /*
1307  * Suspend an intent log.  While in suspended mode, we still honor
1308  * synchronous semantics, but we rely on txg_wait_synced() to do it.
1309  * We suspend the log briefly when taking a snapshot so that the snapshot
1310  * contains all the data it's supposed to, and has an empty intent log.
1311  */
1312 int
1313 zil_suspend(zilog_t *zilog)
1314 {
1315         const zil_header_t *zh = zilog->zl_header;
1316
1317         mutex_enter(&zilog->zl_lock);
1318         if (zh->zh_claim_txg != 0) {            /* unplayed log */
1319                 mutex_exit(&zilog->zl_lock);
1320                 return (EBUSY);
1321         }
1322         if (zilog->zl_suspend++ != 0) {
1323                 /*
1324                  * Someone else already began a suspend.
1325                  * Just wait for them to finish.
1326                  */
1327                 while (zilog->zl_suspending)
1328                         cv_wait(&zilog->zl_cv_suspend, &zilog->zl_lock);
1329                 ASSERT(BP_IS_HOLE(&zh->zh_log));
1330                 mutex_exit(&zilog->zl_lock);
1331                 return (0);
1332         }
1333         zilog->zl_suspending = B_TRUE;
1334         mutex_exit(&zilog->zl_lock);
1335
1336         zil_commit(zilog, UINT64_MAX, 0);
1337
1338         /*
1339          * Wait for any in-flight log writes to complete.
1340          */
1341         mutex_enter(&zilog->zl_lock);
1342         while (zilog->zl_writer)
1343                 cv_wait(&zilog->zl_cv_writer, &zilog->zl_lock);
1344         mutex_exit(&zilog->zl_lock);
1345
1346         zil_destroy(zilog, B_FALSE);
1347
1348         mutex_enter(&zilog->zl_lock);
1349         ASSERT(BP_IS_HOLE(&zh->zh_log));
1350         zilog->zl_suspending = B_FALSE;
1351         cv_broadcast(&zilog->zl_cv_suspend);
1352         mutex_exit(&zilog->zl_lock);
1353
1354         return (0);
1355 }
1356
1357 void
1358 zil_resume(zilog_t *zilog)
1359 {
1360         mutex_enter(&zilog->zl_lock);
1361         ASSERT(zilog->zl_suspend != 0);
1362         zilog->zl_suspend--;
1363         mutex_exit(&zilog->zl_lock);
1364 }
1365
1366 typedef struct zil_replay_arg {
1367         objset_t        *zr_os;
1368         zil_replay_func_t **zr_replay;
1369         void            *zr_arg;
1370         uint64_t        *zr_txgp;
1371         boolean_t       zr_byteswap;
1372         char            *zr_lrbuf;
1373 } zil_replay_arg_t;
1374
1375 static void
1376 zil_replay_log_record(zilog_t *zilog, lr_t *lr, void *zra, uint64_t claim_txg)
1377 {
1378         zil_replay_arg_t *zr = zra;
1379         const zil_header_t *zh = zilog->zl_header;
1380         uint64_t reclen = lr->lrc_reclen;
1381         uint64_t txtype = lr->lrc_txtype;
1382         char *name;
1383         int pass, error, sunk;
1384
1385         if (zilog->zl_stop_replay)
1386                 return;
1387
1388         if (lr->lrc_txg < claim_txg)            /* already committed */
1389                 return;
1390
1391         if (lr->lrc_seq <= zh->zh_replay_seq)   /* already replayed */
1392                 return;
1393
1394         /*
1395          * Make a copy of the data so we can revise and extend it.
1396          */
1397         bcopy(lr, zr->zr_lrbuf, reclen);
1398
1399         /*
1400          * The log block containing this lr may have been byteswapped
1401          * so that we can easily examine common fields like lrc_txtype.
1402          * However, the log is a mix of different data types, and only the
1403          * replay vectors know how to byteswap their records.  Therefore, if
1404          * the lr was byteswapped, undo it before invoking the replay vector.
1405          */
1406         if (zr->zr_byteswap)
1407                 byteswap_uint64_array(zr->zr_lrbuf, reclen);
1408
1409         /*
1410          * If this is a TX_WRITE with a blkptr, suck in the data.
1411          */
1412         if (txtype == TX_WRITE && reclen == sizeof (lr_write_t)) {
1413                 lr_write_t *lrw = (lr_write_t *)lr;
1414                 blkptr_t *wbp = &lrw->lr_blkptr;
1415                 uint64_t wlen = lrw->lr_length;
1416                 char *wbuf = zr->zr_lrbuf + reclen;
1417
1418                 if (BP_IS_HOLE(wbp)) {  /* compressed to a hole */
1419                         bzero(wbuf, wlen);
1420                 } else {
1421                         /*
1422                          * A subsequent write may have overwritten this block,
1423                          * in which case wbp may have been been freed and
1424                          * reallocated, and our read of wbp may fail with a
1425                          * checksum error.  We can safely ignore this because
1426                          * the later write will provide the correct data.
1427                          */
1428                         zbookmark_t zb;
1429
1430                         zb.zb_objset = dmu_objset_id(zilog->zl_os);
1431                         zb.zb_object = lrw->lr_foid;
1432                         zb.zb_level = -1;
1433                         zb.zb_blkid = lrw->lr_offset / BP_GET_LSIZE(wbp);
1434
1435                         (void) zio_wait(zio_read(NULL, zilog->zl_spa,
1436                             wbp, wbuf, BP_GET_LSIZE(wbp), NULL, NULL,
1437                             ZIO_PRIORITY_SYNC_READ,
1438                             ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE, &zb));
1439                         (void) memmove(wbuf, wbuf + lrw->lr_blkoff, wlen);
1440                 }
1441         }
1442
1443         /*
1444          * We must now do two things atomically: replay this log record,
1445          * and update the log header to reflect the fact that we did so.
1446          * We use the DMU's ability to assign into a specific txg to do this.
1447          */
1448         for (pass = 1, sunk = B_FALSE; /* CONSTANTCONDITION */; pass++) {
1449                 uint64_t replay_txg;
1450                 dmu_tx_t *replay_tx;
1451
1452                 replay_tx = dmu_tx_create(zr->zr_os);
1453                 error = dmu_tx_assign(replay_tx, TXG_WAIT);
1454                 if (error) {
1455                         dmu_tx_abort(replay_tx);
1456                         break;
1457                 }
1458
1459                 replay_txg = dmu_tx_get_txg(replay_tx);
1460
1461                 if (txtype == 0 || txtype >= TX_MAX_TYPE) {
1462                         error = EINVAL;
1463                 } else {
1464                         /*
1465                          * On the first pass, arrange for the replay vector
1466                          * to fail its dmu_tx_assign().  That's the only way
1467                          * to ensure that those code paths remain well tested.
1468                          */
1469                         *zr->zr_txgp = replay_txg - (pass == 1);
1470                         error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lrbuf,
1471                             zr->zr_byteswap);
1472                         *zr->zr_txgp = TXG_NOWAIT;
1473                 }
1474
1475                 if (error == 0) {
1476                         dsl_dataset_dirty(dmu_objset_ds(zr->zr_os), replay_tx);
1477                         zilog->zl_replay_seq[replay_txg & TXG_MASK] =
1478                             lr->lrc_seq;
1479                 }
1480
1481                 dmu_tx_commit(replay_tx);
1482
1483                 if (!error)
1484                         return;
1485
1486                 /*
1487                  * The DMU's dnode layer doesn't see removes until the txg
1488                  * commits, so a subsequent claim can spuriously fail with
1489                  * EEXIST. So if we receive any error other than ERESTART
1490                  * we try syncing out any removes then retrying the
1491                  * transaction.
1492                  */
1493                 if (error != ERESTART && !sunk) {
1494                         txg_wait_synced(spa_get_dsl(zilog->zl_spa), 0);
1495                         sunk = B_TRUE;
1496                         continue; /* retry */
1497                 }
1498
1499                 if (error != ERESTART)
1500                         break;
1501
1502                 if (pass != 1)
1503                         txg_wait_open(spa_get_dsl(zilog->zl_spa),
1504                             replay_txg + 1);
1505
1506                 dprintf("pass %d, retrying\n", pass);
1507         }
1508
1509         ASSERT(error && error != ERESTART);
1510         name = kmem_alloc(MAXNAMELEN, KM_SLEEP);
1511         dmu_objset_name(zr->zr_os, name);
1512         cmn_err(CE_WARN, "ZFS replay transaction error %d, "
1513             "dataset %s, seq 0x%llx, txtype %llu\n",
1514             error, name, (u_longlong_t)lr->lrc_seq, (u_longlong_t)txtype);
1515         zilog->zl_stop_replay = 1;
1516         kmem_free(name, MAXNAMELEN);
1517 }
1518
1519 /* ARGSUSED */
1520 static void
1521 zil_incr_blks(zilog_t *zilog, blkptr_t *bp, void *arg, uint64_t claim_txg)
1522 {
1523         zilog->zl_replay_blks++;
1524 }
1525
1526 /*
1527  * If this dataset has a non-empty intent log, replay it and destroy it.
1528  */
1529 void
1530 zil_replay(objset_t *os, void *arg, uint64_t *txgp,
1531         zil_replay_func_t *replay_func[TX_MAX_TYPE])
1532 {
1533         zilog_t *zilog = dmu_objset_zil(os);
1534         const zil_header_t *zh = zilog->zl_header;
1535         zil_replay_arg_t zr;
1536
1537         if (zil_empty(zilog)) {
1538                 zil_destroy(zilog, B_TRUE);
1539                 return;
1540         }
1541         //printf("ZFS: Replaying ZIL on %s...\n", os->os->os_spa->spa_name);
1542
1543         zr.zr_os = os;
1544         zr.zr_replay = replay_func;
1545         zr.zr_arg = arg;
1546         zr.zr_txgp = txgp;
1547         zr.zr_byteswap = BP_SHOULD_BYTESWAP(&zh->zh_log);
1548         zr.zr_lrbuf = kmem_alloc(2 * SPA_MAXBLOCKSIZE, KM_SLEEP);
1549
1550         /*
1551          * Wait for in-progress removes to sync before starting replay.
1552          */
1553         txg_wait_synced(zilog->zl_dmu_pool, 0);
1554
1555         zilog->zl_stop_replay = 0;
1556         zilog->zl_replay_time = LBOLT;
1557         ASSERT(zilog->zl_replay_blks == 0);
1558         (void) zil_parse(zilog, zil_incr_blks, zil_replay_log_record, &zr,
1559             zh->zh_claim_txg);
1560         kmem_free(zr.zr_lrbuf, 2 * SPA_MAXBLOCKSIZE);
1561
1562         zil_destroy(zilog, B_FALSE);
1563         //printf("ZFS: Replay of ZIL on %s finished.\n", os->os->os_spa->spa_name);
1564 }
1565
1566 /*
1567  * Report whether all transactions are committed
1568  */
1569 int
1570 zil_is_committed(zilog_t *zilog)
1571 {
1572         lwb_t *lwb;
1573         int ret;
1574
1575         mutex_enter(&zilog->zl_lock);
1576         while (zilog->zl_writer)
1577                 cv_wait(&zilog->zl_cv_writer, &zilog->zl_lock);
1578
1579         /* recent unpushed intent log transactions? */
1580         if (!list_is_empty(&zilog->zl_itx_list)) {
1581                 ret = B_FALSE;
1582                 goto out;
1583         }
1584
1585         /* intent log never used? */
1586         lwb = list_head(&zilog->zl_lwb_list);
1587         if (lwb == NULL) {
1588                 ret = B_TRUE;
1589                 goto out;
1590         }
1591
1592         /*
1593          * more than 1 log buffer means zil_sync() hasn't yet freed
1594          * entries after a txg has committed
1595          */
1596         if (list_next(&zilog->zl_lwb_list, lwb)) {
1597                 ret = B_FALSE;
1598                 goto out;
1599         }
1600
1601         ASSERT(zil_empty(zilog));
1602         ret = B_TRUE;
1603 out:
1604         cv_broadcast(&zilog->zl_cv_writer);
1605         mutex_exit(&zilog->zl_lock);
1606         return (ret);
1607 }