xref: /titanic_51/usr/src/uts/common/fs/zfs/dsl_dataset.c (revision 9e86db79b7d1bbc5f2f04e99954cbd5eae0e22bb)
1 /*
2  * CDDL HEADER START
3  *
4  * The contents of this file are subject to the terms of the
5  * Common Development and Distribution License (the "License").
6  * You may not use this file except in compliance with the License.
7  *
8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9  * or http://www.opensolaris.org/os/licensing.
10  * See the License for the specific language governing permissions
11  * and limitations under the License.
12  *
13  * When distributing Covered Code, include this CDDL HEADER in each
14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15  * If applicable, add the following below this CDDL HEADER, with the
16  * fields enclosed by brackets "[]" replaced with your own identifying
17  * information: Portions Copyright [yyyy] [name of copyright owner]
18  *
19  * CDDL HEADER END
20  */
21 /*
22  * Copyright 2009 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 #include <sys/dmu_objset.h>
27 #include <sys/dsl_dataset.h>
28 #include <sys/dsl_dir.h>
29 #include <sys/dsl_prop.h>
30 #include <sys/dsl_synctask.h>
31 #include <sys/dmu_traverse.h>
32 #include <sys/dmu_tx.h>
33 #include <sys/arc.h>
34 #include <sys/zio.h>
35 #include <sys/zap.h>
36 #include <sys/unique.h>
37 #include <sys/zfs_context.h>
38 #include <sys/zfs_ioctl.h>
39 #include <sys/spa.h>
40 #include <sys/zfs_znode.h>
41 #include <sys/sunddi.h>
42 #include <sys/zvol.h>
43 
44 static char *dsl_reaper = "the grim reaper";
45 
46 static dsl_checkfunc_t dsl_dataset_destroy_begin_check;
47 static dsl_syncfunc_t dsl_dataset_destroy_begin_sync;
48 static dsl_syncfunc_t dsl_dataset_set_reservation_sync;
49 
50 #define	DS_REF_MAX	(1ULL << 62)
51 
52 #define	DSL_DEADLIST_BLOCKSIZE	SPA_MAXBLOCKSIZE
53 
54 #define	DSL_DATASET_IS_DESTROYED(ds)	((ds)->ds_owner == dsl_reaper)
55 
56 
57 /*
58  * Figure out how much of this delta should be propogated to the dsl_dir
59  * layer.  If there's a refreservation, that space has already been
60  * partially accounted for in our ancestors.
61  */
62 static int64_t
63 parent_delta(dsl_dataset_t *ds, int64_t delta)
64 {
65 	uint64_t old_bytes, new_bytes;
66 
67 	if (ds->ds_reserved == 0)
68 		return (delta);
69 
70 	old_bytes = MAX(ds->ds_phys->ds_unique_bytes, ds->ds_reserved);
71 	new_bytes = MAX(ds->ds_phys->ds_unique_bytes + delta, ds->ds_reserved);
72 
73 	ASSERT3U(ABS((int64_t)(new_bytes - old_bytes)), <=, ABS(delta));
74 	return (new_bytes - old_bytes);
75 }
76 
77 void
78 dsl_dataset_block_born(dsl_dataset_t *ds, blkptr_t *bp, dmu_tx_t *tx)
79 {
80 	int used = bp_get_dasize(tx->tx_pool->dp_spa, bp);
81 	int compressed = BP_GET_PSIZE(bp);
82 	int uncompressed = BP_GET_UCSIZE(bp);
83 	int64_t delta;
84 
85 	dprintf_bp(bp, "born, ds=%p\n", ds);
86 
87 	ASSERT(dmu_tx_is_syncing(tx));
88 	/* It could have been compressed away to nothing */
89 	if (BP_IS_HOLE(bp))
90 		return;
91 	ASSERT(BP_GET_TYPE(bp) != DMU_OT_NONE);
92 	ASSERT3U(BP_GET_TYPE(bp), <, DMU_OT_NUMTYPES);
93 	if (ds == NULL) {
94 		/*
95 		 * Account for the meta-objset space in its placeholder
96 		 * dsl_dir.
97 		 */
98 		ASSERT3U(compressed, ==, uncompressed); /* it's all metadata */
99 		dsl_dir_diduse_space(tx->tx_pool->dp_mos_dir, DD_USED_HEAD,
100 		    used, compressed, uncompressed, tx);
101 		dsl_dir_dirty(tx->tx_pool->dp_mos_dir, tx);
102 		return;
103 	}
104 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
105 	mutex_enter(&ds->ds_dir->dd_lock);
106 	mutex_enter(&ds->ds_lock);
107 	delta = parent_delta(ds, used);
108 	ds->ds_phys->ds_used_bytes += used;
109 	ds->ds_phys->ds_compressed_bytes += compressed;
110 	ds->ds_phys->ds_uncompressed_bytes += uncompressed;
111 	ds->ds_phys->ds_unique_bytes += used;
112 	mutex_exit(&ds->ds_lock);
113 	dsl_dir_diduse_space(ds->ds_dir, DD_USED_HEAD, delta,
114 	    compressed, uncompressed, tx);
115 	dsl_dir_transfer_space(ds->ds_dir, used - delta,
116 	    DD_USED_REFRSRV, DD_USED_HEAD, tx);
117 	mutex_exit(&ds->ds_dir->dd_lock);
118 }
119 
120 int
121 dsl_dataset_block_kill(dsl_dataset_t *ds, blkptr_t *bp, zio_t *pio,
122     dmu_tx_t *tx)
123 {
124 	int used = bp_get_dasize(tx->tx_pool->dp_spa, bp);
125 	int compressed = BP_GET_PSIZE(bp);
126 	int uncompressed = BP_GET_UCSIZE(bp);
127 
128 	ASSERT(pio != NULL);
129 	ASSERT(dmu_tx_is_syncing(tx));
130 	/* No block pointer => nothing to free */
131 	if (BP_IS_HOLE(bp))
132 		return (0);
133 
134 	ASSERT(used > 0);
135 	if (ds == NULL) {
136 		int err;
137 		/*
138 		 * Account for the meta-objset space in its placeholder
139 		 * dataset.
140 		 */
141 		err = dsl_free(pio, tx->tx_pool,
142 		    tx->tx_txg, bp, NULL, NULL, ARC_NOWAIT);
143 		ASSERT(err == 0);
144 
145 		dsl_dir_diduse_space(tx->tx_pool->dp_mos_dir, DD_USED_HEAD,
146 		    -used, -compressed, -uncompressed, tx);
147 		dsl_dir_dirty(tx->tx_pool->dp_mos_dir, tx);
148 		return (used);
149 	}
150 	ASSERT3P(tx->tx_pool, ==, ds->ds_dir->dd_pool);
151 
152 	ASSERT(!dsl_dataset_is_snapshot(ds));
153 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
154 
155 	if (bp->blk_birth > ds->ds_phys->ds_prev_snap_txg) {
156 		int err;
157 		int64_t delta;
158 
159 		dprintf_bp(bp, "freeing: %s", "");
160 		err = dsl_free(pio, tx->tx_pool,
161 		    tx->tx_txg, bp, NULL, NULL, ARC_NOWAIT);
162 		ASSERT(err == 0);
163 
164 		mutex_enter(&ds->ds_dir->dd_lock);
165 		mutex_enter(&ds->ds_lock);
166 		ASSERT(ds->ds_phys->ds_unique_bytes >= used ||
167 		    !DS_UNIQUE_IS_ACCURATE(ds));
168 		delta = parent_delta(ds, -used);
169 		ds->ds_phys->ds_unique_bytes -= used;
170 		mutex_exit(&ds->ds_lock);
171 		dsl_dir_diduse_space(ds->ds_dir, DD_USED_HEAD,
172 		    delta, -compressed, -uncompressed, tx);
173 		dsl_dir_transfer_space(ds->ds_dir, -used - delta,
174 		    DD_USED_REFRSRV, DD_USED_HEAD, tx);
175 		mutex_exit(&ds->ds_dir->dd_lock);
176 	} else {
177 		dprintf_bp(bp, "putting on dead list: %s", "");
178 		VERIFY(0 == bplist_enqueue(&ds->ds_deadlist, bp, tx));
179 		ASSERT3U(ds->ds_prev->ds_object, ==,
180 		    ds->ds_phys->ds_prev_snap_obj);
181 		ASSERT(ds->ds_prev->ds_phys->ds_num_children > 0);
182 		/* if (bp->blk_birth > prev prev snap txg) prev unique += bs */
183 		if (ds->ds_prev->ds_phys->ds_next_snap_obj ==
184 		    ds->ds_object && bp->blk_birth >
185 		    ds->ds_prev->ds_phys->ds_prev_snap_txg) {
186 			dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
187 			mutex_enter(&ds->ds_prev->ds_lock);
188 			ds->ds_prev->ds_phys->ds_unique_bytes += used;
189 			mutex_exit(&ds->ds_prev->ds_lock);
190 		}
191 		if (bp->blk_birth > ds->ds_origin_txg) {
192 			dsl_dir_transfer_space(ds->ds_dir, used,
193 			    DD_USED_HEAD, DD_USED_SNAP, tx);
194 		}
195 	}
196 	mutex_enter(&ds->ds_lock);
197 	ASSERT3U(ds->ds_phys->ds_used_bytes, >=, used);
198 	ds->ds_phys->ds_used_bytes -= used;
199 	ASSERT3U(ds->ds_phys->ds_compressed_bytes, >=, compressed);
200 	ds->ds_phys->ds_compressed_bytes -= compressed;
201 	ASSERT3U(ds->ds_phys->ds_uncompressed_bytes, >=, uncompressed);
202 	ds->ds_phys->ds_uncompressed_bytes -= uncompressed;
203 	mutex_exit(&ds->ds_lock);
204 
205 	return (used);
206 }
207 
208 uint64_t
209 dsl_dataset_prev_snap_txg(dsl_dataset_t *ds)
210 {
211 	uint64_t trysnap = 0;
212 
213 	if (ds == NULL)
214 		return (0);
215 	/*
216 	 * The snapshot creation could fail, but that would cause an
217 	 * incorrect FALSE return, which would only result in an
218 	 * overestimation of the amount of space that an operation would
219 	 * consume, which is OK.
220 	 *
221 	 * There's also a small window where we could miss a pending
222 	 * snapshot, because we could set the sync task in the quiescing
223 	 * phase.  So this should only be used as a guess.
224 	 */
225 	if (ds->ds_trysnap_txg >
226 	    spa_last_synced_txg(ds->ds_dir->dd_pool->dp_spa))
227 		trysnap = ds->ds_trysnap_txg;
228 	return (MAX(ds->ds_phys->ds_prev_snap_txg, trysnap));
229 }
230 
231 boolean_t
232 dsl_dataset_block_freeable(dsl_dataset_t *ds, uint64_t blk_birth)
233 {
234 	return (blk_birth > dsl_dataset_prev_snap_txg(ds));
235 }
236 
237 /* ARGSUSED */
238 static void
239 dsl_dataset_evict(dmu_buf_t *db, void *dsv)
240 {
241 	dsl_dataset_t *ds = dsv;
242 
243 	ASSERT(ds->ds_owner == NULL || DSL_DATASET_IS_DESTROYED(ds));
244 
245 	unique_remove(ds->ds_fsid_guid);
246 
247 	if (ds->ds_objset != NULL)
248 		dmu_objset_evict(ds->ds_objset);
249 
250 	if (ds->ds_prev) {
251 		dsl_dataset_drop_ref(ds->ds_prev, ds);
252 		ds->ds_prev = NULL;
253 	}
254 
255 	bplist_close(&ds->ds_deadlist);
256 	if (ds->ds_dir)
257 		dsl_dir_close(ds->ds_dir, ds);
258 
259 	ASSERT(!list_link_active(&ds->ds_synced_link));
260 
261 	mutex_destroy(&ds->ds_lock);
262 	mutex_destroy(&ds->ds_recvlock);
263 	mutex_destroy(&ds->ds_opening_lock);
264 	mutex_destroy(&ds->ds_deadlist.bpl_lock);
265 	rw_destroy(&ds->ds_rwlock);
266 	cv_destroy(&ds->ds_exclusive_cv);
267 
268 	kmem_free(ds, sizeof (dsl_dataset_t));
269 }
270 
271 static int
272 dsl_dataset_get_snapname(dsl_dataset_t *ds)
273 {
274 	dsl_dataset_phys_t *headphys;
275 	int err;
276 	dmu_buf_t *headdbuf;
277 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
278 	objset_t *mos = dp->dp_meta_objset;
279 
280 	if (ds->ds_snapname[0])
281 		return (0);
282 	if (ds->ds_phys->ds_next_snap_obj == 0)
283 		return (0);
284 
285 	err = dmu_bonus_hold(mos, ds->ds_dir->dd_phys->dd_head_dataset_obj,
286 	    FTAG, &headdbuf);
287 	if (err)
288 		return (err);
289 	headphys = headdbuf->db_data;
290 	err = zap_value_search(dp->dp_meta_objset,
291 	    headphys->ds_snapnames_zapobj, ds->ds_object, 0, ds->ds_snapname);
292 	dmu_buf_rele(headdbuf, FTAG);
293 	return (err);
294 }
295 
296 static int
297 dsl_dataset_snap_lookup(dsl_dataset_t *ds, const char *name, uint64_t *value)
298 {
299 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
300 	uint64_t snapobj = ds->ds_phys->ds_snapnames_zapobj;
301 	matchtype_t mt;
302 	int err;
303 
304 	if (ds->ds_phys->ds_flags & DS_FLAG_CI_DATASET)
305 		mt = MT_FIRST;
306 	else
307 		mt = MT_EXACT;
308 
309 	err = zap_lookup_norm(mos, snapobj, name, 8, 1,
310 	    value, mt, NULL, 0, NULL);
311 	if (err == ENOTSUP && mt == MT_FIRST)
312 		err = zap_lookup(mos, snapobj, name, 8, 1, value);
313 	return (err);
314 }
315 
316 static int
317 dsl_dataset_snap_remove(dsl_dataset_t *ds, char *name, dmu_tx_t *tx)
318 {
319 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
320 	uint64_t snapobj = ds->ds_phys->ds_snapnames_zapobj;
321 	matchtype_t mt;
322 	int err;
323 
324 	dsl_dir_snap_cmtime_update(ds->ds_dir);
325 
326 	if (ds->ds_phys->ds_flags & DS_FLAG_CI_DATASET)
327 		mt = MT_FIRST;
328 	else
329 		mt = MT_EXACT;
330 
331 	err = zap_remove_norm(mos, snapobj, name, mt, tx);
332 	if (err == ENOTSUP && mt == MT_FIRST)
333 		err = zap_remove(mos, snapobj, name, tx);
334 	return (err);
335 }
336 
337 static int
338 dsl_dataset_get_ref(dsl_pool_t *dp, uint64_t dsobj, void *tag,
339     dsl_dataset_t **dsp)
340 {
341 	objset_t *mos = dp->dp_meta_objset;
342 	dmu_buf_t *dbuf;
343 	dsl_dataset_t *ds;
344 	int err;
345 
346 	ASSERT(RW_LOCK_HELD(&dp->dp_config_rwlock) ||
347 	    dsl_pool_sync_context(dp));
348 
349 	err = dmu_bonus_hold(mos, dsobj, tag, &dbuf);
350 	if (err)
351 		return (err);
352 	ds = dmu_buf_get_user(dbuf);
353 	if (ds == NULL) {
354 		dsl_dataset_t *winner;
355 
356 		ds = kmem_zalloc(sizeof (dsl_dataset_t), KM_SLEEP);
357 		ds->ds_dbuf = dbuf;
358 		ds->ds_object = dsobj;
359 		ds->ds_phys = dbuf->db_data;
360 
361 		mutex_init(&ds->ds_lock, NULL, MUTEX_DEFAULT, NULL);
362 		mutex_init(&ds->ds_recvlock, NULL, MUTEX_DEFAULT, NULL);
363 		mutex_init(&ds->ds_opening_lock, NULL, MUTEX_DEFAULT, NULL);
364 		mutex_init(&ds->ds_deadlist.bpl_lock, NULL, MUTEX_DEFAULT,
365 		    NULL);
366 		rw_init(&ds->ds_rwlock, 0, 0, 0);
367 		cv_init(&ds->ds_exclusive_cv, NULL, CV_DEFAULT, NULL);
368 
369 		err = bplist_open(&ds->ds_deadlist,
370 		    mos, ds->ds_phys->ds_deadlist_obj);
371 		if (err == 0) {
372 			err = dsl_dir_open_obj(dp,
373 			    ds->ds_phys->ds_dir_obj, NULL, ds, &ds->ds_dir);
374 		}
375 		if (err) {
376 			/*
377 			 * we don't really need to close the blist if we
378 			 * just opened it.
379 			 */
380 			mutex_destroy(&ds->ds_lock);
381 			mutex_destroy(&ds->ds_recvlock);
382 			mutex_destroy(&ds->ds_opening_lock);
383 			mutex_destroy(&ds->ds_deadlist.bpl_lock);
384 			rw_destroy(&ds->ds_rwlock);
385 			cv_destroy(&ds->ds_exclusive_cv);
386 			kmem_free(ds, sizeof (dsl_dataset_t));
387 			dmu_buf_rele(dbuf, tag);
388 			return (err);
389 		}
390 
391 		if (!dsl_dataset_is_snapshot(ds)) {
392 			ds->ds_snapname[0] = '\0';
393 			if (ds->ds_phys->ds_prev_snap_obj) {
394 				err = dsl_dataset_get_ref(dp,
395 				    ds->ds_phys->ds_prev_snap_obj,
396 				    ds, &ds->ds_prev);
397 			}
398 
399 			if (err == 0 && dsl_dir_is_clone(ds->ds_dir)) {
400 				dsl_dataset_t *origin;
401 
402 				err = dsl_dataset_hold_obj(dp,
403 				    ds->ds_dir->dd_phys->dd_origin_obj,
404 				    FTAG, &origin);
405 				if (err == 0) {
406 					ds->ds_origin_txg =
407 					    origin->ds_phys->ds_creation_txg;
408 					dsl_dataset_rele(origin, FTAG);
409 				}
410 			}
411 		} else {
412 			if (zfs_flags & ZFS_DEBUG_SNAPNAMES)
413 				err = dsl_dataset_get_snapname(ds);
414 			if (err == 0 && ds->ds_phys->ds_userrefs_obj != 0) {
415 				err = zap_count(
416 				    ds->ds_dir->dd_pool->dp_meta_objset,
417 				    ds->ds_phys->ds_userrefs_obj,
418 				    &ds->ds_userrefs);
419 			}
420 		}
421 
422 		if (err == 0 && !dsl_dataset_is_snapshot(ds)) {
423 			/*
424 			 * In sync context, we're called with either no lock
425 			 * or with the write lock.  If we're not syncing,
426 			 * we're always called with the read lock held.
427 			 */
428 			boolean_t need_lock =
429 			    !RW_WRITE_HELD(&dp->dp_config_rwlock) &&
430 			    dsl_pool_sync_context(dp);
431 
432 			if (need_lock)
433 				rw_enter(&dp->dp_config_rwlock, RW_READER);
434 
435 			err = dsl_prop_get_ds(ds,
436 			    "refreservation", sizeof (uint64_t), 1,
437 			    &ds->ds_reserved, NULL);
438 			if (err == 0) {
439 				err = dsl_prop_get_ds(ds,
440 				    "refquota", sizeof (uint64_t), 1,
441 				    &ds->ds_quota, NULL);
442 			}
443 
444 			if (need_lock)
445 				rw_exit(&dp->dp_config_rwlock);
446 		} else {
447 			ds->ds_reserved = ds->ds_quota = 0;
448 		}
449 
450 		if (err == 0) {
451 			winner = dmu_buf_set_user_ie(dbuf, ds, &ds->ds_phys,
452 			    dsl_dataset_evict);
453 		}
454 		if (err || winner) {
455 			bplist_close(&ds->ds_deadlist);
456 			if (ds->ds_prev)
457 				dsl_dataset_drop_ref(ds->ds_prev, ds);
458 			dsl_dir_close(ds->ds_dir, ds);
459 			mutex_destroy(&ds->ds_lock);
460 			mutex_destroy(&ds->ds_recvlock);
461 			mutex_destroy(&ds->ds_opening_lock);
462 			mutex_destroy(&ds->ds_deadlist.bpl_lock);
463 			rw_destroy(&ds->ds_rwlock);
464 			cv_destroy(&ds->ds_exclusive_cv);
465 			kmem_free(ds, sizeof (dsl_dataset_t));
466 			if (err) {
467 				dmu_buf_rele(dbuf, tag);
468 				return (err);
469 			}
470 			ds = winner;
471 		} else {
472 			ds->ds_fsid_guid =
473 			    unique_insert(ds->ds_phys->ds_fsid_guid);
474 		}
475 	}
476 	ASSERT3P(ds->ds_dbuf, ==, dbuf);
477 	ASSERT3P(ds->ds_phys, ==, dbuf->db_data);
478 	ASSERT(ds->ds_phys->ds_prev_snap_obj != 0 ||
479 	    spa_version(dp->dp_spa) < SPA_VERSION_ORIGIN ||
480 	    dp->dp_origin_snap == NULL || ds == dp->dp_origin_snap);
481 	mutex_enter(&ds->ds_lock);
482 	if (!dsl_pool_sync_context(dp) && DSL_DATASET_IS_DESTROYED(ds)) {
483 		mutex_exit(&ds->ds_lock);
484 		dmu_buf_rele(ds->ds_dbuf, tag);
485 		return (ENOENT);
486 	}
487 	mutex_exit(&ds->ds_lock);
488 	*dsp = ds;
489 	return (0);
490 }
491 
492 static int
493 dsl_dataset_hold_ref(dsl_dataset_t *ds, void *tag)
494 {
495 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
496 
497 	/*
498 	 * In syncing context we don't want the rwlock lock: there
499 	 * may be an existing writer waiting for sync phase to
500 	 * finish.  We don't need to worry about such writers, since
501 	 * sync phase is single-threaded, so the writer can't be
502 	 * doing anything while we are active.
503 	 */
504 	if (dsl_pool_sync_context(dp)) {
505 		ASSERT(!DSL_DATASET_IS_DESTROYED(ds));
506 		return (0);
507 	}
508 
509 	/*
510 	 * Normal users will hold the ds_rwlock as a READER until they
511 	 * are finished (i.e., call dsl_dataset_rele()).  "Owners" will
512 	 * drop their READER lock after they set the ds_owner field.
513 	 *
514 	 * If the dataset is being destroyed, the destroy thread will
515 	 * obtain a WRITER lock for exclusive access after it's done its
516 	 * open-context work and then change the ds_owner to
517 	 * dsl_reaper once destruction is assured.  So threads
518 	 * may block here temporarily, until the "destructability" of
519 	 * the dataset is determined.
520 	 */
521 	ASSERT(!RW_WRITE_HELD(&dp->dp_config_rwlock));
522 	mutex_enter(&ds->ds_lock);
523 	while (!rw_tryenter(&ds->ds_rwlock, RW_READER)) {
524 		rw_exit(&dp->dp_config_rwlock);
525 		cv_wait(&ds->ds_exclusive_cv, &ds->ds_lock);
526 		if (DSL_DATASET_IS_DESTROYED(ds)) {
527 			mutex_exit(&ds->ds_lock);
528 			dsl_dataset_drop_ref(ds, tag);
529 			rw_enter(&dp->dp_config_rwlock, RW_READER);
530 			return (ENOENT);
531 		}
532 		rw_enter(&dp->dp_config_rwlock, RW_READER);
533 	}
534 	mutex_exit(&ds->ds_lock);
535 	return (0);
536 }
537 
538 int
539 dsl_dataset_hold_obj(dsl_pool_t *dp, uint64_t dsobj, void *tag,
540     dsl_dataset_t **dsp)
541 {
542 	int err = dsl_dataset_get_ref(dp, dsobj, tag, dsp);
543 
544 	if (err)
545 		return (err);
546 	return (dsl_dataset_hold_ref(*dsp, tag));
547 }
548 
549 int
550 dsl_dataset_own_obj(dsl_pool_t *dp, uint64_t dsobj, boolean_t inconsistentok,
551     void *tag, dsl_dataset_t **dsp)
552 {
553 	int err = dsl_dataset_hold_obj(dp, dsobj, tag, dsp);
554 	if (err)
555 		return (err);
556 	if (!dsl_dataset_tryown(*dsp, inconsistentok, tag)) {
557 		dsl_dataset_rele(*dsp, tag);
558 		*dsp = NULL;
559 		return (EBUSY);
560 	}
561 	return (0);
562 }
563 
564 int
565 dsl_dataset_hold(const char *name, void *tag, dsl_dataset_t **dsp)
566 {
567 	dsl_dir_t *dd;
568 	dsl_pool_t *dp;
569 	const char *snapname;
570 	uint64_t obj;
571 	int err = 0;
572 
573 	err = dsl_dir_open_spa(NULL, name, FTAG, &dd, &snapname);
574 	if (err)
575 		return (err);
576 
577 	dp = dd->dd_pool;
578 	obj = dd->dd_phys->dd_head_dataset_obj;
579 	rw_enter(&dp->dp_config_rwlock, RW_READER);
580 	if (obj)
581 		err = dsl_dataset_get_ref(dp, obj, tag, dsp);
582 	else
583 		err = ENOENT;
584 	if (err)
585 		goto out;
586 
587 	err = dsl_dataset_hold_ref(*dsp, tag);
588 
589 	/* we may be looking for a snapshot */
590 	if (err == 0 && snapname != NULL) {
591 		dsl_dataset_t *ds = NULL;
592 
593 		if (*snapname++ != '@') {
594 			dsl_dataset_rele(*dsp, tag);
595 			err = ENOENT;
596 			goto out;
597 		}
598 
599 		dprintf("looking for snapshot '%s'\n", snapname);
600 		err = dsl_dataset_snap_lookup(*dsp, snapname, &obj);
601 		if (err == 0)
602 			err = dsl_dataset_get_ref(dp, obj, tag, &ds);
603 		dsl_dataset_rele(*dsp, tag);
604 
605 		ASSERT3U((err == 0), ==, (ds != NULL));
606 
607 		if (ds) {
608 			mutex_enter(&ds->ds_lock);
609 			if (ds->ds_snapname[0] == 0)
610 				(void) strlcpy(ds->ds_snapname, snapname,
611 				    sizeof (ds->ds_snapname));
612 			mutex_exit(&ds->ds_lock);
613 			err = dsl_dataset_hold_ref(ds, tag);
614 			*dsp = err ? NULL : ds;
615 		}
616 	}
617 out:
618 	rw_exit(&dp->dp_config_rwlock);
619 	dsl_dir_close(dd, FTAG);
620 	return (err);
621 }
622 
623 int
624 dsl_dataset_own(const char *name, boolean_t inconsistentok,
625     void *tag, dsl_dataset_t **dsp)
626 {
627 	int err = dsl_dataset_hold(name, tag, dsp);
628 	if (err)
629 		return (err);
630 	if (!dsl_dataset_tryown(*dsp, inconsistentok, tag)) {
631 		dsl_dataset_rele(*dsp, tag);
632 		return (EBUSY);
633 	}
634 	return (0);
635 }
636 
637 void
638 dsl_dataset_name(dsl_dataset_t *ds, char *name)
639 {
640 	if (ds == NULL) {
641 		(void) strcpy(name, "mos");
642 	} else {
643 		dsl_dir_name(ds->ds_dir, name);
644 		VERIFY(0 == dsl_dataset_get_snapname(ds));
645 		if (ds->ds_snapname[0]) {
646 			(void) strcat(name, "@");
647 			/*
648 			 * We use a "recursive" mutex so that we
649 			 * can call dprintf_ds() with ds_lock held.
650 			 */
651 			if (!MUTEX_HELD(&ds->ds_lock)) {
652 				mutex_enter(&ds->ds_lock);
653 				(void) strcat(name, ds->ds_snapname);
654 				mutex_exit(&ds->ds_lock);
655 			} else {
656 				(void) strcat(name, ds->ds_snapname);
657 			}
658 		}
659 	}
660 }
661 
662 static int
663 dsl_dataset_namelen(dsl_dataset_t *ds)
664 {
665 	int result;
666 
667 	if (ds == NULL) {
668 		result = 3;	/* "mos" */
669 	} else {
670 		result = dsl_dir_namelen(ds->ds_dir);
671 		VERIFY(0 == dsl_dataset_get_snapname(ds));
672 		if (ds->ds_snapname[0]) {
673 			++result;	/* adding one for the @-sign */
674 			if (!MUTEX_HELD(&ds->ds_lock)) {
675 				mutex_enter(&ds->ds_lock);
676 				result += strlen(ds->ds_snapname);
677 				mutex_exit(&ds->ds_lock);
678 			} else {
679 				result += strlen(ds->ds_snapname);
680 			}
681 		}
682 	}
683 
684 	return (result);
685 }
686 
687 void
688 dsl_dataset_drop_ref(dsl_dataset_t *ds, void *tag)
689 {
690 	dmu_buf_rele(ds->ds_dbuf, tag);
691 }
692 
693 void
694 dsl_dataset_rele(dsl_dataset_t *ds, void *tag)
695 {
696 	if (!dsl_pool_sync_context(ds->ds_dir->dd_pool)) {
697 		rw_exit(&ds->ds_rwlock);
698 	}
699 	dsl_dataset_drop_ref(ds, tag);
700 }
701 
702 void
703 dsl_dataset_disown(dsl_dataset_t *ds, void *tag)
704 {
705 	ASSERT((ds->ds_owner == tag && ds->ds_dbuf) ||
706 	    (DSL_DATASET_IS_DESTROYED(ds) && ds->ds_dbuf == NULL));
707 
708 	mutex_enter(&ds->ds_lock);
709 	ds->ds_owner = NULL;
710 	if (RW_WRITE_HELD(&ds->ds_rwlock)) {
711 		rw_exit(&ds->ds_rwlock);
712 		cv_broadcast(&ds->ds_exclusive_cv);
713 	}
714 	mutex_exit(&ds->ds_lock);
715 	if (ds->ds_dbuf)
716 		dsl_dataset_drop_ref(ds, tag);
717 	else
718 		dsl_dataset_evict(ds->ds_dbuf, ds);
719 }
720 
721 boolean_t
722 dsl_dataset_tryown(dsl_dataset_t *ds, boolean_t inconsistentok, void *tag)
723 {
724 	boolean_t gotit = FALSE;
725 
726 	mutex_enter(&ds->ds_lock);
727 	if (ds->ds_owner == NULL &&
728 	    (!DS_IS_INCONSISTENT(ds) || inconsistentok)) {
729 		ds->ds_owner = tag;
730 		if (!dsl_pool_sync_context(ds->ds_dir->dd_pool))
731 			rw_exit(&ds->ds_rwlock);
732 		gotit = TRUE;
733 	}
734 	mutex_exit(&ds->ds_lock);
735 	return (gotit);
736 }
737 
738 void
739 dsl_dataset_make_exclusive(dsl_dataset_t *ds, void *owner)
740 {
741 	ASSERT3P(owner, ==, ds->ds_owner);
742 	if (!RW_WRITE_HELD(&ds->ds_rwlock))
743 		rw_enter(&ds->ds_rwlock, RW_WRITER);
744 }
745 
746 uint64_t
747 dsl_dataset_create_sync_dd(dsl_dir_t *dd, dsl_dataset_t *origin,
748     uint64_t flags, dmu_tx_t *tx)
749 {
750 	dsl_pool_t *dp = dd->dd_pool;
751 	dmu_buf_t *dbuf;
752 	dsl_dataset_phys_t *dsphys;
753 	uint64_t dsobj;
754 	objset_t *mos = dp->dp_meta_objset;
755 
756 	if (origin == NULL)
757 		origin = dp->dp_origin_snap;
758 
759 	ASSERT(origin == NULL || origin->ds_dir->dd_pool == dp);
760 	ASSERT(origin == NULL || origin->ds_phys->ds_num_children > 0);
761 	ASSERT(dmu_tx_is_syncing(tx));
762 	ASSERT(dd->dd_phys->dd_head_dataset_obj == 0);
763 
764 	dsobj = dmu_object_alloc(mos, DMU_OT_DSL_DATASET, 0,
765 	    DMU_OT_DSL_DATASET, sizeof (dsl_dataset_phys_t), tx);
766 	VERIFY(0 == dmu_bonus_hold(mos, dsobj, FTAG, &dbuf));
767 	dmu_buf_will_dirty(dbuf, tx);
768 	dsphys = dbuf->db_data;
769 	bzero(dsphys, sizeof (dsl_dataset_phys_t));
770 	dsphys->ds_dir_obj = dd->dd_object;
771 	dsphys->ds_flags = flags;
772 	dsphys->ds_fsid_guid = unique_create();
773 	(void) random_get_pseudo_bytes((void*)&dsphys->ds_guid,
774 	    sizeof (dsphys->ds_guid));
775 	dsphys->ds_snapnames_zapobj =
776 	    zap_create_norm(mos, U8_TEXTPREP_TOUPPER, DMU_OT_DSL_DS_SNAP_MAP,
777 	    DMU_OT_NONE, 0, tx);
778 	dsphys->ds_creation_time = gethrestime_sec();
779 	dsphys->ds_creation_txg = tx->tx_txg == TXG_INITIAL ? 1 : tx->tx_txg;
780 	dsphys->ds_deadlist_obj =
781 	    bplist_create(mos, DSL_DEADLIST_BLOCKSIZE, tx);
782 
783 	if (origin) {
784 		dsphys->ds_prev_snap_obj = origin->ds_object;
785 		dsphys->ds_prev_snap_txg =
786 		    origin->ds_phys->ds_creation_txg;
787 		dsphys->ds_used_bytes =
788 		    origin->ds_phys->ds_used_bytes;
789 		dsphys->ds_compressed_bytes =
790 		    origin->ds_phys->ds_compressed_bytes;
791 		dsphys->ds_uncompressed_bytes =
792 		    origin->ds_phys->ds_uncompressed_bytes;
793 		dsphys->ds_bp = origin->ds_phys->ds_bp;
794 		dsphys->ds_flags |= origin->ds_phys->ds_flags;
795 
796 		dmu_buf_will_dirty(origin->ds_dbuf, tx);
797 		origin->ds_phys->ds_num_children++;
798 
799 		if (spa_version(dp->dp_spa) >= SPA_VERSION_NEXT_CLONES) {
800 			if (origin->ds_phys->ds_next_clones_obj == 0) {
801 				origin->ds_phys->ds_next_clones_obj =
802 				    zap_create(mos,
803 				    DMU_OT_NEXT_CLONES, DMU_OT_NONE, 0, tx);
804 			}
805 			VERIFY(0 == zap_add_int(mos,
806 			    origin->ds_phys->ds_next_clones_obj,
807 			    dsobj, tx));
808 		}
809 
810 		dmu_buf_will_dirty(dd->dd_dbuf, tx);
811 		dd->dd_phys->dd_origin_obj = origin->ds_object;
812 	}
813 
814 	if (spa_version(dp->dp_spa) >= SPA_VERSION_UNIQUE_ACCURATE)
815 		dsphys->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
816 
817 	dmu_buf_rele(dbuf, FTAG);
818 
819 	dmu_buf_will_dirty(dd->dd_dbuf, tx);
820 	dd->dd_phys->dd_head_dataset_obj = dsobj;
821 
822 	return (dsobj);
823 }
824 
825 uint64_t
826 dsl_dataset_create_sync(dsl_dir_t *pdd, const char *lastname,
827     dsl_dataset_t *origin, uint64_t flags, cred_t *cr, dmu_tx_t *tx)
828 {
829 	dsl_pool_t *dp = pdd->dd_pool;
830 	uint64_t dsobj, ddobj;
831 	dsl_dir_t *dd;
832 
833 	ASSERT(lastname[0] != '@');
834 
835 	ddobj = dsl_dir_create_sync(dp, pdd, lastname, tx);
836 	VERIFY(0 == dsl_dir_open_obj(dp, ddobj, lastname, FTAG, &dd));
837 
838 	dsobj = dsl_dataset_create_sync_dd(dd, origin, flags, tx);
839 
840 	dsl_deleg_set_create_perms(dd, tx, cr);
841 
842 	dsl_dir_close(dd, FTAG);
843 
844 	return (dsobj);
845 }
846 
847 struct destroyarg {
848 	dsl_sync_task_group_t *dstg;
849 	char *snapname;
850 	char *failed;
851 	boolean_t defer;
852 };
853 
854 static int
855 dsl_snapshot_destroy_one(char *name, void *arg)
856 {
857 	struct destroyarg *da = arg;
858 	dsl_dataset_t *ds;
859 	int err;
860 	char *dsname;
861 
862 	dsname = kmem_asprintf("%s@%s", name, da->snapname);
863 	err = dsl_dataset_own(dsname, B_TRUE, da->dstg, &ds);
864 	strfree(dsname);
865 	if (err == 0) {
866 		struct dsl_ds_destroyarg *dsda;
867 
868 		dsl_dataset_make_exclusive(ds, da->dstg);
869 		if (ds->ds_objset != NULL) {
870 			dmu_objset_evict(ds->ds_objset);
871 			ds->ds_objset = NULL;
872 		}
873 		dsda = kmem_zalloc(sizeof (struct dsl_ds_destroyarg), KM_SLEEP);
874 		dsda->ds = ds;
875 		dsda->defer = da->defer;
876 		dsl_sync_task_create(da->dstg, dsl_dataset_destroy_check,
877 		    dsl_dataset_destroy_sync, dsda, da->dstg, 0);
878 	} else if (err == ENOENT) {
879 		err = 0;
880 	} else {
881 		(void) strcpy(da->failed, name);
882 	}
883 	return (err);
884 }
885 
886 /*
887  * Destroy 'snapname' in all descendants of 'fsname'.
888  */
889 #pragma weak dmu_snapshots_destroy = dsl_snapshots_destroy
890 int
891 dsl_snapshots_destroy(char *fsname, char *snapname, boolean_t defer)
892 {
893 	int err;
894 	struct destroyarg da;
895 	dsl_sync_task_t *dst;
896 	spa_t *spa;
897 
898 	err = spa_open(fsname, &spa, FTAG);
899 	if (err)
900 		return (err);
901 	da.dstg = dsl_sync_task_group_create(spa_get_dsl(spa));
902 	da.snapname = snapname;
903 	da.failed = fsname;
904 	da.defer = defer;
905 
906 	err = dmu_objset_find(fsname,
907 	    dsl_snapshot_destroy_one, &da, DS_FIND_CHILDREN);
908 
909 	if (err == 0)
910 		err = dsl_sync_task_group_wait(da.dstg);
911 
912 	for (dst = list_head(&da.dstg->dstg_tasks); dst;
913 	    dst = list_next(&da.dstg->dstg_tasks, dst)) {
914 		struct dsl_ds_destroyarg *dsda = dst->dst_arg1;
915 		dsl_dataset_t *ds = dsda->ds;
916 
917 		/*
918 		 * Return the file system name that triggered the error
919 		 */
920 		if (dst->dst_err) {
921 			dsl_dataset_name(ds, fsname);
922 			*strchr(fsname, '@') = '\0';
923 		}
924 		ASSERT3P(dsda->rm_origin, ==, NULL);
925 		dsl_dataset_disown(ds, da.dstg);
926 		kmem_free(dsda, sizeof (struct dsl_ds_destroyarg));
927 	}
928 
929 	dsl_sync_task_group_destroy(da.dstg);
930 	spa_close(spa, FTAG);
931 	return (err);
932 }
933 
934 static boolean_t
935 dsl_dataset_might_destroy_origin(dsl_dataset_t *ds)
936 {
937 	boolean_t might_destroy = B_FALSE;
938 
939 	mutex_enter(&ds->ds_lock);
940 	if (ds->ds_phys->ds_num_children == 2 && ds->ds_userrefs == 0 &&
941 	    DS_IS_DEFER_DESTROY(ds))
942 		might_destroy = B_TRUE;
943 	mutex_exit(&ds->ds_lock);
944 
945 	return (might_destroy);
946 }
947 
948 /*
949  * If we're removing a clone, and these three conditions are true:
950  *	1) the clone's origin has no other children
951  *	2) the clone's origin has no user references
952  *	3) the clone's origin has been marked for deferred destruction
953  * Then, prepare to remove the origin as part of this sync task group.
954  */
955 static int
956 dsl_dataset_origin_rm_prep(struct dsl_ds_destroyarg *dsda, void *tag)
957 {
958 	dsl_dataset_t *ds = dsda->ds;
959 	dsl_dataset_t *origin = ds->ds_prev;
960 
961 	if (dsl_dataset_might_destroy_origin(origin)) {
962 		char *name;
963 		int namelen;
964 		int error;
965 
966 		namelen = dsl_dataset_namelen(origin) + 1;
967 		name = kmem_alloc(namelen, KM_SLEEP);
968 		dsl_dataset_name(origin, name);
969 #ifdef _KERNEL
970 		error = zfs_unmount_snap(name, NULL);
971 		if (error) {
972 			kmem_free(name, namelen);
973 			return (error);
974 		}
975 #endif
976 		error = dsl_dataset_own(name, B_TRUE, tag, &origin);
977 		kmem_free(name, namelen);
978 		if (error)
979 			return (error);
980 		dsda->rm_origin = origin;
981 		dsl_dataset_make_exclusive(origin, tag);
982 
983 		if (origin->ds_objset != NULL) {
984 			dmu_objset_evict(origin->ds_objset);
985 			origin->ds_objset = NULL;
986 		}
987 	}
988 
989 	return (0);
990 }
991 
992 /*
993  * ds must be opened as OWNER.  On return (whether successful or not),
994  * ds will be closed and caller can no longer dereference it.
995  */
996 int
997 dsl_dataset_destroy(dsl_dataset_t *ds, void *tag, boolean_t defer)
998 {
999 	int err;
1000 	dsl_sync_task_group_t *dstg;
1001 	objset_t *os;
1002 	dsl_dir_t *dd;
1003 	uint64_t obj;
1004 	struct dsl_ds_destroyarg dsda = {0};
1005 
1006 	dsda.ds = ds;
1007 
1008 	if (dsl_dataset_is_snapshot(ds)) {
1009 		/* Destroying a snapshot is simpler */
1010 		dsl_dataset_make_exclusive(ds, tag);
1011 
1012 		if (ds->ds_objset != NULL) {
1013 			dmu_objset_evict(ds->ds_objset);
1014 			ds->ds_objset = NULL;
1015 		}
1016 		dsda.defer = defer;
1017 		err = dsl_sync_task_do(ds->ds_dir->dd_pool,
1018 		    dsl_dataset_destroy_check, dsl_dataset_destroy_sync,
1019 		    &dsda, tag, 0);
1020 		ASSERT3P(dsda.rm_origin, ==, NULL);
1021 		goto out;
1022 	} else if (defer) {
1023 		err = EINVAL;
1024 		goto out;
1025 	}
1026 
1027 	dd = ds->ds_dir;
1028 
1029 	/*
1030 	 * Check for errors and mark this ds as inconsistent, in
1031 	 * case we crash while freeing the objects.
1032 	 */
1033 	err = dsl_sync_task_do(dd->dd_pool, dsl_dataset_destroy_begin_check,
1034 	    dsl_dataset_destroy_begin_sync, ds, NULL, 0);
1035 	if (err)
1036 		goto out;
1037 
1038 	err = dmu_objset_from_ds(ds, &os);
1039 	if (err)
1040 		goto out;
1041 
1042 	/*
1043 	 * remove the objects in open context, so that we won't
1044 	 * have too much to do in syncing context.
1045 	 */
1046 	for (obj = 0; err == 0; err = dmu_object_next(os, &obj, FALSE,
1047 	    ds->ds_phys->ds_prev_snap_txg)) {
1048 		/*
1049 		 * Ignore errors, if there is not enough disk space
1050 		 * we will deal with it in dsl_dataset_destroy_sync().
1051 		 */
1052 		(void) dmu_free_object(os, obj);
1053 	}
1054 
1055 	/*
1056 	 * We need to sync out all in-flight IO before we try to evict
1057 	 * (the dataset evict func is trying to clear the cached entries
1058 	 * for this dataset in the ARC).
1059 	 */
1060 	txg_wait_synced(dd->dd_pool, 0);
1061 
1062 	/*
1063 	 * If we managed to free all the objects in open
1064 	 * context, the user space accounting should be zero.
1065 	 */
1066 	if (ds->ds_phys->ds_bp.blk_fill == 0 &&
1067 	    dmu_objset_userused_enabled(os)) {
1068 		uint64_t count;
1069 
1070 		ASSERT(zap_count(os, DMU_USERUSED_OBJECT, &count) != 0 ||
1071 		    count == 0);
1072 		ASSERT(zap_count(os, DMU_GROUPUSED_OBJECT, &count) != 0 ||
1073 		    count == 0);
1074 	}
1075 
1076 	if (err != ESRCH)
1077 		goto out;
1078 
1079 	rw_enter(&dd->dd_pool->dp_config_rwlock, RW_READER);
1080 	err = dsl_dir_open_obj(dd->dd_pool, dd->dd_object, NULL, FTAG, &dd);
1081 	rw_exit(&dd->dd_pool->dp_config_rwlock);
1082 
1083 	if (err)
1084 		goto out;
1085 
1086 	if (ds->ds_objset) {
1087 		/*
1088 		 * We need to sync out all in-flight IO before we try
1089 		 * to evict (the dataset evict func is trying to clear
1090 		 * the cached entries for this dataset in the ARC).
1091 		 */
1092 		txg_wait_synced(dd->dd_pool, 0);
1093 	}
1094 
1095 	/*
1096 	 * Blow away the dsl_dir + head dataset.
1097 	 */
1098 	dsl_dataset_make_exclusive(ds, tag);
1099 	if (ds->ds_objset) {
1100 		dmu_objset_evict(ds->ds_objset);
1101 		ds->ds_objset = NULL;
1102 	}
1103 
1104 	/*
1105 	 * If we're removing a clone, we might also need to remove its
1106 	 * origin.
1107 	 */
1108 	do {
1109 		dsda.need_prep = B_FALSE;
1110 		if (dsl_dir_is_clone(dd)) {
1111 			err = dsl_dataset_origin_rm_prep(&dsda, tag);
1112 			if (err) {
1113 				dsl_dir_close(dd, FTAG);
1114 				goto out;
1115 			}
1116 		}
1117 
1118 		dstg = dsl_sync_task_group_create(ds->ds_dir->dd_pool);
1119 		dsl_sync_task_create(dstg, dsl_dataset_destroy_check,
1120 		    dsl_dataset_destroy_sync, &dsda, tag, 0);
1121 		dsl_sync_task_create(dstg, dsl_dir_destroy_check,
1122 		    dsl_dir_destroy_sync, dd, FTAG, 0);
1123 		err = dsl_sync_task_group_wait(dstg);
1124 		dsl_sync_task_group_destroy(dstg);
1125 
1126 		/*
1127 		 * We could be racing against 'zfs release' or 'zfs destroy -d'
1128 		 * on the origin snap, in which case we can get EBUSY if we
1129 		 * needed to destroy the origin snap but were not ready to
1130 		 * do so.
1131 		 */
1132 		if (dsda.need_prep) {
1133 			ASSERT(err == EBUSY);
1134 			ASSERT(dsl_dir_is_clone(dd));
1135 			ASSERT(dsda.rm_origin == NULL);
1136 		}
1137 	} while (dsda.need_prep);
1138 
1139 	if (dsda.rm_origin != NULL)
1140 		dsl_dataset_disown(dsda.rm_origin, tag);
1141 
1142 	/* if it is successful, dsl_dir_destroy_sync will close the dd */
1143 	if (err)
1144 		dsl_dir_close(dd, FTAG);
1145 out:
1146 	dsl_dataset_disown(ds, tag);
1147 	return (err);
1148 }
1149 
1150 blkptr_t *
1151 dsl_dataset_get_blkptr(dsl_dataset_t *ds)
1152 {
1153 	return (&ds->ds_phys->ds_bp);
1154 }
1155 
1156 void
1157 dsl_dataset_set_blkptr(dsl_dataset_t *ds, blkptr_t *bp, dmu_tx_t *tx)
1158 {
1159 	ASSERT(dmu_tx_is_syncing(tx));
1160 	/* If it's the meta-objset, set dp_meta_rootbp */
1161 	if (ds == NULL) {
1162 		tx->tx_pool->dp_meta_rootbp = *bp;
1163 	} else {
1164 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
1165 		ds->ds_phys->ds_bp = *bp;
1166 	}
1167 }
1168 
1169 spa_t *
1170 dsl_dataset_get_spa(dsl_dataset_t *ds)
1171 {
1172 	return (ds->ds_dir->dd_pool->dp_spa);
1173 }
1174 
1175 void
1176 dsl_dataset_dirty(dsl_dataset_t *ds, dmu_tx_t *tx)
1177 {
1178 	dsl_pool_t *dp;
1179 
1180 	if (ds == NULL) /* this is the meta-objset */
1181 		return;
1182 
1183 	ASSERT(ds->ds_objset != NULL);
1184 
1185 	if (ds->ds_phys->ds_next_snap_obj != 0)
1186 		panic("dirtying snapshot!");
1187 
1188 	dp = ds->ds_dir->dd_pool;
1189 
1190 	if (txg_list_add(&dp->dp_dirty_datasets, ds, tx->tx_txg) == 0) {
1191 		/* up the hold count until we can be written out */
1192 		dmu_buf_add_ref(ds->ds_dbuf, ds);
1193 	}
1194 }
1195 
1196 /*
1197  * The unique space in the head dataset can be calculated by subtracting
1198  * the space used in the most recent snapshot, that is still being used
1199  * in this file system, from the space currently in use.  To figure out
1200  * the space in the most recent snapshot still in use, we need to take
1201  * the total space used in the snapshot and subtract out the space that
1202  * has been freed up since the snapshot was taken.
1203  */
1204 static void
1205 dsl_dataset_recalc_head_uniq(dsl_dataset_t *ds)
1206 {
1207 	uint64_t mrs_used;
1208 	uint64_t dlused, dlcomp, dluncomp;
1209 
1210 	ASSERT(ds->ds_object == ds->ds_dir->dd_phys->dd_head_dataset_obj);
1211 
1212 	if (ds->ds_phys->ds_prev_snap_obj != 0)
1213 		mrs_used = ds->ds_prev->ds_phys->ds_used_bytes;
1214 	else
1215 		mrs_used = 0;
1216 
1217 	VERIFY(0 == bplist_space(&ds->ds_deadlist, &dlused, &dlcomp,
1218 	    &dluncomp));
1219 
1220 	ASSERT3U(dlused, <=, mrs_used);
1221 	ds->ds_phys->ds_unique_bytes =
1222 	    ds->ds_phys->ds_used_bytes - (mrs_used - dlused);
1223 
1224 	if (!DS_UNIQUE_IS_ACCURATE(ds) &&
1225 	    spa_version(ds->ds_dir->dd_pool->dp_spa) >=
1226 	    SPA_VERSION_UNIQUE_ACCURATE)
1227 		ds->ds_phys->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
1228 }
1229 
1230 static uint64_t
1231 dsl_dataset_unique(dsl_dataset_t *ds)
1232 {
1233 	if (!DS_UNIQUE_IS_ACCURATE(ds) && !dsl_dataset_is_snapshot(ds))
1234 		dsl_dataset_recalc_head_uniq(ds);
1235 
1236 	return (ds->ds_phys->ds_unique_bytes);
1237 }
1238 
1239 struct killarg {
1240 	dsl_dataset_t *ds;
1241 	zio_t *zio;
1242 	dmu_tx_t *tx;
1243 };
1244 
1245 /* ARGSUSED */
1246 static int
1247 kill_blkptr(spa_t *spa, blkptr_t *bp, const zbookmark_t *zb,
1248     const dnode_phys_t *dnp, void *arg)
1249 {
1250 	struct killarg *ka = arg;
1251 
1252 	if (bp == NULL)
1253 		return (0);
1254 
1255 	if ((zb->zb_level == -1ULL && zb->zb_blkid != 0) ||
1256 	    (zb->zb_object != 0 && dnp == NULL)) {
1257 		/*
1258 		 * It's a block in the intent log.  It has no
1259 		 * accounting, so just free it.
1260 		 */
1261 		VERIFY3U(0, ==, dsl_free(ka->zio, ka->tx->tx_pool,
1262 		    ka->tx->tx_txg, bp, NULL, NULL, ARC_NOWAIT));
1263 	} else {
1264 		ASSERT3U(bp->blk_birth, >, ka->ds->ds_phys->ds_prev_snap_txg);
1265 		(void) dsl_dataset_block_kill(ka->ds, bp, ka->zio, ka->tx);
1266 	}
1267 
1268 	return (0);
1269 }
1270 
1271 /* ARGSUSED */
1272 static int
1273 dsl_dataset_destroy_begin_check(void *arg1, void *arg2, dmu_tx_t *tx)
1274 {
1275 	dsl_dataset_t *ds = arg1;
1276 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
1277 	uint64_t count;
1278 	int err;
1279 
1280 	/*
1281 	 * Can't delete a head dataset if there are snapshots of it.
1282 	 * (Except if the only snapshots are from the branch we cloned
1283 	 * from.)
1284 	 */
1285 	if (ds->ds_prev != NULL &&
1286 	    ds->ds_prev->ds_phys->ds_next_snap_obj == ds->ds_object)
1287 		return (EINVAL);
1288 
1289 	/*
1290 	 * This is really a dsl_dir thing, but check it here so that
1291 	 * we'll be less likely to leave this dataset inconsistent &
1292 	 * nearly destroyed.
1293 	 */
1294 	err = zap_count(mos, ds->ds_dir->dd_phys->dd_child_dir_zapobj, &count);
1295 	if (err)
1296 		return (err);
1297 	if (count != 0)
1298 		return (EEXIST);
1299 
1300 	return (0);
1301 }
1302 
1303 /* ARGSUSED */
1304 static void
1305 dsl_dataset_destroy_begin_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
1306 {
1307 	dsl_dataset_t *ds = arg1;
1308 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1309 
1310 	/* Mark it as inconsistent on-disk, in case we crash */
1311 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
1312 	ds->ds_phys->ds_flags |= DS_FLAG_INCONSISTENT;
1313 
1314 	spa_history_internal_log(LOG_DS_DESTROY_BEGIN, dp->dp_spa, tx,
1315 	    cr, "dataset = %llu", ds->ds_object);
1316 }
1317 
1318 static int
1319 dsl_dataset_origin_check(struct dsl_ds_destroyarg *dsda, void *tag,
1320     dmu_tx_t *tx)
1321 {
1322 	dsl_dataset_t *ds = dsda->ds;
1323 	dsl_dataset_t *ds_prev = ds->ds_prev;
1324 
1325 	if (dsl_dataset_might_destroy_origin(ds_prev)) {
1326 		struct dsl_ds_destroyarg ndsda = {0};
1327 
1328 		/*
1329 		 * If we're not prepared to remove the origin, don't remove
1330 		 * the clone either.
1331 		 */
1332 		if (dsda->rm_origin == NULL) {
1333 			dsda->need_prep = B_TRUE;
1334 			return (EBUSY);
1335 		}
1336 
1337 		ndsda.ds = ds_prev;
1338 		ndsda.is_origin_rm = B_TRUE;
1339 		return (dsl_dataset_destroy_check(&ndsda, tag, tx));
1340 	}
1341 
1342 	/*
1343 	 * If we're not going to remove the origin after all,
1344 	 * undo the open context setup.
1345 	 */
1346 	if (dsda->rm_origin != NULL) {
1347 		dsl_dataset_disown(dsda->rm_origin, tag);
1348 		dsda->rm_origin = NULL;
1349 	}
1350 
1351 	return (0);
1352 }
1353 
1354 /* ARGSUSED */
1355 int
1356 dsl_dataset_destroy_check(void *arg1, void *arg2, dmu_tx_t *tx)
1357 {
1358 	struct dsl_ds_destroyarg *dsda = arg1;
1359 	dsl_dataset_t *ds = dsda->ds;
1360 
1361 	/* we have an owner hold, so noone else can destroy us */
1362 	ASSERT(!DSL_DATASET_IS_DESTROYED(ds));
1363 
1364 	/*
1365 	 * Only allow deferred destroy on pools that support it.
1366 	 * NOTE: deferred destroy is only supported on snapshots.
1367 	 */
1368 	if (dsda->defer) {
1369 		if (spa_version(ds->ds_dir->dd_pool->dp_spa) <
1370 		    SPA_VERSION_USERREFS)
1371 			return (ENOTSUP);
1372 		ASSERT(dsl_dataset_is_snapshot(ds));
1373 		return (0);
1374 	}
1375 
1376 	/*
1377 	 * Can't delete a head dataset if there are snapshots of it.
1378 	 * (Except if the only snapshots are from the branch we cloned
1379 	 * from.)
1380 	 */
1381 	if (ds->ds_prev != NULL &&
1382 	    ds->ds_prev->ds_phys->ds_next_snap_obj == ds->ds_object)
1383 		return (EINVAL);
1384 
1385 	/*
1386 	 * If we made changes this txg, traverse_dsl_dataset won't find
1387 	 * them.  Try again.
1388 	 */
1389 	if (ds->ds_phys->ds_bp.blk_birth >= tx->tx_txg)
1390 		return (EAGAIN);
1391 
1392 	if (dsl_dataset_is_snapshot(ds)) {
1393 		/*
1394 		 * If this snapshot has an elevated user reference count,
1395 		 * we can't destroy it yet.
1396 		 */
1397 		if (ds->ds_userrefs > 0 && !dsda->releasing)
1398 			return (EBUSY);
1399 
1400 		mutex_enter(&ds->ds_lock);
1401 		/*
1402 		 * Can't delete a branch point. However, if we're destroying
1403 		 * a clone and removing its origin due to it having a user
1404 		 * hold count of 0 and having been marked for deferred destroy,
1405 		 * it's OK for the origin to have a single clone.
1406 		 */
1407 		if (ds->ds_phys->ds_num_children >
1408 		    (dsda->is_origin_rm ? 2 : 1)) {
1409 			mutex_exit(&ds->ds_lock);
1410 			return (EEXIST);
1411 		}
1412 		mutex_exit(&ds->ds_lock);
1413 	} else if (dsl_dir_is_clone(ds->ds_dir)) {
1414 		return (dsl_dataset_origin_check(dsda, arg2, tx));
1415 	}
1416 
1417 	/* XXX we should do some i/o error checking... */
1418 	return (0);
1419 }
1420 
1421 struct refsarg {
1422 	kmutex_t lock;
1423 	boolean_t gone;
1424 	kcondvar_t cv;
1425 };
1426 
1427 /* ARGSUSED */
1428 static void
1429 dsl_dataset_refs_gone(dmu_buf_t *db, void *argv)
1430 {
1431 	struct refsarg *arg = argv;
1432 
1433 	mutex_enter(&arg->lock);
1434 	arg->gone = TRUE;
1435 	cv_signal(&arg->cv);
1436 	mutex_exit(&arg->lock);
1437 }
1438 
1439 static void
1440 dsl_dataset_drain_refs(dsl_dataset_t *ds, void *tag)
1441 {
1442 	struct refsarg arg;
1443 
1444 	mutex_init(&arg.lock, NULL, MUTEX_DEFAULT, NULL);
1445 	cv_init(&arg.cv, NULL, CV_DEFAULT, NULL);
1446 	arg.gone = FALSE;
1447 	(void) dmu_buf_update_user(ds->ds_dbuf, ds, &arg, &ds->ds_phys,
1448 	    dsl_dataset_refs_gone);
1449 	dmu_buf_rele(ds->ds_dbuf, tag);
1450 	mutex_enter(&arg.lock);
1451 	while (!arg.gone)
1452 		cv_wait(&arg.cv, &arg.lock);
1453 	ASSERT(arg.gone);
1454 	mutex_exit(&arg.lock);
1455 	ds->ds_dbuf = NULL;
1456 	ds->ds_phys = NULL;
1457 	mutex_destroy(&arg.lock);
1458 	cv_destroy(&arg.cv);
1459 }
1460 
1461 void
1462 dsl_dataset_destroy_sync(void *arg1, void *tag, cred_t *cr, dmu_tx_t *tx)
1463 {
1464 	struct dsl_ds_destroyarg *dsda = arg1;
1465 	dsl_dataset_t *ds = dsda->ds;
1466 	zio_t *zio;
1467 	int err;
1468 	int after_branch_point = FALSE;
1469 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1470 	objset_t *mos = dp->dp_meta_objset;
1471 	dsl_dataset_t *ds_prev = NULL;
1472 	uint64_t obj;
1473 
1474 	ASSERT(ds->ds_owner);
1475 	ASSERT(dsda->defer || ds->ds_phys->ds_num_children <= 1);
1476 	ASSERT(ds->ds_prev == NULL ||
1477 	    ds->ds_prev->ds_phys->ds_next_snap_obj != ds->ds_object);
1478 	ASSERT3U(ds->ds_phys->ds_bp.blk_birth, <=, tx->tx_txg);
1479 
1480 	if (dsda->defer) {
1481 		ASSERT(spa_version(dp->dp_spa) >= SPA_VERSION_USERREFS);
1482 		if (ds->ds_userrefs > 0 || ds->ds_phys->ds_num_children > 1) {
1483 			dmu_buf_will_dirty(ds->ds_dbuf, tx);
1484 			ds->ds_phys->ds_flags |= DS_FLAG_DEFER_DESTROY;
1485 			return;
1486 		}
1487 	}
1488 
1489 	/* signal any waiters that this dataset is going away */
1490 	mutex_enter(&ds->ds_lock);
1491 	ds->ds_owner = dsl_reaper;
1492 	cv_broadcast(&ds->ds_exclusive_cv);
1493 	mutex_exit(&ds->ds_lock);
1494 
1495 	/* Remove our reservation */
1496 	if (ds->ds_reserved != 0) {
1497 		uint64_t val = 0;
1498 		dsl_dataset_set_reservation_sync(ds, &val, cr, tx);
1499 		ASSERT3U(ds->ds_reserved, ==, 0);
1500 	}
1501 
1502 	ASSERT(RW_WRITE_HELD(&dp->dp_config_rwlock));
1503 
1504 	dsl_pool_ds_destroyed(ds, tx);
1505 
1506 	obj = ds->ds_object;
1507 
1508 	if (ds->ds_phys->ds_prev_snap_obj != 0) {
1509 		if (ds->ds_prev) {
1510 			ds_prev = ds->ds_prev;
1511 		} else {
1512 			VERIFY(0 == dsl_dataset_hold_obj(dp,
1513 			    ds->ds_phys->ds_prev_snap_obj, FTAG, &ds_prev));
1514 		}
1515 		after_branch_point =
1516 		    (ds_prev->ds_phys->ds_next_snap_obj != obj);
1517 
1518 		dmu_buf_will_dirty(ds_prev->ds_dbuf, tx);
1519 		if (after_branch_point &&
1520 		    ds_prev->ds_phys->ds_next_clones_obj != 0) {
1521 			VERIFY3U(0, ==, zap_remove_int(mos,
1522 			    ds_prev->ds_phys->ds_next_clones_obj, obj, tx));
1523 			if (ds->ds_phys->ds_next_snap_obj != 0) {
1524 				VERIFY(0 == zap_add_int(mos,
1525 				    ds_prev->ds_phys->ds_next_clones_obj,
1526 				    ds->ds_phys->ds_next_snap_obj, tx));
1527 			}
1528 		}
1529 		if (after_branch_point &&
1530 		    ds->ds_phys->ds_next_snap_obj == 0) {
1531 			/* This clone is toast. */
1532 			ASSERT(ds_prev->ds_phys->ds_num_children > 1);
1533 			ds_prev->ds_phys->ds_num_children--;
1534 
1535 			/*
1536 			 * If the clone's origin has no other clones, no
1537 			 * user holds, and has been marked for deferred
1538 			 * deletion, then we should have done the necessary
1539 			 * destroy setup for it.
1540 			 */
1541 			if (ds_prev->ds_phys->ds_num_children == 1 &&
1542 			    ds_prev->ds_userrefs == 0 &&
1543 			    DS_IS_DEFER_DESTROY(ds_prev)) {
1544 				ASSERT3P(dsda->rm_origin, !=, NULL);
1545 			} else {
1546 				ASSERT3P(dsda->rm_origin, ==, NULL);
1547 			}
1548 		} else if (!after_branch_point) {
1549 			ds_prev->ds_phys->ds_next_snap_obj =
1550 			    ds->ds_phys->ds_next_snap_obj;
1551 		}
1552 	}
1553 
1554 	zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
1555 
1556 	if (ds->ds_phys->ds_next_snap_obj != 0) {
1557 		blkptr_t bp;
1558 		dsl_dataset_t *ds_next;
1559 		uint64_t itor = 0;
1560 		uint64_t old_unique;
1561 		int64_t used = 0, compressed = 0, uncompressed = 0;
1562 
1563 		VERIFY(0 == dsl_dataset_hold_obj(dp,
1564 		    ds->ds_phys->ds_next_snap_obj, FTAG, &ds_next));
1565 		ASSERT3U(ds_next->ds_phys->ds_prev_snap_obj, ==, obj);
1566 
1567 		old_unique = dsl_dataset_unique(ds_next);
1568 
1569 		dmu_buf_will_dirty(ds_next->ds_dbuf, tx);
1570 		ds_next->ds_phys->ds_prev_snap_obj =
1571 		    ds->ds_phys->ds_prev_snap_obj;
1572 		ds_next->ds_phys->ds_prev_snap_txg =
1573 		    ds->ds_phys->ds_prev_snap_txg;
1574 		ASSERT3U(ds->ds_phys->ds_prev_snap_txg, ==,
1575 		    ds_prev ? ds_prev->ds_phys->ds_creation_txg : 0);
1576 
1577 		/*
1578 		 * Transfer to our deadlist (which will become next's
1579 		 * new deadlist) any entries from next's current
1580 		 * deadlist which were born before prev, and free the
1581 		 * other entries.
1582 		 *
1583 		 * XXX we're doing this long task with the config lock held
1584 		 */
1585 		while (bplist_iterate(&ds_next->ds_deadlist, &itor, &bp) == 0) {
1586 			if (bp.blk_birth <= ds->ds_phys->ds_prev_snap_txg) {
1587 				VERIFY(0 == bplist_enqueue(&ds->ds_deadlist,
1588 				    &bp, tx));
1589 				if (ds_prev && !after_branch_point &&
1590 				    bp.blk_birth >
1591 				    ds_prev->ds_phys->ds_prev_snap_txg) {
1592 					ds_prev->ds_phys->ds_unique_bytes +=
1593 					    bp_get_dasize(dp->dp_spa, &bp);
1594 				}
1595 			} else {
1596 				used += bp_get_dasize(dp->dp_spa, &bp);
1597 				compressed += BP_GET_PSIZE(&bp);
1598 				uncompressed += BP_GET_UCSIZE(&bp);
1599 				/* XXX check return value? */
1600 				(void) dsl_free(zio, dp, tx->tx_txg,
1601 				    &bp, NULL, NULL, ARC_NOWAIT);
1602 			}
1603 		}
1604 
1605 		ASSERT3U(used, ==, ds->ds_phys->ds_unique_bytes);
1606 
1607 		/* change snapused */
1608 		dsl_dir_diduse_space(ds->ds_dir, DD_USED_SNAP,
1609 		    -used, -compressed, -uncompressed, tx);
1610 
1611 		/* free next's deadlist */
1612 		bplist_close(&ds_next->ds_deadlist);
1613 		bplist_destroy(mos, ds_next->ds_phys->ds_deadlist_obj, tx);
1614 
1615 		/* set next's deadlist to our deadlist */
1616 		bplist_close(&ds->ds_deadlist);
1617 		ds_next->ds_phys->ds_deadlist_obj =
1618 		    ds->ds_phys->ds_deadlist_obj;
1619 		VERIFY(0 == bplist_open(&ds_next->ds_deadlist, mos,
1620 		    ds_next->ds_phys->ds_deadlist_obj));
1621 		ds->ds_phys->ds_deadlist_obj = 0;
1622 
1623 		if (ds_next->ds_phys->ds_next_snap_obj != 0) {
1624 			/*
1625 			 * Update next's unique to include blocks which
1626 			 * were previously shared by only this snapshot
1627 			 * and it.  Those blocks will be born after the
1628 			 * prev snap and before this snap, and will have
1629 			 * died after the next snap and before the one
1630 			 * after that (ie. be on the snap after next's
1631 			 * deadlist).
1632 			 *
1633 			 * XXX we're doing this long task with the
1634 			 * config lock held
1635 			 */
1636 			dsl_dataset_t *ds_after_next;
1637 			uint64_t space;
1638 
1639 			VERIFY(0 == dsl_dataset_hold_obj(dp,
1640 			    ds_next->ds_phys->ds_next_snap_obj,
1641 			    FTAG, &ds_after_next));
1642 
1643 			VERIFY(0 ==
1644 			    bplist_space_birthrange(&ds_after_next->ds_deadlist,
1645 			    ds->ds_phys->ds_prev_snap_txg,
1646 			    ds->ds_phys->ds_creation_txg, &space));
1647 			ds_next->ds_phys->ds_unique_bytes += space;
1648 
1649 			dsl_dataset_rele(ds_after_next, FTAG);
1650 			ASSERT3P(ds_next->ds_prev, ==, NULL);
1651 		} else {
1652 			ASSERT3P(ds_next->ds_prev, ==, ds);
1653 			dsl_dataset_drop_ref(ds_next->ds_prev, ds_next);
1654 			ds_next->ds_prev = NULL;
1655 			if (ds_prev) {
1656 				VERIFY(0 == dsl_dataset_get_ref(dp,
1657 				    ds->ds_phys->ds_prev_snap_obj,
1658 				    ds_next, &ds_next->ds_prev));
1659 			}
1660 
1661 			dsl_dataset_recalc_head_uniq(ds_next);
1662 
1663 			/*
1664 			 * Reduce the amount of our unconsmed refreservation
1665 			 * being charged to our parent by the amount of
1666 			 * new unique data we have gained.
1667 			 */
1668 			if (old_unique < ds_next->ds_reserved) {
1669 				int64_t mrsdelta;
1670 				uint64_t new_unique =
1671 				    ds_next->ds_phys->ds_unique_bytes;
1672 
1673 				ASSERT(old_unique <= new_unique);
1674 				mrsdelta = MIN(new_unique - old_unique,
1675 				    ds_next->ds_reserved - old_unique);
1676 				dsl_dir_diduse_space(ds->ds_dir,
1677 				    DD_USED_REFRSRV, -mrsdelta, 0, 0, tx);
1678 			}
1679 		}
1680 		dsl_dataset_rele(ds_next, FTAG);
1681 	} else {
1682 		/*
1683 		 * There's no next snapshot, so this is a head dataset.
1684 		 * Destroy the deadlist.  Unless it's a clone, the
1685 		 * deadlist should be empty.  (If it's a clone, it's
1686 		 * safe to ignore the deadlist contents.)
1687 		 */
1688 		struct killarg ka;
1689 
1690 		ASSERT(after_branch_point || bplist_empty(&ds->ds_deadlist));
1691 		bplist_close(&ds->ds_deadlist);
1692 		bplist_destroy(mos, ds->ds_phys->ds_deadlist_obj, tx);
1693 		ds->ds_phys->ds_deadlist_obj = 0;
1694 
1695 		/*
1696 		 * Free everything that we point to (that's born after
1697 		 * the previous snapshot, if we are a clone)
1698 		 *
1699 		 * NB: this should be very quick, because we already
1700 		 * freed all the objects in open context.
1701 		 */
1702 		ka.ds = ds;
1703 		ka.zio = zio;
1704 		ka.tx = tx;
1705 		err = traverse_dataset(ds, ds->ds_phys->ds_prev_snap_txg,
1706 		    TRAVERSE_POST, kill_blkptr, &ka);
1707 		ASSERT3U(err, ==, 0);
1708 		ASSERT(!DS_UNIQUE_IS_ACCURATE(ds) ||
1709 		    ds->ds_phys->ds_unique_bytes == 0);
1710 
1711 		if (ds->ds_prev != NULL) {
1712 			dsl_dataset_rele(ds->ds_prev, ds);
1713 			ds->ds_prev = ds_prev = NULL;
1714 		}
1715 	}
1716 
1717 	err = zio_wait(zio);
1718 	ASSERT3U(err, ==, 0);
1719 
1720 	if (ds->ds_dir->dd_phys->dd_head_dataset_obj == ds->ds_object) {
1721 		/* Erase the link in the dir */
1722 		dmu_buf_will_dirty(ds->ds_dir->dd_dbuf, tx);
1723 		ds->ds_dir->dd_phys->dd_head_dataset_obj = 0;
1724 		ASSERT(ds->ds_phys->ds_snapnames_zapobj != 0);
1725 		err = zap_destroy(mos, ds->ds_phys->ds_snapnames_zapobj, tx);
1726 		ASSERT(err == 0);
1727 	} else {
1728 		/* remove from snapshot namespace */
1729 		dsl_dataset_t *ds_head;
1730 		ASSERT(ds->ds_phys->ds_snapnames_zapobj == 0);
1731 		VERIFY(0 == dsl_dataset_hold_obj(dp,
1732 		    ds->ds_dir->dd_phys->dd_head_dataset_obj, FTAG, &ds_head));
1733 		VERIFY(0 == dsl_dataset_get_snapname(ds));
1734 #ifdef ZFS_DEBUG
1735 		{
1736 			uint64_t val;
1737 
1738 			err = dsl_dataset_snap_lookup(ds_head,
1739 			    ds->ds_snapname, &val);
1740 			ASSERT3U(err, ==, 0);
1741 			ASSERT3U(val, ==, obj);
1742 		}
1743 #endif
1744 		err = dsl_dataset_snap_remove(ds_head, ds->ds_snapname, tx);
1745 		ASSERT(err == 0);
1746 		dsl_dataset_rele(ds_head, FTAG);
1747 	}
1748 
1749 	if (ds_prev && ds->ds_prev != ds_prev)
1750 		dsl_dataset_rele(ds_prev, FTAG);
1751 
1752 	spa_prop_clear_bootfs(dp->dp_spa, ds->ds_object, tx);
1753 	spa_history_internal_log(LOG_DS_DESTROY, dp->dp_spa, tx,
1754 	    cr, "dataset = %llu", ds->ds_object);
1755 
1756 	if (ds->ds_phys->ds_next_clones_obj != 0) {
1757 		uint64_t count;
1758 		ASSERT(0 == zap_count(mos,
1759 		    ds->ds_phys->ds_next_clones_obj, &count) && count == 0);
1760 		VERIFY(0 == dmu_object_free(mos,
1761 		    ds->ds_phys->ds_next_clones_obj, tx));
1762 	}
1763 	if (ds->ds_phys->ds_props_obj != 0)
1764 		VERIFY(0 == zap_destroy(mos, ds->ds_phys->ds_props_obj, tx));
1765 	if (ds->ds_phys->ds_userrefs_obj != 0)
1766 		VERIFY(0 == zap_destroy(mos, ds->ds_phys->ds_userrefs_obj, tx));
1767 	dsl_dir_close(ds->ds_dir, ds);
1768 	ds->ds_dir = NULL;
1769 	dsl_dataset_drain_refs(ds, tag);
1770 	VERIFY(0 == dmu_object_free(mos, obj, tx));
1771 
1772 	if (dsda->rm_origin) {
1773 		/*
1774 		 * Remove the origin of the clone we just destroyed.
1775 		 */
1776 		struct dsl_ds_destroyarg ndsda = {0};
1777 
1778 		ndsda.ds = dsda->rm_origin;
1779 		dsl_dataset_destroy_sync(&ndsda, tag, cr, tx);
1780 	}
1781 }
1782 
1783 static int
1784 dsl_dataset_snapshot_reserve_space(dsl_dataset_t *ds, dmu_tx_t *tx)
1785 {
1786 	uint64_t asize;
1787 
1788 	if (!dmu_tx_is_syncing(tx))
1789 		return (0);
1790 
1791 	/*
1792 	 * If there's an fs-only reservation, any blocks that might become
1793 	 * owned by the snapshot dataset must be accommodated by space
1794 	 * outside of the reservation.
1795 	 */
1796 	asize = MIN(dsl_dataset_unique(ds), ds->ds_reserved);
1797 	if (asize > dsl_dir_space_available(ds->ds_dir, NULL, 0, FALSE))
1798 		return (ENOSPC);
1799 
1800 	/*
1801 	 * Propogate any reserved space for this snapshot to other
1802 	 * snapshot checks in this sync group.
1803 	 */
1804 	if (asize > 0)
1805 		dsl_dir_willuse_space(ds->ds_dir, asize, tx);
1806 
1807 	return (0);
1808 }
1809 
1810 /* ARGSUSED */
1811 int
1812 dsl_dataset_snapshot_check(void *arg1, void *arg2, dmu_tx_t *tx)
1813 {
1814 	dsl_dataset_t *ds = arg1;
1815 	const char *snapname = arg2;
1816 	int err;
1817 	uint64_t value;
1818 
1819 	/*
1820 	 * We don't allow multiple snapshots of the same txg.  If there
1821 	 * is already one, try again.
1822 	 */
1823 	if (ds->ds_phys->ds_prev_snap_txg >= tx->tx_txg)
1824 		return (EAGAIN);
1825 
1826 	/*
1827 	 * Check for conflicting name snapshot name.
1828 	 */
1829 	err = dsl_dataset_snap_lookup(ds, snapname, &value);
1830 	if (err == 0)
1831 		return (EEXIST);
1832 	if (err != ENOENT)
1833 		return (err);
1834 
1835 	/*
1836 	 * Check that the dataset's name is not too long.  Name consists
1837 	 * of the dataset's length + 1 for the @-sign + snapshot name's length
1838 	 */
1839 	if (dsl_dataset_namelen(ds) + 1 + strlen(snapname) >= MAXNAMELEN)
1840 		return (ENAMETOOLONG);
1841 
1842 	err = dsl_dataset_snapshot_reserve_space(ds, tx);
1843 	if (err)
1844 		return (err);
1845 
1846 	ds->ds_trysnap_txg = tx->tx_txg;
1847 	return (0);
1848 }
1849 
1850 void
1851 dsl_dataset_snapshot_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
1852 {
1853 	dsl_dataset_t *ds = arg1;
1854 	const char *snapname = arg2;
1855 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1856 	dmu_buf_t *dbuf;
1857 	dsl_dataset_phys_t *dsphys;
1858 	uint64_t dsobj, crtxg;
1859 	objset_t *mos = dp->dp_meta_objset;
1860 	int err;
1861 
1862 	ASSERT(RW_WRITE_HELD(&dp->dp_config_rwlock));
1863 
1864 	/*
1865 	 * The origin's ds_creation_txg has to be < TXG_INITIAL
1866 	 */
1867 	if (strcmp(snapname, ORIGIN_DIR_NAME) == 0)
1868 		crtxg = 1;
1869 	else
1870 		crtxg = tx->tx_txg;
1871 
1872 	dsobj = dmu_object_alloc(mos, DMU_OT_DSL_DATASET, 0,
1873 	    DMU_OT_DSL_DATASET, sizeof (dsl_dataset_phys_t), tx);
1874 	VERIFY(0 == dmu_bonus_hold(mos, dsobj, FTAG, &dbuf));
1875 	dmu_buf_will_dirty(dbuf, tx);
1876 	dsphys = dbuf->db_data;
1877 	bzero(dsphys, sizeof (dsl_dataset_phys_t));
1878 	dsphys->ds_dir_obj = ds->ds_dir->dd_object;
1879 	dsphys->ds_fsid_guid = unique_create();
1880 	(void) random_get_pseudo_bytes((void*)&dsphys->ds_guid,
1881 	    sizeof (dsphys->ds_guid));
1882 	dsphys->ds_prev_snap_obj = ds->ds_phys->ds_prev_snap_obj;
1883 	dsphys->ds_prev_snap_txg = ds->ds_phys->ds_prev_snap_txg;
1884 	dsphys->ds_next_snap_obj = ds->ds_object;
1885 	dsphys->ds_num_children = 1;
1886 	dsphys->ds_creation_time = gethrestime_sec();
1887 	dsphys->ds_creation_txg = crtxg;
1888 	dsphys->ds_deadlist_obj = ds->ds_phys->ds_deadlist_obj;
1889 	dsphys->ds_used_bytes = ds->ds_phys->ds_used_bytes;
1890 	dsphys->ds_compressed_bytes = ds->ds_phys->ds_compressed_bytes;
1891 	dsphys->ds_uncompressed_bytes = ds->ds_phys->ds_uncompressed_bytes;
1892 	dsphys->ds_flags = ds->ds_phys->ds_flags;
1893 	dsphys->ds_bp = ds->ds_phys->ds_bp;
1894 	dmu_buf_rele(dbuf, FTAG);
1895 
1896 	ASSERT3U(ds->ds_prev != 0, ==, ds->ds_phys->ds_prev_snap_obj != 0);
1897 	if (ds->ds_prev) {
1898 		uint64_t next_clones_obj =
1899 		    ds->ds_prev->ds_phys->ds_next_clones_obj;
1900 		ASSERT(ds->ds_prev->ds_phys->ds_next_snap_obj ==
1901 		    ds->ds_object ||
1902 		    ds->ds_prev->ds_phys->ds_num_children > 1);
1903 		if (ds->ds_prev->ds_phys->ds_next_snap_obj == ds->ds_object) {
1904 			dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
1905 			ASSERT3U(ds->ds_phys->ds_prev_snap_txg, ==,
1906 			    ds->ds_prev->ds_phys->ds_creation_txg);
1907 			ds->ds_prev->ds_phys->ds_next_snap_obj = dsobj;
1908 		} else if (next_clones_obj != 0) {
1909 			VERIFY3U(0, ==, zap_remove_int(mos,
1910 			    next_clones_obj, dsphys->ds_next_snap_obj, tx));
1911 			VERIFY3U(0, ==, zap_add_int(mos,
1912 			    next_clones_obj, dsobj, tx));
1913 		}
1914 	}
1915 
1916 	/*
1917 	 * If we have a reference-reservation on this dataset, we will
1918 	 * need to increase the amount of refreservation being charged
1919 	 * since our unique space is going to zero.
1920 	 */
1921 	if (ds->ds_reserved) {
1922 		int64_t add = MIN(dsl_dataset_unique(ds), ds->ds_reserved);
1923 		dsl_dir_diduse_space(ds->ds_dir, DD_USED_REFRSRV,
1924 		    add, 0, 0, tx);
1925 	}
1926 
1927 	bplist_close(&ds->ds_deadlist);
1928 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
1929 	ASSERT3U(ds->ds_phys->ds_prev_snap_txg, <, tx->tx_txg);
1930 	ds->ds_phys->ds_prev_snap_obj = dsobj;
1931 	ds->ds_phys->ds_prev_snap_txg = crtxg;
1932 	ds->ds_phys->ds_unique_bytes = 0;
1933 	if (spa_version(dp->dp_spa) >= SPA_VERSION_UNIQUE_ACCURATE)
1934 		ds->ds_phys->ds_flags |= DS_FLAG_UNIQUE_ACCURATE;
1935 	ds->ds_phys->ds_deadlist_obj =
1936 	    bplist_create(mos, DSL_DEADLIST_BLOCKSIZE, tx);
1937 	VERIFY(0 == bplist_open(&ds->ds_deadlist, mos,
1938 	    ds->ds_phys->ds_deadlist_obj));
1939 
1940 	dprintf("snap '%s' -> obj %llu\n", snapname, dsobj);
1941 	err = zap_add(mos, ds->ds_phys->ds_snapnames_zapobj,
1942 	    snapname, 8, 1, &dsobj, tx);
1943 	ASSERT(err == 0);
1944 
1945 	if (ds->ds_prev)
1946 		dsl_dataset_drop_ref(ds->ds_prev, ds);
1947 	VERIFY(0 == dsl_dataset_get_ref(dp,
1948 	    ds->ds_phys->ds_prev_snap_obj, ds, &ds->ds_prev));
1949 
1950 	dsl_pool_ds_snapshotted(ds, tx);
1951 
1952 	dsl_dir_snap_cmtime_update(ds->ds_dir);
1953 
1954 	spa_history_internal_log(LOG_DS_SNAPSHOT, dp->dp_spa, tx, cr,
1955 	    "dataset = %llu", dsobj);
1956 }
1957 
1958 void
1959 dsl_dataset_sync(dsl_dataset_t *ds, zio_t *zio, dmu_tx_t *tx)
1960 {
1961 	ASSERT(dmu_tx_is_syncing(tx));
1962 	ASSERT(ds->ds_objset != NULL);
1963 	ASSERT(ds->ds_phys->ds_next_snap_obj == 0);
1964 
1965 	/*
1966 	 * in case we had to change ds_fsid_guid when we opened it,
1967 	 * sync it out now.
1968 	 */
1969 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
1970 	ds->ds_phys->ds_fsid_guid = ds->ds_fsid_guid;
1971 
1972 	dsl_dir_dirty(ds->ds_dir, tx);
1973 	dmu_objset_sync(ds->ds_objset, zio, tx);
1974 }
1975 
1976 void
1977 dsl_dataset_stats(dsl_dataset_t *ds, nvlist_t *nv)
1978 {
1979 	uint64_t refd, avail, uobjs, aobjs;
1980 
1981 	dsl_dir_stats(ds->ds_dir, nv);
1982 
1983 	dsl_dataset_space(ds, &refd, &avail, &uobjs, &aobjs);
1984 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_AVAILABLE, avail);
1985 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFERENCED, refd);
1986 
1987 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_CREATION,
1988 	    ds->ds_phys->ds_creation_time);
1989 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_CREATETXG,
1990 	    ds->ds_phys->ds_creation_txg);
1991 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFQUOTA,
1992 	    ds->ds_quota);
1993 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_REFRESERVATION,
1994 	    ds->ds_reserved);
1995 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_GUID,
1996 	    ds->ds_phys->ds_guid);
1997 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_UNIQUE,
1998 	    dsl_dataset_unique(ds));
1999 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_OBJSETID,
2000 	    ds->ds_object);
2001 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USERREFS, ds->ds_userrefs);
2002 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_DEFER_DESTROY,
2003 	    DS_IS_DEFER_DESTROY(ds) ? 1 : 0);
2004 
2005 	if (ds->ds_phys->ds_next_snap_obj) {
2006 		/*
2007 		 * This is a snapshot; override the dd's space used with
2008 		 * our unique space and compression ratio.
2009 		 */
2010 		dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USED,
2011 		    ds->ds_phys->ds_unique_bytes);
2012 		dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_COMPRESSRATIO,
2013 		    ds->ds_phys->ds_compressed_bytes == 0 ? 100 :
2014 		    (ds->ds_phys->ds_uncompressed_bytes * 100 /
2015 		    ds->ds_phys->ds_compressed_bytes));
2016 	}
2017 }
2018 
2019 void
2020 dsl_dataset_fast_stat(dsl_dataset_t *ds, dmu_objset_stats_t *stat)
2021 {
2022 	stat->dds_creation_txg = ds->ds_phys->ds_creation_txg;
2023 	stat->dds_inconsistent = ds->ds_phys->ds_flags & DS_FLAG_INCONSISTENT;
2024 	stat->dds_guid = ds->ds_phys->ds_guid;
2025 	if (ds->ds_phys->ds_next_snap_obj) {
2026 		stat->dds_is_snapshot = B_TRUE;
2027 		stat->dds_num_clones = ds->ds_phys->ds_num_children - 1;
2028 	} else {
2029 		stat->dds_is_snapshot = B_FALSE;
2030 		stat->dds_num_clones = 0;
2031 	}
2032 
2033 	/* clone origin is really a dsl_dir thing... */
2034 	rw_enter(&ds->ds_dir->dd_pool->dp_config_rwlock, RW_READER);
2035 	if (dsl_dir_is_clone(ds->ds_dir)) {
2036 		dsl_dataset_t *ods;
2037 
2038 		VERIFY(0 == dsl_dataset_get_ref(ds->ds_dir->dd_pool,
2039 		    ds->ds_dir->dd_phys->dd_origin_obj, FTAG, &ods));
2040 		dsl_dataset_name(ods, stat->dds_origin);
2041 		dsl_dataset_drop_ref(ods, FTAG);
2042 	} else {
2043 		stat->dds_origin[0] = '\0';
2044 	}
2045 	rw_exit(&ds->ds_dir->dd_pool->dp_config_rwlock);
2046 }
2047 
2048 uint64_t
2049 dsl_dataset_fsid_guid(dsl_dataset_t *ds)
2050 {
2051 	return (ds->ds_fsid_guid);
2052 }
2053 
2054 void
2055 dsl_dataset_space(dsl_dataset_t *ds,
2056     uint64_t *refdbytesp, uint64_t *availbytesp,
2057     uint64_t *usedobjsp, uint64_t *availobjsp)
2058 {
2059 	*refdbytesp = ds->ds_phys->ds_used_bytes;
2060 	*availbytesp = dsl_dir_space_available(ds->ds_dir, NULL, 0, TRUE);
2061 	if (ds->ds_reserved > ds->ds_phys->ds_unique_bytes)
2062 		*availbytesp += ds->ds_reserved - ds->ds_phys->ds_unique_bytes;
2063 	if (ds->ds_quota != 0) {
2064 		/*
2065 		 * Adjust available bytes according to refquota
2066 		 */
2067 		if (*refdbytesp < ds->ds_quota)
2068 			*availbytesp = MIN(*availbytesp,
2069 			    ds->ds_quota - *refdbytesp);
2070 		else
2071 			*availbytesp = 0;
2072 	}
2073 	*usedobjsp = ds->ds_phys->ds_bp.blk_fill;
2074 	*availobjsp = DN_MAX_OBJECT - *usedobjsp;
2075 }
2076 
2077 boolean_t
2078 dsl_dataset_modified_since_lastsnap(dsl_dataset_t *ds)
2079 {
2080 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
2081 
2082 	ASSERT(RW_LOCK_HELD(&dp->dp_config_rwlock) ||
2083 	    dsl_pool_sync_context(dp));
2084 	if (ds->ds_prev == NULL)
2085 		return (B_FALSE);
2086 	if (ds->ds_phys->ds_bp.blk_birth >
2087 	    ds->ds_prev->ds_phys->ds_creation_txg)
2088 		return (B_TRUE);
2089 	return (B_FALSE);
2090 }
2091 
2092 /* ARGSUSED */
2093 static int
2094 dsl_dataset_snapshot_rename_check(void *arg1, void *arg2, dmu_tx_t *tx)
2095 {
2096 	dsl_dataset_t *ds = arg1;
2097 	char *newsnapname = arg2;
2098 	dsl_dir_t *dd = ds->ds_dir;
2099 	dsl_dataset_t *hds;
2100 	uint64_t val;
2101 	int err;
2102 
2103 	err = dsl_dataset_hold_obj(dd->dd_pool,
2104 	    dd->dd_phys->dd_head_dataset_obj, FTAG, &hds);
2105 	if (err)
2106 		return (err);
2107 
2108 	/* new name better not be in use */
2109 	err = dsl_dataset_snap_lookup(hds, newsnapname, &val);
2110 	dsl_dataset_rele(hds, FTAG);
2111 
2112 	if (err == 0)
2113 		err = EEXIST;
2114 	else if (err == ENOENT)
2115 		err = 0;
2116 
2117 	/* dataset name + 1 for the "@" + the new snapshot name must fit */
2118 	if (dsl_dir_namelen(ds->ds_dir) + 1 + strlen(newsnapname) >= MAXNAMELEN)
2119 		err = ENAMETOOLONG;
2120 
2121 	return (err);
2122 }
2123 
2124 static void
2125 dsl_dataset_snapshot_rename_sync(void *arg1, void *arg2,
2126     cred_t *cr, dmu_tx_t *tx)
2127 {
2128 	dsl_dataset_t *ds = arg1;
2129 	const char *newsnapname = arg2;
2130 	dsl_dir_t *dd = ds->ds_dir;
2131 	objset_t *mos = dd->dd_pool->dp_meta_objset;
2132 	dsl_dataset_t *hds;
2133 	int err;
2134 
2135 	ASSERT(ds->ds_phys->ds_next_snap_obj != 0);
2136 
2137 	VERIFY(0 == dsl_dataset_hold_obj(dd->dd_pool,
2138 	    dd->dd_phys->dd_head_dataset_obj, FTAG, &hds));
2139 
2140 	VERIFY(0 == dsl_dataset_get_snapname(ds));
2141 	err = dsl_dataset_snap_remove(hds, ds->ds_snapname, tx);
2142 	ASSERT3U(err, ==, 0);
2143 	mutex_enter(&ds->ds_lock);
2144 	(void) strcpy(ds->ds_snapname, newsnapname);
2145 	mutex_exit(&ds->ds_lock);
2146 	err = zap_add(mos, hds->ds_phys->ds_snapnames_zapobj,
2147 	    ds->ds_snapname, 8, 1, &ds->ds_object, tx);
2148 	ASSERT3U(err, ==, 0);
2149 
2150 	spa_history_internal_log(LOG_DS_RENAME, dd->dd_pool->dp_spa, tx,
2151 	    cr, "dataset = %llu", ds->ds_object);
2152 	dsl_dataset_rele(hds, FTAG);
2153 }
2154 
2155 struct renamesnaparg {
2156 	dsl_sync_task_group_t *dstg;
2157 	char failed[MAXPATHLEN];
2158 	char *oldsnap;
2159 	char *newsnap;
2160 };
2161 
2162 static int
2163 dsl_snapshot_rename_one(char *name, void *arg)
2164 {
2165 	struct renamesnaparg *ra = arg;
2166 	dsl_dataset_t *ds = NULL;
2167 	char *cp;
2168 	int err;
2169 
2170 	cp = name + strlen(name);
2171 	*cp = '@';
2172 	(void) strcpy(cp + 1, ra->oldsnap);
2173 
2174 	/*
2175 	 * For recursive snapshot renames the parent won't be changing
2176 	 * so we just pass name for both the to/from argument.
2177 	 */
2178 	err = zfs_secpolicy_rename_perms(name, name, CRED());
2179 	if (err == ENOENT) {
2180 		return (0);
2181 	} else if (err) {
2182 		(void) strcpy(ra->failed, name);
2183 		return (err);
2184 	}
2185 
2186 #ifdef _KERNEL
2187 	/*
2188 	 * For all filesystems undergoing rename, we'll need to unmount it.
2189 	 */
2190 	(void) zfs_unmount_snap(name, NULL);
2191 #endif
2192 	err = dsl_dataset_hold(name, ra->dstg, &ds);
2193 	*cp = '\0';
2194 	if (err == ENOENT) {
2195 		return (0);
2196 	} else if (err) {
2197 		(void) strcpy(ra->failed, name);
2198 		return (err);
2199 	}
2200 
2201 	dsl_sync_task_create(ra->dstg, dsl_dataset_snapshot_rename_check,
2202 	    dsl_dataset_snapshot_rename_sync, ds, ra->newsnap, 0);
2203 
2204 	return (0);
2205 }
2206 
2207 static int
2208 dsl_recursive_rename(char *oldname, const char *newname)
2209 {
2210 	int err;
2211 	struct renamesnaparg *ra;
2212 	dsl_sync_task_t *dst;
2213 	spa_t *spa;
2214 	char *cp, *fsname = spa_strdup(oldname);
2215 	int len = strlen(oldname);
2216 
2217 	/* truncate the snapshot name to get the fsname */
2218 	cp = strchr(fsname, '@');
2219 	*cp = '\0';
2220 
2221 	err = spa_open(fsname, &spa, FTAG);
2222 	if (err) {
2223 		kmem_free(fsname, len + 1);
2224 		return (err);
2225 	}
2226 	ra = kmem_alloc(sizeof (struct renamesnaparg), KM_SLEEP);
2227 	ra->dstg = dsl_sync_task_group_create(spa_get_dsl(spa));
2228 
2229 	ra->oldsnap = strchr(oldname, '@') + 1;
2230 	ra->newsnap = strchr(newname, '@') + 1;
2231 	*ra->failed = '\0';
2232 
2233 	err = dmu_objset_find(fsname, dsl_snapshot_rename_one, ra,
2234 	    DS_FIND_CHILDREN);
2235 	kmem_free(fsname, len + 1);
2236 
2237 	if (err == 0) {
2238 		err = dsl_sync_task_group_wait(ra->dstg);
2239 	}
2240 
2241 	for (dst = list_head(&ra->dstg->dstg_tasks); dst;
2242 	    dst = list_next(&ra->dstg->dstg_tasks, dst)) {
2243 		dsl_dataset_t *ds = dst->dst_arg1;
2244 		if (dst->dst_err) {
2245 			dsl_dir_name(ds->ds_dir, ra->failed);
2246 			(void) strcat(ra->failed, "@");
2247 			(void) strcat(ra->failed, ra->newsnap);
2248 		}
2249 		dsl_dataset_rele(ds, ra->dstg);
2250 	}
2251 
2252 	if (err)
2253 		(void) strcpy(oldname, ra->failed);
2254 
2255 	dsl_sync_task_group_destroy(ra->dstg);
2256 	kmem_free(ra, sizeof (struct renamesnaparg));
2257 	spa_close(spa, FTAG);
2258 	return (err);
2259 }
2260 
2261 static int
2262 dsl_valid_rename(char *oldname, void *arg)
2263 {
2264 	int delta = *(int *)arg;
2265 
2266 	if (strlen(oldname) + delta >= MAXNAMELEN)
2267 		return (ENAMETOOLONG);
2268 
2269 	return (0);
2270 }
2271 
2272 #pragma weak dmu_objset_rename = dsl_dataset_rename
2273 int
2274 dsl_dataset_rename(char *oldname, const char *newname, boolean_t recursive)
2275 {
2276 	dsl_dir_t *dd;
2277 	dsl_dataset_t *ds;
2278 	const char *tail;
2279 	int err;
2280 
2281 	err = dsl_dir_open(oldname, FTAG, &dd, &tail);
2282 	if (err)
2283 		return (err);
2284 	/*
2285 	 * If there are more than 2 references there may be holds
2286 	 * hanging around that haven't been cleared out yet.
2287 	 */
2288 	if (dmu_buf_refcount(dd->dd_dbuf) > 2)
2289 		txg_wait_synced(dd->dd_pool, 0);
2290 	if (tail == NULL) {
2291 		int delta = strlen(newname) - strlen(oldname);
2292 
2293 		/* if we're growing, validate child name lengths */
2294 		if (delta > 0)
2295 			err = dmu_objset_find(oldname, dsl_valid_rename,
2296 			    &delta, DS_FIND_CHILDREN | DS_FIND_SNAPSHOTS);
2297 
2298 		if (!err)
2299 			err = dsl_dir_rename(dd, newname);
2300 		dsl_dir_close(dd, FTAG);
2301 		return (err);
2302 	}
2303 	if (tail[0] != '@') {
2304 		/* the name ended in a nonexistent component */
2305 		dsl_dir_close(dd, FTAG);
2306 		return (ENOENT);
2307 	}
2308 
2309 	dsl_dir_close(dd, FTAG);
2310 
2311 	/* new name must be snapshot in same filesystem */
2312 	tail = strchr(newname, '@');
2313 	if (tail == NULL)
2314 		return (EINVAL);
2315 	tail++;
2316 	if (strncmp(oldname, newname, tail - newname) != 0)
2317 		return (EXDEV);
2318 
2319 	if (recursive) {
2320 		err = dsl_recursive_rename(oldname, newname);
2321 	} else {
2322 		err = dsl_dataset_hold(oldname, FTAG, &ds);
2323 		if (err)
2324 			return (err);
2325 
2326 		err = dsl_sync_task_do(ds->ds_dir->dd_pool,
2327 		    dsl_dataset_snapshot_rename_check,
2328 		    dsl_dataset_snapshot_rename_sync, ds, (char *)tail, 1);
2329 
2330 		dsl_dataset_rele(ds, FTAG);
2331 	}
2332 
2333 	return (err);
2334 }
2335 
2336 struct promotenode {
2337 	list_node_t link;
2338 	dsl_dataset_t *ds;
2339 };
2340 
2341 struct promotearg {
2342 	list_t shared_snaps, origin_snaps, clone_snaps;
2343 	dsl_dataset_t *origin_origin, *origin_head;
2344 	uint64_t used, comp, uncomp, unique, cloneusedsnap, originusedsnap;
2345 	char *err_ds;
2346 };
2347 
2348 static int snaplist_space(list_t *l, uint64_t mintxg, uint64_t *spacep);
2349 
2350 /* ARGSUSED */
2351 static int
2352 dsl_dataset_promote_check(void *arg1, void *arg2, dmu_tx_t *tx)
2353 {
2354 	dsl_dataset_t *hds = arg1;
2355 	struct promotearg *pa = arg2;
2356 	struct promotenode *snap = list_head(&pa->shared_snaps);
2357 	dsl_dataset_t *origin_ds = snap->ds;
2358 	int err;
2359 
2360 	/* Check that it is a real clone */
2361 	if (!dsl_dir_is_clone(hds->ds_dir))
2362 		return (EINVAL);
2363 
2364 	/* Since this is so expensive, don't do the preliminary check */
2365 	if (!dmu_tx_is_syncing(tx))
2366 		return (0);
2367 
2368 	if (hds->ds_phys->ds_flags & DS_FLAG_NOPROMOTE)
2369 		return (EXDEV);
2370 
2371 	/* compute origin's new unique space */
2372 	snap = list_tail(&pa->clone_snaps);
2373 	ASSERT3U(snap->ds->ds_phys->ds_prev_snap_obj, ==, origin_ds->ds_object);
2374 	err = bplist_space_birthrange(&snap->ds->ds_deadlist,
2375 	    origin_ds->ds_phys->ds_prev_snap_txg, UINT64_MAX, &pa->unique);
2376 	if (err)
2377 		return (err);
2378 
2379 	/*
2380 	 * Walk the snapshots that we are moving
2381 	 *
2382 	 * Compute space to transfer.  Consider the incremental changes
2383 	 * to used for each snapshot:
2384 	 * (my used) = (prev's used) + (blocks born) - (blocks killed)
2385 	 * So each snapshot gave birth to:
2386 	 * (blocks born) = (my used) - (prev's used) + (blocks killed)
2387 	 * So a sequence would look like:
2388 	 * (uN - u(N-1) + kN) + ... + (u1 - u0 + k1) + (u0 - 0 + k0)
2389 	 * Which simplifies to:
2390 	 * uN + kN + kN-1 + ... + k1 + k0
2391 	 * Note however, if we stop before we reach the ORIGIN we get:
2392 	 * uN + kN + kN-1 + ... + kM - uM-1
2393 	 */
2394 	pa->used = origin_ds->ds_phys->ds_used_bytes;
2395 	pa->comp = origin_ds->ds_phys->ds_compressed_bytes;
2396 	pa->uncomp = origin_ds->ds_phys->ds_uncompressed_bytes;
2397 	for (snap = list_head(&pa->shared_snaps); snap;
2398 	    snap = list_next(&pa->shared_snaps, snap)) {
2399 		uint64_t val, dlused, dlcomp, dluncomp;
2400 		dsl_dataset_t *ds = snap->ds;
2401 
2402 		/* Check that the snapshot name does not conflict */
2403 		VERIFY(0 == dsl_dataset_get_snapname(ds));
2404 		err = dsl_dataset_snap_lookup(hds, ds->ds_snapname, &val);
2405 		if (err == 0) {
2406 			err = EEXIST;
2407 			goto out;
2408 		}
2409 		if (err != ENOENT)
2410 			goto out;
2411 
2412 		/* The very first snapshot does not have a deadlist */
2413 		if (ds->ds_phys->ds_prev_snap_obj == 0)
2414 			continue;
2415 
2416 		if (err = bplist_space(&ds->ds_deadlist,
2417 		    &dlused, &dlcomp, &dluncomp))
2418 			goto out;
2419 		pa->used += dlused;
2420 		pa->comp += dlcomp;
2421 		pa->uncomp += dluncomp;
2422 	}
2423 
2424 	/*
2425 	 * If we are a clone of a clone then we never reached ORIGIN,
2426 	 * so we need to subtract out the clone origin's used space.
2427 	 */
2428 	if (pa->origin_origin) {
2429 		pa->used -= pa->origin_origin->ds_phys->ds_used_bytes;
2430 		pa->comp -= pa->origin_origin->ds_phys->ds_compressed_bytes;
2431 		pa->uncomp -= pa->origin_origin->ds_phys->ds_uncompressed_bytes;
2432 	}
2433 
2434 	/* Check that there is enough space here */
2435 	err = dsl_dir_transfer_possible(origin_ds->ds_dir, hds->ds_dir,
2436 	    pa->used);
2437 	if (err)
2438 		return (err);
2439 
2440 	/*
2441 	 * Compute the amounts of space that will be used by snapshots
2442 	 * after the promotion (for both origin and clone).  For each,
2443 	 * it is the amount of space that will be on all of their
2444 	 * deadlists (that was not born before their new origin).
2445 	 */
2446 	if (hds->ds_dir->dd_phys->dd_flags & DD_FLAG_USED_BREAKDOWN) {
2447 		uint64_t space;
2448 
2449 		/*
2450 		 * Note, typically this will not be a clone of a clone,
2451 		 * so snap->ds->ds_origin_txg will be < TXG_INITIAL, so
2452 		 * these snaplist_space() -> bplist_space_birthrange()
2453 		 * calls will be fast because they do not have to
2454 		 * iterate over all bps.
2455 		 */
2456 		snap = list_head(&pa->origin_snaps);
2457 		err = snaplist_space(&pa->shared_snaps,
2458 		    snap->ds->ds_origin_txg, &pa->cloneusedsnap);
2459 		if (err)
2460 			return (err);
2461 
2462 		err = snaplist_space(&pa->clone_snaps,
2463 		    snap->ds->ds_origin_txg, &space);
2464 		if (err)
2465 			return (err);
2466 		pa->cloneusedsnap += space;
2467 	}
2468 	if (origin_ds->ds_dir->dd_phys->dd_flags & DD_FLAG_USED_BREAKDOWN) {
2469 		err = snaplist_space(&pa->origin_snaps,
2470 		    origin_ds->ds_phys->ds_creation_txg, &pa->originusedsnap);
2471 		if (err)
2472 			return (err);
2473 	}
2474 
2475 	return (0);
2476 out:
2477 	pa->err_ds =  snap->ds->ds_snapname;
2478 	return (err);
2479 }
2480 
2481 static void
2482 dsl_dataset_promote_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
2483 {
2484 	dsl_dataset_t *hds = arg1;
2485 	struct promotearg *pa = arg2;
2486 	struct promotenode *snap = list_head(&pa->shared_snaps);
2487 	dsl_dataset_t *origin_ds = snap->ds;
2488 	dsl_dataset_t *origin_head;
2489 	dsl_dir_t *dd = hds->ds_dir;
2490 	dsl_pool_t *dp = hds->ds_dir->dd_pool;
2491 	dsl_dir_t *odd = NULL;
2492 	uint64_t oldnext_obj;
2493 	int64_t delta;
2494 
2495 	ASSERT(0 == (hds->ds_phys->ds_flags & DS_FLAG_NOPROMOTE));
2496 
2497 	snap = list_head(&pa->origin_snaps);
2498 	origin_head = snap->ds;
2499 
2500 	/*
2501 	 * We need to explicitly open odd, since origin_ds's dd will be
2502 	 * changing.
2503 	 */
2504 	VERIFY(0 == dsl_dir_open_obj(dp, origin_ds->ds_dir->dd_object,
2505 	    NULL, FTAG, &odd));
2506 
2507 	/* change origin's next snap */
2508 	dmu_buf_will_dirty(origin_ds->ds_dbuf, tx);
2509 	oldnext_obj = origin_ds->ds_phys->ds_next_snap_obj;
2510 	snap = list_tail(&pa->clone_snaps);
2511 	ASSERT3U(snap->ds->ds_phys->ds_prev_snap_obj, ==, origin_ds->ds_object);
2512 	origin_ds->ds_phys->ds_next_snap_obj = snap->ds->ds_object;
2513 
2514 	/* change the origin's next clone */
2515 	if (origin_ds->ds_phys->ds_next_clones_obj) {
2516 		VERIFY3U(0, ==, zap_remove_int(dp->dp_meta_objset,
2517 		    origin_ds->ds_phys->ds_next_clones_obj,
2518 		    origin_ds->ds_phys->ds_next_snap_obj, tx));
2519 		VERIFY3U(0, ==, zap_add_int(dp->dp_meta_objset,
2520 		    origin_ds->ds_phys->ds_next_clones_obj,
2521 		    oldnext_obj, tx));
2522 	}
2523 
2524 	/* change origin */
2525 	dmu_buf_will_dirty(dd->dd_dbuf, tx);
2526 	ASSERT3U(dd->dd_phys->dd_origin_obj, ==, origin_ds->ds_object);
2527 	dd->dd_phys->dd_origin_obj = odd->dd_phys->dd_origin_obj;
2528 	hds->ds_origin_txg = origin_head->ds_origin_txg;
2529 	dmu_buf_will_dirty(odd->dd_dbuf, tx);
2530 	odd->dd_phys->dd_origin_obj = origin_ds->ds_object;
2531 	origin_head->ds_origin_txg = origin_ds->ds_phys->ds_creation_txg;
2532 
2533 	/* move snapshots to this dir */
2534 	for (snap = list_head(&pa->shared_snaps); snap;
2535 	    snap = list_next(&pa->shared_snaps, snap)) {
2536 		dsl_dataset_t *ds = snap->ds;
2537 
2538 		/* unregister props as dsl_dir is changing */
2539 		if (ds->ds_objset) {
2540 			dmu_objset_evict(ds->ds_objset);
2541 			ds->ds_objset = NULL;
2542 		}
2543 		/* move snap name entry */
2544 		VERIFY(0 == dsl_dataset_get_snapname(ds));
2545 		VERIFY(0 == dsl_dataset_snap_remove(origin_head,
2546 		    ds->ds_snapname, tx));
2547 		VERIFY(0 == zap_add(dp->dp_meta_objset,
2548 		    hds->ds_phys->ds_snapnames_zapobj, ds->ds_snapname,
2549 		    8, 1, &ds->ds_object, tx));
2550 		/* change containing dsl_dir */
2551 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
2552 		ASSERT3U(ds->ds_phys->ds_dir_obj, ==, odd->dd_object);
2553 		ds->ds_phys->ds_dir_obj = dd->dd_object;
2554 		ASSERT3P(ds->ds_dir, ==, odd);
2555 		dsl_dir_close(ds->ds_dir, ds);
2556 		VERIFY(0 == dsl_dir_open_obj(dp, dd->dd_object,
2557 		    NULL, ds, &ds->ds_dir));
2558 
2559 		ASSERT3U(dsl_prop_numcb(ds), ==, 0);
2560 	}
2561 
2562 	/*
2563 	 * Change space accounting.
2564 	 * Note, pa->*usedsnap and dd_used_breakdown[SNAP] will either
2565 	 * both be valid, or both be 0 (resulting in delta == 0).  This
2566 	 * is true for each of {clone,origin} independently.
2567 	 */
2568 
2569 	delta = pa->cloneusedsnap -
2570 	    dd->dd_phys->dd_used_breakdown[DD_USED_SNAP];
2571 	ASSERT3S(delta, >=, 0);
2572 	ASSERT3U(pa->used, >=, delta);
2573 	dsl_dir_diduse_space(dd, DD_USED_SNAP, delta, 0, 0, tx);
2574 	dsl_dir_diduse_space(dd, DD_USED_HEAD,
2575 	    pa->used - delta, pa->comp, pa->uncomp, tx);
2576 
2577 	delta = pa->originusedsnap -
2578 	    odd->dd_phys->dd_used_breakdown[DD_USED_SNAP];
2579 	ASSERT3S(delta, <=, 0);
2580 	ASSERT3U(pa->used, >=, -delta);
2581 	dsl_dir_diduse_space(odd, DD_USED_SNAP, delta, 0, 0, tx);
2582 	dsl_dir_diduse_space(odd, DD_USED_HEAD,
2583 	    -pa->used - delta, -pa->comp, -pa->uncomp, tx);
2584 
2585 	origin_ds->ds_phys->ds_unique_bytes = pa->unique;
2586 
2587 	/* log history record */
2588 	spa_history_internal_log(LOG_DS_PROMOTE, dd->dd_pool->dp_spa, tx,
2589 	    cr, "dataset = %llu", hds->ds_object);
2590 
2591 	dsl_dir_close(odd, FTAG);
2592 }
2593 
2594 static char *snaplist_tag = "snaplist";
2595 /*
2596  * Make a list of dsl_dataset_t's for the snapshots between first_obj
2597  * (exclusive) and last_obj (inclusive).  The list will be in reverse
2598  * order (last_obj will be the list_head()).  If first_obj == 0, do all
2599  * snapshots back to this dataset's origin.
2600  */
2601 static int
2602 snaplist_make(dsl_pool_t *dp, boolean_t own,
2603     uint64_t first_obj, uint64_t last_obj, list_t *l)
2604 {
2605 	uint64_t obj = last_obj;
2606 
2607 	ASSERT(RW_LOCK_HELD(&dp->dp_config_rwlock));
2608 
2609 	list_create(l, sizeof (struct promotenode),
2610 	    offsetof(struct promotenode, link));
2611 
2612 	while (obj != first_obj) {
2613 		dsl_dataset_t *ds;
2614 		struct promotenode *snap;
2615 		int err;
2616 
2617 		if (own) {
2618 			err = dsl_dataset_own_obj(dp, obj,
2619 			    0, snaplist_tag, &ds);
2620 			if (err == 0)
2621 				dsl_dataset_make_exclusive(ds, snaplist_tag);
2622 		} else {
2623 			err = dsl_dataset_hold_obj(dp, obj, snaplist_tag, &ds);
2624 		}
2625 		if (err == ENOENT) {
2626 			/* lost race with snapshot destroy */
2627 			struct promotenode *last = list_tail(l);
2628 			ASSERT(obj != last->ds->ds_phys->ds_prev_snap_obj);
2629 			obj = last->ds->ds_phys->ds_prev_snap_obj;
2630 			continue;
2631 		} else if (err) {
2632 			return (err);
2633 		}
2634 
2635 		if (first_obj == 0)
2636 			first_obj = ds->ds_dir->dd_phys->dd_origin_obj;
2637 
2638 		snap = kmem_alloc(sizeof (struct promotenode), KM_SLEEP);
2639 		snap->ds = ds;
2640 		list_insert_tail(l, snap);
2641 		obj = ds->ds_phys->ds_prev_snap_obj;
2642 	}
2643 
2644 	return (0);
2645 }
2646 
2647 static int
2648 snaplist_space(list_t *l, uint64_t mintxg, uint64_t *spacep)
2649 {
2650 	struct promotenode *snap;
2651 
2652 	*spacep = 0;
2653 	for (snap = list_head(l); snap; snap = list_next(l, snap)) {
2654 		uint64_t used;
2655 		int err = bplist_space_birthrange(&snap->ds->ds_deadlist,
2656 		    mintxg, UINT64_MAX, &used);
2657 		if (err)
2658 			return (err);
2659 		*spacep += used;
2660 	}
2661 	return (0);
2662 }
2663 
2664 static void
2665 snaplist_destroy(list_t *l, boolean_t own)
2666 {
2667 	struct promotenode *snap;
2668 
2669 	if (!l || !list_link_active(&l->list_head))
2670 		return;
2671 
2672 	while ((snap = list_tail(l)) != NULL) {
2673 		list_remove(l, snap);
2674 		if (own)
2675 			dsl_dataset_disown(snap->ds, snaplist_tag);
2676 		else
2677 			dsl_dataset_rele(snap->ds, snaplist_tag);
2678 		kmem_free(snap, sizeof (struct promotenode));
2679 	}
2680 	list_destroy(l);
2681 }
2682 
2683 /*
2684  * Promote a clone.  Nomenclature note:
2685  * "clone" or "cds": the original clone which is being promoted
2686  * "origin" or "ods": the snapshot which is originally clone's origin
2687  * "origin head" or "ohds": the dataset which is the head
2688  * (filesystem/volume) for the origin
2689  * "origin origin": the origin of the origin's filesystem (typically
2690  * NULL, indicating that the clone is not a clone of a clone).
2691  */
2692 int
2693 dsl_dataset_promote(const char *name, char *conflsnap)
2694 {
2695 	dsl_dataset_t *ds;
2696 	dsl_dir_t *dd;
2697 	dsl_pool_t *dp;
2698 	dmu_object_info_t doi;
2699 	struct promotearg pa = { 0 };
2700 	struct promotenode *snap;
2701 	int err;
2702 
2703 	err = dsl_dataset_hold(name, FTAG, &ds);
2704 	if (err)
2705 		return (err);
2706 	dd = ds->ds_dir;
2707 	dp = dd->dd_pool;
2708 
2709 	err = dmu_object_info(dp->dp_meta_objset,
2710 	    ds->ds_phys->ds_snapnames_zapobj, &doi);
2711 	if (err) {
2712 		dsl_dataset_rele(ds, FTAG);
2713 		return (err);
2714 	}
2715 
2716 	if (dsl_dataset_is_snapshot(ds) || dd->dd_phys->dd_origin_obj == 0) {
2717 		dsl_dataset_rele(ds, FTAG);
2718 		return (EINVAL);
2719 	}
2720 
2721 	/*
2722 	 * We are going to inherit all the snapshots taken before our
2723 	 * origin (i.e., our new origin will be our parent's origin).
2724 	 * Take ownership of them so that we can rename them into our
2725 	 * namespace.
2726 	 */
2727 	rw_enter(&dp->dp_config_rwlock, RW_READER);
2728 
2729 	err = snaplist_make(dp, B_TRUE, 0, dd->dd_phys->dd_origin_obj,
2730 	    &pa.shared_snaps);
2731 	if (err != 0)
2732 		goto out;
2733 
2734 	err = snaplist_make(dp, B_FALSE, 0, ds->ds_object, &pa.clone_snaps);
2735 	if (err != 0)
2736 		goto out;
2737 
2738 	snap = list_head(&pa.shared_snaps);
2739 	ASSERT3U(snap->ds->ds_object, ==, dd->dd_phys->dd_origin_obj);
2740 	err = snaplist_make(dp, B_FALSE, dd->dd_phys->dd_origin_obj,
2741 	    snap->ds->ds_dir->dd_phys->dd_head_dataset_obj, &pa.origin_snaps);
2742 	if (err != 0)
2743 		goto out;
2744 
2745 	if (dsl_dir_is_clone(snap->ds->ds_dir)) {
2746 		err = dsl_dataset_own_obj(dp,
2747 		    snap->ds->ds_dir->dd_phys->dd_origin_obj,
2748 		    0, FTAG, &pa.origin_origin);
2749 		if (err != 0)
2750 			goto out;
2751 	}
2752 
2753 out:
2754 	rw_exit(&dp->dp_config_rwlock);
2755 
2756 	/*
2757 	 * Add in 128x the snapnames zapobj size, since we will be moving
2758 	 * a bunch of snapnames to the promoted ds, and dirtying their
2759 	 * bonus buffers.
2760 	 */
2761 	if (err == 0) {
2762 		err = dsl_sync_task_do(dp, dsl_dataset_promote_check,
2763 		    dsl_dataset_promote_sync, ds, &pa,
2764 		    2 + 2 * doi.doi_physical_blks);
2765 		if (err && pa.err_ds && conflsnap)
2766 			(void) strncpy(conflsnap, pa.err_ds, MAXNAMELEN);
2767 	}
2768 
2769 	snaplist_destroy(&pa.shared_snaps, B_TRUE);
2770 	snaplist_destroy(&pa.clone_snaps, B_FALSE);
2771 	snaplist_destroy(&pa.origin_snaps, B_FALSE);
2772 	if (pa.origin_origin)
2773 		dsl_dataset_disown(pa.origin_origin, FTAG);
2774 	dsl_dataset_rele(ds, FTAG);
2775 	return (err);
2776 }
2777 
2778 struct cloneswaparg {
2779 	dsl_dataset_t *cds; /* clone dataset */
2780 	dsl_dataset_t *ohds; /* origin's head dataset */
2781 	boolean_t force;
2782 	int64_t unused_refres_delta; /* change in unconsumed refreservation */
2783 };
2784 
2785 /* ARGSUSED */
2786 static int
2787 dsl_dataset_clone_swap_check(void *arg1, void *arg2, dmu_tx_t *tx)
2788 {
2789 	struct cloneswaparg *csa = arg1;
2790 
2791 	/* they should both be heads */
2792 	if (dsl_dataset_is_snapshot(csa->cds) ||
2793 	    dsl_dataset_is_snapshot(csa->ohds))
2794 		return (EINVAL);
2795 
2796 	/* the branch point should be just before them */
2797 	if (csa->cds->ds_prev != csa->ohds->ds_prev)
2798 		return (EINVAL);
2799 
2800 	/* cds should be the clone (unless they are unrelated) */
2801 	if (csa->cds->ds_prev != NULL &&
2802 	    csa->cds->ds_prev != csa->cds->ds_dir->dd_pool->dp_origin_snap &&
2803 	    csa->ohds->ds_object !=
2804 	    csa->cds->ds_prev->ds_phys->ds_next_snap_obj)
2805 		return (EINVAL);
2806 
2807 	/* the clone should be a child of the origin */
2808 	if (csa->cds->ds_dir->dd_parent != csa->ohds->ds_dir)
2809 		return (EINVAL);
2810 
2811 	/* ohds shouldn't be modified unless 'force' */
2812 	if (!csa->force && dsl_dataset_modified_since_lastsnap(csa->ohds))
2813 		return (ETXTBSY);
2814 
2815 	/* adjust amount of any unconsumed refreservation */
2816 	csa->unused_refres_delta =
2817 	    (int64_t)MIN(csa->ohds->ds_reserved,
2818 	    csa->ohds->ds_phys->ds_unique_bytes) -
2819 	    (int64_t)MIN(csa->ohds->ds_reserved,
2820 	    csa->cds->ds_phys->ds_unique_bytes);
2821 
2822 	if (csa->unused_refres_delta > 0 &&
2823 	    csa->unused_refres_delta >
2824 	    dsl_dir_space_available(csa->ohds->ds_dir, NULL, 0, TRUE))
2825 		return (ENOSPC);
2826 
2827 	return (0);
2828 }
2829 
2830 /* ARGSUSED */
2831 static void
2832 dsl_dataset_clone_swap_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
2833 {
2834 	struct cloneswaparg *csa = arg1;
2835 	dsl_pool_t *dp = csa->cds->ds_dir->dd_pool;
2836 
2837 	ASSERT(csa->cds->ds_reserved == 0);
2838 	ASSERT(csa->cds->ds_quota == csa->ohds->ds_quota);
2839 
2840 	dmu_buf_will_dirty(csa->cds->ds_dbuf, tx);
2841 	dmu_buf_will_dirty(csa->ohds->ds_dbuf, tx);
2842 
2843 	if (csa->cds->ds_objset != NULL) {
2844 		dmu_objset_evict(csa->cds->ds_objset);
2845 		csa->cds->ds_objset = NULL;
2846 	}
2847 
2848 	if (csa->ohds->ds_objset != NULL) {
2849 		dmu_objset_evict(csa->ohds->ds_objset);
2850 		csa->ohds->ds_objset = NULL;
2851 	}
2852 
2853 	/*
2854 	 * Reset origin's unique bytes, if it exists.
2855 	 */
2856 	if (csa->cds->ds_prev) {
2857 		dsl_dataset_t *origin = csa->cds->ds_prev;
2858 		dmu_buf_will_dirty(origin->ds_dbuf, tx);
2859 		VERIFY(0 == bplist_space_birthrange(&csa->cds->ds_deadlist,
2860 		    origin->ds_phys->ds_prev_snap_txg, UINT64_MAX,
2861 		    &origin->ds_phys->ds_unique_bytes));
2862 	}
2863 
2864 	/* swap blkptrs */
2865 	{
2866 		blkptr_t tmp;
2867 		tmp = csa->ohds->ds_phys->ds_bp;
2868 		csa->ohds->ds_phys->ds_bp = csa->cds->ds_phys->ds_bp;
2869 		csa->cds->ds_phys->ds_bp = tmp;
2870 	}
2871 
2872 	/* set dd_*_bytes */
2873 	{
2874 		int64_t dused, dcomp, duncomp;
2875 		uint64_t cdl_used, cdl_comp, cdl_uncomp;
2876 		uint64_t odl_used, odl_comp, odl_uncomp;
2877 
2878 		ASSERT3U(csa->cds->ds_dir->dd_phys->
2879 		    dd_used_breakdown[DD_USED_SNAP], ==, 0);
2880 
2881 		VERIFY(0 == bplist_space(&csa->cds->ds_deadlist, &cdl_used,
2882 		    &cdl_comp, &cdl_uncomp));
2883 		VERIFY(0 == bplist_space(&csa->ohds->ds_deadlist, &odl_used,
2884 		    &odl_comp, &odl_uncomp));
2885 
2886 		dused = csa->cds->ds_phys->ds_used_bytes + cdl_used -
2887 		    (csa->ohds->ds_phys->ds_used_bytes + odl_used);
2888 		dcomp = csa->cds->ds_phys->ds_compressed_bytes + cdl_comp -
2889 		    (csa->ohds->ds_phys->ds_compressed_bytes + odl_comp);
2890 		duncomp = csa->cds->ds_phys->ds_uncompressed_bytes +
2891 		    cdl_uncomp -
2892 		    (csa->ohds->ds_phys->ds_uncompressed_bytes + odl_uncomp);
2893 
2894 		dsl_dir_diduse_space(csa->ohds->ds_dir, DD_USED_HEAD,
2895 		    dused, dcomp, duncomp, tx);
2896 		dsl_dir_diduse_space(csa->cds->ds_dir, DD_USED_HEAD,
2897 		    -dused, -dcomp, -duncomp, tx);
2898 
2899 		/*
2900 		 * The difference in the space used by snapshots is the
2901 		 * difference in snapshot space due to the head's
2902 		 * deadlist (since that's the only thing that's
2903 		 * changing that affects the snapused).
2904 		 */
2905 		VERIFY(0 == bplist_space_birthrange(&csa->cds->ds_deadlist,
2906 		    csa->ohds->ds_origin_txg, UINT64_MAX, &cdl_used));
2907 		VERIFY(0 == bplist_space_birthrange(&csa->ohds->ds_deadlist,
2908 		    csa->ohds->ds_origin_txg, UINT64_MAX, &odl_used));
2909 		dsl_dir_transfer_space(csa->ohds->ds_dir, cdl_used - odl_used,
2910 		    DD_USED_HEAD, DD_USED_SNAP, tx);
2911 	}
2912 
2913 #define	SWITCH64(x, y) \
2914 	{ \
2915 		uint64_t __tmp = (x); \
2916 		(x) = (y); \
2917 		(y) = __tmp; \
2918 	}
2919 
2920 	/* swap ds_*_bytes */
2921 	SWITCH64(csa->ohds->ds_phys->ds_used_bytes,
2922 	    csa->cds->ds_phys->ds_used_bytes);
2923 	SWITCH64(csa->ohds->ds_phys->ds_compressed_bytes,
2924 	    csa->cds->ds_phys->ds_compressed_bytes);
2925 	SWITCH64(csa->ohds->ds_phys->ds_uncompressed_bytes,
2926 	    csa->cds->ds_phys->ds_uncompressed_bytes);
2927 	SWITCH64(csa->ohds->ds_phys->ds_unique_bytes,
2928 	    csa->cds->ds_phys->ds_unique_bytes);
2929 
2930 	/* apply any parent delta for change in unconsumed refreservation */
2931 	dsl_dir_diduse_space(csa->ohds->ds_dir, DD_USED_REFRSRV,
2932 	    csa->unused_refres_delta, 0, 0, tx);
2933 
2934 	/* swap deadlists */
2935 	bplist_close(&csa->cds->ds_deadlist);
2936 	bplist_close(&csa->ohds->ds_deadlist);
2937 	SWITCH64(csa->ohds->ds_phys->ds_deadlist_obj,
2938 	    csa->cds->ds_phys->ds_deadlist_obj);
2939 	VERIFY(0 == bplist_open(&csa->cds->ds_deadlist, dp->dp_meta_objset,
2940 	    csa->cds->ds_phys->ds_deadlist_obj));
2941 	VERIFY(0 == bplist_open(&csa->ohds->ds_deadlist, dp->dp_meta_objset,
2942 	    csa->ohds->ds_phys->ds_deadlist_obj));
2943 
2944 	dsl_pool_ds_clone_swapped(csa->ohds, csa->cds, tx);
2945 }
2946 
2947 /*
2948  * Swap 'clone' with its origin head datasets.  Used at the end of "zfs
2949  * recv" into an existing fs to swizzle the file system to the new
2950  * version, and by "zfs rollback".  Can also be used to swap two
2951  * independent head datasets if neither has any snapshots.
2952  */
2953 int
2954 dsl_dataset_clone_swap(dsl_dataset_t *clone, dsl_dataset_t *origin_head,
2955     boolean_t force)
2956 {
2957 	struct cloneswaparg csa;
2958 	int error;
2959 
2960 	ASSERT(clone->ds_owner);
2961 	ASSERT(origin_head->ds_owner);
2962 retry:
2963 	/* Need exclusive access for the swap */
2964 	rw_enter(&clone->ds_rwlock, RW_WRITER);
2965 	if (!rw_tryenter(&origin_head->ds_rwlock, RW_WRITER)) {
2966 		rw_exit(&clone->ds_rwlock);
2967 		rw_enter(&origin_head->ds_rwlock, RW_WRITER);
2968 		if (!rw_tryenter(&clone->ds_rwlock, RW_WRITER)) {
2969 			rw_exit(&origin_head->ds_rwlock);
2970 			goto retry;
2971 		}
2972 	}
2973 	csa.cds = clone;
2974 	csa.ohds = origin_head;
2975 	csa.force = force;
2976 	error = dsl_sync_task_do(clone->ds_dir->dd_pool,
2977 	    dsl_dataset_clone_swap_check,
2978 	    dsl_dataset_clone_swap_sync, &csa, NULL, 9);
2979 	return (error);
2980 }
2981 
2982 /*
2983  * Given a pool name and a dataset object number in that pool,
2984  * return the name of that dataset.
2985  */
2986 int
2987 dsl_dsobj_to_dsname(char *pname, uint64_t obj, char *buf)
2988 {
2989 	spa_t *spa;
2990 	dsl_pool_t *dp;
2991 	dsl_dataset_t *ds;
2992 	int error;
2993 
2994 	if ((error = spa_open(pname, &spa, FTAG)) != 0)
2995 		return (error);
2996 	dp = spa_get_dsl(spa);
2997 	rw_enter(&dp->dp_config_rwlock, RW_READER);
2998 	if ((error = dsl_dataset_hold_obj(dp, obj, FTAG, &ds)) == 0) {
2999 		dsl_dataset_name(ds, buf);
3000 		dsl_dataset_rele(ds, FTAG);
3001 	}
3002 	rw_exit(&dp->dp_config_rwlock);
3003 	spa_close(spa, FTAG);
3004 
3005 	return (error);
3006 }
3007 
3008 int
3009 dsl_dataset_check_quota(dsl_dataset_t *ds, boolean_t check_quota,
3010     uint64_t asize, uint64_t inflight, uint64_t *used, uint64_t *ref_rsrv)
3011 {
3012 	int error = 0;
3013 
3014 	ASSERT3S(asize, >, 0);
3015 
3016 	/*
3017 	 * *ref_rsrv is the portion of asize that will come from any
3018 	 * unconsumed refreservation space.
3019 	 */
3020 	*ref_rsrv = 0;
3021 
3022 	mutex_enter(&ds->ds_lock);
3023 	/*
3024 	 * Make a space adjustment for reserved bytes.
3025 	 */
3026 	if (ds->ds_reserved > ds->ds_phys->ds_unique_bytes) {
3027 		ASSERT3U(*used, >=,
3028 		    ds->ds_reserved - ds->ds_phys->ds_unique_bytes);
3029 		*used -= (ds->ds_reserved - ds->ds_phys->ds_unique_bytes);
3030 		*ref_rsrv =
3031 		    asize - MIN(asize, parent_delta(ds, asize + inflight));
3032 	}
3033 
3034 	if (!check_quota || ds->ds_quota == 0) {
3035 		mutex_exit(&ds->ds_lock);
3036 		return (0);
3037 	}
3038 	/*
3039 	 * If they are requesting more space, and our current estimate
3040 	 * is over quota, they get to try again unless the actual
3041 	 * on-disk is over quota and there are no pending changes (which
3042 	 * may free up space for us).
3043 	 */
3044 	if (ds->ds_phys->ds_used_bytes + inflight >= ds->ds_quota) {
3045 		if (inflight > 0 || ds->ds_phys->ds_used_bytes < ds->ds_quota)
3046 			error = ERESTART;
3047 		else
3048 			error = EDQUOT;
3049 	}
3050 	mutex_exit(&ds->ds_lock);
3051 
3052 	return (error);
3053 }
3054 
3055 /* ARGSUSED */
3056 static int
3057 dsl_dataset_set_quota_check(void *arg1, void *arg2, dmu_tx_t *tx)
3058 {
3059 	dsl_dataset_t *ds = arg1;
3060 	uint64_t *quotap = arg2;
3061 	uint64_t new_quota = *quotap;
3062 
3063 	if (spa_version(ds->ds_dir->dd_pool->dp_spa) < SPA_VERSION_REFQUOTA)
3064 		return (ENOTSUP);
3065 
3066 	if (new_quota == 0)
3067 		return (0);
3068 
3069 	if (new_quota < ds->ds_phys->ds_used_bytes ||
3070 	    new_quota < ds->ds_reserved)
3071 		return (ENOSPC);
3072 
3073 	return (0);
3074 }
3075 
3076 /* ARGSUSED */
3077 void
3078 dsl_dataset_set_quota_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
3079 {
3080 	dsl_dataset_t *ds = arg1;
3081 	uint64_t *quotap = arg2;
3082 	uint64_t new_quota = *quotap;
3083 
3084 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
3085 
3086 	ds->ds_quota = new_quota;
3087 
3088 	dsl_dir_prop_set_uint64_sync(ds->ds_dir, "refquota", new_quota, cr, tx);
3089 
3090 	spa_history_internal_log(LOG_DS_REFQUOTA, ds->ds_dir->dd_pool->dp_spa,
3091 	    tx, cr, "%lld dataset = %llu ",
3092 	    (longlong_t)new_quota, ds->ds_object);
3093 }
3094 
3095 int
3096 dsl_dataset_set_quota(const char *dsname, uint64_t quota)
3097 {
3098 	dsl_dataset_t *ds;
3099 	int err;
3100 
3101 	err = dsl_dataset_hold(dsname, FTAG, &ds);
3102 	if (err)
3103 		return (err);
3104 
3105 	if (quota != ds->ds_quota) {
3106 		/*
3107 		 * If someone removes a file, then tries to set the quota, we
3108 		 * want to make sure the file freeing takes effect.
3109 		 */
3110 		txg_wait_open(ds->ds_dir->dd_pool, 0);
3111 
3112 		err = dsl_sync_task_do(ds->ds_dir->dd_pool,
3113 		    dsl_dataset_set_quota_check, dsl_dataset_set_quota_sync,
3114 		    ds, &quota, 0);
3115 	}
3116 	dsl_dataset_rele(ds, FTAG);
3117 	return (err);
3118 }
3119 
3120 static int
3121 dsl_dataset_set_reservation_check(void *arg1, void *arg2, dmu_tx_t *tx)
3122 {
3123 	dsl_dataset_t *ds = arg1;
3124 	uint64_t *reservationp = arg2;
3125 	uint64_t new_reservation = *reservationp;
3126 	uint64_t unique;
3127 
3128 	if (spa_version(ds->ds_dir->dd_pool->dp_spa) <
3129 	    SPA_VERSION_REFRESERVATION)
3130 		return (ENOTSUP);
3131 
3132 	if (dsl_dataset_is_snapshot(ds))
3133 		return (EINVAL);
3134 
3135 	/*
3136 	 * If we are doing the preliminary check in open context, the
3137 	 * space estimates may be inaccurate.
3138 	 */
3139 	if (!dmu_tx_is_syncing(tx))
3140 		return (0);
3141 
3142 	mutex_enter(&ds->ds_lock);
3143 	unique = dsl_dataset_unique(ds);
3144 	mutex_exit(&ds->ds_lock);
3145 
3146 	if (MAX(unique, new_reservation) > MAX(unique, ds->ds_reserved)) {
3147 		uint64_t delta = MAX(unique, new_reservation) -
3148 		    MAX(unique, ds->ds_reserved);
3149 
3150 		if (delta > dsl_dir_space_available(ds->ds_dir, NULL, 0, TRUE))
3151 			return (ENOSPC);
3152 		if (ds->ds_quota > 0 &&
3153 		    new_reservation > ds->ds_quota)
3154 			return (ENOSPC);
3155 	}
3156 
3157 	return (0);
3158 }
3159 
3160 /* ARGSUSED */
3161 static void
3162 dsl_dataset_set_reservation_sync(void *arg1, void *arg2, cred_t *cr,
3163     dmu_tx_t *tx)
3164 {
3165 	dsl_dataset_t *ds = arg1;
3166 	uint64_t *reservationp = arg2;
3167 	uint64_t new_reservation = *reservationp;
3168 	uint64_t unique;
3169 	int64_t delta;
3170 
3171 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
3172 
3173 	mutex_enter(&ds->ds_dir->dd_lock);
3174 	mutex_enter(&ds->ds_lock);
3175 	unique = dsl_dataset_unique(ds);
3176 	delta = MAX(0, (int64_t)(new_reservation - unique)) -
3177 	    MAX(0, (int64_t)(ds->ds_reserved - unique));
3178 	ds->ds_reserved = new_reservation;
3179 	mutex_exit(&ds->ds_lock);
3180 
3181 	dsl_dir_diduse_space(ds->ds_dir, DD_USED_REFRSRV, delta, 0, 0, tx);
3182 	mutex_exit(&ds->ds_dir->dd_lock);
3183 	dsl_dir_prop_set_uint64_sync(ds->ds_dir, "refreservation",
3184 	    new_reservation, cr, tx);
3185 
3186 	spa_history_internal_log(LOG_DS_REFRESERV,
3187 	    ds->ds_dir->dd_pool->dp_spa, tx, cr, "%lld dataset = %llu",
3188 	    (longlong_t)new_reservation, ds->ds_object);
3189 }
3190 
3191 int
3192 dsl_dataset_set_reservation(const char *dsname, uint64_t reservation)
3193 {
3194 	dsl_dataset_t *ds;
3195 	int err;
3196 
3197 	err = dsl_dataset_hold(dsname, FTAG, &ds);
3198 	if (err)
3199 		return (err);
3200 
3201 	err = dsl_sync_task_do(ds->ds_dir->dd_pool,
3202 	    dsl_dataset_set_reservation_check,
3203 	    dsl_dataset_set_reservation_sync, ds, &reservation, 0);
3204 	dsl_dataset_rele(ds, FTAG);
3205 	return (err);
3206 }
3207 
3208 struct dsl_ds_holdarg {
3209 	dsl_sync_task_group_t *dstg;
3210 	char *htag;
3211 	char *snapname;
3212 	boolean_t recursive;
3213 	boolean_t gotone;
3214 	boolean_t temphold;
3215 	char failed[MAXPATHLEN];
3216 };
3217 
3218 static int
3219 dsl_dataset_user_hold_check(void *arg1, void *arg2, dmu_tx_t *tx)
3220 {
3221 	dsl_dataset_t *ds = arg1;
3222 	struct dsl_ds_holdarg *ha = arg2;
3223 	char *htag = ha->htag;
3224 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
3225 	int error = 0;
3226 
3227 	if (spa_version(ds->ds_dir->dd_pool->dp_spa) < SPA_VERSION_USERREFS)
3228 		return (ENOTSUP);
3229 
3230 	if (!dsl_dataset_is_snapshot(ds))
3231 		return (EINVAL);
3232 
3233 	/* tags must be unique */
3234 	mutex_enter(&ds->ds_lock);
3235 	if (ds->ds_phys->ds_userrefs_obj) {
3236 		error = zap_lookup(mos, ds->ds_phys->ds_userrefs_obj, htag,
3237 		    8, 1, tx);
3238 		if (error == 0)
3239 			error = EEXIST;
3240 		else if (error == ENOENT)
3241 			error = 0;
3242 	}
3243 	mutex_exit(&ds->ds_lock);
3244 
3245 	return (error);
3246 }
3247 
3248 static void
3249 dsl_dataset_user_hold_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
3250 {
3251 	dsl_dataset_t *ds = arg1;
3252 	struct dsl_ds_holdarg *ha = arg2;
3253 	char *htag = ha->htag;
3254 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
3255 	objset_t *mos = dp->dp_meta_objset;
3256 	time_t now = gethrestime_sec();
3257 	uint64_t zapobj;
3258 
3259 	mutex_enter(&ds->ds_lock);
3260 	if (ds->ds_phys->ds_userrefs_obj == 0) {
3261 		/*
3262 		 * This is the first user hold for this dataset.  Create
3263 		 * the userrefs zap object.
3264 		 */
3265 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
3266 		zapobj = ds->ds_phys->ds_userrefs_obj =
3267 		    zap_create(mos, DMU_OT_USERREFS, DMU_OT_NONE, 0, tx);
3268 	} else {
3269 		zapobj = ds->ds_phys->ds_userrefs_obj;
3270 	}
3271 	ds->ds_userrefs++;
3272 	mutex_exit(&ds->ds_lock);
3273 
3274 	VERIFY(0 == zap_add(mos, zapobj, htag, 8, 1, &now, tx));
3275 
3276 	if (ha->temphold) {
3277 		VERIFY(0 == dsl_pool_user_hold(dp, ds->ds_object,
3278 		    htag, &now, tx));
3279 	}
3280 
3281 	spa_history_internal_log(LOG_DS_USER_HOLD,
3282 	    dp->dp_spa, tx, cr, "<%s> temp = %d dataset = %llu", htag,
3283 	    (int)ha->temphold, ds->ds_object);
3284 }
3285 
3286 static int
3287 dsl_dataset_user_hold_one(char *dsname, void *arg)
3288 {
3289 	struct dsl_ds_holdarg *ha = arg;
3290 	dsl_dataset_t *ds;
3291 	int error;
3292 	char *name;
3293 
3294 	/* alloc a buffer to hold dsname@snapname plus terminating NULL */
3295 	name = kmem_asprintf("%s@%s", dsname, ha->snapname);
3296 	error = dsl_dataset_hold(name, ha->dstg, &ds);
3297 	strfree(name);
3298 	if (error == 0) {
3299 		ha->gotone = B_TRUE;
3300 		dsl_sync_task_create(ha->dstg, dsl_dataset_user_hold_check,
3301 		    dsl_dataset_user_hold_sync, ds, ha, 0);
3302 	} else if (error == ENOENT && ha->recursive) {
3303 		error = 0;
3304 	} else {
3305 		(void) strcpy(ha->failed, dsname);
3306 	}
3307 	return (error);
3308 }
3309 
3310 int
3311 dsl_dataset_user_hold(char *dsname, char *snapname, char *htag,
3312     boolean_t recursive, boolean_t temphold)
3313 {
3314 	struct dsl_ds_holdarg *ha;
3315 	dsl_sync_task_t *dst;
3316 	spa_t *spa;
3317 	int error;
3318 
3319 	ha = kmem_zalloc(sizeof (struct dsl_ds_holdarg), KM_SLEEP);
3320 
3321 	(void) strlcpy(ha->failed, dsname, sizeof (ha->failed));
3322 
3323 	error = spa_open(dsname, &spa, FTAG);
3324 	if (error) {
3325 		kmem_free(ha, sizeof (struct dsl_ds_holdarg));
3326 		return (error);
3327 	}
3328 
3329 	ha->dstg = dsl_sync_task_group_create(spa_get_dsl(spa));
3330 	ha->htag = htag;
3331 	ha->snapname = snapname;
3332 	ha->recursive = recursive;
3333 	ha->temphold = temphold;
3334 	if (recursive) {
3335 		error = dmu_objset_find(dsname, dsl_dataset_user_hold_one,
3336 		    ha, DS_FIND_CHILDREN);
3337 	} else {
3338 		error = dsl_dataset_user_hold_one(dsname, ha);
3339 	}
3340 	if (error == 0)
3341 		error = dsl_sync_task_group_wait(ha->dstg);
3342 
3343 	for (dst = list_head(&ha->dstg->dstg_tasks); dst;
3344 	    dst = list_next(&ha->dstg->dstg_tasks, dst)) {
3345 		dsl_dataset_t *ds = dst->dst_arg1;
3346 
3347 		if (dst->dst_err) {
3348 			dsl_dataset_name(ds, ha->failed);
3349 			*strchr(ha->failed, '@') = '\0';
3350 		}
3351 		dsl_dataset_rele(ds, ha->dstg);
3352 	}
3353 
3354 	if (error == 0 && recursive && !ha->gotone)
3355 		error = ENOENT;
3356 
3357 	if (error)
3358 		(void) strcpy(dsname, ha->failed);
3359 
3360 	dsl_sync_task_group_destroy(ha->dstg);
3361 	kmem_free(ha, sizeof (struct dsl_ds_holdarg));
3362 	spa_close(spa, FTAG);
3363 	return (error);
3364 }
3365 
3366 struct dsl_ds_releasearg {
3367 	dsl_dataset_t *ds;
3368 	const char *htag;
3369 	boolean_t own;		/* do we own or just hold ds? */
3370 };
3371 
3372 static int
3373 dsl_dataset_release_might_destroy(dsl_dataset_t *ds, const char *htag,
3374     boolean_t *might_destroy)
3375 {
3376 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
3377 	uint64_t zapobj;
3378 	uint64_t tmp;
3379 	int error;
3380 
3381 	*might_destroy = B_FALSE;
3382 
3383 	mutex_enter(&ds->ds_lock);
3384 	zapobj = ds->ds_phys->ds_userrefs_obj;
3385 	if (zapobj == 0) {
3386 		/* The tag can't possibly exist */
3387 		mutex_exit(&ds->ds_lock);
3388 		return (ESRCH);
3389 	}
3390 
3391 	/* Make sure the tag exists */
3392 	error = zap_lookup(mos, zapobj, htag, 8, 1, &tmp);
3393 	if (error) {
3394 		mutex_exit(&ds->ds_lock);
3395 		if (error == ENOENT)
3396 			error = ESRCH;
3397 		return (error);
3398 	}
3399 
3400 	if (ds->ds_userrefs == 1 && ds->ds_phys->ds_num_children == 1 &&
3401 	    DS_IS_DEFER_DESTROY(ds))
3402 		*might_destroy = B_TRUE;
3403 
3404 	mutex_exit(&ds->ds_lock);
3405 	return (0);
3406 }
3407 
3408 static int
3409 dsl_dataset_user_release_check(void *arg1, void *tag, dmu_tx_t *tx)
3410 {
3411 	struct dsl_ds_releasearg *ra = arg1;
3412 	dsl_dataset_t *ds = ra->ds;
3413 	boolean_t might_destroy;
3414 	int error;
3415 
3416 	if (spa_version(ds->ds_dir->dd_pool->dp_spa) < SPA_VERSION_USERREFS)
3417 		return (ENOTSUP);
3418 
3419 	error = dsl_dataset_release_might_destroy(ds, ra->htag, &might_destroy);
3420 	if (error)
3421 		return (error);
3422 
3423 	if (might_destroy) {
3424 		struct dsl_ds_destroyarg dsda = {0};
3425 
3426 		if (dmu_tx_is_syncing(tx)) {
3427 			/*
3428 			 * If we're not prepared to remove the snapshot,
3429 			 * we can't allow the release to happen right now.
3430 			 */
3431 			if (!ra->own)
3432 				return (EBUSY);
3433 			if (ds->ds_objset) {
3434 				dmu_objset_evict(ds->ds_objset);
3435 				ds->ds_objset = NULL;
3436 			}
3437 		}
3438 		dsda.ds = ds;
3439 		dsda.releasing = B_TRUE;
3440 		return (dsl_dataset_destroy_check(&dsda, tag, tx));
3441 	}
3442 
3443 	return (0);
3444 }
3445 
3446 static void
3447 dsl_dataset_user_release_sync(void *arg1, void *tag, cred_t *cr, dmu_tx_t *tx)
3448 {
3449 	struct dsl_ds_releasearg *ra = arg1;
3450 	dsl_dataset_t *ds = ra->ds;
3451 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
3452 	objset_t *mos = dp->dp_meta_objset;
3453 	uint64_t zapobj;
3454 	uint64_t dsobj = ds->ds_object;
3455 	uint64_t refs;
3456 	int error;
3457 
3458 	mutex_enter(&ds->ds_lock);
3459 	ds->ds_userrefs--;
3460 	refs = ds->ds_userrefs;
3461 	mutex_exit(&ds->ds_lock);
3462 	error = dsl_pool_user_release(dp, ds->ds_object, ra->htag, tx);
3463 	VERIFY(error == 0 || error == ENOENT);
3464 	zapobj = ds->ds_phys->ds_userrefs_obj;
3465 	VERIFY(0 == zap_remove(mos, zapobj, ra->htag, tx));
3466 	if (ds->ds_userrefs == 0 && ds->ds_phys->ds_num_children == 1 &&
3467 	    DS_IS_DEFER_DESTROY(ds)) {
3468 		struct dsl_ds_destroyarg dsda = {0};
3469 
3470 		ASSERT(ra->own);
3471 		dsda.ds = ds;
3472 		dsda.releasing = B_TRUE;
3473 		/* We already did the destroy_check */
3474 		dsl_dataset_destroy_sync(&dsda, tag, cr, tx);
3475 	}
3476 
3477 	spa_history_internal_log(LOG_DS_USER_RELEASE,
3478 	    dp->dp_spa, tx, cr, "<%s> %lld dataset = %llu",
3479 	    ra->htag, (longlong_t)refs, dsobj);
3480 }
3481 
3482 static int
3483 dsl_dataset_user_release_one(char *dsname, void *arg)
3484 {
3485 	struct dsl_ds_holdarg *ha = arg;
3486 	struct dsl_ds_releasearg *ra;
3487 	dsl_dataset_t *ds;
3488 	int error;
3489 	void *dtag = ha->dstg;
3490 	char *name;
3491 	boolean_t own = B_FALSE;
3492 	boolean_t might_destroy;
3493 
3494 	/* alloc a buffer to hold dsname@snapname, plus the terminating NULL */
3495 	name = kmem_asprintf("%s@%s", dsname, ha->snapname);
3496 	error = dsl_dataset_hold(name, dtag, &ds);
3497 	strfree(name);
3498 	if (error == ENOENT && ha->recursive)
3499 		return (0);
3500 	(void) strcpy(ha->failed, dsname);
3501 	if (error)
3502 		return (error);
3503 
3504 	ha->gotone = B_TRUE;
3505 
3506 	ASSERT(dsl_dataset_is_snapshot(ds));
3507 
3508 	error = dsl_dataset_release_might_destroy(ds, ha->htag, &might_destroy);
3509 	if (error) {
3510 		dsl_dataset_rele(ds, dtag);
3511 		return (error);
3512 	}
3513 
3514 	if (might_destroy) {
3515 #ifdef _KERNEL
3516 		error = zfs_unmount_snap(name, NULL);
3517 		if (error) {
3518 			dsl_dataset_rele(ds, dtag);
3519 			return (error);
3520 		}
3521 #endif
3522 		if (!dsl_dataset_tryown(ds, B_TRUE, dtag)) {
3523 			dsl_dataset_rele(ds, dtag);
3524 			return (EBUSY);
3525 		} else {
3526 			own = B_TRUE;
3527 			dsl_dataset_make_exclusive(ds, dtag);
3528 		}
3529 	}
3530 
3531 	ra = kmem_alloc(sizeof (struct dsl_ds_releasearg), KM_SLEEP);
3532 	ra->ds = ds;
3533 	ra->htag = ha->htag;
3534 	ra->own = own;
3535 	dsl_sync_task_create(ha->dstg, dsl_dataset_user_release_check,
3536 	    dsl_dataset_user_release_sync, ra, dtag, 0);
3537 
3538 	return (0);
3539 }
3540 
3541 int
3542 dsl_dataset_user_release(char *dsname, char *snapname, char *htag,
3543     boolean_t recursive)
3544 {
3545 	struct dsl_ds_holdarg *ha;
3546 	dsl_sync_task_t *dst;
3547 	spa_t *spa;
3548 	int error;
3549 
3550 	ha = kmem_zalloc(sizeof (struct dsl_ds_holdarg), KM_SLEEP);
3551 
3552 	(void) strlcpy(ha->failed, dsname, sizeof (ha->failed));
3553 
3554 	error = spa_open(dsname, &spa, FTAG);
3555 	if (error) {
3556 		kmem_free(ha, sizeof (struct dsl_ds_holdarg));
3557 		return (error);
3558 	}
3559 
3560 	ha->dstg = dsl_sync_task_group_create(spa_get_dsl(spa));
3561 	ha->htag = htag;
3562 	ha->snapname = snapname;
3563 	ha->recursive = recursive;
3564 	if (recursive) {
3565 		error = dmu_objset_find(dsname, dsl_dataset_user_release_one,
3566 		    ha, DS_FIND_CHILDREN);
3567 	} else {
3568 		error = dsl_dataset_user_release_one(dsname, ha);
3569 	}
3570 	if (error == 0)
3571 		error = dsl_sync_task_group_wait(ha->dstg);
3572 
3573 	for (dst = list_head(&ha->dstg->dstg_tasks); dst;
3574 	    dst = list_next(&ha->dstg->dstg_tasks, dst)) {
3575 		struct dsl_ds_releasearg *ra = dst->dst_arg1;
3576 		dsl_dataset_t *ds = ra->ds;
3577 
3578 		if (dst->dst_err)
3579 			dsl_dataset_name(ds, ha->failed);
3580 
3581 		if (ra->own)
3582 			dsl_dataset_disown(ds, ha->dstg);
3583 		else
3584 			dsl_dataset_rele(ds, ha->dstg);
3585 
3586 		kmem_free(ra, sizeof (struct dsl_ds_releasearg));
3587 	}
3588 
3589 	if (error == 0 && recursive && !ha->gotone)
3590 		error = ENOENT;
3591 
3592 	if (error)
3593 		(void) strcpy(dsname, ha->failed);
3594 
3595 	dsl_sync_task_group_destroy(ha->dstg);
3596 	kmem_free(ha, sizeof (struct dsl_ds_holdarg));
3597 	spa_close(spa, FTAG);
3598 	return (error);
3599 }
3600 
3601 /*
3602  * Called at spa_load time to release a stale temporary user hold.
3603  */
3604 int
3605 dsl_dataset_user_release_tmp(dsl_pool_t *dp, uint64_t dsobj, char *htag)
3606 {
3607 	dsl_dataset_t *ds;
3608 	char *snap;
3609 	char *name;
3610 	int namelen;
3611 	int error;
3612 
3613 	rw_enter(&dp->dp_config_rwlock, RW_READER);
3614 	error = dsl_dataset_hold_obj(dp, dsobj, FTAG, &ds);
3615 	rw_exit(&dp->dp_config_rwlock);
3616 	if (error)
3617 		return (error);
3618 	namelen = dsl_dataset_namelen(ds)+1;
3619 	name = kmem_alloc(namelen, KM_SLEEP);
3620 	dsl_dataset_name(ds, name);
3621 	dsl_dataset_rele(ds, FTAG);
3622 
3623 	snap = strchr(name, '@');
3624 	*snap = '\0';
3625 	++snap;
3626 	return (dsl_dataset_user_release(name, snap, htag, B_FALSE));
3627 }
3628 
3629 int
3630 dsl_dataset_get_holds(const char *dsname, nvlist_t **nvp)
3631 {
3632 	dsl_dataset_t *ds;
3633 	int err;
3634 
3635 	err = dsl_dataset_hold(dsname, FTAG, &ds);
3636 	if (err)
3637 		return (err);
3638 
3639 	VERIFY(0 == nvlist_alloc(nvp, NV_UNIQUE_NAME, KM_SLEEP));
3640 	if (ds->ds_phys->ds_userrefs_obj != 0) {
3641 		zap_attribute_t *za;
3642 		zap_cursor_t zc;
3643 
3644 		za = kmem_alloc(sizeof (zap_attribute_t), KM_SLEEP);
3645 		for (zap_cursor_init(&zc, ds->ds_dir->dd_pool->dp_meta_objset,
3646 		    ds->ds_phys->ds_userrefs_obj);
3647 		    zap_cursor_retrieve(&zc, za) == 0;
3648 		    zap_cursor_advance(&zc)) {
3649 			VERIFY(0 == nvlist_add_uint64(*nvp, za->za_name,
3650 			    za->za_first_integer));
3651 		}
3652 		zap_cursor_fini(&zc);
3653 		kmem_free(za, sizeof (zap_attribute_t));
3654 	}
3655 	dsl_dataset_rele(ds, FTAG);
3656 	return (0);
3657 }
3658 
3659 /*
3660  * Note, this fuction is used as the callback for dmu_objset_find().  We
3661  * always return 0 so that we will continue to find and process
3662  * inconsistent datasets, even if we encounter an error trying to
3663  * process one of them.
3664  */
3665 /* ARGSUSED */
3666 int
3667 dsl_destroy_inconsistent(char *dsname, void *arg)
3668 {
3669 	dsl_dataset_t *ds;
3670 
3671 	if (dsl_dataset_own(dsname, B_TRUE, FTAG, &ds) == 0) {
3672 		if (DS_IS_INCONSISTENT(ds))
3673 			(void) dsl_dataset_destroy(ds, FTAG, B_FALSE);
3674 		else
3675 			dsl_dataset_disown(ds, FTAG);
3676 	}
3677 	return (0);
3678 }
3679