xref: /illumos-gate/usr/src/uts/common/fs/zfs/dsl_dataset.c (revision 77b171372ed21642e04c873ef1e87fe2365520df)
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, 2016 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) 2014 Integros [integros.com]
29  * Copyright 2016, OmniTI Computer Consulting, Inc. All rights reserved.
30  * Copyright 2017 Nexenta Systems, Inc.
31  */
32 
33 #include <sys/dmu_objset.h>
34 #include <sys/dsl_dataset.h>
35 #include <sys/dsl_dir.h>
36 #include <sys/dsl_prop.h>
37 #include <sys/dsl_synctask.h>
38 #include <sys/dmu_traverse.h>
39 #include <sys/dmu_impl.h>
40 #include <sys/dmu_tx.h>
41 #include <sys/arc.h>
42 #include <sys/zio.h>
43 #include <sys/zap.h>
44 #include <sys/zfeature.h>
45 #include <sys/unique.h>
46 #include <sys/zfs_context.h>
47 #include <sys/zfs_ioctl.h>
48 #include <sys/spa.h>
49 #include <sys/zfs_znode.h>
50 #include <sys/zfs_onexit.h>
51 #include <sys/zvol.h>
52 #include <sys/dsl_scan.h>
53 #include <sys/dsl_deadlist.h>
54 #include <sys/dsl_destroy.h>
55 #include <sys/dsl_userhold.h>
56 #include <sys/dsl_bookmark.h>
57 #include <sys/dmu_send.h>
58 #include <sys/zio_checksum.h>
59 #include <sys/zio_compress.h>
60 #include <zfs_fletcher.h>
61 
62 /*
63  * The SPA supports block sizes up to 16MB.  However, very large blocks
64  * can have an impact on i/o latency (e.g. tying up a spinning disk for
65  * ~300ms), and also potentially on the memory allocator.  Therefore,
66  * we do not allow the recordsize to be set larger than zfs_max_recordsize
67  * (default 1MB).  Larger blocks can be created by changing this tunable,
68  * and pools with larger blocks can always be imported and used, regardless
69  * of this setting.
70  */
71 int zfs_max_recordsize = 1 * 1024 * 1024;
72 
73 #define	SWITCH64(x, y) \
74 	{ \
75 		uint64_t __tmp = (x); \
76 		(x) = (y); \
77 		(y) = __tmp; \
78 	}
79 
80 #define	DS_REF_MAX	(1ULL << 62)
81 
82 extern inline dsl_dataset_phys_t *dsl_dataset_phys(dsl_dataset_t *ds);
83 
84 extern int spa_asize_inflation;
85 
86 static zil_header_t zero_zil;
87 
88 /*
89  * Figure out how much of this delta should be propogated to the dsl_dir
90  * layer.  If there's a refreservation, that space has already been
91  * partially accounted for in our ancestors.
92  */
93 static int64_t
94 parent_delta(dsl_dataset_t *ds, int64_t delta)
95 {
96 	dsl_dataset_phys_t *ds_phys;
97 	uint64_t old_bytes, new_bytes;
98 
99 	if (ds->ds_reserved == 0)
100 		return (delta);
101 
102 	ds_phys = dsl_dataset_phys(ds);
103 	old_bytes = MAX(ds_phys->ds_unique_bytes, ds->ds_reserved);
104 	new_bytes = MAX(ds_phys->ds_unique_bytes + delta, ds->ds_reserved);
105 
106 	ASSERT3U(ABS((int64_t)(new_bytes - old_bytes)), <=, ABS(delta));
107 	return (new_bytes - old_bytes);
108 }
109 
110 void
111 dsl_dataset_block_born(dsl_dataset_t *ds, const blkptr_t *bp, dmu_tx_t *tx)
112 {
113 	int used = bp_get_dsize_sync(tx->tx_pool->dp_spa, bp);
114 	int compressed = BP_GET_PSIZE(bp);
115 	int uncompressed = BP_GET_UCSIZE(bp);
116 	int64_t delta;
117 
118 	dprintf_bp(bp, "ds=%p", ds);
119 
120 	ASSERT(dmu_tx_is_syncing(tx));
121 	/* It could have been compressed away to nothing */
122 	if (BP_IS_HOLE(bp))
123 		return;
124 	ASSERT(BP_GET_TYPE(bp) != DMU_OT_NONE);
125 	ASSERT(DMU_OT_IS_VALID(BP_GET_TYPE(bp)));
126 	if (ds == NULL) {
127 		dsl_pool_mos_diduse_space(tx->tx_pool,
128 		    used, compressed, uncompressed);
129 		return;
130 	}
131 
132 	ASSERT3U(bp->blk_birth, >, dsl_dataset_phys(ds)->ds_prev_snap_txg);
133 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
134 	mutex_enter(&ds->ds_lock);
135 	delta = parent_delta(ds, used);
136 	dsl_dataset_phys(ds)->ds_referenced_bytes += used;
137 	dsl_dataset_phys(ds)->ds_compressed_bytes += compressed;
138 	dsl_dataset_phys(ds)->ds_uncompressed_bytes += uncompressed;
139 	dsl_dataset_phys(ds)->ds_unique_bytes += used;
140 
141 	if (BP_GET_LSIZE(bp) > SPA_OLD_MAXBLOCKSIZE) {
142 		ds->ds_feature_activation_needed[SPA_FEATURE_LARGE_BLOCKS] =
143 		    B_TRUE;
144 	}
145 
146 	spa_feature_t f = zio_checksum_to_feature(BP_GET_CHECKSUM(bp));
147 	if (f != SPA_FEATURE_NONE)
148 		ds->ds_feature_activation_needed[f] = B_TRUE;
149 
150 	mutex_exit(&ds->ds_lock);
151 	dsl_dir_diduse_space(ds->ds_dir, DD_USED_HEAD, delta,
152 	    compressed, uncompressed, tx);
153 	dsl_dir_transfer_space(ds->ds_dir, used - delta,
154 	    DD_USED_REFRSRV, DD_USED_HEAD, tx);
155 }
156 
157 int
158 dsl_dataset_block_kill(dsl_dataset_t *ds, const blkptr_t *bp, dmu_tx_t *tx,
159     boolean_t async)
160 {
161 	int used = bp_get_dsize_sync(tx->tx_pool->dp_spa, bp);
162 	int compressed = BP_GET_PSIZE(bp);
163 	int uncompressed = BP_GET_UCSIZE(bp);
164 
165 	if (BP_IS_HOLE(bp))
166 		return (0);
167 
168 	ASSERT(dmu_tx_is_syncing(tx));
169 	ASSERT(bp->blk_birth <= tx->tx_txg);
170 
171 	if (ds == NULL) {
172 		dsl_free(tx->tx_pool, tx->tx_txg, bp);
173 		dsl_pool_mos_diduse_space(tx->tx_pool,
174 		    -used, -compressed, -uncompressed);
175 		return (used);
176 	}
177 	ASSERT3P(tx->tx_pool, ==, ds->ds_dir->dd_pool);
178 
179 	ASSERT(!ds->ds_is_snapshot);
180 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
181 
182 	if (bp->blk_birth > dsl_dataset_phys(ds)->ds_prev_snap_txg) {
183 		int64_t delta;
184 
185 		dprintf_bp(bp, "freeing ds=%llu", ds->ds_object);
186 		dsl_free(tx->tx_pool, tx->tx_txg, bp);
187 
188 		mutex_enter(&ds->ds_lock);
189 		ASSERT(dsl_dataset_phys(ds)->ds_unique_bytes >= used ||
190 		    !DS_UNIQUE_IS_ACCURATE(ds));
191 		delta = parent_delta(ds, -used);
192 		dsl_dataset_phys(ds)->ds_unique_bytes -= used;
193 		mutex_exit(&ds->ds_lock);
194 		dsl_dir_diduse_space(ds->ds_dir, DD_USED_HEAD,
195 		    delta, -compressed, -uncompressed, tx);
196 		dsl_dir_transfer_space(ds->ds_dir, -used - delta,
197 		    DD_USED_REFRSRV, DD_USED_HEAD, tx);
198 	} else {
199 		dprintf_bp(bp, "putting on dead list: %s", "");
200 		if (async) {
201 			/*
202 			 * We are here as part of zio's write done callback,
203 			 * which means we're a zio interrupt thread.  We can't
204 			 * call dsl_deadlist_insert() now because it may block
205 			 * waiting for I/O.  Instead, put bp on the deferred
206 			 * queue and let dsl_pool_sync() finish the job.
207 			 */
208 			bplist_append(&ds->ds_pending_deadlist, bp);
209 		} else {
210 			dsl_deadlist_insert(&ds->ds_deadlist, bp, tx);
211 		}
212 		ASSERT3U(ds->ds_prev->ds_object, ==,
213 		    dsl_dataset_phys(ds)->ds_prev_snap_obj);
214 		ASSERT(dsl_dataset_phys(ds->ds_prev)->ds_num_children > 0);
215 		/* if (bp->blk_birth > prev prev snap txg) prev unique += bs */
216 		if (dsl_dataset_phys(ds->ds_prev)->ds_next_snap_obj ==
217 		    ds->ds_object && bp->blk_birth >
218 		    dsl_dataset_phys(ds->ds_prev)->ds_prev_snap_txg) {
219 			dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
220 			mutex_enter(&ds->ds_prev->ds_lock);
221 			dsl_dataset_phys(ds->ds_prev)->ds_unique_bytes += used;
222 			mutex_exit(&ds->ds_prev->ds_lock);
223 		}
224 		if (bp->blk_birth > ds->ds_dir->dd_origin_txg) {
225 			dsl_dir_transfer_space(ds->ds_dir, used,
226 			    DD_USED_HEAD, DD_USED_SNAP, tx);
227 		}
228 	}
229 	mutex_enter(&ds->ds_lock);
230 	ASSERT3U(dsl_dataset_phys(ds)->ds_referenced_bytes, >=, used);
231 	dsl_dataset_phys(ds)->ds_referenced_bytes -= used;
232 	ASSERT3U(dsl_dataset_phys(ds)->ds_compressed_bytes, >=, compressed);
233 	dsl_dataset_phys(ds)->ds_compressed_bytes -= compressed;
234 	ASSERT3U(dsl_dataset_phys(ds)->ds_uncompressed_bytes, >=, uncompressed);
235 	dsl_dataset_phys(ds)->ds_uncompressed_bytes -= uncompressed;
236 	mutex_exit(&ds->ds_lock);
237 
238 	return (used);
239 }
240 
241 /*
242  * We have to release the fsid syncronously or we risk that a subsequent
243  * mount of the same dataset will fail to unique_insert the fsid.  This
244  * failure would manifest itself as the fsid of this dataset changing
245  * between mounts which makes NFS clients quite unhappy.
246  */
247 static void
248 dsl_dataset_evict_sync(void *dbu)
249 {
250 	dsl_dataset_t *ds = dbu;
251 
252 	ASSERT(ds->ds_owner == NULL);
253 
254 	unique_remove(ds->ds_fsid_guid);
255 }
256 
257 static void
258 dsl_dataset_evict_async(void *dbu)
259 {
260 	dsl_dataset_t *ds = dbu;
261 
262 	ASSERT(ds->ds_owner == NULL);
263 
264 	ds->ds_dbuf = NULL;
265 
266 	if (ds->ds_objset != NULL)
267 		dmu_objset_evict(ds->ds_objset);
268 
269 	if (ds->ds_prev) {
270 		dsl_dataset_rele(ds->ds_prev, ds);
271 		ds->ds_prev = NULL;
272 	}
273 
274 	bplist_destroy(&ds->ds_pending_deadlist);
275 	if (ds->ds_deadlist.dl_os != NULL)
276 		dsl_deadlist_close(&ds->ds_deadlist);
277 	if (ds->ds_dir)
278 		dsl_dir_async_rele(ds->ds_dir, ds);
279 
280 	ASSERT(!list_link_active(&ds->ds_synced_link));
281 
282 	list_destroy(&ds->ds_prop_cbs);
283 	mutex_destroy(&ds->ds_lock);
284 	mutex_destroy(&ds->ds_opening_lock);
285 	mutex_destroy(&ds->ds_sendstream_lock);
286 	refcount_destroy(&ds->ds_longholds);
287 	rrw_destroy(&ds->ds_bp_rwlock);
288 
289 	kmem_free(ds, sizeof (dsl_dataset_t));
290 }
291 
292 int
293 dsl_dataset_get_snapname(dsl_dataset_t *ds)
294 {
295 	dsl_dataset_phys_t *headphys;
296 	int err;
297 	dmu_buf_t *headdbuf;
298 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
299 	objset_t *mos = dp->dp_meta_objset;
300 
301 	if (ds->ds_snapname[0])
302 		return (0);
303 	if (dsl_dataset_phys(ds)->ds_next_snap_obj == 0)
304 		return (0);
305 
306 	err = dmu_bonus_hold(mos, dsl_dir_phys(ds->ds_dir)->dd_head_dataset_obj,
307 	    FTAG, &headdbuf);
308 	if (err != 0)
309 		return (err);
310 	headphys = headdbuf->db_data;
311 	err = zap_value_search(dp->dp_meta_objset,
312 	    headphys->ds_snapnames_zapobj, ds->ds_object, 0, ds->ds_snapname);
313 	dmu_buf_rele(headdbuf, FTAG);
314 	return (err);
315 }
316 
317 int
318 dsl_dataset_snap_lookup(dsl_dataset_t *ds, const char *name, uint64_t *value)
319 {
320 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
321 	uint64_t snapobj = dsl_dataset_phys(ds)->ds_snapnames_zapobj;
322 	matchtype_t mt = 0;
323 	int err;
324 
325 	if (dsl_dataset_phys(ds)->ds_flags & DS_FLAG_CI_DATASET)
326 		mt = MT_NORMALIZE;
327 
328 	err = zap_lookup_norm(mos, snapobj, name, 8, 1,
329 	    value, mt, NULL, 0, NULL);
330 	if (err == ENOTSUP && (mt & MT_NORMALIZE))
331 		err = zap_lookup(mos, snapobj, name, 8, 1, value);
332 	return (err);
333 }
334 
335 int
336 dsl_dataset_snap_remove(dsl_dataset_t *ds, const char *name, dmu_tx_t *tx,
337     boolean_t adj_cnt)
338 {
339 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
340 	uint64_t snapobj = dsl_dataset_phys(ds)->ds_snapnames_zapobj;
341 	matchtype_t mt = 0;
342 	int err;
343 
344 	dsl_dir_snap_cmtime_update(ds->ds_dir);
345 
346 	if (dsl_dataset_phys(ds)->ds_flags & DS_FLAG_CI_DATASET)
347 		mt = MT_NORMALIZE;
348 
349 	err = zap_remove_norm(mos, snapobj, name, mt, tx);
350 	if (err == ENOTSUP && (mt & MT_NORMALIZE))
351 		err = zap_remove(mos, snapobj, name, tx);
352 
353 	if (err == 0 && adj_cnt)
354 		dsl_fs_ss_count_adjust(ds->ds_dir, -1,
355 		    DD_FIELD_SNAPSHOT_COUNT, tx);
356 
357 	return (err);
358 }
359 
360 boolean_t
361 dsl_dataset_try_add_ref(dsl_pool_t *dp, dsl_dataset_t *ds, void *tag)
362 {
363 	dmu_buf_t *dbuf = ds->ds_dbuf;
364 	boolean_t result = B_FALSE;
365 
366 	if (dbuf != NULL && dmu_buf_try_add_ref(dbuf, dp->dp_meta_objset,
367 	    ds->ds_object, DMU_BONUS_BLKID, tag)) {
368 
369 		if (ds == dmu_buf_get_user(dbuf))
370 			result = B_TRUE;
371 		else
372 			dmu_buf_rele(dbuf, tag);
373 	}
374 
375 	return (result);
376 }
377 
378 int
379 dsl_dataset_hold_obj(dsl_pool_t *dp, uint64_t dsobj, void *tag,
380     dsl_dataset_t **dsp)
381 {
382 	objset_t *mos = dp->dp_meta_objset;
383 	dmu_buf_t *dbuf;
384 	dsl_dataset_t *ds;
385 	int err;
386 	dmu_object_info_t doi;
387 
388 	ASSERT(dsl_pool_config_held(dp));
389 
390 	err = dmu_bonus_hold(mos, dsobj, tag, &dbuf);
391 	if (err != 0)
392 		return (err);
393 
394 	/* Make sure dsobj has the correct object type. */
395 	dmu_object_info_from_db(dbuf, &doi);
396 	if (doi.doi_bonus_type != DMU_OT_DSL_DATASET) {
397 		dmu_buf_rele(dbuf, tag);
398 		return (SET_ERROR(EINVAL));
399 	}
400 
401 	ds = dmu_buf_get_user(dbuf);
402 	if (ds == NULL) {
403 		dsl_dataset_t *winner = NULL;
404 
405 		ds = kmem_zalloc(sizeof (dsl_dataset_t), KM_SLEEP);
406 		ds->ds_dbuf = dbuf;
407 		ds->ds_object = dsobj;
408 		ds->ds_is_snapshot = dsl_dataset_phys(ds)->ds_num_children != 0;
409 
410 		mutex_init(&ds->ds_lock, NULL, MUTEX_DEFAULT, NULL);
411 		mutex_init(&ds->ds_opening_lock, NULL, MUTEX_DEFAULT, NULL);
412 		mutex_init(&ds->ds_sendstream_lock, NULL, MUTEX_DEFAULT, NULL);
413 		rrw_init(&ds->ds_bp_rwlock, B_FALSE);
414 		refcount_create(&ds->ds_longholds);
415 
416 		bplist_create(&ds->ds_pending_deadlist);
417 		dsl_deadlist_open(&ds->ds_deadlist,
418 		    mos, dsl_dataset_phys(ds)->ds_deadlist_obj);
419 
420 		list_create(&ds->ds_sendstreams, sizeof (dmu_sendarg_t),
421 		    offsetof(dmu_sendarg_t, dsa_link));
422 
423 		list_create(&ds->ds_prop_cbs, sizeof (dsl_prop_cb_record_t),
424 		    offsetof(dsl_prop_cb_record_t, cbr_ds_node));
425 
426 		if (doi.doi_type == DMU_OTN_ZAP_METADATA) {
427 			for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
428 				if (!(spa_feature_table[f].fi_flags &
429 				    ZFEATURE_FLAG_PER_DATASET))
430 					continue;
431 				err = zap_contains(mos, dsobj,
432 				    spa_feature_table[f].fi_guid);
433 				if (err == 0) {
434 					ds->ds_feature_inuse[f] = B_TRUE;
435 				} else {
436 					ASSERT3U(err, ==, ENOENT);
437 					err = 0;
438 				}
439 			}
440 		}
441 
442 		err = dsl_dir_hold_obj(dp,
443 		    dsl_dataset_phys(ds)->ds_dir_obj, NULL, ds, &ds->ds_dir);
444 		if (err != 0) {
445 			mutex_destroy(&ds->ds_lock);
446 			mutex_destroy(&ds->ds_opening_lock);
447 			mutex_destroy(&ds->ds_sendstream_lock);
448 			refcount_destroy(&ds->ds_longholds);
449 			bplist_destroy(&ds->ds_pending_deadlist);
450 			dsl_deadlist_close(&ds->ds_deadlist);
451 			kmem_free(ds, sizeof (dsl_dataset_t));
452 			dmu_buf_rele(dbuf, tag);
453 			return (err);
454 		}
455 
456 		if (!ds->ds_is_snapshot) {
457 			ds->ds_snapname[0] = '\0';
458 			if (dsl_dataset_phys(ds)->ds_prev_snap_obj != 0) {
459 				err = dsl_dataset_hold_obj(dp,
460 				    dsl_dataset_phys(ds)->ds_prev_snap_obj,
461 				    ds, &ds->ds_prev);
462 			}
463 			if (doi.doi_type == DMU_OTN_ZAP_METADATA) {
464 				int zaperr = zap_lookup(mos, ds->ds_object,
465 				    DS_FIELD_BOOKMARK_NAMES,
466 				    sizeof (ds->ds_bookmarks), 1,
467 				    &ds->ds_bookmarks);
468 				if (zaperr != ENOENT)
469 					VERIFY0(zaperr);
470 			}
471 		} else {
472 			if (zfs_flags & ZFS_DEBUG_SNAPNAMES)
473 				err = dsl_dataset_get_snapname(ds);
474 			if (err == 0 &&
475 			    dsl_dataset_phys(ds)->ds_userrefs_obj != 0) {
476 				err = zap_count(
477 				    ds->ds_dir->dd_pool->dp_meta_objset,
478 				    dsl_dataset_phys(ds)->ds_userrefs_obj,
479 				    &ds->ds_userrefs);
480 			}
481 		}
482 
483 		if (err == 0 && !ds->ds_is_snapshot) {
484 			err = dsl_prop_get_int_ds(ds,
485 			    zfs_prop_to_name(ZFS_PROP_REFRESERVATION),
486 			    &ds->ds_reserved);
487 			if (err == 0) {
488 				err = dsl_prop_get_int_ds(ds,
489 				    zfs_prop_to_name(ZFS_PROP_REFQUOTA),
490 				    &ds->ds_quota);
491 			}
492 		} else {
493 			ds->ds_reserved = ds->ds_quota = 0;
494 		}
495 
496 		dmu_buf_init_user(&ds->ds_dbu, dsl_dataset_evict_sync,
497 		    dsl_dataset_evict_async, &ds->ds_dbuf);
498 		if (err == 0)
499 			winner = dmu_buf_set_user_ie(dbuf, &ds->ds_dbu);
500 
501 		if (err != 0 || winner != NULL) {
502 			bplist_destroy(&ds->ds_pending_deadlist);
503 			dsl_deadlist_close(&ds->ds_deadlist);
504 			if (ds->ds_prev)
505 				dsl_dataset_rele(ds->ds_prev, ds);
506 			dsl_dir_rele(ds->ds_dir, ds);
507 			mutex_destroy(&ds->ds_lock);
508 			mutex_destroy(&ds->ds_opening_lock);
509 			mutex_destroy(&ds->ds_sendstream_lock);
510 			refcount_destroy(&ds->ds_longholds);
511 			kmem_free(ds, sizeof (dsl_dataset_t));
512 			if (err != 0) {
513 				dmu_buf_rele(dbuf, tag);
514 				return (err);
515 			}
516 			ds = winner;
517 		} else {
518 			ds->ds_fsid_guid =
519 			    unique_insert(dsl_dataset_phys(ds)->ds_fsid_guid);
520 			if (ds->ds_fsid_guid !=
521 			    dsl_dataset_phys(ds)->ds_fsid_guid) {
522 				zfs_dbgmsg("ds_fsid_guid changed from "
523 				    "%llx to %llx for pool %s dataset id %llu",
524 				    (long long)
525 				    dsl_dataset_phys(ds)->ds_fsid_guid,
526 				    (long long)ds->ds_fsid_guid,
527 				    spa_name(dp->dp_spa),
528 				    dsobj);
529 			}
530 		}
531 	}
532 	ASSERT3P(ds->ds_dbuf, ==, dbuf);
533 	ASSERT3P(dsl_dataset_phys(ds), ==, dbuf->db_data);
534 	ASSERT(dsl_dataset_phys(ds)->ds_prev_snap_obj != 0 ||
535 	    spa_version(dp->dp_spa) < SPA_VERSION_ORIGIN ||
536 	    dp->dp_origin_snap == NULL || ds == dp->dp_origin_snap);
537 	*dsp = ds;
538 	return (0);
539 }
540 
541 int
542 dsl_dataset_hold(dsl_pool_t *dp, const char *name,
543     void *tag, dsl_dataset_t **dsp)
544 {
545 	dsl_dir_t *dd;
546 	const char *snapname;
547 	uint64_t obj;
548 	int err = 0;
549 	dsl_dataset_t *ds;
550 
551 	err = dsl_dir_hold(dp, name, FTAG, &dd, &snapname);
552 	if (err != 0)
553 		return (err);
554 
555 	ASSERT(dsl_pool_config_held(dp));
556 	obj = dsl_dir_phys(dd)->dd_head_dataset_obj;
557 	if (obj != 0)
558 		err = dsl_dataset_hold_obj(dp, obj, tag, &ds);
559 	else
560 		err = SET_ERROR(ENOENT);
561 
562 	/* we may be looking for a snapshot */
563 	if (err == 0 && snapname != NULL) {
564 		dsl_dataset_t *snap_ds;
565 
566 		if (*snapname++ != '@') {
567 			dsl_dataset_rele(ds, tag);
568 			dsl_dir_rele(dd, FTAG);
569 			return (SET_ERROR(ENOENT));
570 		}
571 
572 		dprintf("looking for snapshot '%s'\n", snapname);
573 		err = dsl_dataset_snap_lookup(ds, snapname, &obj);
574 		if (err == 0)
575 			err = dsl_dataset_hold_obj(dp, obj, tag, &snap_ds);
576 		dsl_dataset_rele(ds, tag);
577 
578 		if (err == 0) {
579 			mutex_enter(&snap_ds->ds_lock);
580 			if (snap_ds->ds_snapname[0] == 0)
581 				(void) strlcpy(snap_ds->ds_snapname, snapname,
582 				    sizeof (snap_ds->ds_snapname));
583 			mutex_exit(&snap_ds->ds_lock);
584 			ds = snap_ds;
585 		}
586 	}
587 	if (err == 0)
588 		*dsp = ds;
589 	dsl_dir_rele(dd, FTAG);
590 	return (err);
591 }
592 
593 int
594 dsl_dataset_own_obj(dsl_pool_t *dp, uint64_t dsobj,
595     void *tag, dsl_dataset_t **dsp)
596 {
597 	int err = dsl_dataset_hold_obj(dp, dsobj, tag, dsp);
598 	if (err != 0)
599 		return (err);
600 	if (!dsl_dataset_tryown(*dsp, tag)) {
601 		dsl_dataset_rele(*dsp, tag);
602 		*dsp = NULL;
603 		return (SET_ERROR(EBUSY));
604 	}
605 	return (0);
606 }
607 
608 int
609 dsl_dataset_own(dsl_pool_t *dp, const char *name,
610     void *tag, dsl_dataset_t **dsp)
611 {
612 	int err = dsl_dataset_hold(dp, name, tag, dsp);
613 	if (err != 0)
614 		return (err);
615 	if (!dsl_dataset_tryown(*dsp, tag)) {
616 		dsl_dataset_rele(*dsp, tag);
617 		return (SET_ERROR(EBUSY));
618 	}
619 	return (0);
620 }
621 
622 /*
623  * See the comment above dsl_pool_hold() for details.  In summary, a long
624  * hold is used to prevent destruction of a dataset while the pool hold
625  * is dropped, allowing other concurrent operations (e.g. spa_sync()).
626  *
627  * The dataset and pool must be held when this function is called.  After it
628  * is called, the pool hold may be released while the dataset is still held
629  * and accessed.
630  */
631 void
632 dsl_dataset_long_hold(dsl_dataset_t *ds, void *tag)
633 {
634 	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
635 	(void) refcount_add(&ds->ds_longholds, tag);
636 }
637 
638 void
639 dsl_dataset_long_rele(dsl_dataset_t *ds, void *tag)
640 {
641 	(void) refcount_remove(&ds->ds_longholds, tag);
642 }
643 
644 /* Return B_TRUE if there are any long holds on this dataset. */
645 boolean_t
646 dsl_dataset_long_held(dsl_dataset_t *ds)
647 {
648 	return (!refcount_is_zero(&ds->ds_longholds));
649 }
650 
651 void
652 dsl_dataset_name(dsl_dataset_t *ds, char *name)
653 {
654 	if (ds == NULL) {
655 		(void) strcpy(name, "mos");
656 	} else {
657 		dsl_dir_name(ds->ds_dir, name);
658 		VERIFY0(dsl_dataset_get_snapname(ds));
659 		if (ds->ds_snapname[0]) {
660 			VERIFY3U(strlcat(name, "@", ZFS_MAX_DATASET_NAME_LEN),
661 			    <, ZFS_MAX_DATASET_NAME_LEN);
662 			/*
663 			 * We use a "recursive" mutex so that we
664 			 * can call dprintf_ds() with ds_lock held.
665 			 */
666 			if (!MUTEX_HELD(&ds->ds_lock)) {
667 				mutex_enter(&ds->ds_lock);
668 				VERIFY3U(strlcat(name, ds->ds_snapname,
669 				    ZFS_MAX_DATASET_NAME_LEN), <,
670 				    ZFS_MAX_DATASET_NAME_LEN);
671 				mutex_exit(&ds->ds_lock);
672 			} else {
673 				VERIFY3U(strlcat(name, ds->ds_snapname,
674 				    ZFS_MAX_DATASET_NAME_LEN), <,
675 				    ZFS_MAX_DATASET_NAME_LEN);
676 			}
677 		}
678 	}
679 }
680 
681 int
682 dsl_dataset_namelen(dsl_dataset_t *ds)
683 {
684 	VERIFY0(dsl_dataset_get_snapname(ds));
685 	mutex_enter(&ds->ds_lock);
686 	int len = dsl_dir_namelen(ds->ds_dir) + 1 + strlen(ds->ds_snapname);
687 	mutex_exit(&ds->ds_lock);
688 	return (len);
689 }
690 
691 void
692 dsl_dataset_rele(dsl_dataset_t *ds, void *tag)
693 {
694 	dmu_buf_rele(ds->ds_dbuf, tag);
695 }
696 
697 void
698 dsl_dataset_disown(dsl_dataset_t *ds, void *tag)
699 {
700 	ASSERT3P(ds->ds_owner, ==, tag);
701 	ASSERT(ds->ds_dbuf != NULL);
702 
703 	mutex_enter(&ds->ds_lock);
704 	ds->ds_owner = NULL;
705 	mutex_exit(&ds->ds_lock);
706 	dsl_dataset_long_rele(ds, tag);
707 	dsl_dataset_rele(ds, tag);
708 }
709 
710 boolean_t
711 dsl_dataset_tryown(dsl_dataset_t *ds, void *tag)
712 {
713 	boolean_t gotit = FALSE;
714 
715 	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
716 	mutex_enter(&ds->ds_lock);
717 	if (ds->ds_owner == NULL && !DS_IS_INCONSISTENT(ds)) {
718 		ds->ds_owner = tag;
719 		dsl_dataset_long_hold(ds, tag);
720 		gotit = TRUE;
721 	}
722 	mutex_exit(&ds->ds_lock);
723 	return (gotit);
724 }
725 
726 boolean_t
727 dsl_dataset_has_owner(dsl_dataset_t *ds)
728 {
729 	boolean_t rv;
730 	mutex_enter(&ds->ds_lock);
731 	rv = (ds->ds_owner != NULL);
732 	mutex_exit(&ds->ds_lock);
733 	return (rv);
734 }
735 
736 static void
737 dsl_dataset_activate_feature(uint64_t dsobj, spa_feature_t f, dmu_tx_t *tx)
738 {
739 	spa_t *spa = dmu_tx_pool(tx)->dp_spa;
740 	objset_t *mos = dmu_tx_pool(tx)->dp_meta_objset;
741 	uint64_t zero = 0;
742 
743 	VERIFY(spa_feature_table[f].fi_flags & ZFEATURE_FLAG_PER_DATASET);
744 
745 	spa_feature_incr(spa, f, tx);
746 	dmu_object_zapify(mos, dsobj, DMU_OT_DSL_DATASET, tx);
747 
748 	VERIFY0(zap_add(mos, dsobj, spa_feature_table[f].fi_guid,
749 	    sizeof (zero), 1, &zero, tx));
750 }
751 
752 void
753 dsl_dataset_deactivate_feature(uint64_t dsobj, spa_feature_t f, dmu_tx_t *tx)
754 {
755 	spa_t *spa = dmu_tx_pool(tx)->dp_spa;
756 	objset_t *mos = dmu_tx_pool(tx)->dp_meta_objset;
757 
758 	VERIFY(spa_feature_table[f].fi_flags & ZFEATURE_FLAG_PER_DATASET);
759 
760 	VERIFY0(zap_remove(mos, dsobj, spa_feature_table[f].fi_guid, tx));
761 	spa_feature_decr(spa, f, tx);
762 }
763 
764 uint64_t
765 dsl_dataset_create_sync_dd(dsl_dir_t *dd, dsl_dataset_t *origin,
766     uint64_t flags, dmu_tx_t *tx)
767 {
768 	dsl_pool_t *dp = dd->dd_pool;
769 	dmu_buf_t *dbuf;
770 	dsl_dataset_phys_t *dsphys;
771 	uint64_t dsobj;
772 	objset_t *mos = dp->dp_meta_objset;
773 
774 	if (origin == NULL)
775 		origin = dp->dp_origin_snap;
776 
777 	ASSERT(origin == NULL || origin->ds_dir->dd_pool == dp);
778 	ASSERT(origin == NULL || dsl_dataset_phys(origin)->ds_num_children > 0);
779 	ASSERT(dmu_tx_is_syncing(tx));
780 	ASSERT(dsl_dir_phys(dd)->dd_head_dataset_obj == 0);
781 
782 	dsobj = dmu_object_alloc(mos, DMU_OT_DSL_DATASET, 0,
783 	    DMU_OT_DSL_DATASET, sizeof (dsl_dataset_phys_t), tx);
784 	VERIFY0(dmu_bonus_hold(mos, dsobj, FTAG, &dbuf));
785 	dmu_buf_will_dirty(dbuf, tx);
786 	dsphys = dbuf->db_data;
787 	bzero(dsphys, sizeof (dsl_dataset_phys_t));
788 	dsphys->ds_dir_obj = dd->dd_object;
789 	dsphys->ds_flags = flags;
790 	dsphys->ds_fsid_guid = unique_create();
791 	(void) random_get_pseudo_bytes((void*)&dsphys->ds_guid,
792 	    sizeof (dsphys->ds_guid));
793 	dsphys->ds_snapnames_zapobj =
794 	    zap_create_norm(mos, U8_TEXTPREP_TOUPPER, DMU_OT_DSL_DS_SNAP_MAP,
795 	    DMU_OT_NONE, 0, tx);
796 	dsphys->ds_creation_time = gethrestime_sec();
797 	dsphys->ds_creation_txg = tx->tx_txg == TXG_INITIAL ? 1 : tx->tx_txg;
798 
799 	if (origin == NULL) {
800 		dsphys->ds_deadlist_obj = dsl_deadlist_alloc(mos, tx);
801 	} else {
802 		dsl_dataset_t *ohds; /* head of the origin snapshot */
803 
804 		dsphys->ds_prev_snap_obj = origin->ds_object;
805 		dsphys->ds_prev_snap_txg =
806 		    dsl_dataset_phys(origin)->ds_creation_txg;
807 		dsphys->ds_referenced_bytes =
808 		    dsl_dataset_phys(origin)->ds_referenced_bytes;
809 		dsphys->ds_compressed_bytes =
810 		    dsl_dataset_phys(origin)->ds_compressed_bytes;
811 		dsphys->ds_uncompressed_bytes =
812 		    dsl_dataset_phys(origin)->ds_uncompressed_bytes;
813 		rrw_enter(&origin->ds_bp_rwlock, RW_READER, FTAG);
814 		dsphys->ds_bp = dsl_dataset_phys(origin)->ds_bp;
815 		rrw_exit(&origin->ds_bp_rwlock, FTAG);
816 
817 		/*
818 		 * Inherit flags that describe the dataset's contents
819 		 * (INCONSISTENT) or properties (Case Insensitive).
820 		 */
821 		dsphys->ds_flags |= dsl_dataset_phys(origin)->ds_flags &
822 		    (DS_FLAG_INCONSISTENT | DS_FLAG_CI_DATASET);
823 
824 		for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
825 			if (origin->ds_feature_inuse[f])
826 				dsl_dataset_activate_feature(dsobj, f, tx);
827 		}
828 
829 		dmu_buf_will_dirty(origin->ds_dbuf, tx);
830 		dsl_dataset_phys(origin)->ds_num_children++;
831 
832 		VERIFY0(dsl_dataset_hold_obj(dp,
833 		    dsl_dir_phys(origin->ds_dir)->dd_head_dataset_obj,
834 		    FTAG, &ohds));
835 		dsphys->ds_deadlist_obj = dsl_deadlist_clone(&ohds->ds_deadlist,
836 		    dsphys->ds_prev_snap_txg, dsphys->ds_prev_snap_obj, tx);
837 		dsl_dataset_rele(ohds, FTAG);
838 
839 		if (spa_version(dp->dp_spa) >= SPA_VERSION_NEXT_CLONES) {
840 			if (dsl_dataset_phys(origin)->ds_next_clones_obj == 0) {
841 				dsl_dataset_phys(origin)->ds_next_clones_obj =
842 				    zap_create(mos,
843 				    DMU_OT_NEXT_CLONES, DMU_OT_NONE, 0, tx);
844 			}
845 			VERIFY0(zap_add_int(mos,
846 			    dsl_dataset_phys(origin)->ds_next_clones_obj,
847 			    dsobj, tx));
848 		}
849 
850 		dmu_buf_will_dirty(dd->dd_dbuf, tx);
851 		dsl_dir_phys(dd)->dd_origin_obj = origin->ds_object;
852 		if (spa_version(dp->dp_spa) >= SPA_VERSION_DIR_CLONES) {
853 			if (dsl_dir_phys(origin->ds_dir)->dd_clones == 0) {
854 				dmu_buf_will_dirty(origin->ds_dir->dd_dbuf, tx);
855 				dsl_dir_phys(origin->ds_dir)->dd_clones =
856 				    zap_create(mos,
857 				    DMU_OT_DSL_CLONES, DMU_OT_NONE, 0, tx);
858 			}
859 			VERIFY0(zap_add_int(mos,
860 			    dsl_dir_phys(origin->ds_dir)->dd_clones,
861 			    dsobj, tx));
862 		}
863 	}
864 
865 	if (spa_version(dp->dp_spa) >= SPA_VERSION_UNIQUE_ACCURATE)
866 		dsphys->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
867 
868 	dmu_buf_rele(dbuf, FTAG);
869 
870 	dmu_buf_will_dirty(dd->dd_dbuf, tx);
871 	dsl_dir_phys(dd)->dd_head_dataset_obj = dsobj;
872 
873 	return (dsobj);
874 }
875 
876 static void
877 dsl_dataset_zero_zil(dsl_dataset_t *ds, dmu_tx_t *tx)
878 {
879 	objset_t *os;
880 
881 	VERIFY0(dmu_objset_from_ds(ds, &os));
882 	if (bcmp(&os->os_zil_header, &zero_zil, sizeof (zero_zil)) != 0) {
883 		dsl_pool_t *dp = ds->ds_dir->dd_pool;
884 		zio_t *zio;
885 
886 		bzero(&os->os_zil_header, sizeof (os->os_zil_header));
887 
888 		zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
889 		dsl_dataset_sync(ds, zio, tx);
890 		VERIFY0(zio_wait(zio));
891 
892 		/* dsl_dataset_sync_done will drop this reference. */
893 		dmu_buf_add_ref(ds->ds_dbuf, ds);
894 		dsl_dataset_sync_done(ds, tx);
895 	}
896 }
897 
898 uint64_t
899 dsl_dataset_create_sync(dsl_dir_t *pdd, const char *lastname,
900     dsl_dataset_t *origin, uint64_t flags, cred_t *cr, dmu_tx_t *tx)
901 {
902 	dsl_pool_t *dp = pdd->dd_pool;
903 	uint64_t dsobj, ddobj;
904 	dsl_dir_t *dd;
905 
906 	ASSERT(dmu_tx_is_syncing(tx));
907 	ASSERT(lastname[0] != '@');
908 
909 	ddobj = dsl_dir_create_sync(dp, pdd, lastname, tx);
910 	VERIFY0(dsl_dir_hold_obj(dp, ddobj, lastname, FTAG, &dd));
911 
912 	dsobj = dsl_dataset_create_sync_dd(dd, origin,
913 	    flags & ~DS_CREATE_FLAG_NODIRTY, tx);
914 
915 	dsl_deleg_set_create_perms(dd, tx, cr);
916 
917 	/*
918 	 * Since we're creating a new node we know it's a leaf, so we can
919 	 * initialize the counts if the limit feature is active.
920 	 */
921 	if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_FS_SS_LIMIT)) {
922 		uint64_t cnt = 0;
923 		objset_t *os = dd->dd_pool->dp_meta_objset;
924 
925 		dsl_dir_zapify(dd, tx);
926 		VERIFY0(zap_add(os, dd->dd_object, DD_FIELD_FILESYSTEM_COUNT,
927 		    sizeof (cnt), 1, &cnt, tx));
928 		VERIFY0(zap_add(os, dd->dd_object, DD_FIELD_SNAPSHOT_COUNT,
929 		    sizeof (cnt), 1, &cnt, tx));
930 	}
931 
932 	dsl_dir_rele(dd, FTAG);
933 
934 	/*
935 	 * If we are creating a clone, make sure we zero out any stale
936 	 * data from the origin snapshots zil header.
937 	 */
938 	if (origin != NULL && !(flags & DS_CREATE_FLAG_NODIRTY)) {
939 		dsl_dataset_t *ds;
940 
941 		VERIFY0(dsl_dataset_hold_obj(dp, dsobj, FTAG, &ds));
942 		dsl_dataset_zero_zil(ds, tx);
943 		dsl_dataset_rele(ds, FTAG);
944 	}
945 
946 	return (dsobj);
947 }
948 
949 /*
950  * The unique space in the head dataset can be calculated by subtracting
951  * the space used in the most recent snapshot, that is still being used
952  * in this file system, from the space currently in use.  To figure out
953  * the space in the most recent snapshot still in use, we need to take
954  * the total space used in the snapshot and subtract out the space that
955  * has been freed up since the snapshot was taken.
956  */
957 void
958 dsl_dataset_recalc_head_uniq(dsl_dataset_t *ds)
959 {
960 	uint64_t mrs_used;
961 	uint64_t dlused, dlcomp, dluncomp;
962 
963 	ASSERT(!ds->ds_is_snapshot);
964 
965 	if (dsl_dataset_phys(ds)->ds_prev_snap_obj != 0)
966 		mrs_used = dsl_dataset_phys(ds->ds_prev)->ds_referenced_bytes;
967 	else
968 		mrs_used = 0;
969 
970 	dsl_deadlist_space(&ds->ds_deadlist, &dlused, &dlcomp, &dluncomp);
971 
972 	ASSERT3U(dlused, <=, mrs_used);
973 	dsl_dataset_phys(ds)->ds_unique_bytes =
974 	    dsl_dataset_phys(ds)->ds_referenced_bytes - (mrs_used - dlused);
975 
976 	if (spa_version(ds->ds_dir->dd_pool->dp_spa) >=
977 	    SPA_VERSION_UNIQUE_ACCURATE)
978 		dsl_dataset_phys(ds)->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
979 }
980 
981 void
982 dsl_dataset_remove_from_next_clones(dsl_dataset_t *ds, uint64_t obj,
983     dmu_tx_t *tx)
984 {
985 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
986 	uint64_t count;
987 	int err;
988 
989 	ASSERT(dsl_dataset_phys(ds)->ds_num_children >= 2);
990 	err = zap_remove_int(mos, dsl_dataset_phys(ds)->ds_next_clones_obj,
991 	    obj, tx);
992 	/*
993 	 * The err should not be ENOENT, but a bug in a previous version
994 	 * of the code could cause upgrade_clones_cb() to not set
995 	 * ds_next_snap_obj when it should, leading to a missing entry.
996 	 * If we knew that the pool was created after
997 	 * SPA_VERSION_NEXT_CLONES, we could assert that it isn't
998 	 * ENOENT.  However, at least we can check that we don't have
999 	 * too many entries in the next_clones_obj even after failing to
1000 	 * remove this one.
1001 	 */
1002 	if (err != ENOENT)
1003 		VERIFY0(err);
1004 	ASSERT0(zap_count(mos, dsl_dataset_phys(ds)->ds_next_clones_obj,
1005 	    &count));
1006 	ASSERT3U(count, <=, dsl_dataset_phys(ds)->ds_num_children - 2);
1007 }
1008 
1009 
1010 blkptr_t *
1011 dsl_dataset_get_blkptr(dsl_dataset_t *ds)
1012 {
1013 	return (&dsl_dataset_phys(ds)->ds_bp);
1014 }
1015 
1016 spa_t *
1017 dsl_dataset_get_spa(dsl_dataset_t *ds)
1018 {
1019 	return (ds->ds_dir->dd_pool->dp_spa);
1020 }
1021 
1022 void
1023 dsl_dataset_dirty(dsl_dataset_t *ds, dmu_tx_t *tx)
1024 {
1025 	dsl_pool_t *dp;
1026 
1027 	if (ds == NULL) /* this is the meta-objset */
1028 		return;
1029 
1030 	ASSERT(ds->ds_objset != NULL);
1031 
1032 	if (dsl_dataset_phys(ds)->ds_next_snap_obj != 0)
1033 		panic("dirtying snapshot!");
1034 
1035 	/* Must not dirty a dataset in the same txg where it got snapshotted. */
1036 	ASSERT3U(tx->tx_txg, >, dsl_dataset_phys(ds)->ds_prev_snap_txg);
1037 
1038 	dp = ds->ds_dir->dd_pool;
1039 	if (txg_list_add(&dp->dp_dirty_datasets, ds, tx->tx_txg)) {
1040 		/* up the hold count until we can be written out */
1041 		dmu_buf_add_ref(ds->ds_dbuf, ds);
1042 	}
1043 }
1044 
1045 boolean_t
1046 dsl_dataset_is_dirty(dsl_dataset_t *ds)
1047 {
1048 	for (int t = 0; t < TXG_SIZE; t++) {
1049 		if (txg_list_member(&ds->ds_dir->dd_pool->dp_dirty_datasets,
1050 		    ds, t))
1051 			return (B_TRUE);
1052 	}
1053 	return (B_FALSE);
1054 }
1055 
1056 static int
1057 dsl_dataset_snapshot_reserve_space(dsl_dataset_t *ds, dmu_tx_t *tx)
1058 {
1059 	uint64_t asize;
1060 
1061 	if (!dmu_tx_is_syncing(tx))
1062 		return (0);
1063 
1064 	/*
1065 	 * If there's an fs-only reservation, any blocks that might become
1066 	 * owned by the snapshot dataset must be accommodated by space
1067 	 * outside of the reservation.
1068 	 */
1069 	ASSERT(ds->ds_reserved == 0 || DS_UNIQUE_IS_ACCURATE(ds));
1070 	asize = MIN(dsl_dataset_phys(ds)->ds_unique_bytes, ds->ds_reserved);
1071 	if (asize > dsl_dir_space_available(ds->ds_dir, NULL, 0, TRUE))
1072 		return (SET_ERROR(ENOSPC));
1073 
1074 	/*
1075 	 * Propagate any reserved space for this snapshot to other
1076 	 * snapshot checks in this sync group.
1077 	 */
1078 	if (asize > 0)
1079 		dsl_dir_willuse_space(ds->ds_dir, asize, tx);
1080 
1081 	return (0);
1082 }
1083 
1084 typedef struct dsl_dataset_snapshot_arg {
1085 	nvlist_t *ddsa_snaps;
1086 	nvlist_t *ddsa_props;
1087 	nvlist_t *ddsa_errors;
1088 	cred_t *ddsa_cr;
1089 } dsl_dataset_snapshot_arg_t;
1090 
1091 int
1092 dsl_dataset_snapshot_check_impl(dsl_dataset_t *ds, const char *snapname,
1093     dmu_tx_t *tx, boolean_t recv, uint64_t cnt, cred_t *cr)
1094 {
1095 	int error;
1096 	uint64_t value;
1097 
1098 	ds->ds_trysnap_txg = tx->tx_txg;
1099 
1100 	if (!dmu_tx_is_syncing(tx))
1101 		return (0);
1102 
1103 	/*
1104 	 * We don't allow multiple snapshots of the same txg.  If there
1105 	 * is already one, try again.
1106 	 */
1107 	if (dsl_dataset_phys(ds)->ds_prev_snap_txg >= tx->tx_txg)
1108 		return (SET_ERROR(EAGAIN));
1109 
1110 	/*
1111 	 * Check for conflicting snapshot name.
1112 	 */
1113 	error = dsl_dataset_snap_lookup(ds, snapname, &value);
1114 	if (error == 0)
1115 		return (SET_ERROR(EEXIST));
1116 	if (error != ENOENT)
1117 		return (error);
1118 
1119 	/*
1120 	 * We don't allow taking snapshots of inconsistent datasets, such as
1121 	 * those into which we are currently receiving.  However, if we are
1122 	 * creating this snapshot as part of a receive, this check will be
1123 	 * executed atomically with respect to the completion of the receive
1124 	 * itself but prior to the clearing of DS_FLAG_INCONSISTENT; in this
1125 	 * case we ignore this, knowing it will be fixed up for us shortly in
1126 	 * dmu_recv_end_sync().
1127 	 */
1128 	if (!recv && DS_IS_INCONSISTENT(ds))
1129 		return (SET_ERROR(EBUSY));
1130 
1131 	/*
1132 	 * Skip the check for temporary snapshots or if we have already checked
1133 	 * the counts in dsl_dataset_snapshot_check. This means we really only
1134 	 * check the count here when we're receiving a stream.
1135 	 */
1136 	if (cnt != 0 && cr != NULL) {
1137 		error = dsl_fs_ss_limit_check(ds->ds_dir, cnt,
1138 		    ZFS_PROP_SNAPSHOT_LIMIT, NULL, cr);
1139 		if (error != 0)
1140 			return (error);
1141 	}
1142 
1143 	error = dsl_dataset_snapshot_reserve_space(ds, tx);
1144 	if (error != 0)
1145 		return (error);
1146 
1147 	return (0);
1148 }
1149 
1150 static int
1151 dsl_dataset_snapshot_check(void *arg, dmu_tx_t *tx)
1152 {
1153 	dsl_dataset_snapshot_arg_t *ddsa = arg;
1154 	dsl_pool_t *dp = dmu_tx_pool(tx);
1155 	nvpair_t *pair;
1156 	int rv = 0;
1157 
1158 	/*
1159 	 * Pre-compute how many total new snapshots will be created for each
1160 	 * level in the tree and below. This is needed for validating the
1161 	 * snapshot limit when either taking a recursive snapshot or when
1162 	 * taking multiple snapshots.
1163 	 *
1164 	 * The problem is that the counts are not actually adjusted when
1165 	 * we are checking, only when we finally sync. For a single snapshot,
1166 	 * this is easy, the count will increase by 1 at each node up the tree,
1167 	 * but its more complicated for the recursive/multiple snapshot case.
1168 	 *
1169 	 * The dsl_fs_ss_limit_check function does recursively check the count
1170 	 * at each level up the tree but since it is validating each snapshot
1171 	 * independently we need to be sure that we are validating the complete
1172 	 * count for the entire set of snapshots. We do this by rolling up the
1173 	 * counts for each component of the name into an nvlist and then
1174 	 * checking each of those cases with the aggregated count.
1175 	 *
1176 	 * This approach properly handles not only the recursive snapshot
1177 	 * case (where we get all of those on the ddsa_snaps list) but also
1178 	 * the sibling case (e.g. snapshot a/b and a/c so that we will also
1179 	 * validate the limit on 'a' using a count of 2).
1180 	 *
1181 	 * We validate the snapshot names in the third loop and only report
1182 	 * name errors once.
1183 	 */
1184 	if (dmu_tx_is_syncing(tx)) {
1185 		nvlist_t *cnt_track = NULL;
1186 		cnt_track = fnvlist_alloc();
1187 
1188 		/* Rollup aggregated counts into the cnt_track list */
1189 		for (pair = nvlist_next_nvpair(ddsa->ddsa_snaps, NULL);
1190 		    pair != NULL;
1191 		    pair = nvlist_next_nvpair(ddsa->ddsa_snaps, pair)) {
1192 			char *pdelim;
1193 			uint64_t val;
1194 			char nm[MAXPATHLEN];
1195 
1196 			(void) strlcpy(nm, nvpair_name(pair), sizeof (nm));
1197 			pdelim = strchr(nm, '@');
1198 			if (pdelim == NULL)
1199 				continue;
1200 			*pdelim = '\0';
1201 
1202 			do {
1203 				if (nvlist_lookup_uint64(cnt_track, nm,
1204 				    &val) == 0) {
1205 					/* update existing entry */
1206 					fnvlist_add_uint64(cnt_track, nm,
1207 					    val + 1);
1208 				} else {
1209 					/* add to list */
1210 					fnvlist_add_uint64(cnt_track, nm, 1);
1211 				}
1212 
1213 				pdelim = strrchr(nm, '/');
1214 				if (pdelim != NULL)
1215 					*pdelim = '\0';
1216 			} while (pdelim != NULL);
1217 		}
1218 
1219 		/* Check aggregated counts at each level */
1220 		for (pair = nvlist_next_nvpair(cnt_track, NULL);
1221 		    pair != NULL; pair = nvlist_next_nvpair(cnt_track, pair)) {
1222 			int error = 0;
1223 			char *name;
1224 			uint64_t cnt = 0;
1225 			dsl_dataset_t *ds;
1226 
1227 			name = nvpair_name(pair);
1228 			cnt = fnvpair_value_uint64(pair);
1229 			ASSERT(cnt > 0);
1230 
1231 			error = dsl_dataset_hold(dp, name, FTAG, &ds);
1232 			if (error == 0) {
1233 				error = dsl_fs_ss_limit_check(ds->ds_dir, cnt,
1234 				    ZFS_PROP_SNAPSHOT_LIMIT, NULL,
1235 				    ddsa->ddsa_cr);
1236 				dsl_dataset_rele(ds, FTAG);
1237 			}
1238 
1239 			if (error != 0) {
1240 				if (ddsa->ddsa_errors != NULL)
1241 					fnvlist_add_int32(ddsa->ddsa_errors,
1242 					    name, error);
1243 				rv = error;
1244 				/* only report one error for this check */
1245 				break;
1246 			}
1247 		}
1248 		nvlist_free(cnt_track);
1249 	}
1250 
1251 	for (pair = nvlist_next_nvpair(ddsa->ddsa_snaps, NULL);
1252 	    pair != NULL; pair = nvlist_next_nvpair(ddsa->ddsa_snaps, pair)) {
1253 		int error = 0;
1254 		dsl_dataset_t *ds;
1255 		char *name, *atp;
1256 		char dsname[ZFS_MAX_DATASET_NAME_LEN];
1257 
1258 		name = nvpair_name(pair);
1259 		if (strlen(name) >= ZFS_MAX_DATASET_NAME_LEN)
1260 			error = SET_ERROR(ENAMETOOLONG);
1261 		if (error == 0) {
1262 			atp = strchr(name, '@');
1263 			if (atp == NULL)
1264 				error = SET_ERROR(EINVAL);
1265 			if (error == 0)
1266 				(void) strlcpy(dsname, name, atp - name + 1);
1267 		}
1268 		if (error == 0)
1269 			error = dsl_dataset_hold(dp, dsname, FTAG, &ds);
1270 		if (error == 0) {
1271 			/* passing 0/NULL skips dsl_fs_ss_limit_check */
1272 			error = dsl_dataset_snapshot_check_impl(ds,
1273 			    atp + 1, tx, B_FALSE, 0, NULL);
1274 			dsl_dataset_rele(ds, FTAG);
1275 		}
1276 
1277 		if (error != 0) {
1278 			if (ddsa->ddsa_errors != NULL) {
1279 				fnvlist_add_int32(ddsa->ddsa_errors,
1280 				    name, error);
1281 			}
1282 			rv = error;
1283 		}
1284 	}
1285 
1286 	return (rv);
1287 }
1288 
1289 void
1290 dsl_dataset_snapshot_sync_impl(dsl_dataset_t *ds, const char *snapname,
1291     dmu_tx_t *tx)
1292 {
1293 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1294 	dmu_buf_t *dbuf;
1295 	dsl_dataset_phys_t *dsphys;
1296 	uint64_t dsobj, crtxg;
1297 	objset_t *mos = dp->dp_meta_objset;
1298 	objset_t *os;
1299 
1300 	ASSERT(RRW_WRITE_HELD(&dp->dp_config_rwlock));
1301 
1302 	/*
1303 	 * If we are on an old pool, the zil must not be active, in which
1304 	 * case it will be zeroed.  Usually zil_suspend() accomplishes this.
1305 	 */
1306 	ASSERT(spa_version(dmu_tx_pool(tx)->dp_spa) >= SPA_VERSION_FAST_SNAP ||
1307 	    dmu_objset_from_ds(ds, &os) != 0 ||
1308 	    bcmp(&os->os_phys->os_zil_header, &zero_zil,
1309 	    sizeof (zero_zil)) == 0);
1310 
1311 	/* Should not snapshot a dirty dataset. */
1312 	ASSERT(!txg_list_member(&ds->ds_dir->dd_pool->dp_dirty_datasets,
1313 	    ds, tx->tx_txg));
1314 
1315 	dsl_fs_ss_count_adjust(ds->ds_dir, 1, DD_FIELD_SNAPSHOT_COUNT, tx);
1316 
1317 	/*
1318 	 * The origin's ds_creation_txg has to be < TXG_INITIAL
1319 	 */
1320 	if (strcmp(snapname, ORIGIN_DIR_NAME) == 0)
1321 		crtxg = 1;
1322 	else
1323 		crtxg = tx->tx_txg;
1324 
1325 	dsobj = dmu_object_alloc(mos, DMU_OT_DSL_DATASET, 0,
1326 	    DMU_OT_DSL_DATASET, sizeof (dsl_dataset_phys_t), tx);
1327 	VERIFY0(dmu_bonus_hold(mos, dsobj, FTAG, &dbuf));
1328 	dmu_buf_will_dirty(dbuf, tx);
1329 	dsphys = dbuf->db_data;
1330 	bzero(dsphys, sizeof (dsl_dataset_phys_t));
1331 	dsphys->ds_dir_obj = ds->ds_dir->dd_object;
1332 	dsphys->ds_fsid_guid = unique_create();
1333 	(void) random_get_pseudo_bytes((void*)&dsphys->ds_guid,
1334 	    sizeof (dsphys->ds_guid));
1335 	dsphys->ds_prev_snap_obj = dsl_dataset_phys(ds)->ds_prev_snap_obj;
1336 	dsphys->ds_prev_snap_txg = dsl_dataset_phys(ds)->ds_prev_snap_txg;
1337 	dsphys->ds_next_snap_obj = ds->ds_object;
1338 	dsphys->ds_num_children = 1;
1339 	dsphys->ds_creation_time = gethrestime_sec();
1340 	dsphys->ds_creation_txg = crtxg;
1341 	dsphys->ds_deadlist_obj = dsl_dataset_phys(ds)->ds_deadlist_obj;
1342 	dsphys->ds_referenced_bytes = dsl_dataset_phys(ds)->ds_referenced_bytes;
1343 	dsphys->ds_compressed_bytes = dsl_dataset_phys(ds)->ds_compressed_bytes;
1344 	dsphys->ds_uncompressed_bytes =
1345 	    dsl_dataset_phys(ds)->ds_uncompressed_bytes;
1346 	dsphys->ds_flags = dsl_dataset_phys(ds)->ds_flags;
1347 	rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
1348 	dsphys->ds_bp = dsl_dataset_phys(ds)->ds_bp;
1349 	rrw_exit(&ds->ds_bp_rwlock, FTAG);
1350 	dmu_buf_rele(dbuf, FTAG);
1351 
1352 	for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
1353 		if (ds->ds_feature_inuse[f])
1354 			dsl_dataset_activate_feature(dsobj, f, tx);
1355 	}
1356 
1357 	ASSERT3U(ds->ds_prev != 0, ==,
1358 	    dsl_dataset_phys(ds)->ds_prev_snap_obj != 0);
1359 	if (ds->ds_prev) {
1360 		uint64_t next_clones_obj =
1361 		    dsl_dataset_phys(ds->ds_prev)->ds_next_clones_obj;
1362 		ASSERT(dsl_dataset_phys(ds->ds_prev)->ds_next_snap_obj ==
1363 		    ds->ds_object ||
1364 		    dsl_dataset_phys(ds->ds_prev)->ds_num_children > 1);
1365 		if (dsl_dataset_phys(ds->ds_prev)->ds_next_snap_obj ==
1366 		    ds->ds_object) {
1367 			dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
1368 			ASSERT3U(dsl_dataset_phys(ds)->ds_prev_snap_txg, ==,
1369 			    dsl_dataset_phys(ds->ds_prev)->ds_creation_txg);
1370 			dsl_dataset_phys(ds->ds_prev)->ds_next_snap_obj = dsobj;
1371 		} else if (next_clones_obj != 0) {
1372 			dsl_dataset_remove_from_next_clones(ds->ds_prev,
1373 			    dsphys->ds_next_snap_obj, tx);
1374 			VERIFY0(zap_add_int(mos,
1375 			    next_clones_obj, dsobj, tx));
1376 		}
1377 	}
1378 
1379 	/*
1380 	 * If we have a reference-reservation on this dataset, we will
1381 	 * need to increase the amount of refreservation being charged
1382 	 * since our unique space is going to zero.
1383 	 */
1384 	if (ds->ds_reserved) {
1385 		int64_t delta;
1386 		ASSERT(DS_UNIQUE_IS_ACCURATE(ds));
1387 		delta = MIN(dsl_dataset_phys(ds)->ds_unique_bytes,
1388 		    ds->ds_reserved);
1389 		dsl_dir_diduse_space(ds->ds_dir, DD_USED_REFRSRV,
1390 		    delta, 0, 0, tx);
1391 	}
1392 
1393 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
1394 	dsl_dataset_phys(ds)->ds_deadlist_obj =
1395 	    dsl_deadlist_clone(&ds->ds_deadlist, UINT64_MAX,
1396 	    dsl_dataset_phys(ds)->ds_prev_snap_obj, tx);
1397 	dsl_deadlist_close(&ds->ds_deadlist);
1398 	dsl_deadlist_open(&ds->ds_deadlist, mos,
1399 	    dsl_dataset_phys(ds)->ds_deadlist_obj);
1400 	dsl_deadlist_add_key(&ds->ds_deadlist,
1401 	    dsl_dataset_phys(ds)->ds_prev_snap_txg, tx);
1402 
1403 	ASSERT3U(dsl_dataset_phys(ds)->ds_prev_snap_txg, <, tx->tx_txg);
1404 	dsl_dataset_phys(ds)->ds_prev_snap_obj = dsobj;
1405 	dsl_dataset_phys(ds)->ds_prev_snap_txg = crtxg;
1406 	dsl_dataset_phys(ds)->ds_unique_bytes = 0;
1407 	if (spa_version(dp->dp_spa) >= SPA_VERSION_UNIQUE_ACCURATE)
1408 		dsl_dataset_phys(ds)->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
1409 
1410 	VERIFY0(zap_add(mos, dsl_dataset_phys(ds)->ds_snapnames_zapobj,
1411 	    snapname, 8, 1, &dsobj, tx));
1412 
1413 	if (ds->ds_prev)
1414 		dsl_dataset_rele(ds->ds_prev, ds);
1415 	VERIFY0(dsl_dataset_hold_obj(dp,
1416 	    dsl_dataset_phys(ds)->ds_prev_snap_obj, ds, &ds->ds_prev));
1417 
1418 	dsl_scan_ds_snapshotted(ds, tx);
1419 
1420 	dsl_dir_snap_cmtime_update(ds->ds_dir);
1421 
1422 	spa_history_log_internal_ds(ds->ds_prev, "snapshot", tx, "");
1423 }
1424 
1425 static void
1426 dsl_dataset_snapshot_sync(void *arg, dmu_tx_t *tx)
1427 {
1428 	dsl_dataset_snapshot_arg_t *ddsa = arg;
1429 	dsl_pool_t *dp = dmu_tx_pool(tx);
1430 	nvpair_t *pair;
1431 
1432 	for (pair = nvlist_next_nvpair(ddsa->ddsa_snaps, NULL);
1433 	    pair != NULL; pair = nvlist_next_nvpair(ddsa->ddsa_snaps, pair)) {
1434 		dsl_dataset_t *ds;
1435 		char *name, *atp;
1436 		char dsname[ZFS_MAX_DATASET_NAME_LEN];
1437 
1438 		name = nvpair_name(pair);
1439 		atp = strchr(name, '@');
1440 		(void) strlcpy(dsname, name, atp - name + 1);
1441 		VERIFY0(dsl_dataset_hold(dp, dsname, FTAG, &ds));
1442 
1443 		dsl_dataset_snapshot_sync_impl(ds, atp + 1, tx);
1444 		if (ddsa->ddsa_props != NULL) {
1445 			dsl_props_set_sync_impl(ds->ds_prev,
1446 			    ZPROP_SRC_LOCAL, ddsa->ddsa_props, tx);
1447 		}
1448 		dsl_dataset_rele(ds, FTAG);
1449 	}
1450 }
1451 
1452 /*
1453  * The snapshots must all be in the same pool.
1454  * All-or-nothing: if there are any failures, nothing will be modified.
1455  */
1456 int
1457 dsl_dataset_snapshot(nvlist_t *snaps, nvlist_t *props, nvlist_t *errors)
1458 {
1459 	dsl_dataset_snapshot_arg_t ddsa;
1460 	nvpair_t *pair;
1461 	boolean_t needsuspend;
1462 	int error;
1463 	spa_t *spa;
1464 	char *firstname;
1465 	nvlist_t *suspended = NULL;
1466 
1467 	pair = nvlist_next_nvpair(snaps, NULL);
1468 	if (pair == NULL)
1469 		return (0);
1470 	firstname = nvpair_name(pair);
1471 
1472 	error = spa_open(firstname, &spa, FTAG);
1473 	if (error != 0)
1474 		return (error);
1475 	needsuspend = (spa_version(spa) < SPA_VERSION_FAST_SNAP);
1476 	spa_close(spa, FTAG);
1477 
1478 	if (needsuspend) {
1479 		suspended = fnvlist_alloc();
1480 		for (pair = nvlist_next_nvpair(snaps, NULL); pair != NULL;
1481 		    pair = nvlist_next_nvpair(snaps, pair)) {
1482 			char fsname[ZFS_MAX_DATASET_NAME_LEN];
1483 			char *snapname = nvpair_name(pair);
1484 			char *atp;
1485 			void *cookie;
1486 
1487 			atp = strchr(snapname, '@');
1488 			if (atp == NULL) {
1489 				error = SET_ERROR(EINVAL);
1490 				break;
1491 			}
1492 			(void) strlcpy(fsname, snapname, atp - snapname + 1);
1493 
1494 			error = zil_suspend(fsname, &cookie);
1495 			if (error != 0)
1496 				break;
1497 			fnvlist_add_uint64(suspended, fsname,
1498 			    (uintptr_t)cookie);
1499 		}
1500 	}
1501 
1502 	ddsa.ddsa_snaps = snaps;
1503 	ddsa.ddsa_props = props;
1504 	ddsa.ddsa_errors = errors;
1505 	ddsa.ddsa_cr = CRED();
1506 
1507 	if (error == 0) {
1508 		error = dsl_sync_task(firstname, dsl_dataset_snapshot_check,
1509 		    dsl_dataset_snapshot_sync, &ddsa,
1510 		    fnvlist_num_pairs(snaps) * 3, ZFS_SPACE_CHECK_NORMAL);
1511 	}
1512 
1513 	if (suspended != NULL) {
1514 		for (pair = nvlist_next_nvpair(suspended, NULL); pair != NULL;
1515 		    pair = nvlist_next_nvpair(suspended, pair)) {
1516 			zil_resume((void *)(uintptr_t)
1517 			    fnvpair_value_uint64(pair));
1518 		}
1519 		fnvlist_free(suspended);
1520 	}
1521 
1522 	return (error);
1523 }
1524 
1525 typedef struct dsl_dataset_snapshot_tmp_arg {
1526 	const char *ddsta_fsname;
1527 	const char *ddsta_snapname;
1528 	minor_t ddsta_cleanup_minor;
1529 	const char *ddsta_htag;
1530 } dsl_dataset_snapshot_tmp_arg_t;
1531 
1532 static int
1533 dsl_dataset_snapshot_tmp_check(void *arg, dmu_tx_t *tx)
1534 {
1535 	dsl_dataset_snapshot_tmp_arg_t *ddsta = arg;
1536 	dsl_pool_t *dp = dmu_tx_pool(tx);
1537 	dsl_dataset_t *ds;
1538 	int error;
1539 
1540 	error = dsl_dataset_hold(dp, ddsta->ddsta_fsname, FTAG, &ds);
1541 	if (error != 0)
1542 		return (error);
1543 
1544 	/* NULL cred means no limit check for tmp snapshot */
1545 	error = dsl_dataset_snapshot_check_impl(ds, ddsta->ddsta_snapname,
1546 	    tx, B_FALSE, 0, NULL);
1547 	if (error != 0) {
1548 		dsl_dataset_rele(ds, FTAG);
1549 		return (error);
1550 	}
1551 
1552 	if (spa_version(dp->dp_spa) < SPA_VERSION_USERREFS) {
1553 		dsl_dataset_rele(ds, FTAG);
1554 		return (SET_ERROR(ENOTSUP));
1555 	}
1556 	error = dsl_dataset_user_hold_check_one(NULL, ddsta->ddsta_htag,
1557 	    B_TRUE, tx);
1558 	if (error != 0) {
1559 		dsl_dataset_rele(ds, FTAG);
1560 		return (error);
1561 	}
1562 
1563 	dsl_dataset_rele(ds, FTAG);
1564 	return (0);
1565 }
1566 
1567 static void
1568 dsl_dataset_snapshot_tmp_sync(void *arg, dmu_tx_t *tx)
1569 {
1570 	dsl_dataset_snapshot_tmp_arg_t *ddsta = arg;
1571 	dsl_pool_t *dp = dmu_tx_pool(tx);
1572 	dsl_dataset_t *ds;
1573 
1574 	VERIFY0(dsl_dataset_hold(dp, ddsta->ddsta_fsname, FTAG, &ds));
1575 
1576 	dsl_dataset_snapshot_sync_impl(ds, ddsta->ddsta_snapname, tx);
1577 	dsl_dataset_user_hold_sync_one(ds->ds_prev, ddsta->ddsta_htag,
1578 	    ddsta->ddsta_cleanup_minor, gethrestime_sec(), tx);
1579 	dsl_destroy_snapshot_sync_impl(ds->ds_prev, B_TRUE, tx);
1580 
1581 	dsl_dataset_rele(ds, FTAG);
1582 }
1583 
1584 int
1585 dsl_dataset_snapshot_tmp(const char *fsname, const char *snapname,
1586     minor_t cleanup_minor, const char *htag)
1587 {
1588 	dsl_dataset_snapshot_tmp_arg_t ddsta;
1589 	int error;
1590 	spa_t *spa;
1591 	boolean_t needsuspend;
1592 	void *cookie;
1593 
1594 	ddsta.ddsta_fsname = fsname;
1595 	ddsta.ddsta_snapname = snapname;
1596 	ddsta.ddsta_cleanup_minor = cleanup_minor;
1597 	ddsta.ddsta_htag = htag;
1598 
1599 	error = spa_open(fsname, &spa, FTAG);
1600 	if (error != 0)
1601 		return (error);
1602 	needsuspend = (spa_version(spa) < SPA_VERSION_FAST_SNAP);
1603 	spa_close(spa, FTAG);
1604 
1605 	if (needsuspend) {
1606 		error = zil_suspend(fsname, &cookie);
1607 		if (error != 0)
1608 			return (error);
1609 	}
1610 
1611 	error = dsl_sync_task(fsname, dsl_dataset_snapshot_tmp_check,
1612 	    dsl_dataset_snapshot_tmp_sync, &ddsta, 3, ZFS_SPACE_CHECK_RESERVED);
1613 
1614 	if (needsuspend)
1615 		zil_resume(cookie);
1616 	return (error);
1617 }
1618 
1619 void
1620 dsl_dataset_sync(dsl_dataset_t *ds, zio_t *zio, dmu_tx_t *tx)
1621 {
1622 	ASSERT(dmu_tx_is_syncing(tx));
1623 	ASSERT(ds->ds_objset != NULL);
1624 	ASSERT(dsl_dataset_phys(ds)->ds_next_snap_obj == 0);
1625 
1626 	/*
1627 	 * in case we had to change ds_fsid_guid when we opened it,
1628 	 * sync it out now.
1629 	 */
1630 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
1631 	dsl_dataset_phys(ds)->ds_fsid_guid = ds->ds_fsid_guid;
1632 
1633 	if (ds->ds_resume_bytes[tx->tx_txg & TXG_MASK] != 0) {
1634 		VERIFY0(zap_update(tx->tx_pool->dp_meta_objset,
1635 		    ds->ds_object, DS_FIELD_RESUME_OBJECT, 8, 1,
1636 		    &ds->ds_resume_object[tx->tx_txg & TXG_MASK], tx));
1637 		VERIFY0(zap_update(tx->tx_pool->dp_meta_objset,
1638 		    ds->ds_object, DS_FIELD_RESUME_OFFSET, 8, 1,
1639 		    &ds->ds_resume_offset[tx->tx_txg & TXG_MASK], tx));
1640 		VERIFY0(zap_update(tx->tx_pool->dp_meta_objset,
1641 		    ds->ds_object, DS_FIELD_RESUME_BYTES, 8, 1,
1642 		    &ds->ds_resume_bytes[tx->tx_txg & TXG_MASK], tx));
1643 		ds->ds_resume_object[tx->tx_txg & TXG_MASK] = 0;
1644 		ds->ds_resume_offset[tx->tx_txg & TXG_MASK] = 0;
1645 		ds->ds_resume_bytes[tx->tx_txg & TXG_MASK] = 0;
1646 	}
1647 
1648 	dmu_objset_sync(ds->ds_objset, zio, tx);
1649 
1650 	for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
1651 		if (ds->ds_feature_activation_needed[f]) {
1652 			if (ds->ds_feature_inuse[f])
1653 				continue;
1654 			dsl_dataset_activate_feature(ds->ds_object, f, tx);
1655 			ds->ds_feature_inuse[f] = B_TRUE;
1656 		}
1657 	}
1658 }
1659 
1660 static int
1661 deadlist_enqueue_cb(void *arg, const blkptr_t *bp, dmu_tx_t *tx)
1662 {
1663 	dsl_deadlist_t *dl = arg;
1664 	dsl_deadlist_insert(dl, bp, tx);
1665 	return (0);
1666 }
1667 
1668 void
1669 dsl_dataset_sync_done(dsl_dataset_t *ds, dmu_tx_t *tx)
1670 {
1671 	objset_t *os = ds->ds_objset;
1672 
1673 	bplist_iterate(&ds->ds_pending_deadlist,
1674 	    deadlist_enqueue_cb, &ds->ds_deadlist, tx);
1675 
1676 	if (os->os_synced_dnodes != NULL) {
1677 		multilist_destroy(os->os_synced_dnodes);
1678 		os->os_synced_dnodes = NULL;
1679 	}
1680 
1681 	ASSERT(!dmu_objset_is_dirty(os, dmu_tx_get_txg(tx)));
1682 
1683 	dmu_buf_rele(ds->ds_dbuf, ds);
1684 }
1685 
1686 int
1687 get_clones_stat_impl(dsl_dataset_t *ds, nvlist_t *val)
1688 {
1689 	uint64_t count = 0;
1690 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
1691 	zap_cursor_t zc;
1692 	zap_attribute_t za;
1693 
1694 	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
1695 
1696 	/*
1697 	 * There may be missing entries in ds_next_clones_obj
1698 	 * due to a bug in a previous version of the code.
1699 	 * Only trust it if it has the right number of entries.
1700 	 */
1701 	if (dsl_dataset_phys(ds)->ds_next_clones_obj != 0) {
1702 		VERIFY0(zap_count(mos, dsl_dataset_phys(ds)->ds_next_clones_obj,
1703 		    &count));
1704 	}
1705 	if (count != dsl_dataset_phys(ds)->ds_num_children - 1) {
1706 		return (ENOENT);
1707 	}
1708 	for (zap_cursor_init(&zc, mos,
1709 	    dsl_dataset_phys(ds)->ds_next_clones_obj);
1710 	    zap_cursor_retrieve(&zc, &za) == 0;
1711 	    zap_cursor_advance(&zc)) {
1712 		dsl_dataset_t *clone;
1713 		char buf[ZFS_MAX_DATASET_NAME_LEN];
1714 		VERIFY0(dsl_dataset_hold_obj(ds->ds_dir->dd_pool,
1715 		    za.za_first_integer, FTAG, &clone));
1716 		dsl_dir_name(clone->ds_dir, buf);
1717 		fnvlist_add_boolean(val, buf);
1718 		dsl_dataset_rele(clone, FTAG);
1719 	}
1720 	zap_cursor_fini(&zc);
1721 	return (0);
1722 }
1723 
1724 void
1725 get_clones_stat(dsl_dataset_t *ds, nvlist_t *nv)
1726 {
1727 	nvlist_t *propval = fnvlist_alloc();
1728 	nvlist_t *val;
1729 
1730 	/*
1731 	 * We use nvlist_alloc() instead of fnvlist_alloc() because the
1732 	 * latter would allocate the list with NV_UNIQUE_NAME flag.
1733 	 * As a result, every time a clone name is appended to the list
1734 	 * it would be (linearly) searched for for a duplicate name.
1735 	 * We already know that all clone names must be unique and we
1736 	 * want avoid the quadratic complexity of double-checking that
1737 	 * because we can have a large number of clones.
1738 	 */
1739 	VERIFY0(nvlist_alloc(&val, 0, KM_SLEEP));
1740 
1741 	if (get_clones_stat_impl(ds, val) == 0) {
1742 		fnvlist_add_nvlist(propval, ZPROP_VALUE, val);
1743 		fnvlist_add_nvlist(nv, zfs_prop_to_name(ZFS_PROP_CLONES),
1744 		    propval);
1745 	} else {
1746 		nvlist_free(val);
1747 		nvlist_free(propval);
1748 	}
1749 }
1750 
1751 /*
1752  * Returns a string that represents the receive resume stats token. It should
1753  * be freed with strfree().
1754  */
1755 char *
1756 get_receive_resume_stats_impl(dsl_dataset_t *ds)
1757 {
1758 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1759 
1760 	if (dsl_dataset_has_resume_receive_state(ds)) {
1761 		char *str;
1762 		void *packed;
1763 		uint8_t *compressed;
1764 		uint64_t val;
1765 		nvlist_t *token_nv = fnvlist_alloc();
1766 		size_t packed_size, compressed_size;
1767 
1768 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
1769 		    DS_FIELD_RESUME_FROMGUID, sizeof (val), 1, &val) == 0) {
1770 			fnvlist_add_uint64(token_nv, "fromguid", val);
1771 		}
1772 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
1773 		    DS_FIELD_RESUME_OBJECT, sizeof (val), 1, &val) == 0) {
1774 			fnvlist_add_uint64(token_nv, "object", val);
1775 		}
1776 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
1777 		    DS_FIELD_RESUME_OFFSET, sizeof (val), 1, &val) == 0) {
1778 			fnvlist_add_uint64(token_nv, "offset", val);
1779 		}
1780 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
1781 		    DS_FIELD_RESUME_BYTES, sizeof (val), 1, &val) == 0) {
1782 			fnvlist_add_uint64(token_nv, "bytes", val);
1783 		}
1784 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
1785 		    DS_FIELD_RESUME_TOGUID, sizeof (val), 1, &val) == 0) {
1786 			fnvlist_add_uint64(token_nv, "toguid", val);
1787 		}
1788 		char buf[256];
1789 		if (zap_lookup(dp->dp_meta_objset, ds->ds_object,
1790 		    DS_FIELD_RESUME_TONAME, 1, sizeof (buf), buf) == 0) {
1791 			fnvlist_add_string(token_nv, "toname", buf);
1792 		}
1793 		if (zap_contains(dp->dp_meta_objset, ds->ds_object,
1794 		    DS_FIELD_RESUME_LARGEBLOCK) == 0) {
1795 			fnvlist_add_boolean(token_nv, "largeblockok");
1796 		}
1797 		if (zap_contains(dp->dp_meta_objset, ds->ds_object,
1798 		    DS_FIELD_RESUME_EMBEDOK) == 0) {
1799 			fnvlist_add_boolean(token_nv, "embedok");
1800 		}
1801 		if (zap_contains(dp->dp_meta_objset, ds->ds_object,
1802 		    DS_FIELD_RESUME_COMPRESSOK) == 0) {
1803 			fnvlist_add_boolean(token_nv, "compressok");
1804 		}
1805 		packed = fnvlist_pack(token_nv, &packed_size);
1806 		fnvlist_free(token_nv);
1807 		compressed = kmem_alloc(packed_size, KM_SLEEP);
1808 
1809 		compressed_size = gzip_compress(packed, compressed,
1810 		    packed_size, packed_size, 6);
1811 
1812 		zio_cksum_t cksum;
1813 		fletcher_4_native(compressed, compressed_size, NULL, &cksum);
1814 
1815 		str = kmem_alloc(compressed_size * 2 + 1, KM_SLEEP);
1816 		for (int i = 0; i < compressed_size; i++) {
1817 			(void) sprintf(str + i * 2, "%02x", compressed[i]);
1818 		}
1819 		str[compressed_size * 2] = '\0';
1820 		char *propval = kmem_asprintf("%u-%llx-%llx-%s",
1821 		    ZFS_SEND_RESUME_TOKEN_VERSION,
1822 		    (longlong_t)cksum.zc_word[0],
1823 		    (longlong_t)packed_size, str);
1824 		kmem_free(packed, packed_size);
1825 		kmem_free(str, compressed_size * 2 + 1);
1826 		kmem_free(compressed, packed_size);
1827 		return (propval);
1828 	}
1829 	return (strdup(""));
1830 }
1831 
1832 /*
1833  * Returns a string that represents the receive resume stats token of the
1834  * dataset's child. It should be freed with strfree().
1835  */
1836 char *
1837 get_child_receive_stats(dsl_dataset_t *ds)
1838 {
1839 	char recvname[ZFS_MAX_DATASET_NAME_LEN + 6];
1840 	dsl_dataset_t *recv_ds;
1841 	dsl_dataset_name(ds, recvname);
1842 	if (strlcat(recvname, "/", sizeof (recvname)) <
1843 	    sizeof (recvname) &&
1844 	    strlcat(recvname, recv_clone_name, sizeof (recvname)) <
1845 	    sizeof (recvname) &&
1846 	    dsl_dataset_hold(ds->ds_dir->dd_pool, recvname, FTAG,
1847 	    &recv_ds)  == 0) {
1848 		char *propval = get_receive_resume_stats_impl(recv_ds);
1849 		dsl_dataset_rele(recv_ds, FTAG);
1850 		return (propval);
1851 	}
1852 	return (strdup(""));
1853 }
1854 
1855 static void
1856 get_receive_resume_stats(dsl_dataset_t *ds, nvlist_t *nv)
1857 {
1858 	char *propval = get_receive_resume_stats_impl(ds);
1859 	if (strcmp(propval, "") != 0) {
1860 		dsl_prop_nvlist_add_string(nv,
1861 		    ZFS_PROP_RECEIVE_RESUME_TOKEN, propval);
1862 	} else {
1863 		char *childval = get_child_receive_stats(ds);
1864 		if (strcmp(childval, "") != 0) {
1865 			dsl_prop_nvlist_add_string(nv,
1866 			    ZFS_PROP_RECEIVE_RESUME_TOKEN, childval);
1867 		}
1868 		strfree(childval);
1869 	}
1870 	strfree(propval);
1871 }
1872 
1873 uint64_t
1874 dsl_get_refratio(dsl_dataset_t *ds)
1875 {
1876 	uint64_t ratio = dsl_dataset_phys(ds)->ds_compressed_bytes == 0 ? 100 :
1877 	    (dsl_dataset_phys(ds)->ds_uncompressed_bytes * 100 /
1878 	    dsl_dataset_phys(ds)->ds_compressed_bytes);
1879 	return (ratio);
1880 }
1881 
1882 uint64_t
1883 dsl_get_logicalreferenced(dsl_dataset_t *ds)
1884 {
1885 	return (dsl_dataset_phys(ds)->ds_uncompressed_bytes);
1886 }
1887 
1888 uint64_t
1889 dsl_get_compressratio(dsl_dataset_t *ds)
1890 {
1891 	if (ds->ds_is_snapshot) {
1892 		return (dsl_get_refratio(ds));
1893 	} else {
1894 		dsl_dir_t *dd = ds->ds_dir;
1895 		mutex_enter(&dd->dd_lock);
1896 		uint64_t val = dsl_dir_get_compressratio(dd);
1897 		mutex_exit(&dd->dd_lock);
1898 		return (val);
1899 	}
1900 }
1901 
1902 uint64_t
1903 dsl_get_used(dsl_dataset_t *ds)
1904 {
1905 	if (ds->ds_is_snapshot) {
1906 		return (dsl_dataset_phys(ds)->ds_unique_bytes);
1907 	} else {
1908 		dsl_dir_t *dd = ds->ds_dir;
1909 		mutex_enter(&dd->dd_lock);
1910 		uint64_t val = dsl_dir_get_used(dd);
1911 		mutex_exit(&dd->dd_lock);
1912 		return (val);
1913 	}
1914 }
1915 
1916 uint64_t
1917 dsl_get_creation(dsl_dataset_t *ds)
1918 {
1919 	return (dsl_dataset_phys(ds)->ds_creation_time);
1920 }
1921 
1922 uint64_t
1923 dsl_get_creationtxg(dsl_dataset_t *ds)
1924 {
1925 	return (dsl_dataset_phys(ds)->ds_creation_txg);
1926 }
1927 
1928 uint64_t
1929 dsl_get_refquota(dsl_dataset_t *ds)
1930 {
1931 	return (ds->ds_quota);
1932 }
1933 
1934 uint64_t
1935 dsl_get_refreservation(dsl_dataset_t *ds)
1936 {
1937 	return (ds->ds_reserved);
1938 }
1939 
1940 uint64_t
1941 dsl_get_guid(dsl_dataset_t *ds)
1942 {
1943 	return (dsl_dataset_phys(ds)->ds_guid);
1944 }
1945 
1946 uint64_t
1947 dsl_get_unique(dsl_dataset_t *ds)
1948 {
1949 	return (dsl_dataset_phys(ds)->ds_unique_bytes);
1950 }
1951 
1952 uint64_t
1953 dsl_get_objsetid(dsl_dataset_t *ds)
1954 {
1955 	return (ds->ds_object);
1956 }
1957 
1958 uint64_t
1959 dsl_get_userrefs(dsl_dataset_t *ds)
1960 {
1961 	return (ds->ds_userrefs);
1962 }
1963 
1964 uint64_t
1965 dsl_get_defer_destroy(dsl_dataset_t *ds)
1966 {
1967 	return (DS_IS_DEFER_DESTROY(ds) ? 1 : 0);
1968 }
1969 
1970 uint64_t
1971 dsl_get_referenced(dsl_dataset_t *ds)
1972 {
1973 	return (dsl_dataset_phys(ds)->ds_referenced_bytes);
1974 }
1975 
1976 uint64_t
1977 dsl_get_numclones(dsl_dataset_t *ds)
1978 {
1979 	ASSERT(ds->ds_is_snapshot);
1980 	return (dsl_dataset_phys(ds)->ds_num_children - 1);
1981 }
1982 
1983 uint64_t
1984 dsl_get_inconsistent(dsl_dataset_t *ds)
1985 {
1986 	return ((dsl_dataset_phys(ds)->ds_flags & DS_FLAG_INCONSISTENT) ?
1987 	    1 : 0);
1988 }
1989 
1990 uint64_t
1991 dsl_get_available(dsl_dataset_t *ds)
1992 {
1993 	uint64_t refdbytes = dsl_get_referenced(ds);
1994 	uint64_t availbytes = dsl_dir_space_available(ds->ds_dir,
1995 	    NULL, 0, TRUE);
1996 	if (ds->ds_reserved > dsl_dataset_phys(ds)->ds_unique_bytes) {
1997 		availbytes +=
1998 		    ds->ds_reserved - dsl_dataset_phys(ds)->ds_unique_bytes;
1999 	}
2000 	if (ds->ds_quota != 0) {
2001 		/*
2002 		 * Adjust available bytes according to refquota
2003 		 */
2004 		if (refdbytes < ds->ds_quota) {
2005 			availbytes = MIN(availbytes,
2006 			    ds->ds_quota - refdbytes);
2007 		} else {
2008 			availbytes = 0;
2009 		}
2010 	}
2011 	return (availbytes);
2012 }
2013 
2014 int
2015 dsl_get_written(dsl_dataset_t *ds, uint64_t *written)
2016 {
2017 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2018 	dsl_dataset_t *prev;
2019 	int err = dsl_dataset_hold_obj(dp,
2020 	    dsl_dataset_phys(ds)->ds_prev_snap_obj, FTAG, &prev);
2021 	if (err == 0) {
2022 		uint64_t comp, uncomp;
2023 		err = dsl_dataset_space_written(prev, ds, written,
2024 		    &comp, &uncomp);
2025 		dsl_dataset_rele(prev, FTAG);
2026 	}
2027 	return (err);
2028 }
2029 
2030 /*
2031  * 'snap' should be a buffer of size ZFS_MAX_DATASET_NAME_LEN.
2032  */
2033 int
2034 dsl_get_prev_snap(dsl_dataset_t *ds, char *snap)
2035 {
2036 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2037 	if (ds->ds_prev != NULL && ds->ds_prev != dp->dp_origin_snap) {
2038 		dsl_dataset_name(ds->ds_prev, snap);
2039 		return (0);
2040 	} else {
2041 		return (ENOENT);
2042 	}
2043 }
2044 
2045 /*
2046  * Returns the mountpoint property and source for the given dataset in the value
2047  * and source buffers. The value buffer must be at least as large as MAXPATHLEN
2048  * and the source buffer as least as large a ZFS_MAX_DATASET_NAME_LEN.
2049  * Returns 0 on success and an error on failure.
2050  */
2051 int
2052 dsl_get_mountpoint(dsl_dataset_t *ds, const char *dsname, char *value,
2053     char *source)
2054 {
2055 	int error;
2056 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2057 
2058 	/* Retrieve the mountpoint value stored in the zap opbject */
2059 	error = dsl_prop_get_ds(ds, zfs_prop_to_name(ZFS_PROP_MOUNTPOINT), 1,
2060 	    ZAP_MAXVALUELEN, value, source);
2061 	if (error != 0) {
2062 		return (error);
2063 	}
2064 
2065 	/* Process the dsname and source to find the full mountpoint string */
2066 	if (value[0] == '/') {
2067 		char *buf = kmem_alloc(ZAP_MAXVALUELEN, KM_SLEEP);
2068 		char *root = buf;
2069 		const char *relpath;
2070 
2071 		/*
2072 		 * If we inherit the mountpoint, even from a dataset
2073 		 * with a received value, the source will be the path of
2074 		 * the dataset we inherit from. If source is
2075 		 * ZPROP_SOURCE_VAL_RECVD, the received value is not
2076 		 * inherited.
2077 		 */
2078 		if (strcmp(source, ZPROP_SOURCE_VAL_RECVD) == 0) {
2079 			relpath = "";
2080 		} else {
2081 			ASSERT0(strncmp(dsname, source, strlen(source)));
2082 			relpath = dsname + strlen(source);
2083 			if (relpath[0] == '/')
2084 				relpath++;
2085 		}
2086 
2087 		spa_altroot(dp->dp_spa, root, ZAP_MAXVALUELEN);
2088 
2089 		/*
2090 		 * Special case an alternate root of '/'. This will
2091 		 * avoid having multiple leading slashes in the
2092 		 * mountpoint path.
2093 		 */
2094 		if (strcmp(root, "/") == 0)
2095 			root++;
2096 
2097 		/*
2098 		 * If the mountpoint is '/' then skip over this
2099 		 * if we are obtaining either an alternate root or
2100 		 * an inherited mountpoint.
2101 		 */
2102 		char *mnt = value;
2103 		if (value[1] == '\0' && (root[0] != '\0' ||
2104 		    relpath[0] != '\0'))
2105 			mnt = value + 1;
2106 
2107 		if (relpath[0] == '\0') {
2108 			(void) snprintf(value, ZAP_MAXVALUELEN, "%s%s",
2109 			    root, mnt);
2110 		} else {
2111 			(void) snprintf(value, ZAP_MAXVALUELEN, "%s%s%s%s",
2112 			    root, mnt, relpath[0] == '@' ? "" : "/",
2113 			    relpath);
2114 		}
2115 		kmem_free(buf, ZAP_MAXVALUELEN);
2116 	} else {
2117 		/* 'legacy' or 'none' */
2118 		(void) snprintf(value, ZAP_MAXVALUELEN, "%s", value);
2119 	}
2120 	return (0);
2121 }
2122 
2123 void
2124 dsl_dataset_stats(dsl_dataset_t *ds, nvlist_t *nv)
2125 {
2126 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2127 
2128 	ASSERT(dsl_pool_config_held(dp));
2129 
2130 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFRATIO,
2131 	    dsl_get_refratio(ds));
2132 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_LOGICALREFERENCED,
2133 	    dsl_get_logicalreferenced(ds));
2134 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_COMPRESSRATIO,
2135 	    dsl_get_compressratio(ds));
2136 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USED,
2137 	    dsl_get_used(ds));
2138 
2139 	if (ds->ds_is_snapshot) {
2140 		get_clones_stat(ds, nv);
2141 	} else {
2142 		char buf[ZFS_MAX_DATASET_NAME_LEN];
2143 		if (dsl_get_prev_snap(ds, buf) == 0)
2144 			dsl_prop_nvlist_add_string(nv, ZFS_PROP_PREV_SNAP,
2145 			    buf);
2146 		dsl_dir_stats(ds->ds_dir, nv);
2147 	}
2148 
2149 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_AVAILABLE,
2150 	    dsl_get_available(ds));
2151 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFERENCED,
2152 	    dsl_get_referenced(ds));
2153 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_CREATION,
2154 	    dsl_get_creation(ds));
2155 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_CREATETXG,
2156 	    dsl_get_creationtxg(ds));
2157 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFQUOTA,
2158 	    dsl_get_refquota(ds));
2159 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFRESERVATION,
2160 	    dsl_get_refreservation(ds));
2161 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_GUID,
2162 	    dsl_get_guid(ds));
2163 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_UNIQUE,
2164 	    dsl_get_unique(ds));
2165 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_OBJSETID,
2166 	    dsl_get_objsetid(ds));
2167 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USERREFS,
2168 	    dsl_get_userrefs(ds));
2169 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_DEFER_DESTROY,
2170 	    dsl_get_defer_destroy(ds));
2171 
2172 	if (dsl_dataset_phys(ds)->ds_prev_snap_obj != 0) {
2173 		uint64_t written;
2174 		if (dsl_get_written(ds, &written) == 0) {
2175 			dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_WRITTEN,
2176 			    written);
2177 		}
2178 	}
2179 
2180 	if (!dsl_dataset_is_snapshot(ds)) {
2181 		/*
2182 		 * A failed "newfs" (e.g. full) resumable receive leaves
2183 		 * the stats set on this dataset.  Check here for the prop.
2184 		 */
2185 		get_receive_resume_stats(ds, nv);
2186 
2187 		/*
2188 		 * A failed incremental resumable receive leaves the
2189 		 * stats set on our child named "%recv".  Check the child
2190 		 * for the prop.
2191 		 */
2192 		/* 6 extra bytes for /%recv */
2193 		char recvname[ZFS_MAX_DATASET_NAME_LEN + 6];
2194 		dsl_dataset_t *recv_ds;
2195 		dsl_dataset_name(ds, recvname);
2196 		if (strlcat(recvname, "/", sizeof (recvname)) <
2197 		    sizeof (recvname) &&
2198 		    strlcat(recvname, recv_clone_name, sizeof (recvname)) <
2199 		    sizeof (recvname) &&
2200 		    dsl_dataset_hold(dp, recvname, FTAG, &recv_ds) == 0) {
2201 			get_receive_resume_stats(recv_ds, nv);
2202 			dsl_dataset_rele(recv_ds, FTAG);
2203 		}
2204 	}
2205 }
2206 
2207 void
2208 dsl_dataset_fast_stat(dsl_dataset_t *ds, dmu_objset_stats_t *stat)
2209 {
2210 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2211 	ASSERT(dsl_pool_config_held(dp));
2212 
2213 	stat->dds_creation_txg = dsl_get_creationtxg(ds);
2214 	stat->dds_inconsistent = dsl_get_inconsistent(ds);
2215 	stat->dds_guid = dsl_get_guid(ds);
2216 	stat->dds_origin[0] = '\0';
2217 	if (ds->ds_is_snapshot) {
2218 		stat->dds_is_snapshot = B_TRUE;
2219 		stat->dds_num_clones = dsl_get_numclones(ds);
2220 	} else {
2221 		stat->dds_is_snapshot = B_FALSE;
2222 		stat->dds_num_clones = 0;
2223 
2224 		if (dsl_dir_is_clone(ds->ds_dir)) {
2225 			dsl_dir_get_origin(ds->ds_dir, stat->dds_origin);
2226 		}
2227 	}
2228 }
2229 
2230 uint64_t
2231 dsl_dataset_fsid_guid(dsl_dataset_t *ds)
2232 {
2233 	return (ds->ds_fsid_guid);
2234 }
2235 
2236 void
2237 dsl_dataset_space(dsl_dataset_t *ds,
2238     uint64_t *refdbytesp, uint64_t *availbytesp,
2239     uint64_t *usedobjsp, uint64_t *availobjsp)
2240 {
2241 	*refdbytesp = dsl_dataset_phys(ds)->ds_referenced_bytes;
2242 	*availbytesp = dsl_dir_space_available(ds->ds_dir, NULL, 0, TRUE);
2243 	if (ds->ds_reserved > dsl_dataset_phys(ds)->ds_unique_bytes)
2244 		*availbytesp +=
2245 		    ds->ds_reserved - dsl_dataset_phys(ds)->ds_unique_bytes;
2246 	if (ds->ds_quota != 0) {
2247 		/*
2248 		 * Adjust available bytes according to refquota
2249 		 */
2250 		if (*refdbytesp < ds->ds_quota)
2251 			*availbytesp = MIN(*availbytesp,
2252 			    ds->ds_quota - *refdbytesp);
2253 		else
2254 			*availbytesp = 0;
2255 	}
2256 	rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
2257 	*usedobjsp = BP_GET_FILL(&dsl_dataset_phys(ds)->ds_bp);
2258 	rrw_exit(&ds->ds_bp_rwlock, FTAG);
2259 	*availobjsp = DN_MAX_OBJECT - *usedobjsp;
2260 }
2261 
2262 boolean_t
2263 dsl_dataset_modified_since_snap(dsl_dataset_t *ds, dsl_dataset_t *snap)
2264 {
2265 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2266 	uint64_t birth;
2267 
2268 	ASSERT(dsl_pool_config_held(dp));
2269 	if (snap == NULL)
2270 		return (B_FALSE);
2271 	rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
2272 	birth = dsl_dataset_get_blkptr(ds)->blk_birth;
2273 	rrw_exit(&ds->ds_bp_rwlock, FTAG);
2274 	if (birth > dsl_dataset_phys(snap)->ds_creation_txg) {
2275 		objset_t *os, *os_snap;
2276 		/*
2277 		 * It may be that only the ZIL differs, because it was
2278 		 * reset in the head.  Don't count that as being
2279 		 * modified.
2280 		 */
2281 		if (dmu_objset_from_ds(ds, &os) != 0)
2282 			return (B_TRUE);
2283 		if (dmu_objset_from_ds(snap, &os_snap) != 0)
2284 			return (B_TRUE);
2285 		return (bcmp(&os->os_phys->os_meta_dnode,
2286 		    &os_snap->os_phys->os_meta_dnode,
2287 		    sizeof (os->os_phys->os_meta_dnode)) != 0);
2288 	}
2289 	return (B_FALSE);
2290 }
2291 
2292 typedef struct dsl_dataset_rename_snapshot_arg {
2293 	const char *ddrsa_fsname;
2294 	const char *ddrsa_oldsnapname;
2295 	const char *ddrsa_newsnapname;
2296 	boolean_t ddrsa_recursive;
2297 	dmu_tx_t *ddrsa_tx;
2298 } dsl_dataset_rename_snapshot_arg_t;
2299 
2300 /* ARGSUSED */
2301 static int
2302 dsl_dataset_rename_snapshot_check_impl(dsl_pool_t *dp,
2303     dsl_dataset_t *hds, void *arg)
2304 {
2305 	dsl_dataset_rename_snapshot_arg_t *ddrsa = arg;
2306 	int error;
2307 	uint64_t val;
2308 
2309 	error = dsl_dataset_snap_lookup(hds, ddrsa->ddrsa_oldsnapname, &val);
2310 	if (error != 0) {
2311 		/* ignore nonexistent snapshots */
2312 		return (error == ENOENT ? 0 : error);
2313 	}
2314 
2315 	/* new name should not exist */
2316 	error = dsl_dataset_snap_lookup(hds, ddrsa->ddrsa_newsnapname, &val);
2317 	if (error == 0)
2318 		error = SET_ERROR(EEXIST);
2319 	else if (error == ENOENT)
2320 		error = 0;
2321 
2322 	/* dataset name + 1 for the "@" + the new snapshot name must fit */
2323 	if (dsl_dir_namelen(hds->ds_dir) + 1 +
2324 	    strlen(ddrsa->ddrsa_newsnapname) >= ZFS_MAX_DATASET_NAME_LEN)
2325 		error = SET_ERROR(ENAMETOOLONG);
2326 
2327 	return (error);
2328 }
2329 
2330 static int
2331 dsl_dataset_rename_snapshot_check(void *arg, dmu_tx_t *tx)
2332 {
2333 	dsl_dataset_rename_snapshot_arg_t *ddrsa = arg;
2334 	dsl_pool_t *dp = dmu_tx_pool(tx);
2335 	dsl_dataset_t *hds;
2336 	int error;
2337 
2338 	error = dsl_dataset_hold(dp, ddrsa->ddrsa_fsname, FTAG, &hds);
2339 	if (error != 0)
2340 		return (error);
2341 
2342 	if (ddrsa->ddrsa_recursive) {
2343 		error = dmu_objset_find_dp(dp, hds->ds_dir->dd_object,
2344 		    dsl_dataset_rename_snapshot_check_impl, ddrsa,
2345 		    DS_FIND_CHILDREN);
2346 	} else {
2347 		error = dsl_dataset_rename_snapshot_check_impl(dp, hds, ddrsa);
2348 	}
2349 	dsl_dataset_rele(hds, FTAG);
2350 	return (error);
2351 }
2352 
2353 static int
2354 dsl_dataset_rename_snapshot_sync_impl(dsl_pool_t *dp,
2355     dsl_dataset_t *hds, void *arg)
2356 {
2357 	dsl_dataset_rename_snapshot_arg_t *ddrsa = arg;
2358 	dsl_dataset_t *ds;
2359 	uint64_t val;
2360 	dmu_tx_t *tx = ddrsa->ddrsa_tx;
2361 	int error;
2362 
2363 	error = dsl_dataset_snap_lookup(hds, ddrsa->ddrsa_oldsnapname, &val);
2364 	ASSERT(error == 0 || error == ENOENT);
2365 	if (error == ENOENT) {
2366 		/* ignore nonexistent snapshots */
2367 		return (0);
2368 	}
2369 
2370 	VERIFY0(dsl_dataset_hold_obj(dp, val, FTAG, &ds));
2371 
2372 	/* log before we change the name */
2373 	spa_history_log_internal_ds(ds, "rename", tx,
2374 	    "-> @%s", ddrsa->ddrsa_newsnapname);
2375 
2376 	VERIFY0(dsl_dataset_snap_remove(hds, ddrsa->ddrsa_oldsnapname, tx,
2377 	    B_FALSE));
2378 	mutex_enter(&ds->ds_lock);
2379 	(void) strcpy(ds->ds_snapname, ddrsa->ddrsa_newsnapname);
2380 	mutex_exit(&ds->ds_lock);
2381 	VERIFY0(zap_add(dp->dp_meta_objset,
2382 	    dsl_dataset_phys(hds)->ds_snapnames_zapobj,
2383 	    ds->ds_snapname, 8, 1, &ds->ds_object, tx));
2384 
2385 	dsl_dataset_rele(ds, FTAG);
2386 	return (0);
2387 }
2388 
2389 static void
2390 dsl_dataset_rename_snapshot_sync(void *arg, dmu_tx_t *tx)
2391 {
2392 	dsl_dataset_rename_snapshot_arg_t *ddrsa = arg;
2393 	dsl_pool_t *dp = dmu_tx_pool(tx);
2394 	dsl_dataset_t *hds;
2395 
2396 	VERIFY0(dsl_dataset_hold(dp, ddrsa->ddrsa_fsname, FTAG, &hds));
2397 	ddrsa->ddrsa_tx = tx;
2398 	if (ddrsa->ddrsa_recursive) {
2399 		VERIFY0(dmu_objset_find_dp(dp, hds->ds_dir->dd_object,
2400 		    dsl_dataset_rename_snapshot_sync_impl, ddrsa,
2401 		    DS_FIND_CHILDREN));
2402 	} else {
2403 		VERIFY0(dsl_dataset_rename_snapshot_sync_impl(dp, hds, ddrsa));
2404 	}
2405 	dsl_dataset_rele(hds, FTAG);
2406 }
2407 
2408 int
2409 dsl_dataset_rename_snapshot(const char *fsname,
2410     const char *oldsnapname, const char *newsnapname, boolean_t recursive)
2411 {
2412 	dsl_dataset_rename_snapshot_arg_t ddrsa;
2413 
2414 	ddrsa.ddrsa_fsname = fsname;
2415 	ddrsa.ddrsa_oldsnapname = oldsnapname;
2416 	ddrsa.ddrsa_newsnapname = newsnapname;
2417 	ddrsa.ddrsa_recursive = recursive;
2418 
2419 	return (dsl_sync_task(fsname, dsl_dataset_rename_snapshot_check,
2420 	    dsl_dataset_rename_snapshot_sync, &ddrsa,
2421 	    1, ZFS_SPACE_CHECK_RESERVED));
2422 }
2423 
2424 /*
2425  * If we're doing an ownership handoff, we need to make sure that there is
2426  * only one long hold on the dataset.  We're not allowed to change anything here
2427  * so we don't permanently release the long hold or regular hold here.  We want
2428  * to do this only when syncing to avoid the dataset unexpectedly going away
2429  * when we release the long hold.
2430  */
2431 static int
2432 dsl_dataset_handoff_check(dsl_dataset_t *ds, void *owner, dmu_tx_t *tx)
2433 {
2434 	boolean_t held;
2435 
2436 	if (!dmu_tx_is_syncing(tx))
2437 		return (0);
2438 
2439 	if (owner != NULL) {
2440 		VERIFY3P(ds->ds_owner, ==, owner);
2441 		dsl_dataset_long_rele(ds, owner);
2442 	}
2443 
2444 	held = dsl_dataset_long_held(ds);
2445 
2446 	if (owner != NULL)
2447 		dsl_dataset_long_hold(ds, owner);
2448 
2449 	if (held)
2450 		return (SET_ERROR(EBUSY));
2451 
2452 	return (0);
2453 }
2454 
2455 typedef struct dsl_dataset_rollback_arg {
2456 	const char *ddra_fsname;
2457 	const char *ddra_tosnap;
2458 	void *ddra_owner;
2459 	nvlist_t *ddra_result;
2460 } dsl_dataset_rollback_arg_t;
2461 
2462 static int
2463 dsl_dataset_rollback_check(void *arg, dmu_tx_t *tx)
2464 {
2465 	dsl_dataset_rollback_arg_t *ddra = arg;
2466 	dsl_pool_t *dp = dmu_tx_pool(tx);
2467 	dsl_dataset_t *ds;
2468 	int64_t unused_refres_delta;
2469 	int error;
2470 
2471 	error = dsl_dataset_hold(dp, ddra->ddra_fsname, FTAG, &ds);
2472 	if (error != 0)
2473 		return (error);
2474 
2475 	/* must not be a snapshot */
2476 	if (ds->ds_is_snapshot) {
2477 		dsl_dataset_rele(ds, FTAG);
2478 		return (SET_ERROR(EINVAL));
2479 	}
2480 
2481 	/* must have a most recent snapshot */
2482 	if (dsl_dataset_phys(ds)->ds_prev_snap_txg < TXG_INITIAL) {
2483 		dsl_dataset_rele(ds, FTAG);
2484 		return (SET_ERROR(EINVAL));
2485 	}
2486 
2487 	/*
2488 	 * No rollback to a snapshot created in the current txg, because
2489 	 * the rollback may dirty the dataset and create blocks that are
2490 	 * not reachable from the rootbp while having a birth txg that
2491 	 * falls into the snapshot's range.
2492 	 */
2493 	if (dmu_tx_is_syncing(tx) &&
2494 	    dsl_dataset_phys(ds)->ds_prev_snap_txg >= tx->tx_txg) {
2495 		dsl_dataset_rele(ds, FTAG);
2496 		return (SET_ERROR(EAGAIN));
2497 	}
2498 
2499 	/*
2500 	 * If the expected target snapshot is specified, then check that
2501 	 * the latest snapshot is it.
2502 	 */
2503 	if (ddra->ddra_tosnap != NULL) {
2504 		char namebuf[ZFS_MAX_DATASET_NAME_LEN];
2505 
2506 		dsl_dataset_name(ds->ds_prev, namebuf);
2507 		if (strcmp(namebuf, ddra->ddra_tosnap) != 0)
2508 			return (SET_ERROR(EXDEV));
2509 	}
2510 
2511 	/* must not have any bookmarks after the most recent snapshot */
2512 	nvlist_t *proprequest = fnvlist_alloc();
2513 	fnvlist_add_boolean(proprequest, zfs_prop_to_name(ZFS_PROP_CREATETXG));
2514 	nvlist_t *bookmarks = fnvlist_alloc();
2515 	error = dsl_get_bookmarks_impl(ds, proprequest, bookmarks);
2516 	fnvlist_free(proprequest);
2517 	if (error != 0)
2518 		return (error);
2519 	for (nvpair_t *pair = nvlist_next_nvpair(bookmarks, NULL);
2520 	    pair != NULL; pair = nvlist_next_nvpair(bookmarks, pair)) {
2521 		nvlist_t *valuenv =
2522 		    fnvlist_lookup_nvlist(fnvpair_value_nvlist(pair),
2523 		    zfs_prop_to_name(ZFS_PROP_CREATETXG));
2524 		uint64_t createtxg = fnvlist_lookup_uint64(valuenv, "value");
2525 		if (createtxg > dsl_dataset_phys(ds)->ds_prev_snap_txg) {
2526 			fnvlist_free(bookmarks);
2527 			dsl_dataset_rele(ds, FTAG);
2528 			return (SET_ERROR(EEXIST));
2529 		}
2530 	}
2531 	fnvlist_free(bookmarks);
2532 
2533 	error = dsl_dataset_handoff_check(ds, ddra->ddra_owner, tx);
2534 	if (error != 0) {
2535 		dsl_dataset_rele(ds, FTAG);
2536 		return (error);
2537 	}
2538 
2539 	/*
2540 	 * Check if the snap we are rolling back to uses more than
2541 	 * the refquota.
2542 	 */
2543 	if (ds->ds_quota != 0 &&
2544 	    dsl_dataset_phys(ds->ds_prev)->ds_referenced_bytes > ds->ds_quota) {
2545 		dsl_dataset_rele(ds, FTAG);
2546 		return (SET_ERROR(EDQUOT));
2547 	}
2548 
2549 	/*
2550 	 * When we do the clone swap, we will temporarily use more space
2551 	 * due to the refreservation (the head will no longer have any
2552 	 * unique space, so the entire amount of the refreservation will need
2553 	 * to be free).  We will immediately destroy the clone, freeing
2554 	 * this space, but the freeing happens over many txg's.
2555 	 */
2556 	unused_refres_delta = (int64_t)MIN(ds->ds_reserved,
2557 	    dsl_dataset_phys(ds)->ds_unique_bytes);
2558 
2559 	if (unused_refres_delta > 0 &&
2560 	    unused_refres_delta >
2561 	    dsl_dir_space_available(ds->ds_dir, NULL, 0, TRUE)) {
2562 		dsl_dataset_rele(ds, FTAG);
2563 		return (SET_ERROR(ENOSPC));
2564 	}
2565 
2566 	dsl_dataset_rele(ds, FTAG);
2567 	return (0);
2568 }
2569 
2570 static void
2571 dsl_dataset_rollback_sync(void *arg, dmu_tx_t *tx)
2572 {
2573 	dsl_dataset_rollback_arg_t *ddra = arg;
2574 	dsl_pool_t *dp = dmu_tx_pool(tx);
2575 	dsl_dataset_t *ds, *clone;
2576 	uint64_t cloneobj;
2577 	char namebuf[ZFS_MAX_DATASET_NAME_LEN];
2578 
2579 	VERIFY0(dsl_dataset_hold(dp, ddra->ddra_fsname, FTAG, &ds));
2580 
2581 	dsl_dataset_name(ds->ds_prev, namebuf);
2582 	fnvlist_add_string(ddra->ddra_result, "target", namebuf);
2583 
2584 	cloneobj = dsl_dataset_create_sync(ds->ds_dir, "%rollback",
2585 	    ds->ds_prev, DS_CREATE_FLAG_NODIRTY, kcred, tx);
2586 
2587 	VERIFY0(dsl_dataset_hold_obj(dp, cloneobj, FTAG, &clone));
2588 
2589 	dsl_dataset_clone_swap_sync_impl(clone, ds, tx);
2590 	dsl_dataset_zero_zil(ds, tx);
2591 
2592 	dsl_destroy_head_sync_impl(clone, tx);
2593 
2594 	dsl_dataset_rele(clone, FTAG);
2595 	dsl_dataset_rele(ds, FTAG);
2596 }
2597 
2598 /*
2599  * Rolls back the given filesystem or volume to the most recent snapshot.
2600  * The name of the most recent snapshot will be returned under key "target"
2601  * in the result nvlist.
2602  *
2603  * If owner != NULL:
2604  * - The existing dataset MUST be owned by the specified owner at entry
2605  * - Upon return, dataset will still be held by the same owner, whether we
2606  *   succeed or not.
2607  *
2608  * This mode is required any time the existing filesystem is mounted.  See
2609  * notes above zfs_suspend_fs() for further details.
2610  */
2611 int
2612 dsl_dataset_rollback(const char *fsname, const char *tosnap, void *owner,
2613     nvlist_t *result)
2614 {
2615 	dsl_dataset_rollback_arg_t ddra;
2616 
2617 	ddra.ddra_fsname = fsname;
2618 	ddra.ddra_tosnap = tosnap;
2619 	ddra.ddra_owner = owner;
2620 	ddra.ddra_result = result;
2621 
2622 	return (dsl_sync_task(fsname, dsl_dataset_rollback_check,
2623 	    dsl_dataset_rollback_sync, &ddra,
2624 	    1, ZFS_SPACE_CHECK_RESERVED));
2625 }
2626 
2627 struct promotenode {
2628 	list_node_t link;
2629 	dsl_dataset_t *ds;
2630 };
2631 
2632 static int snaplist_space(list_t *l, uint64_t mintxg, uint64_t *spacep);
2633 static int promote_hold(dsl_dataset_promote_arg_t *ddpa, dsl_pool_t *dp,
2634     void *tag);
2635 static void promote_rele(dsl_dataset_promote_arg_t *ddpa, void *tag);
2636 
2637 int
2638 dsl_dataset_promote_check(void *arg, dmu_tx_t *tx)
2639 {
2640 	dsl_dataset_promote_arg_t *ddpa = arg;
2641 	dsl_pool_t *dp = dmu_tx_pool(tx);
2642 	dsl_dataset_t *hds;
2643 	struct promotenode *snap;
2644 	dsl_dataset_t *origin_ds;
2645 	int err;
2646 	uint64_t unused;
2647 	uint64_t ss_mv_cnt;
2648 	size_t max_snap_len;
2649 	boolean_t conflicting_snaps;
2650 
2651 	err = promote_hold(ddpa, dp, FTAG);
2652 	if (err != 0)
2653 		return (err);
2654 
2655 	hds = ddpa->ddpa_clone;
2656 	snap = list_head(&ddpa->shared_snaps);
2657 	origin_ds = snap->ds;
2658 	max_snap_len = MAXNAMELEN - strlen(ddpa->ddpa_clonename) - 1;
2659 
2660 	snap = list_head(&ddpa->origin_snaps);
2661 
2662 	if (dsl_dataset_phys(hds)->ds_flags & DS_FLAG_NOPROMOTE) {
2663 		promote_rele(ddpa, FTAG);
2664 		return (SET_ERROR(EXDEV));
2665 	}
2666 
2667 	/*
2668 	 * Compute and check the amount of space to transfer.  Since this is
2669 	 * so expensive, don't do the preliminary check.
2670 	 */
2671 	if (!dmu_tx_is_syncing(tx)) {
2672 		promote_rele(ddpa, FTAG);
2673 		return (0);
2674 	}
2675 
2676 	/* compute origin's new unique space */
2677 	snap = list_tail(&ddpa->clone_snaps);
2678 	ASSERT3U(dsl_dataset_phys(snap->ds)->ds_prev_snap_obj, ==,
2679 	    origin_ds->ds_object);
2680 	dsl_deadlist_space_range(&snap->ds->ds_deadlist,
2681 	    dsl_dataset_phys(origin_ds)->ds_prev_snap_txg, UINT64_MAX,
2682 	    &ddpa->unique, &unused, &unused);
2683 
2684 	/*
2685 	 * Walk the snapshots that we are moving
2686 	 *
2687 	 * Compute space to transfer.  Consider the incremental changes
2688 	 * to used by each snapshot:
2689 	 * (my used) = (prev's used) + (blocks born) - (blocks killed)
2690 	 * So each snapshot gave birth to:
2691 	 * (blocks born) = (my used) - (prev's used) + (blocks killed)
2692 	 * So a sequence would look like:
2693 	 * (uN - u(N-1) + kN) + ... + (u1 - u0 + k1) + (u0 - 0 + k0)
2694 	 * Which simplifies to:
2695 	 * uN + kN + kN-1 + ... + k1 + k0
2696 	 * Note however, if we stop before we reach the ORIGIN we get:
2697 	 * uN + kN + kN-1 + ... + kM - uM-1
2698 	 */
2699 	conflicting_snaps = B_FALSE;
2700 	ss_mv_cnt = 0;
2701 	ddpa->used = dsl_dataset_phys(origin_ds)->ds_referenced_bytes;
2702 	ddpa->comp = dsl_dataset_phys(origin_ds)->ds_compressed_bytes;
2703 	ddpa->uncomp = dsl_dataset_phys(origin_ds)->ds_uncompressed_bytes;
2704 	for (snap = list_head(&ddpa->shared_snaps); snap;
2705 	    snap = list_next(&ddpa->shared_snaps, snap)) {
2706 		uint64_t val, dlused, dlcomp, dluncomp;
2707 		dsl_dataset_t *ds = snap->ds;
2708 
2709 		ss_mv_cnt++;
2710 
2711 		/*
2712 		 * If there are long holds, we won't be able to evict
2713 		 * the objset.
2714 		 */
2715 		if (dsl_dataset_long_held(ds)) {
2716 			err = SET_ERROR(EBUSY);
2717 			goto out;
2718 		}
2719 
2720 		/* Check that the snapshot name does not conflict */
2721 		VERIFY0(dsl_dataset_get_snapname(ds));
2722 		if (strlen(ds->ds_snapname) >= max_snap_len) {
2723 			err = SET_ERROR(ENAMETOOLONG);
2724 			goto out;
2725 		}
2726 		err = dsl_dataset_snap_lookup(hds, ds->ds_snapname, &val);
2727 		if (err == 0) {
2728 			fnvlist_add_boolean(ddpa->err_ds,
2729 			    snap->ds->ds_snapname);
2730 			conflicting_snaps = B_TRUE;
2731 		} else if (err != ENOENT) {
2732 			goto out;
2733 		}
2734 
2735 		/* The very first snapshot does not have a deadlist */
2736 		if (dsl_dataset_phys(ds)->ds_prev_snap_obj == 0)
2737 			continue;
2738 
2739 		dsl_deadlist_space(&ds->ds_deadlist,
2740 		    &dlused, &dlcomp, &dluncomp);
2741 		ddpa->used += dlused;
2742 		ddpa->comp += dlcomp;
2743 		ddpa->uncomp += dluncomp;
2744 	}
2745 
2746 	/*
2747 	 * In order to return the full list of conflicting snapshots, we check
2748 	 * whether there was a conflict after traversing all of them.
2749 	 */
2750 	if (conflicting_snaps) {
2751 		err = SET_ERROR(EEXIST);
2752 		goto out;
2753 	}
2754 
2755 	/*
2756 	 * If we are a clone of a clone then we never reached ORIGIN,
2757 	 * so we need to subtract out the clone origin's used space.
2758 	 */
2759 	if (ddpa->origin_origin) {
2760 		ddpa->used -=
2761 		    dsl_dataset_phys(ddpa->origin_origin)->ds_referenced_bytes;
2762 		ddpa->comp -=
2763 		    dsl_dataset_phys(ddpa->origin_origin)->ds_compressed_bytes;
2764 		ddpa->uncomp -=
2765 		    dsl_dataset_phys(ddpa->origin_origin)->
2766 		    ds_uncompressed_bytes;
2767 	}
2768 
2769 	/* Check that there is enough space and limit headroom here */
2770 	err = dsl_dir_transfer_possible(origin_ds->ds_dir, hds->ds_dir,
2771 	    0, ss_mv_cnt, ddpa->used, ddpa->cr);
2772 	if (err != 0)
2773 		goto out;
2774 
2775 	/*
2776 	 * Compute the amounts of space that will be used by snapshots
2777 	 * after the promotion (for both origin and clone).  For each,
2778 	 * it is the amount of space that will be on all of their
2779 	 * deadlists (that was not born before their new origin).
2780 	 */
2781 	if (dsl_dir_phys(hds->ds_dir)->dd_flags & DD_FLAG_USED_BREAKDOWN) {
2782 		uint64_t space;
2783 
2784 		/*
2785 		 * Note, typically this will not be a clone of a clone,
2786 		 * so dd_origin_txg will be < TXG_INITIAL, so
2787 		 * these snaplist_space() -> dsl_deadlist_space_range()
2788 		 * calls will be fast because they do not have to
2789 		 * iterate over all bps.
2790 		 */
2791 		snap = list_head(&ddpa->origin_snaps);
2792 		err = snaplist_space(&ddpa->shared_snaps,
2793 		    snap->ds->ds_dir->dd_origin_txg, &ddpa->cloneusedsnap);
2794 		if (err != 0)
2795 			goto out;
2796 
2797 		err = snaplist_space(&ddpa->clone_snaps,
2798 		    snap->ds->ds_dir->dd_origin_txg, &space);
2799 		if (err != 0)
2800 			goto out;
2801 		ddpa->cloneusedsnap += space;
2802 	}
2803 	if (dsl_dir_phys(origin_ds->ds_dir)->dd_flags &
2804 	    DD_FLAG_USED_BREAKDOWN) {
2805 		err = snaplist_space(&ddpa->origin_snaps,
2806 		    dsl_dataset_phys(origin_ds)->ds_creation_txg,
2807 		    &ddpa->originusedsnap);
2808 		if (err != 0)
2809 			goto out;
2810 	}
2811 
2812 out:
2813 	promote_rele(ddpa, FTAG);
2814 	return (err);
2815 }
2816 
2817 void
2818 dsl_dataset_promote_sync(void *arg, dmu_tx_t *tx)
2819 {
2820 	dsl_dataset_promote_arg_t *ddpa = arg;
2821 	dsl_pool_t *dp = dmu_tx_pool(tx);
2822 	dsl_dataset_t *hds;
2823 	struct promotenode *snap;
2824 	dsl_dataset_t *origin_ds;
2825 	dsl_dataset_t *origin_head;
2826 	dsl_dir_t *dd;
2827 	dsl_dir_t *odd = NULL;
2828 	uint64_t oldnext_obj;
2829 	int64_t delta;
2830 
2831 	VERIFY0(promote_hold(ddpa, dp, FTAG));
2832 	hds = ddpa->ddpa_clone;
2833 
2834 	ASSERT0(dsl_dataset_phys(hds)->ds_flags & DS_FLAG_NOPROMOTE);
2835 
2836 	snap = list_head(&ddpa->shared_snaps);
2837 	origin_ds = snap->ds;
2838 	dd = hds->ds_dir;
2839 
2840 	snap = list_head(&ddpa->origin_snaps);
2841 	origin_head = snap->ds;
2842 
2843 	/*
2844 	 * We need to explicitly open odd, since origin_ds's dd will be
2845 	 * changing.
2846 	 */
2847 	VERIFY0(dsl_dir_hold_obj(dp, origin_ds->ds_dir->dd_object,
2848 	    NULL, FTAG, &odd));
2849 
2850 	/* change origin's next snap */
2851 	dmu_buf_will_dirty(origin_ds->ds_dbuf, tx);
2852 	oldnext_obj = dsl_dataset_phys(origin_ds)->ds_next_snap_obj;
2853 	snap = list_tail(&ddpa->clone_snaps);
2854 	ASSERT3U(dsl_dataset_phys(snap->ds)->ds_prev_snap_obj, ==,
2855 	    origin_ds->ds_object);
2856 	dsl_dataset_phys(origin_ds)->ds_next_snap_obj = snap->ds->ds_object;
2857 
2858 	/* change the origin's next clone */
2859 	if (dsl_dataset_phys(origin_ds)->ds_next_clones_obj) {
2860 		dsl_dataset_remove_from_next_clones(origin_ds,
2861 		    snap->ds->ds_object, tx);
2862 		VERIFY0(zap_add_int(dp->dp_meta_objset,
2863 		    dsl_dataset_phys(origin_ds)->ds_next_clones_obj,
2864 		    oldnext_obj, tx));
2865 	}
2866 
2867 	/* change origin */
2868 	dmu_buf_will_dirty(dd->dd_dbuf, tx);
2869 	ASSERT3U(dsl_dir_phys(dd)->dd_origin_obj, ==, origin_ds->ds_object);
2870 	dsl_dir_phys(dd)->dd_origin_obj = dsl_dir_phys(odd)->dd_origin_obj;
2871 	dd->dd_origin_txg = origin_head->ds_dir->dd_origin_txg;
2872 	dmu_buf_will_dirty(odd->dd_dbuf, tx);
2873 	dsl_dir_phys(odd)->dd_origin_obj = origin_ds->ds_object;
2874 	origin_head->ds_dir->dd_origin_txg =
2875 	    dsl_dataset_phys(origin_ds)->ds_creation_txg;
2876 
2877 	/* change dd_clone entries */
2878 	if (spa_version(dp->dp_spa) >= SPA_VERSION_DIR_CLONES) {
2879 		VERIFY0(zap_remove_int(dp->dp_meta_objset,
2880 		    dsl_dir_phys(odd)->dd_clones, hds->ds_object, tx));
2881 		VERIFY0(zap_add_int(dp->dp_meta_objset,
2882 		    dsl_dir_phys(ddpa->origin_origin->ds_dir)->dd_clones,
2883 		    hds->ds_object, tx));
2884 
2885 		VERIFY0(zap_remove_int(dp->dp_meta_objset,
2886 		    dsl_dir_phys(ddpa->origin_origin->ds_dir)->dd_clones,
2887 		    origin_head->ds_object, tx));
2888 		if (dsl_dir_phys(dd)->dd_clones == 0) {
2889 			dsl_dir_phys(dd)->dd_clones =
2890 			    zap_create(dp->dp_meta_objset, DMU_OT_DSL_CLONES,
2891 			    DMU_OT_NONE, 0, tx);
2892 		}
2893 		VERIFY0(zap_add_int(dp->dp_meta_objset,
2894 		    dsl_dir_phys(dd)->dd_clones, origin_head->ds_object, tx));
2895 	}
2896 
2897 	/* move snapshots to this dir */
2898 	for (snap = list_head(&ddpa->shared_snaps); snap;
2899 	    snap = list_next(&ddpa->shared_snaps, snap)) {
2900 		dsl_dataset_t *ds = snap->ds;
2901 
2902 		/*
2903 		 * Property callbacks are registered to a particular
2904 		 * dsl_dir.  Since ours is changing, evict the objset
2905 		 * so that they will be unregistered from the old dsl_dir.
2906 		 */
2907 		if (ds->ds_objset) {
2908 			dmu_objset_evict(ds->ds_objset);
2909 			ds->ds_objset = NULL;
2910 		}
2911 
2912 		/* move snap name entry */
2913 		VERIFY0(dsl_dataset_get_snapname(ds));
2914 		VERIFY0(dsl_dataset_snap_remove(origin_head,
2915 		    ds->ds_snapname, tx, B_TRUE));
2916 		VERIFY0(zap_add(dp->dp_meta_objset,
2917 		    dsl_dataset_phys(hds)->ds_snapnames_zapobj, ds->ds_snapname,
2918 		    8, 1, &ds->ds_object, tx));
2919 		dsl_fs_ss_count_adjust(hds->ds_dir, 1,
2920 		    DD_FIELD_SNAPSHOT_COUNT, tx);
2921 
2922 		/* change containing dsl_dir */
2923 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
2924 		ASSERT3U(dsl_dataset_phys(ds)->ds_dir_obj, ==, odd->dd_object);
2925 		dsl_dataset_phys(ds)->ds_dir_obj = dd->dd_object;
2926 		ASSERT3P(ds->ds_dir, ==, odd);
2927 		dsl_dir_rele(ds->ds_dir, ds);
2928 		VERIFY0(dsl_dir_hold_obj(dp, dd->dd_object,
2929 		    NULL, ds, &ds->ds_dir));
2930 
2931 		/* move any clone references */
2932 		if (dsl_dataset_phys(ds)->ds_next_clones_obj &&
2933 		    spa_version(dp->dp_spa) >= SPA_VERSION_DIR_CLONES) {
2934 			zap_cursor_t zc;
2935 			zap_attribute_t za;
2936 
2937 			for (zap_cursor_init(&zc, dp->dp_meta_objset,
2938 			    dsl_dataset_phys(ds)->ds_next_clones_obj);
2939 			    zap_cursor_retrieve(&zc, &za) == 0;
2940 			    zap_cursor_advance(&zc)) {
2941 				dsl_dataset_t *cnds;
2942 				uint64_t o;
2943 
2944 				if (za.za_first_integer == oldnext_obj) {
2945 					/*
2946 					 * We've already moved the
2947 					 * origin's reference.
2948 					 */
2949 					continue;
2950 				}
2951 
2952 				VERIFY0(dsl_dataset_hold_obj(dp,
2953 				    za.za_first_integer, FTAG, &cnds));
2954 				o = dsl_dir_phys(cnds->ds_dir)->
2955 				    dd_head_dataset_obj;
2956 
2957 				VERIFY0(zap_remove_int(dp->dp_meta_objset,
2958 				    dsl_dir_phys(odd)->dd_clones, o, tx));
2959 				VERIFY0(zap_add_int(dp->dp_meta_objset,
2960 				    dsl_dir_phys(dd)->dd_clones, o, tx));
2961 				dsl_dataset_rele(cnds, FTAG);
2962 			}
2963 			zap_cursor_fini(&zc);
2964 		}
2965 
2966 		ASSERT(!dsl_prop_hascb(ds));
2967 	}
2968 
2969 	/*
2970 	 * Change space accounting.
2971 	 * Note, pa->*usedsnap and dd_used_breakdown[SNAP] will either
2972 	 * both be valid, or both be 0 (resulting in delta == 0).  This
2973 	 * is true for each of {clone,origin} independently.
2974 	 */
2975 
2976 	delta = ddpa->cloneusedsnap -
2977 	    dsl_dir_phys(dd)->dd_used_breakdown[DD_USED_SNAP];
2978 	ASSERT3S(delta, >=, 0);
2979 	ASSERT3U(ddpa->used, >=, delta);
2980 	dsl_dir_diduse_space(dd, DD_USED_SNAP, delta, 0, 0, tx);
2981 	dsl_dir_diduse_space(dd, DD_USED_HEAD,
2982 	    ddpa->used - delta, ddpa->comp, ddpa->uncomp, tx);
2983 
2984 	delta = ddpa->originusedsnap -
2985 	    dsl_dir_phys(odd)->dd_used_breakdown[DD_USED_SNAP];
2986 	ASSERT3S(delta, <=, 0);
2987 	ASSERT3U(ddpa->used, >=, -delta);
2988 	dsl_dir_diduse_space(odd, DD_USED_SNAP, delta, 0, 0, tx);
2989 	dsl_dir_diduse_space(odd, DD_USED_HEAD,
2990 	    -ddpa->used - delta, -ddpa->comp, -ddpa->uncomp, tx);
2991 
2992 	dsl_dataset_phys(origin_ds)->ds_unique_bytes = ddpa->unique;
2993 
2994 	/* log history record */
2995 	spa_history_log_internal_ds(hds, "promote", tx, "");
2996 
2997 	dsl_dir_rele(odd, FTAG);
2998 	promote_rele(ddpa, FTAG);
2999 }
3000 
3001 /*
3002  * Make a list of dsl_dataset_t's for the snapshots between first_obj
3003  * (exclusive) and last_obj (inclusive).  The list will be in reverse
3004  * order (last_obj will be the list_head()).  If first_obj == 0, do all
3005  * snapshots back to this dataset's origin.
3006  */
3007 static int
3008 snaplist_make(dsl_pool_t *dp,
3009     uint64_t first_obj, uint64_t last_obj, list_t *l, void *tag)
3010 {
3011 	uint64_t obj = last_obj;
3012 
3013 	list_create(l, sizeof (struct promotenode),
3014 	    offsetof(struct promotenode, link));
3015 
3016 	while (obj != first_obj) {
3017 		dsl_dataset_t *ds;
3018 		struct promotenode *snap;
3019 		int err;
3020 
3021 		err = dsl_dataset_hold_obj(dp, obj, tag, &ds);
3022 		ASSERT(err != ENOENT);
3023 		if (err != 0)
3024 			return (err);
3025 
3026 		if (first_obj == 0)
3027 			first_obj = dsl_dir_phys(ds->ds_dir)->dd_origin_obj;
3028 
3029 		snap = kmem_alloc(sizeof (*snap), KM_SLEEP);
3030 		snap->ds = ds;
3031 		list_insert_tail(l, snap);
3032 		obj = dsl_dataset_phys(ds)->ds_prev_snap_obj;
3033 	}
3034 
3035 	return (0);
3036 }
3037 
3038 static int
3039 snaplist_space(list_t *l, uint64_t mintxg, uint64_t *spacep)
3040 {
3041 	struct promotenode *snap;
3042 
3043 	*spacep = 0;
3044 	for (snap = list_head(l); snap; snap = list_next(l, snap)) {
3045 		uint64_t used, comp, uncomp;
3046 		dsl_deadlist_space_range(&snap->ds->ds_deadlist,
3047 		    mintxg, UINT64_MAX, &used, &comp, &uncomp);
3048 		*spacep += used;
3049 	}
3050 	return (0);
3051 }
3052 
3053 static void
3054 snaplist_destroy(list_t *l, void *tag)
3055 {
3056 	struct promotenode *snap;
3057 
3058 	if (l == NULL || !list_link_active(&l->list_head))
3059 		return;
3060 
3061 	while ((snap = list_tail(l)) != NULL) {
3062 		list_remove(l, snap);
3063 		dsl_dataset_rele(snap->ds, tag);
3064 		kmem_free(snap, sizeof (*snap));
3065 	}
3066 	list_destroy(l);
3067 }
3068 
3069 static int
3070 promote_hold(dsl_dataset_promote_arg_t *ddpa, dsl_pool_t *dp, void *tag)
3071 {
3072 	int error;
3073 	dsl_dir_t *dd;
3074 	struct promotenode *snap;
3075 
3076 	error = dsl_dataset_hold(dp, ddpa->ddpa_clonename, tag,
3077 	    &ddpa->ddpa_clone);
3078 	if (error != 0)
3079 		return (error);
3080 	dd = ddpa->ddpa_clone->ds_dir;
3081 
3082 	if (ddpa->ddpa_clone->ds_is_snapshot ||
3083 	    !dsl_dir_is_clone(dd)) {
3084 		dsl_dataset_rele(ddpa->ddpa_clone, tag);
3085 		return (SET_ERROR(EINVAL));
3086 	}
3087 
3088 	error = snaplist_make(dp, 0, dsl_dir_phys(dd)->dd_origin_obj,
3089 	    &ddpa->shared_snaps, tag);
3090 	if (error != 0)
3091 		goto out;
3092 
3093 	error = snaplist_make(dp, 0, ddpa->ddpa_clone->ds_object,
3094 	    &ddpa->clone_snaps, tag);
3095 	if (error != 0)
3096 		goto out;
3097 
3098 	snap = list_head(&ddpa->shared_snaps);
3099 	ASSERT3U(snap->ds->ds_object, ==, dsl_dir_phys(dd)->dd_origin_obj);
3100 	error = snaplist_make(dp, dsl_dir_phys(dd)->dd_origin_obj,
3101 	    dsl_dir_phys(snap->ds->ds_dir)->dd_head_dataset_obj,
3102 	    &ddpa->origin_snaps, tag);
3103 	if (error != 0)
3104 		goto out;
3105 
3106 	if (dsl_dir_phys(snap->ds->ds_dir)->dd_origin_obj != 0) {
3107 		error = dsl_dataset_hold_obj(dp,
3108 		    dsl_dir_phys(snap->ds->ds_dir)->dd_origin_obj,
3109 		    tag, &ddpa->origin_origin);
3110 		if (error != 0)
3111 			goto out;
3112 	}
3113 out:
3114 	if (error != 0)
3115 		promote_rele(ddpa, tag);
3116 	return (error);
3117 }
3118 
3119 static void
3120 promote_rele(dsl_dataset_promote_arg_t *ddpa, void *tag)
3121 {
3122 	snaplist_destroy(&ddpa->shared_snaps, tag);
3123 	snaplist_destroy(&ddpa->clone_snaps, tag);
3124 	snaplist_destroy(&ddpa->origin_snaps, tag);
3125 	if (ddpa->origin_origin != NULL)
3126 		dsl_dataset_rele(ddpa->origin_origin, tag);
3127 	dsl_dataset_rele(ddpa->ddpa_clone, tag);
3128 }
3129 
3130 /*
3131  * Promote a clone.
3132  *
3133  * If it fails due to a conflicting snapshot name, "conflsnap" will be filled
3134  * in with the name.  (It must be at least ZFS_MAX_DATASET_NAME_LEN bytes long.)
3135  */
3136 int
3137 dsl_dataset_promote(const char *name, char *conflsnap)
3138 {
3139 	dsl_dataset_promote_arg_t ddpa = { 0 };
3140 	uint64_t numsnaps;
3141 	int error;
3142 	nvpair_t *snap_pair;
3143 	objset_t *os;
3144 
3145 	/*
3146 	 * We will modify space proportional to the number of
3147 	 * snapshots.  Compute numsnaps.
3148 	 */
3149 	error = dmu_objset_hold(name, FTAG, &os);
3150 	if (error != 0)
3151 		return (error);
3152 	error = zap_count(dmu_objset_pool(os)->dp_meta_objset,
3153 	    dsl_dataset_phys(dmu_objset_ds(os))->ds_snapnames_zapobj,
3154 	    &numsnaps);
3155 	dmu_objset_rele(os, FTAG);
3156 	if (error != 0)
3157 		return (error);
3158 
3159 	ddpa.ddpa_clonename = name;
3160 	ddpa.err_ds = fnvlist_alloc();
3161 	ddpa.cr = CRED();
3162 
3163 	error = dsl_sync_task(name, dsl_dataset_promote_check,
3164 	    dsl_dataset_promote_sync, &ddpa,
3165 	    2 + numsnaps, ZFS_SPACE_CHECK_RESERVED);
3166 
3167 	/*
3168 	 * Return the first conflicting snapshot found.
3169 	 */
3170 	snap_pair = nvlist_next_nvpair(ddpa.err_ds, NULL);
3171 	if (snap_pair != NULL && conflsnap != NULL)
3172 		(void) strcpy(conflsnap, nvpair_name(snap_pair));
3173 
3174 	fnvlist_free(ddpa.err_ds);
3175 	return (error);
3176 }
3177 
3178 int
3179 dsl_dataset_clone_swap_check_impl(dsl_dataset_t *clone,
3180     dsl_dataset_t *origin_head, boolean_t force, void *owner, dmu_tx_t *tx)
3181 {
3182 	/*
3183 	 * "slack" factor for received datasets with refquota set on them.
3184 	 * See the bottom of this function for details on its use.
3185 	 */
3186 	uint64_t refquota_slack = DMU_MAX_ACCESS * spa_asize_inflation;
3187 	int64_t unused_refres_delta;
3188 
3189 	/* they should both be heads */
3190 	if (clone->ds_is_snapshot ||
3191 	    origin_head->ds_is_snapshot)
3192 		return (SET_ERROR(EINVAL));
3193 
3194 	/* if we are not forcing, the branch point should be just before them */
3195 	if (!force && clone->ds_prev != origin_head->ds_prev)
3196 		return (SET_ERROR(EINVAL));
3197 
3198 	/* clone should be the clone (unless they are unrelated) */
3199 	if (clone->ds_prev != NULL &&
3200 	    clone->ds_prev != clone->ds_dir->dd_pool->dp_origin_snap &&
3201 	    origin_head->ds_dir != clone->ds_prev->ds_dir)
3202 		return (SET_ERROR(EINVAL));
3203 
3204 	/* the clone should be a child of the origin */
3205 	if (clone->ds_dir->dd_parent != origin_head->ds_dir)
3206 		return (SET_ERROR(EINVAL));
3207 
3208 	/* origin_head shouldn't be modified unless 'force' */
3209 	if (!force &&
3210 	    dsl_dataset_modified_since_snap(origin_head, origin_head->ds_prev))
3211 		return (SET_ERROR(ETXTBSY));
3212 
3213 	/* origin_head should have no long holds (e.g. is not mounted) */
3214 	if (dsl_dataset_handoff_check(origin_head, owner, tx))
3215 		return (SET_ERROR(EBUSY));
3216 
3217 	/* check amount of any unconsumed refreservation */
3218 	unused_refres_delta =
3219 	    (int64_t)MIN(origin_head->ds_reserved,
3220 	    dsl_dataset_phys(origin_head)->ds_unique_bytes) -
3221 	    (int64_t)MIN(origin_head->ds_reserved,
3222 	    dsl_dataset_phys(clone)->ds_unique_bytes);
3223 
3224 	if (unused_refres_delta > 0 &&
3225 	    unused_refres_delta >
3226 	    dsl_dir_space_available(origin_head->ds_dir, NULL, 0, TRUE))
3227 		return (SET_ERROR(ENOSPC));
3228 
3229 	/*
3230 	 * The clone can't be too much over the head's refquota.
3231 	 *
3232 	 * To ensure that the entire refquota can be used, we allow one
3233 	 * transaction to exceed the the refquota.  Therefore, this check
3234 	 * needs to also allow for the space referenced to be more than the
3235 	 * refquota.  The maximum amount of space that one transaction can use
3236 	 * on disk is DMU_MAX_ACCESS * spa_asize_inflation.  Allowing this
3237 	 * overage ensures that we are able to receive a filesystem that
3238 	 * exceeds the refquota on the source system.
3239 	 *
3240 	 * So that overage is the refquota_slack we use below.
3241 	 */
3242 	if (origin_head->ds_quota != 0 &&
3243 	    dsl_dataset_phys(clone)->ds_referenced_bytes >
3244 	    origin_head->ds_quota + refquota_slack)
3245 		return (SET_ERROR(EDQUOT));
3246 
3247 	return (0);
3248 }
3249 
3250 void
3251 dsl_dataset_clone_swap_sync_impl(dsl_dataset_t *clone,
3252     dsl_dataset_t *origin_head, dmu_tx_t *tx)
3253 {
3254 	dsl_pool_t *dp = dmu_tx_pool(tx);
3255 	int64_t unused_refres_delta;
3256 
3257 	ASSERT(clone->ds_reserved == 0);
3258 	/*
3259 	 * NOTE: On DEBUG kernels there could be a race between this and
3260 	 * the check function if spa_asize_inflation is adjusted...
3261 	 */
3262 	ASSERT(origin_head->ds_quota == 0 ||
3263 	    dsl_dataset_phys(clone)->ds_unique_bytes <= origin_head->ds_quota +
3264 	    DMU_MAX_ACCESS * spa_asize_inflation);
3265 	ASSERT3P(clone->ds_prev, ==, origin_head->ds_prev);
3266 
3267 	/*
3268 	 * Swap per-dataset feature flags.
3269 	 */
3270 	for (spa_feature_t f = 0; f < SPA_FEATURES; f++) {
3271 		if (!(spa_feature_table[f].fi_flags &
3272 		    ZFEATURE_FLAG_PER_DATASET)) {
3273 			ASSERT(!clone->ds_feature_inuse[f]);
3274 			ASSERT(!origin_head->ds_feature_inuse[f]);
3275 			continue;
3276 		}
3277 
3278 		boolean_t clone_inuse = clone->ds_feature_inuse[f];
3279 		boolean_t origin_head_inuse = origin_head->ds_feature_inuse[f];
3280 
3281 		if (clone_inuse) {
3282 			dsl_dataset_deactivate_feature(clone->ds_object, f, tx);
3283 			clone->ds_feature_inuse[f] = B_FALSE;
3284 		}
3285 		if (origin_head_inuse) {
3286 			dsl_dataset_deactivate_feature(origin_head->ds_object,
3287 			    f, tx);
3288 			origin_head->ds_feature_inuse[f] = B_FALSE;
3289 		}
3290 		if (clone_inuse) {
3291 			dsl_dataset_activate_feature(origin_head->ds_object,
3292 			    f, tx);
3293 			origin_head->ds_feature_inuse[f] = B_TRUE;
3294 		}
3295 		if (origin_head_inuse) {
3296 			dsl_dataset_activate_feature(clone->ds_object, f, tx);
3297 			clone->ds_feature_inuse[f] = B_TRUE;
3298 		}
3299 	}
3300 
3301 	dmu_buf_will_dirty(clone->ds_dbuf, tx);
3302 	dmu_buf_will_dirty(origin_head->ds_dbuf, tx);
3303 
3304 	if (clone->ds_objset != NULL) {
3305 		dmu_objset_evict(clone->ds_objset);
3306 		clone->ds_objset = NULL;
3307 	}
3308 
3309 	if (origin_head->ds_objset != NULL) {
3310 		dmu_objset_evict(origin_head->ds_objset);
3311 		origin_head->ds_objset = NULL;
3312 	}
3313 
3314 	unused_refres_delta =
3315 	    (int64_t)MIN(origin_head->ds_reserved,
3316 	    dsl_dataset_phys(origin_head)->ds_unique_bytes) -
3317 	    (int64_t)MIN(origin_head->ds_reserved,
3318 	    dsl_dataset_phys(clone)->ds_unique_bytes);
3319 
3320 	/*
3321 	 * Reset origin's unique bytes, if it exists.
3322 	 */
3323 	if (clone->ds_prev) {
3324 		dsl_dataset_t *origin = clone->ds_prev;
3325 		uint64_t comp, uncomp;
3326 
3327 		dmu_buf_will_dirty(origin->ds_dbuf, tx);
3328 		dsl_deadlist_space_range(&clone->ds_deadlist,
3329 		    dsl_dataset_phys(origin)->ds_prev_snap_txg, UINT64_MAX,
3330 		    &dsl_dataset_phys(origin)->ds_unique_bytes, &comp, &uncomp);
3331 	}
3332 
3333 	/* swap blkptrs */
3334 	{
3335 		rrw_enter(&clone->ds_bp_rwlock, RW_WRITER, FTAG);
3336 		rrw_enter(&origin_head->ds_bp_rwlock, RW_WRITER, FTAG);
3337 		blkptr_t tmp;
3338 		tmp = dsl_dataset_phys(origin_head)->ds_bp;
3339 		dsl_dataset_phys(origin_head)->ds_bp =
3340 		    dsl_dataset_phys(clone)->ds_bp;
3341 		dsl_dataset_phys(clone)->ds_bp = tmp;
3342 		rrw_exit(&origin_head->ds_bp_rwlock, FTAG);
3343 		rrw_exit(&clone->ds_bp_rwlock, FTAG);
3344 	}
3345 
3346 	/* set dd_*_bytes */
3347 	{
3348 		int64_t dused, dcomp, duncomp;
3349 		uint64_t cdl_used, cdl_comp, cdl_uncomp;
3350 		uint64_t odl_used, odl_comp, odl_uncomp;
3351 
3352 		ASSERT3U(dsl_dir_phys(clone->ds_dir)->
3353 		    dd_used_breakdown[DD_USED_SNAP], ==, 0);
3354 
3355 		dsl_deadlist_space(&clone->ds_deadlist,
3356 		    &cdl_used, &cdl_comp, &cdl_uncomp);
3357 		dsl_deadlist_space(&origin_head->ds_deadlist,
3358 		    &odl_used, &odl_comp, &odl_uncomp);
3359 
3360 		dused = dsl_dataset_phys(clone)->ds_referenced_bytes +
3361 		    cdl_used -
3362 		    (dsl_dataset_phys(origin_head)->ds_referenced_bytes +
3363 		    odl_used);
3364 		dcomp = dsl_dataset_phys(clone)->ds_compressed_bytes +
3365 		    cdl_comp -
3366 		    (dsl_dataset_phys(origin_head)->ds_compressed_bytes +
3367 		    odl_comp);
3368 		duncomp = dsl_dataset_phys(clone)->ds_uncompressed_bytes +
3369 		    cdl_uncomp -
3370 		    (dsl_dataset_phys(origin_head)->ds_uncompressed_bytes +
3371 		    odl_uncomp);
3372 
3373 		dsl_dir_diduse_space(origin_head->ds_dir, DD_USED_HEAD,
3374 		    dused, dcomp, duncomp, tx);
3375 		dsl_dir_diduse_space(clone->ds_dir, DD_USED_HEAD,
3376 		    -dused, -dcomp, -duncomp, tx);
3377 
3378 		/*
3379 		 * The difference in the space used by snapshots is the
3380 		 * difference in snapshot space due to the head's
3381 		 * deadlist (since that's the only thing that's
3382 		 * changing that affects the snapused).
3383 		 */
3384 		dsl_deadlist_space_range(&clone->ds_deadlist,
3385 		    origin_head->ds_dir->dd_origin_txg, UINT64_MAX,
3386 		    &cdl_used, &cdl_comp, &cdl_uncomp);
3387 		dsl_deadlist_space_range(&origin_head->ds_deadlist,
3388 		    origin_head->ds_dir->dd_origin_txg, UINT64_MAX,
3389 		    &odl_used, &odl_comp, &odl_uncomp);
3390 		dsl_dir_transfer_space(origin_head->ds_dir, cdl_used - odl_used,
3391 		    DD_USED_HEAD, DD_USED_SNAP, tx);
3392 	}
3393 
3394 	/* swap ds_*_bytes */
3395 	SWITCH64(dsl_dataset_phys(origin_head)->ds_referenced_bytes,
3396 	    dsl_dataset_phys(clone)->ds_referenced_bytes);
3397 	SWITCH64(dsl_dataset_phys(origin_head)->ds_compressed_bytes,
3398 	    dsl_dataset_phys(clone)->ds_compressed_bytes);
3399 	SWITCH64(dsl_dataset_phys(origin_head)->ds_uncompressed_bytes,
3400 	    dsl_dataset_phys(clone)->ds_uncompressed_bytes);
3401 	SWITCH64(dsl_dataset_phys(origin_head)->ds_unique_bytes,
3402 	    dsl_dataset_phys(clone)->ds_unique_bytes);
3403 
3404 	/* apply any parent delta for change in unconsumed refreservation */
3405 	dsl_dir_diduse_space(origin_head->ds_dir, DD_USED_REFRSRV,
3406 	    unused_refres_delta, 0, 0, tx);
3407 
3408 	/*
3409 	 * Swap deadlists.
3410 	 */
3411 	dsl_deadlist_close(&clone->ds_deadlist);
3412 	dsl_deadlist_close(&origin_head->ds_deadlist);
3413 	SWITCH64(dsl_dataset_phys(origin_head)->ds_deadlist_obj,
3414 	    dsl_dataset_phys(clone)->ds_deadlist_obj);
3415 	dsl_deadlist_open(&clone->ds_deadlist, dp->dp_meta_objset,
3416 	    dsl_dataset_phys(clone)->ds_deadlist_obj);
3417 	dsl_deadlist_open(&origin_head->ds_deadlist, dp->dp_meta_objset,
3418 	    dsl_dataset_phys(origin_head)->ds_deadlist_obj);
3419 
3420 	dsl_scan_ds_clone_swapped(origin_head, clone, tx);
3421 
3422 	spa_history_log_internal_ds(clone, "clone swap", tx,
3423 	    "parent=%s", origin_head->ds_dir->dd_myname);
3424 }
3425 
3426 /*
3427  * Given a pool name and a dataset object number in that pool,
3428  * return the name of that dataset.
3429  */
3430 int
3431 dsl_dsobj_to_dsname(char *pname, uint64_t obj, char *buf)
3432 {
3433 	dsl_pool_t *dp;
3434 	dsl_dataset_t *ds;
3435 	int error;
3436 
3437 	error = dsl_pool_hold(pname, FTAG, &dp);
3438 	if (error != 0)
3439 		return (error);
3440 
3441 	error = dsl_dataset_hold_obj(dp, obj, FTAG, &ds);
3442 	if (error == 0) {
3443 		dsl_dataset_name(ds, buf);
3444 		dsl_dataset_rele(ds, FTAG);
3445 	}
3446 	dsl_pool_rele(dp, FTAG);
3447 
3448 	return (error);
3449 }
3450 
3451 int
3452 dsl_dataset_check_quota(dsl_dataset_t *ds, boolean_t check_quota,
3453     uint64_t asize, uint64_t inflight, uint64_t *used, uint64_t *ref_rsrv)
3454 {
3455 	int error = 0;
3456 
3457 	ASSERT3S(asize, >, 0);
3458 
3459 	/*
3460 	 * *ref_rsrv is the portion of asize that will come from any
3461 	 * unconsumed refreservation space.
3462 	 */
3463 	*ref_rsrv = 0;
3464 
3465 	mutex_enter(&ds->ds_lock);
3466 	/*
3467 	 * Make a space adjustment for reserved bytes.
3468 	 */
3469 	if (ds->ds_reserved > dsl_dataset_phys(ds)->ds_unique_bytes) {
3470 		ASSERT3U(*used, >=,
3471 		    ds->ds_reserved - dsl_dataset_phys(ds)->ds_unique_bytes);
3472 		*used -=
3473 		    (ds->ds_reserved - dsl_dataset_phys(ds)->ds_unique_bytes);
3474 		*ref_rsrv =
3475 		    asize - MIN(asize, parent_delta(ds, asize + inflight));
3476 	}
3477 
3478 	if (!check_quota || ds->ds_quota == 0) {
3479 		mutex_exit(&ds->ds_lock);
3480 		return (0);
3481 	}
3482 	/*
3483 	 * If they are requesting more space, and our current estimate
3484 	 * is over quota, they get to try again unless the actual
3485 	 * on-disk is over quota and there are no pending changes (which
3486 	 * may free up space for us).
3487 	 */
3488 	if (dsl_dataset_phys(ds)->ds_referenced_bytes + inflight >=
3489 	    ds->ds_quota) {
3490 		if (inflight > 0 ||
3491 		    dsl_dataset_phys(ds)->ds_referenced_bytes < ds->ds_quota)
3492 			error = SET_ERROR(ERESTART);
3493 		else
3494 			error = SET_ERROR(EDQUOT);
3495 	}
3496 	mutex_exit(&ds->ds_lock);
3497 
3498 	return (error);
3499 }
3500 
3501 typedef struct dsl_dataset_set_qr_arg {
3502 	const char *ddsqra_name;
3503 	zprop_source_t ddsqra_source;
3504 	uint64_t ddsqra_value;
3505 } dsl_dataset_set_qr_arg_t;
3506 
3507 
3508 /* ARGSUSED */
3509 static int
3510 dsl_dataset_set_refquota_check(void *arg, dmu_tx_t *tx)
3511 {
3512 	dsl_dataset_set_qr_arg_t *ddsqra = arg;
3513 	dsl_pool_t *dp = dmu_tx_pool(tx);
3514 	dsl_dataset_t *ds;
3515 	int error;
3516 	uint64_t newval;
3517 
3518 	if (spa_version(dp->dp_spa) < SPA_VERSION_REFQUOTA)
3519 		return (SET_ERROR(ENOTSUP));
3520 
3521 	error = dsl_dataset_hold(dp, ddsqra->ddsqra_name, FTAG, &ds);
3522 	if (error != 0)
3523 		return (error);
3524 
3525 	if (ds->ds_is_snapshot) {
3526 		dsl_dataset_rele(ds, FTAG);
3527 		return (SET_ERROR(EINVAL));
3528 	}
3529 
3530 	error = dsl_prop_predict(ds->ds_dir,
3531 	    zfs_prop_to_name(ZFS_PROP_REFQUOTA),
3532 	    ddsqra->ddsqra_source, ddsqra->ddsqra_value, &newval);
3533 	if (error != 0) {
3534 		dsl_dataset_rele(ds, FTAG);
3535 		return (error);
3536 	}
3537 
3538 	if (newval == 0) {
3539 		dsl_dataset_rele(ds, FTAG);
3540 		return (0);
3541 	}
3542 
3543 	if (newval < dsl_dataset_phys(ds)->ds_referenced_bytes ||
3544 	    newval < ds->ds_reserved) {
3545 		dsl_dataset_rele(ds, FTAG);
3546 		return (SET_ERROR(ENOSPC));
3547 	}
3548 
3549 	dsl_dataset_rele(ds, FTAG);
3550 	return (0);
3551 }
3552 
3553 static void
3554 dsl_dataset_set_refquota_sync(void *arg, dmu_tx_t *tx)
3555 {
3556 	dsl_dataset_set_qr_arg_t *ddsqra = arg;
3557 	dsl_pool_t *dp = dmu_tx_pool(tx);
3558 	dsl_dataset_t *ds;
3559 	uint64_t newval;
3560 
3561 	VERIFY0(dsl_dataset_hold(dp, ddsqra->ddsqra_name, FTAG, &ds));
3562 
3563 	dsl_prop_set_sync_impl(ds,
3564 	    zfs_prop_to_name(ZFS_PROP_REFQUOTA),
3565 	    ddsqra->ddsqra_source, sizeof (ddsqra->ddsqra_value), 1,
3566 	    &ddsqra->ddsqra_value, tx);
3567 
3568 	VERIFY0(dsl_prop_get_int_ds(ds,
3569 	    zfs_prop_to_name(ZFS_PROP_REFQUOTA), &newval));
3570 
3571 	if (ds->ds_quota != newval) {
3572 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
3573 		ds->ds_quota = newval;
3574 	}
3575 	dsl_dataset_rele(ds, FTAG);
3576 }
3577 
3578 int
3579 dsl_dataset_set_refquota(const char *dsname, zprop_source_t source,
3580     uint64_t refquota)
3581 {
3582 	dsl_dataset_set_qr_arg_t ddsqra;
3583 
3584 	ddsqra.ddsqra_name = dsname;
3585 	ddsqra.ddsqra_source = source;
3586 	ddsqra.ddsqra_value = refquota;
3587 
3588 	return (dsl_sync_task(dsname, dsl_dataset_set_refquota_check,
3589 	    dsl_dataset_set_refquota_sync, &ddsqra, 0, ZFS_SPACE_CHECK_NONE));
3590 }
3591 
3592 static int
3593 dsl_dataset_set_refreservation_check(void *arg, dmu_tx_t *tx)
3594 {
3595 	dsl_dataset_set_qr_arg_t *ddsqra = arg;
3596 	dsl_pool_t *dp = dmu_tx_pool(tx);
3597 	dsl_dataset_t *ds;
3598 	int error;
3599 	uint64_t newval, unique;
3600 
3601 	if (spa_version(dp->dp_spa) < SPA_VERSION_REFRESERVATION)
3602 		return (SET_ERROR(ENOTSUP));
3603 
3604 	error = dsl_dataset_hold(dp, ddsqra->ddsqra_name, FTAG, &ds);
3605 	if (error != 0)
3606 		return (error);
3607 
3608 	if (ds->ds_is_snapshot) {
3609 		dsl_dataset_rele(ds, FTAG);
3610 		return (SET_ERROR(EINVAL));
3611 	}
3612 
3613 	error = dsl_prop_predict(ds->ds_dir,
3614 	    zfs_prop_to_name(ZFS_PROP_REFRESERVATION),
3615 	    ddsqra->ddsqra_source, ddsqra->ddsqra_value, &newval);
3616 	if (error != 0) {
3617 		dsl_dataset_rele(ds, FTAG);
3618 		return (error);
3619 	}
3620 
3621 	/*
3622 	 * If we are doing the preliminary check in open context, the
3623 	 * space estimates may be inaccurate.
3624 	 */
3625 	if (!dmu_tx_is_syncing(tx)) {
3626 		dsl_dataset_rele(ds, FTAG);
3627 		return (0);
3628 	}
3629 
3630 	mutex_enter(&ds->ds_lock);
3631 	if (!DS_UNIQUE_IS_ACCURATE(ds))
3632 		dsl_dataset_recalc_head_uniq(ds);
3633 	unique = dsl_dataset_phys(ds)->ds_unique_bytes;
3634 	mutex_exit(&ds->ds_lock);
3635 
3636 	if (MAX(unique, newval) > MAX(unique, ds->ds_reserved)) {
3637 		uint64_t delta = MAX(unique, newval) -
3638 		    MAX(unique, ds->ds_reserved);
3639 
3640 		if (delta >
3641 		    dsl_dir_space_available(ds->ds_dir, NULL, 0, B_TRUE) ||
3642 		    (ds->ds_quota > 0 && newval > ds->ds_quota)) {
3643 			dsl_dataset_rele(ds, FTAG);
3644 			return (SET_ERROR(ENOSPC));
3645 		}
3646 	}
3647 
3648 	dsl_dataset_rele(ds, FTAG);
3649 	return (0);
3650 }
3651 
3652 void
3653 dsl_dataset_set_refreservation_sync_impl(dsl_dataset_t *ds,
3654     zprop_source_t source, uint64_t value, dmu_tx_t *tx)
3655 {
3656 	uint64_t newval;
3657 	uint64_t unique;
3658 	int64_t delta;
3659 
3660 	dsl_prop_set_sync_impl(ds, zfs_prop_to_name(ZFS_PROP_REFRESERVATION),
3661 	    source, sizeof (value), 1, &value, tx);
3662 
3663 	VERIFY0(dsl_prop_get_int_ds(ds,
3664 	    zfs_prop_to_name(ZFS_PROP_REFRESERVATION), &newval));
3665 
3666 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
3667 	mutex_enter(&ds->ds_dir->dd_lock);
3668 	mutex_enter(&ds->ds_lock);
3669 	ASSERT(DS_UNIQUE_IS_ACCURATE(ds));
3670 	unique = dsl_dataset_phys(ds)->ds_unique_bytes;
3671 	delta = MAX(0, (int64_t)(newval - unique)) -
3672 	    MAX(0, (int64_t)(ds->ds_reserved - unique));
3673 	ds->ds_reserved = newval;
3674 	mutex_exit(&ds->ds_lock);
3675 
3676 	dsl_dir_diduse_space(ds->ds_dir, DD_USED_REFRSRV, delta, 0, 0, tx);
3677 	mutex_exit(&ds->ds_dir->dd_lock);
3678 }
3679 
3680 static void
3681 dsl_dataset_set_refreservation_sync(void *arg, dmu_tx_t *tx)
3682 {
3683 	dsl_dataset_set_qr_arg_t *ddsqra = arg;
3684 	dsl_pool_t *dp = dmu_tx_pool(tx);
3685 	dsl_dataset_t *ds;
3686 
3687 	VERIFY0(dsl_dataset_hold(dp, ddsqra->ddsqra_name, FTAG, &ds));
3688 	dsl_dataset_set_refreservation_sync_impl(ds,
3689 	    ddsqra->ddsqra_source, ddsqra->ddsqra_value, tx);
3690 	dsl_dataset_rele(ds, FTAG);
3691 }
3692 
3693 int
3694 dsl_dataset_set_refreservation(const char *dsname, zprop_source_t source,
3695     uint64_t refreservation)
3696 {
3697 	dsl_dataset_set_qr_arg_t ddsqra;
3698 
3699 	ddsqra.ddsqra_name = dsname;
3700 	ddsqra.ddsqra_source = source;
3701 	ddsqra.ddsqra_value = refreservation;
3702 
3703 	return (dsl_sync_task(dsname, dsl_dataset_set_refreservation_check,
3704 	    dsl_dataset_set_refreservation_sync, &ddsqra,
3705 	    0, ZFS_SPACE_CHECK_NONE));
3706 }
3707 
3708 /*
3709  * Return (in *usedp) the amount of space written in new that is not
3710  * present in oldsnap.  New may be a snapshot or the head.  Old must be
3711  * a snapshot before new, in new's filesystem (or its origin).  If not then
3712  * fail and return EINVAL.
3713  *
3714  * The written space is calculated by considering two components:  First, we
3715  * ignore any freed space, and calculate the written as new's used space
3716  * minus old's used space.  Next, we add in the amount of space that was freed
3717  * between the two snapshots, thus reducing new's used space relative to old's.
3718  * Specifically, this is the space that was born before old->ds_creation_txg,
3719  * and freed before new (ie. on new's deadlist or a previous deadlist).
3720  *
3721  * space freed                         [---------------------]
3722  * snapshots                       ---O-------O--------O-------O------
3723  *                                         oldsnap            new
3724  */
3725 int
3726 dsl_dataset_space_written(dsl_dataset_t *oldsnap, dsl_dataset_t *new,
3727     uint64_t *usedp, uint64_t *compp, uint64_t *uncompp)
3728 {
3729 	int err = 0;
3730 	uint64_t snapobj;
3731 	dsl_pool_t *dp = new->ds_dir->dd_pool;
3732 
3733 	ASSERT(dsl_pool_config_held(dp));
3734 
3735 	*usedp = 0;
3736 	*usedp += dsl_dataset_phys(new)->ds_referenced_bytes;
3737 	*usedp -= dsl_dataset_phys(oldsnap)->ds_referenced_bytes;
3738 
3739 	*compp = 0;
3740 	*compp += dsl_dataset_phys(new)->ds_compressed_bytes;
3741 	*compp -= dsl_dataset_phys(oldsnap)->ds_compressed_bytes;
3742 
3743 	*uncompp = 0;
3744 	*uncompp += dsl_dataset_phys(new)->ds_uncompressed_bytes;
3745 	*uncompp -= dsl_dataset_phys(oldsnap)->ds_uncompressed_bytes;
3746 
3747 	snapobj = new->ds_object;
3748 	while (snapobj != oldsnap->ds_object) {
3749 		dsl_dataset_t *snap;
3750 		uint64_t used, comp, uncomp;
3751 
3752 		if (snapobj == new->ds_object) {
3753 			snap = new;
3754 		} else {
3755 			err = dsl_dataset_hold_obj(dp, snapobj, FTAG, &snap);
3756 			if (err != 0)
3757 				break;
3758 		}
3759 
3760 		if (dsl_dataset_phys(snap)->ds_prev_snap_txg ==
3761 		    dsl_dataset_phys(oldsnap)->ds_creation_txg) {
3762 			/*
3763 			 * The blocks in the deadlist can not be born after
3764 			 * ds_prev_snap_txg, so get the whole deadlist space,
3765 			 * which is more efficient (especially for old-format
3766 			 * deadlists).  Unfortunately the deadlist code
3767 			 * doesn't have enough information to make this
3768 			 * optimization itself.
3769 			 */
3770 			dsl_deadlist_space(&snap->ds_deadlist,
3771 			    &used, &comp, &uncomp);
3772 		} else {
3773 			dsl_deadlist_space_range(&snap->ds_deadlist,
3774 			    0, dsl_dataset_phys(oldsnap)->ds_creation_txg,
3775 			    &used, &comp, &uncomp);
3776 		}
3777 		*usedp += used;
3778 		*compp += comp;
3779 		*uncompp += uncomp;
3780 
3781 		/*
3782 		 * If we get to the beginning of the chain of snapshots
3783 		 * (ds_prev_snap_obj == 0) before oldsnap, then oldsnap
3784 		 * was not a snapshot of/before new.
3785 		 */
3786 		snapobj = dsl_dataset_phys(snap)->ds_prev_snap_obj;
3787 		if (snap != new)
3788 			dsl_dataset_rele(snap, FTAG);
3789 		if (snapobj == 0) {
3790 			err = SET_ERROR(EINVAL);
3791 			break;
3792 		}
3793 
3794 	}
3795 	return (err);
3796 }
3797 
3798 /*
3799  * Return (in *usedp) the amount of space that will be reclaimed if firstsnap,
3800  * lastsnap, and all snapshots in between are deleted.
3801  *
3802  * blocks that would be freed            [---------------------------]
3803  * snapshots                       ---O-------O--------O-------O--------O
3804  *                                        firstsnap        lastsnap
3805  *
3806  * This is the set of blocks that were born after the snap before firstsnap,
3807  * (birth > firstsnap->prev_snap_txg) and died before the snap after the
3808  * last snap (ie, is on lastsnap->ds_next->ds_deadlist or an earlier deadlist).
3809  * We calculate this by iterating over the relevant deadlists (from the snap
3810  * after lastsnap, backward to the snap after firstsnap), summing up the
3811  * space on the deadlist that was born after the snap before firstsnap.
3812  */
3813 int
3814 dsl_dataset_space_wouldfree(dsl_dataset_t *firstsnap,
3815     dsl_dataset_t *lastsnap,
3816     uint64_t *usedp, uint64_t *compp, uint64_t *uncompp)
3817 {
3818 	int err = 0;
3819 	uint64_t snapobj;
3820 	dsl_pool_t *dp = firstsnap->ds_dir->dd_pool;
3821 
3822 	ASSERT(firstsnap->ds_is_snapshot);
3823 	ASSERT(lastsnap->ds_is_snapshot);
3824 
3825 	/*
3826 	 * Check that the snapshots are in the same dsl_dir, and firstsnap
3827 	 * is before lastsnap.
3828 	 */
3829 	if (firstsnap->ds_dir != lastsnap->ds_dir ||
3830 	    dsl_dataset_phys(firstsnap)->ds_creation_txg >
3831 	    dsl_dataset_phys(lastsnap)->ds_creation_txg)
3832 		return (SET_ERROR(EINVAL));
3833 
3834 	*usedp = *compp = *uncompp = 0;
3835 
3836 	snapobj = dsl_dataset_phys(lastsnap)->ds_next_snap_obj;
3837 	while (snapobj != firstsnap->ds_object) {
3838 		dsl_dataset_t *ds;
3839 		uint64_t used, comp, uncomp;
3840 
3841 		err = dsl_dataset_hold_obj(dp, snapobj, FTAG, &ds);
3842 		if (err != 0)
3843 			break;
3844 
3845 		dsl_deadlist_space_range(&ds->ds_deadlist,
3846 		    dsl_dataset_phys(firstsnap)->ds_prev_snap_txg, UINT64_MAX,
3847 		    &used, &comp, &uncomp);
3848 		*usedp += used;
3849 		*compp += comp;
3850 		*uncompp += uncomp;
3851 
3852 		snapobj = dsl_dataset_phys(ds)->ds_prev_snap_obj;
3853 		ASSERT3U(snapobj, !=, 0);
3854 		dsl_dataset_rele(ds, FTAG);
3855 	}
3856 	return (err);
3857 }
3858 
3859 /*
3860  * Return TRUE if 'earlier' is an earlier snapshot in 'later's timeline.
3861  * For example, they could both be snapshots of the same filesystem, and
3862  * 'earlier' is before 'later'.  Or 'earlier' could be the origin of
3863  * 'later's filesystem.  Or 'earlier' could be an older snapshot in the origin's
3864  * filesystem.  Or 'earlier' could be the origin's origin.
3865  *
3866  * If non-zero, earlier_txg is used instead of earlier's ds_creation_txg.
3867  */
3868 boolean_t
3869 dsl_dataset_is_before(dsl_dataset_t *later, dsl_dataset_t *earlier,
3870     uint64_t earlier_txg)
3871 {
3872 	dsl_pool_t *dp = later->ds_dir->dd_pool;
3873 	int error;
3874 	boolean_t ret;
3875 
3876 	ASSERT(dsl_pool_config_held(dp));
3877 	ASSERT(earlier->ds_is_snapshot || earlier_txg != 0);
3878 
3879 	if (earlier_txg == 0)
3880 		earlier_txg = dsl_dataset_phys(earlier)->ds_creation_txg;
3881 
3882 	if (later->ds_is_snapshot &&
3883 	    earlier_txg >= dsl_dataset_phys(later)->ds_creation_txg)
3884 		return (B_FALSE);
3885 
3886 	if (later->ds_dir == earlier->ds_dir)
3887 		return (B_TRUE);
3888 	if (!dsl_dir_is_clone(later->ds_dir))
3889 		return (B_FALSE);
3890 
3891 	if (dsl_dir_phys(later->ds_dir)->dd_origin_obj == earlier->ds_object)
3892 		return (B_TRUE);
3893 	dsl_dataset_t *origin;
3894 	error = dsl_dataset_hold_obj(dp,
3895 	    dsl_dir_phys(later->ds_dir)->dd_origin_obj, FTAG, &origin);
3896 	if (error != 0)
3897 		return (B_FALSE);
3898 	ret = dsl_dataset_is_before(origin, earlier, earlier_txg);
3899 	dsl_dataset_rele(origin, FTAG);
3900 	return (ret);
3901 }
3902 
3903 void
3904 dsl_dataset_zapify(dsl_dataset_t *ds, dmu_tx_t *tx)
3905 {
3906 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
3907 	dmu_object_zapify(mos, ds->ds_object, DMU_OT_DSL_DATASET, tx);
3908 }
3909 
3910 boolean_t
3911 dsl_dataset_is_zapified(dsl_dataset_t *ds)
3912 {
3913 	dmu_object_info_t doi;
3914 
3915 	dmu_object_info_from_db(ds->ds_dbuf, &doi);
3916 	return (doi.doi_type == DMU_OTN_ZAP_METADATA);
3917 }
3918 
3919 boolean_t
3920 dsl_dataset_has_resume_receive_state(dsl_dataset_t *ds)
3921 {
3922 	return (dsl_dataset_is_zapified(ds) &&
3923 	    zap_contains(ds->ds_dir->dd_pool->dp_meta_objset,
3924 	    ds->ds_object, DS_FIELD_RESUME_TOGUID) == 0);
3925 }
3926