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