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