xref: /freebsd/sys/contrib/openzfs/module/zfs/dsl_dataset.c (revision e2eeea75eb8b6dd50c1298067a0655880d186734)
1 /*
2  * CDDL HEADER START
3  *
4  * The contents of this file are subject to the terms of the
5  * Common Development and Distribution License (the "License").
6  * You may not use this file except in compliance with the License.
7  *
8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9  * or http://www.opensolaris.org/os/licensing.
10  * See the License for the specific language governing permissions
11  * and limitations under the License.
12  *
13  * When distributing Covered Code, include this CDDL HEADER in each
14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15  * If applicable, add the following below this CDDL HEADER, with the
16  * fields enclosed by brackets "[]" replaced with your own identifying
17  * information: Portions Copyright [yyyy] [name of copyright owner]
18  *
19  * CDDL HEADER END
20  */
21 
22 /*
23  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Copyright (c) 2011, 2020 by Delphix. All rights reserved.
25  * Copyright (c) 2014, Joyent, Inc. All rights reserved.
26  * Copyright (c) 2014 RackTop Systems.
27  * Copyright (c) 2014 Spectra Logic Corporation, All rights reserved.
28  * Copyright (c) 2016 Actifio, Inc. All rights reserved.
29  * Copyright 2016, OmniTI Computer Consulting, Inc. All rights reserved.
30  * Copyright 2017 Nexenta Systems, Inc.
31  * Copyright (c) 2019, Klara Inc.
32  * Copyright (c) 2019, Allan Jude
33  * Copyright (c) 2020 The FreeBSD Foundation [1]
34  *
35  * [1] Portions of this software were developed by Allan Jude
36  *     under sponsorship from the FreeBSD Foundation.
37  */
38 
39 #include <sys/dmu_objset.h>
40 #include <sys/dsl_dataset.h>
41 #include <sys/dsl_dir.h>
42 #include <sys/dsl_prop.h>
43 #include <sys/dsl_synctask.h>
44 #include <sys/dmu_traverse.h>
45 #include <sys/dmu_impl.h>
46 #include <sys/dmu_tx.h>
47 #include <sys/arc.h>
48 #include <sys/zio.h>
49 #include <sys/zap.h>
50 #include <sys/zfeature.h>
51 #include <sys/unique.h>
52 #include <sys/zfs_context.h>
53 #include <sys/zfs_ioctl.h>
54 #include <sys/spa.h>
55 #include <sys/spa_impl.h>
56 #include <sys/vdev.h>
57 #include <sys/zfs_znode.h>
58 #include <sys/zfs_onexit.h>
59 #include <sys/zvol.h>
60 #include <sys/dsl_scan.h>
61 #include <sys/dsl_deadlist.h>
62 #include <sys/dsl_destroy.h>
63 #include <sys/dsl_userhold.h>
64 #include <sys/dsl_bookmark.h>
65 #include <sys/policy.h>
66 #include <sys/dmu_send.h>
67 #include <sys/dmu_recv.h>
68 #include <sys/zio_compress.h>
69 #include <zfs_fletcher.h>
70 #include <sys/zio_checksum.h>
71 
72 /*
73  * The SPA supports block sizes up to 16MB.  However, very large blocks
74  * can have an impact on i/o latency (e.g. tying up a spinning disk for
75  * ~300ms), and also potentially on the memory allocator.  Therefore,
76  * we do not allow the recordsize to be set larger than zfs_max_recordsize
77  * (default 1MB).  Larger blocks can be created by changing this tunable,
78  * and pools with larger blocks can always be imported and used, regardless
79  * of this setting.
80  */
81 int zfs_max_recordsize = 1 * 1024 * 1024;
82 int zfs_allow_redacted_dataset_mount = 0;
83 
84 #define	SWITCH64(x, y) \
85 	{ \
86 		uint64_t __tmp = (x); \
87 		(x) = (y); \
88 		(y) = __tmp; \
89 	}
90 
91 #define	DS_REF_MAX	(1ULL << 62)
92 
93 extern inline dsl_dataset_phys_t *dsl_dataset_phys(dsl_dataset_t *ds);
94 
95 static void dsl_dataset_set_remap_deadlist_object(dsl_dataset_t *ds,
96     uint64_t obj, dmu_tx_t *tx);
97 static void dsl_dataset_unset_remap_deadlist_object(dsl_dataset_t *ds,
98     dmu_tx_t *tx);
99 
100 static void unload_zfeature(dsl_dataset_t *ds, spa_feature_t f);
101 
102 extern int spa_asize_inflation;
103 
104 static zil_header_t zero_zil;
105 
106 /*
107  * Figure out how much of this delta should be propagated to the dsl_dir
108  * layer.  If there's a refreservation, that space has already been
109  * partially accounted for in our ancestors.
110  */
111 static int64_t
112 parent_delta(dsl_dataset_t *ds, int64_t delta)
113 {
114 	dsl_dataset_phys_t *ds_phys;
115 	uint64_t old_bytes, new_bytes;
116 
117 	if (ds->ds_reserved == 0)
118 		return (delta);
119 
120 	ds_phys = dsl_dataset_phys(ds);
121 	old_bytes = MAX(ds_phys->ds_unique_bytes, ds->ds_reserved);
122 	new_bytes = MAX(ds_phys->ds_unique_bytes + delta, ds->ds_reserved);
123 
124 	ASSERT3U(ABS((int64_t)(new_bytes - old_bytes)), <=, ABS(delta));
125 	return (new_bytes - old_bytes);
126 }
127 
128 void
129 dsl_dataset_block_born(dsl_dataset_t *ds, const blkptr_t *bp, dmu_tx_t *tx)
130 {
131 	spa_t *spa = dmu_tx_pool(tx)->dp_spa;
132 	int used = bp_get_dsize_sync(spa, bp);
133 	int compressed = BP_GET_PSIZE(bp);
134 	int uncompressed = BP_GET_UCSIZE(bp);
135 	int64_t delta;
136 	spa_feature_t f;
137 
138 	dprintf_bp(bp, "ds=%p", ds);
139 
140 	ASSERT(dmu_tx_is_syncing(tx));
141 	/* It could have been compressed away to nothing */
142 	if (BP_IS_HOLE(bp) || BP_IS_REDACTED(bp))
143 		return;
144 	ASSERT(BP_GET_TYPE(bp) != DMU_OT_NONE);
145 	ASSERT(DMU_OT_IS_VALID(BP_GET_TYPE(bp)));
146 	if (ds == NULL) {
147 		dsl_pool_mos_diduse_space(tx->tx_pool,
148 		    used, compressed, uncompressed);
149 		return;
150 	}
151 
152 	ASSERT3U(bp->blk_birth, >, dsl_dataset_phys(ds)->ds_prev_snap_txg);
153 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
154 	mutex_enter(&ds->ds_lock);
155 	delta = parent_delta(ds, used);
156 	dsl_dataset_phys(ds)->ds_referenced_bytes += used;
157 	dsl_dataset_phys(ds)->ds_compressed_bytes += compressed;
158 	dsl_dataset_phys(ds)->ds_uncompressed_bytes += uncompressed;
159 	dsl_dataset_phys(ds)->ds_unique_bytes += used;
160 
161 	if (BP_GET_LSIZE(bp) > SPA_OLD_MAXBLOCKSIZE) {
162 		ds->ds_feature_activation[SPA_FEATURE_LARGE_BLOCKS] =
163 		    (void *)B_TRUE;
164 	}
165 
166 
167 	f = zio_checksum_to_feature(BP_GET_CHECKSUM(bp));
168 	if (f != SPA_FEATURE_NONE) {
169 		ASSERT3S(spa_feature_table[f].fi_type, ==,
170 		    ZFEATURE_TYPE_BOOLEAN);
171 		ds->ds_feature_activation[f] = (void *)B_TRUE;
172 	}
173 
174 	f = zio_compress_to_feature(BP_GET_COMPRESS(bp));
175 	if (f != SPA_FEATURE_NONE) {
176 		ASSERT3S(spa_feature_table[f].fi_type, ==,
177 		    ZFEATURE_TYPE_BOOLEAN);
178 		ds->ds_feature_activation[f] = (void *)B_TRUE;
179 	}
180 
181 	/*
182 	 * Track block for livelist, but ignore embedded blocks because
183 	 * they do not need to be freed.
184 	 */
185 	if (dsl_deadlist_is_open(&ds->ds_dir->dd_livelist) &&
186 	    bp->blk_birth > ds->ds_dir->dd_origin_txg &&
187 	    !(BP_IS_EMBEDDED(bp))) {
188 		ASSERT(dsl_dir_is_clone(ds->ds_dir));
189 		ASSERT(spa_feature_is_enabled(spa,
190 		    SPA_FEATURE_LIVELIST));
191 		bplist_append(&ds->ds_dir->dd_pending_allocs, bp);
192 	}
193 
194 	mutex_exit(&ds->ds_lock);
195 	dsl_dir_diduse_space(ds->ds_dir, DD_USED_HEAD, delta,
196 	    compressed, uncompressed, tx);
197 	dsl_dir_transfer_space(ds->ds_dir, used - delta,
198 	    DD_USED_REFRSRV, DD_USED_HEAD, tx);
199 }
200 
201 /*
202  * Called when the specified segment has been remapped, and is thus no
203  * longer referenced in the head dataset.  The vdev must be indirect.
204  *
205  * If the segment is referenced by a snapshot, put it on the remap deadlist.
206  * Otherwise, add this segment to the obsolete spacemap.
207  */
208 void
209 dsl_dataset_block_remapped(dsl_dataset_t *ds, uint64_t vdev, uint64_t offset,
210     uint64_t size, uint64_t birth, dmu_tx_t *tx)
211 {
212 	spa_t *spa = ds->ds_dir->dd_pool->dp_spa;
213 
214 	ASSERT(dmu_tx_is_syncing(tx));
215 	ASSERT(birth <= tx->tx_txg);
216 	ASSERT(!ds->ds_is_snapshot);
217 
218 	if (birth > dsl_dataset_phys(ds)->ds_prev_snap_txg) {
219 		spa_vdev_indirect_mark_obsolete(spa, vdev, offset, size, tx);
220 	} else {
221 		blkptr_t fakebp;
222 		dva_t *dva = &fakebp.blk_dva[0];
223 
224 		ASSERT(ds != NULL);
225 
226 		mutex_enter(&ds->ds_remap_deadlist_lock);
227 		if (!dsl_dataset_remap_deadlist_exists(ds)) {
228 			dsl_dataset_create_remap_deadlist(ds, tx);
229 		}
230 		mutex_exit(&ds->ds_remap_deadlist_lock);
231 
232 		BP_ZERO(&fakebp);
233 		fakebp.blk_birth = birth;
234 		DVA_SET_VDEV(dva, vdev);
235 		DVA_SET_OFFSET(dva, offset);
236 		DVA_SET_ASIZE(dva, size);
237 		dsl_deadlist_insert(&ds->ds_remap_deadlist, &fakebp, B_FALSE,
238 		    tx);
239 	}
240 }
241 
242 int
243 dsl_dataset_block_kill(dsl_dataset_t *ds, const blkptr_t *bp, dmu_tx_t *tx,
244     boolean_t async)
245 {
246 	spa_t *spa = dmu_tx_pool(tx)->dp_spa;
247 
248 	int used = bp_get_dsize_sync(spa, bp);
249 	int compressed = BP_GET_PSIZE(bp);
250 	int uncompressed = BP_GET_UCSIZE(bp);
251 
252 	if (BP_IS_HOLE(bp) || BP_IS_REDACTED(bp))
253 		return (0);
254 
255 	ASSERT(dmu_tx_is_syncing(tx));
256 	ASSERT(bp->blk_birth <= tx->tx_txg);
257 
258 	if (ds == NULL) {
259 		dsl_free(tx->tx_pool, tx->tx_txg, bp);
260 		dsl_pool_mos_diduse_space(tx->tx_pool,
261 		    -used, -compressed, -uncompressed);
262 		return (used);
263 	}
264 	ASSERT3P(tx->tx_pool, ==, ds->ds_dir->dd_pool);
265 
266 	ASSERT(!ds->ds_is_snapshot);
267 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
268 
269 	/*
270 	 * Track block for livelist, but ignore embedded blocks because
271 	 * they do not need to be freed.
272 	 */
273 	if (dsl_deadlist_is_open(&ds->ds_dir->dd_livelist) &&
274 	    bp->blk_birth > ds->ds_dir->dd_origin_txg &&
275 	    !(BP_IS_EMBEDDED(bp))) {
276 		ASSERT(dsl_dir_is_clone(ds->ds_dir));
277 		ASSERT(spa_feature_is_enabled(spa,
278 		    SPA_FEATURE_LIVELIST));
279 		bplist_append(&ds->ds_dir->dd_pending_frees, bp);
280 	}
281 
282 	if (bp->blk_birth > dsl_dataset_phys(ds)->ds_prev_snap_txg) {
283 		int64_t delta;
284 
285 		dprintf_bp(bp, "freeing ds=%llu", ds->ds_object);
286 		dsl_free(tx->tx_pool, tx->tx_txg, bp);
287 
288 		mutex_enter(&ds->ds_lock);
289 		ASSERT(dsl_dataset_phys(ds)->ds_unique_bytes >= used ||
290 		    !DS_UNIQUE_IS_ACCURATE(ds));
291 		delta = parent_delta(ds, -used);
292 		dsl_dataset_phys(ds)->ds_unique_bytes -= used;
293 		mutex_exit(&ds->ds_lock);
294 		dsl_dir_diduse_space(ds->ds_dir, DD_USED_HEAD,
295 		    delta, -compressed, -uncompressed, tx);
296 		dsl_dir_transfer_space(ds->ds_dir, -used - delta,
297 		    DD_USED_REFRSRV, DD_USED_HEAD, tx);
298 	} else {
299 		dprintf_bp(bp, "putting on dead list: %s", "");
300 		if (async) {
301 			/*
302 			 * We are here as part of zio's write done callback,
303 			 * which means we're a zio interrupt thread.  We can't
304 			 * call dsl_deadlist_insert() now because it may block
305 			 * waiting for I/O.  Instead, put bp on the deferred
306 			 * queue and let dsl_pool_sync() finish the job.
307 			 */
308 			bplist_append(&ds->ds_pending_deadlist, bp);
309 		} else {
310 			dsl_deadlist_insert(&ds->ds_deadlist, bp, B_FALSE, tx);
311 		}
312 		ASSERT3U(ds->ds_prev->ds_object, ==,
313 		    dsl_dataset_phys(ds)->ds_prev_snap_obj);
314 		ASSERT(dsl_dataset_phys(ds->ds_prev)->ds_num_children > 0);
315 		/* if (bp->blk_birth > prev prev snap txg) prev unique += bs */
316 		if (dsl_dataset_phys(ds->ds_prev)->ds_next_snap_obj ==
317 		    ds->ds_object && bp->blk_birth >
318 		    dsl_dataset_phys(ds->ds_prev)->ds_prev_snap_txg) {
319 			dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
320 			mutex_enter(&ds->ds_prev->ds_lock);
321 			dsl_dataset_phys(ds->ds_prev)->ds_unique_bytes += used;
322 			mutex_exit(&ds->ds_prev->ds_lock);
323 		}
324 		if (bp->blk_birth > ds->ds_dir->dd_origin_txg) {
325 			dsl_dir_transfer_space(ds->ds_dir, used,
326 			    DD_USED_HEAD, DD_USED_SNAP, tx);
327 		}
328 	}
329 
330 	dsl_bookmark_block_killed(ds, bp, tx);
331 
332 	mutex_enter(&ds->ds_lock);
333 	ASSERT3U(dsl_dataset_phys(ds)->ds_referenced_bytes, >=, used);
334 	dsl_dataset_phys(ds)->ds_referenced_bytes -= used;
335 	ASSERT3U(dsl_dataset_phys(ds)->ds_compressed_bytes, >=, compressed);
336 	dsl_dataset_phys(ds)->ds_compressed_bytes -= compressed;
337 	ASSERT3U(dsl_dataset_phys(ds)->ds_uncompressed_bytes, >=, uncompressed);
338 	dsl_dataset_phys(ds)->ds_uncompressed_bytes -= uncompressed;
339 	mutex_exit(&ds->ds_lock);
340 
341 	return (used);
342 }
343 
344 struct feature_type_uint64_array_arg {
345 	uint64_t length;
346 	uint64_t *array;
347 };
348 
349 static void
350 unload_zfeature(dsl_dataset_t *ds, spa_feature_t f)
351 {
352 	switch (spa_feature_table[f].fi_type) {
353 	case ZFEATURE_TYPE_BOOLEAN:
354 		break;
355 	case ZFEATURE_TYPE_UINT64_ARRAY:
356 	{
357 		struct feature_type_uint64_array_arg *ftuaa = ds->ds_feature[f];
358 		kmem_free(ftuaa->array, ftuaa->length * sizeof (uint64_t));
359 		kmem_free(ftuaa, sizeof (*ftuaa));
360 		break;
361 	}
362 	default:
363 		panic("Invalid zfeature type %d", spa_feature_table[f].fi_type);
364 	}
365 }
366 
367 static int
368 load_zfeature(objset_t *mos, dsl_dataset_t *ds, spa_feature_t f)
369 {
370 	int err = 0;
371 	switch (spa_feature_table[f].fi_type) {
372 	case ZFEATURE_TYPE_BOOLEAN:
373 		err = zap_contains(mos, ds->ds_object,
374 		    spa_feature_table[f].fi_guid);
375 		if (err == 0) {
376 			ds->ds_feature[f] = (void *)B_TRUE;
377 		} else {
378 			ASSERT3U(err, ==, ENOENT);
379 			err = 0;
380 		}
381 		break;
382 	case ZFEATURE_TYPE_UINT64_ARRAY:
383 	{
384 		uint64_t int_size, num_int;
385 		uint64_t *data;
386 		err = zap_length(mos, ds->ds_object,
387 		    spa_feature_table[f].fi_guid, &int_size, &num_int);
388 		if (err != 0) {
389 			ASSERT3U(err, ==, ENOENT);
390 			err = 0;
391 			break;
392 		}
393 		ASSERT3U(int_size, ==, sizeof (uint64_t));
394 		data = kmem_alloc(int_size * num_int, KM_SLEEP);
395 		VERIFY0(zap_lookup(mos, ds->ds_object,
396 		    spa_feature_table[f].fi_guid, int_size, num_int, data));
397 		struct feature_type_uint64_array_arg *ftuaa =
398 		    kmem_alloc(sizeof (*ftuaa), KM_SLEEP);
399 		ftuaa->length = num_int;
400 		ftuaa->array = data;
401 		ds->ds_feature[f] = ftuaa;
402 		break;
403 	}
404 	default:
405 		panic("Invalid zfeature type %d", spa_feature_table[f].fi_type);
406 	}
407 	return (err);
408 }
409 
410 /*
411  * We have to release the fsid synchronously or we risk that a subsequent
412  * mount of the same dataset will fail to unique_insert the fsid.  This
413  * failure would manifest itself as the fsid of this dataset changing
414  * between mounts which makes NFS clients quite unhappy.
415  */
416 static void
417 dsl_dataset_evict_sync(void *dbu)
418 {
419 	dsl_dataset_t *ds = dbu;
420 
421 	ASSERT(ds->ds_owner == NULL);
422 
423 	unique_remove(ds->ds_fsid_guid);
424 }
425 
426 static void
427 dsl_dataset_evict_async(void *dbu)
428 {
429 	dsl_dataset_t *ds = dbu;
430 
431 	ASSERT(ds->ds_owner == NULL);
432 
433 	ds->ds_dbuf = NULL;
434 
435 	if (ds->ds_objset != NULL)
436 		dmu_objset_evict(ds->ds_objset);
437 
438 	if (ds->ds_prev) {
439 		dsl_dataset_rele(ds->ds_prev, ds);
440 		ds->ds_prev = NULL;
441 	}
442 
443 	dsl_bookmark_fini_ds(ds);
444 
445 	bplist_destroy(&ds->ds_pending_deadlist);
446 	if (dsl_deadlist_is_open(&ds->ds_deadlist))
447 		dsl_deadlist_close(&ds->ds_deadlist);
448 	if (dsl_deadlist_is_open(&ds->ds_remap_deadlist))
449 		dsl_deadlist_close(&ds->ds_remap_deadlist);
450 	if (ds->ds_dir)
451 		dsl_dir_async_rele(ds->ds_dir, ds);
452 
453 	ASSERT(!list_link_active(&ds->ds_synced_link));
454 
455 	for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
456 		if (dsl_dataset_feature_is_active(ds, f))
457 			unload_zfeature(ds, f);
458 	}
459 
460 	list_destroy(&ds->ds_prop_cbs);
461 	mutex_destroy(&ds->ds_lock);
462 	mutex_destroy(&ds->ds_opening_lock);
463 	mutex_destroy(&ds->ds_sendstream_lock);
464 	mutex_destroy(&ds->ds_remap_deadlist_lock);
465 	zfs_refcount_destroy(&ds->ds_longholds);
466 	rrw_destroy(&ds->ds_bp_rwlock);
467 
468 	kmem_free(ds, sizeof (dsl_dataset_t));
469 }
470 
471 int
472 dsl_dataset_get_snapname(dsl_dataset_t *ds)
473 {
474 	dsl_dataset_phys_t *headphys;
475 	int err;
476 	dmu_buf_t *headdbuf;
477 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
478 	objset_t *mos = dp->dp_meta_objset;
479 
480 	if (ds->ds_snapname[0])
481 		return (0);
482 	if (dsl_dataset_phys(ds)->ds_next_snap_obj == 0)
483 		return (0);
484 
485 	err = dmu_bonus_hold(mos, dsl_dir_phys(ds->ds_dir)->dd_head_dataset_obj,
486 	    FTAG, &headdbuf);
487 	if (err != 0)
488 		return (err);
489 	headphys = headdbuf->db_data;
490 	err = zap_value_search(dp->dp_meta_objset,
491 	    headphys->ds_snapnames_zapobj, ds->ds_object, 0, ds->ds_snapname);
492 	if (err != 0 && zfs_recover == B_TRUE) {
493 		err = 0;
494 		(void) snprintf(ds->ds_snapname, sizeof (ds->ds_snapname),
495 		    "SNAPOBJ=%llu-ERR=%d",
496 		    (unsigned long long)ds->ds_object, err);
497 	}
498 	dmu_buf_rele(headdbuf, FTAG);
499 	return (err);
500 }
501 
502 int
503 dsl_dataset_snap_lookup(dsl_dataset_t *ds, const char *name, uint64_t *value)
504 {
505 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
506 	uint64_t snapobj = dsl_dataset_phys(ds)->ds_snapnames_zapobj;
507 	matchtype_t mt = 0;
508 	int err;
509 
510 	if (dsl_dataset_phys(ds)->ds_flags & DS_FLAG_CI_DATASET)
511 		mt = MT_NORMALIZE;
512 
513 	err = zap_lookup_norm(mos, snapobj, name, 8, 1,
514 	    value, mt, NULL, 0, NULL);
515 	if (err == ENOTSUP && (mt & MT_NORMALIZE))
516 		err = zap_lookup(mos, snapobj, name, 8, 1, value);
517 	return (err);
518 }
519 
520 int
521 dsl_dataset_snap_remove(dsl_dataset_t *ds, const char *name, dmu_tx_t *tx,
522     boolean_t adj_cnt)
523 {
524 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
525 	uint64_t snapobj = dsl_dataset_phys(ds)->ds_snapnames_zapobj;
526 	matchtype_t mt = 0;
527 	int err;
528 
529 	dsl_dir_snap_cmtime_update(ds->ds_dir);
530 
531 	if (dsl_dataset_phys(ds)->ds_flags & DS_FLAG_CI_DATASET)
532 		mt = MT_NORMALIZE;
533 
534 	err = zap_remove_norm(mos, snapobj, name, mt, tx);
535 	if (err == ENOTSUP && (mt & MT_NORMALIZE))
536 		err = zap_remove(mos, snapobj, name, tx);
537 
538 	if (err == 0 && adj_cnt)
539 		dsl_fs_ss_count_adjust(ds->ds_dir, -1,
540 		    DD_FIELD_SNAPSHOT_COUNT, tx);
541 
542 	return (err);
543 }
544 
545 boolean_t
546 dsl_dataset_try_add_ref(dsl_pool_t *dp, dsl_dataset_t *ds, void *tag)
547 {
548 	dmu_buf_t *dbuf = ds->ds_dbuf;
549 	boolean_t result = B_FALSE;
550 
551 	if (dbuf != NULL && dmu_buf_try_add_ref(dbuf, dp->dp_meta_objset,
552 	    ds->ds_object, DMU_BONUS_BLKID, tag)) {
553 
554 		if (ds == dmu_buf_get_user(dbuf))
555 			result = B_TRUE;
556 		else
557 			dmu_buf_rele(dbuf, tag);
558 	}
559 
560 	return (result);
561 }
562 
563 int
564 dsl_dataset_hold_obj(dsl_pool_t *dp, uint64_t dsobj, void *tag,
565     dsl_dataset_t **dsp)
566 {
567 	objset_t *mos = dp->dp_meta_objset;
568 	dmu_buf_t *dbuf;
569 	dsl_dataset_t *ds;
570 	int err;
571 	dmu_object_info_t doi;
572 
573 	ASSERT(dsl_pool_config_held(dp));
574 
575 	err = dmu_bonus_hold(mos, dsobj, tag, &dbuf);
576 	if (err != 0)
577 		return (err);
578 
579 	/* Make sure dsobj has the correct object type. */
580 	dmu_object_info_from_db(dbuf, &doi);
581 	if (doi.doi_bonus_type != DMU_OT_DSL_DATASET) {
582 		dmu_buf_rele(dbuf, tag);
583 		return (SET_ERROR(EINVAL));
584 	}
585 
586 	ds = dmu_buf_get_user(dbuf);
587 	if (ds == NULL) {
588 		dsl_dataset_t *winner = NULL;
589 
590 		ds = kmem_zalloc(sizeof (dsl_dataset_t), KM_SLEEP);
591 		ds->ds_dbuf = dbuf;
592 		ds->ds_object = dsobj;
593 		ds->ds_is_snapshot = dsl_dataset_phys(ds)->ds_num_children != 0;
594 		list_link_init(&ds->ds_synced_link);
595 
596 		err = dsl_dir_hold_obj(dp, dsl_dataset_phys(ds)->ds_dir_obj,
597 		    NULL, ds, &ds->ds_dir);
598 		if (err != 0) {
599 			kmem_free(ds, sizeof (dsl_dataset_t));
600 			dmu_buf_rele(dbuf, tag);
601 			return (err);
602 		}
603 
604 		mutex_init(&ds->ds_lock, NULL, MUTEX_DEFAULT, NULL);
605 		mutex_init(&ds->ds_opening_lock, NULL, MUTEX_DEFAULT, NULL);
606 		mutex_init(&ds->ds_sendstream_lock, NULL, MUTEX_DEFAULT, NULL);
607 		mutex_init(&ds->ds_remap_deadlist_lock,
608 		    NULL, MUTEX_DEFAULT, NULL);
609 		rrw_init(&ds->ds_bp_rwlock, B_FALSE);
610 		zfs_refcount_create(&ds->ds_longholds);
611 
612 		bplist_create(&ds->ds_pending_deadlist);
613 
614 		list_create(&ds->ds_sendstreams, sizeof (dmu_sendstatus_t),
615 		    offsetof(dmu_sendstatus_t, dss_link));
616 
617 		list_create(&ds->ds_prop_cbs, sizeof (dsl_prop_cb_record_t),
618 		    offsetof(dsl_prop_cb_record_t, cbr_ds_node));
619 
620 		if (doi.doi_type == DMU_OTN_ZAP_METADATA) {
621 			spa_feature_t f;
622 
623 			for (f = 0; f < SPA_FEATURES; f++) {
624 				if (!(spa_feature_table[f].fi_flags &
625 				    ZFEATURE_FLAG_PER_DATASET))
626 					continue;
627 				err = load_zfeature(mos, ds, f);
628 			}
629 		}
630 
631 		if (!ds->ds_is_snapshot) {
632 			ds->ds_snapname[0] = '\0';
633 			if (dsl_dataset_phys(ds)->ds_prev_snap_obj != 0) {
634 				err = dsl_dataset_hold_obj(dp,
635 				    dsl_dataset_phys(ds)->ds_prev_snap_obj,
636 				    ds, &ds->ds_prev);
637 			}
638 			err = dsl_bookmark_init_ds(ds);
639 		} else {
640 			if (zfs_flags & ZFS_DEBUG_SNAPNAMES)
641 				err = dsl_dataset_get_snapname(ds);
642 			if (err == 0 &&
643 			    dsl_dataset_phys(ds)->ds_userrefs_obj != 0) {
644 				err = zap_count(
645 				    ds->ds_dir->dd_pool->dp_meta_objset,
646 				    dsl_dataset_phys(ds)->ds_userrefs_obj,
647 				    &ds->ds_userrefs);
648 			}
649 		}
650 
651 		if (err == 0 && !ds->ds_is_snapshot) {
652 			err = dsl_prop_get_int_ds(ds,
653 			    zfs_prop_to_name(ZFS_PROP_REFRESERVATION),
654 			    &ds->ds_reserved);
655 			if (err == 0) {
656 				err = dsl_prop_get_int_ds(ds,
657 				    zfs_prop_to_name(ZFS_PROP_REFQUOTA),
658 				    &ds->ds_quota);
659 			}
660 		} else {
661 			ds->ds_reserved = ds->ds_quota = 0;
662 		}
663 
664 		if (err == 0 && ds->ds_dir->dd_crypto_obj != 0 &&
665 		    ds->ds_is_snapshot &&
666 		    zap_contains(mos, dsobj, DS_FIELD_IVSET_GUID) != 0) {
667 			dp->dp_spa->spa_errata =
668 			    ZPOOL_ERRATA_ZOL_8308_ENCRYPTION;
669 		}
670 
671 		dsl_deadlist_open(&ds->ds_deadlist,
672 		    mos, dsl_dataset_phys(ds)->ds_deadlist_obj);
673 		uint64_t remap_deadlist_obj =
674 		    dsl_dataset_get_remap_deadlist_object(ds);
675 		if (remap_deadlist_obj != 0) {
676 			dsl_deadlist_open(&ds->ds_remap_deadlist, mos,
677 			    remap_deadlist_obj);
678 		}
679 
680 		dmu_buf_init_user(&ds->ds_dbu, dsl_dataset_evict_sync,
681 		    dsl_dataset_evict_async, &ds->ds_dbuf);
682 		if (err == 0)
683 			winner = dmu_buf_set_user_ie(dbuf, &ds->ds_dbu);
684 
685 		if (err != 0 || winner != NULL) {
686 			bplist_destroy(&ds->ds_pending_deadlist);
687 			dsl_deadlist_close(&ds->ds_deadlist);
688 			if (dsl_deadlist_is_open(&ds->ds_remap_deadlist))
689 				dsl_deadlist_close(&ds->ds_remap_deadlist);
690 			dsl_bookmark_fini_ds(ds);
691 			if (ds->ds_prev)
692 				dsl_dataset_rele(ds->ds_prev, ds);
693 			dsl_dir_rele(ds->ds_dir, ds);
694 			for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
695 				if (dsl_dataset_feature_is_active(ds, f))
696 					unload_zfeature(ds, f);
697 			}
698 
699 			list_destroy(&ds->ds_prop_cbs);
700 			list_destroy(&ds->ds_sendstreams);
701 			mutex_destroy(&ds->ds_lock);
702 			mutex_destroy(&ds->ds_opening_lock);
703 			mutex_destroy(&ds->ds_sendstream_lock);
704 			mutex_destroy(&ds->ds_remap_deadlist_lock);
705 			zfs_refcount_destroy(&ds->ds_longholds);
706 			rrw_destroy(&ds->ds_bp_rwlock);
707 			kmem_free(ds, sizeof (dsl_dataset_t));
708 			if (err != 0) {
709 				dmu_buf_rele(dbuf, tag);
710 				return (err);
711 			}
712 			ds = winner;
713 		} else {
714 			ds->ds_fsid_guid =
715 			    unique_insert(dsl_dataset_phys(ds)->ds_fsid_guid);
716 			if (ds->ds_fsid_guid !=
717 			    dsl_dataset_phys(ds)->ds_fsid_guid) {
718 				zfs_dbgmsg("ds_fsid_guid changed from "
719 				    "%llx to %llx for pool %s dataset id %llu",
720 				    (long long)
721 				    dsl_dataset_phys(ds)->ds_fsid_guid,
722 				    (long long)ds->ds_fsid_guid,
723 				    spa_name(dp->dp_spa),
724 				    dsobj);
725 			}
726 		}
727 	}
728 
729 	ASSERT3P(ds->ds_dbuf, ==, dbuf);
730 	ASSERT3P(dsl_dataset_phys(ds), ==, dbuf->db_data);
731 	ASSERT(dsl_dataset_phys(ds)->ds_prev_snap_obj != 0 ||
732 	    spa_version(dp->dp_spa) < SPA_VERSION_ORIGIN ||
733 	    dp->dp_origin_snap == NULL || ds == dp->dp_origin_snap);
734 	*dsp = ds;
735 
736 	return (0);
737 }
738 
739 int
740 dsl_dataset_create_key_mapping(dsl_dataset_t *ds)
741 {
742 	dsl_dir_t *dd = ds->ds_dir;
743 
744 	if (dd->dd_crypto_obj == 0)
745 		return (0);
746 
747 	return (spa_keystore_create_mapping(dd->dd_pool->dp_spa,
748 	    ds, ds, &ds->ds_key_mapping));
749 }
750 
751 int
752 dsl_dataset_hold_obj_flags(dsl_pool_t *dp, uint64_t dsobj,
753     ds_hold_flags_t flags, void *tag, dsl_dataset_t **dsp)
754 {
755 	int err;
756 
757 	err = dsl_dataset_hold_obj(dp, dsobj, tag, dsp);
758 	if (err != 0)
759 		return (err);
760 
761 	ASSERT3P(*dsp, !=, NULL);
762 
763 	if (flags & DS_HOLD_FLAG_DECRYPT) {
764 		err = dsl_dataset_create_key_mapping(*dsp);
765 		if (err != 0)
766 			dsl_dataset_rele(*dsp, tag);
767 	}
768 
769 	return (err);
770 }
771 
772 int
773 dsl_dataset_hold_flags(dsl_pool_t *dp, const char *name, ds_hold_flags_t flags,
774     void *tag, dsl_dataset_t **dsp)
775 {
776 	dsl_dir_t *dd;
777 	const char *snapname;
778 	uint64_t obj;
779 	int err = 0;
780 	dsl_dataset_t *ds;
781 
782 	err = dsl_dir_hold(dp, name, FTAG, &dd, &snapname);
783 	if (err != 0)
784 		return (err);
785 
786 	ASSERT(dsl_pool_config_held(dp));
787 	obj = dsl_dir_phys(dd)->dd_head_dataset_obj;
788 	if (obj != 0)
789 		err = dsl_dataset_hold_obj_flags(dp, obj, flags, tag, &ds);
790 	else
791 		err = SET_ERROR(ENOENT);
792 
793 	/* we may be looking for a snapshot */
794 	if (err == 0 && snapname != NULL) {
795 		dsl_dataset_t *snap_ds;
796 
797 		if (*snapname++ != '@') {
798 			dsl_dataset_rele_flags(ds, flags, tag);
799 			dsl_dir_rele(dd, FTAG);
800 			return (SET_ERROR(ENOENT));
801 		}
802 
803 		dprintf("looking for snapshot '%s'\n", snapname);
804 		err = dsl_dataset_snap_lookup(ds, snapname, &obj);
805 		if (err == 0) {
806 			err = dsl_dataset_hold_obj_flags(dp, obj, flags, tag,
807 			    &snap_ds);
808 		}
809 		dsl_dataset_rele_flags(ds, flags, tag);
810 
811 		if (err == 0) {
812 			mutex_enter(&snap_ds->ds_lock);
813 			if (snap_ds->ds_snapname[0] == 0)
814 				(void) strlcpy(snap_ds->ds_snapname, snapname,
815 				    sizeof (snap_ds->ds_snapname));
816 			mutex_exit(&snap_ds->ds_lock);
817 			ds = snap_ds;
818 		}
819 	}
820 	if (err == 0)
821 		*dsp = ds;
822 	dsl_dir_rele(dd, FTAG);
823 	return (err);
824 }
825 
826 int
827 dsl_dataset_hold(dsl_pool_t *dp, const char *name, void *tag,
828     dsl_dataset_t **dsp)
829 {
830 	return (dsl_dataset_hold_flags(dp, name, 0, tag, dsp));
831 }
832 
833 static int
834 dsl_dataset_own_obj_impl(dsl_pool_t *dp, uint64_t dsobj, ds_hold_flags_t flags,
835     void *tag, boolean_t override, dsl_dataset_t **dsp)
836 {
837 	int err = dsl_dataset_hold_obj_flags(dp, dsobj, flags, tag, dsp);
838 	if (err != 0)
839 		return (err);
840 	if (!dsl_dataset_tryown(*dsp, tag, override)) {
841 		dsl_dataset_rele_flags(*dsp, flags, tag);
842 		*dsp = NULL;
843 		return (SET_ERROR(EBUSY));
844 	}
845 	return (0);
846 }
847 
848 
849 int
850 dsl_dataset_own_obj(dsl_pool_t *dp, uint64_t dsobj, ds_hold_flags_t flags,
851     void *tag, dsl_dataset_t **dsp)
852 {
853 	return (dsl_dataset_own_obj_impl(dp, dsobj, flags, tag, B_FALSE, dsp));
854 }
855 
856 int
857 dsl_dataset_own_obj_force(dsl_pool_t *dp, uint64_t dsobj,
858     ds_hold_flags_t flags, void *tag, dsl_dataset_t **dsp)
859 {
860 	return (dsl_dataset_own_obj_impl(dp, dsobj, flags, tag, B_TRUE, dsp));
861 }
862 
863 static int
864 dsl_dataset_own_impl(dsl_pool_t *dp, const char *name, ds_hold_flags_t flags,
865     void *tag, boolean_t override, dsl_dataset_t **dsp)
866 {
867 	int err = dsl_dataset_hold_flags(dp, name, flags, tag, dsp);
868 	if (err != 0)
869 		return (err);
870 	if (!dsl_dataset_tryown(*dsp, tag, override)) {
871 		dsl_dataset_rele_flags(*dsp, flags, tag);
872 		return (SET_ERROR(EBUSY));
873 	}
874 	return (0);
875 }
876 
877 int
878 dsl_dataset_own_force(dsl_pool_t *dp, const char *name, ds_hold_flags_t flags,
879     void *tag, dsl_dataset_t **dsp)
880 {
881 	return (dsl_dataset_own_impl(dp, name, flags, tag, B_TRUE, dsp));
882 }
883 
884 int
885 dsl_dataset_own(dsl_pool_t *dp, const char *name, ds_hold_flags_t flags,
886     void *tag, dsl_dataset_t **dsp)
887 {
888 	return (dsl_dataset_own_impl(dp, name, flags, tag, B_FALSE, dsp));
889 }
890 
891 /*
892  * See the comment above dsl_pool_hold() for details.  In summary, a long
893  * hold is used to prevent destruction of a dataset while the pool hold
894  * is dropped, allowing other concurrent operations (e.g. spa_sync()).
895  *
896  * The dataset and pool must be held when this function is called.  After it
897  * is called, the pool hold may be released while the dataset is still held
898  * and accessed.
899  */
900 void
901 dsl_dataset_long_hold(dsl_dataset_t *ds, void *tag)
902 {
903 	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
904 	(void) zfs_refcount_add(&ds->ds_longholds, tag);
905 }
906 
907 void
908 dsl_dataset_long_rele(dsl_dataset_t *ds, void *tag)
909 {
910 	(void) zfs_refcount_remove(&ds->ds_longholds, tag);
911 }
912 
913 /* Return B_TRUE if there are any long holds on this dataset. */
914 boolean_t
915 dsl_dataset_long_held(dsl_dataset_t *ds)
916 {
917 	return (!zfs_refcount_is_zero(&ds->ds_longholds));
918 }
919 
920 void
921 dsl_dataset_name(dsl_dataset_t *ds, char *name)
922 {
923 	if (ds == NULL) {
924 		(void) strlcpy(name, "mos", ZFS_MAX_DATASET_NAME_LEN);
925 	} else {
926 		dsl_dir_name(ds->ds_dir, name);
927 		VERIFY0(dsl_dataset_get_snapname(ds));
928 		if (ds->ds_snapname[0]) {
929 			VERIFY3U(strlcat(name, "@", ZFS_MAX_DATASET_NAME_LEN),
930 			    <, ZFS_MAX_DATASET_NAME_LEN);
931 			/*
932 			 * We use a "recursive" mutex so that we
933 			 * can call dprintf_ds() with ds_lock held.
934 			 */
935 			if (!MUTEX_HELD(&ds->ds_lock)) {
936 				mutex_enter(&ds->ds_lock);
937 				VERIFY3U(strlcat(name, ds->ds_snapname,
938 				    ZFS_MAX_DATASET_NAME_LEN), <,
939 				    ZFS_MAX_DATASET_NAME_LEN);
940 				mutex_exit(&ds->ds_lock);
941 			} else {
942 				VERIFY3U(strlcat(name, ds->ds_snapname,
943 				    ZFS_MAX_DATASET_NAME_LEN), <,
944 				    ZFS_MAX_DATASET_NAME_LEN);
945 			}
946 		}
947 	}
948 }
949 
950 int
951 dsl_dataset_namelen(dsl_dataset_t *ds)
952 {
953 	VERIFY0(dsl_dataset_get_snapname(ds));
954 	mutex_enter(&ds->ds_lock);
955 	int len = strlen(ds->ds_snapname);
956 	mutex_exit(&ds->ds_lock);
957 	/* add '@' if ds is a snap */
958 	if (len > 0)
959 		len++;
960 	len += dsl_dir_namelen(ds->ds_dir);
961 	return (len);
962 }
963 
964 void
965 dsl_dataset_rele(dsl_dataset_t *ds, void *tag)
966 {
967 	dmu_buf_rele(ds->ds_dbuf, tag);
968 }
969 
970 void
971 dsl_dataset_remove_key_mapping(dsl_dataset_t *ds)
972 {
973 	dsl_dir_t *dd = ds->ds_dir;
974 
975 	if (dd == NULL || dd->dd_crypto_obj == 0)
976 		return;
977 
978 	(void) spa_keystore_remove_mapping(dd->dd_pool->dp_spa,
979 	    ds->ds_object, ds);
980 }
981 
982 void
983 dsl_dataset_rele_flags(dsl_dataset_t *ds, ds_hold_flags_t flags, void *tag)
984 {
985 	if (flags & DS_HOLD_FLAG_DECRYPT)
986 		dsl_dataset_remove_key_mapping(ds);
987 
988 	dsl_dataset_rele(ds, tag);
989 }
990 
991 void
992 dsl_dataset_disown(dsl_dataset_t *ds, ds_hold_flags_t flags, void *tag)
993 {
994 	ASSERT3P(ds->ds_owner, ==, tag);
995 	ASSERT(ds->ds_dbuf != NULL);
996 
997 	mutex_enter(&ds->ds_lock);
998 	ds->ds_owner = NULL;
999 	mutex_exit(&ds->ds_lock);
1000 	dsl_dataset_long_rele(ds, tag);
1001 	dsl_dataset_rele_flags(ds, flags, tag);
1002 }
1003 
1004 boolean_t
1005 dsl_dataset_tryown(dsl_dataset_t *ds, void *tag, boolean_t override)
1006 {
1007 	boolean_t gotit = FALSE;
1008 
1009 	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
1010 	mutex_enter(&ds->ds_lock);
1011 	if (ds->ds_owner == NULL && (override || !(DS_IS_INCONSISTENT(ds) ||
1012 	    (dsl_dataset_feature_is_active(ds,
1013 	    SPA_FEATURE_REDACTED_DATASETS) &&
1014 	    !zfs_allow_redacted_dataset_mount)))) {
1015 		ds->ds_owner = tag;
1016 		dsl_dataset_long_hold(ds, tag);
1017 		gotit = TRUE;
1018 	}
1019 	mutex_exit(&ds->ds_lock);
1020 	return (gotit);
1021 }
1022 
1023 boolean_t
1024 dsl_dataset_has_owner(dsl_dataset_t *ds)
1025 {
1026 	boolean_t rv;
1027 	mutex_enter(&ds->ds_lock);
1028 	rv = (ds->ds_owner != NULL);
1029 	mutex_exit(&ds->ds_lock);
1030 	return (rv);
1031 }
1032 
1033 static boolean_t
1034 zfeature_active(spa_feature_t f, void *arg)
1035 {
1036 	switch (spa_feature_table[f].fi_type) {
1037 	case ZFEATURE_TYPE_BOOLEAN: {
1038 		boolean_t val = (boolean_t)(uintptr_t)arg;
1039 		ASSERT(val == B_FALSE || val == B_TRUE);
1040 		return (val);
1041 	}
1042 	case ZFEATURE_TYPE_UINT64_ARRAY:
1043 		/*
1044 		 * In this case, arg is a uint64_t array.  The feature is active
1045 		 * if the array is non-null.
1046 		 */
1047 		return (arg != NULL);
1048 	default:
1049 		panic("Invalid zfeature type %d", spa_feature_table[f].fi_type);
1050 		return (B_FALSE);
1051 	}
1052 }
1053 
1054 boolean_t
1055 dsl_dataset_feature_is_active(dsl_dataset_t *ds, spa_feature_t f)
1056 {
1057 	return (zfeature_active(f, ds->ds_feature[f]));
1058 }
1059 
1060 /*
1061  * The buffers passed out by this function are references to internal buffers;
1062  * they should not be freed by callers of this function, and they should not be
1063  * used after the dataset has been released.
1064  */
1065 boolean_t
1066 dsl_dataset_get_uint64_array_feature(dsl_dataset_t *ds, spa_feature_t f,
1067     uint64_t *outlength, uint64_t **outp)
1068 {
1069 	VERIFY(spa_feature_table[f].fi_type & ZFEATURE_TYPE_UINT64_ARRAY);
1070 	if (!dsl_dataset_feature_is_active(ds, f)) {
1071 		return (B_FALSE);
1072 	}
1073 	struct feature_type_uint64_array_arg *ftuaa = ds->ds_feature[f];
1074 	*outp = ftuaa->array;
1075 	*outlength = ftuaa->length;
1076 	return (B_TRUE);
1077 }
1078 
1079 void
1080 dsl_dataset_activate_feature(uint64_t dsobj, spa_feature_t f, void *arg,
1081     dmu_tx_t *tx)
1082 {
1083 	spa_t *spa = dmu_tx_pool(tx)->dp_spa;
1084 	objset_t *mos = dmu_tx_pool(tx)->dp_meta_objset;
1085 	uint64_t zero = 0;
1086 
1087 	VERIFY(spa_feature_table[f].fi_flags & ZFEATURE_FLAG_PER_DATASET);
1088 
1089 	spa_feature_incr(spa, f, tx);
1090 	dmu_object_zapify(mos, dsobj, DMU_OT_DSL_DATASET, tx);
1091 
1092 	switch (spa_feature_table[f].fi_type) {
1093 	case ZFEATURE_TYPE_BOOLEAN:
1094 		ASSERT3S((boolean_t)(uintptr_t)arg, ==, B_TRUE);
1095 		VERIFY0(zap_add(mos, dsobj, spa_feature_table[f].fi_guid,
1096 		    sizeof (zero), 1, &zero, tx));
1097 		break;
1098 	case ZFEATURE_TYPE_UINT64_ARRAY:
1099 	{
1100 		struct feature_type_uint64_array_arg *ftuaa = arg;
1101 		VERIFY0(zap_add(mos, dsobj, spa_feature_table[f].fi_guid,
1102 		    sizeof (uint64_t), ftuaa->length, ftuaa->array, tx));
1103 		break;
1104 	}
1105 	default:
1106 		panic("Invalid zfeature type %d", spa_feature_table[f].fi_type);
1107 	}
1108 }
1109 
1110 static void
1111 dsl_dataset_deactivate_feature_impl(dsl_dataset_t *ds, spa_feature_t f,
1112     dmu_tx_t *tx)
1113 {
1114 	spa_t *spa = dmu_tx_pool(tx)->dp_spa;
1115 	objset_t *mos = dmu_tx_pool(tx)->dp_meta_objset;
1116 	uint64_t dsobj = ds->ds_object;
1117 
1118 	VERIFY(spa_feature_table[f].fi_flags & ZFEATURE_FLAG_PER_DATASET);
1119 
1120 	VERIFY0(zap_remove(mos, dsobj, spa_feature_table[f].fi_guid, tx));
1121 	spa_feature_decr(spa, f, tx);
1122 	ds->ds_feature[f] = NULL;
1123 }
1124 
1125 void
1126 dsl_dataset_deactivate_feature(dsl_dataset_t *ds, spa_feature_t f, dmu_tx_t *tx)
1127 {
1128 	unload_zfeature(ds, f);
1129 	dsl_dataset_deactivate_feature_impl(ds, f, tx);
1130 }
1131 
1132 uint64_t
1133 dsl_dataset_create_sync_dd(dsl_dir_t *dd, dsl_dataset_t *origin,
1134     dsl_crypto_params_t *dcp, uint64_t flags, dmu_tx_t *tx)
1135 {
1136 	dsl_pool_t *dp = dd->dd_pool;
1137 	dmu_buf_t *dbuf;
1138 	dsl_dataset_phys_t *dsphys;
1139 	uint64_t dsobj;
1140 	objset_t *mos = dp->dp_meta_objset;
1141 
1142 	if (origin == NULL)
1143 		origin = dp->dp_origin_snap;
1144 
1145 	ASSERT(origin == NULL || origin->ds_dir->dd_pool == dp);
1146 	ASSERT(origin == NULL || dsl_dataset_phys(origin)->ds_num_children > 0);
1147 	ASSERT(dmu_tx_is_syncing(tx));
1148 	ASSERT(dsl_dir_phys(dd)->dd_head_dataset_obj == 0);
1149 
1150 	dsobj = dmu_object_alloc(mos, DMU_OT_DSL_DATASET, 0,
1151 	    DMU_OT_DSL_DATASET, sizeof (dsl_dataset_phys_t), tx);
1152 	VERIFY0(dmu_bonus_hold(mos, dsobj, FTAG, &dbuf));
1153 	dmu_buf_will_dirty(dbuf, tx);
1154 	dsphys = dbuf->db_data;
1155 	bzero(dsphys, sizeof (dsl_dataset_phys_t));
1156 	dsphys->ds_dir_obj = dd->dd_object;
1157 	dsphys->ds_flags = flags;
1158 	dsphys->ds_fsid_guid = unique_create();
1159 	(void) random_get_pseudo_bytes((void*)&dsphys->ds_guid,
1160 	    sizeof (dsphys->ds_guid));
1161 	dsphys->ds_snapnames_zapobj =
1162 	    zap_create_norm(mos, U8_TEXTPREP_TOUPPER, DMU_OT_DSL_DS_SNAP_MAP,
1163 	    DMU_OT_NONE, 0, tx);
1164 	dsphys->ds_creation_time = gethrestime_sec();
1165 	dsphys->ds_creation_txg = tx->tx_txg == TXG_INITIAL ? 1 : tx->tx_txg;
1166 
1167 	if (origin == NULL) {
1168 		dsphys->ds_deadlist_obj = dsl_deadlist_alloc(mos, tx);
1169 	} else {
1170 		dsl_dataset_t *ohds; /* head of the origin snapshot */
1171 
1172 		dsphys->ds_prev_snap_obj = origin->ds_object;
1173 		dsphys->ds_prev_snap_txg =
1174 		    dsl_dataset_phys(origin)->ds_creation_txg;
1175 		dsphys->ds_referenced_bytes =
1176 		    dsl_dataset_phys(origin)->ds_referenced_bytes;
1177 		dsphys->ds_compressed_bytes =
1178 		    dsl_dataset_phys(origin)->ds_compressed_bytes;
1179 		dsphys->ds_uncompressed_bytes =
1180 		    dsl_dataset_phys(origin)->ds_uncompressed_bytes;
1181 		rrw_enter(&origin->ds_bp_rwlock, RW_READER, FTAG);
1182 		dsphys->ds_bp = dsl_dataset_phys(origin)->ds_bp;
1183 		rrw_exit(&origin->ds_bp_rwlock, FTAG);
1184 
1185 		/*
1186 		 * Inherit flags that describe the dataset's contents
1187 		 * (INCONSISTENT) or properties (Case Insensitive).
1188 		 */
1189 		dsphys->ds_flags |= dsl_dataset_phys(origin)->ds_flags &
1190 		    (DS_FLAG_INCONSISTENT | DS_FLAG_CI_DATASET);
1191 
1192 		for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
1193 			if (zfeature_active(f, origin->ds_feature[f])) {
1194 				dsl_dataset_activate_feature(dsobj, f,
1195 				    origin->ds_feature[f], tx);
1196 			}
1197 		}
1198 
1199 		dmu_buf_will_dirty(origin->ds_dbuf, tx);
1200 		dsl_dataset_phys(origin)->ds_num_children++;
1201 
1202 		VERIFY0(dsl_dataset_hold_obj(dp,
1203 		    dsl_dir_phys(origin->ds_dir)->dd_head_dataset_obj,
1204 		    FTAG, &ohds));
1205 		dsphys->ds_deadlist_obj = dsl_deadlist_clone(&ohds->ds_deadlist,
1206 		    dsphys->ds_prev_snap_txg, dsphys->ds_prev_snap_obj, tx);
1207 		dsl_dataset_rele(ohds, FTAG);
1208 
1209 		if (spa_version(dp->dp_spa) >= SPA_VERSION_NEXT_CLONES) {
1210 			if (dsl_dataset_phys(origin)->ds_next_clones_obj == 0) {
1211 				dsl_dataset_phys(origin)->ds_next_clones_obj =
1212 				    zap_create(mos,
1213 				    DMU_OT_NEXT_CLONES, DMU_OT_NONE, 0, tx);
1214 			}
1215 			VERIFY0(zap_add_int(mos,
1216 			    dsl_dataset_phys(origin)->ds_next_clones_obj,
1217 			    dsobj, tx));
1218 		}
1219 
1220 		dmu_buf_will_dirty(dd->dd_dbuf, tx);
1221 		dsl_dir_phys(dd)->dd_origin_obj = origin->ds_object;
1222 		if (spa_version(dp->dp_spa) >= SPA_VERSION_DIR_CLONES) {
1223 			if (dsl_dir_phys(origin->ds_dir)->dd_clones == 0) {
1224 				dmu_buf_will_dirty(origin->ds_dir->dd_dbuf, tx);
1225 				dsl_dir_phys(origin->ds_dir)->dd_clones =
1226 				    zap_create(mos,
1227 				    DMU_OT_DSL_CLONES, DMU_OT_NONE, 0, tx);
1228 			}
1229 			VERIFY0(zap_add_int(mos,
1230 			    dsl_dir_phys(origin->ds_dir)->dd_clones,
1231 			    dsobj, tx));
1232 		}
1233 	}
1234 
1235 	/* handle encryption */
1236 	dsl_dataset_create_crypt_sync(dsobj, dd, origin, dcp, tx);
1237 
1238 	if (spa_version(dp->dp_spa) >= SPA_VERSION_UNIQUE_ACCURATE)
1239 		dsphys->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
1240 
1241 	dmu_buf_rele(dbuf, FTAG);
1242 
1243 	dmu_buf_will_dirty(dd->dd_dbuf, tx);
1244 	dsl_dir_phys(dd)->dd_head_dataset_obj = dsobj;
1245 
1246 	return (dsobj);
1247 }
1248 
1249 static void
1250 dsl_dataset_zero_zil(dsl_dataset_t *ds, dmu_tx_t *tx)
1251 {
1252 	objset_t *os;
1253 
1254 	VERIFY0(dmu_objset_from_ds(ds, &os));
1255 	if (bcmp(&os->os_zil_header, &zero_zil, sizeof (zero_zil)) != 0) {
1256 		dsl_pool_t *dp = ds->ds_dir->dd_pool;
1257 		zio_t *zio;
1258 
1259 		bzero(&os->os_zil_header, sizeof (os->os_zil_header));
1260 		if (os->os_encrypted)
1261 			os->os_next_write_raw[tx->tx_txg & TXG_MASK] = B_TRUE;
1262 
1263 		zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
1264 		dsl_dataset_sync(ds, zio, tx);
1265 		VERIFY0(zio_wait(zio));
1266 
1267 		/* dsl_dataset_sync_done will drop this reference. */
1268 		dmu_buf_add_ref(ds->ds_dbuf, ds);
1269 		dsl_dataset_sync_done(ds, tx);
1270 	}
1271 }
1272 
1273 uint64_t
1274 dsl_dataset_create_sync(dsl_dir_t *pdd, const char *lastname,
1275     dsl_dataset_t *origin, uint64_t flags, cred_t *cr,
1276     dsl_crypto_params_t *dcp, dmu_tx_t *tx)
1277 {
1278 	dsl_pool_t *dp = pdd->dd_pool;
1279 	uint64_t dsobj, ddobj;
1280 	dsl_dir_t *dd;
1281 
1282 	ASSERT(dmu_tx_is_syncing(tx));
1283 	ASSERT(lastname[0] != '@');
1284 	/*
1285 	 * Filesystems will eventually have their origin set to dp_origin_snap,
1286 	 * but that's taken care of in dsl_dataset_create_sync_dd. When
1287 	 * creating a filesystem, this function is called with origin equal to
1288 	 * NULL.
1289 	 */
1290 	if (origin != NULL)
1291 		ASSERT3P(origin, !=, dp->dp_origin_snap);
1292 
1293 	ddobj = dsl_dir_create_sync(dp, pdd, lastname, tx);
1294 	VERIFY0(dsl_dir_hold_obj(dp, ddobj, lastname, FTAG, &dd));
1295 
1296 	dsobj = dsl_dataset_create_sync_dd(dd, origin, dcp,
1297 	    flags & ~DS_CREATE_FLAG_NODIRTY, tx);
1298 
1299 	dsl_deleg_set_create_perms(dd, tx, cr);
1300 
1301 	/*
1302 	 * If we are creating a clone and the livelist feature is enabled,
1303 	 * add the entry DD_FIELD_LIVELIST to ZAP.
1304 	 */
1305 	if (origin != NULL &&
1306 	    spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_LIVELIST)) {
1307 		objset_t *mos = dd->dd_pool->dp_meta_objset;
1308 		dsl_dir_zapify(dd, tx);
1309 		uint64_t obj = dsl_deadlist_alloc(mos, tx);
1310 		VERIFY0(zap_add(mos, dd->dd_object, DD_FIELD_LIVELIST,
1311 		    sizeof (uint64_t), 1, &obj, tx));
1312 		spa_feature_incr(dp->dp_spa, SPA_FEATURE_LIVELIST, tx);
1313 	}
1314 
1315 	/*
1316 	 * Since we're creating a new node we know it's a leaf, so we can
1317 	 * initialize the counts if the limit feature is active.
1318 	 */
1319 	if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_FS_SS_LIMIT)) {
1320 		uint64_t cnt = 0;
1321 		objset_t *os = dd->dd_pool->dp_meta_objset;
1322 
1323 		dsl_dir_zapify(dd, tx);
1324 		VERIFY0(zap_add(os, dd->dd_object, DD_FIELD_FILESYSTEM_COUNT,
1325 		    sizeof (cnt), 1, &cnt, tx));
1326 		VERIFY0(zap_add(os, dd->dd_object, DD_FIELD_SNAPSHOT_COUNT,
1327 		    sizeof (cnt), 1, &cnt, tx));
1328 	}
1329 
1330 	dsl_dir_rele(dd, FTAG);
1331 
1332 	/*
1333 	 * If we are creating a clone, make sure we zero out any stale
1334 	 * data from the origin snapshots zil header.
1335 	 */
1336 	if (origin != NULL && !(flags & DS_CREATE_FLAG_NODIRTY)) {
1337 		dsl_dataset_t *ds;
1338 
1339 		VERIFY0(dsl_dataset_hold_obj(dp, dsobj, FTAG, &ds));
1340 		dsl_dataset_zero_zil(ds, tx);
1341 		dsl_dataset_rele(ds, FTAG);
1342 	}
1343 
1344 	return (dsobj);
1345 }
1346 
1347 /*
1348  * The unique space in the head dataset can be calculated by subtracting
1349  * the space used in the most recent snapshot, that is still being used
1350  * in this file system, from the space currently in use.  To figure out
1351  * the space in the most recent snapshot still in use, we need to take
1352  * the total space used in the snapshot and subtract out the space that
1353  * has been freed up since the snapshot was taken.
1354  */
1355 void
1356 dsl_dataset_recalc_head_uniq(dsl_dataset_t *ds)
1357 {
1358 	uint64_t mrs_used;
1359 	uint64_t dlused, dlcomp, dluncomp;
1360 
1361 	ASSERT(!ds->ds_is_snapshot);
1362 
1363 	if (dsl_dataset_phys(ds)->ds_prev_snap_obj != 0)
1364 		mrs_used = dsl_dataset_phys(ds->ds_prev)->ds_referenced_bytes;
1365 	else
1366 		mrs_used = 0;
1367 
1368 	dsl_deadlist_space(&ds->ds_deadlist, &dlused, &dlcomp, &dluncomp);
1369 
1370 	ASSERT3U(dlused, <=, mrs_used);
1371 	dsl_dataset_phys(ds)->ds_unique_bytes =
1372 	    dsl_dataset_phys(ds)->ds_referenced_bytes - (mrs_used - dlused);
1373 
1374 	if (spa_version(ds->ds_dir->dd_pool->dp_spa) >=
1375 	    SPA_VERSION_UNIQUE_ACCURATE)
1376 		dsl_dataset_phys(ds)->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
1377 }
1378 
1379 void
1380 dsl_dataset_remove_from_next_clones(dsl_dataset_t *ds, uint64_t obj,
1381     dmu_tx_t *tx)
1382 {
1383 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
1384 	uint64_t count __maybe_unused;
1385 	int err;
1386 
1387 	ASSERT(dsl_dataset_phys(ds)->ds_num_children >= 2);
1388 	err = zap_remove_int(mos, dsl_dataset_phys(ds)->ds_next_clones_obj,
1389 	    obj, tx);
1390 	/*
1391 	 * The err should not be ENOENT, but a bug in a previous version
1392 	 * of the code could cause upgrade_clones_cb() to not set
1393 	 * ds_next_snap_obj when it should, leading to a missing entry.
1394 	 * If we knew that the pool was created after
1395 	 * SPA_VERSION_NEXT_CLONES, we could assert that it isn't
1396 	 * ENOENT.  However, at least we can check that we don't have
1397 	 * too many entries in the next_clones_obj even after failing to
1398 	 * remove this one.
1399 	 */
1400 	if (err != ENOENT)
1401 		VERIFY0(err);
1402 	ASSERT0(zap_count(mos, dsl_dataset_phys(ds)->ds_next_clones_obj,
1403 	    &count));
1404 	ASSERT3U(count, <=, dsl_dataset_phys(ds)->ds_num_children - 2);
1405 }
1406 
1407 
1408 blkptr_t *
1409 dsl_dataset_get_blkptr(dsl_dataset_t *ds)
1410 {
1411 	return (&dsl_dataset_phys(ds)->ds_bp);
1412 }
1413 
1414 spa_t *
1415 dsl_dataset_get_spa(dsl_dataset_t *ds)
1416 {
1417 	return (ds->ds_dir->dd_pool->dp_spa);
1418 }
1419 
1420 void
1421 dsl_dataset_dirty(dsl_dataset_t *ds, dmu_tx_t *tx)
1422 {
1423 	dsl_pool_t *dp;
1424 
1425 	if (ds == NULL) /* this is the meta-objset */
1426 		return;
1427 
1428 	ASSERT(ds->ds_objset != NULL);
1429 
1430 	if (dsl_dataset_phys(ds)->ds_next_snap_obj != 0)
1431 		panic("dirtying snapshot!");
1432 
1433 	/* Must not dirty a dataset in the same txg where it got snapshotted. */
1434 	ASSERT3U(tx->tx_txg, >, dsl_dataset_phys(ds)->ds_prev_snap_txg);
1435 
1436 	dp = ds->ds_dir->dd_pool;
1437 	if (txg_list_add(&dp->dp_dirty_datasets, ds, tx->tx_txg)) {
1438 		objset_t *os = ds->ds_objset;
1439 
1440 		/* up the hold count until we can be written out */
1441 		dmu_buf_add_ref(ds->ds_dbuf, ds);
1442 
1443 		/* if this dataset is encrypted, grab a reference to the DCK */
1444 		if (ds->ds_dir->dd_crypto_obj != 0 &&
1445 		    !os->os_raw_receive &&
1446 		    !os->os_next_write_raw[tx->tx_txg & TXG_MASK]) {
1447 			ASSERT3P(ds->ds_key_mapping, !=, NULL);
1448 			key_mapping_add_ref(ds->ds_key_mapping, ds);
1449 		}
1450 	}
1451 }
1452 
1453 static int
1454 dsl_dataset_snapshot_reserve_space(dsl_dataset_t *ds, dmu_tx_t *tx)
1455 {
1456 	uint64_t asize;
1457 
1458 	if (!dmu_tx_is_syncing(tx))
1459 		return (0);
1460 
1461 	/*
1462 	 * If there's an fs-only reservation, any blocks that might become
1463 	 * owned by the snapshot dataset must be accommodated by space
1464 	 * outside of the reservation.
1465 	 */
1466 	ASSERT(ds->ds_reserved == 0 || DS_UNIQUE_IS_ACCURATE(ds));
1467 	asize = MIN(dsl_dataset_phys(ds)->ds_unique_bytes, ds->ds_reserved);
1468 	if (asize > dsl_dir_space_available(ds->ds_dir, NULL, 0, TRUE))
1469 		return (SET_ERROR(ENOSPC));
1470 
1471 	/*
1472 	 * Propagate any reserved space for this snapshot to other
1473 	 * snapshot checks in this sync group.
1474 	 */
1475 	if (asize > 0)
1476 		dsl_dir_willuse_space(ds->ds_dir, asize, tx);
1477 
1478 	return (0);
1479 }
1480 
1481 int
1482 dsl_dataset_snapshot_check_impl(dsl_dataset_t *ds, const char *snapname,
1483     dmu_tx_t *tx, boolean_t recv, uint64_t cnt, cred_t *cr, proc_t *proc)
1484 {
1485 	int error;
1486 	uint64_t value;
1487 
1488 	ds->ds_trysnap_txg = tx->tx_txg;
1489 
1490 	if (!dmu_tx_is_syncing(tx))
1491 		return (0);
1492 
1493 	/*
1494 	 * We don't allow multiple snapshots of the same txg.  If there
1495 	 * is already one, try again.
1496 	 */
1497 	if (dsl_dataset_phys(ds)->ds_prev_snap_txg >= tx->tx_txg)
1498 		return (SET_ERROR(EAGAIN));
1499 
1500 	/*
1501 	 * Check for conflicting snapshot name.
1502 	 */
1503 	error = dsl_dataset_snap_lookup(ds, snapname, &value);
1504 	if (error == 0)
1505 		return (SET_ERROR(EEXIST));
1506 	if (error != ENOENT)
1507 		return (error);
1508 
1509 	/*
1510 	 * We don't allow taking snapshots of inconsistent datasets, such as
1511 	 * those into which we are currently receiving.  However, if we are
1512 	 * creating this snapshot as part of a receive, this check will be
1513 	 * executed atomically with respect to the completion of the receive
1514 	 * itself but prior to the clearing of DS_FLAG_INCONSISTENT; in this
1515 	 * case we ignore this, knowing it will be fixed up for us shortly in
1516 	 * dmu_recv_end_sync().
1517 	 */
1518 	if (!recv && DS_IS_INCONSISTENT(ds))
1519 		return (SET_ERROR(EBUSY));
1520 
1521 	/*
1522 	 * Skip the check for temporary snapshots or if we have already checked
1523 	 * the counts in dsl_dataset_snapshot_check. This means we really only
1524 	 * check the count here when we're receiving a stream.
1525 	 */
1526 	if (cnt != 0 && cr != NULL) {
1527 		error = dsl_fs_ss_limit_check(ds->ds_dir, cnt,
1528 		    ZFS_PROP_SNAPSHOT_LIMIT, NULL, cr, proc);
1529 		if (error != 0)
1530 			return (error);
1531 	}
1532 
1533 	error = dsl_dataset_snapshot_reserve_space(ds, tx);
1534 	if (error != 0)
1535 		return (error);
1536 
1537 	return (0);
1538 }
1539 
1540 int
1541 dsl_dataset_snapshot_check(void *arg, dmu_tx_t *tx)
1542 {
1543 	dsl_dataset_snapshot_arg_t *ddsa = arg;
1544 	dsl_pool_t *dp = dmu_tx_pool(tx);
1545 	nvpair_t *pair;
1546 	int rv = 0;
1547 
1548 	/*
1549 	 * Pre-compute how many total new snapshots will be created for each
1550 	 * level in the tree and below. This is needed for validating the
1551 	 * snapshot limit when either taking a recursive snapshot or when
1552 	 * taking multiple snapshots.
1553 	 *
1554 	 * The problem is that the counts are not actually adjusted when
1555 	 * we are checking, only when we finally sync. For a single snapshot,
1556 	 * this is easy, the count will increase by 1 at each node up the tree,
1557 	 * but its more complicated for the recursive/multiple snapshot case.
1558 	 *
1559 	 * The dsl_fs_ss_limit_check function does recursively check the count
1560 	 * at each level up the tree but since it is validating each snapshot
1561 	 * independently we need to be sure that we are validating the complete
1562 	 * count for the entire set of snapshots. We do this by rolling up the
1563 	 * counts for each component of the name into an nvlist and then
1564 	 * checking each of those cases with the aggregated count.
1565 	 *
1566 	 * This approach properly handles not only the recursive snapshot
1567 	 * case (where we get all of those on the ddsa_snaps list) but also
1568 	 * the sibling case (e.g. snapshot a/b and a/c so that we will also
1569 	 * validate the limit on 'a' using a count of 2).
1570 	 *
1571 	 * We validate the snapshot names in the third loop and only report
1572 	 * name errors once.
1573 	 */
1574 	if (dmu_tx_is_syncing(tx)) {
1575 		char *nm;
1576 		nvlist_t *cnt_track = NULL;
1577 		cnt_track = fnvlist_alloc();
1578 
1579 		nm = kmem_alloc(MAXPATHLEN, KM_SLEEP);
1580 
1581 		/* Rollup aggregated counts into the cnt_track list */
1582 		for (pair = nvlist_next_nvpair(ddsa->ddsa_snaps, NULL);
1583 		    pair != NULL;
1584 		    pair = nvlist_next_nvpair(ddsa->ddsa_snaps, pair)) {
1585 			char *pdelim;
1586 			uint64_t val;
1587 
1588 			(void) strlcpy(nm, nvpair_name(pair), MAXPATHLEN);
1589 			pdelim = strchr(nm, '@');
1590 			if (pdelim == NULL)
1591 				continue;
1592 			*pdelim = '\0';
1593 
1594 			do {
1595 				if (nvlist_lookup_uint64(cnt_track, nm,
1596 				    &val) == 0) {
1597 					/* update existing entry */
1598 					fnvlist_add_uint64(cnt_track, nm,
1599 					    val + 1);
1600 				} else {
1601 					/* add to list */
1602 					fnvlist_add_uint64(cnt_track, nm, 1);
1603 				}
1604 
1605 				pdelim = strrchr(nm, '/');
1606 				if (pdelim != NULL)
1607 					*pdelim = '\0';
1608 			} while (pdelim != NULL);
1609 		}
1610 
1611 		kmem_free(nm, MAXPATHLEN);
1612 
1613 		/* Check aggregated counts at each level */
1614 		for (pair = nvlist_next_nvpair(cnt_track, NULL);
1615 		    pair != NULL; pair = nvlist_next_nvpair(cnt_track, pair)) {
1616 			int error = 0;
1617 			char *name;
1618 			uint64_t cnt = 0;
1619 			dsl_dataset_t *ds;
1620 
1621 			name = nvpair_name(pair);
1622 			cnt = fnvpair_value_uint64(pair);
1623 			ASSERT(cnt > 0);
1624 
1625 			error = dsl_dataset_hold(dp, name, FTAG, &ds);
1626 			if (error == 0) {
1627 				error = dsl_fs_ss_limit_check(ds->ds_dir, cnt,
1628 				    ZFS_PROP_SNAPSHOT_LIMIT, NULL,
1629 				    ddsa->ddsa_cr, ddsa->ddsa_proc);
1630 				dsl_dataset_rele(ds, FTAG);
1631 			}
1632 
1633 			if (error != 0) {
1634 				if (ddsa->ddsa_errors != NULL)
1635 					fnvlist_add_int32(ddsa->ddsa_errors,
1636 					    name, error);
1637 				rv = error;
1638 				/* only report one error for this check */
1639 				break;
1640 			}
1641 		}
1642 		nvlist_free(cnt_track);
1643 	}
1644 
1645 	for (pair = nvlist_next_nvpair(ddsa->ddsa_snaps, NULL);
1646 	    pair != NULL; pair = nvlist_next_nvpair(ddsa->ddsa_snaps, pair)) {
1647 		int error = 0;
1648 		dsl_dataset_t *ds;
1649 		char *name, *atp = NULL;
1650 		char dsname[ZFS_MAX_DATASET_NAME_LEN];
1651 
1652 		name = nvpair_name(pair);
1653 		if (strlen(name) >= ZFS_MAX_DATASET_NAME_LEN)
1654 			error = SET_ERROR(ENAMETOOLONG);
1655 		if (error == 0) {
1656 			atp = strchr(name, '@');
1657 			if (atp == NULL)
1658 				error = SET_ERROR(EINVAL);
1659 			if (error == 0)
1660 				(void) strlcpy(dsname, name, atp - name + 1);
1661 		}
1662 		if (error == 0)
1663 			error = dsl_dataset_hold(dp, dsname, FTAG, &ds);
1664 		if (error == 0) {
1665 			/* passing 0/NULL skips dsl_fs_ss_limit_check */
1666 			error = dsl_dataset_snapshot_check_impl(ds,
1667 			    atp + 1, tx, B_FALSE, 0, NULL, NULL);
1668 			dsl_dataset_rele(ds, FTAG);
1669 		}
1670 
1671 		if (error != 0) {
1672 			if (ddsa->ddsa_errors != NULL) {
1673 				fnvlist_add_int32(ddsa->ddsa_errors,
1674 				    name, error);
1675 			}
1676 			rv = error;
1677 		}
1678 	}
1679 
1680 	return (rv);
1681 }
1682 
1683 void
1684 dsl_dataset_snapshot_sync_impl(dsl_dataset_t *ds, const char *snapname,
1685     dmu_tx_t *tx)
1686 {
1687 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1688 	dmu_buf_t *dbuf;
1689 	dsl_dataset_phys_t *dsphys;
1690 	uint64_t dsobj, crtxg;
1691 	objset_t *mos = dp->dp_meta_objset;
1692 	static zil_header_t zero_zil __maybe_unused;
1693 	objset_t *os __maybe_unused;
1694 
1695 	ASSERT(RRW_WRITE_HELD(&dp->dp_config_rwlock));
1696 
1697 	/*
1698 	 * If we are on an old pool, the zil must not be active, in which
1699 	 * case it will be zeroed.  Usually zil_suspend() accomplishes this.
1700 	 */
1701 	ASSERT(spa_version(dmu_tx_pool(tx)->dp_spa) >= SPA_VERSION_FAST_SNAP ||
1702 	    dmu_objset_from_ds(ds, &os) != 0 ||
1703 	    bcmp(&os->os_phys->os_zil_header, &zero_zil,
1704 	    sizeof (zero_zil)) == 0);
1705 
1706 	/* Should not snapshot a dirty dataset. */
1707 	ASSERT(!txg_list_member(&ds->ds_dir->dd_pool->dp_dirty_datasets,
1708 	    ds, tx->tx_txg));
1709 
1710 	dsl_fs_ss_count_adjust(ds->ds_dir, 1, DD_FIELD_SNAPSHOT_COUNT, tx);
1711 
1712 	/*
1713 	 * The origin's ds_creation_txg has to be < TXG_INITIAL
1714 	 */
1715 	if (strcmp(snapname, ORIGIN_DIR_NAME) == 0)
1716 		crtxg = 1;
1717 	else
1718 		crtxg = tx->tx_txg;
1719 
1720 	dsobj = dmu_object_alloc(mos, DMU_OT_DSL_DATASET, 0,
1721 	    DMU_OT_DSL_DATASET, sizeof (dsl_dataset_phys_t), tx);
1722 	VERIFY0(dmu_bonus_hold(mos, dsobj, FTAG, &dbuf));
1723 	dmu_buf_will_dirty(dbuf, tx);
1724 	dsphys = dbuf->db_data;
1725 	bzero(dsphys, sizeof (dsl_dataset_phys_t));
1726 	dsphys->ds_dir_obj = ds->ds_dir->dd_object;
1727 	dsphys->ds_fsid_guid = unique_create();
1728 	(void) random_get_pseudo_bytes((void*)&dsphys->ds_guid,
1729 	    sizeof (dsphys->ds_guid));
1730 	dsphys->ds_prev_snap_obj = dsl_dataset_phys(ds)->ds_prev_snap_obj;
1731 	dsphys->ds_prev_snap_txg = dsl_dataset_phys(ds)->ds_prev_snap_txg;
1732 	dsphys->ds_next_snap_obj = ds->ds_object;
1733 	dsphys->ds_num_children = 1;
1734 	dsphys->ds_creation_time = gethrestime_sec();
1735 	dsphys->ds_creation_txg = crtxg;
1736 	dsphys->ds_deadlist_obj = dsl_dataset_phys(ds)->ds_deadlist_obj;
1737 	dsphys->ds_referenced_bytes = dsl_dataset_phys(ds)->ds_referenced_bytes;
1738 	dsphys->ds_compressed_bytes = dsl_dataset_phys(ds)->ds_compressed_bytes;
1739 	dsphys->ds_uncompressed_bytes =
1740 	    dsl_dataset_phys(ds)->ds_uncompressed_bytes;
1741 	dsphys->ds_flags = dsl_dataset_phys(ds)->ds_flags;
1742 	rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
1743 	dsphys->ds_bp = dsl_dataset_phys(ds)->ds_bp;
1744 	rrw_exit(&ds->ds_bp_rwlock, FTAG);
1745 	dmu_buf_rele(dbuf, FTAG);
1746 
1747 	for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
1748 		if (zfeature_active(f, ds->ds_feature[f])) {
1749 			dsl_dataset_activate_feature(dsobj, f,
1750 			    ds->ds_feature[f], tx);
1751 		}
1752 	}
1753 
1754 	ASSERT3U(ds->ds_prev != 0, ==,
1755 	    dsl_dataset_phys(ds)->ds_prev_snap_obj != 0);
1756 	if (ds->ds_prev) {
1757 		uint64_t next_clones_obj =
1758 		    dsl_dataset_phys(ds->ds_prev)->ds_next_clones_obj;
1759 		ASSERT(dsl_dataset_phys(ds->ds_prev)->ds_next_snap_obj ==
1760 		    ds->ds_object ||
1761 		    dsl_dataset_phys(ds->ds_prev)->ds_num_children > 1);
1762 		if (dsl_dataset_phys(ds->ds_prev)->ds_next_snap_obj ==
1763 		    ds->ds_object) {
1764 			dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
1765 			ASSERT3U(dsl_dataset_phys(ds)->ds_prev_snap_txg, ==,
1766 			    dsl_dataset_phys(ds->ds_prev)->ds_creation_txg);
1767 			dsl_dataset_phys(ds->ds_prev)->ds_next_snap_obj = dsobj;
1768 		} else if (next_clones_obj != 0) {
1769 			dsl_dataset_remove_from_next_clones(ds->ds_prev,
1770 			    dsphys->ds_next_snap_obj, tx);
1771 			VERIFY0(zap_add_int(mos,
1772 			    next_clones_obj, dsobj, tx));
1773 		}
1774 	}
1775 
1776 	/*
1777 	 * If we have a reference-reservation on this dataset, we will
1778 	 * need to increase the amount of refreservation being charged
1779 	 * since our unique space is going to zero.
1780 	 */
1781 	if (ds->ds_reserved) {
1782 		int64_t delta;
1783 		ASSERT(DS_UNIQUE_IS_ACCURATE(ds));
1784 		delta = MIN(dsl_dataset_phys(ds)->ds_unique_bytes,
1785 		    ds->ds_reserved);
1786 		dsl_dir_diduse_space(ds->ds_dir, DD_USED_REFRSRV,
1787 		    delta, 0, 0, tx);
1788 	}
1789 
1790 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
1791 	dsl_dataset_phys(ds)->ds_deadlist_obj =
1792 	    dsl_deadlist_clone(&ds->ds_deadlist, UINT64_MAX,
1793 	    dsl_dataset_phys(ds)->ds_prev_snap_obj, tx);
1794 	dsl_deadlist_close(&ds->ds_deadlist);
1795 	dsl_deadlist_open(&ds->ds_deadlist, mos,
1796 	    dsl_dataset_phys(ds)->ds_deadlist_obj);
1797 	dsl_deadlist_add_key(&ds->ds_deadlist,
1798 	    dsl_dataset_phys(ds)->ds_prev_snap_txg, tx);
1799 	dsl_bookmark_snapshotted(ds, tx);
1800 
1801 	if (dsl_dataset_remap_deadlist_exists(ds)) {
1802 		uint64_t remap_deadlist_obj =
1803 		    dsl_dataset_get_remap_deadlist_object(ds);
1804 		/*
1805 		 * Move the remap_deadlist to the snapshot.  The head
1806 		 * will create a new remap deadlist on demand, from
1807 		 * dsl_dataset_block_remapped().
1808 		 */
1809 		dsl_dataset_unset_remap_deadlist_object(ds, tx);
1810 		dsl_deadlist_close(&ds->ds_remap_deadlist);
1811 
1812 		dmu_object_zapify(mos, dsobj, DMU_OT_DSL_DATASET, tx);
1813 		VERIFY0(zap_add(mos, dsobj, DS_FIELD_REMAP_DEADLIST,
1814 		    sizeof (remap_deadlist_obj), 1, &remap_deadlist_obj, tx));
1815 	}
1816 
1817 	/*
1818 	 * Create a ivset guid for this snapshot if the dataset is
1819 	 * encrypted. This may be overridden by a raw receive. A
1820 	 * previous implementation of this code did not have this
1821 	 * field as part of the on-disk format for ZFS encryption
1822 	 * (see errata #4). As part of the remediation for this
1823 	 * issue, we ask the user to enable the bookmark_v2 feature
1824 	 * which is now a dependency of the encryption feature. We
1825 	 * use this as a heuristic to determine when the user has
1826 	 * elected to correct any datasets created with the old code.
1827 	 * As a result, we only do this step if the bookmark_v2
1828 	 * feature is enabled, which limits the number of states a
1829 	 * given pool / dataset can be in with regards to terms of
1830 	 * correcting the issue.
1831 	 */
1832 	if (ds->ds_dir->dd_crypto_obj != 0 &&
1833 	    spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_BOOKMARK_V2)) {
1834 		uint64_t ivset_guid = unique_create();
1835 
1836 		dmu_object_zapify(mos, dsobj, DMU_OT_DSL_DATASET, tx);
1837 		VERIFY0(zap_add(mos, dsobj, DS_FIELD_IVSET_GUID,
1838 		    sizeof (ivset_guid), 1, &ivset_guid, tx));
1839 	}
1840 
1841 	ASSERT3U(dsl_dataset_phys(ds)->ds_prev_snap_txg, <, tx->tx_txg);
1842 	dsl_dataset_phys(ds)->ds_prev_snap_obj = dsobj;
1843 	dsl_dataset_phys(ds)->ds_prev_snap_txg = crtxg;
1844 	dsl_dataset_phys(ds)->ds_unique_bytes = 0;
1845 
1846 	if (spa_version(dp->dp_spa) >= SPA_VERSION_UNIQUE_ACCURATE)
1847 		dsl_dataset_phys(ds)->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
1848 
1849 	VERIFY0(zap_add(mos, dsl_dataset_phys(ds)->ds_snapnames_zapobj,
1850 	    snapname, 8, 1, &dsobj, tx));
1851 
1852 	if (ds->ds_prev)
1853 		dsl_dataset_rele(ds->ds_prev, ds);
1854 	VERIFY0(dsl_dataset_hold_obj(dp,
1855 	    dsl_dataset_phys(ds)->ds_prev_snap_obj, ds, &ds->ds_prev));
1856 
1857 	dsl_scan_ds_snapshotted(ds, tx);
1858 
1859 	dsl_dir_snap_cmtime_update(ds->ds_dir);
1860 
1861 	spa_history_log_internal_ds(ds->ds_prev, "snapshot", tx, " ");
1862 }
1863 
1864 void
1865 dsl_dataset_snapshot_sync(void *arg, dmu_tx_t *tx)
1866 {
1867 	dsl_dataset_snapshot_arg_t *ddsa = arg;
1868 	dsl_pool_t *dp = dmu_tx_pool(tx);
1869 	nvpair_t *pair;
1870 
1871 	for (pair = nvlist_next_nvpair(ddsa->ddsa_snaps, NULL);
1872 	    pair != NULL; pair = nvlist_next_nvpair(ddsa->ddsa_snaps, pair)) {
1873 		dsl_dataset_t *ds;
1874 		char *name, *atp;
1875 		char dsname[ZFS_MAX_DATASET_NAME_LEN];
1876 
1877 		name = nvpair_name(pair);
1878 		atp = strchr(name, '@');
1879 		(void) strlcpy(dsname, name, atp - name + 1);
1880 		VERIFY0(dsl_dataset_hold(dp, dsname, FTAG, &ds));
1881 
1882 		dsl_dataset_snapshot_sync_impl(ds, atp + 1, tx);
1883 		if (ddsa->ddsa_props != NULL) {
1884 			dsl_props_set_sync_impl(ds->ds_prev,
1885 			    ZPROP_SRC_LOCAL, ddsa->ddsa_props, tx);
1886 		}
1887 		dsl_dataset_rele(ds, FTAG);
1888 	}
1889 }
1890 
1891 /*
1892  * The snapshots must all be in the same pool.
1893  * All-or-nothing: if there are any failures, nothing will be modified.
1894  */
1895 int
1896 dsl_dataset_snapshot(nvlist_t *snaps, nvlist_t *props, nvlist_t *errors)
1897 {
1898 	dsl_dataset_snapshot_arg_t ddsa;
1899 	nvpair_t *pair;
1900 	boolean_t needsuspend;
1901 	int error;
1902 	spa_t *spa;
1903 	char *firstname;
1904 	nvlist_t *suspended = NULL;
1905 
1906 	pair = nvlist_next_nvpair(snaps, NULL);
1907 	if (pair == NULL)
1908 		return (0);
1909 	firstname = nvpair_name(pair);
1910 
1911 	error = spa_open(firstname, &spa, FTAG);
1912 	if (error != 0)
1913 		return (error);
1914 	needsuspend = (spa_version(spa) < SPA_VERSION_FAST_SNAP);
1915 	spa_close(spa, FTAG);
1916 
1917 	if (needsuspend) {
1918 		suspended = fnvlist_alloc();
1919 		for (pair = nvlist_next_nvpair(snaps, NULL); pair != NULL;
1920 		    pair = nvlist_next_nvpair(snaps, pair)) {
1921 			char fsname[ZFS_MAX_DATASET_NAME_LEN];
1922 			char *snapname = nvpair_name(pair);
1923 			char *atp;
1924 			void *cookie;
1925 
1926 			atp = strchr(snapname, '@');
1927 			if (atp == NULL) {
1928 				error = SET_ERROR(EINVAL);
1929 				break;
1930 			}
1931 			(void) strlcpy(fsname, snapname, atp - snapname + 1);
1932 
1933 			error = zil_suspend(fsname, &cookie);
1934 			if (error != 0)
1935 				break;
1936 			fnvlist_add_uint64(suspended, fsname,
1937 			    (uintptr_t)cookie);
1938 		}
1939 	}
1940 
1941 	ddsa.ddsa_snaps = snaps;
1942 	ddsa.ddsa_props = props;
1943 	ddsa.ddsa_errors = errors;
1944 	ddsa.ddsa_cr = CRED();
1945 	ddsa.ddsa_proc = curproc;
1946 
1947 	if (error == 0) {
1948 		error = dsl_sync_task(firstname, dsl_dataset_snapshot_check,
1949 		    dsl_dataset_snapshot_sync, &ddsa,
1950 		    fnvlist_num_pairs(snaps) * 3, ZFS_SPACE_CHECK_NORMAL);
1951 	}
1952 
1953 	if (suspended != NULL) {
1954 		for (pair = nvlist_next_nvpair(suspended, NULL); pair != NULL;
1955 		    pair = nvlist_next_nvpair(suspended, pair)) {
1956 			zil_resume((void *)(uintptr_t)
1957 			    fnvpair_value_uint64(pair));
1958 		}
1959 		fnvlist_free(suspended);
1960 	}
1961 
1962 	if (error == 0) {
1963 		for (pair = nvlist_next_nvpair(snaps, NULL); pair != NULL;
1964 		    pair = nvlist_next_nvpair(snaps, pair)) {
1965 			zvol_create_minor(nvpair_name(pair));
1966 		}
1967 	}
1968 
1969 	return (error);
1970 }
1971 
1972 typedef struct dsl_dataset_snapshot_tmp_arg {
1973 	const char *ddsta_fsname;
1974 	const char *ddsta_snapname;
1975 	minor_t ddsta_cleanup_minor;
1976 	const char *ddsta_htag;
1977 } dsl_dataset_snapshot_tmp_arg_t;
1978 
1979 static int
1980 dsl_dataset_snapshot_tmp_check(void *arg, dmu_tx_t *tx)
1981 {
1982 	dsl_dataset_snapshot_tmp_arg_t *ddsta = arg;
1983 	dsl_pool_t *dp = dmu_tx_pool(tx);
1984 	dsl_dataset_t *ds;
1985 	int error;
1986 
1987 	error = dsl_dataset_hold(dp, ddsta->ddsta_fsname, FTAG, &ds);
1988 	if (error != 0)
1989 		return (error);
1990 
1991 	/* NULL cred means no limit check for tmp snapshot */
1992 	error = dsl_dataset_snapshot_check_impl(ds, ddsta->ddsta_snapname,
1993 	    tx, B_FALSE, 0, NULL, NULL);
1994 	if (error != 0) {
1995 		dsl_dataset_rele(ds, FTAG);
1996 		return (error);
1997 	}
1998 
1999 	if (spa_version(dp->dp_spa) < SPA_VERSION_USERREFS) {
2000 		dsl_dataset_rele(ds, FTAG);
2001 		return (SET_ERROR(ENOTSUP));
2002 	}
2003 	error = dsl_dataset_user_hold_check_one(NULL, ddsta->ddsta_htag,
2004 	    B_TRUE, tx);
2005 	if (error != 0) {
2006 		dsl_dataset_rele(ds, FTAG);
2007 		return (error);
2008 	}
2009 
2010 	dsl_dataset_rele(ds, FTAG);
2011 	return (0);
2012 }
2013 
2014 static void
2015 dsl_dataset_snapshot_tmp_sync(void *arg, dmu_tx_t *tx)
2016 {
2017 	dsl_dataset_snapshot_tmp_arg_t *ddsta = arg;
2018 	dsl_pool_t *dp = dmu_tx_pool(tx);
2019 	dsl_dataset_t *ds = NULL;
2020 
2021 	VERIFY0(dsl_dataset_hold(dp, ddsta->ddsta_fsname, FTAG, &ds));
2022 
2023 	dsl_dataset_snapshot_sync_impl(ds, ddsta->ddsta_snapname, tx);
2024 	dsl_dataset_user_hold_sync_one(ds->ds_prev, ddsta->ddsta_htag,
2025 	    ddsta->ddsta_cleanup_minor, gethrestime_sec(), tx);
2026 	dsl_destroy_snapshot_sync_impl(ds->ds_prev, B_TRUE, tx);
2027 
2028 	dsl_dataset_rele(ds, FTAG);
2029 }
2030 
2031 int
2032 dsl_dataset_snapshot_tmp(const char *fsname, const char *snapname,
2033     minor_t cleanup_minor, const char *htag)
2034 {
2035 	dsl_dataset_snapshot_tmp_arg_t ddsta;
2036 	int error;
2037 	spa_t *spa;
2038 	boolean_t needsuspend;
2039 	void *cookie;
2040 
2041 	ddsta.ddsta_fsname = fsname;
2042 	ddsta.ddsta_snapname = snapname;
2043 	ddsta.ddsta_cleanup_minor = cleanup_minor;
2044 	ddsta.ddsta_htag = htag;
2045 
2046 	error = spa_open(fsname, &spa, FTAG);
2047 	if (error != 0)
2048 		return (error);
2049 	needsuspend = (spa_version(spa) < SPA_VERSION_FAST_SNAP);
2050 	spa_close(spa, FTAG);
2051 
2052 	if (needsuspend) {
2053 		error = zil_suspend(fsname, &cookie);
2054 		if (error != 0)
2055 			return (error);
2056 	}
2057 
2058 	error = dsl_sync_task(fsname, dsl_dataset_snapshot_tmp_check,
2059 	    dsl_dataset_snapshot_tmp_sync, &ddsta, 3, ZFS_SPACE_CHECK_RESERVED);
2060 
2061 	if (needsuspend)
2062 		zil_resume(cookie);
2063 	return (error);
2064 }
2065 
2066 void
2067 dsl_dataset_sync(dsl_dataset_t *ds, zio_t *zio, dmu_tx_t *tx)
2068 {
2069 	ASSERT(dmu_tx_is_syncing(tx));
2070 	ASSERT(ds->ds_objset != NULL);
2071 	ASSERT(dsl_dataset_phys(ds)->ds_next_snap_obj == 0);
2072 
2073 	/*
2074 	 * in case we had to change ds_fsid_guid when we opened it,
2075 	 * sync it out now.
2076 	 */
2077 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
2078 	dsl_dataset_phys(ds)->ds_fsid_guid = ds->ds_fsid_guid;
2079 
2080 	if (ds->ds_resume_bytes[tx->tx_txg & TXG_MASK] != 0) {
2081 		VERIFY0(zap_update(tx->tx_pool->dp_meta_objset,
2082 		    ds->ds_object, DS_FIELD_RESUME_OBJECT, 8, 1,
2083 		    &ds->ds_resume_object[tx->tx_txg & TXG_MASK], tx));
2084 		VERIFY0(zap_update(tx->tx_pool->dp_meta_objset,
2085 		    ds->ds_object, DS_FIELD_RESUME_OFFSET, 8, 1,
2086 		    &ds->ds_resume_offset[tx->tx_txg & TXG_MASK], tx));
2087 		VERIFY0(zap_update(tx->tx_pool->dp_meta_objset,
2088 		    ds->ds_object, DS_FIELD_RESUME_BYTES, 8, 1,
2089 		    &ds->ds_resume_bytes[tx->tx_txg & TXG_MASK], tx));
2090 		ds->ds_resume_object[tx->tx_txg & TXG_MASK] = 0;
2091 		ds->ds_resume_offset[tx->tx_txg & TXG_MASK] = 0;
2092 		ds->ds_resume_bytes[tx->tx_txg & TXG_MASK] = 0;
2093 	}
2094 
2095 	dmu_objset_sync(ds->ds_objset, zio, tx);
2096 
2097 	for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
2098 		if (zfeature_active(f, ds->ds_feature_activation[f])) {
2099 			if (zfeature_active(f, ds->ds_feature[f]))
2100 				continue;
2101 			dsl_dataset_activate_feature(ds->ds_object, f,
2102 			    ds->ds_feature_activation[f], tx);
2103 			ds->ds_feature[f] = ds->ds_feature_activation[f];
2104 		}
2105 	}
2106 }
2107 
2108 /*
2109  * Check if the percentage of blocks shared between the clone and the
2110  * snapshot (as opposed to those that are clone only) is below a certain
2111  * threshold
2112  */
2113 static boolean_t
2114 dsl_livelist_should_disable(dsl_dataset_t *ds)
2115 {
2116 	uint64_t used, referenced;
2117 	int percent_shared;
2118 
2119 	used = dsl_dir_get_usedds(ds->ds_dir);
2120 	referenced = dsl_get_referenced(ds);
2121 	ASSERT3U(referenced, >=, 0);
2122 	ASSERT3U(used, >=, 0);
2123 	if (referenced == 0)
2124 		return (B_FALSE);
2125 	percent_shared = (100 * (referenced - used)) / referenced;
2126 	if (percent_shared <= zfs_livelist_min_percent_shared)
2127 		return (B_TRUE);
2128 	return (B_FALSE);
2129 }
2130 
2131 /*
2132  *  Check if it is possible to combine two livelist entries into one.
2133  *  This is the case if the combined number of 'live' blkptrs (ALLOCs that
2134  *  don't have a matching FREE) is under the maximum sublist size.
2135  *  We check this by subtracting twice the total number of frees from the total
2136  *  number of blkptrs. FREEs are counted twice because each FREE blkptr
2137  *  will cancel out an ALLOC blkptr when the livelist is processed.
2138  */
2139 static boolean_t
2140 dsl_livelist_should_condense(dsl_deadlist_entry_t *first,
2141     dsl_deadlist_entry_t *next)
2142 {
2143 	uint64_t total_free = first->dle_bpobj.bpo_phys->bpo_num_freed +
2144 	    next->dle_bpobj.bpo_phys->bpo_num_freed;
2145 	uint64_t total_entries = first->dle_bpobj.bpo_phys->bpo_num_blkptrs +
2146 	    next->dle_bpobj.bpo_phys->bpo_num_blkptrs;
2147 	if ((total_entries - (2 * total_free)) < zfs_livelist_max_entries)
2148 		return (B_TRUE);
2149 	return (B_FALSE);
2150 }
2151 
2152 typedef struct try_condense_arg {
2153 	spa_t *spa;
2154 	dsl_dataset_t *ds;
2155 } try_condense_arg_t;
2156 
2157 /*
2158  * Iterate over the livelist entries, searching for a pair to condense.
2159  * A nonzero return value means stop, 0 means keep looking.
2160  */
2161 static int
2162 dsl_livelist_try_condense(void *arg, dsl_deadlist_entry_t *first)
2163 {
2164 	try_condense_arg_t *tca = arg;
2165 	spa_t *spa = tca->spa;
2166 	dsl_dataset_t *ds = tca->ds;
2167 	dsl_deadlist_t *ll = &ds->ds_dir->dd_livelist;
2168 	dsl_deadlist_entry_t *next;
2169 
2170 	/* The condense thread has not yet been created at import */
2171 	if (spa->spa_livelist_condense_zthr == NULL)
2172 		return (1);
2173 
2174 	/* A condense is already in progress */
2175 	if (spa->spa_to_condense.ds != NULL)
2176 		return (1);
2177 
2178 	next = AVL_NEXT(&ll->dl_tree, &first->dle_node);
2179 	/* The livelist has only one entry - don't condense it */
2180 	if (next == NULL)
2181 		return (1);
2182 
2183 	/* Next is the newest entry - don't condense it */
2184 	if (AVL_NEXT(&ll->dl_tree, &next->dle_node) == NULL)
2185 		return (1);
2186 
2187 	/* This pair is not ready to condense but keep looking */
2188 	if (!dsl_livelist_should_condense(first, next))
2189 		return (0);
2190 
2191 	/*
2192 	 * Add a ref to prevent the dataset from being evicted while
2193 	 * the condense zthr or synctask are running. Ref will be
2194 	 * released at the end of the condense synctask
2195 	 */
2196 	dmu_buf_add_ref(ds->ds_dbuf, spa);
2197 
2198 	spa->spa_to_condense.ds = ds;
2199 	spa->spa_to_condense.first = first;
2200 	spa->spa_to_condense.next = next;
2201 	spa->spa_to_condense.syncing = B_FALSE;
2202 	spa->spa_to_condense.cancelled = B_FALSE;
2203 
2204 	zthr_wakeup(spa->spa_livelist_condense_zthr);
2205 	return (1);
2206 }
2207 
2208 static void
2209 dsl_flush_pending_livelist(dsl_dataset_t *ds, dmu_tx_t *tx)
2210 {
2211 	dsl_dir_t *dd = ds->ds_dir;
2212 	spa_t *spa = ds->ds_dir->dd_pool->dp_spa;
2213 	dsl_deadlist_entry_t *last = dsl_deadlist_last(&dd->dd_livelist);
2214 
2215 	/* Check if we need to add a new sub-livelist */
2216 	if (last == NULL) {
2217 		/* The livelist is empty */
2218 		dsl_deadlist_add_key(&dd->dd_livelist,
2219 		    tx->tx_txg - 1, tx);
2220 	} else if (spa_sync_pass(spa) == 1) {
2221 		/*
2222 		 * Check if the newest entry is full. If it is, make a new one.
2223 		 * We only do this once per sync because we could overfill a
2224 		 * sublist in one sync pass and don't want to add another entry
2225 		 * for a txg that is already represented. This ensures that
2226 		 * blkptrs born in the same txg are stored in the same sublist.
2227 		 */
2228 		bpobj_t bpobj = last->dle_bpobj;
2229 		uint64_t all = bpobj.bpo_phys->bpo_num_blkptrs;
2230 		uint64_t free = bpobj.bpo_phys->bpo_num_freed;
2231 		uint64_t alloc = all - free;
2232 		if (alloc > zfs_livelist_max_entries) {
2233 			dsl_deadlist_add_key(&dd->dd_livelist,
2234 			    tx->tx_txg - 1, tx);
2235 		}
2236 	}
2237 
2238 	/* Insert each entry into the on-disk livelist */
2239 	bplist_iterate(&dd->dd_pending_allocs,
2240 	    dsl_deadlist_insert_alloc_cb, &dd->dd_livelist, tx);
2241 	bplist_iterate(&dd->dd_pending_frees,
2242 	    dsl_deadlist_insert_free_cb, &dd->dd_livelist, tx);
2243 
2244 	/* Attempt to condense every pair of adjacent entries */
2245 	try_condense_arg_t arg = {
2246 	    .spa = spa,
2247 	    .ds = ds
2248 	};
2249 	dsl_deadlist_iterate(&dd->dd_livelist, dsl_livelist_try_condense,
2250 	    &arg);
2251 }
2252 
2253 void
2254 dsl_dataset_sync_done(dsl_dataset_t *ds, dmu_tx_t *tx)
2255 {
2256 	objset_t *os = ds->ds_objset;
2257 
2258 	bplist_iterate(&ds->ds_pending_deadlist,
2259 	    dsl_deadlist_insert_alloc_cb, &ds->ds_deadlist, tx);
2260 
2261 	if (dsl_deadlist_is_open(&ds->ds_dir->dd_livelist)) {
2262 		dsl_flush_pending_livelist(ds, tx);
2263 		if (dsl_livelist_should_disable(ds)) {
2264 			dsl_dir_remove_livelist(ds->ds_dir, tx, B_TRUE);
2265 		}
2266 	}
2267 
2268 	dsl_bookmark_sync_done(ds, tx);
2269 
2270 	if (os->os_synced_dnodes != NULL) {
2271 		multilist_destroy(os->os_synced_dnodes);
2272 		os->os_synced_dnodes = NULL;
2273 	}
2274 
2275 	if (os->os_encrypted)
2276 		os->os_next_write_raw[tx->tx_txg & TXG_MASK] = B_FALSE;
2277 	else
2278 		ASSERT0(os->os_next_write_raw[tx->tx_txg & TXG_MASK]);
2279 
2280 	ASSERT(!dmu_objset_is_dirty(os, dmu_tx_get_txg(tx)));
2281 
2282 	dmu_buf_rele(ds->ds_dbuf, ds);
2283 }
2284 
2285 int
2286 get_clones_stat_impl(dsl_dataset_t *ds, nvlist_t *val)
2287 {
2288 	uint64_t count = 0;
2289 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
2290 	zap_cursor_t zc;
2291 	zap_attribute_t za;
2292 
2293 	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
2294 
2295 	/*
2296 	 * There may be missing entries in ds_next_clones_obj
2297 	 * due to a bug in a previous version of the code.
2298 	 * Only trust it if it has the right number of entries.
2299 	 */
2300 	if (dsl_dataset_phys(ds)->ds_next_clones_obj != 0) {
2301 		VERIFY0(zap_count(mos, dsl_dataset_phys(ds)->ds_next_clones_obj,
2302 		    &count));
2303 	}
2304 	if (count != dsl_dataset_phys(ds)->ds_num_children - 1) {
2305 		return (SET_ERROR(ENOENT));
2306 	}
2307 	for (zap_cursor_init(&zc, mos,
2308 	    dsl_dataset_phys(ds)->ds_next_clones_obj);
2309 	    zap_cursor_retrieve(&zc, &za) == 0;
2310 	    zap_cursor_advance(&zc)) {
2311 		dsl_dataset_t *clone;
2312 		char buf[ZFS_MAX_DATASET_NAME_LEN];
2313 		VERIFY0(dsl_dataset_hold_obj(ds->ds_dir->dd_pool,
2314 		    za.za_first_integer, FTAG, &clone));
2315 		dsl_dir_name(clone->ds_dir, buf);
2316 		fnvlist_add_boolean(val, buf);
2317 		dsl_dataset_rele(clone, FTAG);
2318 	}
2319 	zap_cursor_fini(&zc);
2320 	return (0);
2321 }
2322 
2323 void
2324 get_clones_stat(dsl_dataset_t *ds, nvlist_t *nv)
2325 {
2326 	nvlist_t *propval = fnvlist_alloc();
2327 	nvlist_t *val;
2328 
2329 	/*
2330 	 * We use nvlist_alloc() instead of fnvlist_alloc() because the
2331 	 * latter would allocate the list with NV_UNIQUE_NAME flag.
2332 	 * As a result, every time a clone name is appended to the list
2333 	 * it would be (linearly) searched for a duplicate name.
2334 	 * We already know that all clone names must be unique and we
2335 	 * want avoid the quadratic complexity of double-checking that
2336 	 * because we can have a large number of clones.
2337 	 */
2338 	VERIFY0(nvlist_alloc(&val, 0, KM_SLEEP));
2339 
2340 	if (get_clones_stat_impl(ds, val) == 0) {
2341 		fnvlist_add_nvlist(propval, ZPROP_VALUE, val);
2342 		fnvlist_add_nvlist(nv, zfs_prop_to_name(ZFS_PROP_CLONES),
2343 		    propval);
2344 	}
2345 
2346 	nvlist_free(val);
2347 	nvlist_free(propval);
2348 }
2349 
2350 /*
2351  * Returns a string that represents the receive resume stats token. It should
2352  * be freed with strfree().
2353  */
2354 char *
2355 get_receive_resume_stats_impl(dsl_dataset_t *ds)
2356 {
2357 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2358 
2359 	if (dsl_dataset_has_resume_receive_state(ds)) {
2360 		char *str;
2361 		void *packed;
2362 		uint8_t *compressed;
2363 		uint64_t val;
2364 		nvlist_t *token_nv = fnvlist_alloc();
2365 		size_t packed_size, compressed_size;
2366 
2367 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
2368 		    DS_FIELD_RESUME_FROMGUID, sizeof (val), 1, &val) == 0) {
2369 			fnvlist_add_uint64(token_nv, "fromguid", val);
2370 		}
2371 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
2372 		    DS_FIELD_RESUME_OBJECT, sizeof (val), 1, &val) == 0) {
2373 			fnvlist_add_uint64(token_nv, "object", val);
2374 		}
2375 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
2376 		    DS_FIELD_RESUME_OFFSET, sizeof (val), 1, &val) == 0) {
2377 			fnvlist_add_uint64(token_nv, "offset", val);
2378 		}
2379 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
2380 		    DS_FIELD_RESUME_BYTES, sizeof (val), 1, &val) == 0) {
2381 			fnvlist_add_uint64(token_nv, "bytes", val);
2382 		}
2383 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
2384 		    DS_FIELD_RESUME_TOGUID, sizeof (val), 1, &val) == 0) {
2385 			fnvlist_add_uint64(token_nv, "toguid", val);
2386 		}
2387 		char buf[MAXNAMELEN];
2388 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
2389 		    DS_FIELD_RESUME_TONAME, 1, sizeof (buf), buf) == 0) {
2390 			fnvlist_add_string(token_nv, "toname", buf);
2391 		}
2392 		if (zap_contains(dp->dp_meta_objset, ds->ds_object,
2393 		    DS_FIELD_RESUME_LARGEBLOCK) == 0) {
2394 			fnvlist_add_boolean(token_nv, "largeblockok");
2395 		}
2396 		if (zap_contains(dp->dp_meta_objset, ds->ds_object,
2397 		    DS_FIELD_RESUME_EMBEDOK) == 0) {
2398 			fnvlist_add_boolean(token_nv, "embedok");
2399 		}
2400 		if (zap_contains(dp->dp_meta_objset, ds->ds_object,
2401 		    DS_FIELD_RESUME_COMPRESSOK) == 0) {
2402 			fnvlist_add_boolean(token_nv, "compressok");
2403 		}
2404 		if (zap_contains(dp->dp_meta_objset, ds->ds_object,
2405 		    DS_FIELD_RESUME_RAWOK) == 0) {
2406 			fnvlist_add_boolean(token_nv, "rawok");
2407 		}
2408 		if (dsl_dataset_feature_is_active(ds,
2409 		    SPA_FEATURE_REDACTED_DATASETS)) {
2410 			uint64_t num_redact_snaps;
2411 			uint64_t *redact_snaps;
2412 			VERIFY(dsl_dataset_get_uint64_array_feature(ds,
2413 			    SPA_FEATURE_REDACTED_DATASETS, &num_redact_snaps,
2414 			    &redact_snaps));
2415 			fnvlist_add_uint64_array(token_nv, "redact_snaps",
2416 			    redact_snaps, num_redact_snaps);
2417 		}
2418 		if (zap_contains(dp->dp_meta_objset, ds->ds_object,
2419 		    DS_FIELD_RESUME_REDACT_BOOKMARK_SNAPS) == 0) {
2420 			uint64_t num_redact_snaps, int_size;
2421 			uint64_t *redact_snaps;
2422 			VERIFY0(zap_length(dp->dp_meta_objset, ds->ds_object,
2423 			    DS_FIELD_RESUME_REDACT_BOOKMARK_SNAPS, &int_size,
2424 			    &num_redact_snaps));
2425 			ASSERT3U(int_size, ==, sizeof (uint64_t));
2426 
2427 			redact_snaps = kmem_alloc(int_size * num_redact_snaps,
2428 			    KM_SLEEP);
2429 			VERIFY0(zap_lookup(dp->dp_meta_objset, ds->ds_object,
2430 			    DS_FIELD_RESUME_REDACT_BOOKMARK_SNAPS, int_size,
2431 			    num_redact_snaps, redact_snaps));
2432 			fnvlist_add_uint64_array(token_nv, "book_redact_snaps",
2433 			    redact_snaps, num_redact_snaps);
2434 			kmem_free(redact_snaps, int_size * num_redact_snaps);
2435 		}
2436 		packed = fnvlist_pack(token_nv, &packed_size);
2437 		fnvlist_free(token_nv);
2438 		compressed = kmem_alloc(packed_size, KM_SLEEP);
2439 
2440 		compressed_size = gzip_compress(packed, compressed,
2441 		    packed_size, packed_size, 6);
2442 
2443 		zio_cksum_t cksum;
2444 		fletcher_4_native_varsize(compressed, compressed_size, &cksum);
2445 
2446 		size_t alloc_size = compressed_size * 2 + 1;
2447 		str = kmem_alloc(alloc_size, KM_SLEEP);
2448 		for (int i = 0; i < compressed_size; i++) {
2449 			size_t offset = i * 2;
2450 			(void) snprintf(str + offset, alloc_size - offset,
2451 		    "%02x", compressed[i]);
2452 		}
2453 		str[compressed_size * 2] = '\0';
2454 		char *propval = kmem_asprintf("%u-%llx-%llx-%s",
2455 		    ZFS_SEND_RESUME_TOKEN_VERSION,
2456 		    (longlong_t)cksum.zc_word[0],
2457 		    (longlong_t)packed_size, str);
2458 		kmem_free(packed, packed_size);
2459 		kmem_free(str, alloc_size);
2460 		kmem_free(compressed, packed_size);
2461 		return (propval);
2462 	}
2463 	return (kmem_strdup(""));
2464 }
2465 
2466 /*
2467  * Returns a string that represents the receive resume stats token of the
2468  * dataset's child. It should be freed with strfree().
2469  */
2470 char *
2471 get_child_receive_stats(dsl_dataset_t *ds)
2472 {
2473 	char recvname[ZFS_MAX_DATASET_NAME_LEN + 6];
2474 	dsl_dataset_t *recv_ds;
2475 	dsl_dataset_name(ds, recvname);
2476 	if (strlcat(recvname, "/", sizeof (recvname)) <
2477 	    sizeof (recvname) &&
2478 	    strlcat(recvname, recv_clone_name, sizeof (recvname)) <
2479 	    sizeof (recvname) &&
2480 	    dsl_dataset_hold(ds->ds_dir->dd_pool, recvname, FTAG,
2481 	    &recv_ds)  == 0) {
2482 		char *propval = get_receive_resume_stats_impl(recv_ds);
2483 		dsl_dataset_rele(recv_ds, FTAG);
2484 		return (propval);
2485 	}
2486 	return (kmem_strdup(""));
2487 }
2488 
2489 static void
2490 get_receive_resume_stats(dsl_dataset_t *ds, nvlist_t *nv)
2491 {
2492 	char *propval = get_receive_resume_stats_impl(ds);
2493 	if (strcmp(propval, "") != 0) {
2494 		dsl_prop_nvlist_add_string(nv,
2495 		    ZFS_PROP_RECEIVE_RESUME_TOKEN, propval);
2496 	} else {
2497 		char *childval = get_child_receive_stats(ds);
2498 		if (strcmp(childval, "") != 0) {
2499 			dsl_prop_nvlist_add_string(nv,
2500 			    ZFS_PROP_RECEIVE_RESUME_TOKEN, childval);
2501 		}
2502 		kmem_strfree(childval);
2503 	}
2504 	kmem_strfree(propval);
2505 }
2506 
2507 uint64_t
2508 dsl_get_refratio(dsl_dataset_t *ds)
2509 {
2510 	uint64_t ratio = dsl_dataset_phys(ds)->ds_compressed_bytes == 0 ? 100 :
2511 	    (dsl_dataset_phys(ds)->ds_uncompressed_bytes * 100 /
2512 	    dsl_dataset_phys(ds)->ds_compressed_bytes);
2513 	return (ratio);
2514 }
2515 
2516 uint64_t
2517 dsl_get_logicalreferenced(dsl_dataset_t *ds)
2518 {
2519 	return (dsl_dataset_phys(ds)->ds_uncompressed_bytes);
2520 }
2521 
2522 uint64_t
2523 dsl_get_compressratio(dsl_dataset_t *ds)
2524 {
2525 	if (ds->ds_is_snapshot) {
2526 		return (dsl_get_refratio(ds));
2527 	} else {
2528 		dsl_dir_t *dd = ds->ds_dir;
2529 		mutex_enter(&dd->dd_lock);
2530 		uint64_t val = dsl_dir_get_compressratio(dd);
2531 		mutex_exit(&dd->dd_lock);
2532 		return (val);
2533 	}
2534 }
2535 
2536 uint64_t
2537 dsl_get_used(dsl_dataset_t *ds)
2538 {
2539 	if (ds->ds_is_snapshot) {
2540 		return (dsl_dataset_phys(ds)->ds_unique_bytes);
2541 	} else {
2542 		dsl_dir_t *dd = ds->ds_dir;
2543 		mutex_enter(&dd->dd_lock);
2544 		uint64_t val = dsl_dir_get_used(dd);
2545 		mutex_exit(&dd->dd_lock);
2546 		return (val);
2547 	}
2548 }
2549 
2550 uint64_t
2551 dsl_get_creation(dsl_dataset_t *ds)
2552 {
2553 	return (dsl_dataset_phys(ds)->ds_creation_time);
2554 }
2555 
2556 uint64_t
2557 dsl_get_creationtxg(dsl_dataset_t *ds)
2558 {
2559 	return (dsl_dataset_phys(ds)->ds_creation_txg);
2560 }
2561 
2562 uint64_t
2563 dsl_get_refquota(dsl_dataset_t *ds)
2564 {
2565 	return (ds->ds_quota);
2566 }
2567 
2568 uint64_t
2569 dsl_get_refreservation(dsl_dataset_t *ds)
2570 {
2571 	return (ds->ds_reserved);
2572 }
2573 
2574 uint64_t
2575 dsl_get_guid(dsl_dataset_t *ds)
2576 {
2577 	return (dsl_dataset_phys(ds)->ds_guid);
2578 }
2579 
2580 uint64_t
2581 dsl_get_unique(dsl_dataset_t *ds)
2582 {
2583 	return (dsl_dataset_phys(ds)->ds_unique_bytes);
2584 }
2585 
2586 uint64_t
2587 dsl_get_objsetid(dsl_dataset_t *ds)
2588 {
2589 	return (ds->ds_object);
2590 }
2591 
2592 uint64_t
2593 dsl_get_userrefs(dsl_dataset_t *ds)
2594 {
2595 	return (ds->ds_userrefs);
2596 }
2597 
2598 uint64_t
2599 dsl_get_defer_destroy(dsl_dataset_t *ds)
2600 {
2601 	return (DS_IS_DEFER_DESTROY(ds) ? 1 : 0);
2602 }
2603 
2604 uint64_t
2605 dsl_get_referenced(dsl_dataset_t *ds)
2606 {
2607 	return (dsl_dataset_phys(ds)->ds_referenced_bytes);
2608 }
2609 
2610 uint64_t
2611 dsl_get_numclones(dsl_dataset_t *ds)
2612 {
2613 	ASSERT(ds->ds_is_snapshot);
2614 	return (dsl_dataset_phys(ds)->ds_num_children - 1);
2615 }
2616 
2617 uint64_t
2618 dsl_get_inconsistent(dsl_dataset_t *ds)
2619 {
2620 	return ((dsl_dataset_phys(ds)->ds_flags & DS_FLAG_INCONSISTENT) ?
2621 	    1 : 0);
2622 }
2623 
2624 uint64_t
2625 dsl_get_redacted(dsl_dataset_t *ds)
2626 {
2627 	return (dsl_dataset_feature_is_active(ds,
2628 	    SPA_FEATURE_REDACTED_DATASETS));
2629 }
2630 
2631 uint64_t
2632 dsl_get_available(dsl_dataset_t *ds)
2633 {
2634 	uint64_t refdbytes = dsl_get_referenced(ds);
2635 	uint64_t availbytes = dsl_dir_space_available(ds->ds_dir,
2636 	    NULL, 0, TRUE);
2637 	if (ds->ds_reserved > dsl_dataset_phys(ds)->ds_unique_bytes) {
2638 		availbytes +=
2639 		    ds->ds_reserved - dsl_dataset_phys(ds)->ds_unique_bytes;
2640 	}
2641 	if (ds->ds_quota != 0) {
2642 		/*
2643 		 * Adjust available bytes according to refquota
2644 		 */
2645 		if (refdbytes < ds->ds_quota) {
2646 			availbytes = MIN(availbytes,
2647 			    ds->ds_quota - refdbytes);
2648 		} else {
2649 			availbytes = 0;
2650 		}
2651 	}
2652 	return (availbytes);
2653 }
2654 
2655 int
2656 dsl_get_written(dsl_dataset_t *ds, uint64_t *written)
2657 {
2658 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2659 	dsl_dataset_t *prev;
2660 	int err = dsl_dataset_hold_obj(dp,
2661 	    dsl_dataset_phys(ds)->ds_prev_snap_obj, FTAG, &prev);
2662 	if (err == 0) {
2663 		uint64_t comp, uncomp;
2664 		err = dsl_dataset_space_written(prev, ds, written,
2665 		    &comp, &uncomp);
2666 		dsl_dataset_rele(prev, FTAG);
2667 	}
2668 	return (err);
2669 }
2670 
2671 /*
2672  * 'snap' should be a buffer of size ZFS_MAX_DATASET_NAME_LEN.
2673  */
2674 int
2675 dsl_get_prev_snap(dsl_dataset_t *ds, char *snap)
2676 {
2677 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2678 	if (ds->ds_prev != NULL && ds->ds_prev != dp->dp_origin_snap) {
2679 		dsl_dataset_name(ds->ds_prev, snap);
2680 		return (0);
2681 	} else {
2682 		return (SET_ERROR(ENOENT));
2683 	}
2684 }
2685 
2686 void
2687 dsl_get_redact_snaps(dsl_dataset_t *ds, nvlist_t *propval)
2688 {
2689 	uint64_t nsnaps;
2690 	uint64_t *snaps;
2691 	if (dsl_dataset_get_uint64_array_feature(ds,
2692 	    SPA_FEATURE_REDACTED_DATASETS, &nsnaps, &snaps)) {
2693 		fnvlist_add_uint64_array(propval, ZPROP_VALUE, snaps,
2694 		    nsnaps);
2695 	}
2696 }
2697 
2698 /*
2699  * Returns the mountpoint property and source for the given dataset in the value
2700  * and source buffers. The value buffer must be at least as large as MAXPATHLEN
2701  * and the source buffer as least as large a ZFS_MAX_DATASET_NAME_LEN.
2702  * Returns 0 on success and an error on failure.
2703  */
2704 int
2705 dsl_get_mountpoint(dsl_dataset_t *ds, const char *dsname, char *value,
2706     char *source)
2707 {
2708 	int error;
2709 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2710 
2711 	/* Retrieve the mountpoint value stored in the zap object */
2712 	error = dsl_prop_get_ds(ds, zfs_prop_to_name(ZFS_PROP_MOUNTPOINT), 1,
2713 	    ZAP_MAXVALUELEN, value, source);
2714 	if (error != 0) {
2715 		return (error);
2716 	}
2717 
2718 	/*
2719 	 * Process the dsname and source to find the full mountpoint string.
2720 	 * Can be skipped for 'legacy' or 'none'.
2721 	 */
2722 	if (value[0] == '/') {
2723 		char *buf = kmem_alloc(ZAP_MAXVALUELEN, KM_SLEEP);
2724 		char *root = buf;
2725 		const char *relpath;
2726 
2727 		/*
2728 		 * If we inherit the mountpoint, even from a dataset
2729 		 * with a received value, the source will be the path of
2730 		 * the dataset we inherit from. If source is
2731 		 * ZPROP_SOURCE_VAL_RECVD, the received value is not
2732 		 * inherited.
2733 		 */
2734 		if (strcmp(source, ZPROP_SOURCE_VAL_RECVD) == 0) {
2735 			relpath = "";
2736 		} else {
2737 			ASSERT0(strncmp(dsname, source, strlen(source)));
2738 			relpath = dsname + strlen(source);
2739 			if (relpath[0] == '/')
2740 				relpath++;
2741 		}
2742 
2743 		spa_altroot(dp->dp_spa, root, ZAP_MAXVALUELEN);
2744 
2745 		/*
2746 		 * Special case an alternate root of '/'. This will
2747 		 * avoid having multiple leading slashes in the
2748 		 * mountpoint path.
2749 		 */
2750 		if (strcmp(root, "/") == 0)
2751 			root++;
2752 
2753 		/*
2754 		 * If the mountpoint is '/' then skip over this
2755 		 * if we are obtaining either an alternate root or
2756 		 * an inherited mountpoint.
2757 		 */
2758 		char *mnt = value;
2759 		if (value[1] == '\0' && (root[0] != '\0' ||
2760 		    relpath[0] != '\0'))
2761 			mnt = value + 1;
2762 
2763 		if (relpath[0] == '\0') {
2764 			(void) snprintf(value, ZAP_MAXVALUELEN, "%s%s",
2765 			    root, mnt);
2766 		} else {
2767 			(void) snprintf(value, ZAP_MAXVALUELEN, "%s%s%s%s",
2768 			    root, mnt, relpath[0] == '@' ? "" : "/",
2769 			    relpath);
2770 		}
2771 		kmem_free(buf, ZAP_MAXVALUELEN);
2772 	}
2773 
2774 	return (0);
2775 }
2776 
2777 void
2778 dsl_dataset_stats(dsl_dataset_t *ds, nvlist_t *nv)
2779 {
2780 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2781 
2782 	ASSERT(dsl_pool_config_held(dp));
2783 
2784 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFRATIO,
2785 	    dsl_get_refratio(ds));
2786 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_LOGICALREFERENCED,
2787 	    dsl_get_logicalreferenced(ds));
2788 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_COMPRESSRATIO,
2789 	    dsl_get_compressratio(ds));
2790 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USED,
2791 	    dsl_get_used(ds));
2792 
2793 	if (ds->ds_is_snapshot) {
2794 		get_clones_stat(ds, nv);
2795 	} else {
2796 		char buf[ZFS_MAX_DATASET_NAME_LEN];
2797 		if (dsl_get_prev_snap(ds, buf) == 0)
2798 			dsl_prop_nvlist_add_string(nv, ZFS_PROP_PREV_SNAP,
2799 			    buf);
2800 		dsl_dir_stats(ds->ds_dir, nv);
2801 	}
2802 
2803 	nvlist_t *propval = fnvlist_alloc();
2804 	dsl_get_redact_snaps(ds, propval);
2805 	fnvlist_add_nvlist(nv, zfs_prop_to_name(ZFS_PROP_REDACT_SNAPS),
2806 	    propval);
2807 	nvlist_free(propval);
2808 
2809 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_AVAILABLE,
2810 	    dsl_get_available(ds));
2811 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFERENCED,
2812 	    dsl_get_referenced(ds));
2813 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_CREATION,
2814 	    dsl_get_creation(ds));
2815 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_CREATETXG,
2816 	    dsl_get_creationtxg(ds));
2817 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFQUOTA,
2818 	    dsl_get_refquota(ds));
2819 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFRESERVATION,
2820 	    dsl_get_refreservation(ds));
2821 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_GUID,
2822 	    dsl_get_guid(ds));
2823 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_UNIQUE,
2824 	    dsl_get_unique(ds));
2825 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_OBJSETID,
2826 	    dsl_get_objsetid(ds));
2827 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USERREFS,
2828 	    dsl_get_userrefs(ds));
2829 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_DEFER_DESTROY,
2830 	    dsl_get_defer_destroy(ds));
2831 	dsl_dataset_crypt_stats(ds, nv);
2832 
2833 	if (dsl_dataset_phys(ds)->ds_prev_snap_obj != 0) {
2834 		uint64_t written;
2835 		if (dsl_get_written(ds, &written) == 0) {
2836 			dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_WRITTEN,
2837 			    written);
2838 		}
2839 	}
2840 
2841 	if (!dsl_dataset_is_snapshot(ds)) {
2842 		/*
2843 		 * A failed "newfs" (e.g. full) resumable receive leaves
2844 		 * the stats set on this dataset.  Check here for the prop.
2845 		 */
2846 		get_receive_resume_stats(ds, nv);
2847 
2848 		/*
2849 		 * A failed incremental resumable receive leaves the
2850 		 * stats set on our child named "%recv".  Check the child
2851 		 * for the prop.
2852 		 */
2853 		/* 6 extra bytes for /%recv */
2854 		char recvname[ZFS_MAX_DATASET_NAME_LEN + 6];
2855 		dsl_dataset_t *recv_ds;
2856 		dsl_dataset_name(ds, recvname);
2857 		if (strlcat(recvname, "/", sizeof (recvname)) <
2858 		    sizeof (recvname) &&
2859 		    strlcat(recvname, recv_clone_name, sizeof (recvname)) <
2860 		    sizeof (recvname) &&
2861 		    dsl_dataset_hold(dp, recvname, FTAG, &recv_ds) == 0) {
2862 			get_receive_resume_stats(recv_ds, nv);
2863 			dsl_dataset_rele(recv_ds, FTAG);
2864 		}
2865 	}
2866 }
2867 
2868 void
2869 dsl_dataset_fast_stat(dsl_dataset_t *ds, dmu_objset_stats_t *stat)
2870 {
2871 	dsl_pool_t *dp __maybe_unused = ds->ds_dir->dd_pool;
2872 	ASSERT(dsl_pool_config_held(dp));
2873 
2874 	stat->dds_creation_txg = dsl_get_creationtxg(ds);
2875 	stat->dds_inconsistent = dsl_get_inconsistent(ds);
2876 	stat->dds_guid = dsl_get_guid(ds);
2877 	stat->dds_redacted = dsl_get_redacted(ds);
2878 	stat->dds_origin[0] = '\0';
2879 	if (ds->ds_is_snapshot) {
2880 		stat->dds_is_snapshot = B_TRUE;
2881 		stat->dds_num_clones = dsl_get_numclones(ds);
2882 	} else {
2883 		stat->dds_is_snapshot = B_FALSE;
2884 		stat->dds_num_clones = 0;
2885 
2886 		if (dsl_dir_is_clone(ds->ds_dir)) {
2887 			dsl_dir_get_origin(ds->ds_dir, stat->dds_origin);
2888 		}
2889 	}
2890 }
2891 
2892 uint64_t
2893 dsl_dataset_fsid_guid(dsl_dataset_t *ds)
2894 {
2895 	return (ds->ds_fsid_guid);
2896 }
2897 
2898 void
2899 dsl_dataset_space(dsl_dataset_t *ds,
2900     uint64_t *refdbytesp, uint64_t *availbytesp,
2901     uint64_t *usedobjsp, uint64_t *availobjsp)
2902 {
2903 	*refdbytesp = dsl_dataset_phys(ds)->ds_referenced_bytes;
2904 	*availbytesp = dsl_dir_space_available(ds->ds_dir, NULL, 0, TRUE);
2905 	if (ds->ds_reserved > dsl_dataset_phys(ds)->ds_unique_bytes)
2906 		*availbytesp +=
2907 		    ds->ds_reserved - dsl_dataset_phys(ds)->ds_unique_bytes;
2908 	if (ds->ds_quota != 0) {
2909 		/*
2910 		 * Adjust available bytes according to refquota
2911 		 */
2912 		if (*refdbytesp < ds->ds_quota)
2913 			*availbytesp = MIN(*availbytesp,
2914 			    ds->ds_quota - *refdbytesp);
2915 		else
2916 			*availbytesp = 0;
2917 	}
2918 	rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
2919 	*usedobjsp = BP_GET_FILL(&dsl_dataset_phys(ds)->ds_bp);
2920 	rrw_exit(&ds->ds_bp_rwlock, FTAG);
2921 	*availobjsp = DN_MAX_OBJECT - *usedobjsp;
2922 }
2923 
2924 boolean_t
2925 dsl_dataset_modified_since_snap(dsl_dataset_t *ds, dsl_dataset_t *snap)
2926 {
2927 	dsl_pool_t *dp __maybe_unused = ds->ds_dir->dd_pool;
2928 	uint64_t birth;
2929 
2930 	ASSERT(dsl_pool_config_held(dp));
2931 	if (snap == NULL)
2932 		return (B_FALSE);
2933 	rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
2934 	birth = dsl_dataset_get_blkptr(ds)->blk_birth;
2935 	rrw_exit(&ds->ds_bp_rwlock, FTAG);
2936 	if (birth > dsl_dataset_phys(snap)->ds_creation_txg) {
2937 		objset_t *os, *os_snap;
2938 		/*
2939 		 * It may be that only the ZIL differs, because it was
2940 		 * reset in the head.  Don't count that as being
2941 		 * modified.
2942 		 */
2943 		if (dmu_objset_from_ds(ds, &os) != 0)
2944 			return (B_TRUE);
2945 		if (dmu_objset_from_ds(snap, &os_snap) != 0)
2946 			return (B_TRUE);
2947 		return (bcmp(&os->os_phys->os_meta_dnode,
2948 		    &os_snap->os_phys->os_meta_dnode,
2949 		    sizeof (os->os_phys->os_meta_dnode)) != 0);
2950 	}
2951 	return (B_FALSE);
2952 }
2953 
2954 typedef struct dsl_dataset_rename_snapshot_arg {
2955 	const char *ddrsa_fsname;
2956 	const char *ddrsa_oldsnapname;
2957 	const char *ddrsa_newsnapname;
2958 	boolean_t ddrsa_recursive;
2959 	dmu_tx_t *ddrsa_tx;
2960 } dsl_dataset_rename_snapshot_arg_t;
2961 
2962 /* ARGSUSED */
2963 static int
2964 dsl_dataset_rename_snapshot_check_impl(dsl_pool_t *dp,
2965     dsl_dataset_t *hds, void *arg)
2966 {
2967 	dsl_dataset_rename_snapshot_arg_t *ddrsa = arg;
2968 	int error;
2969 	uint64_t val;
2970 
2971 	error = dsl_dataset_snap_lookup(hds, ddrsa->ddrsa_oldsnapname, &val);
2972 	if (error != 0) {
2973 		/* ignore nonexistent snapshots */
2974 		return (error == ENOENT ? 0 : error);
2975 	}
2976 
2977 	/* new name should not exist */
2978 	error = dsl_dataset_snap_lookup(hds, ddrsa->ddrsa_newsnapname, &val);
2979 	if (error == 0)
2980 		error = SET_ERROR(EEXIST);
2981 	else if (error == ENOENT)
2982 		error = 0;
2983 
2984 	/* dataset name + 1 for the "@" + the new snapshot name must fit */
2985 	if (dsl_dir_namelen(hds->ds_dir) + 1 +
2986 	    strlen(ddrsa->ddrsa_newsnapname) >= ZFS_MAX_DATASET_NAME_LEN)
2987 		error = SET_ERROR(ENAMETOOLONG);
2988 
2989 	return (error);
2990 }
2991 
2992 static int
2993 dsl_dataset_rename_snapshot_check(void *arg, dmu_tx_t *tx)
2994 {
2995 	dsl_dataset_rename_snapshot_arg_t *ddrsa = arg;
2996 	dsl_pool_t *dp = dmu_tx_pool(tx);
2997 	dsl_dataset_t *hds;
2998 	int error;
2999 
3000 	error = dsl_dataset_hold(dp, ddrsa->ddrsa_fsname, FTAG, &hds);
3001 	if (error != 0)
3002 		return (error);
3003 
3004 	if (ddrsa->ddrsa_recursive) {
3005 		error = dmu_objset_find_dp(dp, hds->ds_dir->dd_object,
3006 		    dsl_dataset_rename_snapshot_check_impl, ddrsa,
3007 		    DS_FIND_CHILDREN);
3008 	} else {
3009 		error = dsl_dataset_rename_snapshot_check_impl(dp, hds, ddrsa);
3010 	}
3011 	dsl_dataset_rele(hds, FTAG);
3012 	return (error);
3013 }
3014 
3015 static int
3016 dsl_dataset_rename_snapshot_sync_impl(dsl_pool_t *dp,
3017     dsl_dataset_t *hds, void *arg)
3018 {
3019 	dsl_dataset_rename_snapshot_arg_t *ddrsa = arg;
3020 	dsl_dataset_t *ds;
3021 	uint64_t val;
3022 	dmu_tx_t *tx = ddrsa->ddrsa_tx;
3023 	int error;
3024 
3025 	error = dsl_dataset_snap_lookup(hds, ddrsa->ddrsa_oldsnapname, &val);
3026 	ASSERT(error == 0 || error == ENOENT);
3027 	if (error == ENOENT) {
3028 		/* ignore nonexistent snapshots */
3029 		return (0);
3030 	}
3031 
3032 	VERIFY0(dsl_dataset_hold_obj(dp, val, FTAG, &ds));
3033 
3034 	/* log before we change the name */
3035 	spa_history_log_internal_ds(ds, "rename", tx,
3036 	    "-> @%s", ddrsa->ddrsa_newsnapname);
3037 
3038 	VERIFY0(dsl_dataset_snap_remove(hds, ddrsa->ddrsa_oldsnapname, tx,
3039 	    B_FALSE));
3040 	mutex_enter(&ds->ds_lock);
3041 	(void) strlcpy(ds->ds_snapname, ddrsa->ddrsa_newsnapname,
3042 	    sizeof (ds->ds_snapname));
3043 	mutex_exit(&ds->ds_lock);
3044 	VERIFY0(zap_add(dp->dp_meta_objset,
3045 	    dsl_dataset_phys(hds)->ds_snapnames_zapobj,
3046 	    ds->ds_snapname, 8, 1, &ds->ds_object, tx));
3047 	zvol_rename_minors(dp->dp_spa, ddrsa->ddrsa_oldsnapname,
3048 	    ddrsa->ddrsa_newsnapname, B_TRUE);
3049 
3050 	dsl_dataset_rele(ds, FTAG);
3051 	return (0);
3052 }
3053 
3054 static void
3055 dsl_dataset_rename_snapshot_sync(void *arg, dmu_tx_t *tx)
3056 {
3057 	dsl_dataset_rename_snapshot_arg_t *ddrsa = arg;
3058 	dsl_pool_t *dp = dmu_tx_pool(tx);
3059 	dsl_dataset_t *hds = NULL;
3060 
3061 	VERIFY0(dsl_dataset_hold(dp, ddrsa->ddrsa_fsname, FTAG, &hds));
3062 	ddrsa->ddrsa_tx = tx;
3063 	if (ddrsa->ddrsa_recursive) {
3064 		VERIFY0(dmu_objset_find_dp(dp, hds->ds_dir->dd_object,
3065 		    dsl_dataset_rename_snapshot_sync_impl, ddrsa,
3066 		    DS_FIND_CHILDREN));
3067 	} else {
3068 		VERIFY0(dsl_dataset_rename_snapshot_sync_impl(dp, hds, ddrsa));
3069 	}
3070 	dsl_dataset_rele(hds, FTAG);
3071 }
3072 
3073 int
3074 dsl_dataset_rename_snapshot(const char *fsname,
3075     const char *oldsnapname, const char *newsnapname, boolean_t recursive)
3076 {
3077 	dsl_dataset_rename_snapshot_arg_t ddrsa;
3078 
3079 	ddrsa.ddrsa_fsname = fsname;
3080 	ddrsa.ddrsa_oldsnapname = oldsnapname;
3081 	ddrsa.ddrsa_newsnapname = newsnapname;
3082 	ddrsa.ddrsa_recursive = recursive;
3083 
3084 	return (dsl_sync_task(fsname, dsl_dataset_rename_snapshot_check,
3085 	    dsl_dataset_rename_snapshot_sync, &ddrsa,
3086 	    1, ZFS_SPACE_CHECK_RESERVED));
3087 }
3088 
3089 /*
3090  * If we're doing an ownership handoff, we need to make sure that there is
3091  * only one long hold on the dataset.  We're not allowed to change anything here
3092  * so we don't permanently release the long hold or regular hold here.  We want
3093  * to do this only when syncing to avoid the dataset unexpectedly going away
3094  * when we release the long hold.
3095  */
3096 static int
3097 dsl_dataset_handoff_check(dsl_dataset_t *ds, void *owner, dmu_tx_t *tx)
3098 {
3099 	boolean_t held = B_FALSE;
3100 
3101 	if (!dmu_tx_is_syncing(tx))
3102 		return (0);
3103 
3104 	dsl_dir_t *dd = ds->ds_dir;
3105 	mutex_enter(&dd->dd_activity_lock);
3106 	uint64_t holds = zfs_refcount_count(&ds->ds_longholds) -
3107 	    (owner != NULL ? 1 : 0);
3108 	/*
3109 	 * The value of dd_activity_waiters can chance as soon as we drop the
3110 	 * lock, but we're fine with that; new waiters coming in or old
3111 	 * waiters leaving doesn't cause problems, since we're going to cancel
3112 	 * waiters later anyway. The goal of this check is to verify that no
3113 	 * non-waiters have long-holds, and all new long-holds will be
3114 	 * prevented because we're holding the pool config as writer.
3115 	 */
3116 	if (holds != dd->dd_activity_waiters)
3117 		held = B_TRUE;
3118 	mutex_exit(&dd->dd_activity_lock);
3119 
3120 	if (held)
3121 		return (SET_ERROR(EBUSY));
3122 
3123 	return (0);
3124 }
3125 
3126 int
3127 dsl_dataset_rollback_check(void *arg, dmu_tx_t *tx)
3128 {
3129 	dsl_dataset_rollback_arg_t *ddra = arg;
3130 	dsl_pool_t *dp = dmu_tx_pool(tx);
3131 	dsl_dataset_t *ds;
3132 	int64_t unused_refres_delta;
3133 	int error;
3134 
3135 	error = dsl_dataset_hold(dp, ddra->ddra_fsname, FTAG, &ds);
3136 	if (error != 0)
3137 		return (error);
3138 
3139 	/* must not be a snapshot */
3140 	if (ds->ds_is_snapshot) {
3141 		dsl_dataset_rele(ds, FTAG);
3142 		return (SET_ERROR(EINVAL));
3143 	}
3144 
3145 	/* must have a most recent snapshot */
3146 	if (dsl_dataset_phys(ds)->ds_prev_snap_txg < TXG_INITIAL) {
3147 		dsl_dataset_rele(ds, FTAG);
3148 		return (SET_ERROR(ESRCH));
3149 	}
3150 
3151 	/*
3152 	 * No rollback to a snapshot created in the current txg, because
3153 	 * the rollback may dirty the dataset and create blocks that are
3154 	 * not reachable from the rootbp while having a birth txg that
3155 	 * falls into the snapshot's range.
3156 	 */
3157 	if (dmu_tx_is_syncing(tx) &&
3158 	    dsl_dataset_phys(ds)->ds_prev_snap_txg >= tx->tx_txg) {
3159 		dsl_dataset_rele(ds, FTAG);
3160 		return (SET_ERROR(EAGAIN));
3161 	}
3162 
3163 	/*
3164 	 * If the expected target snapshot is specified, then check that
3165 	 * the latest snapshot is it.
3166 	 */
3167 	if (ddra->ddra_tosnap != NULL) {
3168 		dsl_dataset_t *snapds;
3169 
3170 		/* Check if the target snapshot exists at all. */
3171 		error = dsl_dataset_hold(dp, ddra->ddra_tosnap, FTAG, &snapds);
3172 		if (error != 0) {
3173 			/*
3174 			 * ESRCH is used to signal that the target snapshot does
3175 			 * not exist, while ENOENT is used to report that
3176 			 * the rolled back dataset does not exist.
3177 			 * ESRCH is also used to cover other cases where the
3178 			 * target snapshot is not related to the dataset being
3179 			 * rolled back such as being in a different pool.
3180 			 */
3181 			if (error == ENOENT || error == EXDEV)
3182 				error = SET_ERROR(ESRCH);
3183 			dsl_dataset_rele(ds, FTAG);
3184 			return (error);
3185 		}
3186 		ASSERT(snapds->ds_is_snapshot);
3187 
3188 		/* Check if the snapshot is the latest snapshot indeed. */
3189 		if (snapds != ds->ds_prev) {
3190 			/*
3191 			 * Distinguish between the case where the only problem
3192 			 * is intervening snapshots (EEXIST) vs the snapshot
3193 			 * not being a valid target for rollback (ESRCH).
3194 			 */
3195 			if (snapds->ds_dir == ds->ds_dir ||
3196 			    (dsl_dir_is_clone(ds->ds_dir) &&
3197 			    dsl_dir_phys(ds->ds_dir)->dd_origin_obj ==
3198 			    snapds->ds_object)) {
3199 				error = SET_ERROR(EEXIST);
3200 			} else {
3201 				error = SET_ERROR(ESRCH);
3202 			}
3203 			dsl_dataset_rele(snapds, FTAG);
3204 			dsl_dataset_rele(ds, FTAG);
3205 			return (error);
3206 		}
3207 		dsl_dataset_rele(snapds, FTAG);
3208 	}
3209 
3210 	/* must not have any bookmarks after the most recent snapshot */
3211 	if (dsl_bookmark_latest_txg(ds) >
3212 	    dsl_dataset_phys(ds)->ds_prev_snap_txg) {
3213 		dsl_dataset_rele(ds, FTAG);
3214 		return (SET_ERROR(EEXIST));
3215 	}
3216 
3217 	error = dsl_dataset_handoff_check(ds, ddra->ddra_owner, tx);
3218 	if (error != 0) {
3219 		dsl_dataset_rele(ds, FTAG);
3220 		return (error);
3221 	}
3222 
3223 	/*
3224 	 * Check if the snap we are rolling back to uses more than
3225 	 * the refquota.
3226 	 */
3227 	if (ds->ds_quota != 0 &&
3228 	    dsl_dataset_phys(ds->ds_prev)->ds_referenced_bytes > ds->ds_quota) {
3229 		dsl_dataset_rele(ds, FTAG);
3230 		return (SET_ERROR(EDQUOT));
3231 	}
3232 
3233 	/*
3234 	 * When we do the clone swap, we will temporarily use more space
3235 	 * due to the refreservation (the head will no longer have any
3236 	 * unique space, so the entire amount of the refreservation will need
3237 	 * to be free).  We will immediately destroy the clone, freeing
3238 	 * this space, but the freeing happens over many txg's.
3239 	 */
3240 	unused_refres_delta = (int64_t)MIN(ds->ds_reserved,
3241 	    dsl_dataset_phys(ds)->ds_unique_bytes);
3242 
3243 	if (unused_refres_delta > 0 &&
3244 	    unused_refres_delta >
3245 	    dsl_dir_space_available(ds->ds_dir, NULL, 0, TRUE)) {
3246 		dsl_dataset_rele(ds, FTAG);
3247 		return (SET_ERROR(ENOSPC));
3248 	}
3249 
3250 	dsl_dataset_rele(ds, FTAG);
3251 	return (0);
3252 }
3253 
3254 void
3255 dsl_dataset_rollback_sync(void *arg, dmu_tx_t *tx)
3256 {
3257 	dsl_dataset_rollback_arg_t *ddra = arg;
3258 	dsl_pool_t *dp = dmu_tx_pool(tx);
3259 	dsl_dataset_t *ds, *clone;
3260 	uint64_t cloneobj;
3261 	char namebuf[ZFS_MAX_DATASET_NAME_LEN];
3262 
3263 	VERIFY0(dsl_dataset_hold(dp, ddra->ddra_fsname, FTAG, &ds));
3264 
3265 	dsl_dataset_name(ds->ds_prev, namebuf);
3266 	fnvlist_add_string(ddra->ddra_result, "target", namebuf);
3267 
3268 	cloneobj = dsl_dataset_create_sync(ds->ds_dir, "%rollback",
3269 	    ds->ds_prev, DS_CREATE_FLAG_NODIRTY, kcred, NULL, tx);
3270 
3271 	VERIFY0(dsl_dataset_hold_obj(dp, cloneobj, FTAG, &clone));
3272 
3273 	dsl_dataset_clone_swap_sync_impl(clone, ds, tx);
3274 	dsl_dataset_zero_zil(ds, tx);
3275 
3276 	dsl_destroy_head_sync_impl(clone, tx);
3277 
3278 	dsl_dataset_rele(clone, FTAG);
3279 	dsl_dataset_rele(ds, FTAG);
3280 }
3281 
3282 /*
3283  * Rolls back the given filesystem or volume to the most recent snapshot.
3284  * The name of the most recent snapshot will be returned under key "target"
3285  * in the result nvlist.
3286  *
3287  * If owner != NULL:
3288  * - The existing dataset MUST be owned by the specified owner at entry
3289  * - Upon return, dataset will still be held by the same owner, whether we
3290  *   succeed or not.
3291  *
3292  * This mode is required any time the existing filesystem is mounted.  See
3293  * notes above zfs_suspend_fs() for further details.
3294  */
3295 int
3296 dsl_dataset_rollback(const char *fsname, const char *tosnap, void *owner,
3297     nvlist_t *result)
3298 {
3299 	dsl_dataset_rollback_arg_t ddra;
3300 
3301 	ddra.ddra_fsname = fsname;
3302 	ddra.ddra_tosnap = tosnap;
3303 	ddra.ddra_owner = owner;
3304 	ddra.ddra_result = result;
3305 
3306 	return (dsl_sync_task(fsname, dsl_dataset_rollback_check,
3307 	    dsl_dataset_rollback_sync, &ddra,
3308 	    1, ZFS_SPACE_CHECK_RESERVED));
3309 }
3310 
3311 struct promotenode {
3312 	list_node_t link;
3313 	dsl_dataset_t *ds;
3314 };
3315 
3316 static int snaplist_space(list_t *l, uint64_t mintxg, uint64_t *spacep);
3317 static int promote_hold(dsl_dataset_promote_arg_t *ddpa, dsl_pool_t *dp,
3318     void *tag);
3319 static void promote_rele(dsl_dataset_promote_arg_t *ddpa, void *tag);
3320 
3321 int
3322 dsl_dataset_promote_check(void *arg, dmu_tx_t *tx)
3323 {
3324 	dsl_dataset_promote_arg_t *ddpa = arg;
3325 	dsl_pool_t *dp = dmu_tx_pool(tx);
3326 	dsl_dataset_t *hds;
3327 	struct promotenode *snap;
3328 	dsl_dataset_t *origin_ds, *origin_head;
3329 	int err;
3330 	uint64_t unused;
3331 	uint64_t ss_mv_cnt;
3332 	size_t max_snap_len;
3333 	boolean_t conflicting_snaps;
3334 
3335 	err = promote_hold(ddpa, dp, FTAG);
3336 	if (err != 0)
3337 		return (err);
3338 
3339 	hds = ddpa->ddpa_clone;
3340 	max_snap_len = MAXNAMELEN - strlen(ddpa->ddpa_clonename) - 1;
3341 
3342 	if (dsl_dataset_phys(hds)->ds_flags & DS_FLAG_NOPROMOTE) {
3343 		promote_rele(ddpa, FTAG);
3344 		return (SET_ERROR(EXDEV));
3345 	}
3346 
3347 	snap = list_head(&ddpa->shared_snaps);
3348 	origin_head = snap->ds;
3349 	if (snap == NULL) {
3350 		err = SET_ERROR(ENOENT);
3351 		goto out;
3352 	}
3353 	origin_ds = snap->ds;
3354 
3355 	/*
3356 	 * Encrypted clones share a DSL Crypto Key with their origin's dsl dir.
3357 	 * When doing a promote we must make sure the encryption root for
3358 	 * both the target and the target's origin does not change to avoid
3359 	 * needing to rewrap encryption keys
3360 	 */
3361 	err = dsl_dataset_promote_crypt_check(hds->ds_dir, origin_ds->ds_dir);
3362 	if (err != 0)
3363 		goto out;
3364 
3365 	/*
3366 	 * Compute and check the amount of space to transfer.  Since this is
3367 	 * so expensive, don't do the preliminary check.
3368 	 */
3369 	if (!dmu_tx_is_syncing(tx)) {
3370 		promote_rele(ddpa, FTAG);
3371 		return (0);
3372 	}
3373 
3374 	/* compute origin's new unique space */
3375 	snap = list_tail(&ddpa->clone_snaps);
3376 	ASSERT(snap != NULL);
3377 	ASSERT3U(dsl_dataset_phys(snap->ds)->ds_prev_snap_obj, ==,
3378 	    origin_ds->ds_object);
3379 	dsl_deadlist_space_range(&snap->ds->ds_deadlist,
3380 	    dsl_dataset_phys(origin_ds)->ds_prev_snap_txg, UINT64_MAX,
3381 	    &ddpa->unique, &unused, &unused);
3382 
3383 	/*
3384 	 * Walk the snapshots that we are moving
3385 	 *
3386 	 * Compute space to transfer.  Consider the incremental changes
3387 	 * to used by each snapshot:
3388 	 * (my used) = (prev's used) + (blocks born) - (blocks killed)
3389 	 * So each snapshot gave birth to:
3390 	 * (blocks born) = (my used) - (prev's used) + (blocks killed)
3391 	 * So a sequence would look like:
3392 	 * (uN - u(N-1) + kN) + ... + (u1 - u0 + k1) + (u0 - 0 + k0)
3393 	 * Which simplifies to:
3394 	 * uN + kN + kN-1 + ... + k1 + k0
3395 	 * Note however, if we stop before we reach the ORIGIN we get:
3396 	 * uN + kN + kN-1 + ... + kM - uM-1
3397 	 */
3398 	conflicting_snaps = B_FALSE;
3399 	ss_mv_cnt = 0;
3400 	ddpa->used = dsl_dataset_phys(origin_ds)->ds_referenced_bytes;
3401 	ddpa->comp = dsl_dataset_phys(origin_ds)->ds_compressed_bytes;
3402 	ddpa->uncomp = dsl_dataset_phys(origin_ds)->ds_uncompressed_bytes;
3403 	for (snap = list_head(&ddpa->shared_snaps); snap;
3404 	    snap = list_next(&ddpa->shared_snaps, snap)) {
3405 		uint64_t val, dlused, dlcomp, dluncomp;
3406 		dsl_dataset_t *ds = snap->ds;
3407 
3408 		ss_mv_cnt++;
3409 
3410 		/*
3411 		 * If there are long holds, we won't be able to evict
3412 		 * the objset.
3413 		 */
3414 		if (dsl_dataset_long_held(ds)) {
3415 			err = SET_ERROR(EBUSY);
3416 			goto out;
3417 		}
3418 
3419 		/* Check that the snapshot name does not conflict */
3420 		VERIFY0(dsl_dataset_get_snapname(ds));
3421 		if (strlen(ds->ds_snapname) >= max_snap_len) {
3422 			err = SET_ERROR(ENAMETOOLONG);
3423 			goto out;
3424 		}
3425 		err = dsl_dataset_snap_lookup(hds, ds->ds_snapname, &val);
3426 		if (err == 0) {
3427 			fnvlist_add_boolean(ddpa->err_ds,
3428 			    snap->ds->ds_snapname);
3429 			conflicting_snaps = B_TRUE;
3430 		} else if (err != ENOENT) {
3431 			goto out;
3432 		}
3433 
3434 		/* The very first snapshot does not have a deadlist */
3435 		if (dsl_dataset_phys(ds)->ds_prev_snap_obj == 0)
3436 			continue;
3437 
3438 		dsl_deadlist_space(&ds->ds_deadlist,
3439 		    &dlused, &dlcomp, &dluncomp);
3440 		ddpa->used += dlused;
3441 		ddpa->comp += dlcomp;
3442 		ddpa->uncomp += dluncomp;
3443 	}
3444 
3445 	/*
3446 	 * Check that bookmarks that are being transferred don't have
3447 	 * name conflicts.
3448 	 */
3449 	for (dsl_bookmark_node_t *dbn = avl_first(&origin_head->ds_bookmarks);
3450 	    dbn != NULL && dbn->dbn_phys.zbm_creation_txg <=
3451 	    dsl_dataset_phys(origin_ds)->ds_creation_txg;
3452 	    dbn = AVL_NEXT(&origin_head->ds_bookmarks, dbn)) {
3453 		if (strlen(dbn->dbn_name) >= max_snap_len) {
3454 			err = SET_ERROR(ENAMETOOLONG);
3455 			goto out;
3456 		}
3457 		zfs_bookmark_phys_t bm;
3458 		err = dsl_bookmark_lookup_impl(ddpa->ddpa_clone,
3459 		    dbn->dbn_name, &bm);
3460 
3461 		if (err == 0) {
3462 			fnvlist_add_boolean(ddpa->err_ds, dbn->dbn_name);
3463 			conflicting_snaps = B_TRUE;
3464 		} else if (err == ESRCH) {
3465 			err = 0;
3466 		} else if (err != 0) {
3467 			goto out;
3468 		}
3469 	}
3470 
3471 	/*
3472 	 * In order to return the full list of conflicting snapshots, we check
3473 	 * whether there was a conflict after traversing all of them.
3474 	 */
3475 	if (conflicting_snaps) {
3476 		err = SET_ERROR(EEXIST);
3477 		goto out;
3478 	}
3479 
3480 	/*
3481 	 * If we are a clone of a clone then we never reached ORIGIN,
3482 	 * so we need to subtract out the clone origin's used space.
3483 	 */
3484 	if (ddpa->origin_origin) {
3485 		ddpa->used -=
3486 		    dsl_dataset_phys(ddpa->origin_origin)->ds_referenced_bytes;
3487 		ddpa->comp -=
3488 		    dsl_dataset_phys(ddpa->origin_origin)->ds_compressed_bytes;
3489 		ddpa->uncomp -=
3490 		    dsl_dataset_phys(ddpa->origin_origin)->
3491 		    ds_uncompressed_bytes;
3492 	}
3493 
3494 	/* Check that there is enough space and limit headroom here */
3495 	err = dsl_dir_transfer_possible(origin_ds->ds_dir, hds->ds_dir,
3496 	    0, ss_mv_cnt, ddpa->used, ddpa->cr, ddpa->proc);
3497 	if (err != 0)
3498 		goto out;
3499 
3500 	/*
3501 	 * Compute the amounts of space that will be used by snapshots
3502 	 * after the promotion (for both origin and clone).  For each,
3503 	 * it is the amount of space that will be on all of their
3504 	 * deadlists (that was not born before their new origin).
3505 	 */
3506 	if (dsl_dir_phys(hds->ds_dir)->dd_flags & DD_FLAG_USED_BREAKDOWN) {
3507 		uint64_t space;
3508 
3509 		/*
3510 		 * Note, typically this will not be a clone of a clone,
3511 		 * so dd_origin_txg will be < TXG_INITIAL, so
3512 		 * these snaplist_space() -> dsl_deadlist_space_range()
3513 		 * calls will be fast because they do not have to
3514 		 * iterate over all bps.
3515 		 */
3516 		snap = list_head(&ddpa->origin_snaps);
3517 		if (snap == NULL) {
3518 			err = SET_ERROR(ENOENT);
3519 			goto out;
3520 		}
3521 		err = snaplist_space(&ddpa->shared_snaps,
3522 		    snap->ds->ds_dir->dd_origin_txg, &ddpa->cloneusedsnap);
3523 		if (err != 0)
3524 			goto out;
3525 
3526 		err = snaplist_space(&ddpa->clone_snaps,
3527 		    snap->ds->ds_dir->dd_origin_txg, &space);
3528 		if (err != 0)
3529 			goto out;
3530 		ddpa->cloneusedsnap += space;
3531 	}
3532 	if (dsl_dir_phys(origin_ds->ds_dir)->dd_flags &
3533 	    DD_FLAG_USED_BREAKDOWN) {
3534 		err = snaplist_space(&ddpa->origin_snaps,
3535 		    dsl_dataset_phys(origin_ds)->ds_creation_txg,
3536 		    &ddpa->originusedsnap);
3537 		if (err != 0)
3538 			goto out;
3539 	}
3540 
3541 out:
3542 	promote_rele(ddpa, FTAG);
3543 	return (err);
3544 }
3545 
3546 void
3547 dsl_dataset_promote_sync(void *arg, dmu_tx_t *tx)
3548 {
3549 	dsl_dataset_promote_arg_t *ddpa = arg;
3550 	dsl_pool_t *dp = dmu_tx_pool(tx);
3551 	dsl_dataset_t *hds;
3552 	struct promotenode *snap;
3553 	dsl_dataset_t *origin_ds;
3554 	dsl_dataset_t *origin_head;
3555 	dsl_dir_t *dd;
3556 	dsl_dir_t *odd = NULL;
3557 	uint64_t oldnext_obj;
3558 	int64_t delta;
3559 
3560 	ASSERT(nvlist_empty(ddpa->err_ds));
3561 
3562 	VERIFY0(promote_hold(ddpa, dp, FTAG));
3563 	hds = ddpa->ddpa_clone;
3564 
3565 	ASSERT0(dsl_dataset_phys(hds)->ds_flags & DS_FLAG_NOPROMOTE);
3566 
3567 	snap = list_head(&ddpa->shared_snaps);
3568 	origin_ds = snap->ds;
3569 	dd = hds->ds_dir;
3570 
3571 	snap = list_head(&ddpa->origin_snaps);
3572 	origin_head = snap->ds;
3573 
3574 	/*
3575 	 * We need to explicitly open odd, since origin_ds's dd will be
3576 	 * changing.
3577 	 */
3578 	VERIFY0(dsl_dir_hold_obj(dp, origin_ds->ds_dir->dd_object,
3579 	    NULL, FTAG, &odd));
3580 
3581 	dsl_dataset_promote_crypt_sync(hds->ds_dir, odd, tx);
3582 
3583 	/* change origin's next snap */
3584 	dmu_buf_will_dirty(origin_ds->ds_dbuf, tx);
3585 	oldnext_obj = dsl_dataset_phys(origin_ds)->ds_next_snap_obj;
3586 	snap = list_tail(&ddpa->clone_snaps);
3587 	ASSERT3U(dsl_dataset_phys(snap->ds)->ds_prev_snap_obj, ==,
3588 	    origin_ds->ds_object);
3589 	dsl_dataset_phys(origin_ds)->ds_next_snap_obj = snap->ds->ds_object;
3590 
3591 	/* change the origin's next clone */
3592 	if (dsl_dataset_phys(origin_ds)->ds_next_clones_obj) {
3593 		dsl_dataset_remove_from_next_clones(origin_ds,
3594 		    snap->ds->ds_object, tx);
3595 		VERIFY0(zap_add_int(dp->dp_meta_objset,
3596 		    dsl_dataset_phys(origin_ds)->ds_next_clones_obj,
3597 		    oldnext_obj, tx));
3598 	}
3599 
3600 	/* change origin */
3601 	dmu_buf_will_dirty(dd->dd_dbuf, tx);
3602 	ASSERT3U(dsl_dir_phys(dd)->dd_origin_obj, ==, origin_ds->ds_object);
3603 	dsl_dir_phys(dd)->dd_origin_obj = dsl_dir_phys(odd)->dd_origin_obj;
3604 	dd->dd_origin_txg = origin_head->ds_dir->dd_origin_txg;
3605 	dmu_buf_will_dirty(odd->dd_dbuf, tx);
3606 	dsl_dir_phys(odd)->dd_origin_obj = origin_ds->ds_object;
3607 	origin_head->ds_dir->dd_origin_txg =
3608 	    dsl_dataset_phys(origin_ds)->ds_creation_txg;
3609 
3610 	/* change dd_clone entries */
3611 	if (spa_version(dp->dp_spa) >= SPA_VERSION_DIR_CLONES) {
3612 		VERIFY0(zap_remove_int(dp->dp_meta_objset,
3613 		    dsl_dir_phys(odd)->dd_clones, hds->ds_object, tx));
3614 		VERIFY0(zap_add_int(dp->dp_meta_objset,
3615 		    dsl_dir_phys(ddpa->origin_origin->ds_dir)->dd_clones,
3616 		    hds->ds_object, tx));
3617 
3618 		VERIFY0(zap_remove_int(dp->dp_meta_objset,
3619 		    dsl_dir_phys(ddpa->origin_origin->ds_dir)->dd_clones,
3620 		    origin_head->ds_object, tx));
3621 		if (dsl_dir_phys(dd)->dd_clones == 0) {
3622 			dsl_dir_phys(dd)->dd_clones =
3623 			    zap_create(dp->dp_meta_objset, DMU_OT_DSL_CLONES,
3624 			    DMU_OT_NONE, 0, tx);
3625 		}
3626 		VERIFY0(zap_add_int(dp->dp_meta_objset,
3627 		    dsl_dir_phys(dd)->dd_clones, origin_head->ds_object, tx));
3628 	}
3629 
3630 	/*
3631 	 * Move bookmarks to this dir.
3632 	 */
3633 	dsl_bookmark_node_t *dbn_next;
3634 	for (dsl_bookmark_node_t *dbn = avl_first(&origin_head->ds_bookmarks);
3635 	    dbn != NULL && dbn->dbn_phys.zbm_creation_txg <=
3636 	    dsl_dataset_phys(origin_ds)->ds_creation_txg;
3637 	    dbn = dbn_next) {
3638 		dbn_next = AVL_NEXT(&origin_head->ds_bookmarks, dbn);
3639 
3640 		avl_remove(&origin_head->ds_bookmarks, dbn);
3641 		VERIFY0(zap_remove(dp->dp_meta_objset,
3642 		    origin_head->ds_bookmarks_obj, dbn->dbn_name, tx));
3643 
3644 		dsl_bookmark_node_add(hds, dbn, tx);
3645 	}
3646 
3647 	dsl_bookmark_next_changed(hds, origin_ds, tx);
3648 
3649 	/* move snapshots to this dir */
3650 	for (snap = list_head(&ddpa->shared_snaps); snap;
3651 	    snap = list_next(&ddpa->shared_snaps, snap)) {
3652 		dsl_dataset_t *ds = snap->ds;
3653 
3654 		/*
3655 		 * Property callbacks are registered to a particular
3656 		 * dsl_dir.  Since ours is changing, evict the objset
3657 		 * so that they will be unregistered from the old dsl_dir.
3658 		 */
3659 		if (ds->ds_objset) {
3660 			dmu_objset_evict(ds->ds_objset);
3661 			ds->ds_objset = NULL;
3662 		}
3663 
3664 		/* move snap name entry */
3665 		VERIFY0(dsl_dataset_get_snapname(ds));
3666 		VERIFY0(dsl_dataset_snap_remove(origin_head,
3667 		    ds->ds_snapname, tx, B_TRUE));
3668 		VERIFY0(zap_add(dp->dp_meta_objset,
3669 		    dsl_dataset_phys(hds)->ds_snapnames_zapobj, ds->ds_snapname,
3670 		    8, 1, &ds->ds_object, tx));
3671 		dsl_fs_ss_count_adjust(hds->ds_dir, 1,
3672 		    DD_FIELD_SNAPSHOT_COUNT, tx);
3673 
3674 		/* change containing dsl_dir */
3675 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
3676 		ASSERT3U(dsl_dataset_phys(ds)->ds_dir_obj, ==, odd->dd_object);
3677 		dsl_dataset_phys(ds)->ds_dir_obj = dd->dd_object;
3678 		ASSERT3P(ds->ds_dir, ==, odd);
3679 		dsl_dir_rele(ds->ds_dir, ds);
3680 		VERIFY0(dsl_dir_hold_obj(dp, dd->dd_object,
3681 		    NULL, ds, &ds->ds_dir));
3682 
3683 		/* move any clone references */
3684 		if (dsl_dataset_phys(ds)->ds_next_clones_obj &&
3685 		    spa_version(dp->dp_spa) >= SPA_VERSION_DIR_CLONES) {
3686 			zap_cursor_t zc;
3687 			zap_attribute_t za;
3688 
3689 			for (zap_cursor_init(&zc, dp->dp_meta_objset,
3690 			    dsl_dataset_phys(ds)->ds_next_clones_obj);
3691 			    zap_cursor_retrieve(&zc, &za) == 0;
3692 			    zap_cursor_advance(&zc)) {
3693 				dsl_dataset_t *cnds;
3694 				uint64_t o;
3695 
3696 				if (za.za_first_integer == oldnext_obj) {
3697 					/*
3698 					 * We've already moved the
3699 					 * origin's reference.
3700 					 */
3701 					continue;
3702 				}
3703 
3704 				VERIFY0(dsl_dataset_hold_obj(dp,
3705 				    za.za_first_integer, FTAG, &cnds));
3706 				o = dsl_dir_phys(cnds->ds_dir)->
3707 				    dd_head_dataset_obj;
3708 
3709 				VERIFY0(zap_remove_int(dp->dp_meta_objset,
3710 				    dsl_dir_phys(odd)->dd_clones, o, tx));
3711 				VERIFY0(zap_add_int(dp->dp_meta_objset,
3712 				    dsl_dir_phys(dd)->dd_clones, o, tx));
3713 				dsl_dataset_rele(cnds, FTAG);
3714 			}
3715 			zap_cursor_fini(&zc);
3716 		}
3717 
3718 		ASSERT(!dsl_prop_hascb(ds));
3719 	}
3720 
3721 	/*
3722 	 * Change space accounting.
3723 	 * Note, pa->*usedsnap and dd_used_breakdown[SNAP] will either
3724 	 * both be valid, or both be 0 (resulting in delta == 0).  This
3725 	 * is true for each of {clone,origin} independently.
3726 	 */
3727 
3728 	delta = ddpa->cloneusedsnap -
3729 	    dsl_dir_phys(dd)->dd_used_breakdown[DD_USED_SNAP];
3730 	ASSERT3S(delta, >=, 0);
3731 	ASSERT3U(ddpa->used, >=, delta);
3732 	dsl_dir_diduse_space(dd, DD_USED_SNAP, delta, 0, 0, tx);
3733 	dsl_dir_diduse_space(dd, DD_USED_HEAD,
3734 	    ddpa->used - delta, ddpa->comp, ddpa->uncomp, tx);
3735 
3736 	delta = ddpa->originusedsnap -
3737 	    dsl_dir_phys(odd)->dd_used_breakdown[DD_USED_SNAP];
3738 	ASSERT3S(delta, <=, 0);
3739 	ASSERT3U(ddpa->used, >=, -delta);
3740 	dsl_dir_diduse_space(odd, DD_USED_SNAP, delta, 0, 0, tx);
3741 	dsl_dir_diduse_space(odd, DD_USED_HEAD,
3742 	    -ddpa->used - delta, -ddpa->comp, -ddpa->uncomp, tx);
3743 
3744 	dsl_dataset_phys(origin_ds)->ds_unique_bytes = ddpa->unique;
3745 
3746 	/*
3747 	 * Since livelists are specific to a clone's origin txg, they
3748 	 * are no longer accurate. Destroy the livelist from the clone being
3749 	 * promoted. If the origin dataset is a clone, destroy its livelist
3750 	 * as well.
3751 	 */
3752 	dsl_dir_remove_livelist(dd, tx, B_TRUE);
3753 	dsl_dir_remove_livelist(odd, tx, B_TRUE);
3754 
3755 	/* log history record */
3756 	spa_history_log_internal_ds(hds, "promote", tx, " ");
3757 
3758 	dsl_dir_rele(odd, FTAG);
3759 	promote_rele(ddpa, FTAG);
3760 }
3761 
3762 /*
3763  * Make a list of dsl_dataset_t's for the snapshots between first_obj
3764  * (exclusive) and last_obj (inclusive).  The list will be in reverse
3765  * order (last_obj will be the list_head()).  If first_obj == 0, do all
3766  * snapshots back to this dataset's origin.
3767  */
3768 static int
3769 snaplist_make(dsl_pool_t *dp,
3770     uint64_t first_obj, uint64_t last_obj, list_t *l, void *tag)
3771 {
3772 	uint64_t obj = last_obj;
3773 
3774 	list_create(l, sizeof (struct promotenode),
3775 	    offsetof(struct promotenode, link));
3776 
3777 	while (obj != first_obj) {
3778 		dsl_dataset_t *ds;
3779 		struct promotenode *snap;
3780 		int err;
3781 
3782 		err = dsl_dataset_hold_obj(dp, obj, tag, &ds);
3783 		ASSERT(err != ENOENT);
3784 		if (err != 0)
3785 			return (err);
3786 
3787 		if (first_obj == 0)
3788 			first_obj = dsl_dir_phys(ds->ds_dir)->dd_origin_obj;
3789 
3790 		snap = kmem_alloc(sizeof (*snap), KM_SLEEP);
3791 		snap->ds = ds;
3792 		list_insert_tail(l, snap);
3793 		obj = dsl_dataset_phys(ds)->ds_prev_snap_obj;
3794 	}
3795 
3796 	return (0);
3797 }
3798 
3799 static int
3800 snaplist_space(list_t *l, uint64_t mintxg, uint64_t *spacep)
3801 {
3802 	struct promotenode *snap;
3803 
3804 	*spacep = 0;
3805 	for (snap = list_head(l); snap; snap = list_next(l, snap)) {
3806 		uint64_t used, comp, uncomp;
3807 		dsl_deadlist_space_range(&snap->ds->ds_deadlist,
3808 		    mintxg, UINT64_MAX, &used, &comp, &uncomp);
3809 		*spacep += used;
3810 	}
3811 	return (0);
3812 }
3813 
3814 static void
3815 snaplist_destroy(list_t *l, void *tag)
3816 {
3817 	struct promotenode *snap;
3818 
3819 	if (l == NULL || !list_link_active(&l->list_head))
3820 		return;
3821 
3822 	while ((snap = list_tail(l)) != NULL) {
3823 		list_remove(l, snap);
3824 		dsl_dataset_rele(snap->ds, tag);
3825 		kmem_free(snap, sizeof (*snap));
3826 	}
3827 	list_destroy(l);
3828 }
3829 
3830 static int
3831 promote_hold(dsl_dataset_promote_arg_t *ddpa, dsl_pool_t *dp, void *tag)
3832 {
3833 	int error;
3834 	dsl_dir_t *dd;
3835 	struct promotenode *snap;
3836 
3837 	error = dsl_dataset_hold(dp, ddpa->ddpa_clonename, tag,
3838 	    &ddpa->ddpa_clone);
3839 	if (error != 0)
3840 		return (error);
3841 	dd = ddpa->ddpa_clone->ds_dir;
3842 
3843 	if (ddpa->ddpa_clone->ds_is_snapshot ||
3844 	    !dsl_dir_is_clone(dd)) {
3845 		dsl_dataset_rele(ddpa->ddpa_clone, tag);
3846 		return (SET_ERROR(EINVAL));
3847 	}
3848 
3849 	error = snaplist_make(dp, 0, dsl_dir_phys(dd)->dd_origin_obj,
3850 	    &ddpa->shared_snaps, tag);
3851 	if (error != 0)
3852 		goto out;
3853 
3854 	error = snaplist_make(dp, 0, ddpa->ddpa_clone->ds_object,
3855 	    &ddpa->clone_snaps, tag);
3856 	if (error != 0)
3857 		goto out;
3858 
3859 	snap = list_head(&ddpa->shared_snaps);
3860 	ASSERT3U(snap->ds->ds_object, ==, dsl_dir_phys(dd)->dd_origin_obj);
3861 	error = snaplist_make(dp, dsl_dir_phys(dd)->dd_origin_obj,
3862 	    dsl_dir_phys(snap->ds->ds_dir)->dd_head_dataset_obj,
3863 	    &ddpa->origin_snaps, tag);
3864 	if (error != 0)
3865 		goto out;
3866 
3867 	if (dsl_dir_phys(snap->ds->ds_dir)->dd_origin_obj != 0) {
3868 		error = dsl_dataset_hold_obj(dp,
3869 		    dsl_dir_phys(snap->ds->ds_dir)->dd_origin_obj,
3870 		    tag, &ddpa->origin_origin);
3871 		if (error != 0)
3872 			goto out;
3873 	}
3874 out:
3875 	if (error != 0)
3876 		promote_rele(ddpa, tag);
3877 	return (error);
3878 }
3879 
3880 static void
3881 promote_rele(dsl_dataset_promote_arg_t *ddpa, void *tag)
3882 {
3883 	snaplist_destroy(&ddpa->shared_snaps, tag);
3884 	snaplist_destroy(&ddpa->clone_snaps, tag);
3885 	snaplist_destroy(&ddpa->origin_snaps, tag);
3886 	if (ddpa->origin_origin != NULL)
3887 		dsl_dataset_rele(ddpa->origin_origin, tag);
3888 	dsl_dataset_rele(ddpa->ddpa_clone, tag);
3889 }
3890 
3891 /*
3892  * Promote a clone.
3893  *
3894  * If it fails due to a conflicting snapshot name, "conflsnap" will be filled
3895  * in with the name.  (It must be at least ZFS_MAX_DATASET_NAME_LEN bytes long.)
3896  */
3897 int
3898 dsl_dataset_promote(const char *name, char *conflsnap)
3899 {
3900 	dsl_dataset_promote_arg_t ddpa = { 0 };
3901 	uint64_t numsnaps;
3902 	int error;
3903 	nvpair_t *snap_pair;
3904 	objset_t *os;
3905 
3906 	/*
3907 	 * We will modify space proportional to the number of
3908 	 * snapshots.  Compute numsnaps.
3909 	 */
3910 	error = dmu_objset_hold(name, FTAG, &os);
3911 	if (error != 0)
3912 		return (error);
3913 	error = zap_count(dmu_objset_pool(os)->dp_meta_objset,
3914 	    dsl_dataset_phys(dmu_objset_ds(os))->ds_snapnames_zapobj,
3915 	    &numsnaps);
3916 	dmu_objset_rele(os, FTAG);
3917 	if (error != 0)
3918 		return (error);
3919 
3920 	ddpa.ddpa_clonename = name;
3921 	ddpa.err_ds = fnvlist_alloc();
3922 	ddpa.cr = CRED();
3923 	ddpa.proc = curproc;
3924 
3925 	error = dsl_sync_task(name, dsl_dataset_promote_check,
3926 	    dsl_dataset_promote_sync, &ddpa,
3927 	    2 + numsnaps, ZFS_SPACE_CHECK_RESERVED);
3928 
3929 	/*
3930 	 * Return the first conflicting snapshot found.
3931 	 */
3932 	snap_pair = nvlist_next_nvpair(ddpa.err_ds, NULL);
3933 	if (snap_pair != NULL && conflsnap != NULL)
3934 		(void) strlcpy(conflsnap, nvpair_name(snap_pair),
3935 		    ZFS_MAX_DATASET_NAME_LEN);
3936 
3937 	fnvlist_free(ddpa.err_ds);
3938 	return (error);
3939 }
3940 
3941 int
3942 dsl_dataset_clone_swap_check_impl(dsl_dataset_t *clone,
3943     dsl_dataset_t *origin_head, boolean_t force, void *owner, dmu_tx_t *tx)
3944 {
3945 	/*
3946 	 * "slack" factor for received datasets with refquota set on them.
3947 	 * See the bottom of this function for details on its use.
3948 	 */
3949 	uint64_t refquota_slack = (uint64_t)DMU_MAX_ACCESS *
3950 	    spa_asize_inflation;
3951 	int64_t unused_refres_delta;
3952 
3953 	/* they should both be heads */
3954 	if (clone->ds_is_snapshot ||
3955 	    origin_head->ds_is_snapshot)
3956 		return (SET_ERROR(EINVAL));
3957 
3958 	/* if we are not forcing, the branch point should be just before them */
3959 	if (!force && clone->ds_prev != origin_head->ds_prev)
3960 		return (SET_ERROR(EINVAL));
3961 
3962 	/* clone should be the clone (unless they are unrelated) */
3963 	if (clone->ds_prev != NULL &&
3964 	    clone->ds_prev != clone->ds_dir->dd_pool->dp_origin_snap &&
3965 	    origin_head->ds_dir != clone->ds_prev->ds_dir)
3966 		return (SET_ERROR(EINVAL));
3967 
3968 	/* the clone should be a child of the origin */
3969 	if (clone->ds_dir->dd_parent != origin_head->ds_dir)
3970 		return (SET_ERROR(EINVAL));
3971 
3972 	/* origin_head shouldn't be modified unless 'force' */
3973 	if (!force &&
3974 	    dsl_dataset_modified_since_snap(origin_head, origin_head->ds_prev))
3975 		return (SET_ERROR(ETXTBSY));
3976 
3977 	/* origin_head should have no long holds (e.g. is not mounted) */
3978 	if (dsl_dataset_handoff_check(origin_head, owner, tx))
3979 		return (SET_ERROR(EBUSY));
3980 
3981 	/* check amount of any unconsumed refreservation */
3982 	unused_refres_delta =
3983 	    (int64_t)MIN(origin_head->ds_reserved,
3984 	    dsl_dataset_phys(origin_head)->ds_unique_bytes) -
3985 	    (int64_t)MIN(origin_head->ds_reserved,
3986 	    dsl_dataset_phys(clone)->ds_unique_bytes);
3987 
3988 	if (unused_refres_delta > 0 &&
3989 	    unused_refres_delta >
3990 	    dsl_dir_space_available(origin_head->ds_dir, NULL, 0, TRUE))
3991 		return (SET_ERROR(ENOSPC));
3992 
3993 	/*
3994 	 * The clone can't be too much over the head's refquota.
3995 	 *
3996 	 * To ensure that the entire refquota can be used, we allow one
3997 	 * transaction to exceed the refquota.  Therefore, this check
3998 	 * needs to also allow for the space referenced to be more than the
3999 	 * refquota.  The maximum amount of space that one transaction can use
4000 	 * on disk is DMU_MAX_ACCESS * spa_asize_inflation.  Allowing this
4001 	 * overage ensures that we are able to receive a filesystem that
4002 	 * exceeds the refquota on the source system.
4003 	 *
4004 	 * So that overage is the refquota_slack we use below.
4005 	 */
4006 	if (origin_head->ds_quota != 0 &&
4007 	    dsl_dataset_phys(clone)->ds_referenced_bytes >
4008 	    origin_head->ds_quota + refquota_slack)
4009 		return (SET_ERROR(EDQUOT));
4010 
4011 	return (0);
4012 }
4013 
4014 static void
4015 dsl_dataset_swap_remap_deadlists(dsl_dataset_t *clone,
4016     dsl_dataset_t *origin, dmu_tx_t *tx)
4017 {
4018 	uint64_t clone_remap_dl_obj, origin_remap_dl_obj;
4019 	dsl_pool_t *dp = dmu_tx_pool(tx);
4020 
4021 	ASSERT(dsl_pool_sync_context(dp));
4022 
4023 	clone_remap_dl_obj = dsl_dataset_get_remap_deadlist_object(clone);
4024 	origin_remap_dl_obj = dsl_dataset_get_remap_deadlist_object(origin);
4025 
4026 	if (clone_remap_dl_obj != 0) {
4027 		dsl_deadlist_close(&clone->ds_remap_deadlist);
4028 		dsl_dataset_unset_remap_deadlist_object(clone, tx);
4029 	}
4030 	if (origin_remap_dl_obj != 0) {
4031 		dsl_deadlist_close(&origin->ds_remap_deadlist);
4032 		dsl_dataset_unset_remap_deadlist_object(origin, tx);
4033 	}
4034 
4035 	if (clone_remap_dl_obj != 0) {
4036 		dsl_dataset_set_remap_deadlist_object(origin,
4037 		    clone_remap_dl_obj, tx);
4038 		dsl_deadlist_open(&origin->ds_remap_deadlist,
4039 		    dp->dp_meta_objset, clone_remap_dl_obj);
4040 	}
4041 	if (origin_remap_dl_obj != 0) {
4042 		dsl_dataset_set_remap_deadlist_object(clone,
4043 		    origin_remap_dl_obj, tx);
4044 		dsl_deadlist_open(&clone->ds_remap_deadlist,
4045 		    dp->dp_meta_objset, origin_remap_dl_obj);
4046 	}
4047 }
4048 
4049 void
4050 dsl_dataset_clone_swap_sync_impl(dsl_dataset_t *clone,
4051     dsl_dataset_t *origin_head, dmu_tx_t *tx)
4052 {
4053 	dsl_pool_t *dp = dmu_tx_pool(tx);
4054 	int64_t unused_refres_delta;
4055 
4056 	ASSERT(clone->ds_reserved == 0);
4057 	/*
4058 	 * NOTE: On DEBUG kernels there could be a race between this and
4059 	 * the check function if spa_asize_inflation is adjusted...
4060 	 */
4061 	ASSERT(origin_head->ds_quota == 0 ||
4062 	    dsl_dataset_phys(clone)->ds_unique_bytes <= origin_head->ds_quota +
4063 	    DMU_MAX_ACCESS * spa_asize_inflation);
4064 	ASSERT3P(clone->ds_prev, ==, origin_head->ds_prev);
4065 
4066 	dsl_dir_cancel_waiters(origin_head->ds_dir);
4067 
4068 	/*
4069 	 * Swap per-dataset feature flags.
4070 	 */
4071 	for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
4072 		if (!(spa_feature_table[f].fi_flags &
4073 		    ZFEATURE_FLAG_PER_DATASET)) {
4074 			ASSERT(!dsl_dataset_feature_is_active(clone, f));
4075 			ASSERT(!dsl_dataset_feature_is_active(origin_head, f));
4076 			continue;
4077 		}
4078 
4079 		boolean_t clone_inuse = dsl_dataset_feature_is_active(clone, f);
4080 		void *clone_feature = clone->ds_feature[f];
4081 		boolean_t origin_head_inuse =
4082 		    dsl_dataset_feature_is_active(origin_head, f);
4083 		void *origin_head_feature = origin_head->ds_feature[f];
4084 
4085 		if (clone_inuse)
4086 			dsl_dataset_deactivate_feature_impl(clone, f, tx);
4087 		if (origin_head_inuse)
4088 			dsl_dataset_deactivate_feature_impl(origin_head, f, tx);
4089 
4090 		if (clone_inuse) {
4091 			dsl_dataset_activate_feature(origin_head->ds_object, f,
4092 			    clone_feature, tx);
4093 			origin_head->ds_feature[f] = clone_feature;
4094 		}
4095 		if (origin_head_inuse) {
4096 			dsl_dataset_activate_feature(clone->ds_object, f,
4097 			    origin_head_feature, tx);
4098 			clone->ds_feature[f] = origin_head_feature;
4099 		}
4100 	}
4101 
4102 	dmu_buf_will_dirty(clone->ds_dbuf, tx);
4103 	dmu_buf_will_dirty(origin_head->ds_dbuf, tx);
4104 
4105 	if (clone->ds_objset != NULL) {
4106 		dmu_objset_evict(clone->ds_objset);
4107 		clone->ds_objset = NULL;
4108 	}
4109 
4110 	if (origin_head->ds_objset != NULL) {
4111 		dmu_objset_evict(origin_head->ds_objset);
4112 		origin_head->ds_objset = NULL;
4113 	}
4114 
4115 	unused_refres_delta =
4116 	    (int64_t)MIN(origin_head->ds_reserved,
4117 	    dsl_dataset_phys(origin_head)->ds_unique_bytes) -
4118 	    (int64_t)MIN(origin_head->ds_reserved,
4119 	    dsl_dataset_phys(clone)->ds_unique_bytes);
4120 
4121 	/*
4122 	 * Reset origin's unique bytes.
4123 	 */
4124 	{
4125 		dsl_dataset_t *origin = clone->ds_prev;
4126 		uint64_t comp, uncomp;
4127 
4128 		dmu_buf_will_dirty(origin->ds_dbuf, tx);
4129 		dsl_deadlist_space_range(&clone->ds_deadlist,
4130 		    dsl_dataset_phys(origin)->ds_prev_snap_txg, UINT64_MAX,
4131 		    &dsl_dataset_phys(origin)->ds_unique_bytes, &comp, &uncomp);
4132 	}
4133 
4134 	/* swap blkptrs */
4135 	{
4136 		rrw_enter(&clone->ds_bp_rwlock, RW_WRITER, FTAG);
4137 		rrw_enter(&origin_head->ds_bp_rwlock, RW_WRITER, FTAG);
4138 		blkptr_t tmp;
4139 		tmp = dsl_dataset_phys(origin_head)->ds_bp;
4140 		dsl_dataset_phys(origin_head)->ds_bp =
4141 		    dsl_dataset_phys(clone)->ds_bp;
4142 		dsl_dataset_phys(clone)->ds_bp = tmp;
4143 		rrw_exit(&origin_head->ds_bp_rwlock, FTAG);
4144 		rrw_exit(&clone->ds_bp_rwlock, FTAG);
4145 	}
4146 
4147 	/* set dd_*_bytes */
4148 	{
4149 		int64_t dused, dcomp, duncomp;
4150 		uint64_t cdl_used, cdl_comp, cdl_uncomp;
4151 		uint64_t odl_used, odl_comp, odl_uncomp;
4152 
4153 		ASSERT3U(dsl_dir_phys(clone->ds_dir)->
4154 		    dd_used_breakdown[DD_USED_SNAP], ==, 0);
4155 
4156 		dsl_deadlist_space(&clone->ds_deadlist,
4157 		    &cdl_used, &cdl_comp, &cdl_uncomp);
4158 		dsl_deadlist_space(&origin_head->ds_deadlist,
4159 		    &odl_used, &odl_comp, &odl_uncomp);
4160 
4161 		dused = dsl_dataset_phys(clone)->ds_referenced_bytes +
4162 		    cdl_used -
4163 		    (dsl_dataset_phys(origin_head)->ds_referenced_bytes +
4164 		    odl_used);
4165 		dcomp = dsl_dataset_phys(clone)->ds_compressed_bytes +
4166 		    cdl_comp -
4167 		    (dsl_dataset_phys(origin_head)->ds_compressed_bytes +
4168 		    odl_comp);
4169 		duncomp = dsl_dataset_phys(clone)->ds_uncompressed_bytes +
4170 		    cdl_uncomp -
4171 		    (dsl_dataset_phys(origin_head)->ds_uncompressed_bytes +
4172 		    odl_uncomp);
4173 
4174 		dsl_dir_diduse_space(origin_head->ds_dir, DD_USED_HEAD,
4175 		    dused, dcomp, duncomp, tx);
4176 		dsl_dir_diduse_space(clone->ds_dir, DD_USED_HEAD,
4177 		    -dused, -dcomp, -duncomp, tx);
4178 
4179 		/*
4180 		 * The difference in the space used by snapshots is the
4181 		 * difference in snapshot space due to the head's
4182 		 * deadlist (since that's the only thing that's
4183 		 * changing that affects the snapused).
4184 		 */
4185 		dsl_deadlist_space_range(&clone->ds_deadlist,
4186 		    origin_head->ds_dir->dd_origin_txg, UINT64_MAX,
4187 		    &cdl_used, &cdl_comp, &cdl_uncomp);
4188 		dsl_deadlist_space_range(&origin_head->ds_deadlist,
4189 		    origin_head->ds_dir->dd_origin_txg, UINT64_MAX,
4190 		    &odl_used, &odl_comp, &odl_uncomp);
4191 		dsl_dir_transfer_space(origin_head->ds_dir, cdl_used - odl_used,
4192 		    DD_USED_HEAD, DD_USED_SNAP, tx);
4193 	}
4194 
4195 	/* swap ds_*_bytes */
4196 	SWITCH64(dsl_dataset_phys(origin_head)->ds_referenced_bytes,
4197 	    dsl_dataset_phys(clone)->ds_referenced_bytes);
4198 	SWITCH64(dsl_dataset_phys(origin_head)->ds_compressed_bytes,
4199 	    dsl_dataset_phys(clone)->ds_compressed_bytes);
4200 	SWITCH64(dsl_dataset_phys(origin_head)->ds_uncompressed_bytes,
4201 	    dsl_dataset_phys(clone)->ds_uncompressed_bytes);
4202 	SWITCH64(dsl_dataset_phys(origin_head)->ds_unique_bytes,
4203 	    dsl_dataset_phys(clone)->ds_unique_bytes);
4204 
4205 	/* apply any parent delta for change in unconsumed refreservation */
4206 	dsl_dir_diduse_space(origin_head->ds_dir, DD_USED_REFRSRV,
4207 	    unused_refres_delta, 0, 0, tx);
4208 
4209 	/*
4210 	 * Swap deadlists.
4211 	 */
4212 	dsl_deadlist_close(&clone->ds_deadlist);
4213 	dsl_deadlist_close(&origin_head->ds_deadlist);
4214 	SWITCH64(dsl_dataset_phys(origin_head)->ds_deadlist_obj,
4215 	    dsl_dataset_phys(clone)->ds_deadlist_obj);
4216 	dsl_deadlist_open(&clone->ds_deadlist, dp->dp_meta_objset,
4217 	    dsl_dataset_phys(clone)->ds_deadlist_obj);
4218 	dsl_deadlist_open(&origin_head->ds_deadlist, dp->dp_meta_objset,
4219 	    dsl_dataset_phys(origin_head)->ds_deadlist_obj);
4220 	dsl_dataset_swap_remap_deadlists(clone, origin_head, tx);
4221 
4222 	/*
4223 	 * If there is a bookmark at the origin, its "next dataset" is
4224 	 * changing, so we need to reset its FBN.
4225 	 */
4226 	dsl_bookmark_next_changed(origin_head, origin_head->ds_prev, tx);
4227 
4228 	dsl_scan_ds_clone_swapped(origin_head, clone, tx);
4229 
4230 	/*
4231 	 * Destroy any livelists associated with the clone or the origin,
4232 	 * since after the swap the corresponding livelists are no longer
4233 	 * valid.
4234 	 */
4235 	dsl_dir_remove_livelist(clone->ds_dir, tx, B_TRUE);
4236 	dsl_dir_remove_livelist(origin_head->ds_dir, tx, B_TRUE);
4237 
4238 	spa_history_log_internal_ds(clone, "clone swap", tx,
4239 	    "parent=%s", origin_head->ds_dir->dd_myname);
4240 }
4241 
4242 /*
4243  * Given a pool name and a dataset object number in that pool,
4244  * return the name of that dataset.
4245  */
4246 int
4247 dsl_dsobj_to_dsname(char *pname, uint64_t obj, char *buf)
4248 {
4249 	dsl_pool_t *dp;
4250 	dsl_dataset_t *ds;
4251 	int error;
4252 
4253 	error = dsl_pool_hold(pname, FTAG, &dp);
4254 	if (error != 0)
4255 		return (error);
4256 
4257 	error = dsl_dataset_hold_obj(dp, obj, FTAG, &ds);
4258 	if (error == 0) {
4259 		dsl_dataset_name(ds, buf);
4260 		dsl_dataset_rele(ds, FTAG);
4261 	}
4262 	dsl_pool_rele(dp, FTAG);
4263 
4264 	return (error);
4265 }
4266 
4267 int
4268 dsl_dataset_check_quota(dsl_dataset_t *ds, boolean_t check_quota,
4269     uint64_t asize, uint64_t inflight, uint64_t *used, uint64_t *ref_rsrv)
4270 {
4271 	int error = 0;
4272 
4273 	ASSERT3S(asize, >, 0);
4274 
4275 	/*
4276 	 * *ref_rsrv is the portion of asize that will come from any
4277 	 * unconsumed refreservation space.
4278 	 */
4279 	*ref_rsrv = 0;
4280 
4281 	mutex_enter(&ds->ds_lock);
4282 	/*
4283 	 * Make a space adjustment for reserved bytes.
4284 	 */
4285 	if (ds->ds_reserved > dsl_dataset_phys(ds)->ds_unique_bytes) {
4286 		ASSERT3U(*used, >=,
4287 		    ds->ds_reserved - dsl_dataset_phys(ds)->ds_unique_bytes);
4288 		*used -=
4289 		    (ds->ds_reserved - dsl_dataset_phys(ds)->ds_unique_bytes);
4290 		*ref_rsrv =
4291 		    asize - MIN(asize, parent_delta(ds, asize + inflight));
4292 	}
4293 
4294 	if (!check_quota || ds->ds_quota == 0) {
4295 		mutex_exit(&ds->ds_lock);
4296 		return (0);
4297 	}
4298 	/*
4299 	 * If they are requesting more space, and our current estimate
4300 	 * is over quota, they get to try again unless the actual
4301 	 * on-disk is over quota and there are no pending changes (which
4302 	 * may free up space for us).
4303 	 */
4304 	if (dsl_dataset_phys(ds)->ds_referenced_bytes + inflight >=
4305 	    ds->ds_quota) {
4306 		if (inflight > 0 ||
4307 		    dsl_dataset_phys(ds)->ds_referenced_bytes < ds->ds_quota)
4308 			error = SET_ERROR(ERESTART);
4309 		else
4310 			error = SET_ERROR(EDQUOT);
4311 	}
4312 	mutex_exit(&ds->ds_lock);
4313 
4314 	return (error);
4315 }
4316 
4317 typedef struct dsl_dataset_set_qr_arg {
4318 	const char *ddsqra_name;
4319 	zprop_source_t ddsqra_source;
4320 	uint64_t ddsqra_value;
4321 } dsl_dataset_set_qr_arg_t;
4322 
4323 
4324 /* ARGSUSED */
4325 static int
4326 dsl_dataset_set_refquota_check(void *arg, dmu_tx_t *tx)
4327 {
4328 	dsl_dataset_set_qr_arg_t *ddsqra = arg;
4329 	dsl_pool_t *dp = dmu_tx_pool(tx);
4330 	dsl_dataset_t *ds;
4331 	int error;
4332 	uint64_t newval;
4333 
4334 	if (spa_version(dp->dp_spa) < SPA_VERSION_REFQUOTA)
4335 		return (SET_ERROR(ENOTSUP));
4336 
4337 	error = dsl_dataset_hold(dp, ddsqra->ddsqra_name, FTAG, &ds);
4338 	if (error != 0)
4339 		return (error);
4340 
4341 	if (ds->ds_is_snapshot) {
4342 		dsl_dataset_rele(ds, FTAG);
4343 		return (SET_ERROR(EINVAL));
4344 	}
4345 
4346 	error = dsl_prop_predict(ds->ds_dir,
4347 	    zfs_prop_to_name(ZFS_PROP_REFQUOTA),
4348 	    ddsqra->ddsqra_source, ddsqra->ddsqra_value, &newval);
4349 	if (error != 0) {
4350 		dsl_dataset_rele(ds, FTAG);
4351 		return (error);
4352 	}
4353 
4354 	if (newval == 0) {
4355 		dsl_dataset_rele(ds, FTAG);
4356 		return (0);
4357 	}
4358 
4359 	if (newval < dsl_dataset_phys(ds)->ds_referenced_bytes ||
4360 	    newval < ds->ds_reserved) {
4361 		dsl_dataset_rele(ds, FTAG);
4362 		return (SET_ERROR(ENOSPC));
4363 	}
4364 
4365 	dsl_dataset_rele(ds, FTAG);
4366 	return (0);
4367 }
4368 
4369 static void
4370 dsl_dataset_set_refquota_sync(void *arg, dmu_tx_t *tx)
4371 {
4372 	dsl_dataset_set_qr_arg_t *ddsqra = arg;
4373 	dsl_pool_t *dp = dmu_tx_pool(tx);
4374 	dsl_dataset_t *ds = NULL;
4375 	uint64_t newval;
4376 
4377 	VERIFY0(dsl_dataset_hold(dp, ddsqra->ddsqra_name, FTAG, &ds));
4378 
4379 	dsl_prop_set_sync_impl(ds,
4380 	    zfs_prop_to_name(ZFS_PROP_REFQUOTA),
4381 	    ddsqra->ddsqra_source, sizeof (ddsqra->ddsqra_value), 1,
4382 	    &ddsqra->ddsqra_value, tx);
4383 
4384 	VERIFY0(dsl_prop_get_int_ds(ds,
4385 	    zfs_prop_to_name(ZFS_PROP_REFQUOTA), &newval));
4386 
4387 	if (ds->ds_quota != newval) {
4388 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
4389 		ds->ds_quota = newval;
4390 	}
4391 	dsl_dataset_rele(ds, FTAG);
4392 }
4393 
4394 int
4395 dsl_dataset_set_refquota(const char *dsname, zprop_source_t source,
4396     uint64_t refquota)
4397 {
4398 	dsl_dataset_set_qr_arg_t ddsqra;
4399 
4400 	ddsqra.ddsqra_name = dsname;
4401 	ddsqra.ddsqra_source = source;
4402 	ddsqra.ddsqra_value = refquota;
4403 
4404 	return (dsl_sync_task(dsname, dsl_dataset_set_refquota_check,
4405 	    dsl_dataset_set_refquota_sync, &ddsqra, 0,
4406 	    ZFS_SPACE_CHECK_EXTRA_RESERVED));
4407 }
4408 
4409 static int
4410 dsl_dataset_set_refreservation_check(void *arg, dmu_tx_t *tx)
4411 {
4412 	dsl_dataset_set_qr_arg_t *ddsqra = arg;
4413 	dsl_pool_t *dp = dmu_tx_pool(tx);
4414 	dsl_dataset_t *ds;
4415 	int error;
4416 	uint64_t newval, unique;
4417 
4418 	if (spa_version(dp->dp_spa) < SPA_VERSION_REFRESERVATION)
4419 		return (SET_ERROR(ENOTSUP));
4420 
4421 	error = dsl_dataset_hold(dp, ddsqra->ddsqra_name, FTAG, &ds);
4422 	if (error != 0)
4423 		return (error);
4424 
4425 	if (ds->ds_is_snapshot) {
4426 		dsl_dataset_rele(ds, FTAG);
4427 		return (SET_ERROR(EINVAL));
4428 	}
4429 
4430 	error = dsl_prop_predict(ds->ds_dir,
4431 	    zfs_prop_to_name(ZFS_PROP_REFRESERVATION),
4432 	    ddsqra->ddsqra_source, ddsqra->ddsqra_value, &newval);
4433 	if (error != 0) {
4434 		dsl_dataset_rele(ds, FTAG);
4435 		return (error);
4436 	}
4437 
4438 	/*
4439 	 * If we are doing the preliminary check in open context, the
4440 	 * space estimates may be inaccurate.
4441 	 */
4442 	if (!dmu_tx_is_syncing(tx)) {
4443 		dsl_dataset_rele(ds, FTAG);
4444 		return (0);
4445 	}
4446 
4447 	mutex_enter(&ds->ds_lock);
4448 	if (!DS_UNIQUE_IS_ACCURATE(ds))
4449 		dsl_dataset_recalc_head_uniq(ds);
4450 	unique = dsl_dataset_phys(ds)->ds_unique_bytes;
4451 	mutex_exit(&ds->ds_lock);
4452 
4453 	if (MAX(unique, newval) > MAX(unique, ds->ds_reserved)) {
4454 		uint64_t delta = MAX(unique, newval) -
4455 		    MAX(unique, ds->ds_reserved);
4456 
4457 		if (delta >
4458 		    dsl_dir_space_available(ds->ds_dir, NULL, 0, B_TRUE) ||
4459 		    (ds->ds_quota > 0 && newval > ds->ds_quota)) {
4460 			dsl_dataset_rele(ds, FTAG);
4461 			return (SET_ERROR(ENOSPC));
4462 		}
4463 	}
4464 
4465 	dsl_dataset_rele(ds, FTAG);
4466 	return (0);
4467 }
4468 
4469 void
4470 dsl_dataset_set_refreservation_sync_impl(dsl_dataset_t *ds,
4471     zprop_source_t source, uint64_t value, dmu_tx_t *tx)
4472 {
4473 	uint64_t newval;
4474 	uint64_t unique;
4475 	int64_t delta;
4476 
4477 	dsl_prop_set_sync_impl(ds, zfs_prop_to_name(ZFS_PROP_REFRESERVATION),
4478 	    source, sizeof (value), 1, &value, tx);
4479 
4480 	VERIFY0(dsl_prop_get_int_ds(ds,
4481 	    zfs_prop_to_name(ZFS_PROP_REFRESERVATION), &newval));
4482 
4483 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
4484 	mutex_enter(&ds->ds_dir->dd_lock);
4485 	mutex_enter(&ds->ds_lock);
4486 	ASSERT(DS_UNIQUE_IS_ACCURATE(ds));
4487 	unique = dsl_dataset_phys(ds)->ds_unique_bytes;
4488 	delta = MAX(0, (int64_t)(newval - unique)) -
4489 	    MAX(0, (int64_t)(ds->ds_reserved - unique));
4490 	ds->ds_reserved = newval;
4491 	mutex_exit(&ds->ds_lock);
4492 
4493 	dsl_dir_diduse_space(ds->ds_dir, DD_USED_REFRSRV, delta, 0, 0, tx);
4494 	mutex_exit(&ds->ds_dir->dd_lock);
4495 }
4496 
4497 static void
4498 dsl_dataset_set_refreservation_sync(void *arg, dmu_tx_t *tx)
4499 {
4500 	dsl_dataset_set_qr_arg_t *ddsqra = arg;
4501 	dsl_pool_t *dp = dmu_tx_pool(tx);
4502 	dsl_dataset_t *ds = NULL;
4503 
4504 	VERIFY0(dsl_dataset_hold(dp, ddsqra->ddsqra_name, FTAG, &ds));
4505 	dsl_dataset_set_refreservation_sync_impl(ds,
4506 	    ddsqra->ddsqra_source, ddsqra->ddsqra_value, tx);
4507 	dsl_dataset_rele(ds, FTAG);
4508 }
4509 
4510 int
4511 dsl_dataset_set_refreservation(const char *dsname, zprop_source_t source,
4512     uint64_t refreservation)
4513 {
4514 	dsl_dataset_set_qr_arg_t ddsqra;
4515 
4516 	ddsqra.ddsqra_name = dsname;
4517 	ddsqra.ddsqra_source = source;
4518 	ddsqra.ddsqra_value = refreservation;
4519 
4520 	return (dsl_sync_task(dsname, dsl_dataset_set_refreservation_check,
4521 	    dsl_dataset_set_refreservation_sync, &ddsqra, 0,
4522 	    ZFS_SPACE_CHECK_EXTRA_RESERVED));
4523 }
4524 
4525 typedef struct dsl_dataset_set_compression_arg {
4526 	const char *ddsca_name;
4527 	zprop_source_t ddsca_source;
4528 	uint64_t ddsca_value;
4529 } dsl_dataset_set_compression_arg_t;
4530 
4531 /* ARGSUSED */
4532 static int
4533 dsl_dataset_set_compression_check(void *arg, dmu_tx_t *tx)
4534 {
4535 	dsl_dataset_set_compression_arg_t *ddsca = arg;
4536 	dsl_pool_t *dp = dmu_tx_pool(tx);
4537 
4538 	uint64_t compval = ZIO_COMPRESS_ALGO(ddsca->ddsca_value);
4539 	spa_feature_t f = zio_compress_to_feature(compval);
4540 
4541 	if (f == SPA_FEATURE_NONE)
4542 		return (SET_ERROR(EINVAL));
4543 
4544 	if (!spa_feature_is_enabled(dp->dp_spa, f))
4545 		return (SET_ERROR(ENOTSUP));
4546 
4547 	return (0);
4548 }
4549 
4550 static void
4551 dsl_dataset_set_compression_sync(void *arg, dmu_tx_t *tx)
4552 {
4553 	dsl_dataset_set_compression_arg_t *ddsca = arg;
4554 	dsl_pool_t *dp = dmu_tx_pool(tx);
4555 	dsl_dataset_t *ds = NULL;
4556 
4557 	uint64_t compval = ZIO_COMPRESS_ALGO(ddsca->ddsca_value);
4558 	spa_feature_t f = zio_compress_to_feature(compval);
4559 	ASSERT3S(spa_feature_table[f].fi_type, ==, ZFEATURE_TYPE_BOOLEAN);
4560 
4561 	VERIFY0(dsl_dataset_hold(dp, ddsca->ddsca_name, FTAG, &ds));
4562 	if (zfeature_active(f, ds->ds_feature[f]) != B_TRUE) {
4563 		ds->ds_feature_activation[f] = (void *)B_TRUE;
4564 		dsl_dataset_activate_feature(ds->ds_object, f,
4565 		    ds->ds_feature_activation[f], tx);
4566 		ds->ds_feature[f] = ds->ds_feature_activation[f];
4567 	}
4568 	dsl_dataset_rele(ds, FTAG);
4569 }
4570 
4571 int
4572 dsl_dataset_set_compression(const char *dsname, zprop_source_t source,
4573     uint64_t compression)
4574 {
4575 	dsl_dataset_set_compression_arg_t ddsca;
4576 
4577 	/*
4578 	 * The sync task is only required for zstd in order to activate
4579 	 * the feature flag when the property is first set.
4580 	 */
4581 	if (ZIO_COMPRESS_ALGO(compression) != ZIO_COMPRESS_ZSTD)
4582 		return (0);
4583 
4584 	ddsca.ddsca_name = dsname;
4585 	ddsca.ddsca_source = source;
4586 	ddsca.ddsca_value = compression;
4587 
4588 	return (dsl_sync_task(dsname, dsl_dataset_set_compression_check,
4589 	    dsl_dataset_set_compression_sync, &ddsca, 0,
4590 	    ZFS_SPACE_CHECK_EXTRA_RESERVED));
4591 }
4592 
4593 /*
4594  * Return (in *usedp) the amount of space referenced by "new" that was not
4595  * referenced at the time the bookmark corresponds to.  "New" may be a
4596  * snapshot or a head.  The bookmark must be before new, in
4597  * new's filesystem (or its origin) -- caller verifies this.
4598  *
4599  * The written space is calculated by considering two components:  First, we
4600  * ignore any freed space, and calculate the written as new's used space
4601  * minus old's used space.  Next, we add in the amount of space that was freed
4602  * between the two time points, thus reducing new's used space relative to
4603  * old's. Specifically, this is the space that was born before
4604  * zbm_creation_txg, and freed before new (ie. on new's deadlist or a
4605  * previous deadlist).
4606  *
4607  * space freed                         [---------------------]
4608  * snapshots                       ---O-------O--------O-------O------
4609  *                                         bookmark           new
4610  *
4611  * Note, the bookmark's zbm_*_bytes_refd must be valid, but if the HAS_FBN
4612  * flag is not set, we will calculate the freed_before_next based on the
4613  * next snapshot's deadlist, rather than using zbm_*_freed_before_next_snap.
4614  */
4615 static int
4616 dsl_dataset_space_written_impl(zfs_bookmark_phys_t *bmp,
4617     dsl_dataset_t *new, uint64_t *usedp, uint64_t *compp, uint64_t *uncompp)
4618 {
4619 	int err = 0;
4620 	dsl_pool_t *dp = new->ds_dir->dd_pool;
4621 
4622 	ASSERT(dsl_pool_config_held(dp));
4623 	if (dsl_dataset_is_snapshot(new)) {
4624 		ASSERT3U(bmp->zbm_creation_txg, <,
4625 		    dsl_dataset_phys(new)->ds_creation_txg);
4626 	}
4627 
4628 	*usedp = 0;
4629 	*usedp += dsl_dataset_phys(new)->ds_referenced_bytes;
4630 	*usedp -= bmp->zbm_referenced_bytes_refd;
4631 
4632 	*compp = 0;
4633 	*compp += dsl_dataset_phys(new)->ds_compressed_bytes;
4634 	*compp -= bmp->zbm_compressed_bytes_refd;
4635 
4636 	*uncompp = 0;
4637 	*uncompp += dsl_dataset_phys(new)->ds_uncompressed_bytes;
4638 	*uncompp -= bmp->zbm_uncompressed_bytes_refd;
4639 
4640 	dsl_dataset_t *snap = new;
4641 
4642 	while (dsl_dataset_phys(snap)->ds_prev_snap_txg >
4643 	    bmp->zbm_creation_txg) {
4644 		uint64_t used, comp, uncomp;
4645 
4646 		dsl_deadlist_space_range(&snap->ds_deadlist,
4647 		    0, bmp->zbm_creation_txg,
4648 		    &used, &comp, &uncomp);
4649 		*usedp += used;
4650 		*compp += comp;
4651 		*uncompp += uncomp;
4652 
4653 		uint64_t snapobj = dsl_dataset_phys(snap)->ds_prev_snap_obj;
4654 		if (snap != new)
4655 			dsl_dataset_rele(snap, FTAG);
4656 		err = dsl_dataset_hold_obj(dp, snapobj, FTAG, &snap);
4657 		if (err != 0)
4658 			break;
4659 	}
4660 
4661 	/*
4662 	 * We might not have the FBN if we are calculating written from
4663 	 * a snapshot (because we didn't know the correct "next" snapshot
4664 	 * until now).
4665 	 */
4666 	if (bmp->zbm_flags & ZBM_FLAG_HAS_FBN) {
4667 		*usedp += bmp->zbm_referenced_freed_before_next_snap;
4668 		*compp += bmp->zbm_compressed_freed_before_next_snap;
4669 		*uncompp += bmp->zbm_uncompressed_freed_before_next_snap;
4670 	} else {
4671 		ASSERT3U(dsl_dataset_phys(snap)->ds_prev_snap_txg, ==,
4672 		    bmp->zbm_creation_txg);
4673 		uint64_t used, comp, uncomp;
4674 		dsl_deadlist_space(&snap->ds_deadlist, &used, &comp, &uncomp);
4675 		*usedp += used;
4676 		*compp += comp;
4677 		*uncompp += uncomp;
4678 	}
4679 	if (snap != new)
4680 		dsl_dataset_rele(snap, FTAG);
4681 	return (err);
4682 }
4683 
4684 /*
4685  * Return (in *usedp) the amount of space written in new that was not
4686  * present at the time the bookmark corresponds to.  New may be a
4687  * snapshot or the head.  Old must be a bookmark before new, in
4688  * new's filesystem (or its origin) -- caller verifies this.
4689  */
4690 int
4691 dsl_dataset_space_written_bookmark(zfs_bookmark_phys_t *bmp,
4692     dsl_dataset_t *new, uint64_t *usedp, uint64_t *compp, uint64_t *uncompp)
4693 {
4694 	if (!(bmp->zbm_flags & ZBM_FLAG_HAS_FBN))
4695 		return (SET_ERROR(ENOTSUP));
4696 	return (dsl_dataset_space_written_impl(bmp, new,
4697 	    usedp, compp, uncompp));
4698 }
4699 
4700 /*
4701  * Return (in *usedp) the amount of space written in new that is not
4702  * present in oldsnap.  New may be a snapshot or the head.  Old must be
4703  * a snapshot before new, in new's filesystem (or its origin).  If not then
4704  * fail and return EINVAL.
4705  */
4706 int
4707 dsl_dataset_space_written(dsl_dataset_t *oldsnap, dsl_dataset_t *new,
4708     uint64_t *usedp, uint64_t *compp, uint64_t *uncompp)
4709 {
4710 	if (!dsl_dataset_is_before(new, oldsnap, 0))
4711 		return (SET_ERROR(EINVAL));
4712 
4713 	zfs_bookmark_phys_t zbm = { 0 };
4714 	dsl_dataset_phys_t *dsp = dsl_dataset_phys(oldsnap);
4715 	zbm.zbm_guid = dsp->ds_guid;
4716 	zbm.zbm_creation_txg = dsp->ds_creation_txg;
4717 	zbm.zbm_creation_time = dsp->ds_creation_time;
4718 	zbm.zbm_referenced_bytes_refd = dsp->ds_referenced_bytes;
4719 	zbm.zbm_compressed_bytes_refd = dsp->ds_compressed_bytes;
4720 	zbm.zbm_uncompressed_bytes_refd = dsp->ds_uncompressed_bytes;
4721 
4722 	/*
4723 	 * If oldsnap is the origin (or origin's origin, ...) of new,
4724 	 * we can't easily calculate the effective FBN.  Therefore,
4725 	 * we do not set ZBM_FLAG_HAS_FBN, so that the _impl will calculate
4726 	 * it relative to the correct "next": the next snapshot towards "new",
4727 	 * rather than the next snapshot in oldsnap's dsl_dir.
4728 	 */
4729 	return (dsl_dataset_space_written_impl(&zbm, new,
4730 	    usedp, compp, uncompp));
4731 }
4732 
4733 /*
4734  * Return (in *usedp) the amount of space that will be reclaimed if firstsnap,
4735  * lastsnap, and all snapshots in between are deleted.
4736  *
4737  * blocks that would be freed            [---------------------------]
4738  * snapshots                       ---O-------O--------O-------O--------O
4739  *                                        firstsnap        lastsnap
4740  *
4741  * This is the set of blocks that were born after the snap before firstsnap,
4742  * (birth > firstsnap->prev_snap_txg) and died before the snap after the
4743  * last snap (ie, is on lastsnap->ds_next->ds_deadlist or an earlier deadlist).
4744  * We calculate this by iterating over the relevant deadlists (from the snap
4745  * after lastsnap, backward to the snap after firstsnap), summing up the
4746  * space on the deadlist that was born after the snap before firstsnap.
4747  */
4748 int
4749 dsl_dataset_space_wouldfree(dsl_dataset_t *firstsnap,
4750     dsl_dataset_t *lastsnap,
4751     uint64_t *usedp, uint64_t *compp, uint64_t *uncompp)
4752 {
4753 	int err = 0;
4754 	uint64_t snapobj;
4755 	dsl_pool_t *dp = firstsnap->ds_dir->dd_pool;
4756 
4757 	ASSERT(firstsnap->ds_is_snapshot);
4758 	ASSERT(lastsnap->ds_is_snapshot);
4759 
4760 	/*
4761 	 * Check that the snapshots are in the same dsl_dir, and firstsnap
4762 	 * is before lastsnap.
4763 	 */
4764 	if (firstsnap->ds_dir != lastsnap->ds_dir ||
4765 	    dsl_dataset_phys(firstsnap)->ds_creation_txg >
4766 	    dsl_dataset_phys(lastsnap)->ds_creation_txg)
4767 		return (SET_ERROR(EINVAL));
4768 
4769 	*usedp = *compp = *uncompp = 0;
4770 
4771 	snapobj = dsl_dataset_phys(lastsnap)->ds_next_snap_obj;
4772 	while (snapobj != firstsnap->ds_object) {
4773 		dsl_dataset_t *ds;
4774 		uint64_t used, comp, uncomp;
4775 
4776 		err = dsl_dataset_hold_obj(dp, snapobj, FTAG, &ds);
4777 		if (err != 0)
4778 			break;
4779 
4780 		dsl_deadlist_space_range(&ds->ds_deadlist,
4781 		    dsl_dataset_phys(firstsnap)->ds_prev_snap_txg, UINT64_MAX,
4782 		    &used, &comp, &uncomp);
4783 		*usedp += used;
4784 		*compp += comp;
4785 		*uncompp += uncomp;
4786 
4787 		snapobj = dsl_dataset_phys(ds)->ds_prev_snap_obj;
4788 		ASSERT3U(snapobj, !=, 0);
4789 		dsl_dataset_rele(ds, FTAG);
4790 	}
4791 	return (err);
4792 }
4793 
4794 /*
4795  * Return TRUE if 'earlier' is an earlier snapshot in 'later's timeline.
4796  * For example, they could both be snapshots of the same filesystem, and
4797  * 'earlier' is before 'later'.  Or 'earlier' could be the origin of
4798  * 'later's filesystem.  Or 'earlier' could be an older snapshot in the origin's
4799  * filesystem.  Or 'earlier' could be the origin's origin.
4800  *
4801  * If non-zero, earlier_txg is used instead of earlier's ds_creation_txg.
4802  */
4803 boolean_t
4804 dsl_dataset_is_before(dsl_dataset_t *later, dsl_dataset_t *earlier,
4805     uint64_t earlier_txg)
4806 {
4807 	dsl_pool_t *dp = later->ds_dir->dd_pool;
4808 	int error;
4809 	boolean_t ret;
4810 
4811 	ASSERT(dsl_pool_config_held(dp));
4812 	ASSERT(earlier->ds_is_snapshot || earlier_txg != 0);
4813 
4814 	if (earlier_txg == 0)
4815 		earlier_txg = dsl_dataset_phys(earlier)->ds_creation_txg;
4816 
4817 	if (later->ds_is_snapshot &&
4818 	    earlier_txg >= dsl_dataset_phys(later)->ds_creation_txg)
4819 		return (B_FALSE);
4820 
4821 	if (later->ds_dir == earlier->ds_dir)
4822 		return (B_TRUE);
4823 
4824 	/*
4825 	 * We check dd_origin_obj explicitly here rather than using
4826 	 * dsl_dir_is_clone() so that we will return TRUE if "earlier"
4827 	 * is $ORIGIN@$ORIGIN.  dsl_dataset_space_written() depends on
4828 	 * this behavior.
4829 	 */
4830 	if (dsl_dir_phys(later->ds_dir)->dd_origin_obj == 0)
4831 		return (B_FALSE);
4832 
4833 	dsl_dataset_t *origin;
4834 	error = dsl_dataset_hold_obj(dp,
4835 	    dsl_dir_phys(later->ds_dir)->dd_origin_obj, FTAG, &origin);
4836 	if (error != 0)
4837 		return (B_FALSE);
4838 	if (dsl_dataset_phys(origin)->ds_creation_txg == earlier_txg &&
4839 	    origin->ds_dir == earlier->ds_dir) {
4840 		dsl_dataset_rele(origin, FTAG);
4841 		return (B_TRUE);
4842 	}
4843 	ret = dsl_dataset_is_before(origin, earlier, earlier_txg);
4844 	dsl_dataset_rele(origin, FTAG);
4845 	return (ret);
4846 }
4847 
4848 void
4849 dsl_dataset_zapify(dsl_dataset_t *ds, dmu_tx_t *tx)
4850 {
4851 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
4852 	dmu_object_zapify(mos, ds->ds_object, DMU_OT_DSL_DATASET, tx);
4853 }
4854 
4855 boolean_t
4856 dsl_dataset_is_zapified(dsl_dataset_t *ds)
4857 {
4858 	dmu_object_info_t doi;
4859 
4860 	dmu_object_info_from_db(ds->ds_dbuf, &doi);
4861 	return (doi.doi_type == DMU_OTN_ZAP_METADATA);
4862 }
4863 
4864 boolean_t
4865 dsl_dataset_has_resume_receive_state(dsl_dataset_t *ds)
4866 {
4867 	return (dsl_dataset_is_zapified(ds) &&
4868 	    zap_contains(ds->ds_dir->dd_pool->dp_meta_objset,
4869 	    ds->ds_object, DS_FIELD_RESUME_TOGUID) == 0);
4870 }
4871 
4872 uint64_t
4873 dsl_dataset_get_remap_deadlist_object(dsl_dataset_t *ds)
4874 {
4875 	uint64_t remap_deadlist_obj;
4876 	int err;
4877 
4878 	if (!dsl_dataset_is_zapified(ds))
4879 		return (0);
4880 
4881 	err = zap_lookup(ds->ds_dir->dd_pool->dp_meta_objset, ds->ds_object,
4882 	    DS_FIELD_REMAP_DEADLIST, sizeof (remap_deadlist_obj), 1,
4883 	    &remap_deadlist_obj);
4884 
4885 	if (err != 0) {
4886 		VERIFY3S(err, ==, ENOENT);
4887 		return (0);
4888 	}
4889 
4890 	ASSERT(remap_deadlist_obj != 0);
4891 	return (remap_deadlist_obj);
4892 }
4893 
4894 boolean_t
4895 dsl_dataset_remap_deadlist_exists(dsl_dataset_t *ds)
4896 {
4897 	EQUIV(dsl_deadlist_is_open(&ds->ds_remap_deadlist),
4898 	    dsl_dataset_get_remap_deadlist_object(ds) != 0);
4899 	return (dsl_deadlist_is_open(&ds->ds_remap_deadlist));
4900 }
4901 
4902 static void
4903 dsl_dataset_set_remap_deadlist_object(dsl_dataset_t *ds, uint64_t obj,
4904     dmu_tx_t *tx)
4905 {
4906 	ASSERT(obj != 0);
4907 	dsl_dataset_zapify(ds, tx);
4908 	VERIFY0(zap_add(ds->ds_dir->dd_pool->dp_meta_objset, ds->ds_object,
4909 	    DS_FIELD_REMAP_DEADLIST, sizeof (obj), 1, &obj, tx));
4910 }
4911 
4912 static void
4913 dsl_dataset_unset_remap_deadlist_object(dsl_dataset_t *ds, dmu_tx_t *tx)
4914 {
4915 	VERIFY0(zap_remove(ds->ds_dir->dd_pool->dp_meta_objset,
4916 	    ds->ds_object, DS_FIELD_REMAP_DEADLIST, tx));
4917 }
4918 
4919 void
4920 dsl_dataset_destroy_remap_deadlist(dsl_dataset_t *ds, dmu_tx_t *tx)
4921 {
4922 	uint64_t remap_deadlist_object;
4923 	spa_t *spa = ds->ds_dir->dd_pool->dp_spa;
4924 
4925 	ASSERT(dmu_tx_is_syncing(tx));
4926 	ASSERT(dsl_dataset_remap_deadlist_exists(ds));
4927 
4928 	remap_deadlist_object = ds->ds_remap_deadlist.dl_object;
4929 	dsl_deadlist_close(&ds->ds_remap_deadlist);
4930 	dsl_deadlist_free(spa_meta_objset(spa), remap_deadlist_object, tx);
4931 	dsl_dataset_unset_remap_deadlist_object(ds, tx);
4932 	spa_feature_decr(spa, SPA_FEATURE_OBSOLETE_COUNTS, tx);
4933 }
4934 
4935 void
4936 dsl_dataset_create_remap_deadlist(dsl_dataset_t *ds, dmu_tx_t *tx)
4937 {
4938 	uint64_t remap_deadlist_obj;
4939 	spa_t *spa = ds->ds_dir->dd_pool->dp_spa;
4940 
4941 	ASSERT(dmu_tx_is_syncing(tx));
4942 	ASSERT(MUTEX_HELD(&ds->ds_remap_deadlist_lock));
4943 	/*
4944 	 * Currently we only create remap deadlists when there are indirect
4945 	 * vdevs with referenced mappings.
4946 	 */
4947 	ASSERT(spa_feature_is_active(spa, SPA_FEATURE_DEVICE_REMOVAL));
4948 
4949 	remap_deadlist_obj = dsl_deadlist_clone(
4950 	    &ds->ds_deadlist, UINT64_MAX,
4951 	    dsl_dataset_phys(ds)->ds_prev_snap_obj, tx);
4952 	dsl_dataset_set_remap_deadlist_object(ds,
4953 	    remap_deadlist_obj, tx);
4954 	dsl_deadlist_open(&ds->ds_remap_deadlist, spa_meta_objset(spa),
4955 	    remap_deadlist_obj);
4956 	spa_feature_incr(spa, SPA_FEATURE_OBSOLETE_COUNTS, tx);
4957 }
4958 
4959 void
4960 dsl_dataset_activate_redaction(dsl_dataset_t *ds, uint64_t *redact_snaps,
4961     uint64_t num_redact_snaps, dmu_tx_t *tx)
4962 {
4963 	uint64_t dsobj = ds->ds_object;
4964 	struct feature_type_uint64_array_arg *ftuaa =
4965 	    kmem_zalloc(sizeof (*ftuaa), KM_SLEEP);
4966 	ftuaa->length = (int64_t)num_redact_snaps;
4967 	if (num_redact_snaps > 0) {
4968 		ftuaa->array = kmem_alloc(num_redact_snaps * sizeof (uint64_t),
4969 		    KM_SLEEP);
4970 		bcopy(redact_snaps, ftuaa->array, num_redact_snaps *
4971 		    sizeof (uint64_t));
4972 	}
4973 	dsl_dataset_activate_feature(dsobj, SPA_FEATURE_REDACTED_DATASETS,
4974 	    ftuaa, tx);
4975 	ds->ds_feature[SPA_FEATURE_REDACTED_DATASETS] = ftuaa;
4976 }
4977 
4978 /* BEGIN CSTYLED */
4979 #if defined(_LP64)
4980 #define	RECORDSIZE_PERM ZMOD_RW
4981 #else
4982 /* Limited to 1M on 32-bit platforms due to lack of virtual address space */
4983 #define	RECORDSIZE_PERM ZMOD_RD
4984 #endif
4985 ZFS_MODULE_PARAM(zfs, zfs_, max_recordsize, INT, RECORDSIZE_PERM,
4986 	"Max allowed record size");
4987 
4988 ZFS_MODULE_PARAM(zfs, zfs_, allow_redacted_dataset_mount, INT, ZMOD_RW,
4989 	"Allow mounting of redacted datasets");
4990 /* END CSTYLED */
4991 
4992 EXPORT_SYMBOL(dsl_dataset_hold);
4993 EXPORT_SYMBOL(dsl_dataset_hold_flags);
4994 EXPORT_SYMBOL(dsl_dataset_hold_obj);
4995 EXPORT_SYMBOL(dsl_dataset_hold_obj_flags);
4996 EXPORT_SYMBOL(dsl_dataset_own);
4997 EXPORT_SYMBOL(dsl_dataset_own_obj);
4998 EXPORT_SYMBOL(dsl_dataset_name);
4999 EXPORT_SYMBOL(dsl_dataset_rele);
5000 EXPORT_SYMBOL(dsl_dataset_rele_flags);
5001 EXPORT_SYMBOL(dsl_dataset_disown);
5002 EXPORT_SYMBOL(dsl_dataset_tryown);
5003 EXPORT_SYMBOL(dsl_dataset_create_sync);
5004 EXPORT_SYMBOL(dsl_dataset_create_sync_dd);
5005 EXPORT_SYMBOL(dsl_dataset_snapshot_check);
5006 EXPORT_SYMBOL(dsl_dataset_snapshot_sync);
5007 EXPORT_SYMBOL(dsl_dataset_promote);
5008 EXPORT_SYMBOL(dsl_dataset_user_hold);
5009 EXPORT_SYMBOL(dsl_dataset_user_release);
5010 EXPORT_SYMBOL(dsl_dataset_get_holds);
5011 EXPORT_SYMBOL(dsl_dataset_get_blkptr);
5012 EXPORT_SYMBOL(dsl_dataset_get_spa);
5013 EXPORT_SYMBOL(dsl_dataset_modified_since_snap);
5014 EXPORT_SYMBOL(dsl_dataset_space_written);
5015 EXPORT_SYMBOL(dsl_dataset_space_wouldfree);
5016 EXPORT_SYMBOL(dsl_dataset_sync);
5017 EXPORT_SYMBOL(dsl_dataset_block_born);
5018 EXPORT_SYMBOL(dsl_dataset_block_kill);
5019 EXPORT_SYMBOL(dsl_dataset_dirty);
5020 EXPORT_SYMBOL(dsl_dataset_stats);
5021 EXPORT_SYMBOL(dsl_dataset_fast_stat);
5022 EXPORT_SYMBOL(dsl_dataset_space);
5023 EXPORT_SYMBOL(dsl_dataset_fsid_guid);
5024 EXPORT_SYMBOL(dsl_dsobj_to_dsname);
5025 EXPORT_SYMBOL(dsl_dataset_check_quota);
5026 EXPORT_SYMBOL(dsl_dataset_clone_swap_check_impl);
5027 EXPORT_SYMBOL(dsl_dataset_clone_swap_sync_impl);
5028