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 https://opensource.org/licenses/CDDL-1.0. 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, 2018 by Delphix. All rights reserved. 24 * Copyright (c) 2014 Integros [integros.com] 25 * Copyright (c) 2018 Datto Inc. 26 */ 27 28 /* Portions Copyright 2010 Robert Milkowski */ 29 30 #include <sys/zfs_context.h> 31 #include <sys/spa.h> 32 #include <sys/spa_impl.h> 33 #include <sys/dmu.h> 34 #include <sys/zap.h> 35 #include <sys/arc.h> 36 #include <sys/stat.h> 37 #include <sys/zil.h> 38 #include <sys/zil_impl.h> 39 #include <sys/dsl_dataset.h> 40 #include <sys/vdev_impl.h> 41 #include <sys/dmu_tx.h> 42 #include <sys/dsl_pool.h> 43 #include <sys/metaslab.h> 44 #include <sys/trace_zfs.h> 45 #include <sys/abd.h> 46 #include <sys/brt.h> 47 #include <sys/wmsum.h> 48 49 /* 50 * The ZFS Intent Log (ZIL) saves "transaction records" (itxs) of system 51 * calls that change the file system. Each itx has enough information to 52 * be able to replay them after a system crash, power loss, or 53 * equivalent failure mode. These are stored in memory until either: 54 * 55 * 1. they are committed to the pool by the DMU transaction group 56 * (txg), at which point they can be discarded; or 57 * 2. they are committed to the on-disk ZIL for the dataset being 58 * modified (e.g. due to an fsync, O_DSYNC, or other synchronous 59 * requirement). 60 * 61 * In the event of a crash or power loss, the itxs contained by each 62 * dataset's on-disk ZIL will be replayed when that dataset is first 63 * instantiated (e.g. if the dataset is a normal filesystem, when it is 64 * first mounted). 65 * 66 * As hinted at above, there is one ZIL per dataset (both the in-memory 67 * representation, and the on-disk representation). The on-disk format 68 * consists of 3 parts: 69 * 70 * - a single, per-dataset, ZIL header; which points to a chain of 71 * - zero or more ZIL blocks; each of which contains 72 * - zero or more ZIL records 73 * 74 * A ZIL record holds the information necessary to replay a single 75 * system call transaction. A ZIL block can hold many ZIL records, and 76 * the blocks are chained together, similarly to a singly linked list. 77 * 78 * Each ZIL block contains a block pointer (blkptr_t) to the next ZIL 79 * block in the chain, and the ZIL header points to the first block in 80 * the chain. 81 * 82 * Note, there is not a fixed place in the pool to hold these ZIL 83 * blocks; they are dynamically allocated and freed as needed from the 84 * blocks available on the pool, though they can be preferentially 85 * allocated from a dedicated "log" vdev. 86 */ 87 88 /* 89 * This controls the amount of time that a ZIL block (lwb) will remain 90 * "open" when it isn't "full", and it has a thread waiting for it to be 91 * committed to stable storage. Please refer to the zil_commit_waiter() 92 * function (and the comments within it) for more details. 93 */ 94 static uint_t zfs_commit_timeout_pct = 10; 95 96 /* 97 * See zil.h for more information about these fields. 98 */ 99 static zil_kstat_values_t zil_stats = { 100 { "zil_commit_count", KSTAT_DATA_UINT64 }, 101 { "zil_commit_writer_count", KSTAT_DATA_UINT64 }, 102 { "zil_itx_count", KSTAT_DATA_UINT64 }, 103 { "zil_itx_indirect_count", KSTAT_DATA_UINT64 }, 104 { "zil_itx_indirect_bytes", KSTAT_DATA_UINT64 }, 105 { "zil_itx_copied_count", KSTAT_DATA_UINT64 }, 106 { "zil_itx_copied_bytes", KSTAT_DATA_UINT64 }, 107 { "zil_itx_needcopy_count", KSTAT_DATA_UINT64 }, 108 { "zil_itx_needcopy_bytes", KSTAT_DATA_UINT64 }, 109 { "zil_itx_metaslab_normal_count", KSTAT_DATA_UINT64 }, 110 { "zil_itx_metaslab_normal_bytes", KSTAT_DATA_UINT64 }, 111 { "zil_itx_metaslab_normal_write", KSTAT_DATA_UINT64 }, 112 { "zil_itx_metaslab_normal_alloc", KSTAT_DATA_UINT64 }, 113 { "zil_itx_metaslab_slog_count", KSTAT_DATA_UINT64 }, 114 { "zil_itx_metaslab_slog_bytes", KSTAT_DATA_UINT64 }, 115 { "zil_itx_metaslab_slog_write", KSTAT_DATA_UINT64 }, 116 { "zil_itx_metaslab_slog_alloc", KSTAT_DATA_UINT64 }, 117 }; 118 119 static zil_sums_t zil_sums_global; 120 static kstat_t *zil_kstats_global; 121 122 /* 123 * Disable intent logging replay. This global ZIL switch affects all pools. 124 */ 125 int zil_replay_disable = 0; 126 127 /* 128 * Disable the flush commands that are normally sent to the disk(s) by the ZIL 129 * after an LWB write has completed. Setting this will cause ZIL corruption on 130 * power loss if a volatile out-of-order write cache is enabled. 131 */ 132 static int zil_nocacheflush = 0; 133 134 /* 135 * Limit SLOG write size per commit executed with synchronous priority. 136 * Any writes above that will be executed with lower (asynchronous) priority 137 * to limit potential SLOG device abuse by single active ZIL writer. 138 */ 139 static uint64_t zil_slog_bulk = 64 * 1024 * 1024; 140 141 static kmem_cache_t *zil_lwb_cache; 142 static kmem_cache_t *zil_zcw_cache; 143 144 static void zil_lwb_commit(zilog_t *zilog, lwb_t *lwb, itx_t *itx); 145 static itx_t *zil_itx_clone(itx_t *oitx); 146 static uint64_t zil_max_waste_space(zilog_t *zilog); 147 148 static int 149 zil_bp_compare(const void *x1, const void *x2) 150 { 151 const dva_t *dva1 = &((zil_bp_node_t *)x1)->zn_dva; 152 const dva_t *dva2 = &((zil_bp_node_t *)x2)->zn_dva; 153 154 int cmp = TREE_CMP(DVA_GET_VDEV(dva1), DVA_GET_VDEV(dva2)); 155 if (likely(cmp)) 156 return (cmp); 157 158 return (TREE_CMP(DVA_GET_OFFSET(dva1), DVA_GET_OFFSET(dva2))); 159 } 160 161 static void 162 zil_bp_tree_init(zilog_t *zilog) 163 { 164 avl_create(&zilog->zl_bp_tree, zil_bp_compare, 165 sizeof (zil_bp_node_t), offsetof(zil_bp_node_t, zn_node)); 166 } 167 168 static void 169 zil_bp_tree_fini(zilog_t *zilog) 170 { 171 avl_tree_t *t = &zilog->zl_bp_tree; 172 zil_bp_node_t *zn; 173 void *cookie = NULL; 174 175 while ((zn = avl_destroy_nodes(t, &cookie)) != NULL) 176 kmem_free(zn, sizeof (zil_bp_node_t)); 177 178 avl_destroy(t); 179 } 180 181 int 182 zil_bp_tree_add(zilog_t *zilog, const blkptr_t *bp) 183 { 184 avl_tree_t *t = &zilog->zl_bp_tree; 185 const dva_t *dva; 186 zil_bp_node_t *zn; 187 avl_index_t where; 188 189 if (BP_IS_EMBEDDED(bp)) 190 return (0); 191 192 dva = BP_IDENTITY(bp); 193 194 if (avl_find(t, dva, &where) != NULL) 195 return (SET_ERROR(EEXIST)); 196 197 zn = kmem_alloc(sizeof (zil_bp_node_t), KM_SLEEP); 198 zn->zn_dva = *dva; 199 avl_insert(t, zn, where); 200 201 return (0); 202 } 203 204 static zil_header_t * 205 zil_header_in_syncing_context(zilog_t *zilog) 206 { 207 return ((zil_header_t *)zilog->zl_header); 208 } 209 210 static void 211 zil_init_log_chain(zilog_t *zilog, blkptr_t *bp) 212 { 213 zio_cksum_t *zc = &bp->blk_cksum; 214 215 (void) random_get_pseudo_bytes((void *)&zc->zc_word[ZIL_ZC_GUID_0], 216 sizeof (zc->zc_word[ZIL_ZC_GUID_0])); 217 (void) random_get_pseudo_bytes((void *)&zc->zc_word[ZIL_ZC_GUID_1], 218 sizeof (zc->zc_word[ZIL_ZC_GUID_1])); 219 zc->zc_word[ZIL_ZC_OBJSET] = dmu_objset_id(zilog->zl_os); 220 zc->zc_word[ZIL_ZC_SEQ] = 1ULL; 221 } 222 223 static int 224 zil_kstats_global_update(kstat_t *ksp, int rw) 225 { 226 zil_kstat_values_t *zs = ksp->ks_data; 227 ASSERT3P(&zil_stats, ==, zs); 228 229 if (rw == KSTAT_WRITE) { 230 return (SET_ERROR(EACCES)); 231 } 232 233 zil_kstat_values_update(zs, &zil_sums_global); 234 235 return (0); 236 } 237 238 /* 239 * Read a log block and make sure it's valid. 240 */ 241 static int 242 zil_read_log_block(zilog_t *zilog, boolean_t decrypt, const blkptr_t *bp, 243 blkptr_t *nbp, char **begin, char **end, arc_buf_t **abuf) 244 { 245 zio_flag_t zio_flags = ZIO_FLAG_CANFAIL; 246 arc_flags_t aflags = ARC_FLAG_WAIT; 247 zbookmark_phys_t zb; 248 int error; 249 250 if (zilog->zl_header->zh_claim_txg == 0) 251 zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB; 252 253 if (!(zilog->zl_header->zh_flags & ZIL_CLAIM_LR_SEQ_VALID)) 254 zio_flags |= ZIO_FLAG_SPECULATIVE; 255 256 if (!decrypt) 257 zio_flags |= ZIO_FLAG_RAW; 258 259 SET_BOOKMARK(&zb, bp->blk_cksum.zc_word[ZIL_ZC_OBJSET], 260 ZB_ZIL_OBJECT, ZB_ZIL_LEVEL, bp->blk_cksum.zc_word[ZIL_ZC_SEQ]); 261 262 error = arc_read(NULL, zilog->zl_spa, bp, arc_getbuf_func, 263 abuf, ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb); 264 265 if (error == 0) { 266 zio_cksum_t cksum = bp->blk_cksum; 267 268 /* 269 * Validate the checksummed log block. 270 * 271 * Sequence numbers should be... sequential. The checksum 272 * verifier for the next block should be bp's checksum plus 1. 273 * 274 * Also check the log chain linkage and size used. 275 */ 276 cksum.zc_word[ZIL_ZC_SEQ]++; 277 278 uint64_t size = BP_GET_LSIZE(bp); 279 if (BP_GET_CHECKSUM(bp) == ZIO_CHECKSUM_ZILOG2) { 280 zil_chain_t *zilc = (*abuf)->b_data; 281 char *lr = (char *)(zilc + 1); 282 283 if (memcmp(&cksum, &zilc->zc_next_blk.blk_cksum, 284 sizeof (cksum)) || 285 zilc->zc_nused < sizeof (*zilc) || 286 zilc->zc_nused > size) { 287 error = SET_ERROR(ECKSUM); 288 } else { 289 *begin = lr; 290 *end = lr + zilc->zc_nused - sizeof (*zilc); 291 *nbp = zilc->zc_next_blk; 292 } 293 } else { 294 char *lr = (*abuf)->b_data; 295 zil_chain_t *zilc = (zil_chain_t *)(lr + size) - 1; 296 297 if (memcmp(&cksum, &zilc->zc_next_blk.blk_cksum, 298 sizeof (cksum)) || 299 (zilc->zc_nused > (size - sizeof (*zilc)))) { 300 error = SET_ERROR(ECKSUM); 301 } else { 302 *begin = lr; 303 *end = lr + zilc->zc_nused; 304 *nbp = zilc->zc_next_blk; 305 } 306 } 307 } 308 309 return (error); 310 } 311 312 /* 313 * Read a TX_WRITE log data block. 314 */ 315 static int 316 zil_read_log_data(zilog_t *zilog, const lr_write_t *lr, void *wbuf) 317 { 318 zio_flag_t zio_flags = ZIO_FLAG_CANFAIL; 319 const blkptr_t *bp = &lr->lr_blkptr; 320 arc_flags_t aflags = ARC_FLAG_WAIT; 321 arc_buf_t *abuf = NULL; 322 zbookmark_phys_t zb; 323 int error; 324 325 if (BP_IS_HOLE(bp)) { 326 if (wbuf != NULL) 327 memset(wbuf, 0, MAX(BP_GET_LSIZE(bp), lr->lr_length)); 328 return (0); 329 } 330 331 if (zilog->zl_header->zh_claim_txg == 0) 332 zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB; 333 334 /* 335 * If we are not using the resulting data, we are just checking that 336 * it hasn't been corrupted so we don't need to waste CPU time 337 * decompressing and decrypting it. 338 */ 339 if (wbuf == NULL) 340 zio_flags |= ZIO_FLAG_RAW; 341 342 ASSERT3U(BP_GET_LSIZE(bp), !=, 0); 343 SET_BOOKMARK(&zb, dmu_objset_id(zilog->zl_os), lr->lr_foid, 344 ZB_ZIL_LEVEL, lr->lr_offset / BP_GET_LSIZE(bp)); 345 346 error = arc_read(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf, 347 ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb); 348 349 if (error == 0) { 350 if (wbuf != NULL) 351 memcpy(wbuf, abuf->b_data, arc_buf_size(abuf)); 352 arc_buf_destroy(abuf, &abuf); 353 } 354 355 return (error); 356 } 357 358 void 359 zil_sums_init(zil_sums_t *zs) 360 { 361 wmsum_init(&zs->zil_commit_count, 0); 362 wmsum_init(&zs->zil_commit_writer_count, 0); 363 wmsum_init(&zs->zil_itx_count, 0); 364 wmsum_init(&zs->zil_itx_indirect_count, 0); 365 wmsum_init(&zs->zil_itx_indirect_bytes, 0); 366 wmsum_init(&zs->zil_itx_copied_count, 0); 367 wmsum_init(&zs->zil_itx_copied_bytes, 0); 368 wmsum_init(&zs->zil_itx_needcopy_count, 0); 369 wmsum_init(&zs->zil_itx_needcopy_bytes, 0); 370 wmsum_init(&zs->zil_itx_metaslab_normal_count, 0); 371 wmsum_init(&zs->zil_itx_metaslab_normal_bytes, 0); 372 wmsum_init(&zs->zil_itx_metaslab_normal_write, 0); 373 wmsum_init(&zs->zil_itx_metaslab_normal_alloc, 0); 374 wmsum_init(&zs->zil_itx_metaslab_slog_count, 0); 375 wmsum_init(&zs->zil_itx_metaslab_slog_bytes, 0); 376 wmsum_init(&zs->zil_itx_metaslab_slog_write, 0); 377 wmsum_init(&zs->zil_itx_metaslab_slog_alloc, 0); 378 } 379 380 void 381 zil_sums_fini(zil_sums_t *zs) 382 { 383 wmsum_fini(&zs->zil_commit_count); 384 wmsum_fini(&zs->zil_commit_writer_count); 385 wmsum_fini(&zs->zil_itx_count); 386 wmsum_fini(&zs->zil_itx_indirect_count); 387 wmsum_fini(&zs->zil_itx_indirect_bytes); 388 wmsum_fini(&zs->zil_itx_copied_count); 389 wmsum_fini(&zs->zil_itx_copied_bytes); 390 wmsum_fini(&zs->zil_itx_needcopy_count); 391 wmsum_fini(&zs->zil_itx_needcopy_bytes); 392 wmsum_fini(&zs->zil_itx_metaslab_normal_count); 393 wmsum_fini(&zs->zil_itx_metaslab_normal_bytes); 394 wmsum_fini(&zs->zil_itx_metaslab_normal_write); 395 wmsum_fini(&zs->zil_itx_metaslab_normal_alloc); 396 wmsum_fini(&zs->zil_itx_metaslab_slog_count); 397 wmsum_fini(&zs->zil_itx_metaslab_slog_bytes); 398 wmsum_fini(&zs->zil_itx_metaslab_slog_write); 399 wmsum_fini(&zs->zil_itx_metaslab_slog_alloc); 400 } 401 402 void 403 zil_kstat_values_update(zil_kstat_values_t *zs, zil_sums_t *zil_sums) 404 { 405 zs->zil_commit_count.value.ui64 = 406 wmsum_value(&zil_sums->zil_commit_count); 407 zs->zil_commit_writer_count.value.ui64 = 408 wmsum_value(&zil_sums->zil_commit_writer_count); 409 zs->zil_itx_count.value.ui64 = 410 wmsum_value(&zil_sums->zil_itx_count); 411 zs->zil_itx_indirect_count.value.ui64 = 412 wmsum_value(&zil_sums->zil_itx_indirect_count); 413 zs->zil_itx_indirect_bytes.value.ui64 = 414 wmsum_value(&zil_sums->zil_itx_indirect_bytes); 415 zs->zil_itx_copied_count.value.ui64 = 416 wmsum_value(&zil_sums->zil_itx_copied_count); 417 zs->zil_itx_copied_bytes.value.ui64 = 418 wmsum_value(&zil_sums->zil_itx_copied_bytes); 419 zs->zil_itx_needcopy_count.value.ui64 = 420 wmsum_value(&zil_sums->zil_itx_needcopy_count); 421 zs->zil_itx_needcopy_bytes.value.ui64 = 422 wmsum_value(&zil_sums->zil_itx_needcopy_bytes); 423 zs->zil_itx_metaslab_normal_count.value.ui64 = 424 wmsum_value(&zil_sums->zil_itx_metaslab_normal_count); 425 zs->zil_itx_metaslab_normal_bytes.value.ui64 = 426 wmsum_value(&zil_sums->zil_itx_metaslab_normal_bytes); 427 zs->zil_itx_metaslab_normal_write.value.ui64 = 428 wmsum_value(&zil_sums->zil_itx_metaslab_normal_write); 429 zs->zil_itx_metaslab_normal_alloc.value.ui64 = 430 wmsum_value(&zil_sums->zil_itx_metaslab_normal_alloc); 431 zs->zil_itx_metaslab_slog_count.value.ui64 = 432 wmsum_value(&zil_sums->zil_itx_metaslab_slog_count); 433 zs->zil_itx_metaslab_slog_bytes.value.ui64 = 434 wmsum_value(&zil_sums->zil_itx_metaslab_slog_bytes); 435 zs->zil_itx_metaslab_slog_write.value.ui64 = 436 wmsum_value(&zil_sums->zil_itx_metaslab_slog_write); 437 zs->zil_itx_metaslab_slog_alloc.value.ui64 = 438 wmsum_value(&zil_sums->zil_itx_metaslab_slog_alloc); 439 } 440 441 /* 442 * Parse the intent log, and call parse_func for each valid record within. 443 */ 444 int 445 zil_parse(zilog_t *zilog, zil_parse_blk_func_t *parse_blk_func, 446 zil_parse_lr_func_t *parse_lr_func, void *arg, uint64_t txg, 447 boolean_t decrypt) 448 { 449 const zil_header_t *zh = zilog->zl_header; 450 boolean_t claimed = !!zh->zh_claim_txg; 451 uint64_t claim_blk_seq = claimed ? zh->zh_claim_blk_seq : UINT64_MAX; 452 uint64_t claim_lr_seq = claimed ? zh->zh_claim_lr_seq : UINT64_MAX; 453 uint64_t max_blk_seq = 0; 454 uint64_t max_lr_seq = 0; 455 uint64_t blk_count = 0; 456 uint64_t lr_count = 0; 457 blkptr_t blk, next_blk = {{{{0}}}}; 458 int error = 0; 459 460 /* 461 * Old logs didn't record the maximum zh_claim_lr_seq. 462 */ 463 if (!(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID)) 464 claim_lr_seq = UINT64_MAX; 465 466 /* 467 * Starting at the block pointed to by zh_log we read the log chain. 468 * For each block in the chain we strongly check that block to 469 * ensure its validity. We stop when an invalid block is found. 470 * For each block pointer in the chain we call parse_blk_func(). 471 * For each record in each valid block we call parse_lr_func(). 472 * If the log has been claimed, stop if we encounter a sequence 473 * number greater than the highest claimed sequence number. 474 */ 475 zil_bp_tree_init(zilog); 476 477 for (blk = zh->zh_log; !BP_IS_HOLE(&blk); blk = next_blk) { 478 uint64_t blk_seq = blk.blk_cksum.zc_word[ZIL_ZC_SEQ]; 479 int reclen; 480 char *lrp, *end; 481 arc_buf_t *abuf = NULL; 482 483 if (blk_seq > claim_blk_seq) 484 break; 485 486 error = parse_blk_func(zilog, &blk, arg, txg); 487 if (error != 0) 488 break; 489 ASSERT3U(max_blk_seq, <, blk_seq); 490 max_blk_seq = blk_seq; 491 blk_count++; 492 493 if (max_lr_seq == claim_lr_seq && max_blk_seq == claim_blk_seq) 494 break; 495 496 error = zil_read_log_block(zilog, decrypt, &blk, &next_blk, 497 &lrp, &end, &abuf); 498 if (error != 0) { 499 if (abuf) 500 arc_buf_destroy(abuf, &abuf); 501 if (claimed) { 502 char name[ZFS_MAX_DATASET_NAME_LEN]; 503 504 dmu_objset_name(zilog->zl_os, name); 505 506 cmn_err(CE_WARN, "ZFS read log block error %d, " 507 "dataset %s, seq 0x%llx\n", error, name, 508 (u_longlong_t)blk_seq); 509 } 510 break; 511 } 512 513 for (; lrp < end; lrp += reclen) { 514 lr_t *lr = (lr_t *)lrp; 515 reclen = lr->lrc_reclen; 516 ASSERT3U(reclen, >=, sizeof (lr_t)); 517 ASSERT3U(reclen, <=, end - lrp); 518 if (lr->lrc_seq > claim_lr_seq) { 519 arc_buf_destroy(abuf, &abuf); 520 goto done; 521 } 522 523 error = parse_lr_func(zilog, lr, arg, txg); 524 if (error != 0) { 525 arc_buf_destroy(abuf, &abuf); 526 goto done; 527 } 528 ASSERT3U(max_lr_seq, <, lr->lrc_seq); 529 max_lr_seq = lr->lrc_seq; 530 lr_count++; 531 } 532 arc_buf_destroy(abuf, &abuf); 533 } 534 done: 535 zilog->zl_parse_error = error; 536 zilog->zl_parse_blk_seq = max_blk_seq; 537 zilog->zl_parse_lr_seq = max_lr_seq; 538 zilog->zl_parse_blk_count = blk_count; 539 zilog->zl_parse_lr_count = lr_count; 540 541 zil_bp_tree_fini(zilog); 542 543 return (error); 544 } 545 546 static int 547 zil_clear_log_block(zilog_t *zilog, const blkptr_t *bp, void *tx, 548 uint64_t first_txg) 549 { 550 (void) tx; 551 ASSERT(!BP_IS_HOLE(bp)); 552 553 /* 554 * As we call this function from the context of a rewind to a 555 * checkpoint, each ZIL block whose txg is later than the txg 556 * that we rewind to is invalid. Thus, we return -1 so 557 * zil_parse() doesn't attempt to read it. 558 */ 559 if (BP_GET_LOGICAL_BIRTH(bp) >= first_txg) 560 return (-1); 561 562 if (zil_bp_tree_add(zilog, bp) != 0) 563 return (0); 564 565 zio_free(zilog->zl_spa, first_txg, bp); 566 return (0); 567 } 568 569 static int 570 zil_noop_log_record(zilog_t *zilog, const lr_t *lrc, void *tx, 571 uint64_t first_txg) 572 { 573 (void) zilog, (void) lrc, (void) tx, (void) first_txg; 574 return (0); 575 } 576 577 static int 578 zil_claim_log_block(zilog_t *zilog, const blkptr_t *bp, void *tx, 579 uint64_t first_txg) 580 { 581 /* 582 * Claim log block if not already committed and not already claimed. 583 * If tx == NULL, just verify that the block is claimable. 584 */ 585 if (BP_IS_HOLE(bp) || BP_GET_LOGICAL_BIRTH(bp) < first_txg || 586 zil_bp_tree_add(zilog, bp) != 0) 587 return (0); 588 589 return (zio_wait(zio_claim(NULL, zilog->zl_spa, 590 tx == NULL ? 0 : first_txg, bp, spa_claim_notify, NULL, 591 ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB))); 592 } 593 594 static int 595 zil_claim_write(zilog_t *zilog, const lr_t *lrc, void *tx, uint64_t first_txg) 596 { 597 lr_write_t *lr = (lr_write_t *)lrc; 598 int error; 599 600 ASSERT3U(lrc->lrc_reclen, >=, sizeof (*lr)); 601 602 /* 603 * If the block is not readable, don't claim it. This can happen 604 * in normal operation when a log block is written to disk before 605 * some of the dmu_sync() blocks it points to. In this case, the 606 * transaction cannot have been committed to anyone (we would have 607 * waited for all writes to be stable first), so it is semantically 608 * correct to declare this the end of the log. 609 */ 610 if (BP_GET_LOGICAL_BIRTH(&lr->lr_blkptr) >= first_txg) { 611 error = zil_read_log_data(zilog, lr, NULL); 612 if (error != 0) 613 return (error); 614 } 615 616 return (zil_claim_log_block(zilog, &lr->lr_blkptr, tx, first_txg)); 617 } 618 619 static int 620 zil_claim_clone_range(zilog_t *zilog, const lr_t *lrc, void *tx, 621 uint64_t first_txg) 622 { 623 const lr_clone_range_t *lr = (const lr_clone_range_t *)lrc; 624 const blkptr_t *bp; 625 spa_t *spa = zilog->zl_spa; 626 uint_t ii; 627 628 ASSERT3U(lrc->lrc_reclen, >=, sizeof (*lr)); 629 ASSERT3U(lrc->lrc_reclen, >=, offsetof(lr_clone_range_t, 630 lr_bps[lr->lr_nbps])); 631 632 if (tx == NULL) { 633 return (0); 634 } 635 636 /* 637 * XXX: Do we need to byteswap lr? 638 */ 639 640 for (ii = 0; ii < lr->lr_nbps; ii++) { 641 bp = &lr->lr_bps[ii]; 642 643 /* 644 * When data is embedded into the BP there is no need to create 645 * BRT entry as there is no data block. Just copy the BP as it 646 * contains the data. 647 */ 648 if (BP_IS_HOLE(bp) || BP_IS_EMBEDDED(bp)) 649 continue; 650 651 /* 652 * We can not handle block pointers from the future, since they 653 * are not yet allocated. It should not normally happen, but 654 * just in case lets be safe and just stop here now instead of 655 * corrupting the pool. 656 */ 657 if (BP_GET_BIRTH(bp) >= first_txg) 658 return (SET_ERROR(ENOENT)); 659 660 /* 661 * Assert the block is really allocated before we reference it. 662 */ 663 metaslab_check_free(spa, bp); 664 } 665 666 for (ii = 0; ii < lr->lr_nbps; ii++) { 667 bp = &lr->lr_bps[ii]; 668 if (!BP_IS_HOLE(bp) && !BP_IS_EMBEDDED(bp)) 669 brt_pending_add(spa, bp, tx); 670 } 671 672 return (0); 673 } 674 675 static int 676 zil_claim_log_record(zilog_t *zilog, const lr_t *lrc, void *tx, 677 uint64_t first_txg) 678 { 679 680 switch (lrc->lrc_txtype) { 681 case TX_WRITE: 682 return (zil_claim_write(zilog, lrc, tx, first_txg)); 683 case TX_CLONE_RANGE: 684 return (zil_claim_clone_range(zilog, lrc, tx, first_txg)); 685 default: 686 return (0); 687 } 688 } 689 690 static int 691 zil_free_log_block(zilog_t *zilog, const blkptr_t *bp, void *tx, 692 uint64_t claim_txg) 693 { 694 (void) claim_txg; 695 696 zio_free(zilog->zl_spa, dmu_tx_get_txg(tx), bp); 697 698 return (0); 699 } 700 701 static int 702 zil_free_write(zilog_t *zilog, const lr_t *lrc, void *tx, uint64_t claim_txg) 703 { 704 lr_write_t *lr = (lr_write_t *)lrc; 705 blkptr_t *bp = &lr->lr_blkptr; 706 707 ASSERT3U(lrc->lrc_reclen, >=, sizeof (*lr)); 708 709 /* 710 * If we previously claimed it, we need to free it. 711 */ 712 if (BP_GET_LOGICAL_BIRTH(bp) >= claim_txg && 713 zil_bp_tree_add(zilog, bp) == 0 && !BP_IS_HOLE(bp)) { 714 zio_free(zilog->zl_spa, dmu_tx_get_txg(tx), bp); 715 } 716 717 return (0); 718 } 719 720 static int 721 zil_free_clone_range(zilog_t *zilog, const lr_t *lrc, void *tx) 722 { 723 const lr_clone_range_t *lr = (const lr_clone_range_t *)lrc; 724 const blkptr_t *bp; 725 spa_t *spa; 726 uint_t ii; 727 728 ASSERT3U(lrc->lrc_reclen, >=, sizeof (*lr)); 729 ASSERT3U(lrc->lrc_reclen, >=, offsetof(lr_clone_range_t, 730 lr_bps[lr->lr_nbps])); 731 732 if (tx == NULL) { 733 return (0); 734 } 735 736 spa = zilog->zl_spa; 737 738 for (ii = 0; ii < lr->lr_nbps; ii++) { 739 bp = &lr->lr_bps[ii]; 740 741 if (!BP_IS_HOLE(bp)) { 742 zio_free(spa, dmu_tx_get_txg(tx), bp); 743 } 744 } 745 746 return (0); 747 } 748 749 static int 750 zil_free_log_record(zilog_t *zilog, const lr_t *lrc, void *tx, 751 uint64_t claim_txg) 752 { 753 754 if (claim_txg == 0) { 755 return (0); 756 } 757 758 switch (lrc->lrc_txtype) { 759 case TX_WRITE: 760 return (zil_free_write(zilog, lrc, tx, claim_txg)); 761 case TX_CLONE_RANGE: 762 return (zil_free_clone_range(zilog, lrc, tx)); 763 default: 764 return (0); 765 } 766 } 767 768 static int 769 zil_lwb_vdev_compare(const void *x1, const void *x2) 770 { 771 const uint64_t v1 = ((zil_vdev_node_t *)x1)->zv_vdev; 772 const uint64_t v2 = ((zil_vdev_node_t *)x2)->zv_vdev; 773 774 return (TREE_CMP(v1, v2)); 775 } 776 777 /* 778 * Allocate a new lwb. We may already have a block pointer for it, in which 779 * case we get size and version from there. Or we may not yet, in which case 780 * we choose them here and later make the block allocation match. 781 */ 782 static lwb_t * 783 zil_alloc_lwb(zilog_t *zilog, int sz, blkptr_t *bp, boolean_t slog, 784 uint64_t txg, lwb_state_t state) 785 { 786 lwb_t *lwb; 787 788 lwb = kmem_cache_alloc(zil_lwb_cache, KM_SLEEP); 789 lwb->lwb_zilog = zilog; 790 if (bp) { 791 lwb->lwb_blk = *bp; 792 lwb->lwb_slim = (BP_GET_CHECKSUM(bp) == ZIO_CHECKSUM_ZILOG2); 793 sz = BP_GET_LSIZE(bp); 794 } else { 795 BP_ZERO(&lwb->lwb_blk); 796 lwb->lwb_slim = (spa_version(zilog->zl_spa) >= 797 SPA_VERSION_SLIM_ZIL); 798 } 799 lwb->lwb_slog = slog; 800 lwb->lwb_error = 0; 801 if (lwb->lwb_slim) { 802 lwb->lwb_nmax = sz; 803 lwb->lwb_nused = lwb->lwb_nfilled = sizeof (zil_chain_t); 804 } else { 805 lwb->lwb_nmax = sz - sizeof (zil_chain_t); 806 lwb->lwb_nused = lwb->lwb_nfilled = 0; 807 } 808 lwb->lwb_sz = sz; 809 lwb->lwb_state = state; 810 lwb->lwb_buf = zio_buf_alloc(sz); 811 lwb->lwb_child_zio = NULL; 812 lwb->lwb_write_zio = NULL; 813 lwb->lwb_root_zio = NULL; 814 lwb->lwb_issued_timestamp = 0; 815 lwb->lwb_issued_txg = 0; 816 lwb->lwb_alloc_txg = txg; 817 lwb->lwb_max_txg = 0; 818 819 mutex_enter(&zilog->zl_lock); 820 list_insert_tail(&zilog->zl_lwb_list, lwb); 821 if (state != LWB_STATE_NEW) 822 zilog->zl_last_lwb_opened = lwb; 823 mutex_exit(&zilog->zl_lock); 824 825 return (lwb); 826 } 827 828 static void 829 zil_free_lwb(zilog_t *zilog, lwb_t *lwb) 830 { 831 ASSERT(MUTEX_HELD(&zilog->zl_lock)); 832 ASSERT(lwb->lwb_state == LWB_STATE_NEW || 833 lwb->lwb_state == LWB_STATE_FLUSH_DONE); 834 ASSERT3P(lwb->lwb_child_zio, ==, NULL); 835 ASSERT3P(lwb->lwb_write_zio, ==, NULL); 836 ASSERT3P(lwb->lwb_root_zio, ==, NULL); 837 ASSERT3U(lwb->lwb_alloc_txg, <=, spa_syncing_txg(zilog->zl_spa)); 838 ASSERT3U(lwb->lwb_max_txg, <=, spa_syncing_txg(zilog->zl_spa)); 839 VERIFY(list_is_empty(&lwb->lwb_itxs)); 840 VERIFY(list_is_empty(&lwb->lwb_waiters)); 841 ASSERT(avl_is_empty(&lwb->lwb_vdev_tree)); 842 ASSERT(!MUTEX_HELD(&lwb->lwb_vdev_lock)); 843 844 /* 845 * Clear the zilog's field to indicate this lwb is no longer 846 * valid, and prevent use-after-free errors. 847 */ 848 if (zilog->zl_last_lwb_opened == lwb) 849 zilog->zl_last_lwb_opened = NULL; 850 851 kmem_cache_free(zil_lwb_cache, lwb); 852 } 853 854 /* 855 * Called when we create in-memory log transactions so that we know 856 * to cleanup the itxs at the end of spa_sync(). 857 */ 858 static void 859 zilog_dirty(zilog_t *zilog, uint64_t txg) 860 { 861 dsl_pool_t *dp = zilog->zl_dmu_pool; 862 dsl_dataset_t *ds = dmu_objset_ds(zilog->zl_os); 863 864 ASSERT(spa_writeable(zilog->zl_spa)); 865 866 if (ds->ds_is_snapshot) 867 panic("dirtying snapshot!"); 868 869 if (txg_list_add(&dp->dp_dirty_zilogs, zilog, txg)) { 870 /* up the hold count until we can be written out */ 871 dmu_buf_add_ref(ds->ds_dbuf, zilog); 872 873 zilog->zl_dirty_max_txg = MAX(txg, zilog->zl_dirty_max_txg); 874 } 875 } 876 877 /* 878 * Determine if the zil is dirty in the specified txg. Callers wanting to 879 * ensure that the dirty state does not change must hold the itxg_lock for 880 * the specified txg. Holding the lock will ensure that the zil cannot be 881 * dirtied (zil_itx_assign) or cleaned (zil_clean) while we check its current 882 * state. 883 */ 884 static boolean_t __maybe_unused 885 zilog_is_dirty_in_txg(zilog_t *zilog, uint64_t txg) 886 { 887 dsl_pool_t *dp = zilog->zl_dmu_pool; 888 889 if (txg_list_member(&dp->dp_dirty_zilogs, zilog, txg & TXG_MASK)) 890 return (B_TRUE); 891 return (B_FALSE); 892 } 893 894 /* 895 * Determine if the zil is dirty. The zil is considered dirty if it has 896 * any pending itx records that have not been cleaned by zil_clean(). 897 */ 898 static boolean_t 899 zilog_is_dirty(zilog_t *zilog) 900 { 901 dsl_pool_t *dp = zilog->zl_dmu_pool; 902 903 for (int t = 0; t < TXG_SIZE; t++) { 904 if (txg_list_member(&dp->dp_dirty_zilogs, zilog, t)) 905 return (B_TRUE); 906 } 907 return (B_FALSE); 908 } 909 910 /* 911 * Its called in zil_commit context (zil_process_commit_list()/zil_create()). 912 * It activates SPA_FEATURE_ZILSAXATTR feature, if its enabled. 913 * Check dsl_dataset_feature_is_active to avoid txg_wait_synced() on every 914 * zil_commit. 915 */ 916 static void 917 zil_commit_activate_saxattr_feature(zilog_t *zilog) 918 { 919 dsl_dataset_t *ds = dmu_objset_ds(zilog->zl_os); 920 uint64_t txg = 0; 921 dmu_tx_t *tx = NULL; 922 923 if (spa_feature_is_enabled(zilog->zl_spa, SPA_FEATURE_ZILSAXATTR) && 924 dmu_objset_type(zilog->zl_os) != DMU_OST_ZVOL && 925 !dsl_dataset_feature_is_active(ds, SPA_FEATURE_ZILSAXATTR)) { 926 tx = dmu_tx_create(zilog->zl_os); 927 VERIFY0(dmu_tx_assign(tx, TXG_WAIT)); 928 dsl_dataset_dirty(ds, tx); 929 txg = dmu_tx_get_txg(tx); 930 931 mutex_enter(&ds->ds_lock); 932 ds->ds_feature_activation[SPA_FEATURE_ZILSAXATTR] = 933 (void *)B_TRUE; 934 mutex_exit(&ds->ds_lock); 935 dmu_tx_commit(tx); 936 txg_wait_synced(zilog->zl_dmu_pool, txg); 937 } 938 } 939 940 /* 941 * Create an on-disk intent log. 942 */ 943 static lwb_t * 944 zil_create(zilog_t *zilog) 945 { 946 const zil_header_t *zh = zilog->zl_header; 947 lwb_t *lwb = NULL; 948 uint64_t txg = 0; 949 dmu_tx_t *tx = NULL; 950 blkptr_t blk; 951 int error = 0; 952 boolean_t slog = FALSE; 953 dsl_dataset_t *ds = dmu_objset_ds(zilog->zl_os); 954 955 956 /* 957 * Wait for any previous destroy to complete. 958 */ 959 txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg); 960 961 ASSERT(zh->zh_claim_txg == 0); 962 ASSERT(zh->zh_replay_seq == 0); 963 964 blk = zh->zh_log; 965 966 /* 967 * Allocate an initial log block if: 968 * - there isn't one already 969 * - the existing block is the wrong endianness 970 */ 971 if (BP_IS_HOLE(&blk) || BP_SHOULD_BYTESWAP(&blk)) { 972 tx = dmu_tx_create(zilog->zl_os); 973 VERIFY0(dmu_tx_assign(tx, TXG_WAIT)); 974 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx); 975 txg = dmu_tx_get_txg(tx); 976 977 if (!BP_IS_HOLE(&blk)) { 978 zio_free(zilog->zl_spa, txg, &blk); 979 BP_ZERO(&blk); 980 } 981 982 error = zio_alloc_zil(zilog->zl_spa, zilog->zl_os, txg, &blk, 983 ZIL_MIN_BLKSZ, &slog); 984 if (error == 0) 985 zil_init_log_chain(zilog, &blk); 986 } 987 988 /* 989 * Allocate a log write block (lwb) for the first log block. 990 */ 991 if (error == 0) 992 lwb = zil_alloc_lwb(zilog, 0, &blk, slog, txg, LWB_STATE_NEW); 993 994 /* 995 * If we just allocated the first log block, commit our transaction 996 * and wait for zil_sync() to stuff the block pointer into zh_log. 997 * (zh is part of the MOS, so we cannot modify it in open context.) 998 */ 999 if (tx != NULL) { 1000 /* 1001 * If "zilsaxattr" feature is enabled on zpool, then activate 1002 * it now when we're creating the ZIL chain. We can't wait with 1003 * this until we write the first xattr log record because we 1004 * need to wait for the feature activation to sync out. 1005 */ 1006 if (spa_feature_is_enabled(zilog->zl_spa, 1007 SPA_FEATURE_ZILSAXATTR) && dmu_objset_type(zilog->zl_os) != 1008 DMU_OST_ZVOL) { 1009 mutex_enter(&ds->ds_lock); 1010 ds->ds_feature_activation[SPA_FEATURE_ZILSAXATTR] = 1011 (void *)B_TRUE; 1012 mutex_exit(&ds->ds_lock); 1013 } 1014 1015 dmu_tx_commit(tx); 1016 txg_wait_synced(zilog->zl_dmu_pool, txg); 1017 } else { 1018 /* 1019 * This branch covers the case where we enable the feature on a 1020 * zpool that has existing ZIL headers. 1021 */ 1022 zil_commit_activate_saxattr_feature(zilog); 1023 } 1024 IMPLY(spa_feature_is_enabled(zilog->zl_spa, SPA_FEATURE_ZILSAXATTR) && 1025 dmu_objset_type(zilog->zl_os) != DMU_OST_ZVOL, 1026 dsl_dataset_feature_is_active(ds, SPA_FEATURE_ZILSAXATTR)); 1027 1028 ASSERT(error != 0 || memcmp(&blk, &zh->zh_log, sizeof (blk)) == 0); 1029 IMPLY(error == 0, lwb != NULL); 1030 1031 return (lwb); 1032 } 1033 1034 /* 1035 * In one tx, free all log blocks and clear the log header. If keep_first 1036 * is set, then we're replaying a log with no content. We want to keep the 1037 * first block, however, so that the first synchronous transaction doesn't 1038 * require a txg_wait_synced() in zil_create(). We don't need to 1039 * txg_wait_synced() here either when keep_first is set, because both 1040 * zil_create() and zil_destroy() will wait for any in-progress destroys 1041 * to complete. 1042 * Return B_TRUE if there were any entries to replay. 1043 */ 1044 boolean_t 1045 zil_destroy(zilog_t *zilog, boolean_t keep_first) 1046 { 1047 const zil_header_t *zh = zilog->zl_header; 1048 lwb_t *lwb; 1049 dmu_tx_t *tx; 1050 uint64_t txg; 1051 1052 /* 1053 * Wait for any previous destroy to complete. 1054 */ 1055 txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg); 1056 1057 zilog->zl_old_header = *zh; /* debugging aid */ 1058 1059 if (BP_IS_HOLE(&zh->zh_log)) 1060 return (B_FALSE); 1061 1062 tx = dmu_tx_create(zilog->zl_os); 1063 VERIFY0(dmu_tx_assign(tx, TXG_WAIT)); 1064 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx); 1065 txg = dmu_tx_get_txg(tx); 1066 1067 mutex_enter(&zilog->zl_lock); 1068 1069 ASSERT3U(zilog->zl_destroy_txg, <, txg); 1070 zilog->zl_destroy_txg = txg; 1071 zilog->zl_keep_first = keep_first; 1072 1073 if (!list_is_empty(&zilog->zl_lwb_list)) { 1074 ASSERT(zh->zh_claim_txg == 0); 1075 VERIFY(!keep_first); 1076 while ((lwb = list_remove_head(&zilog->zl_lwb_list)) != NULL) { 1077 if (lwb->lwb_buf != NULL) 1078 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz); 1079 if (!BP_IS_HOLE(&lwb->lwb_blk)) 1080 zio_free(zilog->zl_spa, txg, &lwb->lwb_blk); 1081 zil_free_lwb(zilog, lwb); 1082 } 1083 } else if (!keep_first) { 1084 zil_destroy_sync(zilog, tx); 1085 } 1086 mutex_exit(&zilog->zl_lock); 1087 1088 dmu_tx_commit(tx); 1089 1090 return (B_TRUE); 1091 } 1092 1093 void 1094 zil_destroy_sync(zilog_t *zilog, dmu_tx_t *tx) 1095 { 1096 ASSERT(list_is_empty(&zilog->zl_lwb_list)); 1097 (void) zil_parse(zilog, zil_free_log_block, 1098 zil_free_log_record, tx, zilog->zl_header->zh_claim_txg, B_FALSE); 1099 } 1100 1101 int 1102 zil_claim(dsl_pool_t *dp, dsl_dataset_t *ds, void *txarg) 1103 { 1104 dmu_tx_t *tx = txarg; 1105 zilog_t *zilog; 1106 uint64_t first_txg; 1107 zil_header_t *zh; 1108 objset_t *os; 1109 int error; 1110 1111 error = dmu_objset_own_obj(dp, ds->ds_object, 1112 DMU_OST_ANY, B_FALSE, B_FALSE, FTAG, &os); 1113 if (error != 0) { 1114 /* 1115 * EBUSY indicates that the objset is inconsistent, in which 1116 * case it can not have a ZIL. 1117 */ 1118 if (error != EBUSY) { 1119 cmn_err(CE_WARN, "can't open objset for %llu, error %u", 1120 (unsigned long long)ds->ds_object, error); 1121 } 1122 1123 return (0); 1124 } 1125 1126 zilog = dmu_objset_zil(os); 1127 zh = zil_header_in_syncing_context(zilog); 1128 ASSERT3U(tx->tx_txg, ==, spa_first_txg(zilog->zl_spa)); 1129 first_txg = spa_min_claim_txg(zilog->zl_spa); 1130 1131 /* 1132 * If the spa_log_state is not set to be cleared, check whether 1133 * the current uberblock is a checkpoint one and if the current 1134 * header has been claimed before moving on. 1135 * 1136 * If the current uberblock is a checkpointed uberblock then 1137 * one of the following scenarios took place: 1138 * 1139 * 1] We are currently rewinding to the checkpoint of the pool. 1140 * 2] We crashed in the middle of a checkpoint rewind but we 1141 * did manage to write the checkpointed uberblock to the 1142 * vdev labels, so when we tried to import the pool again 1143 * the checkpointed uberblock was selected from the import 1144 * procedure. 1145 * 1146 * In both cases we want to zero out all the ZIL blocks, except 1147 * the ones that have been claimed at the time of the checkpoint 1148 * (their zh_claim_txg != 0). The reason is that these blocks 1149 * may be corrupted since we may have reused their locations on 1150 * disk after we took the checkpoint. 1151 * 1152 * We could try to set spa_log_state to SPA_LOG_CLEAR earlier 1153 * when we first figure out whether the current uberblock is 1154 * checkpointed or not. Unfortunately, that would discard all 1155 * the logs, including the ones that are claimed, and we would 1156 * leak space. 1157 */ 1158 if (spa_get_log_state(zilog->zl_spa) == SPA_LOG_CLEAR || 1159 (zilog->zl_spa->spa_uberblock.ub_checkpoint_txg != 0 && 1160 zh->zh_claim_txg == 0)) { 1161 if (!BP_IS_HOLE(&zh->zh_log)) { 1162 (void) zil_parse(zilog, zil_clear_log_block, 1163 zil_noop_log_record, tx, first_txg, B_FALSE); 1164 } 1165 BP_ZERO(&zh->zh_log); 1166 if (os->os_encrypted) 1167 os->os_next_write_raw[tx->tx_txg & TXG_MASK] = B_TRUE; 1168 dsl_dataset_dirty(dmu_objset_ds(os), tx); 1169 dmu_objset_disown(os, B_FALSE, FTAG); 1170 return (0); 1171 } 1172 1173 /* 1174 * If we are not rewinding and opening the pool normally, then 1175 * the min_claim_txg should be equal to the first txg of the pool. 1176 */ 1177 ASSERT3U(first_txg, ==, spa_first_txg(zilog->zl_spa)); 1178 1179 /* 1180 * Claim all log blocks if we haven't already done so, and remember 1181 * the highest claimed sequence number. This ensures that if we can 1182 * read only part of the log now (e.g. due to a missing device), 1183 * but we can read the entire log later, we will not try to replay 1184 * or destroy beyond the last block we successfully claimed. 1185 */ 1186 ASSERT3U(zh->zh_claim_txg, <=, first_txg); 1187 if (zh->zh_claim_txg == 0 && !BP_IS_HOLE(&zh->zh_log)) { 1188 (void) zil_parse(zilog, zil_claim_log_block, 1189 zil_claim_log_record, tx, first_txg, B_FALSE); 1190 zh->zh_claim_txg = first_txg; 1191 zh->zh_claim_blk_seq = zilog->zl_parse_blk_seq; 1192 zh->zh_claim_lr_seq = zilog->zl_parse_lr_seq; 1193 if (zilog->zl_parse_lr_count || zilog->zl_parse_blk_count > 1) 1194 zh->zh_flags |= ZIL_REPLAY_NEEDED; 1195 zh->zh_flags |= ZIL_CLAIM_LR_SEQ_VALID; 1196 if (os->os_encrypted) 1197 os->os_next_write_raw[tx->tx_txg & TXG_MASK] = B_TRUE; 1198 dsl_dataset_dirty(dmu_objset_ds(os), tx); 1199 } 1200 1201 ASSERT3U(first_txg, ==, (spa_last_synced_txg(zilog->zl_spa) + 1)); 1202 dmu_objset_disown(os, B_FALSE, FTAG); 1203 return (0); 1204 } 1205 1206 /* 1207 * Check the log by walking the log chain. 1208 * Checksum errors are ok as they indicate the end of the chain. 1209 * Any other error (no device or read failure) returns an error. 1210 */ 1211 int 1212 zil_check_log_chain(dsl_pool_t *dp, dsl_dataset_t *ds, void *tx) 1213 { 1214 (void) dp; 1215 zilog_t *zilog; 1216 objset_t *os; 1217 blkptr_t *bp; 1218 int error; 1219 1220 ASSERT(tx == NULL); 1221 1222 error = dmu_objset_from_ds(ds, &os); 1223 if (error != 0) { 1224 cmn_err(CE_WARN, "can't open objset %llu, error %d", 1225 (unsigned long long)ds->ds_object, error); 1226 return (0); 1227 } 1228 1229 zilog = dmu_objset_zil(os); 1230 bp = (blkptr_t *)&zilog->zl_header->zh_log; 1231 1232 if (!BP_IS_HOLE(bp)) { 1233 vdev_t *vd; 1234 boolean_t valid = B_TRUE; 1235 1236 /* 1237 * Check the first block and determine if it's on a log device 1238 * which may have been removed or faulted prior to loading this 1239 * pool. If so, there's no point in checking the rest of the 1240 * log as its content should have already been synced to the 1241 * pool. 1242 */ 1243 spa_config_enter(os->os_spa, SCL_STATE, FTAG, RW_READER); 1244 vd = vdev_lookup_top(os->os_spa, DVA_GET_VDEV(&bp->blk_dva[0])); 1245 if (vd->vdev_islog && vdev_is_dead(vd)) 1246 valid = vdev_log_state_valid(vd); 1247 spa_config_exit(os->os_spa, SCL_STATE, FTAG); 1248 1249 if (!valid) 1250 return (0); 1251 1252 /* 1253 * Check whether the current uberblock is checkpointed (e.g. 1254 * we are rewinding) and whether the current header has been 1255 * claimed or not. If it hasn't then skip verifying it. We 1256 * do this because its ZIL blocks may be part of the pool's 1257 * state before the rewind, which is no longer valid. 1258 */ 1259 zil_header_t *zh = zil_header_in_syncing_context(zilog); 1260 if (zilog->zl_spa->spa_uberblock.ub_checkpoint_txg != 0 && 1261 zh->zh_claim_txg == 0) 1262 return (0); 1263 } 1264 1265 /* 1266 * Because tx == NULL, zil_claim_log_block() will not actually claim 1267 * any blocks, but just determine whether it is possible to do so. 1268 * In addition to checking the log chain, zil_claim_log_block() 1269 * will invoke zio_claim() with a done func of spa_claim_notify(), 1270 * which will update spa_max_claim_txg. See spa_load() for details. 1271 */ 1272 error = zil_parse(zilog, zil_claim_log_block, zil_claim_log_record, tx, 1273 zilog->zl_header->zh_claim_txg ? -1ULL : 1274 spa_min_claim_txg(os->os_spa), B_FALSE); 1275 1276 return ((error == ECKSUM || error == ENOENT) ? 0 : error); 1277 } 1278 1279 /* 1280 * When an itx is "skipped", this function is used to properly mark the 1281 * waiter as "done, and signal any thread(s) waiting on it. An itx can 1282 * be skipped (and not committed to an lwb) for a variety of reasons, 1283 * one of them being that the itx was committed via spa_sync(), prior to 1284 * it being committed to an lwb; this can happen if a thread calling 1285 * zil_commit() is racing with spa_sync(). 1286 */ 1287 static void 1288 zil_commit_waiter_skip(zil_commit_waiter_t *zcw) 1289 { 1290 mutex_enter(&zcw->zcw_lock); 1291 ASSERT3B(zcw->zcw_done, ==, B_FALSE); 1292 zcw->zcw_done = B_TRUE; 1293 cv_broadcast(&zcw->zcw_cv); 1294 mutex_exit(&zcw->zcw_lock); 1295 } 1296 1297 /* 1298 * This function is used when the given waiter is to be linked into an 1299 * lwb's "lwb_waiter" list; i.e. when the itx is committed to the lwb. 1300 * At this point, the waiter will no longer be referenced by the itx, 1301 * and instead, will be referenced by the lwb. 1302 */ 1303 static void 1304 zil_commit_waiter_link_lwb(zil_commit_waiter_t *zcw, lwb_t *lwb) 1305 { 1306 /* 1307 * The lwb_waiters field of the lwb is protected by the zilog's 1308 * zl_issuer_lock while the lwb is open and zl_lock otherwise. 1309 * zl_issuer_lock also protects leaving the open state. 1310 * zcw_lwb setting is protected by zl_issuer_lock and state != 1311 * flush_done, which transition is protected by zl_lock. 1312 */ 1313 ASSERT(MUTEX_HELD(&lwb->lwb_zilog->zl_issuer_lock)); 1314 IMPLY(lwb->lwb_state != LWB_STATE_OPENED, 1315 MUTEX_HELD(&lwb->lwb_zilog->zl_lock)); 1316 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_NEW); 1317 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_FLUSH_DONE); 1318 1319 ASSERT(!list_link_active(&zcw->zcw_node)); 1320 list_insert_tail(&lwb->lwb_waiters, zcw); 1321 ASSERT3P(zcw->zcw_lwb, ==, NULL); 1322 zcw->zcw_lwb = lwb; 1323 } 1324 1325 /* 1326 * This function is used when zio_alloc_zil() fails to allocate a ZIL 1327 * block, and the given waiter must be linked to the "nolwb waiters" 1328 * list inside of zil_process_commit_list(). 1329 */ 1330 static void 1331 zil_commit_waiter_link_nolwb(zil_commit_waiter_t *zcw, list_t *nolwb) 1332 { 1333 ASSERT(!list_link_active(&zcw->zcw_node)); 1334 list_insert_tail(nolwb, zcw); 1335 ASSERT3P(zcw->zcw_lwb, ==, NULL); 1336 } 1337 1338 void 1339 zil_lwb_add_block(lwb_t *lwb, const blkptr_t *bp) 1340 { 1341 avl_tree_t *t = &lwb->lwb_vdev_tree; 1342 avl_index_t where; 1343 zil_vdev_node_t *zv, zvsearch; 1344 int ndvas = BP_GET_NDVAS(bp); 1345 int i; 1346 1347 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_WRITE_DONE); 1348 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_FLUSH_DONE); 1349 1350 if (zil_nocacheflush) 1351 return; 1352 1353 mutex_enter(&lwb->lwb_vdev_lock); 1354 for (i = 0; i < ndvas; i++) { 1355 zvsearch.zv_vdev = DVA_GET_VDEV(&bp->blk_dva[i]); 1356 if (avl_find(t, &zvsearch, &where) == NULL) { 1357 zv = kmem_alloc(sizeof (*zv), KM_SLEEP); 1358 zv->zv_vdev = zvsearch.zv_vdev; 1359 avl_insert(t, zv, where); 1360 } 1361 } 1362 mutex_exit(&lwb->lwb_vdev_lock); 1363 } 1364 1365 static void 1366 zil_lwb_flush_defer(lwb_t *lwb, lwb_t *nlwb) 1367 { 1368 avl_tree_t *src = &lwb->lwb_vdev_tree; 1369 avl_tree_t *dst = &nlwb->lwb_vdev_tree; 1370 void *cookie = NULL; 1371 zil_vdev_node_t *zv; 1372 1373 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_WRITE_DONE); 1374 ASSERT3S(nlwb->lwb_state, !=, LWB_STATE_WRITE_DONE); 1375 ASSERT3S(nlwb->lwb_state, !=, LWB_STATE_FLUSH_DONE); 1376 1377 /* 1378 * While 'lwb' is at a point in its lifetime where lwb_vdev_tree does 1379 * not need the protection of lwb_vdev_lock (it will only be modified 1380 * while holding zilog->zl_lock) as its writes and those of its 1381 * children have all completed. The younger 'nlwb' may be waiting on 1382 * future writes to additional vdevs. 1383 */ 1384 mutex_enter(&nlwb->lwb_vdev_lock); 1385 /* 1386 * Tear down the 'lwb' vdev tree, ensuring that entries which do not 1387 * exist in 'nlwb' are moved to it, freeing any would-be duplicates. 1388 */ 1389 while ((zv = avl_destroy_nodes(src, &cookie)) != NULL) { 1390 avl_index_t where; 1391 1392 if (avl_find(dst, zv, &where) == NULL) { 1393 avl_insert(dst, zv, where); 1394 } else { 1395 kmem_free(zv, sizeof (*zv)); 1396 } 1397 } 1398 mutex_exit(&nlwb->lwb_vdev_lock); 1399 } 1400 1401 void 1402 zil_lwb_add_txg(lwb_t *lwb, uint64_t txg) 1403 { 1404 lwb->lwb_max_txg = MAX(lwb->lwb_max_txg, txg); 1405 } 1406 1407 /* 1408 * This function is a called after all vdevs associated with a given lwb write 1409 * have completed their flush command; or as soon as the lwb write completes, 1410 * if "zil_nocacheflush" is set. Further, all "previous" lwb's will have 1411 * completed before this function is called; i.e. this function is called for 1412 * all previous lwbs before it's called for "this" lwb (enforced via zio the 1413 * dependencies configured in zil_lwb_set_zio_dependency()). 1414 * 1415 * The intention is for this function to be called as soon as the contents of 1416 * an lwb are considered "stable" on disk, and will survive any sudden loss of 1417 * power. At this point, any threads waiting for the lwb to reach this state 1418 * are signalled, and the "waiter" structures are marked "done". 1419 */ 1420 static void 1421 zil_lwb_flush_vdevs_done(zio_t *zio) 1422 { 1423 lwb_t *lwb = zio->io_private; 1424 zilog_t *zilog = lwb->lwb_zilog; 1425 zil_commit_waiter_t *zcw; 1426 itx_t *itx; 1427 1428 spa_config_exit(zilog->zl_spa, SCL_STATE, lwb); 1429 1430 hrtime_t t = gethrtime() - lwb->lwb_issued_timestamp; 1431 1432 mutex_enter(&zilog->zl_lock); 1433 1434 zilog->zl_last_lwb_latency = (zilog->zl_last_lwb_latency * 7 + t) / 8; 1435 1436 lwb->lwb_root_zio = NULL; 1437 1438 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_WRITE_DONE); 1439 lwb->lwb_state = LWB_STATE_FLUSH_DONE; 1440 1441 if (zilog->zl_last_lwb_opened == lwb) { 1442 /* 1443 * Remember the highest committed log sequence number 1444 * for ztest. We only update this value when all the log 1445 * writes succeeded, because ztest wants to ASSERT that 1446 * it got the whole log chain. 1447 */ 1448 zilog->zl_commit_lr_seq = zilog->zl_lr_seq; 1449 } 1450 1451 while ((itx = list_remove_head(&lwb->lwb_itxs)) != NULL) 1452 zil_itx_destroy(itx); 1453 1454 while ((zcw = list_remove_head(&lwb->lwb_waiters)) != NULL) { 1455 mutex_enter(&zcw->zcw_lock); 1456 1457 ASSERT3P(zcw->zcw_lwb, ==, lwb); 1458 zcw->zcw_lwb = NULL; 1459 /* 1460 * We expect any ZIO errors from child ZIOs to have been 1461 * propagated "up" to this specific LWB's root ZIO, in 1462 * order for this error handling to work correctly. This 1463 * includes ZIO errors from either this LWB's write or 1464 * flush, as well as any errors from other dependent LWBs 1465 * (e.g. a root LWB ZIO that might be a child of this LWB). 1466 * 1467 * With that said, it's important to note that LWB flush 1468 * errors are not propagated up to the LWB root ZIO. 1469 * This is incorrect behavior, and results in VDEV flush 1470 * errors not being handled correctly here. See the 1471 * comment above the call to "zio_flush" for details. 1472 */ 1473 1474 zcw->zcw_zio_error = zio->io_error; 1475 1476 ASSERT3B(zcw->zcw_done, ==, B_FALSE); 1477 zcw->zcw_done = B_TRUE; 1478 cv_broadcast(&zcw->zcw_cv); 1479 1480 mutex_exit(&zcw->zcw_lock); 1481 } 1482 1483 uint64_t txg = lwb->lwb_issued_txg; 1484 1485 /* Once we drop the lock, lwb may be freed by zil_sync(). */ 1486 mutex_exit(&zilog->zl_lock); 1487 1488 mutex_enter(&zilog->zl_lwb_io_lock); 1489 ASSERT3U(zilog->zl_lwb_inflight[txg & TXG_MASK], >, 0); 1490 zilog->zl_lwb_inflight[txg & TXG_MASK]--; 1491 if (zilog->zl_lwb_inflight[txg & TXG_MASK] == 0) 1492 cv_broadcast(&zilog->zl_lwb_io_cv); 1493 mutex_exit(&zilog->zl_lwb_io_lock); 1494 } 1495 1496 /* 1497 * Wait for the completion of all issued write/flush of that txg provided. 1498 * It guarantees zil_lwb_flush_vdevs_done() is called and returned. 1499 */ 1500 static void 1501 zil_lwb_flush_wait_all(zilog_t *zilog, uint64_t txg) 1502 { 1503 ASSERT3U(txg, ==, spa_syncing_txg(zilog->zl_spa)); 1504 1505 mutex_enter(&zilog->zl_lwb_io_lock); 1506 while (zilog->zl_lwb_inflight[txg & TXG_MASK] > 0) 1507 cv_wait(&zilog->zl_lwb_io_cv, &zilog->zl_lwb_io_lock); 1508 mutex_exit(&zilog->zl_lwb_io_lock); 1509 1510 #ifdef ZFS_DEBUG 1511 mutex_enter(&zilog->zl_lock); 1512 mutex_enter(&zilog->zl_lwb_io_lock); 1513 lwb_t *lwb = list_head(&zilog->zl_lwb_list); 1514 while (lwb != NULL) { 1515 if (lwb->lwb_issued_txg <= txg) { 1516 ASSERT(lwb->lwb_state != LWB_STATE_ISSUED); 1517 ASSERT(lwb->lwb_state != LWB_STATE_WRITE_DONE); 1518 IMPLY(lwb->lwb_issued_txg > 0, 1519 lwb->lwb_state == LWB_STATE_FLUSH_DONE); 1520 } 1521 IMPLY(lwb->lwb_state == LWB_STATE_WRITE_DONE || 1522 lwb->lwb_state == LWB_STATE_FLUSH_DONE, 1523 lwb->lwb_buf == NULL); 1524 lwb = list_next(&zilog->zl_lwb_list, lwb); 1525 } 1526 mutex_exit(&zilog->zl_lwb_io_lock); 1527 mutex_exit(&zilog->zl_lock); 1528 #endif 1529 } 1530 1531 /* 1532 * This is called when an lwb's write zio completes. The callback's purpose is 1533 * to issue the flush commands for the vdevs in the lwb's lwb_vdev_tree. The 1534 * tree will contain the vdevs involved in writing out this specific lwb's 1535 * data, and in the case that cache flushes have been deferred, vdevs involved 1536 * in writing the data for previous lwbs. The writes corresponding to all the 1537 * vdevs in the lwb_vdev_tree will have completed by the time this is called, 1538 * due to the zio dependencies configured in zil_lwb_set_zio_dependency(), 1539 * which takes deferred flushes into account. The lwb will be "done" once 1540 * zil_lwb_flush_vdevs_done() is called, which occurs in the zio completion 1541 * callback for the lwb's root zio. 1542 */ 1543 static void 1544 zil_lwb_write_done(zio_t *zio) 1545 { 1546 lwb_t *lwb = zio->io_private; 1547 spa_t *spa = zio->io_spa; 1548 zilog_t *zilog = lwb->lwb_zilog; 1549 avl_tree_t *t = &lwb->lwb_vdev_tree; 1550 void *cookie = NULL; 1551 zil_vdev_node_t *zv; 1552 lwb_t *nlwb; 1553 1554 ASSERT3S(spa_config_held(spa, SCL_STATE, RW_READER), !=, 0); 1555 1556 abd_free(zio->io_abd); 1557 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz); 1558 lwb->lwb_buf = NULL; 1559 1560 mutex_enter(&zilog->zl_lock); 1561 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_ISSUED); 1562 lwb->lwb_state = LWB_STATE_WRITE_DONE; 1563 lwb->lwb_child_zio = NULL; 1564 lwb->lwb_write_zio = NULL; 1565 1566 /* 1567 * If nlwb is not yet issued, zil_lwb_set_zio_dependency() is not 1568 * called for it yet, and when it will be, it won't be able to make 1569 * its write ZIO a parent this ZIO. In such case we can not defer 1570 * our flushes or below may be a race between the done callbacks. 1571 */ 1572 nlwb = list_next(&zilog->zl_lwb_list, lwb); 1573 if (nlwb && nlwb->lwb_state != LWB_STATE_ISSUED) 1574 nlwb = NULL; 1575 mutex_exit(&zilog->zl_lock); 1576 1577 if (avl_numnodes(t) == 0) 1578 return; 1579 1580 /* 1581 * If there was an IO error, we're not going to call zio_flush() 1582 * on these vdevs, so we simply empty the tree and free the 1583 * nodes. We avoid calling zio_flush() since there isn't any 1584 * good reason for doing so, after the lwb block failed to be 1585 * written out. 1586 * 1587 * Additionally, we don't perform any further error handling at 1588 * this point (e.g. setting "zcw_zio_error" appropriately), as 1589 * we expect that to occur in "zil_lwb_flush_vdevs_done" (thus, 1590 * we expect any error seen here, to have been propagated to 1591 * that function). 1592 */ 1593 if (zio->io_error != 0) { 1594 while ((zv = avl_destroy_nodes(t, &cookie)) != NULL) 1595 kmem_free(zv, sizeof (*zv)); 1596 return; 1597 } 1598 1599 /* 1600 * If this lwb does not have any threads waiting for it to complete, we 1601 * want to defer issuing the flush command to the vdevs written to by 1602 * "this" lwb, and instead rely on the "next" lwb to handle the flush 1603 * command for those vdevs. Thus, we merge the vdev tree of "this" lwb 1604 * with the vdev tree of the "next" lwb in the list, and assume the 1605 * "next" lwb will handle flushing the vdevs (or deferring the flush(s) 1606 * again). 1607 * 1608 * This is a useful performance optimization, especially for workloads 1609 * with lots of async write activity and few sync write and/or fsync 1610 * activity, as it has the potential to coalesce multiple flush 1611 * commands to a vdev into one. 1612 */ 1613 if (list_is_empty(&lwb->lwb_waiters) && nlwb != NULL) { 1614 zil_lwb_flush_defer(lwb, nlwb); 1615 ASSERT(avl_is_empty(&lwb->lwb_vdev_tree)); 1616 return; 1617 } 1618 1619 while ((zv = avl_destroy_nodes(t, &cookie)) != NULL) { 1620 vdev_t *vd = vdev_lookup_top(spa, zv->zv_vdev); 1621 if (vd != NULL) { 1622 /* 1623 * The "ZIO_FLAG_DONT_PROPAGATE" is currently 1624 * always used within "zio_flush". This means, 1625 * any errors when flushing the vdev(s), will 1626 * (unfortunately) not be handled correctly, 1627 * since these "zio_flush" errors will not be 1628 * propagated up to "zil_lwb_flush_vdevs_done". 1629 */ 1630 zio_flush(lwb->lwb_root_zio, vd); 1631 } 1632 kmem_free(zv, sizeof (*zv)); 1633 } 1634 } 1635 1636 /* 1637 * Build the zio dependency chain, which is used to preserve the ordering of 1638 * lwb completions that is required by the semantics of the ZIL. Each new lwb 1639 * zio becomes a parent of the previous lwb zio, such that the new lwb's zio 1640 * cannot complete until the previous lwb's zio completes. 1641 * 1642 * This is required by the semantics of zil_commit(): the commit waiters 1643 * attached to the lwbs will be woken in the lwb zio's completion callback, 1644 * so this zio dependency graph ensures the waiters are woken in the correct 1645 * order (the same order the lwbs were created). 1646 */ 1647 static void 1648 zil_lwb_set_zio_dependency(zilog_t *zilog, lwb_t *lwb) 1649 { 1650 ASSERT(MUTEX_HELD(&zilog->zl_lock)); 1651 1652 lwb_t *prev_lwb = list_prev(&zilog->zl_lwb_list, lwb); 1653 if (prev_lwb == NULL || 1654 prev_lwb->lwb_state == LWB_STATE_FLUSH_DONE) 1655 return; 1656 1657 /* 1658 * If the previous lwb's write hasn't already completed, we also want 1659 * to order the completion of the lwb write zios (above, we only order 1660 * the completion of the lwb root zios). This is required because of 1661 * how we can defer the flush commands for each lwb. 1662 * 1663 * When the flush commands are deferred, the previous lwb will rely on 1664 * this lwb to flush the vdevs written to by that previous lwb. Thus, 1665 * we need to ensure this lwb doesn't issue the flush until after the 1666 * previous lwb's write completes. We ensure this ordering by setting 1667 * the zio parent/child relationship here. 1668 * 1669 * Without this relationship on the lwb's write zio, it's possible for 1670 * this lwb's write to complete prior to the previous lwb's write 1671 * completing; and thus, the vdevs for the previous lwb would be 1672 * flushed prior to that lwb's data being written to those vdevs (the 1673 * vdevs are flushed in the lwb write zio's completion handler, 1674 * zil_lwb_write_done()). 1675 */ 1676 if (prev_lwb->lwb_state == LWB_STATE_ISSUED) { 1677 ASSERT3P(prev_lwb->lwb_write_zio, !=, NULL); 1678 zio_add_child(lwb->lwb_write_zio, prev_lwb->lwb_write_zio); 1679 } else { 1680 ASSERT3S(prev_lwb->lwb_state, ==, LWB_STATE_WRITE_DONE); 1681 } 1682 1683 ASSERT3P(prev_lwb->lwb_root_zio, !=, NULL); 1684 zio_add_child(lwb->lwb_root_zio, prev_lwb->lwb_root_zio); 1685 } 1686 1687 1688 /* 1689 * This function's purpose is to "open" an lwb such that it is ready to 1690 * accept new itxs being committed to it. This function is idempotent; if 1691 * the passed in lwb has already been opened, it is essentially a no-op. 1692 */ 1693 static void 1694 zil_lwb_write_open(zilog_t *zilog, lwb_t *lwb) 1695 { 1696 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock)); 1697 1698 if (lwb->lwb_state != LWB_STATE_NEW) { 1699 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_OPENED); 1700 return; 1701 } 1702 1703 mutex_enter(&zilog->zl_lock); 1704 lwb->lwb_state = LWB_STATE_OPENED; 1705 zilog->zl_last_lwb_opened = lwb; 1706 mutex_exit(&zilog->zl_lock); 1707 } 1708 1709 /* 1710 * Maximum block size used by the ZIL. This is picked up when the ZIL is 1711 * initialized. Otherwise this should not be used directly; see 1712 * zl_max_block_size instead. 1713 */ 1714 static uint_t zil_maxblocksize = SPA_OLD_MAXBLOCKSIZE; 1715 1716 /* 1717 * Plan splitting of the provided burst size between several blocks. 1718 */ 1719 static uint_t 1720 zil_lwb_plan(zilog_t *zilog, uint64_t size, uint_t *minsize) 1721 { 1722 uint_t md = zilog->zl_max_block_size - sizeof (zil_chain_t); 1723 1724 if (size <= md) { 1725 /* 1726 * Small bursts are written as-is in one block. 1727 */ 1728 *minsize = size; 1729 return (size); 1730 } else if (size > 8 * md) { 1731 /* 1732 * Big bursts use maximum blocks. The first block size 1733 * is hard to predict, but it does not really matter. 1734 */ 1735 *minsize = 0; 1736 return (md); 1737 } 1738 1739 /* 1740 * Medium bursts try to divide evenly to better utilize several SLOG 1741 * VDEVs. The first block size we predict assuming the worst case of 1742 * maxing out others. Fall back to using maximum blocks if due to 1743 * large records or wasted space we can not predict anything better. 1744 */ 1745 uint_t s = size; 1746 uint_t n = DIV_ROUND_UP(s, md - sizeof (lr_write_t)); 1747 uint_t chunk = DIV_ROUND_UP(s, n); 1748 uint_t waste = zil_max_waste_space(zilog); 1749 waste = MAX(waste, zilog->zl_cur_max); 1750 if (chunk <= md - waste) { 1751 *minsize = MAX(s - (md - waste) * (n - 1), waste); 1752 return (chunk); 1753 } else { 1754 *minsize = 0; 1755 return (md); 1756 } 1757 } 1758 1759 /* 1760 * Try to predict next block size based on previous history. Make prediction 1761 * sufficient for 7 of 8 previous bursts. Don't try to save if the saving is 1762 * less then 50%, extra writes may cost more, but we don't want single spike 1763 * to badly affect our predictions. 1764 */ 1765 static uint_t 1766 zil_lwb_predict(zilog_t *zilog) 1767 { 1768 uint_t m, o; 1769 1770 /* If we are in the middle of a burst, take it into account also. */ 1771 if (zilog->zl_cur_size > 0) { 1772 o = zil_lwb_plan(zilog, zilog->zl_cur_size, &m); 1773 } else { 1774 o = UINT_MAX; 1775 m = 0; 1776 } 1777 1778 /* Find minimum optimal size. We don't need to go below that. */ 1779 for (int i = 0; i < ZIL_BURSTS; i++) 1780 o = MIN(o, zilog->zl_prev_opt[i]); 1781 1782 /* Find two biggest minimal first block sizes above the optimal. */ 1783 uint_t m1 = MAX(m, o), m2 = o; 1784 for (int i = 0; i < ZIL_BURSTS; i++) { 1785 m = zilog->zl_prev_min[i]; 1786 if (m >= m1) { 1787 m2 = m1; 1788 m1 = m; 1789 } else if (m > m2) { 1790 m2 = m; 1791 } 1792 } 1793 1794 /* 1795 * If second minimum size gives 50% saving -- use it. It may cost us 1796 * one additional write later, but the space saving is just too big. 1797 */ 1798 return ((m1 < m2 * 2) ? m1 : m2); 1799 } 1800 1801 /* 1802 * Close the log block for being issued and allocate the next one. 1803 * Has to be called under zl_issuer_lock to chain more lwbs. 1804 */ 1805 static lwb_t * 1806 zil_lwb_write_close(zilog_t *zilog, lwb_t *lwb, lwb_state_t state) 1807 { 1808 uint64_t blksz, plan, plan2; 1809 1810 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock)); 1811 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_OPENED); 1812 lwb->lwb_state = LWB_STATE_CLOSED; 1813 1814 /* 1815 * If there was an allocation failure then returned NULL will trigger 1816 * zil_commit_writer_stall() at the caller. This is inherently racy, 1817 * since allocation may not have happened yet. 1818 */ 1819 if (lwb->lwb_error != 0) 1820 return (NULL); 1821 1822 /* 1823 * Log blocks are pre-allocated. Here we select the size of the next 1824 * block, based on what's left of this burst and the previous history. 1825 * While we try to only write used part of the block, we can't just 1826 * always allocate the maximum block size because we can exhaust all 1827 * available pool log space, so we try to be reasonable. 1828 */ 1829 if (zilog->zl_cur_left > 0) { 1830 /* 1831 * We are in the middle of a burst and know how much is left. 1832 * But if workload is multi-threaded there may be more soon. 1833 * Try to predict what can it be and plan for the worst case. 1834 */ 1835 uint_t m; 1836 plan = zil_lwb_plan(zilog, zilog->zl_cur_left, &m); 1837 if (zilog->zl_parallel) { 1838 plan2 = zil_lwb_plan(zilog, zilog->zl_cur_left + 1839 zil_lwb_predict(zilog), &m); 1840 if (plan < plan2) 1841 plan = plan2; 1842 } 1843 } else { 1844 /* 1845 * The previous burst is done and we can only predict what 1846 * will come next. 1847 */ 1848 plan = zil_lwb_predict(zilog); 1849 } 1850 blksz = plan + sizeof (zil_chain_t); 1851 blksz = P2ROUNDUP_TYPED(blksz, ZIL_MIN_BLKSZ, uint64_t); 1852 blksz = MIN(blksz, zilog->zl_max_block_size); 1853 DTRACE_PROBE3(zil__block__size, zilog_t *, zilog, uint64_t, blksz, 1854 uint64_t, plan); 1855 1856 return (zil_alloc_lwb(zilog, blksz, NULL, 0, 0, state)); 1857 } 1858 1859 /* 1860 * Finalize previously closed block and issue the write zio. 1861 */ 1862 static void 1863 zil_lwb_write_issue(zilog_t *zilog, lwb_t *lwb) 1864 { 1865 spa_t *spa = zilog->zl_spa; 1866 zil_chain_t *zilc; 1867 boolean_t slog; 1868 zbookmark_phys_t zb; 1869 zio_priority_t prio; 1870 int error; 1871 1872 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_CLOSED); 1873 1874 /* Actually fill the lwb with the data. */ 1875 for (itx_t *itx = list_head(&lwb->lwb_itxs); itx; 1876 itx = list_next(&lwb->lwb_itxs, itx)) 1877 zil_lwb_commit(zilog, lwb, itx); 1878 lwb->lwb_nused = lwb->lwb_nfilled; 1879 ASSERT3U(lwb->lwb_nused, <=, lwb->lwb_nmax); 1880 1881 lwb->lwb_root_zio = zio_root(spa, zil_lwb_flush_vdevs_done, lwb, 1882 ZIO_FLAG_CANFAIL); 1883 1884 /* 1885 * The lwb is now ready to be issued, but it can be only if it already 1886 * got its block pointer allocated or the allocation has failed. 1887 * Otherwise leave it as-is, relying on some other thread to issue it 1888 * after allocating its block pointer via calling zil_lwb_write_issue() 1889 * for the previous lwb(s) in the chain. 1890 */ 1891 mutex_enter(&zilog->zl_lock); 1892 lwb->lwb_state = LWB_STATE_READY; 1893 if (BP_IS_HOLE(&lwb->lwb_blk) && lwb->lwb_error == 0) { 1894 mutex_exit(&zilog->zl_lock); 1895 return; 1896 } 1897 mutex_exit(&zilog->zl_lock); 1898 1899 next_lwb: 1900 if (lwb->lwb_slim) 1901 zilc = (zil_chain_t *)lwb->lwb_buf; 1902 else 1903 zilc = (zil_chain_t *)(lwb->lwb_buf + lwb->lwb_nmax); 1904 int wsz = lwb->lwb_sz; 1905 if (lwb->lwb_error == 0) { 1906 abd_t *lwb_abd = abd_get_from_buf(lwb->lwb_buf, lwb->lwb_sz); 1907 if (!lwb->lwb_slog || zilog->zl_cur_size <= zil_slog_bulk) 1908 prio = ZIO_PRIORITY_SYNC_WRITE; 1909 else 1910 prio = ZIO_PRIORITY_ASYNC_WRITE; 1911 SET_BOOKMARK(&zb, lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_OBJSET], 1912 ZB_ZIL_OBJECT, ZB_ZIL_LEVEL, 1913 lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_SEQ]); 1914 lwb->lwb_write_zio = zio_rewrite(lwb->lwb_root_zio, spa, 0, 1915 &lwb->lwb_blk, lwb_abd, lwb->lwb_sz, zil_lwb_write_done, 1916 lwb, prio, ZIO_FLAG_CANFAIL, &zb); 1917 zil_lwb_add_block(lwb, &lwb->lwb_blk); 1918 1919 if (lwb->lwb_slim) { 1920 /* For Slim ZIL only write what is used. */ 1921 wsz = P2ROUNDUP_TYPED(lwb->lwb_nused, ZIL_MIN_BLKSZ, 1922 int); 1923 ASSERT3S(wsz, <=, lwb->lwb_sz); 1924 zio_shrink(lwb->lwb_write_zio, wsz); 1925 wsz = lwb->lwb_write_zio->io_size; 1926 } 1927 memset(lwb->lwb_buf + lwb->lwb_nused, 0, wsz - lwb->lwb_nused); 1928 zilc->zc_pad = 0; 1929 zilc->zc_nused = lwb->lwb_nused; 1930 zilc->zc_eck.zec_cksum = lwb->lwb_blk.blk_cksum; 1931 } else { 1932 /* 1933 * We can't write the lwb if there was an allocation failure, 1934 * so create a null zio instead just to maintain dependencies. 1935 */ 1936 lwb->lwb_write_zio = zio_null(lwb->lwb_root_zio, spa, NULL, 1937 zil_lwb_write_done, lwb, ZIO_FLAG_CANFAIL); 1938 lwb->lwb_write_zio->io_error = lwb->lwb_error; 1939 } 1940 if (lwb->lwb_child_zio) 1941 zio_add_child(lwb->lwb_write_zio, lwb->lwb_child_zio); 1942 1943 /* 1944 * Open transaction to allocate the next block pointer. 1945 */ 1946 dmu_tx_t *tx = dmu_tx_create(zilog->zl_os); 1947 VERIFY0(dmu_tx_assign(tx, TXG_WAIT | TXG_NOTHROTTLE)); 1948 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx); 1949 uint64_t txg = dmu_tx_get_txg(tx); 1950 1951 /* 1952 * Allocate next the block pointer unless we are already in error. 1953 */ 1954 lwb_t *nlwb = list_next(&zilog->zl_lwb_list, lwb); 1955 blkptr_t *bp = &zilc->zc_next_blk; 1956 BP_ZERO(bp); 1957 error = lwb->lwb_error; 1958 if (error == 0) { 1959 error = zio_alloc_zil(spa, zilog->zl_os, txg, bp, nlwb->lwb_sz, 1960 &slog); 1961 } 1962 if (error == 0) { 1963 ASSERT3U(BP_GET_LOGICAL_BIRTH(bp), ==, txg); 1964 BP_SET_CHECKSUM(bp, nlwb->lwb_slim ? ZIO_CHECKSUM_ZILOG2 : 1965 ZIO_CHECKSUM_ZILOG); 1966 bp->blk_cksum = lwb->lwb_blk.blk_cksum; 1967 bp->blk_cksum.zc_word[ZIL_ZC_SEQ]++; 1968 } 1969 1970 /* 1971 * Reduce TXG open time by incrementing inflight counter and committing 1972 * the transaciton. zil_sync() will wait for it to return to zero. 1973 */ 1974 mutex_enter(&zilog->zl_lwb_io_lock); 1975 lwb->lwb_issued_txg = txg; 1976 zilog->zl_lwb_inflight[txg & TXG_MASK]++; 1977 zilog->zl_lwb_max_issued_txg = MAX(txg, zilog->zl_lwb_max_issued_txg); 1978 mutex_exit(&zilog->zl_lwb_io_lock); 1979 dmu_tx_commit(tx); 1980 1981 spa_config_enter(spa, SCL_STATE, lwb, RW_READER); 1982 1983 /* 1984 * We've completed all potentially blocking operations. Update the 1985 * nlwb and allow it proceed without possible lock order reversals. 1986 */ 1987 mutex_enter(&zilog->zl_lock); 1988 zil_lwb_set_zio_dependency(zilog, lwb); 1989 lwb->lwb_state = LWB_STATE_ISSUED; 1990 1991 if (nlwb) { 1992 nlwb->lwb_blk = *bp; 1993 nlwb->lwb_error = error; 1994 nlwb->lwb_slog = slog; 1995 nlwb->lwb_alloc_txg = txg; 1996 if (nlwb->lwb_state != LWB_STATE_READY) 1997 nlwb = NULL; 1998 } 1999 mutex_exit(&zilog->zl_lock); 2000 2001 if (lwb->lwb_slog) { 2002 ZIL_STAT_BUMP(zilog, zil_itx_metaslab_slog_count); 2003 ZIL_STAT_INCR(zilog, zil_itx_metaslab_slog_bytes, 2004 lwb->lwb_nused); 2005 ZIL_STAT_INCR(zilog, zil_itx_metaslab_slog_write, 2006 wsz); 2007 ZIL_STAT_INCR(zilog, zil_itx_metaslab_slog_alloc, 2008 BP_GET_LSIZE(&lwb->lwb_blk)); 2009 } else { 2010 ZIL_STAT_BUMP(zilog, zil_itx_metaslab_normal_count); 2011 ZIL_STAT_INCR(zilog, zil_itx_metaslab_normal_bytes, 2012 lwb->lwb_nused); 2013 ZIL_STAT_INCR(zilog, zil_itx_metaslab_normal_write, 2014 wsz); 2015 ZIL_STAT_INCR(zilog, zil_itx_metaslab_normal_alloc, 2016 BP_GET_LSIZE(&lwb->lwb_blk)); 2017 } 2018 lwb->lwb_issued_timestamp = gethrtime(); 2019 if (lwb->lwb_child_zio) 2020 zio_nowait(lwb->lwb_child_zio); 2021 zio_nowait(lwb->lwb_write_zio); 2022 zio_nowait(lwb->lwb_root_zio); 2023 2024 /* 2025 * If nlwb was ready when we gave it the block pointer, 2026 * it is on us to issue it and possibly following ones. 2027 */ 2028 lwb = nlwb; 2029 if (lwb) 2030 goto next_lwb; 2031 } 2032 2033 /* 2034 * Maximum amount of data that can be put into single log block. 2035 */ 2036 uint64_t 2037 zil_max_log_data(zilog_t *zilog, size_t hdrsize) 2038 { 2039 return (zilog->zl_max_block_size - sizeof (zil_chain_t) - hdrsize); 2040 } 2041 2042 /* 2043 * Maximum amount of log space we agree to waste to reduce number of 2044 * WR_NEED_COPY chunks to reduce zl_get_data() overhead (~6%). 2045 */ 2046 static inline uint64_t 2047 zil_max_waste_space(zilog_t *zilog) 2048 { 2049 return (zil_max_log_data(zilog, sizeof (lr_write_t)) / 16); 2050 } 2051 2052 /* 2053 * Maximum amount of write data for WR_COPIED. For correctness, consumers 2054 * must fall back to WR_NEED_COPY if we can't fit the entire record into one 2055 * maximum sized log block, because each WR_COPIED record must fit in a 2056 * single log block. Below that it is a tradeoff of additional memory copy 2057 * and possibly worse log space efficiency vs additional range lock/unlock. 2058 */ 2059 static uint_t zil_maxcopied = 7680; 2060 2061 uint64_t 2062 zil_max_copied_data(zilog_t *zilog) 2063 { 2064 uint64_t max_data = zil_max_log_data(zilog, sizeof (lr_write_t)); 2065 return (MIN(max_data, zil_maxcopied)); 2066 } 2067 2068 static uint64_t 2069 zil_itx_record_size(itx_t *itx) 2070 { 2071 lr_t *lr = &itx->itx_lr; 2072 2073 if (lr->lrc_txtype == TX_COMMIT) 2074 return (0); 2075 ASSERT3U(lr->lrc_reclen, >=, sizeof (lr_t)); 2076 return (lr->lrc_reclen); 2077 } 2078 2079 static uint64_t 2080 zil_itx_data_size(itx_t *itx) 2081 { 2082 lr_t *lr = &itx->itx_lr; 2083 lr_write_t *lrw = (lr_write_t *)lr; 2084 2085 if (lr->lrc_txtype == TX_WRITE && itx->itx_wr_state == WR_NEED_COPY) { 2086 ASSERT3U(lr->lrc_reclen, ==, sizeof (lr_write_t)); 2087 return (P2ROUNDUP_TYPED(lrw->lr_length, sizeof (uint64_t), 2088 uint64_t)); 2089 } 2090 return (0); 2091 } 2092 2093 static uint64_t 2094 zil_itx_full_size(itx_t *itx) 2095 { 2096 lr_t *lr = &itx->itx_lr; 2097 2098 if (lr->lrc_txtype == TX_COMMIT) 2099 return (0); 2100 ASSERT3U(lr->lrc_reclen, >=, sizeof (lr_t)); 2101 return (lr->lrc_reclen + zil_itx_data_size(itx)); 2102 } 2103 2104 /* 2105 * Estimate space needed in the lwb for the itx. Allocate more lwbs or 2106 * split the itx as needed, but don't touch the actual transaction data. 2107 * Has to be called under zl_issuer_lock to call zil_lwb_write_close() 2108 * to chain more lwbs. 2109 */ 2110 static lwb_t * 2111 zil_lwb_assign(zilog_t *zilog, lwb_t *lwb, itx_t *itx, list_t *ilwbs) 2112 { 2113 itx_t *citx; 2114 lr_t *lr, *clr; 2115 lr_write_t *lrw; 2116 uint64_t dlen, dnow, lwb_sp, reclen, max_log_data; 2117 2118 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock)); 2119 ASSERT3P(lwb, !=, NULL); 2120 ASSERT3P(lwb->lwb_buf, !=, NULL); 2121 2122 zil_lwb_write_open(zilog, lwb); 2123 2124 lr = &itx->itx_lr; 2125 lrw = (lr_write_t *)lr; 2126 2127 /* 2128 * A commit itx doesn't represent any on-disk state; instead 2129 * it's simply used as a place holder on the commit list, and 2130 * provides a mechanism for attaching a "commit waiter" onto the 2131 * correct lwb (such that the waiter can be signalled upon 2132 * completion of that lwb). Thus, we don't process this itx's 2133 * log record if it's a commit itx (these itx's don't have log 2134 * records), and instead link the itx's waiter onto the lwb's 2135 * list of waiters. 2136 * 2137 * For more details, see the comment above zil_commit(). 2138 */ 2139 if (lr->lrc_txtype == TX_COMMIT) { 2140 zil_commit_waiter_link_lwb(itx->itx_private, lwb); 2141 list_insert_tail(&lwb->lwb_itxs, itx); 2142 return (lwb); 2143 } 2144 2145 reclen = lr->lrc_reclen; 2146 ASSERT3U(reclen, >=, sizeof (lr_t)); 2147 ASSERT3U(reclen, <=, zil_max_log_data(zilog, 0)); 2148 dlen = zil_itx_data_size(itx); 2149 2150 cont: 2151 /* 2152 * If this record won't fit in the current log block, start a new one. 2153 * For WR_NEED_COPY optimize layout for minimal number of chunks. 2154 */ 2155 lwb_sp = lwb->lwb_nmax - lwb->lwb_nused; 2156 max_log_data = zil_max_log_data(zilog, sizeof (lr_write_t)); 2157 if (reclen > lwb_sp || (reclen + dlen > lwb_sp && 2158 lwb_sp < zil_max_waste_space(zilog) && 2159 (dlen % max_log_data == 0 || 2160 lwb_sp < reclen + dlen % max_log_data))) { 2161 list_insert_tail(ilwbs, lwb); 2162 lwb = zil_lwb_write_close(zilog, lwb, LWB_STATE_OPENED); 2163 if (lwb == NULL) 2164 return (NULL); 2165 lwb_sp = lwb->lwb_nmax - lwb->lwb_nused; 2166 } 2167 2168 /* 2169 * There must be enough space in the log block to hold reclen. 2170 * For WR_COPIED, we need to fit the whole record in one block, 2171 * and reclen is the write record header size + the data size. 2172 * For WR_NEED_COPY, we can create multiple records, splitting 2173 * the data into multiple blocks, so we only need to fit one 2174 * word of data per block; in this case reclen is just the header 2175 * size (no data). 2176 */ 2177 ASSERT3U(reclen + MIN(dlen, sizeof (uint64_t)), <=, lwb_sp); 2178 2179 dnow = MIN(dlen, lwb_sp - reclen); 2180 if (dlen > dnow) { 2181 ASSERT3U(lr->lrc_txtype, ==, TX_WRITE); 2182 ASSERT3U(itx->itx_wr_state, ==, WR_NEED_COPY); 2183 citx = zil_itx_clone(itx); 2184 clr = &citx->itx_lr; 2185 lr_write_t *clrw = (lr_write_t *)clr; 2186 clrw->lr_length = dnow; 2187 lrw->lr_offset += dnow; 2188 lrw->lr_length -= dnow; 2189 zilog->zl_cur_left -= dnow; 2190 } else { 2191 citx = itx; 2192 clr = lr; 2193 } 2194 2195 /* 2196 * We're actually making an entry, so update lrc_seq to be the 2197 * log record sequence number. Note that this is generally not 2198 * equal to the itx sequence number because not all transactions 2199 * are synchronous, and sometimes spa_sync() gets there first. 2200 */ 2201 clr->lrc_seq = ++zilog->zl_lr_seq; 2202 2203 lwb->lwb_nused += reclen + dnow; 2204 ASSERT3U(lwb->lwb_nused, <=, lwb->lwb_nmax); 2205 ASSERT0(P2PHASE(lwb->lwb_nused, sizeof (uint64_t))); 2206 2207 zil_lwb_add_txg(lwb, lr->lrc_txg); 2208 list_insert_tail(&lwb->lwb_itxs, citx); 2209 2210 dlen -= dnow; 2211 if (dlen > 0) 2212 goto cont; 2213 2214 if (lr->lrc_txtype == TX_WRITE && 2215 lr->lrc_txg > spa_freeze_txg(zilog->zl_spa)) 2216 txg_wait_synced(zilog->zl_dmu_pool, lr->lrc_txg); 2217 2218 return (lwb); 2219 } 2220 2221 /* 2222 * Fill the actual transaction data into the lwb, following zil_lwb_assign(). 2223 * Does not require locking. 2224 */ 2225 static void 2226 zil_lwb_commit(zilog_t *zilog, lwb_t *lwb, itx_t *itx) 2227 { 2228 lr_t *lr, *lrb; 2229 lr_write_t *lrw, *lrwb; 2230 char *lr_buf; 2231 uint64_t dlen, reclen; 2232 2233 lr = &itx->itx_lr; 2234 lrw = (lr_write_t *)lr; 2235 2236 if (lr->lrc_txtype == TX_COMMIT) 2237 return; 2238 2239 reclen = lr->lrc_reclen; 2240 dlen = zil_itx_data_size(itx); 2241 ASSERT3U(reclen + dlen, <=, lwb->lwb_nused - lwb->lwb_nfilled); 2242 2243 lr_buf = lwb->lwb_buf + lwb->lwb_nfilled; 2244 memcpy(lr_buf, lr, reclen); 2245 lrb = (lr_t *)lr_buf; /* Like lr, but inside lwb. */ 2246 lrwb = (lr_write_t *)lrb; /* Like lrw, but inside lwb. */ 2247 2248 ZIL_STAT_BUMP(zilog, zil_itx_count); 2249 2250 /* 2251 * If it's a write, fetch the data or get its blkptr as appropriate. 2252 */ 2253 if (lr->lrc_txtype == TX_WRITE) { 2254 if (itx->itx_wr_state == WR_COPIED) { 2255 ZIL_STAT_BUMP(zilog, zil_itx_copied_count); 2256 ZIL_STAT_INCR(zilog, zil_itx_copied_bytes, 2257 lrw->lr_length); 2258 } else { 2259 char *dbuf; 2260 int error; 2261 2262 if (itx->itx_wr_state == WR_NEED_COPY) { 2263 dbuf = lr_buf + reclen; 2264 lrb->lrc_reclen += dlen; 2265 ZIL_STAT_BUMP(zilog, zil_itx_needcopy_count); 2266 ZIL_STAT_INCR(zilog, zil_itx_needcopy_bytes, 2267 dlen); 2268 } else { 2269 ASSERT3S(itx->itx_wr_state, ==, WR_INDIRECT); 2270 dbuf = NULL; 2271 ZIL_STAT_BUMP(zilog, zil_itx_indirect_count); 2272 ZIL_STAT_INCR(zilog, zil_itx_indirect_bytes, 2273 lrw->lr_length); 2274 if (lwb->lwb_child_zio == NULL) { 2275 lwb->lwb_child_zio = zio_null(NULL, 2276 zilog->zl_spa, NULL, NULL, NULL, 2277 ZIO_FLAG_CANFAIL); 2278 } 2279 } 2280 2281 /* 2282 * The "lwb_child_zio" we pass in will become a child of 2283 * "lwb_write_zio", when one is created, so one will be 2284 * a parent of any zio's created by the "zl_get_data". 2285 * This way "lwb_write_zio" will first wait for children 2286 * block pointers before own writing, and then for their 2287 * writing completion before the vdev cache flushing. 2288 */ 2289 error = zilog->zl_get_data(itx->itx_private, 2290 itx->itx_gen, lrwb, dbuf, lwb, 2291 lwb->lwb_child_zio); 2292 if (dbuf != NULL && error == 0) { 2293 /* Zero any padding bytes in the last block. */ 2294 memset((char *)dbuf + lrwb->lr_length, 0, 2295 dlen - lrwb->lr_length); 2296 } 2297 2298 /* 2299 * Typically, the only return values we should see from 2300 * ->zl_get_data() are 0, EIO, ENOENT, EEXIST or 2301 * EALREADY. However, it is also possible to see other 2302 * error values such as ENOSPC or EINVAL from 2303 * dmu_read() -> dnode_hold() -> dnode_hold_impl() or 2304 * ENXIO as well as a multitude of others from the 2305 * block layer through dmu_buf_hold() -> dbuf_read() 2306 * -> zio_wait(), as well as through dmu_read() -> 2307 * dnode_hold() -> dnode_hold_impl() -> dbuf_read() -> 2308 * zio_wait(). When these errors happen, we can assume 2309 * that neither an immediate write nor an indirect 2310 * write occurred, so we need to fall back to 2311 * txg_wait_synced(). This is unusual, so we print to 2312 * dmesg whenever one of these errors occurs. 2313 */ 2314 switch (error) { 2315 case 0: 2316 break; 2317 default: 2318 cmn_err(CE_WARN, "zil_lwb_commit() received " 2319 "unexpected error %d from ->zl_get_data()" 2320 ". Falling back to txg_wait_synced().", 2321 error); 2322 zfs_fallthrough; 2323 case EIO: 2324 txg_wait_synced(zilog->zl_dmu_pool, 2325 lr->lrc_txg); 2326 zfs_fallthrough; 2327 case ENOENT: 2328 zfs_fallthrough; 2329 case EEXIST: 2330 zfs_fallthrough; 2331 case EALREADY: 2332 return; 2333 } 2334 } 2335 } 2336 2337 lwb->lwb_nfilled += reclen + dlen; 2338 ASSERT3S(lwb->lwb_nfilled, <=, lwb->lwb_nused); 2339 ASSERT0(P2PHASE(lwb->lwb_nfilled, sizeof (uint64_t))); 2340 } 2341 2342 itx_t * 2343 zil_itx_create(uint64_t txtype, size_t olrsize) 2344 { 2345 size_t itxsize, lrsize; 2346 itx_t *itx; 2347 2348 ASSERT3U(olrsize, >=, sizeof (lr_t)); 2349 lrsize = P2ROUNDUP_TYPED(olrsize, sizeof (uint64_t), size_t); 2350 ASSERT3U(lrsize, >=, olrsize); 2351 itxsize = offsetof(itx_t, itx_lr) + lrsize; 2352 2353 itx = zio_data_buf_alloc(itxsize); 2354 itx->itx_lr.lrc_txtype = txtype; 2355 itx->itx_lr.lrc_reclen = lrsize; 2356 itx->itx_lr.lrc_seq = 0; /* defensive */ 2357 memset((char *)&itx->itx_lr + olrsize, 0, lrsize - olrsize); 2358 itx->itx_sync = B_TRUE; /* default is synchronous */ 2359 itx->itx_callback = NULL; 2360 itx->itx_callback_data = NULL; 2361 itx->itx_size = itxsize; 2362 2363 return (itx); 2364 } 2365 2366 static itx_t * 2367 zil_itx_clone(itx_t *oitx) 2368 { 2369 ASSERT3U(oitx->itx_size, >=, sizeof (itx_t)); 2370 ASSERT3U(oitx->itx_size, ==, 2371 offsetof(itx_t, itx_lr) + oitx->itx_lr.lrc_reclen); 2372 2373 itx_t *itx = zio_data_buf_alloc(oitx->itx_size); 2374 memcpy(itx, oitx, oitx->itx_size); 2375 itx->itx_callback = NULL; 2376 itx->itx_callback_data = NULL; 2377 return (itx); 2378 } 2379 2380 void 2381 zil_itx_destroy(itx_t *itx) 2382 { 2383 ASSERT3U(itx->itx_size, >=, sizeof (itx_t)); 2384 ASSERT3U(itx->itx_lr.lrc_reclen, ==, 2385 itx->itx_size - offsetof(itx_t, itx_lr)); 2386 IMPLY(itx->itx_lr.lrc_txtype == TX_COMMIT, itx->itx_callback == NULL); 2387 IMPLY(itx->itx_callback != NULL, itx->itx_lr.lrc_txtype != TX_COMMIT); 2388 2389 if (itx->itx_callback != NULL) 2390 itx->itx_callback(itx->itx_callback_data); 2391 2392 zio_data_buf_free(itx, itx->itx_size); 2393 } 2394 2395 /* 2396 * Free up the sync and async itxs. The itxs_t has already been detached 2397 * so no locks are needed. 2398 */ 2399 static void 2400 zil_itxg_clean(void *arg) 2401 { 2402 itx_t *itx; 2403 list_t *list; 2404 avl_tree_t *t; 2405 void *cookie; 2406 itxs_t *itxs = arg; 2407 itx_async_node_t *ian; 2408 2409 list = &itxs->i_sync_list; 2410 while ((itx = list_remove_head(list)) != NULL) { 2411 /* 2412 * In the general case, commit itxs will not be found 2413 * here, as they'll be committed to an lwb via 2414 * zil_lwb_assign(), and free'd in that function. Having 2415 * said that, it is still possible for commit itxs to be 2416 * found here, due to the following race: 2417 * 2418 * - a thread calls zil_commit() which assigns the 2419 * commit itx to a per-txg i_sync_list 2420 * - zil_itxg_clean() is called (e.g. via spa_sync()) 2421 * while the waiter is still on the i_sync_list 2422 * 2423 * There's nothing to prevent syncing the txg while the 2424 * waiter is on the i_sync_list. This normally doesn't 2425 * happen because spa_sync() is slower than zil_commit(), 2426 * but if zil_commit() calls txg_wait_synced() (e.g. 2427 * because zil_create() or zil_commit_writer_stall() is 2428 * called) we will hit this case. 2429 */ 2430 if (itx->itx_lr.lrc_txtype == TX_COMMIT) 2431 zil_commit_waiter_skip(itx->itx_private); 2432 2433 zil_itx_destroy(itx); 2434 } 2435 2436 cookie = NULL; 2437 t = &itxs->i_async_tree; 2438 while ((ian = avl_destroy_nodes(t, &cookie)) != NULL) { 2439 list = &ian->ia_list; 2440 while ((itx = list_remove_head(list)) != NULL) { 2441 /* commit itxs should never be on the async lists. */ 2442 ASSERT3U(itx->itx_lr.lrc_txtype, !=, TX_COMMIT); 2443 zil_itx_destroy(itx); 2444 } 2445 list_destroy(list); 2446 kmem_free(ian, sizeof (itx_async_node_t)); 2447 } 2448 avl_destroy(t); 2449 2450 kmem_free(itxs, sizeof (itxs_t)); 2451 } 2452 2453 static int 2454 zil_aitx_compare(const void *x1, const void *x2) 2455 { 2456 const uint64_t o1 = ((itx_async_node_t *)x1)->ia_foid; 2457 const uint64_t o2 = ((itx_async_node_t *)x2)->ia_foid; 2458 2459 return (TREE_CMP(o1, o2)); 2460 } 2461 2462 /* 2463 * Remove all async itx with the given oid. 2464 */ 2465 void 2466 zil_remove_async(zilog_t *zilog, uint64_t oid) 2467 { 2468 uint64_t otxg, txg; 2469 itx_async_node_t *ian, ian_search; 2470 avl_tree_t *t; 2471 avl_index_t where; 2472 list_t clean_list; 2473 itx_t *itx; 2474 2475 ASSERT(oid != 0); 2476 list_create(&clean_list, sizeof (itx_t), offsetof(itx_t, itx_node)); 2477 2478 if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */ 2479 otxg = ZILTEST_TXG; 2480 else 2481 otxg = spa_last_synced_txg(zilog->zl_spa) + 1; 2482 2483 for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) { 2484 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK]; 2485 2486 mutex_enter(&itxg->itxg_lock); 2487 if (itxg->itxg_txg != txg) { 2488 mutex_exit(&itxg->itxg_lock); 2489 continue; 2490 } 2491 2492 /* 2493 * Locate the object node and append its list. 2494 */ 2495 t = &itxg->itxg_itxs->i_async_tree; 2496 ian_search.ia_foid = oid; 2497 ian = avl_find(t, &ian_search, &where); 2498 if (ian != NULL) 2499 list_move_tail(&clean_list, &ian->ia_list); 2500 mutex_exit(&itxg->itxg_lock); 2501 } 2502 while ((itx = list_remove_head(&clean_list)) != NULL) { 2503 /* commit itxs should never be on the async lists. */ 2504 ASSERT3U(itx->itx_lr.lrc_txtype, !=, TX_COMMIT); 2505 zil_itx_destroy(itx); 2506 } 2507 list_destroy(&clean_list); 2508 } 2509 2510 void 2511 zil_itx_assign(zilog_t *zilog, itx_t *itx, dmu_tx_t *tx) 2512 { 2513 uint64_t txg; 2514 itxg_t *itxg; 2515 itxs_t *itxs, *clean = NULL; 2516 2517 /* 2518 * Ensure the data of a renamed file is committed before the rename. 2519 */ 2520 if ((itx->itx_lr.lrc_txtype & ~TX_CI) == TX_RENAME) 2521 zil_async_to_sync(zilog, itx->itx_oid); 2522 2523 if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) 2524 txg = ZILTEST_TXG; 2525 else 2526 txg = dmu_tx_get_txg(tx); 2527 2528 itxg = &zilog->zl_itxg[txg & TXG_MASK]; 2529 mutex_enter(&itxg->itxg_lock); 2530 itxs = itxg->itxg_itxs; 2531 if (itxg->itxg_txg != txg) { 2532 if (itxs != NULL) { 2533 /* 2534 * The zil_clean callback hasn't got around to cleaning 2535 * this itxg. Save the itxs for release below. 2536 * This should be rare. 2537 */ 2538 zfs_dbgmsg("zil_itx_assign: missed itx cleanup for " 2539 "txg %llu", (u_longlong_t)itxg->itxg_txg); 2540 clean = itxg->itxg_itxs; 2541 } 2542 itxg->itxg_txg = txg; 2543 itxs = itxg->itxg_itxs = kmem_zalloc(sizeof (itxs_t), 2544 KM_SLEEP); 2545 2546 list_create(&itxs->i_sync_list, sizeof (itx_t), 2547 offsetof(itx_t, itx_node)); 2548 avl_create(&itxs->i_async_tree, zil_aitx_compare, 2549 sizeof (itx_async_node_t), 2550 offsetof(itx_async_node_t, ia_node)); 2551 } 2552 if (itx->itx_sync) { 2553 list_insert_tail(&itxs->i_sync_list, itx); 2554 } else { 2555 avl_tree_t *t = &itxs->i_async_tree; 2556 uint64_t foid = 2557 LR_FOID_GET_OBJ(((lr_ooo_t *)&itx->itx_lr)->lr_foid); 2558 itx_async_node_t *ian; 2559 avl_index_t where; 2560 2561 ian = avl_find(t, &foid, &where); 2562 if (ian == NULL) { 2563 ian = kmem_alloc(sizeof (itx_async_node_t), 2564 KM_SLEEP); 2565 list_create(&ian->ia_list, sizeof (itx_t), 2566 offsetof(itx_t, itx_node)); 2567 ian->ia_foid = foid; 2568 avl_insert(t, ian, where); 2569 } 2570 list_insert_tail(&ian->ia_list, itx); 2571 } 2572 2573 itx->itx_lr.lrc_txg = dmu_tx_get_txg(tx); 2574 2575 /* 2576 * We don't want to dirty the ZIL using ZILTEST_TXG, because 2577 * zil_clean() will never be called using ZILTEST_TXG. Thus, we 2578 * need to be careful to always dirty the ZIL using the "real" 2579 * TXG (not itxg_txg) even when the SPA is frozen. 2580 */ 2581 zilog_dirty(zilog, dmu_tx_get_txg(tx)); 2582 mutex_exit(&itxg->itxg_lock); 2583 2584 /* Release the old itxs now we've dropped the lock */ 2585 if (clean != NULL) 2586 zil_itxg_clean(clean); 2587 } 2588 2589 /* 2590 * If there are any in-memory intent log transactions which have now been 2591 * synced then start up a taskq to free them. We should only do this after we 2592 * have written out the uberblocks (i.e. txg has been committed) so that 2593 * don't inadvertently clean out in-memory log records that would be required 2594 * by zil_commit(). 2595 */ 2596 void 2597 zil_clean(zilog_t *zilog, uint64_t synced_txg) 2598 { 2599 itxg_t *itxg = &zilog->zl_itxg[synced_txg & TXG_MASK]; 2600 itxs_t *clean_me; 2601 2602 ASSERT3U(synced_txg, <, ZILTEST_TXG); 2603 2604 mutex_enter(&itxg->itxg_lock); 2605 if (itxg->itxg_itxs == NULL || itxg->itxg_txg == ZILTEST_TXG) { 2606 mutex_exit(&itxg->itxg_lock); 2607 return; 2608 } 2609 ASSERT3U(itxg->itxg_txg, <=, synced_txg); 2610 ASSERT3U(itxg->itxg_txg, !=, 0); 2611 clean_me = itxg->itxg_itxs; 2612 itxg->itxg_itxs = NULL; 2613 itxg->itxg_txg = 0; 2614 mutex_exit(&itxg->itxg_lock); 2615 /* 2616 * Preferably start a task queue to free up the old itxs but 2617 * if taskq_dispatch can't allocate resources to do that then 2618 * free it in-line. This should be rare. Note, using TQ_SLEEP 2619 * created a bad performance problem. 2620 */ 2621 ASSERT3P(zilog->zl_dmu_pool, !=, NULL); 2622 ASSERT3P(zilog->zl_dmu_pool->dp_zil_clean_taskq, !=, NULL); 2623 taskqid_t id = taskq_dispatch(zilog->zl_dmu_pool->dp_zil_clean_taskq, 2624 zil_itxg_clean, clean_me, TQ_NOSLEEP); 2625 if (id == TASKQID_INVALID) 2626 zil_itxg_clean(clean_me); 2627 } 2628 2629 /* 2630 * This function will traverse the queue of itxs that need to be 2631 * committed, and move them onto the ZIL's zl_itx_commit_list. 2632 */ 2633 static uint64_t 2634 zil_get_commit_list(zilog_t *zilog) 2635 { 2636 uint64_t otxg, txg, wtxg = 0; 2637 list_t *commit_list = &zilog->zl_itx_commit_list; 2638 2639 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock)); 2640 2641 if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */ 2642 otxg = ZILTEST_TXG; 2643 else 2644 otxg = spa_last_synced_txg(zilog->zl_spa) + 1; 2645 2646 /* 2647 * This is inherently racy, since there is nothing to prevent 2648 * the last synced txg from changing. That's okay since we'll 2649 * only commit things in the future. 2650 */ 2651 for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) { 2652 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK]; 2653 2654 mutex_enter(&itxg->itxg_lock); 2655 if (itxg->itxg_txg != txg) { 2656 mutex_exit(&itxg->itxg_lock); 2657 continue; 2658 } 2659 2660 /* 2661 * If we're adding itx records to the zl_itx_commit_list, 2662 * then the zil better be dirty in this "txg". We can assert 2663 * that here since we're holding the itxg_lock which will 2664 * prevent spa_sync from cleaning it. Once we add the itxs 2665 * to the zl_itx_commit_list we must commit it to disk even 2666 * if it's unnecessary (i.e. the txg was synced). 2667 */ 2668 ASSERT(zilog_is_dirty_in_txg(zilog, txg) || 2669 spa_freeze_txg(zilog->zl_spa) != UINT64_MAX); 2670 list_t *sync_list = &itxg->itxg_itxs->i_sync_list; 2671 itx_t *itx = NULL; 2672 if (unlikely(zilog->zl_suspend > 0)) { 2673 /* 2674 * ZIL was just suspended, but we lost the race. 2675 * Allow all earlier itxs to be committed, but ask 2676 * caller to do txg_wait_synced(txg) for any new. 2677 */ 2678 if (!list_is_empty(sync_list)) 2679 wtxg = MAX(wtxg, txg); 2680 } else { 2681 itx = list_head(sync_list); 2682 list_move_tail(commit_list, sync_list); 2683 } 2684 2685 mutex_exit(&itxg->itxg_lock); 2686 2687 while (itx != NULL) { 2688 uint64_t s = zil_itx_full_size(itx); 2689 zilog->zl_cur_size += s; 2690 zilog->zl_cur_left += s; 2691 s = zil_itx_record_size(itx); 2692 zilog->zl_cur_max = MAX(zilog->zl_cur_max, s); 2693 itx = list_next(commit_list, itx); 2694 } 2695 } 2696 return (wtxg); 2697 } 2698 2699 /* 2700 * Move the async itxs for a specified object to commit into sync lists. 2701 */ 2702 void 2703 zil_async_to_sync(zilog_t *zilog, uint64_t foid) 2704 { 2705 uint64_t otxg, txg; 2706 itx_async_node_t *ian, ian_search; 2707 avl_tree_t *t; 2708 avl_index_t where; 2709 2710 if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */ 2711 otxg = ZILTEST_TXG; 2712 else 2713 otxg = spa_last_synced_txg(zilog->zl_spa) + 1; 2714 2715 /* 2716 * This is inherently racy, since there is nothing to prevent 2717 * the last synced txg from changing. 2718 */ 2719 for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) { 2720 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK]; 2721 2722 mutex_enter(&itxg->itxg_lock); 2723 if (itxg->itxg_txg != txg) { 2724 mutex_exit(&itxg->itxg_lock); 2725 continue; 2726 } 2727 2728 /* 2729 * If a foid is specified then find that node and append its 2730 * list. Otherwise walk the tree appending all the lists 2731 * to the sync list. We add to the end rather than the 2732 * beginning to ensure the create has happened. 2733 */ 2734 t = &itxg->itxg_itxs->i_async_tree; 2735 if (foid != 0) { 2736 ian_search.ia_foid = foid; 2737 ian = avl_find(t, &ian_search, &where); 2738 if (ian != NULL) { 2739 list_move_tail(&itxg->itxg_itxs->i_sync_list, 2740 &ian->ia_list); 2741 } 2742 } else { 2743 void *cookie = NULL; 2744 2745 while ((ian = avl_destroy_nodes(t, &cookie)) != NULL) { 2746 list_move_tail(&itxg->itxg_itxs->i_sync_list, 2747 &ian->ia_list); 2748 list_destroy(&ian->ia_list); 2749 kmem_free(ian, sizeof (itx_async_node_t)); 2750 } 2751 } 2752 mutex_exit(&itxg->itxg_lock); 2753 } 2754 } 2755 2756 /* 2757 * This function will prune commit itxs that are at the head of the 2758 * commit list (it won't prune past the first non-commit itx), and 2759 * either: a) attach them to the last lwb that's still pending 2760 * completion, or b) skip them altogether. 2761 * 2762 * This is used as a performance optimization to prevent commit itxs 2763 * from generating new lwbs when it's unnecessary to do so. 2764 */ 2765 static void 2766 zil_prune_commit_list(zilog_t *zilog) 2767 { 2768 itx_t *itx; 2769 2770 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock)); 2771 2772 while ((itx = list_head(&zilog->zl_itx_commit_list)) != NULL) { 2773 lr_t *lrc = &itx->itx_lr; 2774 if (lrc->lrc_txtype != TX_COMMIT) 2775 break; 2776 2777 mutex_enter(&zilog->zl_lock); 2778 2779 lwb_t *last_lwb = zilog->zl_last_lwb_opened; 2780 if (last_lwb == NULL || 2781 last_lwb->lwb_state == LWB_STATE_FLUSH_DONE) { 2782 /* 2783 * All of the itxs this waiter was waiting on 2784 * must have already completed (or there were 2785 * never any itx's for it to wait on), so it's 2786 * safe to skip this waiter and mark it done. 2787 */ 2788 zil_commit_waiter_skip(itx->itx_private); 2789 } else { 2790 zil_commit_waiter_link_lwb(itx->itx_private, last_lwb); 2791 } 2792 2793 mutex_exit(&zilog->zl_lock); 2794 2795 list_remove(&zilog->zl_itx_commit_list, itx); 2796 zil_itx_destroy(itx); 2797 } 2798 2799 IMPLY(itx != NULL, itx->itx_lr.lrc_txtype != TX_COMMIT); 2800 } 2801 2802 static void 2803 zil_commit_writer_stall(zilog_t *zilog) 2804 { 2805 /* 2806 * When zio_alloc_zil() fails to allocate the next lwb block on 2807 * disk, we must call txg_wait_synced() to ensure all of the 2808 * lwbs in the zilog's zl_lwb_list are synced and then freed (in 2809 * zil_sync()), such that any subsequent ZIL writer (i.e. a call 2810 * to zil_process_commit_list()) will have to call zil_create(), 2811 * and start a new ZIL chain. 2812 * 2813 * Since zil_alloc_zil() failed, the lwb that was previously 2814 * issued does not have a pointer to the "next" lwb on disk. 2815 * Thus, if another ZIL writer thread was to allocate the "next" 2816 * on-disk lwb, that block could be leaked in the event of a 2817 * crash (because the previous lwb on-disk would not point to 2818 * it). 2819 * 2820 * We must hold the zilog's zl_issuer_lock while we do this, to 2821 * ensure no new threads enter zil_process_commit_list() until 2822 * all lwb's in the zl_lwb_list have been synced and freed 2823 * (which is achieved via the txg_wait_synced() call). 2824 */ 2825 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock)); 2826 txg_wait_synced(zilog->zl_dmu_pool, 0); 2827 ASSERT(list_is_empty(&zilog->zl_lwb_list)); 2828 } 2829 2830 static void 2831 zil_burst_done(zilog_t *zilog) 2832 { 2833 if (!list_is_empty(&zilog->zl_itx_commit_list) || 2834 zilog->zl_cur_size == 0) 2835 return; 2836 2837 if (zilog->zl_parallel) 2838 zilog->zl_parallel--; 2839 2840 uint_t r = (zilog->zl_prev_rotor + 1) & (ZIL_BURSTS - 1); 2841 zilog->zl_prev_rotor = r; 2842 zilog->zl_prev_opt[r] = zil_lwb_plan(zilog, zilog->zl_cur_size, 2843 &zilog->zl_prev_min[r]); 2844 2845 zilog->zl_cur_size = 0; 2846 zilog->zl_cur_max = 0; 2847 zilog->zl_cur_left = 0; 2848 } 2849 2850 /* 2851 * This function will traverse the commit list, creating new lwbs as 2852 * needed, and committing the itxs from the commit list to these newly 2853 * created lwbs. Additionally, as a new lwb is created, the previous 2854 * lwb will be issued to the zio layer to be written to disk. 2855 */ 2856 static void 2857 zil_process_commit_list(zilog_t *zilog, zil_commit_waiter_t *zcw, list_t *ilwbs) 2858 { 2859 spa_t *spa = zilog->zl_spa; 2860 list_t nolwb_itxs; 2861 list_t nolwb_waiters; 2862 lwb_t *lwb, *plwb; 2863 itx_t *itx; 2864 2865 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock)); 2866 2867 /* 2868 * Return if there's nothing to commit before we dirty the fs by 2869 * calling zil_create(). 2870 */ 2871 if (list_is_empty(&zilog->zl_itx_commit_list)) 2872 return; 2873 2874 list_create(&nolwb_itxs, sizeof (itx_t), offsetof(itx_t, itx_node)); 2875 list_create(&nolwb_waiters, sizeof (zil_commit_waiter_t), 2876 offsetof(zil_commit_waiter_t, zcw_node)); 2877 2878 lwb = list_tail(&zilog->zl_lwb_list); 2879 if (lwb == NULL) { 2880 lwb = zil_create(zilog); 2881 } else { 2882 /* 2883 * Activate SPA_FEATURE_ZILSAXATTR for the cases where ZIL will 2884 * have already been created (zl_lwb_list not empty). 2885 */ 2886 zil_commit_activate_saxattr_feature(zilog); 2887 ASSERT(lwb->lwb_state == LWB_STATE_NEW || 2888 lwb->lwb_state == LWB_STATE_OPENED); 2889 2890 /* 2891 * If the lwb is still opened, it means the workload is really 2892 * multi-threaded and we won the chance of write aggregation. 2893 * If it is not opened yet, but previous lwb is still not 2894 * flushed, it still means the workload is multi-threaded, but 2895 * there was too much time between the commits to aggregate, so 2896 * we try aggregation next times, but without too much hopes. 2897 */ 2898 if (lwb->lwb_state == LWB_STATE_OPENED) { 2899 zilog->zl_parallel = ZIL_BURSTS; 2900 } else if ((plwb = list_prev(&zilog->zl_lwb_list, lwb)) 2901 != NULL && plwb->lwb_state != LWB_STATE_FLUSH_DONE) { 2902 zilog->zl_parallel = MAX(zilog->zl_parallel, 2903 ZIL_BURSTS / 2); 2904 } 2905 } 2906 2907 while ((itx = list_remove_head(&zilog->zl_itx_commit_list)) != NULL) { 2908 lr_t *lrc = &itx->itx_lr; 2909 uint64_t txg = lrc->lrc_txg; 2910 2911 ASSERT3U(txg, !=, 0); 2912 2913 if (lrc->lrc_txtype == TX_COMMIT) { 2914 DTRACE_PROBE2(zil__process__commit__itx, 2915 zilog_t *, zilog, itx_t *, itx); 2916 } else { 2917 DTRACE_PROBE2(zil__process__normal__itx, 2918 zilog_t *, zilog, itx_t *, itx); 2919 } 2920 2921 boolean_t synced = txg <= spa_last_synced_txg(spa); 2922 boolean_t frozen = txg > spa_freeze_txg(spa); 2923 2924 /* 2925 * If the txg of this itx has already been synced out, then 2926 * we don't need to commit this itx to an lwb. This is 2927 * because the data of this itx will have already been 2928 * written to the main pool. This is inherently racy, and 2929 * it's still ok to commit an itx whose txg has already 2930 * been synced; this will result in a write that's 2931 * unnecessary, but will do no harm. 2932 * 2933 * With that said, we always want to commit TX_COMMIT itxs 2934 * to an lwb, regardless of whether or not that itx's txg 2935 * has been synced out. We do this to ensure any OPENED lwb 2936 * will always have at least one zil_commit_waiter_t linked 2937 * to the lwb. 2938 * 2939 * As a counter-example, if we skipped TX_COMMIT itx's 2940 * whose txg had already been synced, the following 2941 * situation could occur if we happened to be racing with 2942 * spa_sync: 2943 * 2944 * 1. We commit a non-TX_COMMIT itx to an lwb, where the 2945 * itx's txg is 10 and the last synced txg is 9. 2946 * 2. spa_sync finishes syncing out txg 10. 2947 * 3. We move to the next itx in the list, it's a TX_COMMIT 2948 * whose txg is 10, so we skip it rather than committing 2949 * it to the lwb used in (1). 2950 * 2951 * If the itx that is skipped in (3) is the last TX_COMMIT 2952 * itx in the commit list, than it's possible for the lwb 2953 * used in (1) to remain in the OPENED state indefinitely. 2954 * 2955 * To prevent the above scenario from occurring, ensuring 2956 * that once an lwb is OPENED it will transition to ISSUED 2957 * and eventually DONE, we always commit TX_COMMIT itx's to 2958 * an lwb here, even if that itx's txg has already been 2959 * synced. 2960 * 2961 * Finally, if the pool is frozen, we _always_ commit the 2962 * itx. The point of freezing the pool is to prevent data 2963 * from being written to the main pool via spa_sync, and 2964 * instead rely solely on the ZIL to persistently store the 2965 * data; i.e. when the pool is frozen, the last synced txg 2966 * value can't be trusted. 2967 */ 2968 if (frozen || !synced || lrc->lrc_txtype == TX_COMMIT) { 2969 if (lwb != NULL) { 2970 lwb = zil_lwb_assign(zilog, lwb, itx, ilwbs); 2971 if (lwb == NULL) { 2972 list_insert_tail(&nolwb_itxs, itx); 2973 } else if ((zcw->zcw_lwb != NULL && 2974 zcw->zcw_lwb != lwb) || zcw->zcw_done) { 2975 /* 2976 * Our lwb is done, leave the rest of 2977 * itx list to somebody else who care. 2978 */ 2979 zilog->zl_parallel = ZIL_BURSTS; 2980 zilog->zl_cur_left -= 2981 zil_itx_full_size(itx); 2982 break; 2983 } 2984 } else { 2985 if (lrc->lrc_txtype == TX_COMMIT) { 2986 zil_commit_waiter_link_nolwb( 2987 itx->itx_private, &nolwb_waiters); 2988 } 2989 list_insert_tail(&nolwb_itxs, itx); 2990 } 2991 zilog->zl_cur_left -= zil_itx_full_size(itx); 2992 } else { 2993 ASSERT3S(lrc->lrc_txtype, !=, TX_COMMIT); 2994 zilog->zl_cur_left -= zil_itx_full_size(itx); 2995 zil_itx_destroy(itx); 2996 } 2997 } 2998 2999 if (lwb == NULL) { 3000 /* 3001 * This indicates zio_alloc_zil() failed to allocate the 3002 * "next" lwb on-disk. When this happens, we must stall 3003 * the ZIL write pipeline; see the comment within 3004 * zil_commit_writer_stall() for more details. 3005 */ 3006 while ((lwb = list_remove_head(ilwbs)) != NULL) 3007 zil_lwb_write_issue(zilog, lwb); 3008 zil_commit_writer_stall(zilog); 3009 3010 /* 3011 * Additionally, we have to signal and mark the "nolwb" 3012 * waiters as "done" here, since without an lwb, we 3013 * can't do this via zil_lwb_flush_vdevs_done() like 3014 * normal. 3015 */ 3016 zil_commit_waiter_t *zcw; 3017 while ((zcw = list_remove_head(&nolwb_waiters)) != NULL) 3018 zil_commit_waiter_skip(zcw); 3019 3020 /* 3021 * And finally, we have to destroy the itx's that 3022 * couldn't be committed to an lwb; this will also call 3023 * the itx's callback if one exists for the itx. 3024 */ 3025 while ((itx = list_remove_head(&nolwb_itxs)) != NULL) 3026 zil_itx_destroy(itx); 3027 } else { 3028 ASSERT(list_is_empty(&nolwb_waiters)); 3029 ASSERT3P(lwb, !=, NULL); 3030 ASSERT(lwb->lwb_state == LWB_STATE_NEW || 3031 lwb->lwb_state == LWB_STATE_OPENED); 3032 3033 /* 3034 * At this point, the ZIL block pointed at by the "lwb" 3035 * variable is in "new" or "opened" state. 3036 * 3037 * If it's "new", then no itxs have been committed to it, so 3038 * there's no point in issuing its zio (i.e. it's "empty"). 3039 * 3040 * If it's "opened", then it contains one or more itxs that 3041 * eventually need to be committed to stable storage. In 3042 * this case we intentionally do not issue the lwb's zio 3043 * to disk yet, and instead rely on one of the following 3044 * two mechanisms for issuing the zio: 3045 * 3046 * 1. Ideally, there will be more ZIL activity occurring on 3047 * the system, such that this function will be immediately 3048 * called again by different thread and this lwb will be 3049 * closed by zil_lwb_assign(). This way, the lwb will be 3050 * "full" when it is issued to disk, and we'll make use of 3051 * the lwb's size the best we can. 3052 * 3053 * 2. If there isn't sufficient ZIL activity occurring on 3054 * the system, zil_commit_waiter() will close it and issue 3055 * the zio. If this occurs, the lwb is not guaranteed 3056 * to be "full" by the time its zio is issued, and means 3057 * the size of the lwb was "too large" given the amount 3058 * of ZIL activity occurring on the system at that time. 3059 * 3060 * We do this for a couple of reasons: 3061 * 3062 * 1. To try and reduce the number of IOPs needed to 3063 * write the same number of itxs. If an lwb has space 3064 * available in its buffer for more itxs, and more itxs 3065 * will be committed relatively soon (relative to the 3066 * latency of performing a write), then it's beneficial 3067 * to wait for these "next" itxs. This way, more itxs 3068 * can be committed to stable storage with fewer writes. 3069 * 3070 * 2. To try and use the largest lwb block size that the 3071 * incoming rate of itxs can support. Again, this is to 3072 * try and pack as many itxs into as few lwbs as 3073 * possible, without significantly impacting the latency 3074 * of each individual itx. 3075 */ 3076 if (lwb->lwb_state == LWB_STATE_OPENED && !zilog->zl_parallel) { 3077 zil_burst_done(zilog); 3078 list_insert_tail(ilwbs, lwb); 3079 lwb = zil_lwb_write_close(zilog, lwb, LWB_STATE_NEW); 3080 if (lwb == NULL) { 3081 while ((lwb = list_remove_head(ilwbs)) != NULL) 3082 zil_lwb_write_issue(zilog, lwb); 3083 zil_commit_writer_stall(zilog); 3084 } 3085 } 3086 } 3087 } 3088 3089 /* 3090 * This function is responsible for ensuring the passed in commit waiter 3091 * (and associated commit itx) is committed to an lwb. If the waiter is 3092 * not already committed to an lwb, all itxs in the zilog's queue of 3093 * itxs will be processed. The assumption is the passed in waiter's 3094 * commit itx will found in the queue just like the other non-commit 3095 * itxs, such that when the entire queue is processed, the waiter will 3096 * have been committed to an lwb. 3097 * 3098 * The lwb associated with the passed in waiter is not guaranteed to 3099 * have been issued by the time this function completes. If the lwb is 3100 * not issued, we rely on future calls to zil_commit_writer() to issue 3101 * the lwb, or the timeout mechanism found in zil_commit_waiter(). 3102 */ 3103 static uint64_t 3104 zil_commit_writer(zilog_t *zilog, zil_commit_waiter_t *zcw) 3105 { 3106 list_t ilwbs; 3107 lwb_t *lwb; 3108 uint64_t wtxg = 0; 3109 3110 ASSERT(!MUTEX_HELD(&zilog->zl_lock)); 3111 ASSERT(spa_writeable(zilog->zl_spa)); 3112 3113 list_create(&ilwbs, sizeof (lwb_t), offsetof(lwb_t, lwb_issue_node)); 3114 mutex_enter(&zilog->zl_issuer_lock); 3115 3116 if (zcw->zcw_lwb != NULL || zcw->zcw_done) { 3117 /* 3118 * It's possible that, while we were waiting to acquire 3119 * the "zl_issuer_lock", another thread committed this 3120 * waiter to an lwb. If that occurs, we bail out early, 3121 * without processing any of the zilog's queue of itxs. 3122 * 3123 * On certain workloads and system configurations, the 3124 * "zl_issuer_lock" can become highly contended. In an 3125 * attempt to reduce this contention, we immediately drop 3126 * the lock if the waiter has already been processed. 3127 * 3128 * We've measured this optimization to reduce CPU spent 3129 * contending on this lock by up to 5%, using a system 3130 * with 32 CPUs, low latency storage (~50 usec writes), 3131 * and 1024 threads performing sync writes. 3132 */ 3133 goto out; 3134 } 3135 3136 ZIL_STAT_BUMP(zilog, zil_commit_writer_count); 3137 3138 wtxg = zil_get_commit_list(zilog); 3139 zil_prune_commit_list(zilog); 3140 zil_process_commit_list(zilog, zcw, &ilwbs); 3141 3142 out: 3143 mutex_exit(&zilog->zl_issuer_lock); 3144 while ((lwb = list_remove_head(&ilwbs)) != NULL) 3145 zil_lwb_write_issue(zilog, lwb); 3146 list_destroy(&ilwbs); 3147 return (wtxg); 3148 } 3149 3150 static void 3151 zil_commit_waiter_timeout(zilog_t *zilog, zil_commit_waiter_t *zcw) 3152 { 3153 ASSERT(!MUTEX_HELD(&zilog->zl_issuer_lock)); 3154 ASSERT(MUTEX_HELD(&zcw->zcw_lock)); 3155 ASSERT3B(zcw->zcw_done, ==, B_FALSE); 3156 3157 lwb_t *lwb = zcw->zcw_lwb; 3158 ASSERT3P(lwb, !=, NULL); 3159 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_NEW); 3160 3161 /* 3162 * If the lwb has already been issued by another thread, we can 3163 * immediately return since there's no work to be done (the 3164 * point of this function is to issue the lwb). Additionally, we 3165 * do this prior to acquiring the zl_issuer_lock, to avoid 3166 * acquiring it when it's not necessary to do so. 3167 */ 3168 if (lwb->lwb_state != LWB_STATE_OPENED) 3169 return; 3170 3171 /* 3172 * In order to call zil_lwb_write_close() we must hold the 3173 * zilog's "zl_issuer_lock". We can't simply acquire that lock, 3174 * since we're already holding the commit waiter's "zcw_lock", 3175 * and those two locks are acquired in the opposite order 3176 * elsewhere. 3177 */ 3178 mutex_exit(&zcw->zcw_lock); 3179 mutex_enter(&zilog->zl_issuer_lock); 3180 mutex_enter(&zcw->zcw_lock); 3181 3182 /* 3183 * Since we just dropped and re-acquired the commit waiter's 3184 * lock, we have to re-check to see if the waiter was marked 3185 * "done" during that process. If the waiter was marked "done", 3186 * the "lwb" pointer is no longer valid (it can be free'd after 3187 * the waiter is marked "done"), so without this check we could 3188 * wind up with a use-after-free error below. 3189 */ 3190 if (zcw->zcw_done) { 3191 mutex_exit(&zilog->zl_issuer_lock); 3192 return; 3193 } 3194 3195 ASSERT3P(lwb, ==, zcw->zcw_lwb); 3196 3197 /* 3198 * We've already checked this above, but since we hadn't acquired 3199 * the zilog's zl_issuer_lock, we have to perform this check a 3200 * second time while holding the lock. 3201 * 3202 * We don't need to hold the zl_lock since the lwb cannot transition 3203 * from OPENED to CLOSED while we hold the zl_issuer_lock. The lwb 3204 * _can_ transition from CLOSED to DONE, but it's OK to race with 3205 * that transition since we treat the lwb the same, whether it's in 3206 * the CLOSED, ISSUED or DONE states. 3207 * 3208 * The important thing, is we treat the lwb differently depending on 3209 * if it's OPENED or CLOSED, and block any other threads that might 3210 * attempt to close/issue this lwb. For that reason we hold the 3211 * zl_issuer_lock when checking the lwb_state; we must not call 3212 * zil_lwb_write_close() if the lwb had already been closed/issued. 3213 * 3214 * See the comment above the lwb_state_t structure definition for 3215 * more details on the lwb states, and locking requirements. 3216 */ 3217 if (lwb->lwb_state != LWB_STATE_OPENED) { 3218 mutex_exit(&zilog->zl_issuer_lock); 3219 return; 3220 } 3221 3222 /* 3223 * We do not need zcw_lock once we hold zl_issuer_lock and know lwb 3224 * is still open. But we have to drop it to avoid a deadlock in case 3225 * callback of zio issued by zil_lwb_write_issue() try to get it, 3226 * while zil_lwb_write_issue() is blocked on attempt to issue next 3227 * lwb it found in LWB_STATE_READY state. 3228 */ 3229 mutex_exit(&zcw->zcw_lock); 3230 3231 /* 3232 * As described in the comments above zil_commit_waiter() and 3233 * zil_process_commit_list(), we need to issue this lwb's zio 3234 * since we've reached the commit waiter's timeout and it still 3235 * hasn't been issued. 3236 */ 3237 zil_burst_done(zilog); 3238 lwb_t *nlwb = zil_lwb_write_close(zilog, lwb, LWB_STATE_NEW); 3239 3240 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_CLOSED); 3241 3242 if (nlwb == NULL) { 3243 /* 3244 * When zil_lwb_write_close() returns NULL, this 3245 * indicates zio_alloc_zil() failed to allocate the 3246 * "next" lwb on-disk. When this occurs, the ZIL write 3247 * pipeline must be stalled; see the comment within the 3248 * zil_commit_writer_stall() function for more details. 3249 */ 3250 zil_lwb_write_issue(zilog, lwb); 3251 zil_commit_writer_stall(zilog); 3252 mutex_exit(&zilog->zl_issuer_lock); 3253 } else { 3254 mutex_exit(&zilog->zl_issuer_lock); 3255 zil_lwb_write_issue(zilog, lwb); 3256 } 3257 mutex_enter(&zcw->zcw_lock); 3258 } 3259 3260 /* 3261 * This function is responsible for performing the following two tasks: 3262 * 3263 * 1. its primary responsibility is to block until the given "commit 3264 * waiter" is considered "done". 3265 * 3266 * 2. its secondary responsibility is to issue the zio for the lwb that 3267 * the given "commit waiter" is waiting on, if this function has 3268 * waited "long enough" and the lwb is still in the "open" state. 3269 * 3270 * Given a sufficient amount of itxs being generated and written using 3271 * the ZIL, the lwb's zio will be issued via the zil_lwb_assign() 3272 * function. If this does not occur, this secondary responsibility will 3273 * ensure the lwb is issued even if there is not other synchronous 3274 * activity on the system. 3275 * 3276 * For more details, see zil_process_commit_list(); more specifically, 3277 * the comment at the bottom of that function. 3278 */ 3279 static void 3280 zil_commit_waiter(zilog_t *zilog, zil_commit_waiter_t *zcw) 3281 { 3282 ASSERT(!MUTEX_HELD(&zilog->zl_lock)); 3283 ASSERT(!MUTEX_HELD(&zilog->zl_issuer_lock)); 3284 ASSERT(spa_writeable(zilog->zl_spa)); 3285 3286 mutex_enter(&zcw->zcw_lock); 3287 3288 /* 3289 * The timeout is scaled based on the lwb latency to avoid 3290 * significantly impacting the latency of each individual itx. 3291 * For more details, see the comment at the bottom of the 3292 * zil_process_commit_list() function. 3293 */ 3294 int pct = MAX(zfs_commit_timeout_pct, 1); 3295 hrtime_t sleep = (zilog->zl_last_lwb_latency * pct) / 100; 3296 hrtime_t wakeup = gethrtime() + sleep; 3297 boolean_t timedout = B_FALSE; 3298 3299 while (!zcw->zcw_done) { 3300 ASSERT(MUTEX_HELD(&zcw->zcw_lock)); 3301 3302 lwb_t *lwb = zcw->zcw_lwb; 3303 3304 /* 3305 * Usually, the waiter will have a non-NULL lwb field here, 3306 * but it's possible for it to be NULL as a result of 3307 * zil_commit() racing with spa_sync(). 3308 * 3309 * When zil_clean() is called, it's possible for the itxg 3310 * list (which may be cleaned via a taskq) to contain 3311 * commit itxs. When this occurs, the commit waiters linked 3312 * off of these commit itxs will not be committed to an 3313 * lwb. Additionally, these commit waiters will not be 3314 * marked done until zil_commit_waiter_skip() is called via 3315 * zil_itxg_clean(). 3316 * 3317 * Thus, it's possible for this commit waiter (i.e. the 3318 * "zcw" variable) to be found in this "in between" state; 3319 * where it's "zcw_lwb" field is NULL, and it hasn't yet 3320 * been skipped, so it's "zcw_done" field is still B_FALSE. 3321 */ 3322 IMPLY(lwb != NULL, lwb->lwb_state != LWB_STATE_NEW); 3323 3324 if (lwb != NULL && lwb->lwb_state == LWB_STATE_OPENED) { 3325 ASSERT3B(timedout, ==, B_FALSE); 3326 3327 /* 3328 * If the lwb hasn't been issued yet, then we 3329 * need to wait with a timeout, in case this 3330 * function needs to issue the lwb after the 3331 * timeout is reached; responsibility (2) from 3332 * the comment above this function. 3333 */ 3334 int rc = cv_timedwait_hires(&zcw->zcw_cv, 3335 &zcw->zcw_lock, wakeup, USEC2NSEC(1), 3336 CALLOUT_FLAG_ABSOLUTE); 3337 3338 if (rc != -1 || zcw->zcw_done) 3339 continue; 3340 3341 timedout = B_TRUE; 3342 zil_commit_waiter_timeout(zilog, zcw); 3343 3344 if (!zcw->zcw_done) { 3345 /* 3346 * If the commit waiter has already been 3347 * marked "done", it's possible for the 3348 * waiter's lwb structure to have already 3349 * been freed. Thus, we can only reliably 3350 * make these assertions if the waiter 3351 * isn't done. 3352 */ 3353 ASSERT3P(lwb, ==, zcw->zcw_lwb); 3354 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_OPENED); 3355 } 3356 } else { 3357 /* 3358 * If the lwb isn't open, then it must have already 3359 * been issued. In that case, there's no need to 3360 * use a timeout when waiting for the lwb to 3361 * complete. 3362 * 3363 * Additionally, if the lwb is NULL, the waiter 3364 * will soon be signaled and marked done via 3365 * zil_clean() and zil_itxg_clean(), so no timeout 3366 * is required. 3367 */ 3368 3369 IMPLY(lwb != NULL, 3370 lwb->lwb_state == LWB_STATE_CLOSED || 3371 lwb->lwb_state == LWB_STATE_READY || 3372 lwb->lwb_state == LWB_STATE_ISSUED || 3373 lwb->lwb_state == LWB_STATE_WRITE_DONE || 3374 lwb->lwb_state == LWB_STATE_FLUSH_DONE); 3375 cv_wait(&zcw->zcw_cv, &zcw->zcw_lock); 3376 } 3377 } 3378 3379 mutex_exit(&zcw->zcw_lock); 3380 } 3381 3382 static zil_commit_waiter_t * 3383 zil_alloc_commit_waiter(void) 3384 { 3385 zil_commit_waiter_t *zcw = kmem_cache_alloc(zil_zcw_cache, KM_SLEEP); 3386 3387 cv_init(&zcw->zcw_cv, NULL, CV_DEFAULT, NULL); 3388 mutex_init(&zcw->zcw_lock, NULL, MUTEX_DEFAULT, NULL); 3389 list_link_init(&zcw->zcw_node); 3390 zcw->zcw_lwb = NULL; 3391 zcw->zcw_done = B_FALSE; 3392 zcw->zcw_zio_error = 0; 3393 3394 return (zcw); 3395 } 3396 3397 static void 3398 zil_free_commit_waiter(zil_commit_waiter_t *zcw) 3399 { 3400 ASSERT(!list_link_active(&zcw->zcw_node)); 3401 ASSERT3P(zcw->zcw_lwb, ==, NULL); 3402 ASSERT3B(zcw->zcw_done, ==, B_TRUE); 3403 mutex_destroy(&zcw->zcw_lock); 3404 cv_destroy(&zcw->zcw_cv); 3405 kmem_cache_free(zil_zcw_cache, zcw); 3406 } 3407 3408 /* 3409 * This function is used to create a TX_COMMIT itx and assign it. This 3410 * way, it will be linked into the ZIL's list of synchronous itxs, and 3411 * then later committed to an lwb (or skipped) when 3412 * zil_process_commit_list() is called. 3413 */ 3414 static void 3415 zil_commit_itx_assign(zilog_t *zilog, zil_commit_waiter_t *zcw) 3416 { 3417 dmu_tx_t *tx = dmu_tx_create(zilog->zl_os); 3418 3419 /* 3420 * Since we are not going to create any new dirty data, and we 3421 * can even help with clearing the existing dirty data, we 3422 * should not be subject to the dirty data based delays. We 3423 * use TXG_NOTHROTTLE to bypass the delay mechanism. 3424 */ 3425 VERIFY0(dmu_tx_assign(tx, TXG_WAIT | TXG_NOTHROTTLE)); 3426 3427 itx_t *itx = zil_itx_create(TX_COMMIT, sizeof (lr_t)); 3428 itx->itx_sync = B_TRUE; 3429 itx->itx_private = zcw; 3430 3431 zil_itx_assign(zilog, itx, tx); 3432 3433 dmu_tx_commit(tx); 3434 } 3435 3436 /* 3437 * Commit ZFS Intent Log transactions (itxs) to stable storage. 3438 * 3439 * When writing ZIL transactions to the on-disk representation of the 3440 * ZIL, the itxs are committed to a Log Write Block (lwb). Multiple 3441 * itxs can be committed to a single lwb. Once a lwb is written and 3442 * committed to stable storage (i.e. the lwb is written, and vdevs have 3443 * been flushed), each itx that was committed to that lwb is also 3444 * considered to be committed to stable storage. 3445 * 3446 * When an itx is committed to an lwb, the log record (lr_t) contained 3447 * by the itx is copied into the lwb's zio buffer, and once this buffer 3448 * is written to disk, it becomes an on-disk ZIL block. 3449 * 3450 * As itxs are generated, they're inserted into the ZIL's queue of 3451 * uncommitted itxs. The semantics of zil_commit() are such that it will 3452 * block until all itxs that were in the queue when it was called, are 3453 * committed to stable storage. 3454 * 3455 * If "foid" is zero, this means all "synchronous" and "asynchronous" 3456 * itxs, for all objects in the dataset, will be committed to stable 3457 * storage prior to zil_commit() returning. If "foid" is non-zero, all 3458 * "synchronous" itxs for all objects, but only "asynchronous" itxs 3459 * that correspond to the foid passed in, will be committed to stable 3460 * storage prior to zil_commit() returning. 3461 * 3462 * Generally speaking, when zil_commit() is called, the consumer doesn't 3463 * actually care about _all_ of the uncommitted itxs. Instead, they're 3464 * simply trying to waiting for a specific itx to be committed to disk, 3465 * but the interface(s) for interacting with the ZIL don't allow such 3466 * fine-grained communication. A better interface would allow a consumer 3467 * to create and assign an itx, and then pass a reference to this itx to 3468 * zil_commit(); such that zil_commit() would return as soon as that 3469 * specific itx was committed to disk (instead of waiting for _all_ 3470 * itxs to be committed). 3471 * 3472 * When a thread calls zil_commit() a special "commit itx" will be 3473 * generated, along with a corresponding "waiter" for this commit itx. 3474 * zil_commit() will wait on this waiter's CV, such that when the waiter 3475 * is marked done, and signaled, zil_commit() will return. 3476 * 3477 * This commit itx is inserted into the queue of uncommitted itxs. This 3478 * provides an easy mechanism for determining which itxs were in the 3479 * queue prior to zil_commit() having been called, and which itxs were 3480 * added after zil_commit() was called. 3481 * 3482 * The commit itx is special; it doesn't have any on-disk representation. 3483 * When a commit itx is "committed" to an lwb, the waiter associated 3484 * with it is linked onto the lwb's list of waiters. Then, when that lwb 3485 * completes, each waiter on the lwb's list is marked done and signaled 3486 * -- allowing the thread waiting on the waiter to return from zil_commit(). 3487 * 3488 * It's important to point out a few critical factors that allow us 3489 * to make use of the commit itxs, commit waiters, per-lwb lists of 3490 * commit waiters, and zio completion callbacks like we're doing: 3491 * 3492 * 1. The list of waiters for each lwb is traversed, and each commit 3493 * waiter is marked "done" and signaled, in the zio completion 3494 * callback of the lwb's zio[*]. 3495 * 3496 * * Actually, the waiters are signaled in the zio completion 3497 * callback of the root zio for the flush commands that are sent to 3498 * the vdevs upon completion of the lwb zio. 3499 * 3500 * 2. When the itxs are inserted into the ZIL's queue of uncommitted 3501 * itxs, the order in which they are inserted is preserved[*]; as 3502 * itxs are added to the queue, they are added to the tail of 3503 * in-memory linked lists. 3504 * 3505 * When committing the itxs to lwbs (to be written to disk), they 3506 * are committed in the same order in which the itxs were added to 3507 * the uncommitted queue's linked list(s); i.e. the linked list of 3508 * itxs to commit is traversed from head to tail, and each itx is 3509 * committed to an lwb in that order. 3510 * 3511 * * To clarify: 3512 * 3513 * - the order of "sync" itxs is preserved w.r.t. other 3514 * "sync" itxs, regardless of the corresponding objects. 3515 * - the order of "async" itxs is preserved w.r.t. other 3516 * "async" itxs corresponding to the same object. 3517 * - the order of "async" itxs is *not* preserved w.r.t. other 3518 * "async" itxs corresponding to different objects. 3519 * - the order of "sync" itxs w.r.t. "async" itxs (or vice 3520 * versa) is *not* preserved, even for itxs that correspond 3521 * to the same object. 3522 * 3523 * For more details, see: zil_itx_assign(), zil_async_to_sync(), 3524 * zil_get_commit_list(), and zil_process_commit_list(). 3525 * 3526 * 3. The lwbs represent a linked list of blocks on disk. Thus, any 3527 * lwb cannot be considered committed to stable storage, until its 3528 * "previous" lwb is also committed to stable storage. This fact, 3529 * coupled with the fact described above, means that itxs are 3530 * committed in (roughly) the order in which they were generated. 3531 * This is essential because itxs are dependent on prior itxs. 3532 * Thus, we *must not* deem an itx as being committed to stable 3533 * storage, until *all* prior itxs have also been committed to 3534 * stable storage. 3535 * 3536 * To enforce this ordering of lwb zio's, while still leveraging as 3537 * much of the underlying storage performance as possible, we rely 3538 * on two fundamental concepts: 3539 * 3540 * 1. The creation and issuance of lwb zio's is protected by 3541 * the zilog's "zl_issuer_lock", which ensures only a single 3542 * thread is creating and/or issuing lwb's at a time 3543 * 2. The "previous" lwb is a child of the "current" lwb 3544 * (leveraging the zio parent-child dependency graph) 3545 * 3546 * By relying on this parent-child zio relationship, we can have 3547 * many lwb zio's concurrently issued to the underlying storage, 3548 * but the order in which they complete will be the same order in 3549 * which they were created. 3550 */ 3551 void 3552 zil_commit(zilog_t *zilog, uint64_t foid) 3553 { 3554 /* 3555 * We should never attempt to call zil_commit on a snapshot for 3556 * a couple of reasons: 3557 * 3558 * 1. A snapshot may never be modified, thus it cannot have any 3559 * in-flight itxs that would have modified the dataset. 3560 * 3561 * 2. By design, when zil_commit() is called, a commit itx will 3562 * be assigned to this zilog; as a result, the zilog will be 3563 * dirtied. We must not dirty the zilog of a snapshot; there's 3564 * checks in the code that enforce this invariant, and will 3565 * cause a panic if it's not upheld. 3566 */ 3567 ASSERT3B(dmu_objset_is_snapshot(zilog->zl_os), ==, B_FALSE); 3568 3569 if (zilog->zl_sync == ZFS_SYNC_DISABLED) 3570 return; 3571 3572 if (!spa_writeable(zilog->zl_spa)) { 3573 /* 3574 * If the SPA is not writable, there should never be any 3575 * pending itxs waiting to be committed to disk. If that 3576 * weren't true, we'd skip writing those itxs out, and 3577 * would break the semantics of zil_commit(); thus, we're 3578 * verifying that truth before we return to the caller. 3579 */ 3580 ASSERT(list_is_empty(&zilog->zl_lwb_list)); 3581 ASSERT3P(zilog->zl_last_lwb_opened, ==, NULL); 3582 for (int i = 0; i < TXG_SIZE; i++) 3583 ASSERT3P(zilog->zl_itxg[i].itxg_itxs, ==, NULL); 3584 return; 3585 } 3586 3587 /* 3588 * If the ZIL is suspended, we don't want to dirty it by calling 3589 * zil_commit_itx_assign() below, nor can we write out 3590 * lwbs like would be done in zil_commit_write(). Thus, we 3591 * simply rely on txg_wait_synced() to maintain the necessary 3592 * semantics, and avoid calling those functions altogether. 3593 */ 3594 if (zilog->zl_suspend > 0) { 3595 txg_wait_synced(zilog->zl_dmu_pool, 0); 3596 return; 3597 } 3598 3599 zil_commit_impl(zilog, foid); 3600 } 3601 3602 void 3603 zil_commit_impl(zilog_t *zilog, uint64_t foid) 3604 { 3605 ZIL_STAT_BUMP(zilog, zil_commit_count); 3606 3607 /* 3608 * Move the "async" itxs for the specified foid to the "sync" 3609 * queues, such that they will be later committed (or skipped) 3610 * to an lwb when zil_process_commit_list() is called. 3611 * 3612 * Since these "async" itxs must be committed prior to this 3613 * call to zil_commit returning, we must perform this operation 3614 * before we call zil_commit_itx_assign(). 3615 */ 3616 zil_async_to_sync(zilog, foid); 3617 3618 /* 3619 * We allocate a new "waiter" structure which will initially be 3620 * linked to the commit itx using the itx's "itx_private" field. 3621 * Since the commit itx doesn't represent any on-disk state, 3622 * when it's committed to an lwb, rather than copying the its 3623 * lr_t into the lwb's buffer, the commit itx's "waiter" will be 3624 * added to the lwb's list of waiters. Then, when the lwb is 3625 * committed to stable storage, each waiter in the lwb's list of 3626 * waiters will be marked "done", and signalled. 3627 * 3628 * We must create the waiter and assign the commit itx prior to 3629 * calling zil_commit_writer(), or else our specific commit itx 3630 * is not guaranteed to be committed to an lwb prior to calling 3631 * zil_commit_waiter(). 3632 */ 3633 zil_commit_waiter_t *zcw = zil_alloc_commit_waiter(); 3634 zil_commit_itx_assign(zilog, zcw); 3635 3636 uint64_t wtxg = zil_commit_writer(zilog, zcw); 3637 zil_commit_waiter(zilog, zcw); 3638 3639 if (zcw->zcw_zio_error != 0) { 3640 /* 3641 * If there was an error writing out the ZIL blocks that 3642 * this thread is waiting on, then we fallback to 3643 * relying on spa_sync() to write out the data this 3644 * thread is waiting on. Obviously this has performance 3645 * implications, but the expectation is for this to be 3646 * an exceptional case, and shouldn't occur often. 3647 */ 3648 DTRACE_PROBE2(zil__commit__io__error, 3649 zilog_t *, zilog, zil_commit_waiter_t *, zcw); 3650 txg_wait_synced(zilog->zl_dmu_pool, 0); 3651 } else if (wtxg != 0) { 3652 txg_wait_synced(zilog->zl_dmu_pool, wtxg); 3653 } 3654 3655 zil_free_commit_waiter(zcw); 3656 } 3657 3658 /* 3659 * Called in syncing context to free committed log blocks and update log header. 3660 */ 3661 void 3662 zil_sync(zilog_t *zilog, dmu_tx_t *tx) 3663 { 3664 zil_header_t *zh = zil_header_in_syncing_context(zilog); 3665 uint64_t txg = dmu_tx_get_txg(tx); 3666 spa_t *spa = zilog->zl_spa; 3667 uint64_t *replayed_seq = &zilog->zl_replayed_seq[txg & TXG_MASK]; 3668 lwb_t *lwb; 3669 3670 /* 3671 * We don't zero out zl_destroy_txg, so make sure we don't try 3672 * to destroy it twice. 3673 */ 3674 if (spa_sync_pass(spa) != 1) 3675 return; 3676 3677 zil_lwb_flush_wait_all(zilog, txg); 3678 3679 mutex_enter(&zilog->zl_lock); 3680 3681 ASSERT(zilog->zl_stop_sync == 0); 3682 3683 if (*replayed_seq != 0) { 3684 ASSERT(zh->zh_replay_seq < *replayed_seq); 3685 zh->zh_replay_seq = *replayed_seq; 3686 *replayed_seq = 0; 3687 } 3688 3689 if (zilog->zl_destroy_txg == txg) { 3690 blkptr_t blk = zh->zh_log; 3691 dsl_dataset_t *ds = dmu_objset_ds(zilog->zl_os); 3692 3693 ASSERT(list_is_empty(&zilog->zl_lwb_list)); 3694 3695 memset(zh, 0, sizeof (zil_header_t)); 3696 memset(zilog->zl_replayed_seq, 0, 3697 sizeof (zilog->zl_replayed_seq)); 3698 3699 if (zilog->zl_keep_first) { 3700 /* 3701 * If this block was part of log chain that couldn't 3702 * be claimed because a device was missing during 3703 * zil_claim(), but that device later returns, 3704 * then this block could erroneously appear valid. 3705 * To guard against this, assign a new GUID to the new 3706 * log chain so it doesn't matter what blk points to. 3707 */ 3708 zil_init_log_chain(zilog, &blk); 3709 zh->zh_log = blk; 3710 } else { 3711 /* 3712 * A destroyed ZIL chain can't contain any TX_SETSAXATTR 3713 * records. So, deactivate the feature for this dataset. 3714 * We activate it again when we start a new ZIL chain. 3715 */ 3716 if (dsl_dataset_feature_is_active(ds, 3717 SPA_FEATURE_ZILSAXATTR)) 3718 dsl_dataset_deactivate_feature(ds, 3719 SPA_FEATURE_ZILSAXATTR, tx); 3720 } 3721 } 3722 3723 while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) { 3724 zh->zh_log = lwb->lwb_blk; 3725 if (lwb->lwb_state != LWB_STATE_FLUSH_DONE || 3726 lwb->lwb_alloc_txg > txg || lwb->lwb_max_txg > txg) 3727 break; 3728 list_remove(&zilog->zl_lwb_list, lwb); 3729 if (!BP_IS_HOLE(&lwb->lwb_blk)) 3730 zio_free(spa, txg, &lwb->lwb_blk); 3731 zil_free_lwb(zilog, lwb); 3732 3733 /* 3734 * If we don't have anything left in the lwb list then 3735 * we've had an allocation failure and we need to zero 3736 * out the zil_header blkptr so that we don't end 3737 * up freeing the same block twice. 3738 */ 3739 if (list_is_empty(&zilog->zl_lwb_list)) 3740 BP_ZERO(&zh->zh_log); 3741 } 3742 3743 mutex_exit(&zilog->zl_lock); 3744 } 3745 3746 static int 3747 zil_lwb_cons(void *vbuf, void *unused, int kmflag) 3748 { 3749 (void) unused, (void) kmflag; 3750 lwb_t *lwb = vbuf; 3751 list_create(&lwb->lwb_itxs, sizeof (itx_t), offsetof(itx_t, itx_node)); 3752 list_create(&lwb->lwb_waiters, sizeof (zil_commit_waiter_t), 3753 offsetof(zil_commit_waiter_t, zcw_node)); 3754 avl_create(&lwb->lwb_vdev_tree, zil_lwb_vdev_compare, 3755 sizeof (zil_vdev_node_t), offsetof(zil_vdev_node_t, zv_node)); 3756 mutex_init(&lwb->lwb_vdev_lock, NULL, MUTEX_DEFAULT, NULL); 3757 return (0); 3758 } 3759 3760 static void 3761 zil_lwb_dest(void *vbuf, void *unused) 3762 { 3763 (void) unused; 3764 lwb_t *lwb = vbuf; 3765 mutex_destroy(&lwb->lwb_vdev_lock); 3766 avl_destroy(&lwb->lwb_vdev_tree); 3767 list_destroy(&lwb->lwb_waiters); 3768 list_destroy(&lwb->lwb_itxs); 3769 } 3770 3771 void 3772 zil_init(void) 3773 { 3774 zil_lwb_cache = kmem_cache_create("zil_lwb_cache", 3775 sizeof (lwb_t), 0, zil_lwb_cons, zil_lwb_dest, NULL, NULL, NULL, 0); 3776 3777 zil_zcw_cache = kmem_cache_create("zil_zcw_cache", 3778 sizeof (zil_commit_waiter_t), 0, NULL, NULL, NULL, NULL, NULL, 0); 3779 3780 zil_sums_init(&zil_sums_global); 3781 zil_kstats_global = kstat_create("zfs", 0, "zil", "misc", 3782 KSTAT_TYPE_NAMED, sizeof (zil_stats) / sizeof (kstat_named_t), 3783 KSTAT_FLAG_VIRTUAL); 3784 3785 if (zil_kstats_global != NULL) { 3786 zil_kstats_global->ks_data = &zil_stats; 3787 zil_kstats_global->ks_update = zil_kstats_global_update; 3788 zil_kstats_global->ks_private = NULL; 3789 kstat_install(zil_kstats_global); 3790 } 3791 } 3792 3793 void 3794 zil_fini(void) 3795 { 3796 kmem_cache_destroy(zil_zcw_cache); 3797 kmem_cache_destroy(zil_lwb_cache); 3798 3799 if (zil_kstats_global != NULL) { 3800 kstat_delete(zil_kstats_global); 3801 zil_kstats_global = NULL; 3802 } 3803 3804 zil_sums_fini(&zil_sums_global); 3805 } 3806 3807 void 3808 zil_set_sync(zilog_t *zilog, uint64_t sync) 3809 { 3810 zilog->zl_sync = sync; 3811 } 3812 3813 void 3814 zil_set_logbias(zilog_t *zilog, uint64_t logbias) 3815 { 3816 zilog->zl_logbias = logbias; 3817 } 3818 3819 zilog_t * 3820 zil_alloc(objset_t *os, zil_header_t *zh_phys) 3821 { 3822 zilog_t *zilog; 3823 3824 zilog = kmem_zalloc(sizeof (zilog_t), KM_SLEEP); 3825 3826 zilog->zl_header = zh_phys; 3827 zilog->zl_os = os; 3828 zilog->zl_spa = dmu_objset_spa(os); 3829 zilog->zl_dmu_pool = dmu_objset_pool(os); 3830 zilog->zl_destroy_txg = TXG_INITIAL - 1; 3831 zilog->zl_logbias = dmu_objset_logbias(os); 3832 zilog->zl_sync = dmu_objset_syncprop(os); 3833 zilog->zl_dirty_max_txg = 0; 3834 zilog->zl_last_lwb_opened = NULL; 3835 zilog->zl_last_lwb_latency = 0; 3836 zilog->zl_max_block_size = MIN(MAX(P2ALIGN_TYPED(zil_maxblocksize, 3837 ZIL_MIN_BLKSZ, uint64_t), ZIL_MIN_BLKSZ), 3838 spa_maxblocksize(dmu_objset_spa(os))); 3839 3840 mutex_init(&zilog->zl_lock, NULL, MUTEX_DEFAULT, NULL); 3841 mutex_init(&zilog->zl_issuer_lock, NULL, MUTEX_DEFAULT, NULL); 3842 mutex_init(&zilog->zl_lwb_io_lock, NULL, MUTEX_DEFAULT, NULL); 3843 3844 for (int i = 0; i < TXG_SIZE; i++) { 3845 mutex_init(&zilog->zl_itxg[i].itxg_lock, NULL, 3846 MUTEX_DEFAULT, NULL); 3847 } 3848 3849 list_create(&zilog->zl_lwb_list, sizeof (lwb_t), 3850 offsetof(lwb_t, lwb_node)); 3851 3852 list_create(&zilog->zl_itx_commit_list, sizeof (itx_t), 3853 offsetof(itx_t, itx_node)); 3854 3855 cv_init(&zilog->zl_cv_suspend, NULL, CV_DEFAULT, NULL); 3856 cv_init(&zilog->zl_lwb_io_cv, NULL, CV_DEFAULT, NULL); 3857 3858 for (int i = 0; i < ZIL_BURSTS; i++) { 3859 zilog->zl_prev_opt[i] = zilog->zl_max_block_size - 3860 sizeof (zil_chain_t); 3861 } 3862 3863 return (zilog); 3864 } 3865 3866 void 3867 zil_free(zilog_t *zilog) 3868 { 3869 int i; 3870 3871 zilog->zl_stop_sync = 1; 3872 3873 ASSERT0(zilog->zl_suspend); 3874 ASSERT0(zilog->zl_suspending); 3875 3876 ASSERT(list_is_empty(&zilog->zl_lwb_list)); 3877 list_destroy(&zilog->zl_lwb_list); 3878 3879 ASSERT(list_is_empty(&zilog->zl_itx_commit_list)); 3880 list_destroy(&zilog->zl_itx_commit_list); 3881 3882 for (i = 0; i < TXG_SIZE; i++) { 3883 /* 3884 * It's possible for an itx to be generated that doesn't dirty 3885 * a txg (e.g. ztest TX_TRUNCATE). So there's no zil_clean() 3886 * callback to remove the entry. We remove those here. 3887 * 3888 * Also free up the ziltest itxs. 3889 */ 3890 if (zilog->zl_itxg[i].itxg_itxs) 3891 zil_itxg_clean(zilog->zl_itxg[i].itxg_itxs); 3892 mutex_destroy(&zilog->zl_itxg[i].itxg_lock); 3893 } 3894 3895 mutex_destroy(&zilog->zl_issuer_lock); 3896 mutex_destroy(&zilog->zl_lock); 3897 mutex_destroy(&zilog->zl_lwb_io_lock); 3898 3899 cv_destroy(&zilog->zl_cv_suspend); 3900 cv_destroy(&zilog->zl_lwb_io_cv); 3901 3902 kmem_free(zilog, sizeof (zilog_t)); 3903 } 3904 3905 /* 3906 * Open an intent log. 3907 */ 3908 zilog_t * 3909 zil_open(objset_t *os, zil_get_data_t *get_data, zil_sums_t *zil_sums) 3910 { 3911 zilog_t *zilog = dmu_objset_zil(os); 3912 3913 ASSERT3P(zilog->zl_get_data, ==, NULL); 3914 ASSERT3P(zilog->zl_last_lwb_opened, ==, NULL); 3915 ASSERT(list_is_empty(&zilog->zl_lwb_list)); 3916 3917 zilog->zl_get_data = get_data; 3918 zilog->zl_sums = zil_sums; 3919 3920 return (zilog); 3921 } 3922 3923 /* 3924 * Close an intent log. 3925 */ 3926 void 3927 zil_close(zilog_t *zilog) 3928 { 3929 lwb_t *lwb; 3930 uint64_t txg; 3931 3932 if (!dmu_objset_is_snapshot(zilog->zl_os)) { 3933 zil_commit(zilog, 0); 3934 } else { 3935 ASSERT(list_is_empty(&zilog->zl_lwb_list)); 3936 ASSERT0(zilog->zl_dirty_max_txg); 3937 ASSERT3B(zilog_is_dirty(zilog), ==, B_FALSE); 3938 } 3939 3940 mutex_enter(&zilog->zl_lock); 3941 txg = zilog->zl_dirty_max_txg; 3942 lwb = list_tail(&zilog->zl_lwb_list); 3943 if (lwb != NULL) { 3944 txg = MAX(txg, lwb->lwb_alloc_txg); 3945 txg = MAX(txg, lwb->lwb_max_txg); 3946 } 3947 mutex_exit(&zilog->zl_lock); 3948 3949 /* 3950 * zl_lwb_max_issued_txg may be larger than lwb_max_txg. It depends 3951 * on the time when the dmu_tx transaction is assigned in 3952 * zil_lwb_write_issue(). 3953 */ 3954 mutex_enter(&zilog->zl_lwb_io_lock); 3955 txg = MAX(zilog->zl_lwb_max_issued_txg, txg); 3956 mutex_exit(&zilog->zl_lwb_io_lock); 3957 3958 /* 3959 * We need to use txg_wait_synced() to wait until that txg is synced. 3960 * zil_sync() will guarantee all lwbs up to that txg have been 3961 * written out, flushed, and cleaned. 3962 */ 3963 if (txg != 0) 3964 txg_wait_synced(zilog->zl_dmu_pool, txg); 3965 3966 if (zilog_is_dirty(zilog)) 3967 zfs_dbgmsg("zil (%px) is dirty, txg %llu", zilog, 3968 (u_longlong_t)txg); 3969 if (txg < spa_freeze_txg(zilog->zl_spa)) 3970 VERIFY(!zilog_is_dirty(zilog)); 3971 3972 zilog->zl_get_data = NULL; 3973 3974 /* 3975 * We should have only one lwb left on the list; remove it now. 3976 */ 3977 mutex_enter(&zilog->zl_lock); 3978 lwb = list_remove_head(&zilog->zl_lwb_list); 3979 if (lwb != NULL) { 3980 ASSERT(list_is_empty(&zilog->zl_lwb_list)); 3981 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_NEW); 3982 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz); 3983 zil_free_lwb(zilog, lwb); 3984 } 3985 mutex_exit(&zilog->zl_lock); 3986 } 3987 3988 static const char *suspend_tag = "zil suspending"; 3989 3990 /* 3991 * Suspend an intent log. While in suspended mode, we still honor 3992 * synchronous semantics, but we rely on txg_wait_synced() to do it. 3993 * On old version pools, we suspend the log briefly when taking a 3994 * snapshot so that it will have an empty intent log. 3995 * 3996 * Long holds are not really intended to be used the way we do here -- 3997 * held for such a short time. A concurrent caller of dsl_dataset_long_held() 3998 * could fail. Therefore we take pains to only put a long hold if it is 3999 * actually necessary. Fortunately, it will only be necessary if the 4000 * objset is currently mounted (or the ZVOL equivalent). In that case it 4001 * will already have a long hold, so we are not really making things any worse. 4002 * 4003 * Ideally, we would locate the existing long-holder (i.e. the zfsvfs_t or 4004 * zvol_state_t), and use their mechanism to prevent their hold from being 4005 * dropped (e.g. VFS_HOLD()). However, that would be even more pain for 4006 * very little gain. 4007 * 4008 * if cookiep == NULL, this does both the suspend & resume. 4009 * Otherwise, it returns with the dataset "long held", and the cookie 4010 * should be passed into zil_resume(). 4011 */ 4012 int 4013 zil_suspend(const char *osname, void **cookiep) 4014 { 4015 objset_t *os; 4016 zilog_t *zilog; 4017 const zil_header_t *zh; 4018 int error; 4019 4020 error = dmu_objset_hold(osname, suspend_tag, &os); 4021 if (error != 0) 4022 return (error); 4023 zilog = dmu_objset_zil(os); 4024 4025 mutex_enter(&zilog->zl_lock); 4026 zh = zilog->zl_header; 4027 4028 if (zh->zh_flags & ZIL_REPLAY_NEEDED) { /* unplayed log */ 4029 mutex_exit(&zilog->zl_lock); 4030 dmu_objset_rele(os, suspend_tag); 4031 return (SET_ERROR(EBUSY)); 4032 } 4033 4034 /* 4035 * Don't put a long hold in the cases where we can avoid it. This 4036 * is when there is no cookie so we are doing a suspend & resume 4037 * (i.e. called from zil_vdev_offline()), and there's nothing to do 4038 * for the suspend because it's already suspended, or there's no ZIL. 4039 */ 4040 if (cookiep == NULL && !zilog->zl_suspending && 4041 (zilog->zl_suspend > 0 || BP_IS_HOLE(&zh->zh_log))) { 4042 mutex_exit(&zilog->zl_lock); 4043 dmu_objset_rele(os, suspend_tag); 4044 return (0); 4045 } 4046 4047 dsl_dataset_long_hold(dmu_objset_ds(os), suspend_tag); 4048 dsl_pool_rele(dmu_objset_pool(os), suspend_tag); 4049 4050 zilog->zl_suspend++; 4051 4052 if (zilog->zl_suspend > 1) { 4053 /* 4054 * Someone else is already suspending it. 4055 * Just wait for them to finish. 4056 */ 4057 4058 while (zilog->zl_suspending) 4059 cv_wait(&zilog->zl_cv_suspend, &zilog->zl_lock); 4060 mutex_exit(&zilog->zl_lock); 4061 4062 if (cookiep == NULL) 4063 zil_resume(os); 4064 else 4065 *cookiep = os; 4066 return (0); 4067 } 4068 4069 /* 4070 * If there is no pointer to an on-disk block, this ZIL must not 4071 * be active (e.g. filesystem not mounted), so there's nothing 4072 * to clean up. 4073 */ 4074 if (BP_IS_HOLE(&zh->zh_log)) { 4075 ASSERT(cookiep != NULL); /* fast path already handled */ 4076 4077 *cookiep = os; 4078 mutex_exit(&zilog->zl_lock); 4079 return (0); 4080 } 4081 4082 /* 4083 * The ZIL has work to do. Ensure that the associated encryption 4084 * key will remain mapped while we are committing the log by 4085 * grabbing a reference to it. If the key isn't loaded we have no 4086 * choice but to return an error until the wrapping key is loaded. 4087 */ 4088 if (os->os_encrypted && 4089 dsl_dataset_create_key_mapping(dmu_objset_ds(os)) != 0) { 4090 zilog->zl_suspend--; 4091 mutex_exit(&zilog->zl_lock); 4092 dsl_dataset_long_rele(dmu_objset_ds(os), suspend_tag); 4093 dsl_dataset_rele(dmu_objset_ds(os), suspend_tag); 4094 return (SET_ERROR(EACCES)); 4095 } 4096 4097 zilog->zl_suspending = B_TRUE; 4098 mutex_exit(&zilog->zl_lock); 4099 4100 /* 4101 * We need to use zil_commit_impl to ensure we wait for all 4102 * LWB_STATE_OPENED, _CLOSED and _READY lwbs to be committed 4103 * to disk before proceeding. If we used zil_commit instead, it 4104 * would just call txg_wait_synced(), because zl_suspend is set. 4105 * txg_wait_synced() doesn't wait for these lwb's to be 4106 * LWB_STATE_FLUSH_DONE before returning. 4107 */ 4108 zil_commit_impl(zilog, 0); 4109 4110 /* 4111 * Now that we've ensured all lwb's are LWB_STATE_FLUSH_DONE, we 4112 * use txg_wait_synced() to ensure the data from the zilog has 4113 * migrated to the main pool before calling zil_destroy(). 4114 */ 4115 txg_wait_synced(zilog->zl_dmu_pool, 0); 4116 4117 zil_destroy(zilog, B_FALSE); 4118 4119 mutex_enter(&zilog->zl_lock); 4120 zilog->zl_suspending = B_FALSE; 4121 cv_broadcast(&zilog->zl_cv_suspend); 4122 mutex_exit(&zilog->zl_lock); 4123 4124 if (os->os_encrypted) 4125 dsl_dataset_remove_key_mapping(dmu_objset_ds(os)); 4126 4127 if (cookiep == NULL) 4128 zil_resume(os); 4129 else 4130 *cookiep = os; 4131 return (0); 4132 } 4133 4134 void 4135 zil_resume(void *cookie) 4136 { 4137 objset_t *os = cookie; 4138 zilog_t *zilog = dmu_objset_zil(os); 4139 4140 mutex_enter(&zilog->zl_lock); 4141 ASSERT(zilog->zl_suspend != 0); 4142 zilog->zl_suspend--; 4143 mutex_exit(&zilog->zl_lock); 4144 dsl_dataset_long_rele(dmu_objset_ds(os), suspend_tag); 4145 dsl_dataset_rele(dmu_objset_ds(os), suspend_tag); 4146 } 4147 4148 typedef struct zil_replay_arg { 4149 zil_replay_func_t *const *zr_replay; 4150 void *zr_arg; 4151 boolean_t zr_byteswap; 4152 char *zr_lr; 4153 } zil_replay_arg_t; 4154 4155 static int 4156 zil_replay_error(zilog_t *zilog, const lr_t *lr, int error) 4157 { 4158 char name[ZFS_MAX_DATASET_NAME_LEN]; 4159 4160 zilog->zl_replaying_seq--; /* didn't actually replay this one */ 4161 4162 dmu_objset_name(zilog->zl_os, name); 4163 4164 cmn_err(CE_WARN, "ZFS replay transaction error %d, " 4165 "dataset %s, seq 0x%llx, txtype %llu %s\n", error, name, 4166 (u_longlong_t)lr->lrc_seq, 4167 (u_longlong_t)(lr->lrc_txtype & ~TX_CI), 4168 (lr->lrc_txtype & TX_CI) ? "CI" : ""); 4169 4170 return (error); 4171 } 4172 4173 static int 4174 zil_replay_log_record(zilog_t *zilog, const lr_t *lr, void *zra, 4175 uint64_t claim_txg) 4176 { 4177 zil_replay_arg_t *zr = zra; 4178 const zil_header_t *zh = zilog->zl_header; 4179 uint64_t reclen = lr->lrc_reclen; 4180 uint64_t txtype = lr->lrc_txtype; 4181 int error = 0; 4182 4183 zilog->zl_replaying_seq = lr->lrc_seq; 4184 4185 if (lr->lrc_seq <= zh->zh_replay_seq) /* already replayed */ 4186 return (0); 4187 4188 if (lr->lrc_txg < claim_txg) /* already committed */ 4189 return (0); 4190 4191 /* Strip case-insensitive bit, still present in log record */ 4192 txtype &= ~TX_CI; 4193 4194 if (txtype == 0 || txtype >= TX_MAX_TYPE) 4195 return (zil_replay_error(zilog, lr, EINVAL)); 4196 4197 /* 4198 * If this record type can be logged out of order, the object 4199 * (lr_foid) may no longer exist. That's legitimate, not an error. 4200 */ 4201 if (TX_OOO(txtype)) { 4202 error = dmu_object_info(zilog->zl_os, 4203 LR_FOID_GET_OBJ(((lr_ooo_t *)lr)->lr_foid), NULL); 4204 if (error == ENOENT || error == EEXIST) 4205 return (0); 4206 } 4207 4208 /* 4209 * Make a copy of the data so we can revise and extend it. 4210 */ 4211 memcpy(zr->zr_lr, lr, reclen); 4212 4213 /* 4214 * If this is a TX_WRITE with a blkptr, suck in the data. 4215 */ 4216 if (txtype == TX_WRITE && reclen == sizeof (lr_write_t)) { 4217 error = zil_read_log_data(zilog, (lr_write_t *)lr, 4218 zr->zr_lr + reclen); 4219 if (error != 0) 4220 return (zil_replay_error(zilog, lr, error)); 4221 } 4222 4223 /* 4224 * The log block containing this lr may have been byteswapped 4225 * so that we can easily examine common fields like lrc_txtype. 4226 * However, the log is a mix of different record types, and only the 4227 * replay vectors know how to byteswap their records. Therefore, if 4228 * the lr was byteswapped, undo it before invoking the replay vector. 4229 */ 4230 if (zr->zr_byteswap) 4231 byteswap_uint64_array(zr->zr_lr, reclen); 4232 4233 /* 4234 * We must now do two things atomically: replay this log record, 4235 * and update the log header sequence number to reflect the fact that 4236 * we did so. At the end of each replay function the sequence number 4237 * is updated if we are in replay mode. 4238 */ 4239 error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lr, zr->zr_byteswap); 4240 if (error != 0) { 4241 /* 4242 * The DMU's dnode layer doesn't see removes until the txg 4243 * commits, so a subsequent claim can spuriously fail with 4244 * EEXIST. So if we receive any error we try syncing out 4245 * any removes then retry the transaction. Note that we 4246 * specify B_FALSE for byteswap now, so we don't do it twice. 4247 */ 4248 txg_wait_synced(spa_get_dsl(zilog->zl_spa), 0); 4249 error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lr, B_FALSE); 4250 if (error != 0) 4251 return (zil_replay_error(zilog, lr, error)); 4252 } 4253 return (0); 4254 } 4255 4256 static int 4257 zil_incr_blks(zilog_t *zilog, const blkptr_t *bp, void *arg, uint64_t claim_txg) 4258 { 4259 (void) bp, (void) arg, (void) claim_txg; 4260 4261 zilog->zl_replay_blks++; 4262 4263 return (0); 4264 } 4265 4266 /* 4267 * If this dataset has a non-empty intent log, replay it and destroy it. 4268 * Return B_TRUE if there were any entries to replay. 4269 */ 4270 boolean_t 4271 zil_replay(objset_t *os, void *arg, 4272 zil_replay_func_t *const replay_func[TX_MAX_TYPE]) 4273 { 4274 zilog_t *zilog = dmu_objset_zil(os); 4275 const zil_header_t *zh = zilog->zl_header; 4276 zil_replay_arg_t zr; 4277 4278 if ((zh->zh_flags & ZIL_REPLAY_NEEDED) == 0) { 4279 return (zil_destroy(zilog, B_TRUE)); 4280 } 4281 4282 zr.zr_replay = replay_func; 4283 zr.zr_arg = arg; 4284 zr.zr_byteswap = BP_SHOULD_BYTESWAP(&zh->zh_log); 4285 zr.zr_lr = vmem_alloc(2 * SPA_MAXBLOCKSIZE, KM_SLEEP); 4286 4287 /* 4288 * Wait for in-progress removes to sync before starting replay. 4289 */ 4290 txg_wait_synced(zilog->zl_dmu_pool, 0); 4291 4292 zilog->zl_replay = B_TRUE; 4293 zilog->zl_replay_time = ddi_get_lbolt(); 4294 ASSERT(zilog->zl_replay_blks == 0); 4295 (void) zil_parse(zilog, zil_incr_blks, zil_replay_log_record, &zr, 4296 zh->zh_claim_txg, B_TRUE); 4297 vmem_free(zr.zr_lr, 2 * SPA_MAXBLOCKSIZE); 4298 4299 zil_destroy(zilog, B_FALSE); 4300 txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg); 4301 zilog->zl_replay = B_FALSE; 4302 4303 return (B_TRUE); 4304 } 4305 4306 boolean_t 4307 zil_replaying(zilog_t *zilog, dmu_tx_t *tx) 4308 { 4309 if (zilog->zl_sync == ZFS_SYNC_DISABLED) 4310 return (B_TRUE); 4311 4312 if (zilog->zl_replay) { 4313 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx); 4314 zilog->zl_replayed_seq[dmu_tx_get_txg(tx) & TXG_MASK] = 4315 zilog->zl_replaying_seq; 4316 return (B_TRUE); 4317 } 4318 4319 return (B_FALSE); 4320 } 4321 4322 int 4323 zil_reset(const char *osname, void *arg) 4324 { 4325 (void) arg; 4326 4327 int error = zil_suspend(osname, NULL); 4328 /* EACCES means crypto key not loaded */ 4329 if ((error == EACCES) || (error == EBUSY)) 4330 return (SET_ERROR(error)); 4331 if (error != 0) 4332 return (SET_ERROR(EEXIST)); 4333 return (0); 4334 } 4335 4336 EXPORT_SYMBOL(zil_alloc); 4337 EXPORT_SYMBOL(zil_free); 4338 EXPORT_SYMBOL(zil_open); 4339 EXPORT_SYMBOL(zil_close); 4340 EXPORT_SYMBOL(zil_replay); 4341 EXPORT_SYMBOL(zil_replaying); 4342 EXPORT_SYMBOL(zil_destroy); 4343 EXPORT_SYMBOL(zil_destroy_sync); 4344 EXPORT_SYMBOL(zil_itx_create); 4345 EXPORT_SYMBOL(zil_itx_destroy); 4346 EXPORT_SYMBOL(zil_itx_assign); 4347 EXPORT_SYMBOL(zil_commit); 4348 EXPORT_SYMBOL(zil_claim); 4349 EXPORT_SYMBOL(zil_check_log_chain); 4350 EXPORT_SYMBOL(zil_sync); 4351 EXPORT_SYMBOL(zil_clean); 4352 EXPORT_SYMBOL(zil_suspend); 4353 EXPORT_SYMBOL(zil_resume); 4354 EXPORT_SYMBOL(zil_lwb_add_block); 4355 EXPORT_SYMBOL(zil_bp_tree_add); 4356 EXPORT_SYMBOL(zil_set_sync); 4357 EXPORT_SYMBOL(zil_set_logbias); 4358 EXPORT_SYMBOL(zil_sums_init); 4359 EXPORT_SYMBOL(zil_sums_fini); 4360 EXPORT_SYMBOL(zil_kstat_values_update); 4361 4362 ZFS_MODULE_PARAM(zfs, zfs_, commit_timeout_pct, UINT, ZMOD_RW, 4363 "ZIL block open timeout percentage"); 4364 4365 ZFS_MODULE_PARAM(zfs_zil, zil_, replay_disable, INT, ZMOD_RW, 4366 "Disable intent logging replay"); 4367 4368 ZFS_MODULE_PARAM(zfs_zil, zil_, nocacheflush, INT, ZMOD_RW, 4369 "Disable ZIL cache flushes"); 4370 4371 ZFS_MODULE_PARAM(zfs_zil, zil_, slog_bulk, U64, ZMOD_RW, 4372 "Limit in bytes slog sync writes per commit"); 4373 4374 ZFS_MODULE_PARAM(zfs_zil, zil_, maxblocksize, UINT, ZMOD_RW, 4375 "Limit in bytes of ZIL log block size"); 4376 4377 ZFS_MODULE_PARAM(zfs_zil, zil_, maxcopied, UINT, ZMOD_RW, 4378 "Limit in bytes WR_COPIED size"); 4379