zfs: merge openzfs/zfs@9c56b8ec7

Notable upstream pull request merges:
 #15817 5536c0dee Sync AUX label during pool import
 #15889 c7ada64bb ddt: dedup table quota enforcement
 #15890 62e7d3c89 ddt: add support for prefetching tables into the ARC
 #15894 e26b3771e spa_preferred_class: pass the entire zio
 #15894 d54d0fff3 dnode: allow storage class to be overridden by object type
 #16197 55427add3 Several improvements to ARC shrinking
 #16217 -multiple JSON output for various zfs and zpool subcommands
 #16248 24e6585e7 libzfs.h: Set ZFS_MAXPROPLEN and ZPOOL_MAXPROPLEN
                  to ZAP_MAXVALUELEN
 #16264 9dfc5c4a0 Fix long_free_dirty accounting for small files
 #16268 ed0db1cc8 Make txg_wait_synced conditional in zfsvfs_teardown,
                  for FreeBSD
 #16288 d60debbf5 Fix sa_add_projid to lookup and update SA_ZPL_DXATTR
 #16308 ec580bc52 zfs: add bounds checking to zil_parse
 #16310 c21dc56ea Fix zdb_dump_block for little endian
 #16315 7ddc1f737 zil: add stats for commit failure/fallback
 #16326 b0bf14cdb abd: lift ABD zero scan from zio_compress_data()
                  to abd_cmp_zero()
 #16337 c8184d714 Block cloning conditionally destroy ARC buffer
 #16338 dbe07928b Add support for multiple lines to the sharenfs property
                  for FreeBSD
 #16374 1a3e32e6a Cleanup DB_DNODE() macros usage
 #16374 ed87d456e Skip dnode handles use when not needed
 #16346 fb6d8cf22 Add some missing vdev properties
 #16364 670147be5 zvol: ensure device minors are properly cleaned up
 #16382 dea8fabf7 FreeBSD: Fix RLIMIT_FSIZE handling for block cloning
 #16387 aef452f10 Improve zfs_blkptr_verify()
 #16395 cbcb52243 Fix the names of some FreeBSD sysctls in
                  include/tunables.cfg
 #16401 5b9f3b766 Soften pruning threshold on not evictable metadata
 #16404 cdd53fea1 FreeBSD: Add missing memory reclamation accounting
 #16404 1fdcb653b Once more refactor arc_summary output
 #16419 1f5bf91a8 Fix memory corruption during parallel zpool import
                  with -o cachefile
 #16426 cf6e8b218 zstream: remove duplicate highbit64 definition

Obtained from:	OpenZFS
OpenZFS commit:	9c56b8ec7855119c8948c551ee28086a11465a56

23 files changed, 1287 insertions, 313 deletions

diff --git a/sys/contrib/openzfs/module/zfs/abd.c b/sys/contrib/openzfs/module/zfs/abd.c
index 2c0cda25dbc6..94f492522f0d 100644
--- a/sys/contrib/openzfs/module/zfs/abd.c
+++ b/sys/contrib/openzfs/module/zfs/abd.c
@@ -1051,6 +1051,31 @@ abd_cmp(abd_t *dabd, abd_t *sabd)
 }
 
 /*
+ * Check if ABD content is all-zeroes.
+ */
+static int
+abd_cmp_zero_off_cb(void *data, size_t len, void *private)
+{
+	(void) private;
+
+	/* This function can only check whole uint64s. Enforce that. */
+	ASSERT0(P2PHASE(len, 8));
+
+	uint64_t *end = (uint64_t *)((char *)data + len);
+	for (uint64_t *word = (uint64_t *)data; word < end; word++)
+		if (*word != 0)
+			return (1);
+
+	return (0);
+}
+
+int
+abd_cmp_zero_off(abd_t *abd, size_t off, size_t size)
+{
+	return (abd_iterate_func(abd, off, size, abd_cmp_zero_off_cb, NULL));
+}
+
+/*
  * Iterate over code ABDs and a data ABD and call @func_raidz_gen.
  *
  * @cabds          parity ABDs, must have equal size
diff --git a/sys/contrib/openzfs/module/zfs/arc.c b/sys/contrib/openzfs/module/zfs/arc.c
index 30d30b98a6c6..78c2cf8ec5c3 100644
--- a/sys/contrib/openzfs/module/zfs/arc.c
+++ b/sys/contrib/openzfs/module/zfs/arc.c
@@ -26,7 +26,7 @@
  * Copyright (c) 2017, Nexenta Systems, Inc.  All rights reserved.
  * Copyright (c) 2019, loli10K <ezomori.nozomu@gmail.com>. All rights reserved.
  * Copyright (c) 2020, George Amanakis. All rights reserved.
- * Copyright (c) 2019, Klara Inc.
+ * Copyright (c) 2019, 2023, Klara Inc.
  * Copyright (c) 2019, Allan Jude
  * Copyright (c) 2020, The FreeBSD Foundation [1]
  *
@@ -1258,7 +1258,7 @@ retry:
 	}
 
 	hdr_full_cache = kmem_cache_create("arc_buf_hdr_t_full", HDR_FULL_SIZE,
-	    0, hdr_full_cons, hdr_full_dest, NULL, NULL, NULL, 0);
+	    0, hdr_full_cons, hdr_full_dest, NULL, NULL, NULL, KMC_RECLAIMABLE);
 	hdr_l2only_cache = kmem_cache_create("arc_buf_hdr_t_l2only",
 	    HDR_L2ONLY_SIZE, 0, hdr_l2only_cons, hdr_l2only_dest, NULL,
 	    NULL, NULL, 0);
@@ -4236,6 +4236,18 @@ arc_evict_adj(uint64_t frac, uint64_t total, uint64_t up, uint64_t down,
 }
 
 /*
+ * Calculate (x * multiplier / divisor) without unnecesary overflows.
+ */
+static uint64_t
+arc_mf(uint64_t x, uint64_t multiplier, uint64_t divisor)
+{
+	uint64_t q = (x / divisor);
+	uint64_t r = (x % divisor);
+
+	return ((q * multiplier) + ((r * multiplier) / divisor));
+}
+
+/*
  * Evict buffers from the cache, such that arcstat_size is capped by arc_c.
  */
 static uint64_t
@@ -4287,17 +4299,20 @@ arc_evict(void)
 	 */
 	int64_t prune = 0;
 	int64_t dn = wmsum_value(&arc_sums.arcstat_dnode_size);
+	int64_t nem = zfs_refcount_count(&arc_mru->arcs_size[ARC_BUFC_METADATA])
+	    + zfs_refcount_count(&arc_mfu->arcs_size[ARC_BUFC_METADATA])
+	    - zfs_refcount_count(&arc_mru->arcs_esize[ARC_BUFC_METADATA])
+	    - zfs_refcount_count(&arc_mfu->arcs_esize[ARC_BUFC_METADATA]);
 	w = wt * (int64_t)(arc_meta >> 16) >> 16;
-	if (zfs_refcount_count(&arc_mru->arcs_size[ARC_BUFC_METADATA]) +
-	    zfs_refcount_count(&arc_mfu->arcs_size[ARC_BUFC_METADATA]) -
-	    zfs_refcount_count(&arc_mru->arcs_esize[ARC_BUFC_METADATA]) -
-	    zfs_refcount_count(&arc_mfu->arcs_esize[ARC_BUFC_METADATA]) >
-	    w * 3 / 4) {
+	if (nem > w * 3 / 4) {
 		prune = dn / sizeof (dnode_t) *
 		    zfs_arc_dnode_reduce_percent / 100;
-	} else if (dn > arc_dnode_limit) {
-		prune = (dn - arc_dnode_limit) / sizeof (dnode_t) *
-		    zfs_arc_dnode_reduce_percent / 100;
+		if (nem < w && w > 4)
+			prune = arc_mf(prune, nem - w * 3 / 4, w / 4);
+	}
+	if (dn > arc_dnode_limit) {
+		prune = MAX(prune, (dn - arc_dnode_limit) / sizeof (dnode_t) *
+		    zfs_arc_dnode_reduce_percent / 100);
 	}
 	if (prune > 0)
 		arc_prune_async(prune);
@@ -4398,13 +4413,14 @@ arc_flush(spa_t *spa, boolean_t retry)
 	(void) arc_flush_state(arc_uncached, guid, ARC_BUFC_METADATA, retry);
 }
 
-void
-arc_reduce_target_size(int64_t to_free)
+uint64_t
+arc_reduce_target_size(uint64_t to_free)
 {
-	uint64_t c = arc_c;
-
-	if (c <= arc_c_min)
-		return;
+	/*
+	 * Get the actual arc size.  Even if we don't need it, this updates
+	 * the aggsum lower bound estimate for arc_is_overflowing().
+	 */
+	uint64_t asize = aggsum_value(&arc_sums.arcstat_size);
 
 	/*
 	 * All callers want the ARC to actually evict (at least) this much
@@ -4414,16 +4430,28 @@ arc_reduce_target_size(int64_t to_free)
 	 * immediately have arc_c < arc_size and therefore the arc_evict_zthr
 	 * will evict.
 	 */
-	uint64_t asize = aggsum_value(&arc_sums.arcstat_size);
-	if (asize < c)
-		to_free += c - asize;
-	arc_c = MAX((int64_t)c - to_free, (int64_t)arc_c_min);
+	uint64_t c = arc_c;
+	if (c > arc_c_min) {
+		c = MIN(c, MAX(asize, arc_c_min));
+		to_free = MIN(to_free, c - arc_c_min);
+		arc_c = c - to_free;
+	} else {
+		to_free = 0;
+	}
 
-	/* See comment in arc_evict_cb_check() on why lock+flag */
-	mutex_enter(&arc_evict_lock);
-	arc_evict_needed = B_TRUE;
-	mutex_exit(&arc_evict_lock);
-	zthr_wakeup(arc_evict_zthr);
+	/*
+	 * Whether or not we reduced the target size, request eviction if the
+	 * current size is over it now, since caller obviously wants some RAM.
+	 */
+	if (asize > arc_c) {
+		/* See comment in arc_evict_cb_check() on why lock+flag */
+		mutex_enter(&arc_evict_lock);
+		arc_evict_needed = B_TRUE;
+		mutex_exit(&arc_evict_lock);
+		zthr_wakeup(arc_evict_zthr);
+	}
+
+	return (to_free);
 }
 
 /*
@@ -4630,9 +4658,9 @@ arc_reap_cb_check(void *arg, zthr_t *zthr)
 static void
 arc_reap_cb(void *arg, zthr_t *zthr)
 {
-	(void) arg, (void) zthr;
+	int64_t can_free, free_memory, to_free;
 
-	int64_t free_memory;
+	(void) arg, (void) zthr;
 	fstrans_cookie_t cookie = spl_fstrans_mark();
 
 	/*
@@ -4660,13 +4688,10 @@ arc_reap_cb(void *arg, zthr_t *zthr)
 	 * amount, reduce by what is needed to hit the fractional amount.
 	 */
 	free_memory = arc_available_memory();
-
-	int64_t can_free = arc_c - arc_c_min;
-	if (can_free > 0) {
-		int64_t to_free = (can_free >> arc_shrink_shift) - free_memory;
-		if (to_free > 0)
-			arc_reduce_target_size(to_free);
-	}
+	can_free = arc_c - arc_c_min;
+	to_free = (MAX(can_free, 0) >> arc_shrink_shift) - free_memory;
+	if (to_free > 0)
+		arc_reduce_target_size(to_free);
 	spl_fstrans_unmark(cookie);
 }
 
@@ -4754,16 +4779,11 @@ arc_adapt(uint64_t bytes)
 }
 
 /*
- * Check if arc_size has grown past our upper threshold, determined by
- * zfs_arc_overflow_shift.
+ * Check if ARC current size has grown past our upper thresholds.
  */
 static arc_ovf_level_t
-arc_is_overflowing(boolean_t use_reserve)
+arc_is_overflowing(boolean_t lax, boolean_t use_reserve)
 {
-	/* Always allow at least one block of overflow */
-	int64_t overflow = MAX(SPA_MAXBLOCKSIZE,
-	    arc_c >> zfs_arc_overflow_shift);
-
 	/*
 	 * We just compare the lower bound here for performance reasons. Our
 	 * primary goals are to make sure that the arc never grows without
@@ -4773,12 +4793,22 @@ arc_is_overflowing(boolean_t use_reserve)
 	 * in the ARC. In practice, that's in the tens of MB, which is low
 	 * enough to be safe.
 	 */
-	int64_t over = aggsum_lower_bound(&arc_sums.arcstat_size) -
-	    arc_c - overflow / 2;
-	if (!use_reserve)
-		overflow /= 2;
-	return (over < 0 ? ARC_OVF_NONE :
-	    over < overflow ? ARC_OVF_SOME : ARC_OVF_SEVERE);
+	int64_t over = aggsum_lower_bound(&arc_sums.arcstat_size) - arc_c -
+	    zfs_max_recordsize;
+
+	/* Always allow at least one block of overflow. */
+	if (over < 0)
+		return (ARC_OVF_NONE);
+
+	/* If we are under memory pressure, report severe overflow. */
+	if (!lax)
+		return (ARC_OVF_SEVERE);
+
+	/* We are not under pressure, so be more or less relaxed. */
+	int64_t overflow = (arc_c >> zfs_arc_overflow_shift) / 2;
+	if (use_reserve)
+		overflow *= 3;
+	return (over < overflow ? ARC_OVF_SOME : ARC_OVF_SEVERE);
 }
 
 static abd_t *
@@ -4810,15 +4840,17 @@ arc_get_data_buf(arc_buf_hdr_t *hdr, uint64_t size, const void *tag)
 
 /*
  * Wait for the specified amount of data (in bytes) to be evicted from the
- * ARC, and for there to be sufficient free memory in the system.  Waiting for
- * eviction ensures that the memory used by the ARC decreases.  Waiting for
- * free memory ensures that the system won't run out of free pages, regardless
- * of ARC behavior and settings.  See arc_lowmem_init().
+ * ARC, and for there to be sufficient free memory in the system.
+ * The lax argument specifies that caller does not have a specific reason
+ * to wait, not aware of any memory pressure.  Low memory handlers though
+ * should set it to B_FALSE to wait for all required evictions to complete.
+ * The use_reserve argument allows some callers to wait less than others
+ * to not block critical code paths, possibly blocking other resources.
  */
 void
