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