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 (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
  23  * Copyright (c) 2013 by Delphix. All rights reserved.
  24  */
  25 
  26 /* Portions Copyright 2010 Robert Milkowski */
  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_impl.h>
  39 #include <sys/dmu_tx.h>
  40 #include <sys/dsl_pool.h>
  41 
  42 /*
  43  * The zfs intent log (ZIL) saves transaction records of system calls
  44  * that change the file system in memory with enough information
  45  * to be able to replay them. These are stored in memory until
  46  * either the DMU transaction group (txg) commits them to the stable pool
  47  * and they can be discarded, or they are flushed to the stable log
  48  * (also in the pool) due to a fsync, O_DSYNC or other synchronous
  49  * requirement. In the event of a panic or power fail then those log
  50  * records (transactions) are replayed.
  51  *
  52  * There is one ZIL per file system. Its on-disk (pool) format consists
  53  * of 3 parts:
  54  *
  55  *      - ZIL header
  56  *      - ZIL blocks
  57  *      - ZIL records
  58  *
  59  * A log record holds a system call transaction. Log blocks can
  60  * hold many log records and the blocks are chained together.
  61  * Each ZIL block contains a block pointer (blkptr_t) to the next
  62  * ZIL block in the chain. The ZIL header points to the first
  63  * block in the chain. Note there is not a fixed place in the pool
  64  * to hold blocks. They are dynamically allocated and freed as
  65  * needed from the blocks available. Figure X shows the ZIL structure:
  66  */
  67 
  68 /*
  69  * Disable intent logging replay.  This global ZIL switch affects all pools.
  70  */
  71 int zil_replay_disable = 0;
  72 
  73 /*
  74  * Tunable parameter for debugging or performance analysis.  Setting
  75  * zfs_nocacheflush will cause corruption on power loss if a volatile
  76  * out-of-order write cache is enabled.
  77  */
  78 boolean_t zfs_nocacheflush = B_FALSE;
  79 
  80 static kmem_cache_t *zil_lwb_cache;
  81 
  82 static void zil_async_to_sync(zilog_t *zilog, uint64_t foid);
  83 
  84 #define LWB_EMPTY(lwb) ((BP_GET_LSIZE(&lwb->lwb_blk) - \
  85     sizeof (zil_chain_t)) == (lwb->lwb_sz - lwb->lwb_nused))
  86 
  87 
  88 /*
  89  * ziltest is by and large an ugly hack, but very useful in
  90  * checking replay without tedious work.
  91  * When running ziltest we want to keep all itx's and so maintain
  92  * a single list in the zl_itxg[] that uses a high txg: ZILTEST_TXG
  93  * We subtract TXG_CONCURRENT_STATES to allow for common code.
  94  */
  95 #define ZILTEST_TXG (UINT64_MAX - TXG_CONCURRENT_STATES)
  96 
  97 static int
  98 zil_bp_compare(const void *x1, const void *x2)
  99 {
 100         const dva_t *dva1 = &((zil_bp_node_t *)x1)->zn_dva;
 101         const dva_t *dva2 = &((zil_bp_node_t *)x2)->zn_dva;
 102 
 103         if (DVA_GET_VDEV(dva1) < DVA_GET_VDEV(dva2))
 104                 return (-1);
 105         if (DVA_GET_VDEV(dva1) > DVA_GET_VDEV(dva2))
 106                 return (1);
 107 
 108         if (DVA_GET_OFFSET(dva1) < DVA_GET_OFFSET(dva2))
 109                 return (-1);
 110         if (DVA_GET_OFFSET(dva1) > DVA_GET_OFFSET(dva2))
 111                 return (1);
 112 
 113         return (0);
 114 }
 115 
 116 static void
 117 zil_bp_tree_init(zilog_t *zilog)
 118 {
 119         avl_create(&zilog->zl_bp_tree, zil_bp_compare,
 120             sizeof (zil_bp_node_t), offsetof(zil_bp_node_t, zn_node));
 121 }
 122 
 123 static void
 124 zil_bp_tree_fini(zilog_t *zilog)
 125 {
 126         avl_tree_t *t = &zilog->zl_bp_tree;
 127         zil_bp_node_t *zn;
 128         void *cookie = NULL;
 129 
 130         while ((zn = avl_destroy_nodes(t, &cookie)) != NULL)
 131                 kmem_free(zn, sizeof (zil_bp_node_t));
 132 
 133         avl_destroy(t);
 134 }
 135 
 136 int
 137 zil_bp_tree_add(zilog_t *zilog, const blkptr_t *bp)
 138 {
 139         avl_tree_t *t = &zilog->zl_bp_tree;
 140         const dva_t *dva;
 141         zil_bp_node_t *zn;
 142         avl_index_t where;
 143 
 144         if (BP_IS_EMBEDDED(bp))
 145                 return (0);
 146 
 147         dva = BP_IDENTITY(bp);
 148 
 149         if (avl_find(t, dva, &where) != NULL)
 150                 return (SET_ERROR(EEXIST));
 151 
 152         zn = kmem_alloc(sizeof (zil_bp_node_t), KM_SLEEP);
 153         zn->zn_dva = *dva;
 154         avl_insert(t, zn, where);
 155 
 156         return (0);
 157 }
 158 
 159 static zil_header_t *
 160 zil_header_in_syncing_context(zilog_t *zilog)
 161 {
 162         return ((zil_header_t *)zilog->zl_header);
 163 }
 164 
 165 static void
 166 zil_init_log_chain(zilog_t *zilog, blkptr_t *bp)
 167 {
 168         zio_cksum_t *zc = &bp->blk_cksum;
 169 
 170         zc->zc_word[ZIL_ZC_GUID_0] = spa_get_random(-1ULL);
 171         zc->zc_word[ZIL_ZC_GUID_1] = spa_get_random(-1ULL);
 172         zc->zc_word[ZIL_ZC_OBJSET] = dmu_objset_id(zilog->zl_os);
 173         zc->zc_word[ZIL_ZC_SEQ] = 1ULL;
 174 }
 175 
 176 /*
 177  * Read a log block and make sure it's valid.
 178  */
 179 static int
 180 zil_read_log_block(zilog_t *zilog, const blkptr_t *bp, blkptr_t *nbp, void *dst,
 181     char **end)
 182 {
 183         enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;
 184         uint32_t aflags = ARC_WAIT;
 185         arc_buf_t *abuf = NULL;
 186         zbookmark_t zb;
 187         int error;
 188 
 189         if (zilog->zl_header->zh_claim_txg == 0)
 190                 zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB;
 191 
 192         if (!(zilog->zl_header->zh_flags & ZIL_CLAIM_LR_SEQ_VALID))
 193                 zio_flags |= ZIO_FLAG_SPECULATIVE;
 194 
 195         SET_BOOKMARK(&zb, bp->blk_cksum.zc_word[ZIL_ZC_OBJSET],
 196             ZB_ZIL_OBJECT, ZB_ZIL_LEVEL, bp->blk_cksum.zc_word[ZIL_ZC_SEQ]);
 197 
 198         error = arc_read(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf,
 199             ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);
 200 
 201         if (error == 0) {
 202                 zio_cksum_t cksum = bp->blk_cksum;
 203 
 204                 /*
 205                  * Validate the checksummed log block.
 206                  *
 207                  * Sequence numbers should be... sequential.  The checksum
 208                  * verifier for the next block should be bp's checksum plus 1.
 209                  *
 210                  * Also check the log chain linkage and size used.
 211                  */
 212                 cksum.zc_word[ZIL_ZC_SEQ]++;
 213 
 214                 if (BP_GET_CHECKSUM(bp) == ZIO_CHECKSUM_ZILOG2) {
 215                         zil_chain_t *zilc = abuf->b_data;
 216                         char *lr = (char *)(zilc + 1);
 217                         uint64_t len = zilc->zc_nused - sizeof (zil_chain_t);
 218 
 219                         if (bcmp(&cksum, &zilc->zc_next_blk.blk_cksum,
 220                             sizeof (cksum)) || BP_IS_HOLE(&zilc->zc_next_blk)) {
 221                                 error = SET_ERROR(ECKSUM);
 222                         } else {
 223                                 bcopy(lr, dst, len);
 224                                 *end = (char *)dst + len;
 225                                 *nbp = zilc->zc_next_blk;
 226                         }
 227                 } else {
 228                         char *lr = abuf->b_data;
 229                         uint64_t size = BP_GET_LSIZE(bp);
 230                         zil_chain_t *zilc = (zil_chain_t *)(lr + size) - 1;
 231 
 232                         if (bcmp(&cksum, &zilc->zc_next_blk.blk_cksum,
 233                             sizeof (cksum)) || BP_IS_HOLE(&zilc->zc_next_blk) ||
 234                             (zilc->zc_nused > (size - sizeof (*zilc)))) {
 235                                 error = SET_ERROR(ECKSUM);
 236                         } else {
 237                                 bcopy(lr, dst, zilc->zc_nused);
 238                                 *end = (char *)dst + zilc->zc_nused;
 239                                 *nbp = zilc->zc_next_blk;
 240                         }
 241                 }
 242 
 243                 VERIFY(arc_buf_remove_ref(abuf, &abuf));
 244         }
 245 
 246         return (error);
 247 }
 248 
 249 /*
 250  * Read a TX_WRITE log data block.
 251  */
 252 static int
 253 zil_read_log_data(zilog_t *zilog, const lr_write_t *lr, void *wbuf)
 254 {
 255         enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;
 256         const blkptr_t *bp = &lr->lr_blkptr;
 257         uint32_t aflags = ARC_WAIT;
 258         arc_buf_t *abuf = NULL;
 259         zbookmark_t zb;
 260         int error;
 261 
 262         if (BP_IS_HOLE(bp)) {
 263                 if (wbuf != NULL)
 264                         bzero(wbuf, MAX(BP_GET_LSIZE(bp), lr->lr_length));
 265                 return (0);
 266         }
 267 
 268         if (zilog->zl_header->zh_claim_txg == 0)
 269                 zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB;
 270 
 271         SET_BOOKMARK(&zb, dmu_objset_id(zilog->zl_os), lr->lr_foid,
 272             ZB_ZIL_LEVEL, lr->lr_offset / BP_GET_LSIZE(bp));
 273 
 274         error = arc_read(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf,
 275             ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);
 276 
 277         if (error == 0) {
 278                 if (wbuf != NULL)
 279                         bcopy(abuf->b_data, wbuf, arc_buf_size(abuf));
 280                 (void) arc_buf_remove_ref(abuf, &abuf);
 281         }
 282 
 283         return (error);
 284 }
 285 
 286 /*
 287  * Parse the intent log, and call parse_func for each valid record within.
 288  */
 289 int
 290 zil_parse(zilog_t *zilog, zil_parse_blk_func_t *parse_blk_func,
 291     zil_parse_lr_func_t *parse_lr_func, void *arg, uint64_t txg)
 292 {
 293         const zil_header_t *zh = zilog->zl_header;
 294         boolean_t claimed = !!zh->zh_claim_txg;
 295         uint64_t claim_blk_seq = claimed ? zh->zh_claim_blk_seq : UINT64_MAX;
 296         uint64_t claim_lr_seq = claimed ? zh->zh_claim_lr_seq : UINT64_MAX;
 297         uint64_t max_blk_seq = 0;
 298         uint64_t max_lr_seq = 0;
 299         uint64_t blk_count = 0;
 300         uint64_t lr_count = 0;
 301         blkptr_t blk, next_blk;
 302         char *lrbuf, *lrp;
 303         int error = 0;
 304 
 305         /*
 306          * Old logs didn't record the maximum zh_claim_lr_seq.
 307          */
 308         if (!(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID))
 309                 claim_lr_seq = UINT64_MAX;
 310 
 311         /*
 312          * Starting at the block pointed to by zh_log we read the log chain.
 313          * For each block in the chain we strongly check that block to
 314          * ensure its validity.  We stop when an invalid block is found.
 315          * For each block pointer in the chain we call parse_blk_func().
 316          * For each record in each valid block we call parse_lr_func().
 317          * If the log has been claimed, stop if we encounter a sequence
 318          * number greater than the highest claimed sequence number.
 319          */
 320         lrbuf = zio_buf_alloc(SPA_MAXBLOCKSIZE);
 321         zil_bp_tree_init(zilog);
 322 
 323         for (blk = zh->zh_log; !BP_IS_HOLE(&blk); blk = next_blk) {
 324                 uint64_t blk_seq = blk.blk_cksum.zc_word[ZIL_ZC_SEQ];
 325                 int reclen;
 326                 char *end;
 327 
 328                 if (blk_seq > claim_blk_seq)
 329                         break;
 330                 if ((error = parse_blk_func(zilog, &blk, arg, txg)) != 0)
 331                         break;
 332                 ASSERT3U(max_blk_seq, <, blk_seq);
 333                 max_blk_seq = blk_seq;
 334                 blk_count++;
 335 
 336                 if (max_lr_seq == claim_lr_seq && max_blk_seq == claim_blk_seq)
 337                         break;
 338 
 339                 error = zil_read_log_block(zilog, &blk, &next_blk, lrbuf, &end);
 340                 if (error != 0)
 341                         break;
 342 
 343                 for (lrp = lrbuf; lrp < end; lrp += reclen) {
 344                         lr_t *lr = (lr_t *)lrp;
 345                         reclen = lr->lrc_reclen;
 346                         ASSERT3U(reclen, >=, sizeof (lr_t));
 347                         if (lr->lrc_seq > claim_lr_seq)
 348                                 goto done;
 349                         if ((error = parse_lr_func(zilog, lr, arg, txg)) != 0)
 350                                 goto done;
 351                         ASSERT3U(max_lr_seq, <, lr->lrc_seq);
 352                         max_lr_seq = lr->lrc_seq;
 353                         lr_count++;
 354                 }
 355         }
 356 done:
 357         zilog->zl_parse_error = error;
 358         zilog->zl_parse_blk_seq = max_blk_seq;
 359         zilog->zl_parse_lr_seq = max_lr_seq;
 360         zilog->zl_parse_blk_count = blk_count;
 361         zilog->zl_parse_lr_count = lr_count;
 362 
 363         ASSERT(!claimed || !(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID) ||
 364             (max_blk_seq == claim_blk_seq && max_lr_seq == claim_lr_seq));
 365 
 366         zil_bp_tree_fini(zilog);
 367         zio_buf_free(lrbuf, SPA_MAXBLOCKSIZE);
 368 
 369         return (error);
 370 }
 371 
 372 static int
 373 zil_claim_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t first_txg)
 374 {
 375         /*
 376          * Claim log block if not already committed and not already claimed.
 377          * If tx == NULL, just verify that the block is claimable.
 378          */
 379         if (BP_IS_HOLE(bp) || bp->blk_birth < first_txg ||
 380             zil_bp_tree_add(zilog, bp) != 0)
 381                 return (0);
 382 
 383         return (zio_wait(zio_claim(NULL, zilog->zl_spa,
 384             tx == NULL ? 0 : first_txg, bp, spa_claim_notify, NULL,
 385             ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB)));
 386 }
 387 
 388 static int
 389 zil_claim_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t first_txg)
 390 {
 391         lr_write_t *lr = (lr_write_t *)lrc;
 392         int error;
 393 
 394         if (lrc->lrc_txtype != TX_WRITE)
 395                 return (0);
 396 
 397         /*
 398          * If the block is not readable, don't claim it.  This can happen
 399          * in normal operation when a log block is written to disk before
 400          * some of the dmu_sync() blocks it points to.  In this case, the
 401          * transaction cannot have been committed to anyone (we would have
 402          * waited for all writes to be stable first), so it is semantically
 403          * correct to declare this the end of the log.
 404          */
 405         if (lr->lr_blkptr.blk_birth >= first_txg &&
 406             (error = zil_read_log_data(zilog, lr, NULL)) != 0)
 407                 return (error);
 408         return (zil_claim_log_block(zilog, &lr->lr_blkptr, tx, first_txg));
 409 }
 410 
 411 /* ARGSUSED */
 412 static int
 413 zil_free_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t claim_txg)
 414 {
 415         zio_free_zil(zilog->zl_spa, dmu_tx_get_txg(tx), bp);
 416 
 417         return (0);
 418 }
 419 
 420 static int
 421 zil_free_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t claim_txg)
 422 {
 423         lr_write_t *lr = (lr_write_t *)lrc;
 424         blkptr_t *bp = &lr->lr_blkptr;
 425 
 426         /*
 427          * If we previously claimed it, we need to free it.
 428          */
 429         if (claim_txg != 0 && lrc->lrc_txtype == TX_WRITE &&
 430             bp->blk_birth >= claim_txg && zil_bp_tree_add(zilog, bp) == 0 &&
 431             !BP_IS_HOLE(bp))
 432                 zio_free(zilog->zl_spa, dmu_tx_get_txg(tx), bp);
 433 
 434         return (0);
 435 }
 436 
 437 static lwb_t *
 438 zil_alloc_lwb(zilog_t *zilog, blkptr_t *bp, uint64_t txg)
 439 {
 440         lwb_t *lwb;
 441 
 442         lwb = kmem_cache_alloc(zil_lwb_cache, KM_SLEEP);
 443         lwb->lwb_zilog = zilog;
 444         lwb->lwb_blk = *bp;
 445         lwb->lwb_buf = zio_buf_alloc(BP_GET_LSIZE(bp));
 446         lwb->lwb_max_txg = txg;
 447         lwb->lwb_zio = NULL;
 448         lwb->lwb_tx = NULL;
 449         if (BP_GET_CHECKSUM(bp) == ZIO_CHECKSUM_ZILOG2) {
 450                 lwb->lwb_nused = sizeof (zil_chain_t);
 451                 lwb->lwb_sz = BP_GET_LSIZE(bp);
 452         } else {
 453                 lwb->lwb_nused = 0;
 454                 lwb->lwb_sz = BP_GET_LSIZE(bp) - sizeof (zil_chain_t);
 455         }
 456 
 457         mutex_enter(&zilog->zl_lock);
 458         list_insert_tail(&zilog->zl_lwb_list, lwb);
 459         mutex_exit(&zilog->zl_lock);
 460 
 461         return (lwb);
 462 }
 463 
 464 /*
 465  * Called when we create in-memory log transactions so that we know
 466  * to cleanup the itxs at the end of spa_sync().
 467  */
 468 void
 469 zilog_dirty(zilog_t *zilog, uint64_t txg)
 470 {
 471         dsl_pool_t *dp = zilog->zl_dmu_pool;
 472         dsl_dataset_t *ds = dmu_objset_ds(zilog->zl_os);
 473 
 474         if (dsl_dataset_is_snapshot(ds))
 475                 panic("dirtying snapshot!");
 476 
 477         if (txg_list_add(&dp->dp_dirty_zilogs, zilog, txg)) {
 478                 /* up the hold count until we can be written out */
 479                 dmu_buf_add_ref(ds->ds_dbuf, zilog);
 480         }
 481 }
 482 
 483 boolean_t
 484 zilog_is_dirty(zilog_t *zilog)
 485 {
 486         dsl_pool_t *dp = zilog->zl_dmu_pool;
 487 
 488         for (int t = 0; t < TXG_SIZE; t++) {
 489                 if (txg_list_member(&dp->dp_dirty_zilogs, zilog, t))
 490                         return (B_TRUE);
 491         }
 492         return (B_FALSE);
 493 }
 494 
 495 /*
 496  * Create an on-disk intent log.
 497  */
 498 static lwb_t *
 499 zil_create(zilog_t *zilog)
 500 {
 501         const zil_header_t *zh = zilog->zl_header;
 502         lwb_t *lwb = NULL;
 503         uint64_t txg = 0;
 504         dmu_tx_t *tx = NULL;
 505         blkptr_t blk;
 506         int error = 0;
 507 
 508         /*
 509          * Wait for any previous destroy to complete.
 510          */
 511         txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
 512 
 513         ASSERT(zh->zh_claim_txg == 0);
 514         ASSERT(zh->zh_replay_seq == 0);
 515 
 516         blk = zh->zh_log;
 517 
 518         /*
 519          * Allocate an initial log block if:
 520          *    - there isn't one already
 521          *    - the existing block is the wrong endianess
 522          */
 523         if (BP_IS_HOLE(&blk) || BP_SHOULD_BYTESWAP(&blk)) {
 524                 tx = dmu_tx_create(zilog->zl_os);
 525                 VERIFY(dmu_tx_assign(tx, TXG_WAIT) == 0);
 526                 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
 527                 txg = dmu_tx_get_txg(tx);
 528 
 529                 if (!BP_IS_HOLE(&blk)) {
 530                         zio_free_zil(zilog->zl_spa, txg, &blk);
 531                         BP_ZERO(&blk);
 532                 }
 533 
 534                 error = zio_alloc_zil(zilog->zl_spa, txg, &blk, NULL,
 535                     ZIL_MIN_BLKSZ, zilog->zl_logbias == ZFS_LOGBIAS_LATENCY);
 536 
 537                 if (error == 0)
 538                         zil_init_log_chain(zilog, &blk);
 539         }
 540 
 541         /*
 542          * Allocate a log write buffer (lwb) for the first log block.
 543          */
 544         if (error == 0)
 545                 lwb = zil_alloc_lwb(zilog, &blk, txg);
 546 
 547         /*
 548          * If we just allocated the first log block, commit our transaction
 549          * and wait for zil_sync() to stuff the block poiner into zh_log.
 550          * (zh is part of the MOS, so we cannot modify it in open context.)
 551          */
 552         if (tx != NULL) {
 553                 dmu_tx_commit(tx);
 554                 txg_wait_synced(zilog->zl_dmu_pool, txg);
 555         }
 556 
 557         ASSERT(bcmp(&blk, &zh->zh_log, sizeof (blk)) == 0);
 558 
 559         return (lwb);
 560 }
 561 
 562 /*
 563  * In one tx, free all log blocks and clear the log header.
 564  * If keep_first is set, then we're replaying a log with no content.
 565  * We want to keep the first block, however, so that the first
 566  * synchronous transaction doesn't require a txg_wait_synced()
 567  * in zil_create().  We don't need to txg_wait_synced() here either
 568  * when keep_first is set, because both zil_create() and zil_destroy()
 569  * will wait for any in-progress destroys to complete.
 570  */
 571 void
 572 zil_destroy(zilog_t *zilog, boolean_t keep_first)
 573 {
 574         const zil_header_t *zh = zilog->zl_header;
 575         lwb_t *lwb;
 576         dmu_tx_t *tx;
 577         uint64_t txg;
 578 
 579         /*
 580          * Wait for any previous destroy to complete.
 581          */
 582         txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
 583 
 584         zilog->zl_old_header = *zh;          /* debugging aid */
 585 
 586         if (BP_IS_HOLE(&zh->zh_log))
 587                 return;
 588 
 589         tx = dmu_tx_create(zilog->zl_os);
 590         VERIFY(dmu_tx_assign(tx, TXG_WAIT) == 0);
 591         dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
 592         txg = dmu_tx_get_txg(tx);
 593 
 594         mutex_enter(&zilog->zl_lock);
 595 
 596         ASSERT3U(zilog->zl_destroy_txg, <, txg);
 597         zilog->zl_destroy_txg = txg;
 598         zilog->zl_keep_first = keep_first;
 599 
 600         if (!list_is_empty(&zilog->zl_lwb_list)) {
 601                 ASSERT(zh->zh_claim_txg == 0);
 602                 VERIFY(!keep_first);
 603                 while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
 604                         list_remove(&zilog->zl_lwb_list, lwb);
 605                         if (lwb->lwb_buf != NULL)
 606                                 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
 607                         zio_free_zil(zilog->zl_spa, txg, &lwb->lwb_blk);
 608                         kmem_cache_free(zil_lwb_cache, lwb);
 609                 }
 610         } else if (!keep_first) {
 611                 zil_destroy_sync(zilog, tx);
 612         }
 613         mutex_exit(&zilog->zl_lock);
 614 
 615         dmu_tx_commit(tx);
 616 }
 617 
 618 void
 619 zil_destroy_sync(zilog_t *zilog, dmu_tx_t *tx)
 620 {
 621         ASSERT(list_is_empty(&zilog->zl_lwb_list));
 622         (void) zil_parse(zilog, zil_free_log_block,
 623             zil_free_log_record, tx, zilog->zl_header->zh_claim_txg);
 624 }
 625 
 626 int
 627 zil_claim(const char *osname, void *txarg)
 628 {
 629         dmu_tx_t *tx = txarg;
 630         uint64_t first_txg = dmu_tx_get_txg(tx);
 631         zilog_t *zilog;
 632         zil_header_t *zh;
 633         objset_t *os;
 634         int error;
 635 
 636         error = dmu_objset_own(osname, DMU_OST_ANY, B_FALSE, FTAG, &os);
 637         if (error != 0) {
 638                 cmn_err(CE_WARN, "can't open objset for %s", osname);
 639                 return (0);
 640         }
 641 
 642         zilog = dmu_objset_zil(os);
 643         zh = zil_header_in_syncing_context(zilog);
 644 
 645         if (spa_get_log_state(zilog->zl_spa) == SPA_LOG_CLEAR) {
 646                 if (!BP_IS_HOLE(&zh->zh_log))
 647                         zio_free_zil(zilog->zl_spa, first_txg, &zh->zh_log);
 648                 BP_ZERO(&zh->zh_log);
 649                 dsl_dataset_dirty(dmu_objset_ds(os), tx);
 650                 dmu_objset_disown(os, FTAG);
 651                 return (0);
 652         }
 653 
 654         /*
 655          * Claim all log blocks if we haven't already done so, and remember
 656          * the highest claimed sequence number.  This ensures that if we can
 657          * read only part of the log now (e.g. due to a missing device),
 658          * but we can read the entire log later, we will not try to replay
 659          * or destroy beyond the last block we successfully claimed.
 660          */
 661         ASSERT3U(zh->zh_claim_txg, <=, first_txg);
 662         if (zh->zh_claim_txg == 0 && !BP_IS_HOLE(&zh->zh_log)) {
 663                 (void) zil_parse(zilog, zil_claim_log_block,
 664                     zil_claim_log_record, tx, first_txg);
 665                 zh->zh_claim_txg = first_txg;
 666                 zh->zh_claim_blk_seq = zilog->zl_parse_blk_seq;
 667                 zh->zh_claim_lr_seq = zilog->zl_parse_lr_seq;
 668                 if (zilog->zl_parse_lr_count || zilog->zl_parse_blk_count > 1)
 669                         zh->zh_flags |= ZIL_REPLAY_NEEDED;
 670                 zh->zh_flags |= ZIL_CLAIM_LR_SEQ_VALID;
 671                 dsl_dataset_dirty(dmu_objset_ds(os), tx);
 672         }
 673 
 674         ASSERT3U(first_txg, ==, (spa_last_synced_txg(zilog->zl_spa) + 1));
 675         dmu_objset_disown(os, FTAG);
 676         return (0);
 677 }
 678 
 679 /*
 680  * Check the log by walking the log chain.
 681  * Checksum errors are ok as they indicate the end of the chain.
 682  * Any other error (no device or read failure) returns an error.
 683  */
 684 int
 685 zil_check_log_chain(const char *osname, void *tx)
 686 {
 687         zilog_t *zilog;
 688         objset_t *os;
 689         blkptr_t *bp;
 690         int error;
 691 
 692         ASSERT(tx == NULL);
 693 
 694         error = dmu_objset_hold(osname, FTAG, &os);
 695         if (error != 0) {
 696                 cmn_err(CE_WARN, "can't open objset for %s", osname);
 697                 return (0);
 698         }
 699 
 700         zilog = dmu_objset_zil(os);
 701         bp = (blkptr_t *)&zilog->zl_header->zh_log;
 702 
 703         /*
 704          * Check the first block and determine if it's on a log device
 705          * which may have been removed or faulted prior to loading this
 706          * pool.  If so, there's no point in checking the rest of the log
 707          * as its content should have already been synced to the pool.
 708          */
 709         if (!BP_IS_HOLE(bp)) {
 710                 vdev_t *vd;
 711                 boolean_t valid = B_TRUE;
 712 
 713                 spa_config_enter(os->os_spa, SCL_STATE, FTAG, RW_READER);
 714                 vd = vdev_lookup_top(os->os_spa, DVA_GET_VDEV(&bp->blk_dva[0]));
 715                 if (vd->vdev_islog && vdev_is_dead(vd))
 716                         valid = vdev_log_state_valid(vd);
 717                 spa_config_exit(os->os_spa, SCL_STATE, FTAG);
 718 
 719                 if (!valid) {
 720                         dmu_objset_rele(os, FTAG);
 721                         return (0);
 722                 }
 723         }
 724 
 725         /*
 726          * Because tx == NULL, zil_claim_log_block() will not actually claim
 727          * any blocks, but just determine whether it is possible to do so.
 728          * In addition to checking the log chain, zil_claim_log_block()
 729          * will invoke zio_claim() with a done func of spa_claim_notify(),
 730          * which will update spa_max_claim_txg.  See spa_load() for details.
 731          */
 732         error = zil_parse(zilog, zil_claim_log_block, zil_claim_log_record, tx,
 733             zilog->zl_header->zh_claim_txg ? -1ULL : spa_first_txg(os->os_spa));
 734 
 735         dmu_objset_rele(os, FTAG);
 736 
 737         return ((error == ECKSUM || error == ENOENT) ? 0 : error);
 738 }
 739 
 740 static int
 741 zil_vdev_compare(const void *x1, const void *x2)
 742 {
 743         const uint64_t v1 = ((zil_vdev_node_t *)x1)->zv_vdev;
 744         const uint64_t v2 = ((zil_vdev_node_t *)x2)->zv_vdev;
 745 
 746         if (v1 < v2)
 747                 return (-1);
 748         if (v1 > v2)
 749                 return (1);
 750 
 751         return (0);
 752 }
 753 
 754 void
 755 zil_add_block(zilog_t *zilog, const blkptr_t *bp)
 756 {
 757         avl_tree_t *t = &zilog->zl_vdev_tree;
 758         avl_index_t where;
 759         zil_vdev_node_t *zv, zvsearch;
 760         int ndvas = BP_GET_NDVAS(bp);
 761         int i;
 762 
 763         if (zfs_nocacheflush)
 764                 return;
 765 
 766         ASSERT(zilog->zl_writer);
 767 
 768         /*
 769          * Even though we're zl_writer, we still need a lock because the
 770          * zl_get_data() callbacks may have dmu_sync() done callbacks
 771          * that will run concurrently.
 772          */
 773         mutex_enter(&zilog->zl_vdev_lock);
 774         for (i = 0; i < ndvas; i++) {
 775                 zvsearch.zv_vdev = DVA_GET_VDEV(&bp->blk_dva[i]);
 776                 if (avl_find(t, &zvsearch, &where) == NULL) {
 777                         zv = kmem_alloc(sizeof (*zv), KM_SLEEP);
 778                         zv->zv_vdev = zvsearch.zv_vdev;
 779                         avl_insert(t, zv, where);
 780                 }
 781         }
 782         mutex_exit(&zilog->zl_vdev_lock);
 783 }
 784 
 785 static void
 786 zil_flush_vdevs(zilog_t *zilog)
 787 {
 788         spa_t *spa = zilog->zl_spa;
 789         avl_tree_t *t = &zilog->zl_vdev_tree;
 790         void *cookie = NULL;
 791         zil_vdev_node_t *zv;
 792         zio_t *zio;
 793 
 794         ASSERT(zilog->zl_writer);
 795 
 796         /*
 797          * We don't need zl_vdev_lock here because we're the zl_writer,
 798          * and all zl_get_data() callbacks are done.
 799          */
 800         if (avl_numnodes(t) == 0)
 801                 return;
 802 
 803         spa_config_enter(spa, SCL_STATE, FTAG, RW_READER);
 804 
 805         zio = zio_root(spa, NULL, NULL, ZIO_FLAG_CANFAIL);
 806 
 807         while ((zv = avl_destroy_nodes(t, &cookie)) != NULL) {
 808                 vdev_t *vd = vdev_lookup_top(spa, zv->zv_vdev);
 809                 if (vd != NULL)
 810                         zio_flush(zio, vd);
 811                 kmem_free(zv, sizeof (*zv));
 812         }
 813 
 814         /*
 815          * Wait for all the flushes to complete.  Not all devices actually
 816          * support the DKIOCFLUSHWRITECACHE ioctl, so it's OK if it fails.
 817          */
 818         (void) zio_wait(zio);
 819 
 820         spa_config_exit(spa, SCL_STATE, FTAG);
 821 }
 822 
 823 /*
 824  * Function called when a log block write completes
 825  */
 826 static void
 827 zil_lwb_write_done(zio_t *zio)
 828 {
 829         lwb_t *lwb = zio->io_private;
 830         zilog_t *zilog = lwb->lwb_zilog;
 831         dmu_tx_t *tx = lwb->lwb_tx;
 832 
 833         ASSERT(BP_GET_COMPRESS(zio->io_bp) == ZIO_COMPRESS_OFF);
 834         ASSERT(BP_GET_TYPE(zio->io_bp) == DMU_OT_INTENT_LOG);
 835         ASSERT(BP_GET_LEVEL(zio->io_bp) == 0);
 836         ASSERT(BP_GET_BYTEORDER(zio->io_bp) == ZFS_HOST_BYTEORDER);
 837         ASSERT(!BP_IS_GANG(zio->io_bp));
 838         ASSERT(!BP_IS_HOLE(zio->io_bp));
 839         ASSERT(BP_GET_FILL(zio->io_bp) == 0);
 840 
 841         /*
 842          * Ensure the lwb buffer pointer is cleared before releasing
 843          * the txg. If we have had an allocation failure and
 844          * the txg is waiting to sync then we want want zil_sync()
 845          * to remove the lwb so that it's not picked up as the next new
 846          * one in zil_commit_writer(). zil_sync() will only remove
 847          * the lwb if lwb_buf is null.
 848          */
 849         zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
 850         mutex_enter(&zilog->zl_lock);
 851         lwb->lwb_buf = NULL;
 852         lwb->lwb_tx = NULL;
 853         mutex_exit(&zilog->zl_lock);
 854 
 855         /*
 856          * Now that we've written this log block, we have a stable pointer
 857          * to the next block in the chain, so it's OK to let the txg in
 858          * which we allocated the next block sync.
 859          */
 860         dmu_tx_commit(tx);
 861 }
 862 
 863 /*
 864  * Initialize the io for a log block.
 865  */
 866 static void
 867 zil_lwb_write_init(zilog_t *zilog, lwb_t *lwb)
 868 {
 869         zbookmark_t zb;
 870 
 871         SET_BOOKMARK(&zb, lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_OBJSET],
 872             ZB_ZIL_OBJECT, ZB_ZIL_LEVEL,
 873             lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_SEQ]);
 874 
 875         if (zilog->zl_root_zio == NULL) {
 876                 zilog->zl_root_zio = zio_root(zilog->zl_spa, NULL, NULL,
 877                     ZIO_FLAG_CANFAIL);
 878         }
 879         if (lwb->lwb_zio == NULL) {
 880                 lwb->lwb_zio = zio_rewrite(zilog->zl_root_zio, zilog->zl_spa,
 881                     0, &lwb->lwb_blk, lwb->lwb_buf, BP_GET_LSIZE(&lwb->lwb_blk),
 882                     zil_lwb_write_done, lwb, ZIO_PRIORITY_SYNC_WRITE,
 883                     ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_PROPAGATE, &zb);
 884         }
 885 }
 886 
 887 /*
 888  * Define a limited set of intent log block sizes.
 889  *
 890  * These must be a multiple of 4KB. Note only the amount used (again
 891  * aligned to 4KB) actually gets written. However, we can't always just
 892  * allocate SPA_MAXBLOCKSIZE as the slog space could be exhausted.
 893  */
 894 uint64_t zil_block_buckets[] = {
 895     4096,               /* non TX_WRITE */
 896     8192+4096,          /* data base */
 897     32*1024 + 4096,     /* NFS writes */
 898     UINT64_MAX
 899 };
 900 
 901 /*
 902  * Use the slog as long as the logbias is 'latency' and the current commit size
 903  * is less than the limit or the total list size is less than 2X the limit.
 904  * Limit checking is disabled by setting zil_slog_limit to UINT64_MAX.
 905  */
 906 uint64_t zil_slog_limit = 1024 * 1024;
 907 #define USE_SLOG(zilog) (((zilog)->zl_logbias == ZFS_LOGBIAS_LATENCY) && \
 908         (((zilog)->zl_cur_used < zil_slog_limit) || \
 909         ((zilog)->zl_itx_list_sz < (zil_slog_limit << 1))))
 910 
 911 /*
 912  * Start a log block write and advance to the next log block.
 913  * Calls are serialized.
 914  */
 915 static lwb_t *
 916 zil_lwb_write_start(zilog_t *zilog, lwb_t *lwb)
 917 {
 918         lwb_t *nlwb = NULL;
 919         zil_chain_t *zilc;
 920         spa_t *spa = zilog->zl_spa;
 921         blkptr_t *bp;
 922         dmu_tx_t *tx;
 923         uint64_t txg;
 924         uint64_t zil_blksz, wsz;
 925         int i, error;
 926 
 927         if (BP_GET_CHECKSUM(&lwb->lwb_blk) == ZIO_CHECKSUM_ZILOG2) {
 928                 zilc = (zil_chain_t *)lwb->lwb_buf;
 929                 bp = &zilc->zc_next_blk;
 930         } else {
 931                 zilc = (zil_chain_t *)(lwb->lwb_buf + lwb->lwb_sz);
 932                 bp = &zilc->zc_next_blk;
 933         }
 934 
 935         ASSERT(lwb->lwb_nused <= lwb->lwb_sz);
 936 
 937         /*
 938          * Allocate the next block and save its address in this block
 939          * before writing it in order to establish the log chain.
 940          * Note that if the allocation of nlwb synced before we wrote
 941          * the block that points at it (lwb), we'd leak it if we crashed.
 942          * Therefore, we don't do dmu_tx_commit() until zil_lwb_write_done().
 943          * We dirty the dataset to ensure that zil_sync() will be called
 944          * to clean up in the event of allocation failure or I/O failure.
 945          */
 946         tx = dmu_tx_create(zilog->zl_os);
 947         VERIFY(dmu_tx_assign(tx, TXG_WAIT) == 0);
 948         dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
 949         txg = dmu_tx_get_txg(tx);
 950 
 951         lwb->lwb_tx = tx;
 952 
 953         /*
 954          * Log blocks are pre-allocated. Here we select the size of the next
 955          * block, based on size used in the last block.
 956          * - first find the smallest bucket that will fit the block from a
 957          *   limited set of block sizes. This is because it's faster to write
 958          *   blocks allocated from the same metaslab as they are adjacent or
 959          *   close.
 960          * - next find the maximum from the new suggested size and an array of
 961          *   previous sizes. This lessens a picket fence effect of wrongly
 962          *   guesssing the size if we have a stream of say 2k, 64k, 2k, 64k
 963          *   requests.
 964          *
 965          * Note we only write what is used, but we can't just allocate
 966          * the maximum block size because we can exhaust the available
 967          * pool log space.
 968          */
 969         zil_blksz = zilog->zl_cur_used + sizeof (zil_chain_t);
 970         for (i = 0; zil_blksz > zil_block_buckets[i]; i++)
 971                 continue;
 972         zil_blksz = zil_block_buckets[i];
 973         if (zil_blksz == UINT64_MAX)
 974                 zil_blksz = SPA_MAXBLOCKSIZE;
 975         zilog->zl_prev_blks[zilog->zl_prev_rotor] = zil_blksz;
 976         for (i = 0; i < ZIL_PREV_BLKS; i++)
 977                 zil_blksz = MAX(zil_blksz, zilog->zl_prev_blks[i]);
 978         zilog->zl_prev_rotor = (zilog->zl_prev_rotor + 1) & (ZIL_PREV_BLKS - 1);
 979 
 980         BP_ZERO(bp);
 981         /* pass the old blkptr in order to spread log blocks across devs */
 982         error = zio_alloc_zil(spa, txg, bp, &lwb->lwb_blk, zil_blksz,
 983             USE_SLOG(zilog));
 984         if (error == 0) {
 985                 ASSERT3U(bp->blk_birth, ==, txg);
 986                 bp->blk_cksum = lwb->lwb_blk.blk_cksum;
 987                 bp->blk_cksum.zc_word[ZIL_ZC_SEQ]++;
 988 
 989                 /*
 990                  * Allocate a new log write buffer (lwb).
 991                  */
 992                 nlwb = zil_alloc_lwb(zilog, bp, txg);
 993 
 994                 /* Record the block for later vdev flushing */
 995                 zil_add_block(zilog, &lwb->lwb_blk);
 996         }
 997 
 998         if (BP_GET_CHECKSUM(&lwb->lwb_blk) == ZIO_CHECKSUM_ZILOG2) {
 999                 /* For Slim ZIL only write what is used. */
1000                 wsz = P2ROUNDUP_TYPED(lwb->lwb_nused, ZIL_MIN_BLKSZ, uint64_t);
1001                 ASSERT3U(wsz, <=, lwb->lwb_sz);
1002                 zio_shrink(lwb->lwb_zio, wsz);
1003 
1004         } else {
1005                 wsz = lwb->lwb_sz;
1006         }
1007 
1008         zilc->zc_pad = 0;
1009         zilc->zc_nused = lwb->lwb_nused;
1010         zilc->zc_eck.zec_cksum = lwb->lwb_blk.blk_cksum;
1011 
1012         /*
1013          * clear unused data for security
1014          */
1015         bzero(lwb->lwb_buf + lwb->lwb_nused, wsz - lwb->lwb_nused);
1016 
1017         zio_nowait(lwb->lwb_zio); /* Kick off the write for the old log block */
1018 
1019         /*
1020          * If there was an allocation failure then nlwb will be null which
1021          * forces a txg_wait_synced().
1022          */
1023         return (nlwb);
1024 }
1025 
1026 static lwb_t *
1027 zil_lwb_commit(zilog_t *zilog, itx_t *itx, lwb_t *lwb)
1028 {
1029         lr_t *lrc = &itx->itx_lr; /* common log record */
1030         lr_write_t *lrw = (lr_write_t *)lrc;
1031         char *lr_buf;
1032         uint64_t txg = lrc->lrc_txg;
1033         uint64_t reclen = lrc->lrc_reclen;
1034         uint64_t dlen = 0;
1035 
1036         if (lwb == NULL)
1037                 return (NULL);
1038 
1039         ASSERT(lwb->lwb_buf != NULL);
1040         ASSERT(zilog_is_dirty(zilog) ||
1041             spa_freeze_txg(zilog->zl_spa) != UINT64_MAX);
1042 
1043         if (lrc->lrc_txtype == TX_WRITE && itx->itx_wr_state == WR_NEED_COPY)
1044                 dlen = P2ROUNDUP_TYPED(
1045                     lrw->lr_length, sizeof (uint64_t), uint64_t);
1046 
1047         zilog->zl_cur_used += (reclen + dlen);
1048 
1049         zil_lwb_write_init(zilog, lwb);
1050 
1051         /*
1052          * If this record won't fit in the current log block, start a new one.
1053          */
1054         if (lwb->lwb_nused + reclen + dlen > lwb->lwb_sz) {
1055                 lwb = zil_lwb_write_start(zilog, lwb);
1056                 if (lwb == NULL)
1057                         return (NULL);
1058                 zil_lwb_write_init(zilog, lwb);
1059                 ASSERT(LWB_EMPTY(lwb));
1060                 if (lwb->lwb_nused + reclen + dlen > lwb->lwb_sz) {
1061                         txg_wait_synced(zilog->zl_dmu_pool, txg);
1062                         return (lwb);
1063                 }
1064         }
1065 
1066         lr_buf = lwb->lwb_buf + lwb->lwb_nused;
1067         bcopy(lrc, lr_buf, reclen);
1068         lrc = (lr_t *)lr_buf;
1069         lrw = (lr_write_t *)lrc;
1070 
1071         /*
1072          * If it's a write, fetch the data or get its blkptr as appropriate.
1073          */
1074         if (lrc->lrc_txtype == TX_WRITE) {
1075                 if (txg > spa_freeze_txg(zilog->zl_spa))
1076                         txg_wait_synced(zilog->zl_dmu_pool, txg);
1077                 if (itx->itx_wr_state != WR_COPIED) {
1078                         char *dbuf;
1079                         int error;
1080 
1081                         if (dlen) {
1082                                 ASSERT(itx->itx_wr_state == WR_NEED_COPY);
1083                                 dbuf = lr_buf + reclen;
1084                                 lrw->lr_common.lrc_reclen += dlen;
1085                         } else {
1086                                 ASSERT(itx->itx_wr_state == WR_INDIRECT);
1087                                 dbuf = NULL;
1088                         }
1089                         error = zilog->zl_get_data(
1090                             itx->itx_private, lrw, dbuf, lwb->lwb_zio);
1091                         if (error == EIO) {
1092                                 txg_wait_synced(zilog->zl_dmu_pool, txg);
1093                                 return (lwb);
1094                         }
1095                         if (error != 0) {
1096                                 ASSERT(error == ENOENT || error == EEXIST ||
1097                                     error == EALREADY);
1098                                 return (lwb);
1099                         }
1100                 }
1101         }
1102 
1103         /*
1104          * We're actually making an entry, so update lrc_seq to be the
1105          * log record sequence number.  Note that this is generally not
1106          * equal to the itx sequence number because not all transactions
1107          * are synchronous, and sometimes spa_sync() gets there first.
1108          */
1109         lrc->lrc_seq = ++zilog->zl_lr_seq; /* we are single threaded */
1110         lwb->lwb_nused += reclen + dlen;
1111         lwb->lwb_max_txg = MAX(lwb->lwb_max_txg, txg);
1112         ASSERT3U(lwb->lwb_nused, <=, lwb->lwb_sz);
1113         ASSERT0(P2PHASE(lwb->lwb_nused, sizeof (uint64_t)));
1114 
1115         return (lwb);
1116 }
1117 
1118 itx_t *
1119 zil_itx_create(uint64_t txtype, size_t lrsize)
1120 {
1121         itx_t *itx;
1122 
1123         lrsize = P2ROUNDUP_TYPED(lrsize, sizeof (uint64_t), size_t);
1124 
1125         itx = kmem_alloc(offsetof(itx_t, itx_lr) + lrsize, KM_SLEEP);
1126         itx->itx_lr.lrc_txtype = txtype;
1127         itx->itx_lr.lrc_reclen = lrsize;
1128         itx->itx_sod = lrsize; /* if write & WR_NEED_COPY will be increased */
1129         itx->itx_lr.lrc_seq = 0;     /* defensive */
1130         itx->itx_sync = B_TRUE;              /* default is synchronous */
1131 
1132         return (itx);
1133 }
1134 
1135 void
1136 zil_itx_destroy(itx_t *itx)
1137 {
1138         kmem_free(itx, offsetof(itx_t, itx_lr) + itx->itx_lr.lrc_reclen);
1139 }
1140 
1141 /*
1142  * Free up the sync and async itxs. The itxs_t has already been detached
1143  * so no locks are needed.
1144  */
1145 static void
1146 zil_itxg_clean(itxs_t *itxs)
1147 {
1148         itx_t *itx;
1149         list_t *list;
1150         avl_tree_t *t;
1151         void *cookie;
1152         itx_async_node_t *ian;
1153 
1154         list = &itxs->i_sync_list;
1155         while ((itx = list_head(list)) != NULL) {
1156                 list_remove(list, itx);
1157                 kmem_free(itx, offsetof(itx_t, itx_lr) +
1158                     itx->itx_lr.lrc_reclen);
1159         }
1160 
1161         cookie = NULL;
1162         t = &itxs->i_async_tree;
1163         while ((ian = avl_destroy_nodes(t, &cookie)) != NULL) {
1164                 list = &ian->ia_list;
1165                 while ((itx = list_head(list)) != NULL) {
1166                         list_remove(list, itx);
1167                         kmem_free(itx, offsetof(itx_t, itx_lr) +
1168                             itx->itx_lr.lrc_reclen);
1169                 }
1170                 list_destroy(list);
1171                 kmem_free(ian, sizeof (itx_async_node_t));
1172         }
1173         avl_destroy(t);
1174 
1175         kmem_free(itxs, sizeof (itxs_t));
1176 }
1177 
1178 static int
1179 zil_aitx_compare(const void *x1, const void *x2)
1180 {
1181         const uint64_t o1 = ((itx_async_node_t *)x1)->ia_foid;
1182         const uint64_t o2 = ((itx_async_node_t *)x2)->ia_foid;
1183 
1184         if (o1 < o2)
1185                 return (-1);
1186         if (o1 > o2)
1187                 return (1);
1188 
1189         return (0);
1190 }
1191 
1192 /*
1193  * Remove all async itx with the given oid.
1194  */
1195 static void
1196 zil_remove_async(zilog_t *zilog, uint64_t oid)
1197 {
1198         uint64_t otxg, txg;
1199         itx_async_node_t *ian;
1200         avl_tree_t *t;
1201         avl_index_t where;
1202         list_t clean_list;
1203         itx_t *itx;
1204 
1205         ASSERT(oid != 0);
1206         list_create(&clean_list, sizeof (itx_t), offsetof(itx_t, itx_node));
1207 
1208         if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1209                 otxg = ZILTEST_TXG;
1210         else
1211                 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1212 
1213         for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1214                 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1215 
1216                 mutex_enter(&itxg->itxg_lock);
1217                 if (itxg->itxg_txg != txg) {
1218                         mutex_exit(&itxg->itxg_lock);
1219                         continue;
1220                 }
1221 
1222                 /*
1223                  * Locate the object node and append its list.
1224                  */
1225                 t = &itxg->itxg_itxs->i_async_tree;
1226                 ian = avl_find(t, &oid, &where);
1227                 if (ian != NULL)
1228                         list_move_tail(&clean_list, &ian->ia_list);
1229                 mutex_exit(&itxg->itxg_lock);
1230         }
1231         while ((itx = list_head(&clean_list)) != NULL) {
1232                 list_remove(&clean_list, itx);
1233                 kmem_free(itx, offsetof(itx_t, itx_lr) +
1234                     itx->itx_lr.lrc_reclen);
1235         }
1236         list_destroy(&clean_list);
1237 }
1238 
1239 void
1240 zil_itx_assign(zilog_t *zilog, itx_t *itx, dmu_tx_t *tx)
1241 {
1242         uint64_t txg;
1243         itxg_t *itxg;
1244         itxs_t *itxs, *clean = NULL;
1245 
1246         /*
1247          * Object ids can be re-instantiated in the next txg so
1248          * remove any async transactions to avoid future leaks.
1249          * This can happen if a fsync occurs on the re-instantiated
1250          * object for a WR_INDIRECT or WR_NEED_COPY write, which gets
1251          * the new file data and flushes a write record for the old object.
1252          */
1253         if ((itx->itx_lr.lrc_txtype & ~TX_CI) == TX_REMOVE)
1254                 zil_remove_async(zilog, itx->itx_oid);
1255 
1256         /*
1257          * Ensure the data of a renamed file is committed before the rename.
1258          */
1259         if ((itx->itx_lr.lrc_txtype & ~TX_CI) == TX_RENAME)
1260                 zil_async_to_sync(zilog, itx->itx_oid);
1261 
1262         if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX)
1263                 txg = ZILTEST_TXG;
1264         else
1265                 txg = dmu_tx_get_txg(tx);
1266 
1267         itxg = &zilog->zl_itxg[txg & TXG_MASK];
1268         mutex_enter(&itxg->itxg_lock);
1269         itxs = itxg->itxg_itxs;
1270         if (itxg->itxg_txg != txg) {
1271                 if (itxs != NULL) {
1272                         /*
1273                          * The zil_clean callback hasn't got around to cleaning
1274                          * this itxg. Save the itxs for release below.
1275                          * This should be rare.
1276                          */
1277                         atomic_add_64(&zilog->zl_itx_list_sz, -itxg->itxg_sod);
1278                         itxg->itxg_sod = 0;
1279                         clean = itxg->itxg_itxs;
1280                 }
1281                 ASSERT(itxg->itxg_sod == 0);
1282                 itxg->itxg_txg = txg;
1283                 itxs = itxg->itxg_itxs = kmem_zalloc(sizeof (itxs_t), KM_SLEEP);
1284 
1285                 list_create(&itxs->i_sync_list, sizeof (itx_t),
1286                     offsetof(itx_t, itx_node));
1287                 avl_create(&itxs->i_async_tree, zil_aitx_compare,
1288                     sizeof (itx_async_node_t),
1289                     offsetof(itx_async_node_t, ia_node));
1290         }
1291         if (itx->itx_sync) {
1292                 list_insert_tail(&itxs->i_sync_list, itx);
1293                 atomic_add_64(&zilog->zl_itx_list_sz, itx->itx_sod);
1294                 itxg->itxg_sod += itx->itx_sod;
1295         } else {
1296                 avl_tree_t *t = &itxs->i_async_tree;
1297                 uint64_t foid = ((lr_ooo_t *)&itx->itx_lr)->lr_foid;
1298                 itx_async_node_t *ian;
1299                 avl_index_t where;
1300 
1301                 ian = avl_find(t, &foid, &where);
1302                 if (ian == NULL) {
1303                         ian = kmem_alloc(sizeof (itx_async_node_t), KM_SLEEP);
1304                         list_create(&ian->ia_list, sizeof (itx_t),
1305                             offsetof(itx_t, itx_node));
1306                         ian->ia_foid = foid;
1307                         avl_insert(t, ian, where);
1308                 }
1309                 list_insert_tail(&ian->ia_list, itx);
1310         }
1311 
1312         itx->itx_lr.lrc_txg = dmu_tx_get_txg(tx);
1313         zilog_dirty(zilog, txg);
1314         mutex_exit(&itxg->itxg_lock);
1315 
1316         /* Release the old itxs now we've dropped the lock */
1317         if (clean != NULL)
1318                 zil_itxg_clean(clean);
1319 }
1320 
1321 /*
1322  * If there are any in-memory intent log transactions which have now been
1323  * synced then start up a taskq to free them. We should only do this after we
1324  * have written out the uberblocks (i.e. txg has been comitted) so that
1325  * don't inadvertently clean out in-memory log records that would be required
1326  * by zil_commit().
1327  */
1328 void
1329 zil_clean(zilog_t *zilog, uint64_t synced_txg)
1330 {
1331         itxg_t *itxg = &zilog->zl_itxg[synced_txg & TXG_MASK];
1332         itxs_t *clean_me;
1333 
1334         mutex_enter(&itxg->itxg_lock);
1335         if (itxg->itxg_itxs == NULL || itxg->itxg_txg == ZILTEST_TXG) {
1336                 mutex_exit(&itxg->itxg_lock);
1337                 return;
1338         }
1339         ASSERT3U(itxg->itxg_txg, <=, synced_txg);
1340         ASSERT(itxg->itxg_txg != 0);
1341         ASSERT(zilog->zl_clean_taskq != NULL);
1342         atomic_add_64(&zilog->zl_itx_list_sz, -itxg->itxg_sod);
1343         itxg->itxg_sod = 0;
1344         clean_me = itxg->itxg_itxs;
1345         itxg->itxg_itxs = NULL;
1346         itxg->itxg_txg = 0;
1347         mutex_exit(&itxg->itxg_lock);
1348         /*
1349          * Preferably start a task queue to free up the old itxs but
1350          * if taskq_dispatch can't allocate resources to do that then
1351          * free it in-line. This should be rare. Note, using TQ_SLEEP
1352          * created a bad performance problem.
1353          */
1354         if (taskq_dispatch(zilog->zl_clean_taskq,
1355             (void (*)(void *))zil_itxg_clean, clean_me, TQ_NOSLEEP) == NULL)
1356                 zil_itxg_clean(clean_me);
1357 }
1358 
1359 /*
1360  * Get the list of itxs to commit into zl_itx_commit_list.
1361  */
1362 static void
1363 zil_get_commit_list(zilog_t *zilog)
1364 {
1365         uint64_t otxg, txg;
1366         list_t *commit_list = &zilog->zl_itx_commit_list;
1367         uint64_t push_sod = 0;
1368 
1369         if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1370                 otxg = ZILTEST_TXG;
1371         else
1372                 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1373 
1374         for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1375                 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1376 
1377                 mutex_enter(&itxg->itxg_lock);
1378                 if (itxg->itxg_txg != txg) {
1379                         mutex_exit(&itxg->itxg_lock);
1380                         continue;
1381                 }
1382 
1383                 list_move_tail(commit_list, &itxg->itxg_itxs->i_sync_list);
1384                 push_sod += itxg->itxg_sod;
1385                 itxg->itxg_sod = 0;
1386 
1387                 mutex_exit(&itxg->itxg_lock);
1388         }
1389         atomic_add_64(&zilog->zl_itx_list_sz, -push_sod);
1390 }
1391 
1392 /*
1393  * Move the async itxs for a specified object to commit into sync lists.
1394  */
1395 static void
1396 zil_async_to_sync(zilog_t *zilog, uint64_t foid)
1397 {
1398         uint64_t otxg, txg;
1399         itx_async_node_t *ian;
1400         avl_tree_t *t;
1401         avl_index_t where;
1402 
1403         if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1404                 otxg = ZILTEST_TXG;
1405         else
1406                 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1407 
1408         for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1409                 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1410 
1411                 mutex_enter(&itxg->itxg_lock);
1412                 if (itxg->itxg_txg != txg) {
1413                         mutex_exit(&itxg->itxg_lock);
1414                         continue;
1415                 }
1416 
1417                 /*
1418                  * If a foid is specified then find that node and append its
1419                  * list. Otherwise walk the tree appending all the lists
1420                  * to the sync list. We add to the end rather than the
1421                  * beginning to ensure the create has happened.
1422                  */
1423                 t = &itxg->itxg_itxs->i_async_tree;
1424                 if (foid != 0) {
1425                         ian = avl_find(t, &foid, &where);
1426                         if (ian != NULL) {
1427                                 list_move_tail(&itxg->itxg_itxs->i_sync_list,
1428                                     &ian->ia_list);
1429                         }
1430                 } else {
1431                         void *cookie = NULL;
1432 
1433                         while ((ian = avl_destroy_nodes(t, &cookie)) != NULL) {
1434                                 list_move_tail(&itxg->itxg_itxs->i_sync_list,
1435                                     &ian->ia_list);
1436                                 list_destroy(&ian->ia_list);
1437                                 kmem_free(ian, sizeof (itx_async_node_t));
1438                         }
1439                 }
1440                 mutex_exit(&itxg->itxg_lock);
1441         }
1442 }
1443 
1444 static void
1445 zil_commit_writer(zilog_t *zilog)
1446 {
1447         uint64_t txg;
1448         itx_t *itx;
1449         lwb_t *lwb;
1450         spa_t *spa = zilog->zl_spa;
1451         int error = 0;
1452 
1453         ASSERT(zilog->zl_root_zio == NULL);
1454 
1455         mutex_exit(&zilog->zl_lock);
1456 
1457         zil_get_commit_list(zilog);
1458 
1459         /*
1460          * Return if there's nothing to commit before we dirty the fs by
1461          * calling zil_create().
1462          */
1463         if (list_head(&zilog->zl_itx_commit_list) == NULL) {
1464                 mutex_enter(&zilog->zl_lock);
1465                 return;
1466         }
1467 
1468         if (zilog->zl_suspend) {
1469                 lwb = NULL;
1470         } else {
1471                 lwb = list_tail(&zilog->zl_lwb_list);
1472                 if (lwb == NULL)
1473                         lwb = zil_create(zilog);
1474         }
1475 
1476         DTRACE_PROBE1(zil__cw1, zilog_t *, zilog);
1477         while (itx = list_head(&zilog->zl_itx_commit_list)) {
1478                 txg = itx->itx_lr.lrc_txg;
1479                 ASSERT(txg);
1480 
1481                 if (txg > spa_last_synced_txg(spa) || txg > spa_freeze_txg(spa))
1482                         lwb = zil_lwb_commit(zilog, itx, lwb);
1483                 list_remove(&zilog->zl_itx_commit_list, itx);
1484                 kmem_free(itx, offsetof(itx_t, itx_lr)
1485                     + itx->itx_lr.lrc_reclen);
1486         }
1487         DTRACE_PROBE1(zil__cw2, zilog_t *, zilog);
1488 
1489         /* write the last block out */
1490         if (lwb != NULL && lwb->lwb_zio != NULL)
1491                 lwb = zil_lwb_write_start(zilog, lwb);
1492 
1493         zilog->zl_cur_used = 0;
1494 
1495         /*
1496          * Wait if necessary for the log blocks to be on stable storage.
1497          */
1498         if (zilog->zl_root_zio) {
1499                 error = zio_wait(zilog->zl_root_zio);
1500                 zilog->zl_root_zio = NULL;
1501                 zil_flush_vdevs(zilog);
1502         }
1503 
1504         if (error || lwb == NULL)
1505                 txg_wait_synced(zilog->zl_dmu_pool, 0);
1506 
1507         mutex_enter(&zilog->zl_lock);
1508 
1509         /*
1510          * Remember the highest committed log sequence number for ztest.
1511          * We only update this value when all the log writes succeeded,
1512          * because ztest wants to ASSERT that it got the whole log chain.
1513          */
1514         if (error == 0 && lwb != NULL)
1515                 zilog->zl_commit_lr_seq = zilog->zl_lr_seq;
1516 }
1517 
1518 /*
1519  * Commit zfs transactions to stable storage.
1520  * If foid is 0 push out all transactions, otherwise push only those
1521  * for that object or might reference that object.
1522  *
1523  * itxs are committed in batches. In a heavily stressed zil there will be
1524  * a commit writer thread who is writing out a bunch of itxs to the log
1525  * for a set of committing threads (cthreads) in the same batch as the writer.
1526  * Those cthreads are all waiting on the same cv for that batch.
1527  *
1528  * There will also be a different and growing batch of threads that are
1529  * waiting to commit (qthreads). When the committing batch completes
1530  * a transition occurs such that the cthreads exit and the qthreads become
1531  * cthreads. One of the new cthreads becomes the writer thread for the
1532  * batch. Any new threads arriving become new qthreads.
1533  *
1534  * Only 2 condition variables are needed and there's no transition
1535  * between the two cvs needed. They just flip-flop between qthreads
1536  * and cthreads.
1537  *
1538  * Using this scheme we can efficiently wakeup up only those threads
1539  * that have been committed.
1540  */
1541 void
1542 zil_commit(zilog_t *zilog, uint64_t foid)
1543 {
1544         uint64_t mybatch;
1545 
1546         if (zilog->zl_sync == ZFS_SYNC_DISABLED)
1547                 return;
1548 
1549         /* move the async itxs for the foid to the sync queues */
1550         zil_async_to_sync(zilog, foid);
1551 
1552         mutex_enter(&zilog->zl_lock);
1553         mybatch = zilog->zl_next_batch;
1554         while (zilog->zl_writer) {
1555                 cv_wait(&zilog->zl_cv_batch[mybatch & 1], &zilog->zl_lock);
1556                 if (mybatch <= zilog->zl_com_batch) {
1557                         mutex_exit(&zilog->zl_lock);
1558                         return;
1559                 }
1560         }
1561 
1562         zilog->zl_next_batch++;
1563         zilog->zl_writer = B_TRUE;
1564         zil_commit_writer(zilog);
1565         zilog->zl_com_batch = mybatch;
1566         zilog->zl_writer = B_FALSE;
1567         mutex_exit(&zilog->zl_lock);
1568 
1569         /* wake up one thread to become the next writer */
1570         cv_signal(&zilog->zl_cv_batch[(mybatch+1) & 1]);
1571 
1572         /* wake up all threads waiting for this batch to be committed */
1573         cv_broadcast(&zilog->zl_cv_batch[mybatch & 1]);
1574 }
1575 
1576 /*
1577  * Called in syncing context to free committed log blocks and update log header.
1578  */
1579 void
1580 zil_sync(zilog_t *zilog, dmu_tx_t *tx)
1581 {
1582         zil_header_t *zh = zil_header_in_syncing_context(zilog);
1583         uint64_t txg = dmu_tx_get_txg(tx);
1584         spa_t *spa = zilog->zl_spa;
1585         uint64_t *replayed_seq = &zilog->zl_replayed_seq[txg & TXG_MASK];
1586         lwb_t *lwb;
1587 
1588         /*
1589          * We don't zero out zl_destroy_txg, so make sure we don't try
1590          * to destroy it twice.
1591          */
1592         if (spa_sync_pass(spa) != 1)
1593                 return;
1594 
1595         mutex_enter(&zilog->zl_lock);
1596 
1597         ASSERT(zilog->zl_stop_sync == 0);
1598 
1599         if (*replayed_seq != 0) {
1600                 ASSERT(zh->zh_replay_seq < *replayed_seq);
1601                 zh->zh_replay_seq = *replayed_seq;
1602                 *replayed_seq = 0;
1603         }
1604 
1605         if (zilog->zl_destroy_txg == txg) {
1606                 blkptr_t blk = zh->zh_log;
1607 
1608                 ASSERT(list_head(&zilog->zl_lwb_list) == NULL);
1609 
1610                 bzero(zh, sizeof (zil_header_t));
1611                 bzero(zilog->zl_replayed_seq, sizeof (zilog->zl_replayed_seq));
1612 
1613                 if (zilog->zl_keep_first) {
1614                         /*
1615                          * If this block was part of log chain that couldn't
1616                          * be claimed because a device was missing during
1617                          * zil_claim(), but that device later returns,
1618                          * then this block could erroneously appear valid.
1619                          * To guard against this, assign a new GUID to the new
1620                          * log chain so it doesn't matter what blk points to.
1621                          */
1622                         zil_init_log_chain(zilog, &blk);
1623                         zh->zh_log = blk;
1624                 }
1625         }
1626 
1627         while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
1628                 zh->zh_log = lwb->lwb_blk;
1629                 if (lwb->lwb_buf != NULL || lwb->lwb_max_txg > txg)
1630                         break;
1631                 list_remove(&zilog->zl_lwb_list, lwb);
1632                 zio_free_zil(spa, txg, &lwb->lwb_blk);
1633                 kmem_cache_free(zil_lwb_cache, lwb);
1634 
1635                 /*
1636                  * If we don't have anything left in the lwb list then
1637                  * we've had an allocation failure and we need to zero
1638                  * out the zil_header blkptr so that we don't end
1639                  * up freeing the same block twice.
1640                  */
1641                 if (list_head(&zilog->zl_lwb_list) == NULL)
1642                         BP_ZERO(&zh->zh_log);
1643         }
1644         mutex_exit(&zilog->zl_lock);
1645 }
1646 
1647 void
1648 zil_init(void)
1649 {
1650         zil_lwb_cache = kmem_cache_create("zil_lwb_cache",
1651             sizeof (struct lwb), 0, NULL, NULL, NULL, NULL, NULL, 0);
1652 }
1653 
1654 void
1655 zil_fini(void)
1656 {
1657         kmem_cache_destroy(zil_lwb_cache);
1658 }
1659 
1660 void
1661 zil_set_sync(zilog_t *zilog, uint64_t sync)
1662 {
1663         zilog->zl_sync = sync;
1664 }
1665 
1666 void
1667 zil_set_logbias(zilog_t *zilog, uint64_t logbias)
1668 {
1669         zilog->zl_logbias = logbias;
1670 }
1671 
1672 zilog_t *
1673 zil_alloc(objset_t *os, zil_header_t *zh_phys)
1674 {
1675         zilog_t *zilog;
1676 
1677         zilog = kmem_zalloc(sizeof (zilog_t), KM_SLEEP);
1678 
1679         zilog->zl_header = zh_phys;
1680         zilog->zl_os = os;
1681         zilog->zl_spa = dmu_objset_spa(os);
1682         zilog->zl_dmu_pool = dmu_objset_pool(os);
1683         zilog->zl_destroy_txg = TXG_INITIAL - 1;
1684         zilog->zl_logbias = dmu_objset_logbias(os);
1685         zilog->zl_sync = dmu_objset_syncprop(os);
1686         zilog->zl_next_batch = 1;
1687 
1688         mutex_init(&zilog->zl_lock, NULL, MUTEX_DEFAULT, NULL);
1689 
1690         for (int i = 0; i < TXG_SIZE; i++) {
1691                 mutex_init(&zilog->zl_itxg[i].itxg_lock, NULL,
1692                     MUTEX_DEFAULT, NULL);
1693         }
1694 
1695         list_create(&zilog->zl_lwb_list, sizeof (lwb_t),
1696             offsetof(lwb_t, lwb_node));
1697 
1698         list_create(&zilog->zl_itx_commit_list, sizeof (itx_t),
1699             offsetof(itx_t, itx_node));
1700 
1701         mutex_init(&zilog->zl_vdev_lock, NULL, MUTEX_DEFAULT, NULL);
1702 
1703         avl_create(&zilog->zl_vdev_tree, zil_vdev_compare,
1704             sizeof (zil_vdev_node_t), offsetof(zil_vdev_node_t, zv_node));
1705 
1706         cv_init(&zilog->zl_cv_writer, NULL, CV_DEFAULT, NULL);
1707         cv_init(&zilog->zl_cv_suspend, NULL, CV_DEFAULT, NULL);
1708         cv_init(&zilog->zl_cv_batch[0], NULL, CV_DEFAULT, NULL);
1709         cv_init(&zilog->zl_cv_batch[1], NULL, CV_DEFAULT, NULL);
1710 
1711         return (zilog);
1712 }
1713 
1714 void
1715 zil_free(zilog_t *zilog)
1716 {
1717         zilog->zl_stop_sync = 1;
1718 
1719         ASSERT0(zilog->zl_suspend);
1720         ASSERT0(zilog->zl_suspending);
1721 
1722         ASSERT(list_is_empty(&zilog->zl_lwb_list));
1723         list_destroy(&zilog->zl_lwb_list);
1724 
1725         avl_destroy(&zilog->zl_vdev_tree);
1726         mutex_destroy(&zilog->zl_vdev_lock);
1727 
1728         ASSERT(list_is_empty(&zilog->zl_itx_commit_list));
1729         list_destroy(&zilog->zl_itx_commit_list);
1730 
1731         for (int i = 0; i < TXG_SIZE; i++) {
1732                 /*
1733                  * It's possible for an itx to be generated that doesn't dirty
1734                  * a txg (e.g. ztest TX_TRUNCATE). So there's no zil_clean()
1735                  * callback to remove the entry. We remove those here.
1736                  *
1737                  * Also free up the ziltest itxs.
1738                  */
1739                 if (zilog->zl_itxg[i].itxg_itxs)
1740                         zil_itxg_clean(zilog->zl_itxg[i].itxg_itxs);
1741                 mutex_destroy(&zilog->zl_itxg[i].itxg_lock);
1742         }
1743 
1744         mutex_destroy(&zilog->zl_lock);
1745 
1746         cv_destroy(&zilog->zl_cv_writer);
1747         cv_destroy(&zilog->zl_cv_suspend);
1748         cv_destroy(&zilog->zl_cv_batch[0]);
1749         cv_destroy(&zilog->zl_cv_batch[1]);
1750 
1751         kmem_free(zilog, sizeof (zilog_t));
1752 }
1753 
1754 /*
1755  * Open an intent log.
1756  */
1757 zilog_t *
1758 zil_open(objset_t *os, zil_get_data_t *get_data)
1759 {
1760         zilog_t *zilog = dmu_objset_zil(os);
1761 
1762         ASSERT(zilog->zl_clean_taskq == NULL);
1763         ASSERT(zilog->zl_get_data == NULL);
1764         ASSERT(list_is_empty(&zilog->zl_lwb_list));
1765 
1766         zilog->zl_get_data = get_data;
1767         zilog->zl_clean_taskq = taskq_create("zil_clean", 1, minclsyspri,
1768             2, 2, TASKQ_PREPOPULATE);
1769 
1770         return (zilog);
1771 }
1772 
1773 /*
1774  * Close an intent log.
1775  */
1776 void
1777 zil_close(zilog_t *zilog)
1778 {
1779         lwb_t *lwb;
1780         uint64_t txg = 0;
1781 
1782         zil_commit(zilog, 0); /* commit all itx */
1783 
1784         /*
1785          * The lwb_max_txg for the stubby lwb will reflect the last activity
1786          * for the zil.  After a txg_wait_synced() on the txg we know all the
1787          * callbacks have occurred that may clean the zil.  Only then can we
1788          * destroy the zl_clean_taskq.
1789          */
1790         mutex_enter(&zilog->zl_lock);
1791         lwb = list_tail(&zilog->zl_lwb_list);
1792         if (lwb != NULL)
1793                 txg = lwb->lwb_max_txg;
1794         mutex_exit(&zilog->zl_lock);
1795         if (txg)
1796                 txg_wait_synced(zilog->zl_dmu_pool, txg);
1797         ASSERT(!zilog_is_dirty(zilog));
1798 
1799         taskq_destroy(zilog->zl_clean_taskq);
1800         zilog->zl_clean_taskq = NULL;
1801         zilog->zl_get_data = NULL;
1802 
1803         /*
1804          * We should have only one LWB left on the list; remove it now.
1805          */
1806         mutex_enter(&zilog->zl_lock);
1807         lwb = list_head(&zilog->zl_lwb_list);
1808         if (lwb != NULL) {
1809                 ASSERT(lwb == list_tail(&zilog->zl_lwb_list));
1810                 list_remove(&zilog->zl_lwb_list, lwb);
1811                 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
1812                 kmem_cache_free(zil_lwb_cache, lwb);
1813         }
1814         mutex_exit(&zilog->zl_lock);
1815 }
1816 
1817 static char *suspend_tag = "zil suspending";
1818 
1819 /*
1820  * Suspend an intent log.  While in suspended mode, we still honor
1821  * synchronous semantics, but we rely on txg_wait_synced() to do it.
1822  * On old version pools, we suspend the log briefly when taking a
1823  * snapshot so that it will have an empty intent log.
1824  *
1825  * Long holds are not really intended to be used the way we do here --
1826  * held for such a short time.  A concurrent caller of dsl_dataset_long_held()
1827  * could fail.  Therefore we take pains to only put a long hold if it is
1828  * actually necessary.  Fortunately, it will only be necessary if the
1829  * objset is currently mounted (or the ZVOL equivalent).  In that case it
1830  * will already have a long hold, so we are not really making things any worse.
1831  *
1832  * Ideally, we would locate the existing long-holder (i.e. the zfsvfs_t or
1833  * zvol_state_t), and use their mechanism to prevent their hold from being
1834  * dropped (e.g. VFS_HOLD()).  However, that would be even more pain for
1835  * very little gain.
1836  *
1837  * if cookiep == NULL, this does both the suspend & resume.
1838  * Otherwise, it returns with the dataset "long held", and the cookie
1839  * should be passed into zil_resume().
1840  */
1841 int
1842 zil_suspend(const char *osname, void **cookiep)
1843 {
1844         objset_t *os;
1845         zilog_t *zilog;
1846         const zil_header_t *zh;
1847         int error;
1848 
1849         error = dmu_objset_hold(osname, suspend_tag, &os);
1850         if (error != 0)
1851                 return (error);
1852         zilog = dmu_objset_zil(os);
1853 
1854         mutex_enter(&zilog->zl_lock);
1855         zh = zilog->zl_header;
1856 
1857         if (zh->zh_flags & ZIL_REPLAY_NEEDED) {          /* unplayed log */
1858                 mutex_exit(&zilog->zl_lock);
1859                 dmu_objset_rele(os, suspend_tag);
1860                 return (SET_ERROR(EBUSY));
1861         }
1862 
1863         /*
1864          * Don't put a long hold in the cases where we can avoid it.  This
1865          * is when there is no cookie so we are doing a suspend & resume
1866          * (i.e. called from zil_vdev_offline()), and there's nothing to do
1867          * for the suspend because it's already suspended, or there's no ZIL.
1868          */
1869         if (cookiep == NULL && !zilog->zl_suspending &&
1870             (zilog->zl_suspend > 0 || BP_IS_HOLE(&zh->zh_log))) {
1871                 mutex_exit(&zilog->zl_lock);
1872                 dmu_objset_rele(os, suspend_tag);
1873                 return (0);
1874         }
1875 
1876         dsl_dataset_long_hold(dmu_objset_ds(os), suspend_tag);
1877         dsl_pool_rele(dmu_objset_pool(os), suspend_tag);
1878 
1879         zilog->zl_suspend++;
1880 
1881         if (zilog->zl_suspend > 1) {
1882                 /*
1883                  * Someone else is already suspending it.
1884                  * Just wait for them to finish.
1885                  */
1886 
1887                 while (zilog->zl_suspending)
1888                         cv_wait(&zilog->zl_cv_suspend, &zilog->zl_lock);
1889                 mutex_exit(&zilog->zl_lock);
1890 
1891                 if (cookiep == NULL)
1892                         zil_resume(os);
1893                 else
1894                         *cookiep = os;
1895                 return (0);
1896         }
1897 
1898         /*
1899          * If there is no pointer to an on-disk block, this ZIL must not
1900          * be active (e.g. filesystem not mounted), so there's nothing
1901          * to clean up.
1902          */
1903         if (BP_IS_HOLE(&zh->zh_log)) {
1904                 ASSERT(cookiep != NULL); /* fast path already handled */
1905 
1906                 *cookiep = os;
1907                 mutex_exit(&zilog->zl_lock);
1908                 return (0);
1909         }
1910 
1911         zilog->zl_suspending = B_TRUE;
1912         mutex_exit(&zilog->zl_lock);
1913 
1914         zil_commit(zilog, 0);
1915 
1916         zil_destroy(zilog, B_FALSE);
1917 
1918         mutex_enter(&zilog->zl_lock);
1919         zilog->zl_suspending = B_FALSE;
1920         cv_broadcast(&zilog->zl_cv_suspend);
1921         mutex_exit(&zilog->zl_lock);
1922 
1923         if (cookiep == NULL)
1924                 zil_resume(os);
1925         else
1926                 *cookiep = os;
1927         return (0);
1928 }
1929 
1930 void
1931 zil_resume(void *cookie)
1932 {
1933         objset_t *os = cookie;
1934         zilog_t *zilog = dmu_objset_zil(os);
1935 
1936         mutex_enter(&zilog->zl_lock);
1937         ASSERT(zilog->zl_suspend != 0);
1938         zilog->zl_suspend--;
1939         mutex_exit(&zilog->zl_lock);
1940         dsl_dataset_long_rele(dmu_objset_ds(os), suspend_tag);
1941         dsl_dataset_rele(dmu_objset_ds(os), suspend_tag);
1942 }
1943 
1944 typedef struct zil_replay_arg {
1945         zil_replay_func_t **zr_replay;
1946         void            *zr_arg;
1947         boolean_t       zr_byteswap;
1948         char            *zr_lr;
1949 } zil_replay_arg_t;
1950 
1951 static int
1952 zil_replay_error(zilog_t *zilog, lr_t *lr, int error)
1953 {
1954         char name[MAXNAMELEN];
1955 
1956         zilog->zl_replaying_seq--;   /* didn't actually replay this one */
1957 
1958         dmu_objset_name(zilog->zl_os, name);
1959 
1960         cmn_err(CE_WARN, "ZFS replay transaction error %d, "
1961             "dataset %s, seq 0x%llx, txtype %llu %s\n", error, name,
1962             (u_longlong_t)lr->lrc_seq,
1963             (u_longlong_t)(lr->lrc_txtype & ~TX_CI),
1964             (lr->lrc_txtype & TX_CI) ? "CI" : "");
1965 
1966         return (error);
1967 }
1968 
1969 static int
1970 zil_replay_log_record(zilog_t *zilog, lr_t *lr, void *zra, uint64_t claim_txg)
1971 {
1972         zil_replay_arg_t *zr = zra;
1973         const zil_header_t *zh = zilog->zl_header;
1974         uint64_t reclen = lr->lrc_reclen;
1975         uint64_t txtype = lr->lrc_txtype;
1976         int error = 0;
1977 
1978         zilog->zl_replaying_seq = lr->lrc_seq;
1979 
1980         if (lr->lrc_seq <= zh->zh_replay_seq)  /* already replayed */
1981                 return (0);
1982 
1983         if (lr->lrc_txg < claim_txg)              /* already committed */
1984                 return (0);
1985 
1986         /* Strip case-insensitive bit, still present in log record */
1987         txtype &= ~TX_CI;
1988 
1989         if (txtype == 0 || txtype >= TX_MAX_TYPE)
1990                 return (zil_replay_error(zilog, lr, EINVAL));
1991 
1992         /*
1993          * If this record type can be logged out of order, the object
1994          * (lr_foid) may no longer exist.  That's legitimate, not an error.
1995          */
1996         if (TX_OOO(txtype)) {
1997                 error = dmu_object_info(zilog->zl_os,
1998                     ((lr_ooo_t *)lr)->lr_foid, NULL);
1999                 if (error == ENOENT || error == EEXIST)
2000                         return (0);
2001         }
2002 
2003         /*
2004          * Make a copy of the data so we can revise and extend it.
2005          */
2006         bcopy(lr, zr->zr_lr, reclen);
2007 
2008         /*
2009          * If this is a TX_WRITE with a blkptr, suck in the data.
2010          */
2011         if (txtype == TX_WRITE && reclen == sizeof (lr_write_t)) {
2012                 error = zil_read_log_data(zilog, (lr_write_t *)lr,
2013                     zr->zr_lr + reclen);
2014                 if (error != 0)
2015                         return (zil_replay_error(zilog, lr, error));
2016         }
2017 
2018         /*
2019          * The log block containing this lr may have been byteswapped
2020          * so that we can easily examine common fields like lrc_txtype.
2021          * However, the log is a mix of different record types, and only the
2022          * replay vectors know how to byteswap their records.  Therefore, if
2023          * the lr was byteswapped, undo it before invoking the replay vector.
2024          */
2025         if (zr->zr_byteswap)
2026                 byteswap_uint64_array(zr->zr_lr, reclen);
2027 
2028         /*
2029          * We must now do two things atomically: replay this log record,
2030          * and update the log header sequence number to reflect the fact that
2031          * we did so. At the end of each replay function the sequence number
2032          * is updated if we are in replay mode.
2033          */
2034         error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lr, zr->zr_byteswap);
2035         if (error != 0) {
2036                 /*
2037                  * The DMU's dnode layer doesn't see removes until the txg
2038                  * commits, so a subsequent claim can spuriously fail with
2039                  * EEXIST. So if we receive any error we try syncing out
2040                  * any removes then retry the transaction.  Note that we
2041                  * specify B_FALSE for byteswap now, so we don't do it twice.
2042                  */
2043                 txg_wait_synced(spa_get_dsl(zilog->zl_spa), 0);
2044                 error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lr, B_FALSE);
2045                 if (error != 0)
2046                         return (zil_replay_error(zilog, lr, error));
2047         }
2048         return (0);
2049 }
2050 
2051 /* ARGSUSED */
2052 static int
2053 zil_incr_blks(zilog_t *zilog, blkptr_t *bp, void *arg, uint64_t claim_txg)
2054 {
2055         zilog->zl_replay_blks++;
2056 
2057         return (0);
2058 }
2059 
2060 /*
2061  * If this dataset has a non-empty intent log, replay it and destroy it.
2062  */
2063 void
2064 zil_replay(objset_t *os, void *arg, zil_replay_func_t *replay_func[TX_MAX_TYPE])
2065 {
2066         zilog_t *zilog = dmu_objset_zil(os);
2067         const zil_header_t *zh = zilog->zl_header;
2068         zil_replay_arg_t zr;
2069 
2070         if ((zh->zh_flags & ZIL_REPLAY_NEEDED) == 0) {
2071                 zil_destroy(zilog, B_TRUE);
2072                 return;
2073         }
2074 
2075         zr.zr_replay = replay_func;
2076         zr.zr_arg = arg;
2077         zr.zr_byteswap = BP_SHOULD_BYTESWAP(&zh->zh_log);
2078         zr.zr_lr = kmem_alloc(2 * SPA_MAXBLOCKSIZE, KM_SLEEP);
2079 
2080         /*
2081          * Wait for in-progress removes to sync before starting replay.
2082          */
2083         txg_wait_synced(zilog->zl_dmu_pool, 0);
2084 
2085         zilog->zl_replay = B_TRUE;
2086         zilog->zl_replay_time = ddi_get_lbolt();
2087         ASSERT(zilog->zl_replay_blks == 0);
2088         (void) zil_parse(zilog, zil_incr_blks, zil_replay_log_record, &zr,
2089             zh->zh_claim_txg);
2090         kmem_free(zr.zr_lr, 2 * SPA_MAXBLOCKSIZE);
2091 
2092         zil_destroy(zilog, B_FALSE);
2093         txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
2094         zilog->zl_replay = B_FALSE;
2095 }
2096 
2097 boolean_t
2098 zil_replaying(zilog_t *zilog, dmu_tx_t *tx)
2099 {
2100         if (zilog->zl_sync == ZFS_SYNC_DISABLED)
2101                 return (B_TRUE);
2102 
2103         if (zilog->zl_replay) {
2104                 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
2105                 zilog->zl_replayed_seq[dmu_tx_get_txg(tx) & TXG_MASK] =
2106                     zilog->zl_replaying_seq;
2107                 return (B_TRUE);
2108         }
2109 
2110         return (B_FALSE);
2111 }
2112 
2113 /* ARGSUSED */
2114 int
2115 zil_vdev_offline(const char *osname, void *arg)
2116 {
2117         int error;
2118 
2119         error = zil_suspend(osname, NULL);
2120         if (error != 0)
2121                 return (SET_ERROR(EEXIST));
2122         return (0);
2123 }