-arc_wait_for_eviction(uint64_t amount, boolean_t use_reserve)
+arc_wait_for_eviction(uint64_t amount, boolean_t lax, boolean_t use_reserve)
 {
-	switch (arc_is_overflowing(use_reserve)) {
+	switch (arc_is_overflowing(lax, use_reserve)) {
 	case ARC_OVF_NONE:
 		return;
 	case ARC_OVF_SOME:
@@ -4913,7 +4945,7 @@ arc_get_data_impl(arc_buf_hdr_t *hdr, uint64_t size, const void *tag,
 	 * under arc_c.  See the comment above zfs_arc_eviction_pct.
 	 */
 	arc_wait_for_eviction(size * zfs_arc_eviction_pct / 100,
-	    alloc_flags & ARC_HDR_USE_RESERVE);
+	    B_TRUE, alloc_flags & ARC_HDR_USE_RESERVE);
 
 	arc_buf_contents_t type = arc_buf_type(hdr);
 	if (type == ARC_BUFC_METADATA) {
@@ -5455,6 +5487,57 @@ arc_read_done(zio_t *zio)
 }
 
 /*
+ * Lookup the block at the specified DVA (in bp), and return the manner in
+ * which the block is cached. A zero return indicates not cached.
+ */
+int
+arc_cached(spa_t *spa, const blkptr_t *bp)
+{
+	arc_buf_hdr_t *hdr = NULL;
+	kmutex_t *hash_lock = NULL;
+	uint64_t guid = spa_load_guid(spa);
+	int flags = 0;
+
+	if (BP_IS_EMBEDDED(bp))
+		return (ARC_CACHED_EMBEDDED);
+
+	hdr = buf_hash_find(guid, bp, &hash_lock);
+	if (hdr == NULL)
+		return (0);
+
+	if (HDR_HAS_L1HDR(hdr)) {
+		arc_state_t *state = hdr->b_l1hdr.b_state;
+		/*
+		 * We switch to ensure that any future arc_state_type_t
+		 * changes are handled. This is just a shift to promote
+		 * more compile-time checking.
+		 */
+		switch (state->arcs_state) {
+		case ARC_STATE_ANON:
+			break;
+		case ARC_STATE_MRU:
+			flags |= ARC_CACHED_IN_MRU | ARC_CACHED_IN_L1;
+			break;
+		case ARC_STATE_MFU:
+			flags |= ARC_CACHED_IN_MFU | ARC_CACHED_IN_L1;
+			break;
+		case ARC_STATE_UNCACHED:
+			/* The header is still in L1, probably not for long */
+			flags |= ARC_CACHED_IN_L1;
+			break;
+		default:
+			break;
+		}
+	}
+	if (HDR_HAS_L2HDR(hdr))
+		flags |= ARC_CACHED_IN_L2;
+
+	mutex_exit(hash_lock);
+
+	return (flags);
+}
+
+/*
  * "Read" the block at the specified DVA (in bp) via the
  * cache.  If the block is found in the cache, invoke the provided
  * callback immediately and return.  Note that the `zio' parameter
@@ -5508,19 +5591,6 @@ arc_read(zio_t *pio, spa_t *spa, const blkptr_t *bp,
 	 */
 	fstrans_cookie_t cookie = spl_fstrans_mark();
 top:
-	/*
-	 * Verify the block pointer contents are reasonable.  This should
-	 * always be the case since the blkptr is protected by a checksum.
-	 * However, if there is damage it's desirable to detect this early
-	 * and treat it as a checksum error.  This allows an alternate blkptr
-	 * to be tried when one is available (e.g. ditto blocks).
-	 */
-	if (!zfs_blkptr_verify(spa, bp, (zio_flags & ZIO_FLAG_CONFIG_WRITER) ?
-	    BLK_CONFIG_HELD : BLK_CONFIG_NEEDED, BLK_VERIFY_LOG)) {
-		rc = SET_ERROR(ECKSUM);
-		goto done;
-	}
-
 	if (!embedded_bp) {
 		/*
 		 * Embedded BP's have no DVA and require no I/O to "read".
@@ -5540,6 +5610,18 @@ top:
 	    (hdr->b_l1hdr.b_pabd != NULL && !encrypted_read))) {
 		boolean_t is_data = !HDR_ISTYPE_METADATA(hdr);
 
+		/*
+		 * Verify the block pointer contents are reasonable.  This
+		 * should always be the case since the blkptr is protected by
+		 * a checksum.
+		 */
+		if (!zfs_blkptr_verify(spa, bp, BLK_CONFIG_SKIP,
+		    BLK_VERIFY_LOG)) {
+			mutex_exit(hash_lock);
+			rc = SET_ERROR(ECKSUM);
+			goto done;
+		}
+
 		if (HDR_IO_IN_PROGRESS(hdr)) {
 			if (*arc_flags & ARC_FLAG_CACHED_ONLY) {
 				mutex_exit(hash_lock);
@@ -5693,6 +5775,20 @@ top:
 			goto done;
 		}
 
+		/*
+		 * Verify the block pointer contents are reasonable.  This
+		 * should always be the case since the blkptr is protected by
+		 * a checksum.
+		 */
+		if (!zfs_blkptr_verify(spa, bp,
+		    (zio_flags & ZIO_FLAG_CONFIG_WRITER) ?
+		    BLK_CONFIG_HELD : BLK_CONFIG_NEEDED, BLK_VERIFY_LOG)) {
+			if (hash_lock != NULL)
+				mutex_exit(hash_lock);
+			rc = SET_ERROR(ECKSUM);
+			goto done;
+		}
+
 		if (hdr == NULL) {
 			/*
 			 * This block is not in the cache or it has
diff --git a/sys/contrib/openzfs/module/zfs/dbuf.c b/sys/contrib/openzfs/module/zfs/dbuf.c
index 56fe2c4dbe30..099883ba2652 100644
--- a/sys/contrib/openzfs/module/zfs/dbuf.c
+++ b/sys/contrib/openzfs/module/zfs/dbuf.c
@@ -2705,6 +2705,9 @@ void
 dmu_buf_will_clone(dmu_buf_t *db_fake, dmu_tx_t *tx)
 {
 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
+	ASSERT0(db->db_level);
+	ASSERT(db->db_blkid != DMU_BONUS_BLKID);
+	ASSERT(db->db.db_object != DMU_META_DNODE_OBJECT);
 
 	/*
 	 * Block cloning: We are going to clone into this block, so undirty
@@ -2716,11 +2719,22 @@ dmu_buf_will_clone(dmu_buf_t *db_fake, dmu_tx_t *tx)
 	VERIFY(!dbuf_undirty(db, tx));
 	ASSERT0P(dbuf_find_dirty_eq(db, tx->tx_txg));
 	if (db->db_buf != NULL) {
-		arc_buf_destroy(db->db_buf, db);
+		/*
+		 * If there is an associated ARC buffer with this dbuf we can
+		 * only destroy it if the previous dirty record does not
+		 * reference it.
+		 */
+		dbuf_dirty_record_t *dr = list_head(&db->db_dirty_records);
+		if (dr == NULL || dr->dt.dl.dr_data != db->db_buf)
+			arc_buf_destroy(db->db_buf, db);
+
 		db->db_buf = NULL;
 		dbuf_clear_data(db);
 	}
 
+	ASSERT3P(db->db_buf, ==, NULL);
+	ASSERT3P(db->db.db_data, ==, NULL);
+
 	db->db_state = DB_NOFILL;
 	DTRACE_SET_STATE(db, "allocating NOFILL buffer for clone");
 
@@ -3103,7 +3117,11 @@ dbuf_destroy(dmu_buf_impl_t *db)
 		 */
 		mutex_enter(&dn->dn_mtx);
 		dnode_rele_and_unlock(dn, db, B_TRUE);
+#ifdef USE_DNODE_HANDLE
 		db->db_dnode_handle = NULL;
+#else
+		db->db_dnode = NULL;
+#endif
 
 		dbuf_hash_remove(db);
 	} else {
@@ -3252,7 +3270,11 @@ dbuf_create(dnode_t *dn, uint8_t level, uint64_t blkid,
 	db->db_level = level;
 	db->db_blkid = blkid;
 	db->db_dirtycnt = 0;
+#ifdef USE_DNODE_HANDLE
 	db->db_dnode_handle = dn->dn_handle;
+#else
+	db->db_dnode = dn;
+#endif
 	db->db_parent = parent;
 	db->db_blkptr = blkptr;
 	db->db_hash = hash;
@@ -4390,7 +4412,7 @@ dbuf_lightweight_bp(dbuf_dirty_record_t *dr)
 		dmu_buf_impl_t *parent_db = dr->dr_parent->dr_dbuf;
 		int epbs = dn->dn_indblkshift - SPA_BLKPTRSHIFT;
 		VERIFY3U(parent_db->db_level, ==, 1);
-		VERIFY3P(parent_db->db_dnode_handle->dnh_dnode, ==, dn);
+		VERIFY3P(DB_DNODE(parent_db), ==, dn);
 		VERIFY3U(dr->dt.dll.dr_blkid >> epbs, ==, parent_db->db_blkid);
 		blkptr_t *bp = parent_db->db.db_data;
 		return (&bp[dr->dt.dll.dr_blkid & ((1 << epbs) - 1)]);
@@ -4813,14 +4835,13 @@ dbuf_write_children_ready(zio_t *zio, arc_buf_t *buf, void *vdb)
 {
 	(void) zio, (void) buf;
 	dmu_buf_impl_t *db = vdb;
-	dnode_t *dn;
 	blkptr_t *bp;
 	unsigned int epbs, i;
 
 	ASSERT3U(db->db_level, >, 0);
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	epbs = dn->dn_phys->dn_indblkshift - SPA_BLKPTRSHIFT;
+	epbs = DB_DNODE(db)->dn_phys->dn_indblkshift - SPA_BLKPTRSHIFT;
+	DB_DNODE_EXIT(db);
 	ASSERT3U(epbs, <, 31);
 
 	/* Determine if all our children are holes */
@@ -4843,7 +4864,6 @@ dbuf_write_children_ready(zio_t *zio, arc_buf_t *buf, void *vdb)
 		memset(db->db.db_data, 0, db->db.db_size);
 		rw_exit(&db->db_rwlock);
 	}
-	DB_DNODE_EXIT(db);
 }
 
 static void
@@ -5062,8 +5082,7 @@ dbuf_remap(dnode_t *dn, dmu_buf_impl_t *db, dmu_tx_t *tx)
 		}
 	} else if (db->db.db_object == DMU_META_DNODE_OBJECT) {
 		dnode_phys_t *dnp = db->db.db_data;
-		ASSERT3U(db->db_dnode_handle->dnh_dnode->dn_type, ==,
-		    DMU_OT_DNODE);
+		ASSERT3U(dn->dn_type, ==, DMU_OT_DNODE);
 		for (int i = 0; i < db->db.db_size >> DNODE_SHIFT;
 		    i += dnp[i].dn_extra_slots + 1) {
 			for (int j = 0; j < dnp[i].dn_nblkptr; j++) {
diff --git a/sys/contrib/openzfs/module/zfs/ddt.c b/sys/contrib/openzfs/module/zfs/ddt.c
index 4c53cb0a2f9b..d70ae1a031d5 100644
--- a/sys/contrib/openzfs/module/zfs/ddt.c
+++ b/sys/contrib/openzfs/module/zfs/ddt.c
@@ -23,7 +23,7 @@
  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  * Copyright (c) 2012, 2016 by Delphix. All rights reserved.
  * Copyright (c) 2022 by Pawel Jakub Dawidek
- * Copyright (c) 2023, Klara Inc.
+ * Copyright (c) 2019, 2023, Klara Inc.
  */
 
 #include <sys/zfs_context.h>
@@ -101,6 +101,22 @@
  * object and (if necessary), removed from an old one. ddt_tree is cleared and
  * the next txg can start.
  *
+ * ## Dedup quota
+ *
+ * A maximum size for all DDTs on the pool can be set with the
+ * dedup_table_quota property. This is determined in ddt_over_quota() and
+ * enforced during ddt_lookup(). If the pool is at or over its quota limit,
+ * ddt_lookup() will only return entries for existing blocks, as updates are
+ * still possible. New entries will not be created; instead, ddt_lookup() will
+ * return NULL. In response, the DDT write stage (zio_ddt_write()) will remove
+ * the D bit on the block and reissue the IO as a regular write. The block will
+ * not be deduplicated.
+ *
+ * Note that this is based on the on-disk size of the dedup store. Reclaiming
+ * this space after deleting entries relies on the ZAP "shrinking" behaviour,
+ * without which, no space would be recovered and the DDT would continue to be
+ * considered "over quota". See zap_shrink_enabled.
+ *
  * ## Repair IO
  *
  * If a read on a dedup block fails, but there are other copies of the block in
@@ -152,6 +168,13 @@ static kmem_cache_t *ddt_entry_cache;
  */
 int zfs_dedup_prefetch = 0;
 
+/*
+ * If the dedup class cannot satisfy a DDT allocation, treat as over quota
+ * for this many TXGs.
+ */
+uint_t dedup_class_wait_txgs = 5;
+
+
 static const ddt_ops_t *const ddt_ops[DDT_TYPES] = {
 	&ddt_zap_ops,
 };
@@ -317,6 +340,16 @@ ddt_object_prefetch(ddt_t *ddt, ddt_type_t type, ddt_class_t class,
 	    ddt->ddt_object[type][class], ddk);
 }
 
+static void
+ddt_object_prefetch_all(ddt_t *ddt, ddt_type_t type, ddt_class_t class)
+{
+	if (!ddt_object_exists(ddt, type, class))
+		return;
+
+	ddt_ops[type]->ddt_op_prefetch_all(ddt->ddt_os,
+	    ddt->ddt_object[type][class]);
+}
+
 static int
 ddt_object_update(ddt_t *ddt, ddt_type_t type, ddt_class_t class,
     ddt_entry_t *dde, dmu_tx_t *tx)
@@ -554,8 +587,6 @@ ddt_alloc(const ddt_key_t *ddk)
 static void
 ddt_free(ddt_entry_t *dde)
 {
-	ASSERT(dde->dde_flags & DDE_FLAG_LOADED);
-
 	for (int p = 0; p < DDT_PHYS_TYPES; p++)
 		ASSERT3P(dde->dde_lead_zio[p], ==, NULL);
 
@@ -575,9 +606,88 @@ ddt_remove(ddt_t *ddt, ddt_entry_t *dde)
 	ddt_free(dde);
 }
 
+static boolean_t
+ddt_special_over_quota(spa_t *spa, metaslab_class_t *mc)
+{
+	if (mc != NULL && metaslab_class_get_space(mc) > 0) {
+		/* Over quota if allocating outside of this special class */
+		if (spa_syncing_txg(spa) <= spa->spa_dedup_class_full_txg +
+		    dedup_class_wait_txgs) {
+			/* Waiting for some deferred frees to be processed */
+			return (B_TRUE);
+		}
+
+		/*
+		 * We're considered over quota when we hit 85% full, or for
+		 * larger drives, when there is less than 8GB free.
+		 */
+		uint64_t allocated = metaslab_class_get_alloc(mc);
+		uint64_t capacity = metaslab_class_get_space(mc);
+		uint64_t limit = MAX(capacity * 85 / 100,
+		    (capacity > (1LL<<33)) ? capacity - (1LL<<33) : 0);
+
+		return (allocated >= limit);
+	}
+	return (B_FALSE);
+}
+
+/*
+ * Check if the DDT is over its quota.  This can be due to a few conditions:
+ *   1. 'dedup_table_quota' property is not 0 (none) and the dedup dsize
+ *       exceeds this limit
+ *
+ *   2. 'dedup_table_quota' property is set to automatic and
+ *      a. the dedup or special allocation class could not satisfy a DDT
+ *         allocation in a recent transaction
+ *      b. the dedup or special allocation class has exceeded its 85% limit
+ */
+static boolean_t
+ddt_over_quota(spa_t *spa)
+{
+	if (spa->spa_dedup_table_quota == 0)
+		return (B_FALSE);
+
+	if (spa->spa_dedup_table_quota != UINT64_MAX)
+		return (ddt_get_ddt_dsize(spa) > spa->spa_dedup_table_quota);
+
+	/*
+	 * For automatic quota, table size is limited by dedup or special class
+	 */
+	if (ddt_special_over_quota(spa, spa_dedup_class(spa)))
+		return (B_TRUE);
+	else if (spa_special_has_ddt(spa) &&
+	    ddt_special_over_quota(spa, spa_special_class(spa)))
+		return (B_TRUE);
+
+	return (B_FALSE);
+}
+
+void
+ddt_prefetch_all(spa_t *spa)
+{
+	/*
+	 * Load all DDT entries for each type/class combination. This is
+	 * indended to perform a prefetch on all such blocks. For the same
+	 * reason that ddt_prefetch isn't locked, this is also not locked.
+	 */
+	for (enum zio_checksum c = 0; c < ZIO_CHECKSUM_FUNCTIONS; c++) {
+		ddt_t *ddt = spa->spa_ddt[c];
+		if (!ddt)
+			continue;
+
+		for (ddt_type_t type = 0; type < DDT_TYPES; type++) {
+			for (ddt_class_t class = 0; class < DDT_CLASSES;
+			    class++) {
+				ddt_object_prefetch_all(ddt, type, class);
+			}
+		}
+	}
+}
+
 ddt_entry_t *
 ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 {
+	spa_t *spa = ddt->ddt_spa;
 	ddt_key_t search;
 	ddt_entry_t *dde;
 	ddt_type_t type;
@@ -592,13 +702,28 @@ ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 	/* Find an existing live entry */
 	dde = avl_find(&ddt->ddt_tree, &search, &where);
 	if (dde != NULL) {
-		/* Found it. If it's already loaded, we can just return it. */
+		/* If we went over quota, act like we didn't find it */
+		if (dde->dde_flags & DDE_FLAG_OVERQUOTA)
+			return (NULL);
+
+		/* If it's already loaded, we can just return it. */
 		if (dde->dde_flags & DDE_FLAG_LOADED)
 			return (dde);
 
 		/* Someone else is loading it, wait for it. */
+		dde->dde_waiters++;
 		while (!(dde->dde_flags & DDE_FLAG_LOADED))
 			cv_wait(&dde->dde_cv, &ddt->ddt_lock);
+		dde->dde_waiters--;
+
+		/* Loaded but over quota, forget we were ever here */
+		if (dde->dde_flags & DDE_FLAG_OVERQUOTA) {
+			if (dde->dde_waiters == 0) {
+				avl_remove(&ddt->ddt_tree, dde);
+				ddt_free(dde);
+			}
+			return (NULL);
+		}
 
 		return (dde);
 	}
@@ -639,14 +764,27 @@ ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 	dde->dde_type = type;	/* will be DDT_TYPES if no entry found */
 	dde->dde_class = class;	/* will be DDT_CLASSES if no entry found */
 
-	if (error == 0)
+	if (dde->dde_type == DDT_TYPES &&
+	    dde->dde_class == DDT_CLASSES &&
+	    ddt_over_quota(spa)) {
+		/* Over quota. If no one is waiting, clean up right now. */
+		if (dde->dde_waiters == 0) {
+			avl_remove(&ddt->ddt_tree, dde);
+			ddt_free(dde);
+			return (NULL);
+		}
+
+		/* Flag cleanup required */
+		dde->dde_flags |= DDE_FLAG_OVERQUOTA;
+	} else if (error == 0) {
 		ddt_stat_update(ddt, dde, -1ULL);
+	}
 
 	/* Entry loaded, everyone can proceed now */
 	dde->dde_flags |= DDE_FLAG_LOADED;
 	cv_broadcast(&dde->dde_cv);
 
-	return (dde);
+	return (dde->dde_flags & DDE_FLAG_OVERQUOTA ? NULL : dde);
 }
 
 void
@@ -775,6 +913,7 @@ ddt_load(spa_t *spa)
 		memcpy(&ddt->ddt_histogram_cache, ddt->ddt_histogram,
 		    sizeof (ddt->ddt_histogram));
 		spa->spa_dedup_dspace = ~0ULL;
+		spa->spa_dedup_dsize = ~0ULL;
 	}
 
 	return (0);
@@ -1032,6 +1171,7 @@ ddt_sync_table(ddt_t *ddt, dmu_tx_t *tx, uint64_t txg)
 	memcpy(&ddt->ddt_histogram_cache, ddt->ddt_histogram,
 	    sizeof (ddt->ddt_histogram));
 	spa->spa_dedup_dspace = ~0ULL;
+	spa->spa_dedup_dsize = ~0ULL;
 }
 
 void
@@ -1123,7 +1263,13 @@ ddt_addref(spa_t *spa, const blkptr_t *bp)
 	ddt_enter(ddt);
 
 	dde = ddt_lookup(ddt, bp, B_TRUE);
-	ASSERT3P(dde, !=, NULL);
+
+	/* Can be NULL if the entry for this block was pruned. */
+	if (dde == NULL) {
+		ddt_exit(ddt);
+		spa_config_exit(spa, SCL_ZIO, FTAG);
+		return (B_FALSE);
+	}
 
 	if (dde->dde_type < DDT_TYPES) {
 		ddt_phys_t *ddp;
diff --git a/sys/contrib/openzfs/module/zfs/ddt_stats.c b/sys/contrib/openzfs/module/zfs/ddt_stats.c
index af5365a1d114..82b682019ae9 100644
--- a/sys/contrib/openzfs/module/zfs/ddt_stats.c
+++ b/sys/contrib/openzfs/module/zfs/ddt_stats.c
@@ -129,7 +129,8 @@ ddt_histogram_empty(const ddt_histogram_t *ddh)
 void
 ddt_get_dedup_object_stats(spa_t *spa, ddt_object_t *ddo_total)
 {
-	/* Sum the statistics we cached in ddt_object_sync(). */
+	memset(ddo_total, 0, sizeof (*ddo_total));
+
 	for (enum zio_checksum c = 0; c < ZIO_CHECKSUM_FUNCTIONS; c++) {
 		ddt_t *ddt = spa->spa_ddt[c];
 		if (!ddt)
@@ -138,8 +139,32 @@ ddt_get_dedup_object_stats(spa_t *spa, ddt_object_t *ddo_total)
 		for (ddt_type_t type = 0; type < DDT_TYPES; type++) {
 			for (ddt_class_t class = 0; class < DDT_CLASSES;
 			    class++) {
+				dmu_object_info_t doi;
+				uint64_t cnt;
+				int err;
+
+				/*
+				 * These stats were originally calculated
+				 * during ddt_object_load().
+				 */
+
+				err = ddt_object_info(ddt, type, class, &doi);
+				if (err != 0)
+					continue;
+
+				err = ddt_object_count(ddt, type, class, &cnt);
+				if (err != 0)
+					continue;
+
 				ddt_object_t *ddo =
 				    &ddt->ddt_object_stats[type][class];
+
+				ddo->ddo_count = cnt;
+				ddo->ddo_dspace =
+				    doi.doi_physical_blocks_512 << 9;
+				ddo->ddo_mspace = doi.doi_fill_count *
+				    doi.doi_data_block_size;
+
 				ddo_total->ddo_count += ddo->ddo_count;
 				ddo_total->ddo_dspace += ddo->ddo_dspace;
 				ddo_total->ddo_mspace += ddo->ddo_mspace;
@@ -147,11 +172,24 @@ ddt_get_dedup_object_stats(spa_t *spa, ddt_object_t *ddo_total)
 		}
 	}
 
-	/* ... and compute the averages. */
-	if (ddo_total->ddo_count != 0) {
-		ddo_total->ddo_dspace /= ddo_total->ddo_count;
-		ddo_total->ddo_mspace /= ddo_total->ddo_count;
-	}
+	/*
+	 * This returns raw counts (not averages). One of the consumers,
+	 * print_dedup_stats(), historically has expected raw counts.
+	 */
+
+	spa->spa_dedup_dsize = ddo_total->ddo_dspace;
+}
+
+uint64_t
+ddt_get_ddt_dsize(spa_t *spa)
+{
+	ddt_object_t ddo_total;
+
+	/* recalculate after each txg sync */
+	if (spa->spa_dedup_dsize == ~0ULL)
+		ddt_get_dedup_object_stats(spa, &ddo_total);
+
+	return (spa->spa_dedup_dsize);
 }
 
 void
@@ -210,3 +248,32 @@ ddt_get_pool_dedup_ratio(spa_t *spa)
 
 	return (dds_total.dds_ref_dsize * 100 / dds_total.dds_dsize);
 }
+
+int
+ddt_get_pool_dedup_cached(spa_t *spa, uint64_t *psize)
+{
+	uint64_t l1sz, l1tot, l2sz, l2tot;
+	int err = 0;
+
+	l1tot = l2tot = 0;
+	*psize = 0;
+	for (enum zio_checksum c = 0; c < ZIO_CHECKSUM_FUNCTIONS; c++) {
+		ddt_t *ddt = spa->spa_ddt[c];
+		if (ddt == NULL)
+			continue;
+		for (ddt_type_t type = 0; type < DDT_TYPES; type++) {
+			for (ddt_class_t class = 0; class < DDT_CLASSES;
+			    class++) {
+				err = dmu_object_cached_size(ddt->ddt_os,
+				    ddt->ddt_object[type][class], &l1sz, &l2sz);
+				if (err != 0)
+					return (err);
+				l1tot += l1sz;
+				l2tot += l2sz;
+			}
+		}
+	}
+
+	*psize = l1tot + l2tot;
+	return (err);
+}
diff --git a/sys/contrib/openzfs/module/zfs/ddt_zap.c b/sys/contrib/openzfs/module/zfs/ddt_zap.c
index 741554de3c60..7ce7461a2b25 100644
--- a/sys/contrib/openzfs/module/zfs/ddt_zap.c
+++ b/sys/contrib/openzfs/module/zfs/ddt_zap.c
@@ -147,6 +147,12 @@ ddt_zap_prefetch(objset_t *os, uint64_t object, const ddt_key_t *ddk)
 	(void) zap_prefetch_uint64(os, object, (uint64_t *)ddk, DDT_KEY_WORDS);
 }
 
+static void
+ddt_zap_prefetch_all(objset_t *os, uint64_t object)
+{
+	(void) zap_prefetch_object(os, object);
+}
+
 static int
 ddt_zap_update(objset_t *os, uint64_t object, const ddt_key_t *ddk,
     const ddt_phys_t *phys, size_t psize, dmu_tx_t *tx)
@@ -231,6 +237,7 @@ const ddt_ops_t ddt_zap_ops = {
 	ddt_zap_lookup,
 	ddt_zap_contains,
 	ddt_zap_prefetch,
+	ddt_zap_prefetch_all,
 	ddt_zap_update,
 	ddt_zap_remove,
 	ddt_zap_walk,
diff --git a/sys/contrib/openzfs/module/zfs/dmu.c b/sys/contrib/openzfs/module/zfs/dmu.c
index 8b440aafba43..3dcf49ceb64e 100644
--- a/sys/contrib/openzfs/module/zfs/dmu.c
+++ b/sys/contrib/openzfs/module/zfs/dmu.c
@@ -26,7 +26,7 @@
  * Copyright (c) 2016, Nexenta Systems, Inc. All rights reserved.
  * Copyright (c) 2015 by Chunwei Chen. All rights reserved.
  * Copyright (c) 2019 Datto Inc.
- * Copyright (c) 2019, Klara Inc.
+ * Copyright (c) 2019, 2023, Klara Inc.
  * Copyright (c) 2019, Allan Jude
  * Copyright (c) 2022 Hewlett Packard Enterprise Development LP.
  * Copyright (c) 2021, 2022 by Pawel Jakub Dawidek
@@ -276,13 +276,14 @@ dmu_set_bonus(dmu_buf_t *db_fake, int newsize, dmu_tx_t *tx)
 	dnode_t *dn;
 	int error;
 
+	if (newsize < 0 || newsize > db_fake->db_size)
+		return (SET_ERROR(EINVAL));
+
 	DB_DNODE_ENTER(db);
 	dn = DB_DNODE(db);
 
 	if (dn->dn_bonus != db) {
 		error = SET_ERROR(EINVAL);
-	} else if (newsize < 0 || newsize > db_fake->db_size) {
-		error = SET_ERROR(EINVAL);
 	} else {
 		dnode_setbonuslen(dn, newsize, tx);
 		error = 0;
@@ -299,12 +300,13 @@ dmu_set_bonustype(dmu_buf_t *db_fake, dmu_object_type_t type, dmu_tx_t *tx)
 	dnode_t *dn;
 	int error;
 
+	if (!DMU_OT_IS_VALID(type))
+		return (SET_ERROR(EINVAL));
+
 	DB_DNODE_ENTER(db);
 	dn = DB_DNODE(db);
 
-	if (!DMU_OT_IS_VALID(type)) {
-		error = SET_ERROR(EINVAL);
-	} else if (dn->dn_bonus != db) {
+	if (dn->dn_bonus != db) {
 		error = SET_ERROR(EINVAL);
 	} else {
 		dnode_setbonus_type(dn, type, tx);
@@ -319,12 +321,10 @@ dmu_object_type_t
 dmu_get_bonustype(dmu_buf_t *db_fake)
 {
 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
-	dnode_t *dn;
 	dmu_object_type_t type;
 
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	type = dn->dn_bonustype;
+	type = DB_DNODE(db)->dn_bonustype;
 	DB_DNODE_EXIT(db);
 
 	return (type);
@@ -486,7 +486,6 @@ dmu_spill_hold_by_bonus(dmu_buf_t *bonus, uint32_t flags, const void *tag,
     dmu_buf_t **dbp)
 {
 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)bonus;
-	dnode_t *dn;
 	int err;
 	uint32_t db_flags = DB_RF_CANFAIL;
 
@@ -494,8 +493,7 @@ dmu_spill_hold_by_bonus(dmu_buf_t *bonus, uint32_t flags, const void *tag,
 		db_flags |= DB_RF_NO_DECRYPT;
 
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	err = dmu_spill_hold_by_dnode(dn, db_flags, tag, dbp);
+	err = dmu_spill_hold_by_dnode(DB_DNODE(db), db_flags, tag, dbp);
 	DB_DNODE_EXIT(db);
 
 	return (err);
@@ -668,13 +666,11 @@ dmu_buf_hold_array_by_bonus(dmu_buf_t *db_fake, uint64_t offset,
     dmu_buf_t ***dbpp)
 {
 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
-	dnode_t *dn;
 	int err;
 
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,
-	    numbufsp, dbpp, DMU_READ_PREFETCH);
+	err = dmu_buf_hold_array_by_dnode(DB_DNODE(db), offset, length, read,
+	    tag, numbufsp, dbpp, DMU_READ_PREFETCH);
 	DB_DNODE_EXIT(db);
 
 	return (err);
@@ -701,7 +697,7 @@ dmu_buf_rele_array(dmu_buf_t **dbp_fake, int numbufs, const void *tag)
  * Issue prefetch I/Os for the given blocks.  If level is greater than 0, the
  * indirect blocks prefetched will be those that point to the blocks containing
  * the data starting at offset, and continuing to offset + len.  If the range
- * it too long, prefetch the first dmu_prefetch_max bytes as requested, while
+ * is too long, prefetch the first dmu_prefetch_max bytes as requested, while
  * for the rest only a higher level, also fitting within dmu_prefetch_max.  It
  * should primarily help random reads, since for long sequential reads there is
  * a speculative prefetcher.
@@ -777,6 +773,106 @@ dmu_prefetch_by_dnode(dnode_t *dn, int64_t level, uint64_t offset,
 	rw_exit(&dn->dn_struct_rwlock);
 }
 
+typedef struct {
+	kmutex_t	dpa_lock;
+	kcondvar_t	dpa_cv;
+	uint64_t	dpa_pending_io;
+} dmu_prefetch_arg_t;
+
+static void
+dmu_prefetch_done(void *arg, uint64_t level, uint64_t blkid, boolean_t issued)
+{
+	(void) level; (void) blkid; (void)issued;
+	dmu_prefetch_arg_t *dpa = arg;
+
+	ASSERT0(level);
+
+	mutex_enter(&dpa->dpa_lock);
+	ASSERT3U(dpa->dpa_pending_io, >, 0);
+	if (--dpa->dpa_pending_io == 0)
+		cv_broadcast(&dpa->dpa_cv);
+	mutex_exit(&dpa->dpa_lock);
+}
+
+static void
+dmu_prefetch_wait_by_dnode(dnode_t *dn, uint64_t offset, uint64_t len)
+{
+	dmu_prefetch_arg_t dpa;
+
+	mutex_init(&dpa.dpa_lock, NULL, MUTEX_DEFAULT, NULL);
+	cv_init(&dpa.dpa_cv, NULL, CV_DEFAULT, NULL);
+
+	rw_enter(&dn->dn_struct_rwlock, RW_READER);
+
+	uint64_t start = dbuf_whichblock(dn, 0, offset);
+	uint64_t end = dbuf_whichblock(dn, 0, offset + len - 1) + 1;
+	dpa.dpa_pending_io = end - start;
+
+	for (uint64_t blk = start; blk < end; blk++) {
+		(void) dbuf_prefetch_impl(dn, 0, blk, ZIO_PRIORITY_ASYNC_READ,
+		    0, dmu_prefetch_done, &dpa);
+	}
+
+	rw_exit(&dn->dn_struct_rwlock);
+
+	/* wait for prefetch L0 reads to finish */
+	mutex_enter(&dpa.dpa_lock);
+	while (dpa.dpa_pending_io > 0) {
+		cv_wait(&dpa.dpa_cv, &dpa.dpa_lock);
+
+	}
+	mutex_exit(&dpa.dpa_lock);
+
+	mutex_destroy(&dpa.dpa_lock);
+	cv_destroy(&dpa.dpa_cv);
+}
+
+/*
+ * Issue prefetch I/Os for the given L0 block range and wait for the I/O
+ * to complete. This does not enforce dmu_prefetch_max and will prefetch
+ * the entire range. The blocks are read from disk into the ARC but no
+ * decompression occurs (i.e., the dbuf cache is not required).
+ */
+int
+dmu_prefetch_wait(objset_t *os, uint64_t object, uint64_t offset, uint64_t size)
+{
+	dnode_t *dn;
+	int err = 0;
+
+	err = dnode_hold(os, object, FTAG, &dn);
+	if (err != 0)
+		return (err);
+
+	/*
+	 * Chunk the requests (16 indirects worth) so that we can be interrupted
+	 */
+	uint64_t chunksize;
+	if (dn->dn_indblkshift) {
+		uint64_t nbps = bp_span_in_blocks(dn->dn_indblkshift, 1);
+		chunksize = (nbps * 16) << dn->dn_datablkshift;
+	} else {
+		chunksize = dn->dn_datablksz;
+	}
+
+	while (size > 0) {
+		uint64_t mylen = MIN(size, chunksize);
+
+		dmu_prefetch_wait_by_dnode(dn, offset, mylen);
+
+		offset += mylen;
+		size -= mylen;
+
+		if (issig()) {
+			err = SET_ERROR(EINTR);
+			break;
+		}
+	}
+
+	dnode_rele(dn, FTAG);
+
+	return (err);
+}
+
 /*
  * Issue prefetch I/Os for the given object's dnode.
  */
@@ -815,6 +911,13 @@ get_next_chunk(dnode_t *dn, uint64_t *start, uint64_t minimum, uint64_t *l1blks)
 
 	ASSERT3U(minimum, <=, *start);
 
+	/* dn_nlevels == 1 means we don't have any L1 blocks */
+	if (dn->dn_nlevels <= 1) {
+		*l1blks = 0;
+		*start = minimum;
+		return (0);
+	}
+
 	/*
 	 * Check if we can free the entire range assuming that all of the
 	 * L1 blocks in this range have data. If we can, we use this
@@ -1301,15 +1404,13 @@ int
 dmu_read_uio_dbuf(dmu_buf_t *zdb, zfs_uio_t *uio, uint64_t size)
 {
 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)zdb;
-	dnode_t *dn;
 	int err;
 
 	if (size == 0)
 		return (0);
 
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	err = dmu_read_uio_dnode(dn, uio, size);
+	err = dmu_read_uio_dnode(DB_DNODE(db), uio, size);
 	DB_DNODE_EXIT(db);
 
 	return (err);
@@ -1403,15 +1504,13 @@ dmu_write_uio_dbuf(dmu_buf_t *zdb, zfs_uio_t *uio, uint64_t size,
     dmu_tx_t *tx)
 {
 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)zdb;
-	dnode_t *dn;
 	int err;
 
 	if (size == 0)
 		return (0);
 
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	err = dmu_write_uio_dnode(dn, uio, size, tx);
+	err = dmu_write_uio_dnode(DB_DNODE(db), uio, size, tx);
 	DB_DNODE_EXIT(db);
 
 	return (err);
@@ -1444,6 +1543,114 @@ dmu_write_uio(objset_t *os, uint64_t object, zfs_uio_t *uio, uint64_t size,
 }
 #endif /* _KERNEL */
 
+static void
+dmu_cached_bps(spa_t *spa, blkptr_t *bps, uint_t nbps,
+    uint64_t *l1sz, uint64_t *l2sz)
+{
+	int cached_flags;
+
+	if (bps == NULL)
+		return;
+
+	for (size_t blk_off = 0; blk_off < nbps; blk_off++) {
+		blkptr_t *bp = &bps[blk_off];
+
+		if (BP_IS_HOLE(bp))
+			continue;
+
+		cached_flags = arc_cached(spa, bp);
+		if (cached_flags == 0)
+			continue;
+
+		if ((cached_flags & (ARC_CACHED_IN_L1 | ARC_CACHED_IN_L2)) ==
+		    ARC_CACHED_IN_L2)
+			*l2sz += BP_GET_LSIZE(bp);
+		else
+			*l1sz += BP_GET_LSIZE(bp);
+	}
+}
+
+/*
+ * Estimate DMU object cached size.
+ */
+int
+dmu_object_cached_size(objset_t *os, uint64_t object,
+    uint64_t *l1sz, uint64_t *l2sz)
+{
+	dnode_t *dn;
+	dmu_object_info_t doi;
+	int err = 0;
+
+	*l1sz = *l2sz = 0;
+
+	if (dnode_hold(os, object, FTAG, &dn) != 0)
+		return (0);
+
+	if (dn->dn_nlevels < 2) {
+		dnode_rele(dn, FTAG);
+		return (0);
+	}
+
+	dmu_object_info_from_dnode(dn, &doi);
+
+	for (uint64_t off = 0; off < doi.doi_max_offset;
+	    off += dmu_prefetch_max) {
+		/* dbuf_read doesn't prefetch L1 blocks. */
+		dmu_prefetch_by_dnode(dn, 1, off,
+		    dmu_prefetch_max, ZIO_PRIORITY_SYNC_READ);
+	}
+
+	/*
+	 * Hold all valid L1 blocks, asking ARC the status of each BP
+	 * contained in each such L1 block.
+	 */
+	uint_t nbps = bp_span_in_blocks(dn->dn_indblkshift, 1);
+	uint64_t l1blks = 1 + (dn->dn_maxblkid / nbps);
+
+	rw_enter(&dn->dn_struct_rwlock, RW_READER);
+	for (uint64_t blk = 0; blk < l1blks; blk++) {
+		dmu_buf_impl_t *db = NULL;
+
+		if (issig()) {
+			/*
+			 * On interrupt, get out, and bubble up EINTR
+			 */
+			err = EINTR;
+			break;
+		}
+
+		/*
+		 * If we get an i/o error here, the L1 can't be read,
+		 * and nothing under it could be cached, so we just
+		 * continue. Ignoring the error from dbuf_hold_impl
+		 * or from dbuf_read is then a reasonable choice.
+		 */
+		err = dbuf_hold_impl(dn, 1, blk, B_TRUE, B_FALSE, FTAG, &db);
+		if (err != 0) {
+			/*
+			 * ignore error and continue
+			 */
+			err = 0;
+			continue;
+		}
+
+		err = dbuf_read(db, NULL, DB_RF_CANFAIL);
+		if (err == 0) {
+			dmu_cached_bps(dmu_objset_spa(os), db->db.db_data,
+			    nbps, l1sz, l2sz);
+		}
+		/*
+		 * error may be ignored, and we continue
+		 */
+		err = 0;
+		dbuf_rele(db, FTAG);
+	}
+	rw_exit(&dn->dn_struct_rwlock);
+
+	dnode_rele(dn, FTAG);
+	return (err);
+}
+
 /*
  * Allocate a loaned anonymous arc buffer.
  */
@@ -1539,11 +1746,11 @@ dmu_assign_arcbuf_by_dbuf(dmu_buf_t *handle, uint64_t offset, arc_buf_t *buf,
     dmu_tx_t *tx)
 {
 	int err;
-	dmu_buf_impl_t *dbuf = (dmu_buf_impl_t *)handle;
+	dmu_buf_impl_t *db = (dmu_buf_impl_t *)handle;
 
-	DB_DNODE_ENTER(dbuf);
-	err = dmu_assign_arcbuf_by_dnode(DB_DNODE(dbuf), offset, buf, tx);
-	DB_DNODE_EXIT(dbuf);
+	DB_DNODE_ENTER(db);
+	err = dmu_assign_arcbuf_by_dnode(DB_DNODE(db), offset, buf, tx);
+	DB_DNODE_EXIT(db);
 
 	return (err);
 }
@@ -1782,7 +1989,6 @@ dmu_sync(zio_t *pio, uint64_t txg, dmu_sync_cb_t *done, zgd_t *zgd)
 	dmu_sync_arg_t *dsa;
 	zbookmark_phys_t zb;
 	zio_prop_t zp;
-	dnode_t *dn;
 
 	ASSERT(pio != NULL);
 	ASSERT(txg != 0);
@@ -1791,8 +1997,7 @@ dmu_sync(zio_t *pio, uint64_t txg, dmu_sync_cb_t *done, zgd_t *zgd)
 	    db->db.db_object, db->db_level, db->db_blkid);
 
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	dmu_write_policy(os, dn, db->db_level, WP_DMU_SYNC, &zp);
+	dmu_write_policy(os, DB_DNODE(db), db->db_level, WP_DMU_SYNC, &zp);
 	DB_DNODE_EXIT(db);
 
 	/*
@@ -1877,11 +2082,14 @@ dmu_sync(zio_t *pio, uint64_t txg, dmu_sync_cb_t *done, zgd_t *zgd)
 	 * zio_done(), which VERIFYs that the override BP is identical
 	 * to the on-disk BP.
 	 */
-	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	if (dr_next != NULL || dnode_block_freed(dn, db->db_blkid))
+	if (dr_next != NULL) {
 		zp.zp_nopwrite = B_FALSE;
-	DB_DNODE_EXIT(db);
+	} else {
+		DB_DNODE_ENTER(db);
+		if (dnode_block_freed(DB_DNODE(db), db->db_blkid))
+			zp.zp_nopwrite = B_FALSE;
+		DB_DNODE_EXIT(db);
+	}
 
 	ASSERT(dr->dr_txg == txg);
 	if (dr->dt.dl.dr_override_state == DR_IN_DMU_SYNC ||
@@ -2154,6 +2362,7 @@ dmu_write_policy(objset_t *os, dnode_t *dn, int level, int wp, zio_prop_t *zp)
 	memset(zp->zp_mac, 0, ZIO_DATA_MAC_LEN);
 	zp->zp_zpl_smallblk = DMU_OT_IS_FILE(zp->zp_type) ?
 	    os->os_zpl_special_smallblock : 0;
+	zp->zp_storage_type = dn ? dn->dn_storage_type : DMU_OT_NONE;
 
 	ASSERT3U(zp->zp_compress, !=, ZIO_COMPRESS_INHERIT);
 }
@@ -2487,11 +2696,9 @@ void
 dmu_object_dnsize_from_db(dmu_buf_t *db_fake, int *dnsize)
 {
 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
-	dnode_t *dn;
 
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	*dnsize = dn->dn_num_slots << DNODE_SHIFT;
+	*dnsize = DB_DNODE(db)->dn_num_slots << DNODE_SHIFT;
 	DB_DNODE_EXIT(db);
 }
 
diff --git a/sys/contrib/openzfs/module/zfs/dmu_tx.c b/sys/contrib/openzfs/module/zfs/dmu_tx.c
index 8451b5082e86..2c2a6c7642a5 100644
--- a/sys/contrib/openzfs/module/zfs/dmu_tx.c
+++ b/sys/contrib/openzfs/module/zfs/dmu_tx.c
@@ -1520,11 +1520,8 @@ dmu_tx_hold_sa(dmu_tx_t *tx, sa_handle_t *hdl, boolean_t may_grow)
 		ASSERT(tx->tx_txg == 0);
 		dmu_tx_hold_spill(tx, object);
 	} else {
-		dnode_t *dn;
-
 		DB_DNODE_ENTER(db);
-		dn = DB_DNODE(db);
-		if (dn->dn_have_spill) {
+		if (DB_DNODE(db)->dn_have_spill) {
 			ASSERT(tx->tx_txg == 0);
 			dmu_tx_hold_spill(tx, object);
 		}
diff --git a/sys/contrib/openzfs/module/zfs/dnode.c b/sys/contrib/openzfs/module/zfs/dnode.c
index a703fd414f87..ecc6761f8fa4 100644
--- a/sys/contrib/openzfs/module/zfs/dnode.c
+++ b/sys/contrib/openzfs/module/zfs/dnode.c
@@ -306,7 +306,7 @@ dnode_init(void)
 {
 	ASSERT(dnode_cache == NULL);
 	dnode_cache = kmem_cache_create("dnode_t", sizeof (dnode_t),
-	    0, dnode_cons, dnode_dest, NULL, NULL, NULL, 0);
+	    0, dnode_cons, dnode_dest, NULL, NULL, NULL, KMC_RECLAIMABLE);
 	kmem_cache_set_move(dnode_cache, dnode_move);
 
 	wmsum_init(&dnode_sums.dnode_hold_dbuf_hold, 0);
@@ -544,6 +544,17 @@ dnode_setbonus_type(dnode_t *dn, dmu_object_type_t newtype, dmu_tx_t *tx)
 }
 
 void
+dnode_set_storage_type(dnode_t *dn, dmu_object_type_t newtype)
+{
+	/*
+	 * This is not in the dnode_phys, but it should be, and perhaps one day
+	 * will. For now we require it be set after taking a hold.
+	 */
+	ASSERT3U(zfs_refcount_count(&dn->dn_holds), >=, 1);
+	dn->dn_storage_type = newtype;
+}
+
+void
 dnode_rm_spill(dnode_t *dn, dmu_tx_t *tx)
 {
 	ASSERT3U(zfs_refcount_count(&dn->dn_holds), >=, 1);
@@ -604,6 +615,8 @@ dnode_create(objset_t *os, dnode_phys_t *dnp, dmu_buf_impl_t *db,
 	dn->dn_have_spill = ((dnp->dn_flags & DNODE_FLAG_SPILL_BLKPTR) != 0);
 	dn->dn_id_flags = 0;
 
+	dn->dn_storage_type = DMU_OT_NONE;
+
 	dmu_zfetch_init(&dn->dn_zfetch, dn);
 
 	ASSERT(DMU_OT_IS_VALID(dn->dn_phys->dn_type));
@@ -687,6 +700,8 @@ dnode_destroy(dnode_t *dn)
 	dn->dn_newprojid = ZFS_DEFAULT_PROJID;
 	dn->dn_id_flags = 0;
 
+	dn->dn_storage_type = DMU_OT_NONE;
+
 	dmu_zfetch_fini(&dn->dn_zfetch);
 	kmem_cache_free(dnode_cache, dn);
 	arc_space_return(sizeof (dnode_t), ARC_SPACE_DNODE);
@@ -946,6 +961,7 @@ dnode_move_impl(dnode_t *odn, dnode_t *ndn)
 	ndn->dn_newgid = odn->dn_newgid;
 	ndn->dn_newprojid = odn->dn_newprojid;
 	ndn->dn_id_flags = odn->dn_id_flags;
+	ndn->dn_storage_type = odn->dn_storage_type;
 	dmu_zfetch_init(&ndn->dn_zfetch, ndn);
 
 	/*
@@ -1004,6 +1020,7 @@ dnode_move_impl(dnode_t *odn, dnode_t *ndn)
 	odn->dn_newgid = 0;
 	odn->dn_newprojid = ZFS_DEFAULT_PROJID;
 	odn->dn_id_flags = 0;
+	odn->dn_storage_type = DMU_OT_NONE;
 
 	/*
 	 * Mark the dnode.
@@ -1020,6 +1037,19 @@ dnode_move(void *buf, void *newbuf, size_t size, void *arg)
 	int64_t refcount;
 	uint32_t dbufs;
 
+#ifndef USE_DNODE_HANDLE
+	/*
+	 * We can't move dnodes if dbufs reference them directly without
+	 * using handles and respecitve locking.  Unless USE_DNODE_HANDLE
+	 * is defined the code below is only to make sure it still builds,
+	 * but it should never be used, since it is unsafe.
+	 */
+#ifdef ZFS_DEBUG
+	PANIC("dnode_move() called without USE_DNODE_HANDLE");
+#endif
+	return (KMEM_CBRC_NO);
+#endif
+
 	/*
 	 * The dnode is on the objset's list of known dnodes if the objset
 	 * pointer is valid. We set the low bit of the objset pointer when
@@ -1757,7 +1787,7 @@ dnode_rele_and_unlock(dnode_t *dn, const void *tag, boolean_t evicting)
 	 * handle.
 	 */
 #ifdef ZFS_DEBUG
-	ASSERT(refs > 0 || dnh->dnh_zrlock.zr_owner != curthread);
+	ASSERT(refs > 0 || zrl_owner(&dnh->dnh_zrlock) != curthread);
 #endif
 
 	/* NOTE: the DNODE_DNODE does not have a dn_dbuf */
diff --git a/sys/contrib/openzfs/module/zfs/lz4_zfs.c b/sys/contrib/openzfs/module/zfs/lz4_zfs.c
index 820556effb8b..de90c45f2f07 100644
--- a/sys/contrib/openzfs/module/zfs/lz4_zfs.c
+++ b/sys/contrib/openzfs/module/zfs/lz4_zfs.c
@@ -886,7 +886,8 @@ void
 lz4_init(void)
 {
 	lz4_cache = kmem_cache_create("lz4_cache",
-	    sizeof (struct refTables), 0, NULL, NULL, NULL, NULL, NULL, 0);
+	    sizeof (struct refTables), 0, NULL, NULL, NULL, NULL, NULL,
+	    KMC_RECLAIMABLE);
 }
 
 void
diff --git a/sys/contrib/openzfs/module/zfs/sa.c b/sys/contrib/openzfs/module/zfs/sa.c
index 0ae4c331dd36..bc4c9dff31e7 100644
--- a/sys/contrib/openzfs/module/zfs/sa.c
+++ b/sys/contrib/openzfs/module/zfs/sa.c
@@ -236,7 +236,7 @@ sa_cache_init(void)
 {
 	sa_cache = kmem_cache_create("sa_cache",
 	    sizeof (sa_handle_t), 0, sa_cache_constructor,
-	    sa_cache_destructor, NULL, NULL, NULL, 0);
+	    sa_cache_destructor, NULL, NULL, NULL, KMC_RECLAIMABLE);
 }
 
 void
@@ -1501,6 +1501,42 @@ sa_lookup(sa_handle_t *hdl, sa_attr_type_t attr, void *buf, uint32_t buflen)
 	return (error);
 }
 
+/*
+ * Return size of an attribute
+ */
+
+static int
+sa_size_locked(sa_handle_t *hdl, sa_attr_type_t attr, int *size)
+{
+	sa_bulk_attr_t bulk;
+	int error;
+
+	bulk.sa_data = NULL;
+	bulk.sa_attr = attr;
+	bulk.sa_data_func = NULL;
+
+	ASSERT(hdl);
+	ASSERT(MUTEX_HELD(&hdl->sa_lock));
+	if ((error = sa_attr_op(hdl, &bulk, 1, SA_LOOKUP, NULL)) != 0) {
+		return (error);
+	}
+	*size = bulk.sa_size;
+
+	return (0);
+}
+
+int
+sa_size(sa_handle_t *hdl, sa_attr_type_t attr, int *size)
+{
+	int error;
+
+	mutex_enter(&hdl->sa_lock);
+	error = sa_size_locked(hdl, attr, size);
+	mutex_exit(&hdl->sa_lock);
+
+	return (error);
+}
+
 #ifdef _KERNEL
 int
 sa_lookup_uio(sa_handle_t *hdl, sa_attr_type_t attr, zfs_uio_t *uio)
@@ -1542,6 +1578,8 @@ sa_add_projid(sa_handle_t *hdl, dmu_tx_t *tx, uint64_t projid)
 	uint64_t crtime[2], mtime[2], ctime[2], atime[2];
 	zfs_acl_phys_t znode_acl = { 0 };
 	char scanstamp[AV_SCANSTAMP_SZ];
+	char *dxattr_obj = NULL;
+	int dxattr_size = 0;
 
 	if (zp->z_acl_cached == NULL) {
 		zfs_acl_t *aclp;
@@ -1623,6 +1661,17 @@ sa_add_projid(sa_handle_t *hdl, dmu_tx_t *tx, uint64_t projid)
 	if (err != 0 && err != ENOENT)
 		goto out;
 
+	err = sa_size_locked(hdl, SA_ZPL_DXATTR(zfsvfs), &dxattr_size);
+	if (err != 0 && err != ENOENT)
+		goto out;
+	if (dxattr_size != 0) {
+		dxattr_obj = vmem_alloc(dxattr_size, KM_SLEEP);
+		err = sa_lookup_locked(hdl, SA_ZPL_DXATTR(zfsvfs), dxattr_obj,
+		    dxattr_size);
+		if (err != 0 && err != ENOENT)
+			goto out;
+	}
+
 	zp->z_projid = projid;
 	zp->z_pflags |= ZFS_PROJID;
 	links = ZTONLNK(zp);
@@ -1674,6 +1723,11 @@ sa_add_projid(sa_handle_t *hdl, dmu_tx_t *tx, uint64_t projid)
 		zp->z_pflags &= ~ZFS_BONUS_SCANSTAMP;
 	}
 
+	if (dxattr_obj) {
+		SA_ADD_BULK_ATTR(attrs, count, SA_ZPL_DXATTR(zfsvfs),
+		    NULL, dxattr_obj, dxattr_size);
+	}
+
 	VERIFY(dmu_set_bonustype(db, DMU_OT_SA, tx) == 0);
 	VERIFY(sa_replace_all_by_template_locked(hdl, attrs, count, tx) == 0);
 	if (znode_acl.z_acl_extern_obj) {
@@ -1688,6 +1742,8 @@ out:
 	mutex_exit(&hdl->sa_lock);
 	kmem_free(attrs, sizeof (sa_bulk_attr_t) * ZPL_END);
 	kmem_free(bulk, sizeof (sa_bulk_attr_t) * ZPL_END);
+	if (dxattr_obj)
+		vmem_free(dxattr_obj, dxattr_size);
 	return (err);
 }
 #endif
@@ -1852,7 +1908,6 @@ sa_modify_attrs(sa_handle_t *hdl, sa_attr_type_t newattr,
 {
 	sa_os_t *sa = hdl->sa_os->os_sa;
 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)hdl->sa_bonus;
-	dnode_t *dn;
 	sa_bulk_attr_t *attr_desc;
 	void *old_data[2];
 	int bonus_attr_count = 0;
@@ -1872,8 +1927,7 @@ sa_modify_attrs(sa_handle_t *hdl, sa_attr_type_t newattr,
 	/* First make of copy of the old data */
 
 	DB_DNODE_ENTER(db);
-	dn = DB_DNODE(db);
-	if (dn->dn_bonuslen != 0) {
+	if (DB_DNODE(db)->dn_bonuslen != 0) {
 		bonus_data_size = hdl->sa_bonus->db_size;
 		old_data[0] = kmem_alloc(bonus_data_size, KM_SLEEP);
 		memcpy(old_data[0], hdl->sa_bonus->db_data,
@@ -2059,32 +2113,6 @@ sa_update(sa_handle_t *hdl, sa_attr_type_t type,
 	return (error);
 }
 
-/*
- * Return size of an attribute
- */
-
-int
-sa_size(sa_handle_t *hdl, sa_attr_type_t attr, int *size)
-{
-	sa_bulk_attr_t bulk;
-	int error;
-
-	bulk.sa_data = NULL;
-	bulk.sa_attr = attr;
-	bulk.sa_data_func = NULL;
-
-	ASSERT(hdl);
-	mutex_enter(&hdl->sa_lock);
-	if ((error = sa_attr_op(hdl, &bulk, 1, SA_LOOKUP, NULL)) != 0) {
-		mutex_exit(&hdl->sa_lock);
-		return (error);
-	}
-	*size = bulk.sa_size;
-
-	mutex_exit(&hdl->sa_lock);
-	return (0);
-}
-
 int
 sa_bulk_lookup_locked(sa_handle_t *hdl, sa_bulk_attr_t *attrs, int count)
 {
diff --git a/sys/contrib/openzfs/module/zfs/spa.c b/sys/contrib/openzfs/module/zfs/spa.c
index 638572996c3a..cafc7196c354 100644
--- a/sys/contrib/openzfs/module/zfs/spa.c
+++ b/sys/contrib/openzfs/module/zfs/spa.c
@@ -34,7 +34,7 @@
  * Copyright (c) 2017, Intel Corporation.
  * Copyright (c) 2021, Colm Buckley <colm@tuatha.org>
  * Copyright (c) 2023 Hewlett Packard Enterprise Development LP.
- * Copyright (c) 2024, Klara Inc.
+ * Copyright (c) 2023, 2024, Klara Inc.
  */
 
 /*
@@ -337,6 +337,55 @@ spa_prop_add_list(nvlist_t *nvl, zpool_prop_t prop, const char *strval,
 	nvlist_free(propval);
 }
 
+static int
+spa_prop_add(spa_t *spa, const char *propname, nvlist_t *outnvl)
+{
+	zpool_prop_t prop = zpool_name_to_prop(propname);
+	zprop_source_t src = ZPROP_SRC_NONE;
+	uint64_t intval;
+	int err;
+
+	/*
+	 * NB: Not all properties lookups via this API require
+	 * the spa props lock, so they must explicitly grab it here.
+	 */
+	switch (prop) {
+	case ZPOOL_PROP_DEDUPCACHED:
+		err = ddt_get_pool_dedup_cached(spa, &intval);
+		if (err != 0)
+			return (SET_ERROR(err));
+		break;
+	default:
+		return (SET_ERROR(EINVAL));
+	}
+
+	spa_prop_add_list(outnvl, prop, NULL, intval, src);
+
+	return (0);
+}
+
+int
+spa_prop_get_nvlist(spa_t *spa, char **props, unsigned int n_props,
+    nvlist_t **outnvl)
+{
+	int err = 0;
+
+	if (props == NULL)
+		return (0);
+
+	if (*outnvl == NULL) {
+		err = nvlist_alloc(outnvl, NV_UNIQUE_NAME, KM_SLEEP);
+		if (err)
+			return (err);
+	}
+
+	for (unsigned int i = 0; i < n_props && err == 0; i++) {
+		err = spa_prop_add(spa, props[i], *outnvl);
+	}
+
+	return (err);
+}
+
 /*
  * Add a user property (source=src, propname=propval) to an nvlist.
  */
@@ -406,6 +455,9 @@ spa_prop_get_config(spa_t *spa, nvlist_t **nvp)
 		spa_prop_add_list(*nvp, ZPOOL_PROP_BCLONERATIO, NULL,
 		    brt_get_ratio(spa), src);
 
+		spa_prop_add_list(*nvp, ZPOOL_PROP_DEDUP_TABLE_SIZE, NULL,
+		    ddt_get_ddt_dsize(spa), src);
+
 		spa_prop_add_list(*nvp, ZPOOL_PROP_HEALTH, NULL,
 		    rvd->vdev_state, src);
 
@@ -500,9 +552,11 @@ spa_prop_get(spa_t *spa, nvlist_t **nvp)
 	dsl_pool_t *dp;
 	int err;
 
-	err = nvlist_alloc(nvp, NV_UNIQUE_NAME, KM_SLEEP);
-	if (err)
-		return (err);
+	if (*nvp == NULL) {
+		err = nvlist_alloc(nvp, NV_UNIQUE_NAME, KM_SLEEP);
+		if (err)
+			return (err);
+	}
 
 	dp = spa_get_dsl(spa);
 	dsl_pool_config_enter(dp, FTAG);
@@ -672,6 +726,10 @@ spa_prop_validate(spa_t *spa, nvlist_t *props)
 				error = SET_ERROR(EINVAL);
 			break;
 
+		case ZPOOL_PROP_DEDUP_TABLE_QUOTA:
+			error = nvpair_value_uint64(elem, &intval);
+			break;
+
 		case ZPOOL_PROP_DELEGATION:
 		case ZPOOL_PROP_AUTOREPLACE:
 		case ZPOOL_PROP_LISTSNAPS:
@@ -4732,6 +4790,8 @@ spa_ld_get_props(spa_t *spa)
 		spa_prop_find(spa, ZPOOL_PROP_DELEGATION, &spa->spa_delegation);
 		spa_prop_find(spa, ZPOOL_PROP_FAILUREMODE, &spa->spa_failmode);
 		spa_prop_find(spa, ZPOOL_PROP_AUTOEXPAND, &spa->spa_autoexpand);
+		spa_prop_find(spa, ZPOOL_PROP_DEDUP_TABLE_QUOTA,
+		    &spa->spa_dedup_table_quota);
 		spa_prop_find(spa, ZPOOL_PROP_MULTIHOST, &spa->spa_multihost);
 		spa_prop_find(spa, ZPOOL_PROP_AUTOTRIM, &spa->spa_autotrim);
 		spa->spa_autoreplace = (autoreplace != 0);
@@ -6588,6 +6648,8 @@ spa_create(const char *pool, nvlist_t *nvroot, nvlist_t *props,
 	spa->spa_autoexpand = zpool_prop_default_numeric(ZPOOL_PROP_AUTOEXPAND);
 	spa->spa_multihost = zpool_prop_default_numeric(ZPOOL_PROP_MULTIHOST);
 	spa->spa_autotrim = zpool_prop_default_numeric(ZPOOL_PROP_AUTOTRIM);
+	spa->spa_dedup_table_quota =
+	    zpool_prop_default_numeric(ZPOOL_PROP_DEDUP_TABLE_QUOTA);
 
 	if (props != NULL) {
 		spa_configfile_set(spa, props, B_FALSE);
@@ -6755,6 +6817,7 @@ spa_import(char *pool, nvlist_t *config, nvlist_t *props, uint64_t flags)
 		spa_load_spares(spa);
 		spa_config_exit(spa, SCL_ALL, FTAG);
 		spa->spa_spares.sav_sync = B_TRUE;
+		spa->spa_spares.sav_label_sync = B_TRUE;
 	}
 	if (nvlist_lookup_nvlist_array(nvroot, ZPOOL_CONFIG_L2CACHE,
 	    &l2cache, &nl2cache) == 0) {
@@ -6770,6 +6833,7 @@ spa_import(char *pool, nvlist_t *config, nvlist_t *props, uint64_t flags)
 		spa_load_l2cache(spa);
 		spa_config_exit(spa, SCL_ALL, FTAG);
 		spa->spa_l2cache.sav_sync = B_TRUE;
+		spa->spa_l2cache.sav_label_sync = B_TRUE;
 	}
 
 	/*
@@ -9631,6 +9695,9 @@ spa_sync_props(void *arg, dmu_tx_t *tx)
 				case ZPOOL_PROP_MULTIHOST:
 					spa->spa_multihost = intval;
 					break;
+				case ZPOOL_PROP_DEDUP_TABLE_QUOTA:
+					spa->spa_dedup_table_quota = intval;
+					break;
 				default:
 					break;
 				}
diff --git a/sys/contrib/openzfs/module/zfs/spa_misc.c b/sys/contrib/openzfs/module/zfs/spa_misc.c
index d1d41bbe7214..97191e768549 100644
--- a/sys/contrib/openzfs/module/zfs/spa_misc.c
+++ b/sys/contrib/openzfs/module/zfs/spa_misc.c
@@ -1996,13 +1996,31 @@ spa_dedup_class(spa_t *spa)
 	return (spa->spa_dedup_class);
 }
 
+boolean_t
+spa_special_has_ddt(spa_t *spa)
+{
+	return (zfs_ddt_data_is_special &&
+	    spa->spa_special_class->mc_groups != 0);
+}
+
 /*
  * Locate an appropriate allocation class
  */
 metaslab_class_t *
-spa_preferred_class(spa_t *spa, uint64_t size, dmu_object_type_t objtype,
-    uint_t level, uint_t special_smallblk)
+spa_preferred_class(spa_t *spa, const zio_t *zio)
 {
+	const zio_prop_t *zp = &zio->io_prop;
+
+	/*
+	 * Override object type for the purposes of selecting a storage class.
+	 * Primarily for DMU_OTN_ types where we can't explicitly control their
+	 * storage class; instead, choose a static type most closely matches
+	 * what we want.
+	 */
+	dmu_object_type_t objtype =
+	    zp->zp_storage_type == DMU_OT_NONE ?
+	    zp->zp_type : zp->zp_storage_type;
+
 	/*
 	 * ZIL allocations determine their class in zio_alloc_zil().
 	 */
@@ -2020,14 +2038,15 @@ spa_preferred_class(spa_t *spa, uint64_t size, dmu_object_type_t objtype,
 	}
 
 	/* Indirect blocks for user data can land in special if allowed */
-	if (level > 0 && (DMU_OT_IS_FILE(objtype) || objtype == DMU_OT_ZVOL)) {
+	if (zp->zp_level > 0 &&
+	    (DMU_OT_IS_FILE(objtype) || objtype == DMU_OT_ZVOL)) {
 		if (has_special_class && zfs_user_indirect_is_special)
 			return (spa_special_class(spa));
 		else
 			return (spa_normal_class(spa));
 	}
 
-	if (DMU_OT_IS_METADATA(objtype) || level > 0) {
+	if (DMU_OT_IS_METADATA(objtype) || zp->zp_level > 0) {
 		if (has_special_class)
 			return (spa_special_class(spa));
 		else
@@ -2040,7 +2059,7 @@ spa_preferred_class(spa_t *spa, uint64_t size, dmu_object_type_t objtype,
 	 * zfs_special_class_metadata_reserve_pct exclusively for metadata.
 	 */
 	if (DMU_OT_IS_FILE(objtype) &&
-	    has_special_class && size <= special_smallblk) {
+	    has_special_class && zio->io_size <= zp->zp_zpl_smallblk) {
 		metaslab_class_t *special = spa_special_class(spa);
 		uint64_t alloc = metaslab_class_get_alloc(special);
 		uint64_t space = metaslab_class_get_space(special);
diff --git a/sys/contrib/openzfs/module/zfs/vdev.c b/sys/contrib/openzfs/module/zfs/vdev.c
index 11cc39ba3527..6ae0a14127bf 100644
--- a/sys/contrib/openzfs/module/zfs/vdev.c
+++ b/sys/contrib/openzfs/module/zfs/vdev.c
@@ -6222,6 +6222,16 @@ vdev_prop_get(vdev_t *vd, nvlist_t *innvl, nvlist_t *outnvl)
 				    vd->vdev_stat.vs_initialize_errors,
 				    ZPROP_SRC_NONE);
 				continue;
+			case VDEV_PROP_TRIM_ERRORS:
+				vdev_prop_add_list(outnvl, propname, NULL,
+				    vd->vdev_stat.vs_trim_errors,
+				    ZPROP_SRC_NONE);
+				continue;
+			case VDEV_PROP_SLOW_IOS:
+				vdev_prop_add_list(outnvl, propname, NULL,
+				    vd->vdev_stat.vs_slow_ios,
+				    ZPROP_SRC_NONE);
+				continue;
 			case VDEV_PROP_OPS_NULL:
 				vdev_prop_add_list(outnvl, propname, NULL,
 				    vd->vdev_stat.vs_ops[ZIO_TYPE_NULL],
@@ -6306,6 +6316,14 @@ vdev_prop_get(vdev_t *vd, nvlist_t *innvl, nvlist_t *outnvl)
 					    ZPROP_SRC_NONE);
 				}
 				continue;
+			case VDEV_PROP_TRIM_SUPPORT:
+				/* only valid for leaf vdevs */
+				if (vd->vdev_ops->vdev_op_leaf) {
+					vdev_prop_add_list(outnvl, propname,
+					    NULL, vd->vdev_has_trim,
+					    ZPROP_SRC_NONE);
+				}
+				continue;
 			/* Numeric Properites */
 			case VDEV_PROP_ALLOCATING:
 				/* Leaf vdevs cannot have this property */
diff --git a/sys/contrib/openzfs/module/zfs/vdev_label.c b/sys/contrib/openzfs/module/zfs/vdev_label.c
index ed592514fded..47346dd5acff 100644
--- a/sys/contrib/openzfs/module/zfs/vdev_label.c
+++ b/sys/contrib/openzfs/module/zfs/vdev_label.c
@@ -1007,6 +1007,47 @@ vdev_inuse(vdev_t *vd, uint64_t crtxg, vdev_labeltype_t reason,
 	return (state == POOL_STATE_ACTIVE);
 }
 
+static nvlist_t *
+vdev_aux_label_generate(vdev_t *vd, boolean_t reason_spare)
+{
+	/*
+	 * For inactive hot spares and level 2 ARC devices, we generate
+	 * a special label that identifies as a mutually shared hot
+	 * spare or l2cache device. We write the label in case of
+	 * addition or removal of hot spare or l2cache vdev (in which
+	 * case we want to revert the labels).
+	 */
+	nvlist_t *label = fnvlist_alloc();
+	fnvlist_add_uint64(label, ZPOOL_CONFIG_VERSION,
+	    spa_version(vd->vdev_spa));
+	fnvlist_add_uint64(label, ZPOOL_CONFIG_POOL_STATE, reason_spare ?
+	    POOL_STATE_SPARE : POOL_STATE_L2CACHE);
+	fnvlist_add_uint64(label, ZPOOL_CONFIG_GUID, vd->vdev_guid);
+
+	/*
+	 * This is merely to facilitate reporting the ashift of the
+	 * cache device through zdb. The actual retrieval of the
+	 * ashift (in vdev_alloc()) uses the nvlist
+	 * spa->spa_l2cache->sav_config (populated in
+	 * spa_ld_open_aux_vdevs()).
+	 */
+	if (!reason_spare)
+		fnvlist_add_uint64(label, ZPOOL_CONFIG_ASHIFT, vd->vdev_ashift);
+
+	/*
+	 * Add path information to help find it during pool import
+	 */
+	if (vd->vdev_path != NULL)
+		fnvlist_add_string(label, ZPOOL_CONFIG_PATH, vd->vdev_path);
+	if (vd->vdev_devid != NULL)
+		fnvlist_add_string(label, ZPOOL_CONFIG_DEVID, vd->vdev_devid);
+	if (vd->vdev_physpath != NULL) {
+		fnvlist_add_string(label, ZPOOL_CONFIG_PHYS_PATH,
+		    vd->vdev_physpath);
+	}
+	return (label);
+}
+
 /*
  * Initialize a vdev label.  We check to make sure each leaf device is not in
  * use, and writable.  We put down an initial label which we will later
@@ -1121,49 +1162,7 @@ vdev_label_init(vdev_t *vd, uint64_t crtxg, vdev_labeltype_t reason)
 	 * be written again with a meaningful txg by spa_sync().
 	 */
 	if (reason_spare || reason_l2cache) {
-		/*
-		 * For inactive hot spares and level 2 ARC devices, we generate
-		 * a special label that identifies as a mutually shared hot
-		 * spare or l2cache device. We write the label in case of
-		 * addition or removal of hot spare or l2cache vdev (in which
-		 * case we want to revert the labels).
-		 */
-		VERIFY(nvlist_alloc(&label, NV_UNIQUE_NAME, KM_SLEEP) == 0);
-
-		VERIFY(nvlist_add_uint64(label, ZPOOL_CONFIG_VERSION,
-		    spa_version(spa)) == 0);
-		VERIFY(nvlist_add_uint64(label, ZPOOL_CONFIG_POOL_STATE,
-		    reason_spare ? POOL_STATE_SPARE : POOL_STATE_L2CACHE) == 0);
-		VERIFY(nvlist_add_uint64(label, ZPOOL_CONFIG_GUID,
-		    vd->vdev_guid) == 0);
-
-		/*
-		 * This is merely to facilitate reporting the ashift of the
-		 * cache device through zdb. The actual retrieval of the
-		 * ashift (in vdev_alloc()) uses the nvlist
-		 * spa->spa_l2cache->sav_config (populated in
-		 * spa_ld_open_aux_vdevs()).
-		 */
-		if (reason_l2cache) {
-			VERIFY(nvlist_add_uint64(label, ZPOOL_CONFIG_ASHIFT,
-			    vd->vdev_ashift) == 0);
-		}
-
-		/*
-		 * Add path information to help find it during pool import
-		 */
-		if (vd->vdev_path != NULL) {
-			VERIFY(nvlist_add_string(label, ZPOOL_CONFIG_PATH,
-			    vd->vdev_path) == 0);
-		}
-		if (vd->vdev_devid != NULL) {
-			VERIFY(nvlist_add_string(label, ZPOOL_CONFIG_DEVID,
-			    vd->vdev_devid) == 0);
-		}
-		if (vd->vdev_physpath != NULL) {
-			VERIFY(nvlist_add_string(label, ZPOOL_CONFIG_PHYS_PATH,
-			    vd->vdev_physpath) == 0);
-		}
+		label = vdev_aux_label_generate(vd, reason_spare);
 
 		/*
 		 * When spare or l2cache (aux) vdev is added during pool
@@ -1900,6 +1899,8 @@ vdev_label_sync(zio_t *zio, uint64_t *good_writes,
 	abd_t *vp_abd;
 	char *buf;
 	size_t buflen;
+	vdev_t *pvd = vd->vdev_parent;
+	boolean_t spare_in_use = B_FALSE;
 
 	for (int c = 0; c < vd->vdev_children; c++) {
 		vdev_label_sync(zio, good_writes,
@@ -1920,10 +1921,17 @@ vdev_label_sync(zio_t *zio, uint64_t *good_writes,
 	if (vd->vdev_ops == &vdev_draid_spare_ops)
 		return;
 
+	if (pvd && pvd->vdev_ops == &vdev_spare_ops)
+		spare_in_use = B_TRUE;
+
 	/*
 	 * Generate a label describing the top-level config to which we belong.
 	 */
-	label = spa_config_generate(vd->vdev_spa, vd, txg, B_FALSE);
+	if ((vd->vdev_isspare && !spare_in_use) || vd->vdev_isl2cache) {
+		label = vdev_aux_label_generate(vd, vd->vdev_isspare);
+	} else {
+		label = spa_config_generate(vd->vdev_spa, vd, txg, B_FALSE);
+	}
 
 	vp_abd = abd_alloc_linear(sizeof (vdev_phys_t), B_TRUE);
 	abd_zero(vp_abd, sizeof (vdev_phys_t));
@@ -1973,6 +1981,24 @@ vdev_label_sync_list(spa_t *spa, int l, uint64_t txg, int flags)
 		zio_nowait(vio);
 	}
 
+	/*
+	 * AUX path may have changed during import
+	 */
+	spa_aux_vdev_t *sav[2] = {&spa->spa_spares, &spa->spa_l2cache};
+	for (int i = 0; i < 2; i++) {
+		for (int v = 0; v < sav[i]->sav_count; v++) {
+			uint64_t *good_writes;
+			if (!sav[i]->sav_label_sync)
+				continue;
+			good_writes = kmem_zalloc(sizeof (uint64_t), KM_SLEEP);
+			zio_t *vio = zio_null(zio, spa, NULL,
+			    vdev_label_sync_ignore_done, good_writes, flags);
+			vdev_label_sync(vio, good_writes, sav[i]->sav_vdevs[v],
+			    l, txg, flags);
+			zio_nowait(vio);
+		}
+	}
+
 	error = zio_wait(zio);
 
 	/*
@@ -1983,6 +2009,15 @@ vdev_label_sync_list(spa_t *spa, int l, uint64_t txg, int flags)
 	for (vd = list_head(dl); vd != NULL; vd = list_next(dl, vd))
 		zio_flush(zio, vd);
 
+	for (int i = 0; i < 2; i++) {
+		if (!sav[i]->sav_label_sync)
+			continue;
+		for (int v = 0; v < sav[i]->sav_count; v++)
+			zio_flush(zio, sav[i]->sav_vdevs[v]);
+		if (l == 1)
+			sav[i]->sav_label_sync = B_FALSE;
+	}
+
 	(void) zio_wait(zio);
 
 	return (error);
diff --git a/sys/contrib/openzfs/module/zfs/zap_micro.c b/sys/contrib/openzfs/module/zfs/zap_micro.c
index d806988af96d..0d6533b0e131 100644
--- a/sys/contrib/openzfs/module/zfs/zap_micro.c
+++ b/sys/contrib/openzfs/module/zfs/zap_micro.c
@@ -1073,6 +1073,21 @@ zap_prefetch(objset_t *os, uint64_t zapobj, const char *name)
 }
 
 int
+zap_prefetch_object(objset_t *os, uint64_t zapobj)
+{
+	int error;
+	dmu_object_info_t doi;
+
+	error = dmu_object_info(os, zapobj, &doi);
+	if (error == 0 && DMU_OT_BYTESWAP(doi.doi_type) != DMU_BSWAP_ZAP)
+		error = SET_ERROR(EINVAL);
+	if (error == 0)
+		dmu_prefetch_wait(os, zapobj, 0, doi.doi_max_offset);
+
+	return (error);
+}
+
+int
 zap_lookup_by_dnode(dnode_t *dn, const char *name,
     uint64_t integer_size, uint64_t num_integers, void *buf)
 {
@@ -1784,6 +1799,7 @@ EXPORT_SYMBOL(zap_lookup_uint64);
 EXPORT_SYMBOL(zap_contains);
 EXPORT_SYMBOL(zap_prefetch);
 EXPORT_SYMBOL(zap_prefetch_uint64);
+EXPORT_SYMBOL(zap_prefetch_object);
 EXPORT_SYMBOL(zap_add);
 EXPORT_SYMBOL(zap_add_by_dnode);
 EXPORT_SYMBOL(zap_add_uint64);
diff --git a/sys/contrib/openzfs/module/zfs/zfs_ioctl.c b/sys/contrib/openzfs/module/zfs/zfs_ioctl.c
index 7b527eb75e83..897335dd4e4f 100644
--- a/sys/contrib/openzfs/module/zfs/zfs_ioctl.c
+++ b/sys/contrib/openzfs/module/zfs/zfs_ioctl.c
@@ -38,7 +38,7 @@
  * Copyright (c) 2017 Open-E, Inc. All Rights Reserved.
  * Copyright (c) 2019 Datto Inc.
  * Copyright (c) 2019, 2020 by Christian Schwarz. All rights reserved.
- * Copyright (c) 2019, 2021, 2024, Klara Inc.
+ * Copyright (c) 2019, 2021, 2023, 2024, Klara Inc.
  * Copyright (c) 2019, Allan Jude
  * Copyright 2024 Oxide Computer Company
  */
@@ -3009,34 +3009,51 @@ zfs_ioc_pool_set_props(zfs_cmd_t *zc)
 	return (error);
 }
 
+/*
+ * innvl: {
+ *	"get_props_names": [ "prop1", "prop2", ..., "propN" ]
+ * }
+ */
+
+static const zfs_ioc_key_t zfs_keys_get_props[] = {
+	{ ZPOOL_GET_PROPS_NAMES,	DATA_TYPE_STRING_ARRAY,	ZK_OPTIONAL },
+};
+
 static int
-zfs_ioc_pool_get_props(zfs_cmd_t *zc)
+zfs_ioc_pool_get_props(const char *pool, nvlist_t *innvl, nvlist_t *outnvl)
 {
+	nvlist_t *nvp = outnvl;
 	spa_t *spa;
+	char **props = NULL;
+	unsigned int n_props = 0;
 	int error;
-	nvlist_t *nvp = NULL;
 
-	if ((error = spa_open(zc->zc_name, &spa, FTAG)) != 0) {
+	if (nvlist_lookup_string_array(innvl, ZPOOL_GET_PROPS_NAMES,
+	    &props, &n_props) != 0) {
+		props = NULL;
+	}
+
+	if ((error = spa_open(pool, &spa, FTAG)) != 0) {
 		/*
 		 * If the pool is faulted, there may be properties we can still
 		 * get (such as altroot and cachefile), so attempt to get them
 		 * anyway.
 		 */
 		mutex_enter(&spa_namespace_lock);
-		if ((spa = spa_lookup(zc->zc_name)) != NULL)
+		if ((spa = spa_lookup(pool)) != NULL) {
 			error = spa_prop_get(spa, &nvp);
+			if (error == 0 && props != NULL)
+				error = spa_prop_get_nvlist(spa, props, n_props,
+				    &nvp);
+		}
 		mutex_exit(&spa_namespace_lock);
 	} else {
 		error = spa_prop_get(spa, &nvp);
+		if (error == 0 && props != NULL)
+			error = spa_prop_get_nvlist(spa, props, n_props, &nvp);
 		spa_close(spa, FTAG);
 	}
 
-	if (error == 0 && zc->zc_nvlist_dst != 0)
-		error = put_nvlist(zc, nvp);
-	else
-		error = SET_ERROR(EFAULT);
-
-	nvlist_free(nvp);
 	return (error);
 }
 
@@ -4032,6 +4049,52 @@ zfs_ioc_pool_discard_checkpoint(const char *poolname, nvlist_t *innvl,
 }
 
 /*
+ * Loads specific types of data for the given pool
+ *
+ * innvl: {
+ *     "prefetch_type" -> int32_t
+ * }
+ *
+ * outnvl: empty
+ */
+static const zfs_ioc_key_t zfs_keys_pool_prefetch[] = {
+	{ZPOOL_PREFETCH_TYPE,	DATA_TYPE_INT32,	0},
+};
+
+static int
+zfs_ioc_pool_prefetch(const char *poolname, nvlist_t *innvl, nvlist_t *outnvl)
+{
+	(void) outnvl;
+
+	int error;
+	spa_t *spa;
+	int32_t type;
+
+	/*
+	 * Currently, only ZPOOL_PREFETCH_DDT is supported
+	 */
+	if (nvlist_lookup_int32(innvl, ZPOOL_PREFETCH_TYPE, &type) != 0 ||
+	    type != ZPOOL_PREFETCH_DDT) {
+		return (EINVAL);
+	}
+
+	error = spa_open(poolname, &spa, FTAG);
+	if (error != 0)
+		return (error);
+
+	hrtime_t start_time = gethrtime();
+
+	ddt_prefetch_all(spa);
+
+	zfs_dbgmsg("pool '%s': loaded ddt into ARC in %llu ms", spa->spa_name,
+	    (u_longlong_t)NSEC2MSEC(gethrtime() - start_time));
+
+	spa_close(spa, FTAG);
+
+	return (error);
+}
+
+/*
  * inputs:
  * zc_name		name of dataset to destroy
  * zc_defer_destroy	mark for deferred destroy
@@ -7283,6 +7346,12 @@ zfs_ioctl_init(void)
 	    zfs_keys_pool_discard_checkpoint,
 	    ARRAY_SIZE(zfs_keys_pool_discard_checkpoint));
 
+	zfs_ioctl_register("zpool_prefetch",
+	    ZFS_IOC_POOL_PREFETCH, zfs_ioc_pool_prefetch,
+	    zfs_secpolicy_config, POOL_NAME,
+	    POOL_CHECK_SUSPENDED, B_TRUE, B_TRUE,
+	    zfs_keys_pool_prefetch, ARRAY_SIZE(zfs_keys_pool_prefetch));
+
 	zfs_ioctl_register("initialize", ZFS_IOC_POOL_INITIALIZE,
 	    zfs_ioc_pool_initialize, zfs_secpolicy_config, POOL_NAME,
 	    POOL_CHECK_SUSPENDED | POOL_CHECK_READONLY, B_TRUE, B_TRUE,
@@ -7328,6 +7397,11 @@ zfs_ioctl_init(void)
 	    POOL_CHECK_NONE, B_TRUE, B_TRUE,
 	    zfs_keys_pool_scrub, ARRAY_SIZE(zfs_keys_pool_scrub));
 
+	zfs_ioctl_register("get_props", ZFS_IOC_POOL_GET_PROPS,
+	    zfs_ioc_pool_get_props, zfs_secpolicy_read, POOL_NAME,
+	    POOL_CHECK_NONE, B_FALSE, B_FALSE,
+	    zfs_keys_get_props, ARRAY_SIZE(zfs_keys_get_props));
+
 	/* IOCTLS that use the legacy function signature */
 
 	zfs_ioctl_register_legacy(ZFS_IOC_POOL_FREEZE, zfs_ioc_pool_freeze,
@@ -7383,8 +7457,6 @@ zfs_ioctl_init(void)
 
 	zfs_ioctl_register_pool(ZFS_IOC_POOL_STATS, zfs_ioc_pool_stats,
 	    zfs_secpolicy_read, B_FALSE, POOL_CHECK_NONE);
-	zfs_ioctl_register_pool(ZFS_IOC_POOL_GET_PROPS, zfs_ioc_pool_get_props,
-	    zfs_secpolicy_read, B_FALSE, POOL_CHECK_NONE);
 
 	zfs_ioctl_register_pool(ZFS_IOC_ERROR_LOG, zfs_ioc_error_log,
 	    zfs_secpolicy_inject, B_FALSE, POOL_CHECK_SUSPENDED);
diff --git a/sys/contrib/openzfs/module/zfs/zfs_log.c b/sys/contrib/openzfs/module/zfs/zfs_log.c
index fa4e7093ca46..399f5a0117bb 100644
--- a/sys/contrib/openzfs/module/zfs/zfs_log.c
+++ b/sys/contrib/openzfs/module/zfs/zfs_log.c
@@ -665,13 +665,13 @@ zfs_log_write(zilog_t *zilog, dmu_tx_t *tx, int txtype,
 			DB_DNODE_ENTER(db);
 			err = dmu_read_by_dnode(DB_DNODE(db), off, len, lr + 1,
 			    DMU_READ_NO_PREFETCH);
+			DB_DNODE_EXIT(db);
 			if (err != 0) {
 				zil_itx_destroy(itx);
 				itx = zil_itx_create(txtype, sizeof (*lr));
 				lr = (lr_write_t *)&itx->itx_lr;
 				wr_state = WR_NEED_COPY;
 			}
-			DB_DNODE_EXIT(db);
 		}
 
 		itx->itx_wr_state = wr_state;
diff --git a/sys/contrib/openzfs/module/zfs/zfs_replay.c b/sys/contrib/openzfs/module/zfs/zfs_replay.c
index 2e0af60f6db4..810550161f8b 100644
--- a/sys/contrib/openzfs/module/zfs/zfs_replay.c
+++ b/sys/contrib/openzfs/module/zfs/zfs_replay.c
@@ -439,7 +439,7 @@ zfs_replay_create_acl(void *arg1, void *arg2, boolean_t byteswap)
 bail:
 	if (error == 0 && zp != NULL) {
 #ifdef __FreeBSD__
-		VOP_UNLOCK1(ZTOV(zp));
+		VOP_UNLOCK(ZTOV(zp));
 #endif
 		zrele(zp);
 	}
@@ -595,7 +595,7 @@ zfs_replay_create(void *arg1, void *arg2, boolean_t byteswap)
 out:
 	if (error == 0 && zp != NULL) {
 #ifdef __FreeBSD__
-		VOP_UNLOCK1(ZTOV(zp));
+		VOP_UNLOCK(ZTOV(zp));
 #endif
 		zrele(zp);
 	}
diff --git a/sys/contrib/openzfs/module/zfs/zil.c b/sys/contrib/openzfs/module/zfs/zil.c
index 34be54b337fd..3983da6aa424 100644
--- a/sys/contrib/openzfs/module/zfs/zil.c
+++ b/sys/contrib/openzfs/module/zfs/zil.c
@@ -99,6 +99,9 @@ static uint_t zfs_commit_timeout_pct = 10;
 static zil_kstat_values_t zil_stats = {
 	{ "zil_commit_count",			KSTAT_DATA_UINT64 },
 	{ "zil_commit_writer_count",		KSTAT_DATA_UINT64 },
+	{ "zil_commit_error_count",		KSTAT_DATA_UINT64 },
+	{ "zil_commit_stall_count",		KSTAT_DATA_UINT64 },
+	{ "zil_commit_suspend_count",		KSTAT_DATA_UINT64 },
 	{ "zil_itx_count",			KSTAT_DATA_UINT64 },
 	{ "zil_itx_indirect_count",		KSTAT_DATA_UINT64 },
 	{ "zil_itx_indirect_bytes",		KSTAT_DATA_UINT64 },
@@ -360,6 +363,9 @@ zil_sums_init(zil_sums_t *zs)
 {
 	wmsum_init(&zs->zil_commit_count, 0);
 	wmsum_init(&zs->zil_commit_writer_count, 0);
+	wmsum_init(&zs->zil_commit_error_count, 0);
+	wmsum_init(&zs->zil_commit_stall_count, 0);
+	wmsum_init(&zs->zil_commit_suspend_count, 0);
 	wmsum_init(&zs->zil_itx_count, 0);
 	wmsum_init(&zs->zil_itx_indirect_count, 0);
 	wmsum_init(&zs->zil_itx_indirect_bytes, 0);
@@ -382,6 +388,9 @@ zil_sums_fini(zil_sums_t *zs)
 {
 	wmsum_fini(&zs->zil_commit_count);
 	wmsum_fini(&zs->zil_commit_writer_count);
+	wmsum_fini(&zs->zil_commit_error_count);
+	wmsum_fini(&zs->zil_commit_stall_count);
+	wmsum_fini(&zs->zil_commit_suspend_count);
 	wmsum_fini(&zs->zil_itx_count);
 	wmsum_fini(&zs->zil_itx_indirect_count);
 	wmsum_fini(&zs->zil_itx_indirect_bytes);
@@ -406,6 +415,12 @@ zil_kstat_values_update(zil_kstat_values_t *zs, zil_sums_t *zil_sums)
 	    wmsum_value(&zil_sums->zil_commit_count);
 	zs->zil_commit_writer_count.value.ui64 =
 	    wmsum_value(&zil_sums->zil_commit_writer_count);
+	zs->zil_commit_error_count.value.ui64 =
+	    wmsum_value(&zil_sums->zil_commit_error_count);
+	zs->zil_commit_stall_count.value.ui64 =
+	    wmsum_value(&zil_sums->zil_commit_stall_count);
+	zs->zil_commit_suspend_count.value.ui64 =
+	    wmsum_value(&zil_sums->zil_commit_suspend_count);
 	zs->zil_itx_count.value.ui64 =
 	    wmsum_value(&zil_sums->zil_itx_count);
 	zs->zil_itx_indirect_count.value.ui64 =
@@ -512,9 +527,26 @@ zil_parse(zilog_t *zilog, zil_parse_blk_func_t *parse_blk_func,
 
 		for (; lrp < end; lrp += reclen) {
 			lr_t *lr = (lr_t *)lrp;
+
+			/*
+			 * Are the remaining bytes large enough to hold an
+			 * log record?
+			 */
+			if ((char *)(lr + 1) > end) {
+				cmn_err(CE_WARN, "zil_parse: lr_t overrun");
+				error = SET_ERROR(ECKSUM);
+				arc_buf_destroy(abuf, &abuf);
+				goto done;
+			}
 			reclen = lr->lrc_reclen;
-			ASSERT3U(reclen, >=, sizeof (lr_t));
-			ASSERT3U(reclen, <=, end - lrp);
+			if (reclen < sizeof (lr_t) || reclen > end - lrp) {
+				cmn_err(CE_WARN,
+				    "zil_parse: lr_t has an invalid reclen");
+				error = SET_ERROR(ECKSUM);
+				arc_buf_destroy(abuf, &abuf);
+				goto done;
+			}
+
 			if (lr->lrc_seq > claim_lr_seq) {
 				arc_buf_destroy(abuf, &abuf);
 				goto done;
@@ -2823,6 +2855,7 @@ zil_commit_writer_stall(zilog_t *zilog)
 	 * (which is achieved via the txg_wait_synced() call).
 	 */
 	ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
+	ZIL_STAT_BUMP(zilog, zil_commit_stall_count);
 	txg_wait_synced(zilog->zl_dmu_pool, 0);
 	ASSERT(list_is_empty(&zilog->zl_lwb_list));
 }
@@ -3592,6 +3625,7 @@ zil_commit(zilog_t *zilog, uint64_t foid)
 	 * semantics, and avoid calling those functions altogether.
 	 */
 	if (zilog->zl_suspend > 0) {
+		ZIL_STAT_BUMP(zilog, zil_commit_suspend_count);
 		txg_wait_synced(zilog->zl_dmu_pool, 0);
 		return;
 	}
@@ -3645,10 +3679,12 @@ zil_commit_impl(zilog_t *zilog, uint64_t foid)
 		 * implications, but the expectation is for this to be
 		 * an exceptional case, and shouldn't occur often.
 		 */
+		ZIL_STAT_BUMP(zilog, zil_commit_error_count);
 		DTRACE_PROBE2(zil__commit__io__error,
 		    zilog_t *, zilog, zil_commit_waiter_t *, zcw);
 		txg_wait_synced(zilog->zl_dmu_pool, 0);
 	} else if (wtxg != 0) {
+		ZIL_STAT_BUMP(zilog, zil_commit_suspend_count);
 		txg_wait_synced(zilog->zl_dmu_pool, wtxg);
 	}
 
diff --git a/sys/contrib/openzfs/module/zfs/zio.c b/sys/contrib/openzfs/module/zfs/zio.c
index d68d5ababe79..6d08d4bd1633 100644
--- a/sys/contrib/openzfs/module/zfs/zio.c
+++ b/sys/contrib/openzfs/module/zfs/zio.c
@@ -194,6 +194,10 @@ zio_init(void)
 		cflags = (zio_exclude_metadata || size > zio_buf_debug_limit) ?
 		    KMC_NODEBUG : 0;
 		data_cflags = KMC_NODEBUG;
+		if (abd_size_alloc_linear(size)) {
+			cflags |= KMC_RECLAIMABLE;
+			data_cflags |= KMC_RECLAIMABLE;
+		}
 		if (cflags == data_cflags) {
 			/*
 			 * Resulting kmem caches would be identical.
@@ -1101,45 +1105,50 @@ zfs_blkptr_verify(spa_t *spa, const blkptr_t *bp,
 {
 	int errors = 0;
 
-	if (!DMU_OT_IS_VALID(BP_GET_TYPE(bp))) {
+	if (unlikely(!DMU_OT_IS_VALID(BP_GET_TYPE(bp)))) {
 		errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
 		    "blkptr at %px has invalid TYPE %llu",
 		    bp, (longlong_t)BP_GET_TYPE(bp));
 	}
-	if (BP_GET_CHECKSUM(bp) >= ZIO_CHECKSUM_FUNCTIONS) {
-		errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
-		    "blkptr at %px has invalid CHECKSUM %llu",
-		    bp, (longlong_t)BP_GET_CHECKSUM(bp));
-	}
-	if (BP_GET_COMPRESS(bp) >= ZIO_COMPRESS_FUNCTIONS) {
+	if (unlikely(BP_GET_COMPRESS(bp) >= ZIO_COMPRESS_FUNCTIONS)) {
 		errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
 		    "blkptr at %px has invalid COMPRESS %llu",
 		    bp, (longlong_t)BP_GET_COMPRESS(bp));
 	}
-	if (BP_GET_LSIZE(bp) > SPA_MAXBLOCKSIZE) {
+	if (unlikely(BP_GET_LSIZE(bp) > SPA_MAXBLOCKSIZE)) {
 		errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
 		    "blkptr at %px has invalid LSIZE %llu",
 		    bp, (longlong_t)BP_GET_LSIZE(bp));
 	}
-	if (BP_GET_PSIZE(bp) > SPA_MAXBLOCKSIZE) {
-		errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
-		    "blkptr at %px has invalid PSIZE %llu",
-		    bp, (longlong_t)BP_GET_PSIZE(bp));
-	}
-
 	if (BP_IS_EMBEDDED(bp)) {
-		if (BPE_GET_ETYPE(bp) >= NUM_BP_EMBEDDED_TYPES) {
+		if (unlikely(BPE_GET_ETYPE(bp) >= NUM_BP_EMBEDDED_TYPES)) {
 			errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
 			    "blkptr at %px has invalid ETYPE %llu",
 			    bp, (longlong_t)BPE_GET_ETYPE(bp));
 		}
+		if (unlikely(BPE_GET_PSIZE(bp) > BPE_PAYLOAD_SIZE)) {
+			errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
+			    "blkptr at %px has invalid PSIZE %llu",
+			    bp, (longlong_t)BPE_GET_PSIZE(bp));
+		}
+		return (errors == 0);
+	}
+	if (unlikely(BP_GET_CHECKSUM(bp) >= ZIO_CHECKSUM_FUNCTIONS)) {
+		errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
+		    "blkptr at %px has invalid CHECKSUM %llu",
+		    bp, (longlong_t)BP_GET_CHECKSUM(bp));
+	}
+	if (unlikely(BP_GET_PSIZE(bp) > SPA_MAXBLOCKSIZE)) {
+		errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
+		    "blkptr at %px has invalid PSIZE %llu",
+		    bp, (longlong_t)BP_GET_PSIZE(bp));
 	}
 
 	/*
 	 * Do not verify individual DVAs if the config is not trusted. This
 	 * will be done once the zio is executed in vdev_mirror_map_alloc.
 	 */
-	if (!spa->spa_trust_config)
+	if (unlikely(!spa->spa_trust_config))
 		return (errors == 0);
 
 	switch (blk_config) {
@@ -1168,20 +1177,20 @@ zfs_blkptr_verify(spa_t *spa, const blkptr_t *bp,
 		const dva_t *dva = &bp->blk_dva[i];
 		uint64_t vdevid = DVA_GET_VDEV(dva);
 
-		if (vdevid >= spa->spa_root_vdev->vdev_children) {
+		if (unlikely(vdevid >= spa->spa_root_vdev->vdev_children)) {
 			errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
 			    "blkptr at %px DVA %u has invalid VDEV %llu",
 			    bp, i, (longlong_t)vdevid);
 			continue;
 		}
 		vdev_t *vd = spa->spa_root_vdev->vdev_child[vdevid];
-		if (vd == NULL) {
+		if (unlikely(vd == NULL)) {
 			errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
 			    "blkptr at %px DVA %u has invalid VDEV %llu",
 			    bp, i, (longlong_t)vdevid);
 			continue;
 		}
-		if (vd->vdev_ops == &vdev_hole_ops) {
+		if (unlikely(vd->vdev_ops == &vdev_hole_ops)) {
 			errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
 			    "blkptr at %px DVA %u has hole VDEV %llu",
 			    bp, i, (longlong_t)vdevid);
@@ -1199,7 +1208,7 @@ zfs_blkptr_verify(spa_t *spa, const blkptr_t *bp,
 		uint64_t asize = DVA_GET_ASIZE(dva);
 		if (DVA_GET_GANG(dva))
 			asize = vdev_gang_header_asize(vd);
-		if (offset + asize > vd->vdev_asize) {
+		if (unlikely(offset + asize > vd->vdev_asize)) {
 			errors += zfs_blkptr_verify_log(spa, bp, blk_verify,
 			    "blkptr at %px DVA %u has invalid OFFSET %llu",
 			    bp, i, (longlong_t)offset);
@@ -1850,8 +1859,13 @@ zio_write_compress(zio_t *zio)
 	if (compress != ZIO_COMPRESS_OFF &&
 	    !(zio->io_flags & ZIO_FLAG_RAW_COMPRESS)) {
 		void *cbuf = NULL;
-		psize = zio_compress_data(compress, zio->io_abd, &cbuf, lsize,
-		    zp->zp_complevel);
+		if (abd_cmp_zero(zio->io_abd, lsize) == 0)
+			psize = 0;
+		else if (compress == ZIO_COMPRESS_EMPTY)
+			psize = lsize;
+		else
+			psize = zio_compress_data(compress, zio->io_abd, &cbuf,
+			    lsize, zp->zp_complevel);
 		if (psize == 0) {
 			compress = ZIO_COMPRESS_OFF;
 		} else if (psize >= lsize) {
@@ -1915,10 +1929,12 @@ zio_write_compress(zio_t *zio)
 		 * receive, we must check whether the block can be compressed
 		 * to a hole.
 		 */
-		psize = zio_compress_data(ZIO_COMPRESS_EMPTY,
-		    zio->io_abd, NULL, lsize, zp->zp_complevel);
-		if (psize == 0 || psize >= lsize)
+		if (abd_cmp_zero(zio->io_abd, lsize) == 0) {
+			psize = 0;
 			compress = ZIO_COMPRESS_OFF;
+		} else {
+			psize = lsize;
+		}
 	} else if (zio->io_flags & ZIO_FLAG_RAW_COMPRESS &&
 	    !(zio->io_flags & ZIO_FLAG_RAW_ENCRYPT)) {
 		/*
@@ -3069,7 +3085,7 @@ zio_write_gang_block(zio_t *pio, metaslab_class_t *mc)
 		zp.zp_checksum = gio->io_prop.zp_checksum;
 		zp.zp_compress = ZIO_COMPRESS_OFF;
 		zp.zp_complevel = gio->io_prop.zp_complevel;
-		zp.zp_type = DMU_OT_NONE;
+		zp.zp_type = zp.zp_storage_type = DMU_OT_NONE;
 		zp.zp_level = 0;
 		zp.zp_copies = gio->io_prop.zp_copies;
 		zp.zp_dedup = B_FALSE;
@@ -3503,6 +3519,15 @@ zio_ddt_write(zio_t *zio)
 
 	ddt_enter(ddt);
 	dde = ddt_lookup(ddt, bp, B_TRUE);
+	if (dde == NULL) {
+		/* DDT size is over its quota so no new entries */
+		zp->zp_dedup = B_FALSE;
+		BP_SET_DEDUP(bp, B_FALSE);
+		if (zio->io_bp_override == NULL)
+			zio->io_pipeline = ZIO_WRITE_PIPELINE;
+		ddt_exit(ddt);
+		return (zio);
+	}
 	ddp = &dde->dde_phys[p];
 
 	if (zp->zp_dedup_verify && zio_ddt_collision(zio, ddt, dde)) {
@@ -3628,8 +3653,7 @@ zio_dva_throttle(zio_t *zio)
 	metaslab_class_t *mc;
 
 	/* locate an appropriate allocation class */
-	mc = spa_preferred_class(spa, zio->io_size, zio->io_prop.zp_type,
-	    zio->io_prop.zp_level, zio->io_prop.zp_zpl_smallblk);
+	mc = spa_preferred_class(spa, zio);
 
 	if (zio->io_priority == ZIO_PRIORITY_SYNC_WRITE ||
 	    !mc->mc_alloc_throttle_enabled ||
@@ -3701,9 +3725,7 @@ zio_dva_allocate(zio_t *zio)
 	 */
 	mc = zio->io_metaslab_class;
 	if (mc == NULL) {
-		mc = spa_preferred_class(spa, zio->io_size,
-		    zio->io_prop.zp_type, zio->io_prop.zp_level,
-		    zio->io_prop.zp_zpl_smallblk);
+		mc = spa_preferred_class(spa, zio);
 		zio->io_metaslab_class = mc;
 	}
 
@@ -3728,6 +3750,26 @@ zio_dva_allocate(zio_t *zio)
 	 */
 	if (error == ENOSPC && mc != spa_normal_class(spa)) {
 		/*
+		 * When the dedup or special class is spilling into the  normal
+		 * class, there can still be significant space available due
+		 * to deferred frees that are in-flight.  We track the txg when
+		 * this occurred and back off adding new DDT entries for a few
+		 * txgs to allow the free blocks to be processed.
+		 */
+		if ((mc == spa_dedup_class(spa) || (spa_special_has_ddt(spa) &&
+		    mc == spa_special_class(spa))) &&
+		    spa->spa_dedup_class_full_txg != zio->io_txg) {
+			spa->spa_dedup_class_full_txg = zio->io_txg;
+			zfs_dbgmsg("%s[%d]: %s class spilling, req size %d, "
+			    "%llu allocated of %llu",
+			    spa_name(spa), (int)zio->io_txg,
+			    mc == spa_dedup_class(spa) ? "dedup" : "special",
+			    (int)zio->io_size,
+			    (u_longlong_t)metaslab_class_get_alloc(mc),
+			    (u_longlong_t)metaslab_class_get_space(mc));
+		}
+
+		/*
 		 * If throttling, transfer reservation over to normal class.
 		 * The io_allocator slot can remain the same even though we
 		 * are switching classes.
diff --git a/sys/contrib/openzfs/module/zfs/zio_compress.c b/sys/contrib/openzfs/module/zfs/zio_compress.c
index c8a10db7483b..e12d5498ccda 100644
--- a/sys/contrib/openzfs/module/zfs/zio_compress.c
+++ b/sys/contrib/openzfs/module/zfs/zio_compress.c
@@ -111,19 +111,6 @@ zio_compress_select(spa_t *spa, enum zio_compress child,
 	return (result);
 }
 
-static int
-zio_compress_zeroed_cb(void *data, size_t len, void *private)
-{
-	(void) private;
-
-	uint64_t *end = (uint64_t *)((char *)data + len);
-	for (uint64_t *word = (uint64_t *)data; word < end; word++)
-		if (*word != 0)
-			return (1);
-
-	return (0);
-}
-
 size_t
 zio_compress_data(enum zio_compress c, abd_t *src, void **dst, size_t s_len,
     uint8_t level)
@@ -132,18 +119,9 @@ zio_compress_data(enum zio_compress c, abd_t *src, void **dst, size_t s_len,
 	uint8_t complevel;
 	zio_compress_info_t *ci = &zio_compress_table[c];
 
-	ASSERT((uint_t)c < ZIO_COMPRESS_FUNCTIONS);
-	ASSERT((uint_t)c == ZIO_COMPRESS_EMPTY || ci->ci_compress != NULL);
-
-	/*
-	 * If the data is all zeroes, we don't even need to allocate
-	 * a block for it.  We indicate this by returning zero size.
-	 */
-	if (abd_iterate_func(src, 0, s_len, zio_compress_zeroed_cb, NULL) == 0)
-		return (0);
-
-	if (c == ZIO_COMPRESS_EMPTY)
-		return (s_len);
+	ASSERT3U(c, <, ZIO_COMPRESS_FUNCTIONS);
+	ASSERT3U(ci->ci_compress, !=, NULL);
+	ASSERT3U(s_len, >, 0);
 
 	/* Compress at least 12.5% */
 	d_len = s_len - (s_len >> 3);
diff --git a/sys/contrib/openzfs/module/zfs/zvol.c b/sys/contrib/openzfs/module/zfs/zvol.c
index 5b6a3f5cb410..001f774a6d16 100644
--- a/sys/contrib/openzfs/module/zfs/zvol.c
+++ b/sys/contrib/openzfs/module/zfs/zvol.c
@@ -37,6 +37,7 @@
  * Copyright 2014 Nexenta Systems, Inc.  All rights reserved.
  * Copyright (c) 2016 Actifio, Inc. All rights reserved.
  * Copyright (c) 2012, 2019 by Delphix. All rights reserved.
+ * Copyright (c) 2024, Klara, Inc.
  */
 
 /*
@@ -894,6 +895,9 @@ zvol_resume(zvol_state_t *zv)
 	 */
 	atomic_dec(&zv->zv_suspend_ref);
 
+	if (zv->zv_flags & ZVOL_REMOVING)
+		cv_broadcast(&zv->zv_removing_cv);
+
 	return (SET_ERROR(error));
 }
 
@@ -929,6 +933,9 @@ zvol_last_close(zvol_state_t *zv)
 	ASSERT(RW_READ_HELD(&zv->zv_suspend_lock));
 	ASSERT(MUTEX_HELD(&zv->zv_state_lock));
 
+	if (zv->zv_flags & ZVOL_REMOVING)
+		cv_broadcast(&zv->zv_removing_cv);
+
 	zvol_shutdown_zv(zv);
 
 	dmu_objset_disown(zv->zv_objset, 1, zv);
@@ -1221,6 +1228,41 @@ zvol_create_minor(const char *name)
  * Remove minors for specified dataset including children and snapshots.
  */
 
+/*
+ * Remove the minor for a given zvol. This will do it all:
+ *  - flag the zvol for removal, so new requests are rejected
+ *  - wait until outstanding requests are completed
+ *  - remove it from lists
+ *  - free it
+ * It's also usable as a taskq task, and smells nice too.
+ */
+static void
+zvol_remove_minor_task(void *arg)
+{
+	zvol_state_t *zv = (zvol_state_t *)arg;
+
+	ASSERT(!RW_LOCK_HELD(&zvol_state_lock));
+	ASSERT(!MUTEX_HELD(&zv->zv_state_lock));
+
+	mutex_enter(&zv->zv_state_lock);
+	while (zv->zv_open_count > 0 || atomic_read(&zv->zv_suspend_ref)) {
+		zv->zv_flags |= ZVOL_REMOVING;
+		cv_wait(&zv->zv_removing_cv, &zv->zv_state_lock);
+	}
+	mutex_exit(&zv->zv_state_lock);
+
+	rw_enter(&zvol_state_lock, RW_WRITER);
+	mutex_enter(&zv->zv_state_lock);
+
+	zvol_remove(zv);
+	zvol_os_clear_private(zv);
+
+	mutex_exit(&zv->zv_state_lock);
+	rw_exit(&zvol_state_lock);
+
+	zvol_os_free(zv);
+}
+
 static void
 zvol_free_task(void *arg)
 {
@@ -1233,11 +1275,13 @@ zvol_remove_minors_impl(const char *name)
 	zvol_state_t *zv, *zv_next;
 	int namelen = ((name) ? strlen(name) : 0);
 	taskqid_t t;
-	list_t free_list;
+	list_t delay_list, free_list;
 
 	if (zvol_inhibit_dev)
 		return;
 
+	list_create(&delay_list, sizeof (zvol_state_t),
+	    offsetof(zvol_state_t, zv_next));
 	list_create(&free_list, sizeof (zvol_state_t),
 	    offsetof(zvol_state_t, zv_next));
 
@@ -1256,9 +1300,24 @@ zvol_remove_minors_impl(const char *name)
 			 * one is currently using this zv
 			 */
 
-			/* If in use, leave alone */
+			/*
+			 * If in use, try to throw everyone off and try again
+			 * later.
+			 */
 			if (zv->zv_open_count > 0 ||
 			    atomic_read(&zv->zv_suspend_ref)) {
+				zv->zv_flags |= ZVOL_REMOVING;
+				t = taskq_dispatch(
+				    zv->zv_objset->os_spa->spa_zvol_taskq,
+				    zvol_remove_minor_task, zv, TQ_SLEEP);
+				if (t == TASKQID_INVALID) {
+					/*
+					 * Couldn't create the task, so we'll
+					 * do it in place once the loop is
+					 * finished.
+					 */
+					list_insert_head(&delay_list, zv);
+				}
 				mutex_exit(&zv->zv_state_lock);
 				continue;
 			}
@@ -1285,7 +1344,11 @@ zvol_remove_minors_impl(const char *name)
 	}
 	rw_exit(&zvol_state_lock);
 
-	/* Drop zvol_state_lock before calling zvol_free() */
+	/* Wait for zvols that we couldn't create a remove task for */
+	while ((zv = list_remove_head(&delay_list)) != NULL)
+		zvol_remove_minor_task(zv);
+
+	/* Free any that we couldn't free in parallel earlier */
 	while ((zv = list_remove_head(&free_list)) != NULL)
 		zvol_os_free(zv);
 }
@@ -1305,33 +1368,38 @@ zvol_remove_minor_impl(const char *name)
 		zv_next = list_next(&zvol_state_list, zv);
 
 		mutex_enter(&zv->zv_state_lock);
-		if (strcmp(zv->zv_name, name) == 0) {
-			/*
-			 * By holding zv_state_lock here, we guarantee that no
-			 * one is currently using this zv
-			 */
+		if (strcmp(zv->zv_name, name) == 0)
+			/* Found, leave the the loop with zv_lock held */
+			break;
+		mutex_exit(&zv->zv_state_lock);
+	}
 
-			/* If in use, leave alone */
-			if (zv->zv_open_count > 0 ||
-			    atomic_read(&zv->zv_suspend_ref)) {
-				mutex_exit(&zv->zv_state_lock);
-				continue;
-			}
-			zvol_remove(zv);
+	if (zv == NULL) {
+		rw_exit(&zvol_state_lock);
+		return;
+	}
 
-			zvol_os_clear_private(zv);
-			mutex_exit(&zv->zv_state_lock);
-			break;
-		} else {
-			mutex_exit(&zv->zv_state_lock);
-		}
+	ASSERT(MUTEX_HELD(&zv->zv_state_lock));
+
+	if (zv->zv_open_count > 0 || atomic_read(&zv->zv_suspend_ref)) {
+		/*
+		 * In use, so try to throw everyone off, then wait
+		 * until finished.
+		 */
+		zv->zv_flags |= ZVOL_REMOVING;
+		mutex_exit(&zv->zv_state_lock);
+		rw_exit(&zvol_state_lock);
+		zvol_remove_minor_task(zv);
+		return;
 	}
 
-	/* Drop zvol_state_lock before calling zvol_free() */
+	zvol_remove(zv);
+	zvol_os_clear_private(zv);
+
+	mutex_exit(&zv->zv_state_lock);
 	rw_exit(&zvol_state_lock);
 
-	if (zv != NULL)
-		zvol_os_free(zv);
+	zvol_os_free(zv);
 }
 
 /*