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