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