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