xref: /titanic_41/usr/src/uts/common/fs/zfs/dmu_objset.c (revision 391647d5ee9b25dc5307abb55f583388e08b2dd7)
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 2008 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 #pragma ident	"%Z%%M%	%I%	%E% SMI"
27 
28 #include <sys/cred.h>
29 #include <sys/zfs_context.h>
30 #include <sys/dmu_objset.h>
31 #include <sys/dsl_dir.h>
32 #include <sys/dsl_dataset.h>
33 #include <sys/dsl_prop.h>
34 #include <sys/dsl_pool.h>
35 #include <sys/dsl_synctask.h>
36 #include <sys/dsl_deleg.h>
37 #include <sys/dnode.h>
38 #include <sys/dbuf.h>
39 #include <sys/zvol.h>
40 #include <sys/dmu_tx.h>
41 #include <sys/zio_checksum.h>
42 #include <sys/zap.h>
43 #include <sys/zil.h>
44 #include <sys/dmu_impl.h>
45 #include <sys/zfs_ioctl.h>
46 
47 spa_t *
48 dmu_objset_spa(objset_t *os)
49 {
50 	return (os->os->os_spa);
51 }
52 
53 zilog_t *
54 dmu_objset_zil(objset_t *os)
55 {
56 	return (os->os->os_zil);
57 }
58 
59 dsl_pool_t *
60 dmu_objset_pool(objset_t *os)
61 {
62 	dsl_dataset_t *ds;
63 
64 	if ((ds = os->os->os_dsl_dataset) != NULL && ds->ds_dir)
65 		return (ds->ds_dir->dd_pool);
66 	else
67 		return (spa_get_dsl(os->os->os_spa));
68 }
69 
70 dsl_dataset_t *
71 dmu_objset_ds(objset_t *os)
72 {
73 	return (os->os->os_dsl_dataset);
74 }
75 
76 dmu_objset_type_t
77 dmu_objset_type(objset_t *os)
78 {
79 	return (os->os->os_phys->os_type);
80 }
81 
82 void
83 dmu_objset_name(objset_t *os, char *buf)
84 {
85 	dsl_dataset_name(os->os->os_dsl_dataset, buf);
86 }
87 
88 uint64_t
89 dmu_objset_id(objset_t *os)
90 {
91 	dsl_dataset_t *ds = os->os->os_dsl_dataset;
92 
93 	return (ds ? ds->ds_object : 0);
94 }
95 
96 static void
97 checksum_changed_cb(void *arg, uint64_t newval)
98 {
99 	objset_impl_t *osi = arg;
100 
101 	/*
102 	 * Inheritance should have been done by now.
103 	 */
104 	ASSERT(newval != ZIO_CHECKSUM_INHERIT);
105 
106 	osi->os_checksum = zio_checksum_select(newval, ZIO_CHECKSUM_ON_VALUE);
107 }
108 
109 static void
110 compression_changed_cb(void *arg, uint64_t newval)
111 {
112 	objset_impl_t *osi = arg;
113 
114 	/*
115 	 * Inheritance and range checking should have been done by now.
116 	 */
117 	ASSERT(newval != ZIO_COMPRESS_INHERIT);
118 
119 	osi->os_compress = zio_compress_select(newval, ZIO_COMPRESS_ON_VALUE);
120 }
121 
122 static void
123 copies_changed_cb(void *arg, uint64_t newval)
124 {
125 	objset_impl_t *osi = arg;
126 
127 	/*
128 	 * Inheritance and range checking should have been done by now.
129 	 */
130 	ASSERT(newval > 0);
131 	ASSERT(newval <= spa_max_replication(osi->os_spa));
132 
133 	osi->os_copies = newval;
134 }
135 
136 static void
137 primary_cache_changed_cb(void *arg, uint64_t newval)
138 {
139 	objset_impl_t *osi = arg;
140 
141 	/*
142 	 * Inheritance and range checking should have been done by now.
143 	 */
144 	ASSERT(newval == ZFS_CACHE_ALL || newval == ZFS_CACHE_NONE ||
145 	    newval == ZFS_CACHE_METADATA);
146 
147 	osi->os_primary_cache = newval;
148 }
149 
150 static void
151 secondary_cache_changed_cb(void *arg, uint64_t newval)
152 {
153 	objset_impl_t *osi = arg;
154 
155 	/*
156 	 * Inheritance and range checking should have been done by now.
157 	 */
158 	ASSERT(newval == ZFS_CACHE_ALL || newval == ZFS_CACHE_NONE ||
159 	    newval == ZFS_CACHE_METADATA);
160 
161 	osi->os_secondary_cache = newval;
162 }
163 
164 void
165 dmu_objset_byteswap(void *buf, size_t size)
166 {
167 	objset_phys_t *osp = buf;
168 
169 	ASSERT(size == sizeof (objset_phys_t));
170 	dnode_byteswap(&osp->os_meta_dnode);
171 	byteswap_uint64_array(&osp->os_zil_header, sizeof (zil_header_t));
172 	osp->os_type = BSWAP_64(osp->os_type);
173 }
174 
175 int
176 dmu_objset_open_impl(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
177     objset_impl_t **osip)
178 {
179 	objset_impl_t *osi;
180 	int i, err;
181 
182 	ASSERT(ds == NULL || MUTEX_HELD(&ds->ds_opening_lock));
183 
184 	osi = kmem_zalloc(sizeof (objset_impl_t), KM_SLEEP);
185 	osi->os.os = osi;
186 	osi->os_dsl_dataset = ds;
187 	osi->os_spa = spa;
188 	osi->os_rootbp = bp;
189 	if (!BP_IS_HOLE(osi->os_rootbp)) {
190 		uint32_t aflags = ARC_WAIT;
191 		zbookmark_t zb;
192 		zb.zb_objset = ds ? ds->ds_object : 0;
193 		zb.zb_object = 0;
194 		zb.zb_level = -1;
195 		zb.zb_blkid = 0;
196 		if (DMU_OS_IS_L2CACHEABLE(osi))
197 			aflags |= ARC_L2CACHE;
198 
199 		dprintf_bp(osi->os_rootbp, "reading %s", "");
200 		/*
201 		 * NB: when bprewrite scrub can change the bp,
202 		 * and this is called from dmu_objset_open_ds_os, the bp
203 		 * could change, and we'll need a lock.
204 		 */
205 		err = arc_read_nolock(NULL, spa, osi->os_rootbp,
206 		    arc_getbuf_func, &osi->os_phys_buf,
207 		    ZIO_PRIORITY_SYNC_READ, ZIO_FLAG_CANFAIL, &aflags, &zb);
208 		if (err) {
209 			kmem_free(osi, sizeof (objset_impl_t));
210 			/* convert checksum errors into IO errors */
211 			if (err == ECKSUM)
212 				err = EIO;
213 			return (err);
214 		}
215 		osi->os_phys = osi->os_phys_buf->b_data;
216 	} else {
217 		osi->os_phys_buf = arc_buf_alloc(spa, sizeof (objset_phys_t),
218 		    &osi->os_phys_buf, ARC_BUFC_METADATA);
219 		osi->os_phys = osi->os_phys_buf->b_data;
220 		bzero(osi->os_phys, sizeof (objset_phys_t));
221 	}
222 
223 	/*
224 	 * Note: the changed_cb will be called once before the register
225 	 * func returns, thus changing the checksum/compression from the
226 	 * default (fletcher2/off).  Snapshots don't need to know about
227 	 * checksum/compression/copies.
228 	 */
229 	if (ds) {
230 		err = dsl_prop_register(ds, "primarycache",
231 		    primary_cache_changed_cb, osi);
232 		if (err == 0)
233 			err = dsl_prop_register(ds, "secondarycache",
234 			    secondary_cache_changed_cb, osi);
235 		if (!dsl_dataset_is_snapshot(ds)) {
236 			if (err == 0)
237 				err = dsl_prop_register(ds, "checksum",
238 				    checksum_changed_cb, osi);
239 			if (err == 0)
240 				err = dsl_prop_register(ds, "compression",
241 				    compression_changed_cb, osi);
242 			if (err == 0)
243 				err = dsl_prop_register(ds, "copies",
244 				    copies_changed_cb, osi);
245 		}
246 		if (err) {
247 			VERIFY(arc_buf_remove_ref(osi->os_phys_buf,
248 			    &osi->os_phys_buf) == 1);
249 			kmem_free(osi, sizeof (objset_impl_t));
250 			return (err);
251 		}
252 	} else if (ds == NULL) {
253 		/* It's the meta-objset. */
254 		osi->os_checksum = ZIO_CHECKSUM_FLETCHER_4;
255 		osi->os_compress = ZIO_COMPRESS_LZJB;
256 		osi->os_copies = spa_max_replication(spa);
257 		osi->os_primary_cache = ZFS_CACHE_ALL;
258 		osi->os_secondary_cache = ZFS_CACHE_ALL;
259 	}
260 
261 	osi->os_zil_header = osi->os_phys->os_zil_header;
262 	osi->os_zil = zil_alloc(&osi->os, &osi->os_zil_header);
263 
264 	for (i = 0; i < TXG_SIZE; i++) {
265 		list_create(&osi->os_dirty_dnodes[i], sizeof (dnode_t),
266 		    offsetof(dnode_t, dn_dirty_link[i]));
267 		list_create(&osi->os_free_dnodes[i], sizeof (dnode_t),
268 		    offsetof(dnode_t, dn_dirty_link[i]));
269 	}
270 	list_create(&osi->os_dnodes, sizeof (dnode_t),
271 	    offsetof(dnode_t, dn_link));
272 	list_create(&osi->os_downgraded_dbufs, sizeof (dmu_buf_impl_t),
273 	    offsetof(dmu_buf_impl_t, db_link));
274 
275 	mutex_init(&osi->os_lock, NULL, MUTEX_DEFAULT, NULL);
276 	mutex_init(&osi->os_obj_lock, NULL, MUTEX_DEFAULT, NULL);
277 	mutex_init(&osi->os_user_ptr_lock, NULL, MUTEX_DEFAULT, NULL);
278 
279 	osi->os_meta_dnode = dnode_special_open(osi,
280 	    &osi->os_phys->os_meta_dnode, DMU_META_DNODE_OBJECT);
281 
282 	/*
283 	 * We should be the only thread trying to do this because we
284 	 * have ds_opening_lock
285 	 */
286 	if (ds) {
287 		VERIFY(NULL == dsl_dataset_set_user_ptr(ds, osi,
288 		    dmu_objset_evict));
289 	}
290 
291 	*osip = osi;
292 	return (0);
293 }
294 
295 static int
296 dmu_objset_open_ds_os(dsl_dataset_t *ds, objset_t *os, dmu_objset_type_t type)
297 {
298 	objset_impl_t *osi;
299 
300 	mutex_enter(&ds->ds_opening_lock);
301 	osi = dsl_dataset_get_user_ptr(ds);
302 	if (osi == NULL) {
303 		int err;
304 
305 		err = dmu_objset_open_impl(dsl_dataset_get_spa(ds),
306 		    ds, &ds->ds_phys->ds_bp, &osi);
307 		if (err) {
308 			mutex_exit(&ds->ds_opening_lock);
309 			return (err);
310 		}
311 	}
312 	mutex_exit(&ds->ds_opening_lock);
313 
314 	os->os = osi;
315 	os->os_mode = DS_MODE_NOHOLD;
316 
317 	if (type != DMU_OST_ANY && type != os->os->os_phys->os_type)
318 		return (EINVAL);
319 	return (0);
320 }
321 
322 int
323 dmu_objset_open_ds(dsl_dataset_t *ds, dmu_objset_type_t type, objset_t **osp)
324 {
325 	objset_t *os;
326 	int err;
327 
328 	os = kmem_alloc(sizeof (objset_t), KM_SLEEP);
329 	err = dmu_objset_open_ds_os(ds, os, type);
330 	if (err)
331 		kmem_free(os, sizeof (objset_t));
332 	else
333 		*osp = os;
334 	return (err);
335 }
336 
337 /* called from zpl */
338 int
339 dmu_objset_open(const char *name, dmu_objset_type_t type, int mode,
340     objset_t **osp)
341 {
342 	objset_t *os;
343 	dsl_dataset_t *ds;
344 	int err;
345 
346 	ASSERT(DS_MODE_TYPE(mode) == DS_MODE_USER ||
347 	    DS_MODE_TYPE(mode) == DS_MODE_OWNER);
348 
349 	os = kmem_alloc(sizeof (objset_t), KM_SLEEP);
350 	if (DS_MODE_TYPE(mode) == DS_MODE_USER)
351 		err = dsl_dataset_hold(name, os, &ds);
352 	else
353 		err = dsl_dataset_own(name, mode, os, &ds);
354 	if (err) {
355 		kmem_free(os, sizeof (objset_t));
356 		return (err);
357 	}
358 
359 	err = dmu_objset_open_ds_os(ds, os, type);
360 	if (err) {
361 		if (DS_MODE_TYPE(mode) == DS_MODE_USER)
362 			dsl_dataset_rele(ds, os);
363 		else
364 			dsl_dataset_disown(ds, os);
365 		kmem_free(os, sizeof (objset_t));
366 	} else {
367 		os->os_mode = mode;
368 		*osp = os;
369 	}
370 	return (err);
371 }
372 
373 void
374 dmu_objset_close(objset_t *os)
375 {
376 	ASSERT(DS_MODE_TYPE(os->os_mode) == DS_MODE_USER ||
377 	    DS_MODE_TYPE(os->os_mode) == DS_MODE_OWNER ||
378 	    DS_MODE_TYPE(os->os_mode) == DS_MODE_NOHOLD);
379 
380 	if (DS_MODE_TYPE(os->os_mode) == DS_MODE_USER)
381 		dsl_dataset_rele(os->os->os_dsl_dataset, os);
382 	else if (DS_MODE_TYPE(os->os_mode) == DS_MODE_OWNER)
383 		dsl_dataset_disown(os->os->os_dsl_dataset, os);
384 	kmem_free(os, sizeof (objset_t));
385 }
386 
387 int
388 dmu_objset_evict_dbufs(objset_t *os)
389 {
390 	objset_impl_t *osi = os->os;
391 	dnode_t *dn;
392 
393 	mutex_enter(&osi->os_lock);
394 
395 	/* process the mdn last, since the other dnodes have holds on it */
396 	list_remove(&osi->os_dnodes, osi->os_meta_dnode);
397 	list_insert_tail(&osi->os_dnodes, osi->os_meta_dnode);
398 
399 	/*
400 	 * Find the first dnode with holds.  We have to do this dance
401 	 * because dnode_add_ref() only works if you already have a
402 	 * hold.  If there are no holds then it has no dbufs so OK to
403 	 * skip.
404 	 */
405 	for (dn = list_head(&osi->os_dnodes);
406 	    dn && !dnode_add_ref(dn, FTAG);
407 	    dn = list_next(&osi->os_dnodes, dn))
408 		continue;
409 
410 	while (dn) {
411 		dnode_t *next_dn = dn;
412 
413 		do {
414 			next_dn = list_next(&osi->os_dnodes, next_dn);
415 		} while (next_dn && !dnode_add_ref(next_dn, FTAG));
416 
417 		mutex_exit(&osi->os_lock);
418 		dnode_evict_dbufs(dn);
419 		dnode_rele(dn, FTAG);
420 		mutex_enter(&osi->os_lock);
421 		dn = next_dn;
422 	}
423 	mutex_exit(&osi->os_lock);
424 	return (list_head(&osi->os_dnodes) != osi->os_meta_dnode);
425 }
426 
427 void
428 dmu_objset_evict(dsl_dataset_t *ds, void *arg)
429 {
430 	objset_impl_t *osi = arg;
431 	objset_t os;
432 	int i;
433 
434 	for (i = 0; i < TXG_SIZE; i++) {
435 		ASSERT(list_head(&osi->os_dirty_dnodes[i]) == NULL);
436 		ASSERT(list_head(&osi->os_free_dnodes[i]) == NULL);
437 	}
438 
439 	if (ds) {
440 		if (!dsl_dataset_is_snapshot(ds)) {
441 			VERIFY(0 == dsl_prop_unregister(ds, "checksum",
442 			    checksum_changed_cb, osi));
443 			VERIFY(0 == dsl_prop_unregister(ds, "compression",
444 			    compression_changed_cb, osi));
445 			VERIFY(0 == dsl_prop_unregister(ds, "copies",
446 			    copies_changed_cb, osi));
447 		}
448 		VERIFY(0 == dsl_prop_unregister(ds, "primarycache",
449 		    primary_cache_changed_cb, osi));
450 		VERIFY(0 == dsl_prop_unregister(ds, "secondarycache",
451 		    secondary_cache_changed_cb, osi));
452 	}
453 
454 	/*
455 	 * We should need only a single pass over the dnode list, since
456 	 * nothing can be added to the list at this point.
457 	 */
458 	os.os = osi;
459 	(void) dmu_objset_evict_dbufs(&os);
460 
461 	ASSERT3P(list_head(&osi->os_dnodes), ==, osi->os_meta_dnode);
462 	ASSERT3P(list_tail(&osi->os_dnodes), ==, osi->os_meta_dnode);
463 	ASSERT3P(list_head(&osi->os_meta_dnode->dn_dbufs), ==, NULL);
464 
465 	dnode_special_close(osi->os_meta_dnode);
466 	zil_free(osi->os_zil);
467 
468 	VERIFY(arc_buf_remove_ref(osi->os_phys_buf, &osi->os_phys_buf) == 1);
469 	mutex_destroy(&osi->os_lock);
470 	mutex_destroy(&osi->os_obj_lock);
471 	mutex_destroy(&osi->os_user_ptr_lock);
472 	kmem_free(osi, sizeof (objset_impl_t));
473 }
474 
475 /* called from dsl for meta-objset */
476 objset_impl_t *
477 dmu_objset_create_impl(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
478     dmu_objset_type_t type, dmu_tx_t *tx)
479 {
480 	objset_impl_t *osi;
481 	dnode_t *mdn;
482 
483 	ASSERT(dmu_tx_is_syncing(tx));
484 	if (ds)
485 		mutex_enter(&ds->ds_opening_lock);
486 	VERIFY(0 == dmu_objset_open_impl(spa, ds, bp, &osi));
487 	if (ds)
488 		mutex_exit(&ds->ds_opening_lock);
489 	mdn = osi->os_meta_dnode;
490 
491 	dnode_allocate(mdn, DMU_OT_DNODE, 1 << DNODE_BLOCK_SHIFT,
492 	    DN_MAX_INDBLKSHIFT, DMU_OT_NONE, 0, tx);
493 
494 	/*
495 	 * We don't want to have to increase the meta-dnode's nlevels
496 	 * later, because then we could do it in quescing context while
497 	 * we are also accessing it in open context.
498 	 *
499 	 * This precaution is not necessary for the MOS (ds == NULL),
500 	 * because the MOS is only updated in syncing context.
501 	 * This is most fortunate: the MOS is the only objset that
502 	 * needs to be synced multiple times as spa_sync() iterates
503 	 * to convergence, so minimizing its dn_nlevels matters.
504 	 */
505 	if (ds != NULL) {
506 		int levels = 1;
507 
508 		/*
509 		 * Determine the number of levels necessary for the meta-dnode
510 		 * to contain DN_MAX_OBJECT dnodes.
511 		 */
512 		while ((uint64_t)mdn->dn_nblkptr << (mdn->dn_datablkshift +
513 		    (levels - 1) * (mdn->dn_indblkshift - SPA_BLKPTRSHIFT)) <
514 		    DN_MAX_OBJECT * sizeof (dnode_phys_t))
515 			levels++;
516 
517 		mdn->dn_next_nlevels[tx->tx_txg & TXG_MASK] =
518 		    mdn->dn_nlevels = levels;
519 	}
520 
521 	ASSERT(type != DMU_OST_NONE);
522 	ASSERT(type != DMU_OST_ANY);
523 	ASSERT(type < DMU_OST_NUMTYPES);
524 	osi->os_phys->os_type = type;
525 
526 	dsl_dataset_dirty(ds, tx);
527 
528 	return (osi);
529 }
530 
531 struct oscarg {
532 	void (*userfunc)(objset_t *os, void *arg, cred_t *cr, dmu_tx_t *tx);
533 	void *userarg;
534 	dsl_dataset_t *clone_parent;
535 	const char *lastname;
536 	dmu_objset_type_t type;
537 	uint64_t flags;
538 };
539 
540 /*ARGSUSED*/
541 static int
542 dmu_objset_create_check(void *arg1, void *arg2, dmu_tx_t *tx)
543 {
544 	dsl_dir_t *dd = arg1;
545 	struct oscarg *oa = arg2;
546 	objset_t *mos = dd->dd_pool->dp_meta_objset;
547 	int err;
548 	uint64_t ddobj;
549 
550 	err = zap_lookup(mos, dd->dd_phys->dd_child_dir_zapobj,
551 	    oa->lastname, sizeof (uint64_t), 1, &ddobj);
552 	if (err != ENOENT)
553 		return (err ? err : EEXIST);
554 
555 	if (oa->clone_parent != NULL) {
556 		/*
557 		 * You can't clone across pools.
558 		 */
559 		if (oa->clone_parent->ds_dir->dd_pool != dd->dd_pool)
560 			return (EXDEV);
561 
562 		/*
563 		 * You can only clone snapshots, not the head datasets.
564 		 */
565 		if (oa->clone_parent->ds_phys->ds_num_children == 0)
566 			return (EINVAL);
567 	}
568 
569 	return (0);
570 }
571 
572 static void
573 dmu_objset_create_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
574 {
575 	dsl_dir_t *dd = arg1;
576 	struct oscarg *oa = arg2;
577 	dsl_dataset_t *ds;
578 	blkptr_t *bp;
579 	uint64_t dsobj;
580 
581 	ASSERT(dmu_tx_is_syncing(tx));
582 
583 	dsobj = dsl_dataset_create_sync(dd, oa->lastname,
584 	    oa->clone_parent, oa->flags, cr, tx);
585 
586 	VERIFY(0 == dsl_dataset_hold_obj(dd->dd_pool, dsobj, FTAG, &ds));
587 	bp = dsl_dataset_get_blkptr(ds);
588 	if (BP_IS_HOLE(bp)) {
589 		objset_impl_t *osi;
590 
591 		/* This is an empty dmu_objset; not a clone. */
592 		osi = dmu_objset_create_impl(dsl_dataset_get_spa(ds),
593 		    ds, bp, oa->type, tx);
594 
595 		if (oa->userfunc)
596 			oa->userfunc(&osi->os, oa->userarg, cr, tx);
597 	}
598 
599 	spa_history_internal_log(LOG_DS_CREATE, dd->dd_pool->dp_spa,
600 	    tx, cr, "dataset = %llu", dsobj);
601 
602 	dsl_dataset_rele(ds, FTAG);
603 }
604 
605 int
606 dmu_objset_create(const char *name, dmu_objset_type_t type,
607     objset_t *clone_parent, uint64_t flags,
608     void (*func)(objset_t *os, void *arg, cred_t *cr, dmu_tx_t *tx), void *arg)
609 {
610 	dsl_dir_t *pdd;
611 	const char *tail;
612 	int err = 0;
613 	struct oscarg oa = { 0 };
614 
615 	ASSERT(strchr(name, '@') == NULL);
616 	err = dsl_dir_open(name, FTAG, &pdd, &tail);
617 	if (err)
618 		return (err);
619 	if (tail == NULL) {
620 		dsl_dir_close(pdd, FTAG);
621 		return (EEXIST);
622 	}
623 
624 	dprintf("name=%s\n", name);
625 
626 	oa.userfunc = func;
627 	oa.userarg = arg;
628 	oa.lastname = tail;
629 	oa.type = type;
630 	oa.flags = flags;
631 
632 	if (clone_parent != NULL) {
633 		/*
634 		 * You can't clone to a different type.
635 		 */
636 		if (clone_parent->os->os_phys->os_type != type) {
637 			dsl_dir_close(pdd, FTAG);
638 			return (EINVAL);
639 		}
640 		oa.clone_parent = clone_parent->os->os_dsl_dataset;
641 	}
642 	err = dsl_sync_task_do(pdd->dd_pool, dmu_objset_create_check,
643 	    dmu_objset_create_sync, pdd, &oa, 5);
644 	dsl_dir_close(pdd, FTAG);
645 	return (err);
646 }
647 
648 int
649 dmu_objset_destroy(const char *name)
650 {
651 	objset_t *os;
652 	int error;
653 
654 	/*
655 	 * If it looks like we'll be able to destroy it, and there's
656 	 * an unplayed replay log sitting around, destroy the log.
657 	 * It would be nicer to do this in dsl_dataset_destroy_sync(),
658 	 * but the replay log objset is modified in open context.
659 	 */
660 	error = dmu_objset_open(name, DMU_OST_ANY,
661 	    DS_MODE_OWNER|DS_MODE_READONLY|DS_MODE_INCONSISTENT, &os);
662 	if (error == 0) {
663 		dsl_dataset_t *ds = os->os->os_dsl_dataset;
664 		zil_destroy(dmu_objset_zil(os), B_FALSE);
665 
666 		error = dsl_dataset_destroy(ds, os);
667 		/*
668 		 * dsl_dataset_destroy() closes the ds.
669 		 */
670 		kmem_free(os, sizeof (objset_t));
671 	}
672 
673 	return (error);
674 }
675 
676 /*
677  * This will close the objset.
678  */
679 int
680 dmu_objset_rollback(objset_t *os)
681 {
682 	int err;
683 	dsl_dataset_t *ds;
684 
685 	ds = os->os->os_dsl_dataset;
686 
687 	if (!dsl_dataset_tryown(ds, TRUE, os)) {
688 		dmu_objset_close(os);
689 		return (EBUSY);
690 	}
691 
692 	err = dsl_dataset_rollback(ds, os->os->os_phys->os_type);
693 
694 	/*
695 	 * NB: we close the objset manually because the rollback
696 	 * actually implicitly called dmu_objset_evict(), thus freeing
697 	 * the objset_impl_t.
698 	 */
699 	dsl_dataset_disown(ds, os);
700 	kmem_free(os, sizeof (objset_t));
701 	return (err);
702 }
703 
704 struct snaparg {
705 	dsl_sync_task_group_t *dstg;
706 	char *snapname;
707 	char failed[MAXPATHLEN];
708 	boolean_t checkperms;
709 	list_t objsets;
710 };
711 
712 struct osnode {
713 	list_node_t node;
714 	objset_t *os;
715 };
716 
717 static int
718 dmu_objset_snapshot_one(char *name, void *arg)
719 {
720 	struct snaparg *sn = arg;
721 	objset_t *os;
722 	int err;
723 
724 	(void) strcpy(sn->failed, name);
725 
726 	/*
727 	 * Check permissions only when requested.  This only applies when
728 	 * doing a recursive snapshot.  The permission checks for the starting
729 	 * dataset have already been performed in zfs_secpolicy_snapshot()
730 	 */
731 	if (sn->checkperms == B_TRUE &&
732 	    (err = zfs_secpolicy_snapshot_perms(name, CRED())))
733 		return (err);
734 
735 	err = dmu_objset_open(name, DMU_OST_ANY, DS_MODE_USER, &os);
736 	if (err != 0)
737 		return (err);
738 
739 	/* If the objset is in an inconsistent state, return busy */
740 	if (os->os->os_dsl_dataset->ds_phys->ds_flags & DS_FLAG_INCONSISTENT) {
741 		dmu_objset_close(os);
742 		return (EBUSY);
743 	}
744 
745 	/*
746 	 * NB: we need to wait for all in-flight changes to get to disk,
747 	 * so that we snapshot those changes.  zil_suspend does this as
748 	 * a side effect.
749 	 */
750 	err = zil_suspend(dmu_objset_zil(os));
751 	if (err == 0) {
752 		struct osnode *osn;
753 		dsl_sync_task_create(sn->dstg, dsl_dataset_snapshot_check,
754 		    dsl_dataset_snapshot_sync, os->os->os_dsl_dataset,
755 		    sn->snapname, 3);
756 		osn = kmem_alloc(sizeof (struct osnode), KM_SLEEP);
757 		osn->os = os;
758 		list_insert_tail(&sn->objsets, osn);
759 	} else {
760 		dmu_objset_close(os);
761 	}
762 
763 	return (err);
764 }
765 
766 int
767 dmu_objset_snapshot(char *fsname, char *snapname, boolean_t recursive)
768 {
769 	dsl_sync_task_t *dst;
770 	struct osnode *osn;
771 	struct snaparg sn = { 0 };
772 	spa_t *spa;
773 	int err;
774 
775 	(void) strcpy(sn.failed, fsname);
776 
777 	err = spa_open(fsname, &spa, FTAG);
778 	if (err)
779 		return (err);
780 
781 	sn.dstg = dsl_sync_task_group_create(spa_get_dsl(spa));
782 	sn.snapname = snapname;
783 	list_create(&sn.objsets, sizeof (struct osnode),
784 	    offsetof(struct osnode, node));
785 
786 	if (recursive) {
787 		sn.checkperms = B_TRUE;
788 		err = dmu_objset_find(fsname,
789 		    dmu_objset_snapshot_one, &sn, DS_FIND_CHILDREN);
790 	} else {
791 		sn.checkperms = B_FALSE;
792 		err = dmu_objset_snapshot_one(fsname, &sn);
793 	}
794 
795 	if (err)
796 		goto out;
797 
798 	err = dsl_sync_task_group_wait(sn.dstg);
799 
800 	for (dst = list_head(&sn.dstg->dstg_tasks); dst;
801 	    dst = list_next(&sn.dstg->dstg_tasks, dst)) {
802 		dsl_dataset_t *ds = dst->dst_arg1;
803 		if (dst->dst_err)
804 			dsl_dataset_name(ds, sn.failed);
805 	}
806 
807 out:
808 	while (osn = list_head(&sn.objsets)) {
809 		list_remove(&sn.objsets, osn);
810 		zil_resume(dmu_objset_zil(osn->os));
811 		dmu_objset_close(osn->os);
812 		kmem_free(osn, sizeof (struct osnode));
813 	}
814 	list_destroy(&sn.objsets);
815 
816 	if (err)
817 		(void) strcpy(fsname, sn.failed);
818 	dsl_sync_task_group_destroy(sn.dstg);
819 	spa_close(spa, FTAG);
820 	return (err);
821 }
822 
823 static void
824 dmu_objset_sync_dnodes(list_t *list, dmu_tx_t *tx)
825 {
826 	dnode_t *dn;
827 
828 	while (dn = list_head(list)) {
829 		ASSERT(dn->dn_object != DMU_META_DNODE_OBJECT);
830 		ASSERT(dn->dn_dbuf->db_data_pending);
831 		/*
832 		 * Initialize dn_zio outside dnode_sync()
833 		 * to accomodate meta-dnode
834 		 */
835 		dn->dn_zio = dn->dn_dbuf->db_data_pending->dr_zio;
836 		ASSERT(dn->dn_zio);
837 
838 		ASSERT3U(dn->dn_nlevels, <=, DN_MAX_LEVELS);
839 		list_remove(list, dn);
840 		dnode_sync(dn, tx);
841 	}
842 }
843 
844 /* ARGSUSED */
845 static void
846 ready(zio_t *zio, arc_buf_t *abuf, void *arg)
847 {
848 	objset_impl_t *os = arg;
849 	blkptr_t *bp = os->os_rootbp;
850 	dnode_phys_t *dnp = &os->os_phys->os_meta_dnode;
851 	int i;
852 
853 	ASSERT(bp == zio->io_bp);
854 
855 	/*
856 	 * Update rootbp fill count.
857 	 */
858 	bp->blk_fill = 1;	/* count the meta-dnode */
859 	for (i = 0; i < dnp->dn_nblkptr; i++)
860 		bp->blk_fill += dnp->dn_blkptr[i].blk_fill;
861 
862 	BP_SET_TYPE(bp, DMU_OT_OBJSET);
863 	BP_SET_LEVEL(bp, 0);
864 
865 	/* We must do this after we've set the bp's type and level */
866 	if (!DVA_EQUAL(BP_IDENTITY(bp),
867 	    BP_IDENTITY(&zio->io_bp_orig))) {
868 		if (zio->io_bp_orig.blk_birth == os->os_synctx->tx_txg)
869 			(void) dsl_dataset_block_kill(os->os_dsl_dataset,
870 			    &zio->io_bp_orig, NULL, os->os_synctx);
871 		dsl_dataset_block_born(os->os_dsl_dataset, bp, os->os_synctx);
872 	}
873 }
874 
875 /* called from dsl */
876 void
877 dmu_objset_sync(objset_impl_t *os, zio_t *pio, dmu_tx_t *tx)
878 {
879 	int txgoff;
880 	zbookmark_t zb;
881 	writeprops_t wp = { 0 };
882 	zio_t *zio;
883 	list_t *list;
884 	dbuf_dirty_record_t *dr;
885 
886 	dprintf_ds(os->os_dsl_dataset, "txg=%llu\n", tx->tx_txg);
887 
888 	ASSERT(dmu_tx_is_syncing(tx));
889 	/* XXX the write_done callback should really give us the tx... */
890 	os->os_synctx = tx;
891 
892 	if (os->os_dsl_dataset == NULL) {
893 		/*
894 		 * This is the MOS.  If we have upgraded,
895 		 * spa_max_replication() could change, so reset
896 		 * os_copies here.
897 		 */
898 		os->os_copies = spa_max_replication(os->os_spa);
899 	}
900 
901 	/*
902 	 * Create the root block IO
903 	 */
904 	zb.zb_objset = os->os_dsl_dataset ? os->os_dsl_dataset->ds_object : 0;
905 	zb.zb_object = 0;
906 	zb.zb_level = -1;
907 	zb.zb_blkid = 0;
908 	if (BP_IS_OLDER(os->os_rootbp, tx->tx_txg)) {
909 		(void) dsl_dataset_block_kill(os->os_dsl_dataset,
910 		    os->os_rootbp, pio, tx);
911 	}
912 	wp.wp_type = DMU_OT_OBJSET;
913 	wp.wp_copies = os->os_copies;
914 	wp.wp_level = (uint8_t)-1;
915 	wp.wp_oschecksum = os->os_checksum;
916 	wp.wp_oscompress = os->os_compress;
917 	arc_release(os->os_phys_buf, &os->os_phys_buf);
918 	zio = arc_write(pio, os->os_spa, &wp,
919 	    DMU_OS_IS_L2CACHEABLE(os), tx->tx_txg, os->os_rootbp,
920 	    os->os_phys_buf, ready, NULL, os, ZIO_PRIORITY_ASYNC_WRITE,
921 	    ZIO_FLAG_MUSTSUCCEED | ZIO_FLAG_METADATA, &zb);
922 
923 	/*
924 	 * Sync meta-dnode - the parent IO for the sync is the root block
925 	 */
926 	os->os_meta_dnode->dn_zio = zio;
927 	dnode_sync(os->os_meta_dnode, tx);
928 
929 	txgoff = tx->tx_txg & TXG_MASK;
930 
931 	dmu_objset_sync_dnodes(&os->os_free_dnodes[txgoff], tx);
932 	dmu_objset_sync_dnodes(&os->os_dirty_dnodes[txgoff], tx);
933 
934 	list = &os->os_meta_dnode->dn_dirty_records[txgoff];
935 	while (dr = list_head(list)) {
936 		ASSERT(dr->dr_dbuf->db_level == 0);
937 		list_remove(list, dr);
938 		if (dr->dr_zio)
939 			zio_nowait(dr->dr_zio);
940 	}
941 	/*
942 	 * Free intent log blocks up to this tx.
943 	 */
944 	zil_sync(os->os_zil, tx);
945 	os->os_phys->os_zil_header = os->os_zil_header;
946 	zio_nowait(zio);
947 }
948 
949 void
950 dmu_objset_space(objset_t *os, uint64_t *refdbytesp, uint64_t *availbytesp,
951     uint64_t *usedobjsp, uint64_t *availobjsp)
952 {
953 	dsl_dataset_space(os->os->os_dsl_dataset, refdbytesp, availbytesp,
954 	    usedobjsp, availobjsp);
955 }
956 
957 uint64_t
958 dmu_objset_fsid_guid(objset_t *os)
959 {
960 	return (dsl_dataset_fsid_guid(os->os->os_dsl_dataset));
961 }
962 
963 void
964 dmu_objset_fast_stat(objset_t *os, dmu_objset_stats_t *stat)
965 {
966 	stat->dds_type = os->os->os_phys->os_type;
967 	if (os->os->os_dsl_dataset)
968 		dsl_dataset_fast_stat(os->os->os_dsl_dataset, stat);
969 }
970 
971 void
972 dmu_objset_stats(objset_t *os, nvlist_t *nv)
973 {
974 	ASSERT(os->os->os_dsl_dataset ||
975 	    os->os->os_phys->os_type == DMU_OST_META);
976 
977 	if (os->os->os_dsl_dataset != NULL)
978 		dsl_dataset_stats(os->os->os_dsl_dataset, nv);
979 
980 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_TYPE,
981 	    os->os->os_phys->os_type);
982 }
983 
984 int
985 dmu_objset_is_snapshot(objset_t *os)
986 {
987 	if (os->os->os_dsl_dataset != NULL)
988 		return (dsl_dataset_is_snapshot(os->os->os_dsl_dataset));
989 	else
990 		return (B_FALSE);
991 }
992 
993 int
994 dmu_snapshot_realname(objset_t *os, char *name, char *real, int maxlen,
995     boolean_t *conflict)
996 {
997 	dsl_dataset_t *ds = os->os->os_dsl_dataset;
998 	uint64_t ignored;
999 
1000 	if (ds->ds_phys->ds_snapnames_zapobj == 0)
1001 		return (ENOENT);
1002 
1003 	return (zap_lookup_norm(ds->ds_dir->dd_pool->dp_meta_objset,
1004 	    ds->ds_phys->ds_snapnames_zapobj, name, 8, 1, &ignored, MT_FIRST,
1005 	    real, maxlen, conflict));
1006 }
1007 
1008 int
1009 dmu_snapshot_list_next(objset_t *os, int namelen, char *name,
1010     uint64_t *idp, uint64_t *offp, boolean_t *case_conflict)
1011 {
1012 	dsl_dataset_t *ds = os->os->os_dsl_dataset;
1013 	zap_cursor_t cursor;
1014 	zap_attribute_t attr;
1015 
1016 	if (ds->ds_phys->ds_snapnames_zapobj == 0)
1017 		return (ENOENT);
1018 
1019 	zap_cursor_init_serialized(&cursor,
1020 	    ds->ds_dir->dd_pool->dp_meta_objset,
1021 	    ds->ds_phys->ds_snapnames_zapobj, *offp);
1022 
1023 	if (zap_cursor_retrieve(&cursor, &attr) != 0) {
1024 		zap_cursor_fini(&cursor);
1025 		return (ENOENT);
1026 	}
1027 
1028 	if (strlen(attr.za_name) + 1 > namelen) {
1029 		zap_cursor_fini(&cursor);
1030 		return (ENAMETOOLONG);
1031 	}
1032 
1033 	(void) strcpy(name, attr.za_name);
1034 	if (idp)
1035 		*idp = attr.za_first_integer;
1036 	if (case_conflict)
1037 		*case_conflict = attr.za_normalization_conflict;
1038 	zap_cursor_advance(&cursor);
1039 	*offp = zap_cursor_serialize(&cursor);
1040 	zap_cursor_fini(&cursor);
1041 
1042 	return (0);
1043 }
1044 
1045 int
1046 dmu_dir_list_next(objset_t *os, int namelen, char *name,
1047     uint64_t *idp, uint64_t *offp)
1048 {
1049 	dsl_dir_t *dd = os->os->os_dsl_dataset->ds_dir;
1050 	zap_cursor_t cursor;
1051 	zap_attribute_t attr;
1052 
1053 	/* there is no next dir on a snapshot! */
1054 	if (os->os->os_dsl_dataset->ds_object !=
1055 	    dd->dd_phys->dd_head_dataset_obj)
1056 		return (ENOENT);
1057 
1058 	zap_cursor_init_serialized(&cursor,
1059 	    dd->dd_pool->dp_meta_objset,
1060 	    dd->dd_phys->dd_child_dir_zapobj, *offp);
1061 
1062 	if (zap_cursor_retrieve(&cursor, &attr) != 0) {
1063 		zap_cursor_fini(&cursor);
1064 		return (ENOENT);
1065 	}
1066 
1067 	if (strlen(attr.za_name) + 1 > namelen) {
1068 		zap_cursor_fini(&cursor);
1069 		return (ENAMETOOLONG);
1070 	}
1071 
1072 	(void) strcpy(name, attr.za_name);
1073 	if (idp)
1074 		*idp = attr.za_first_integer;
1075 	zap_cursor_advance(&cursor);
1076 	*offp = zap_cursor_serialize(&cursor);
1077 	zap_cursor_fini(&cursor);
1078 
1079 	return (0);
1080 }
1081 
1082 struct findarg {
1083 	int (*func)(char *, void *);
1084 	void *arg;
1085 };
1086 
1087 /* ARGSUSED */
1088 static int
1089 findfunc(spa_t *spa, uint64_t dsobj, const char *dsname, void *arg)
1090 {
1091 	struct findarg *fa = arg;
1092 	return (fa->func((char *)dsname, fa->arg));
1093 }
1094 
1095 /*
1096  * Find all objsets under name, and for each, call 'func(child_name, arg)'.
1097  * Perhaps change all callers to use dmu_objset_find_spa()?
1098  */
1099 int
1100 dmu_objset_find(char *name, int func(char *, void *), void *arg, int flags)
1101 {
1102 	struct findarg fa;
1103 	fa.func = func;
1104 	fa.arg = arg;
1105 	return (dmu_objset_find_spa(NULL, name, findfunc, &fa, flags));
1106 }
1107 
1108 /*
1109  * Find all objsets under name, call func on each
1110  */
1111 int
1112 dmu_objset_find_spa(spa_t *spa, const char *name,
1113     int func(spa_t *, uint64_t, const char *, void *), void *arg, int flags)
1114 {
1115 	dsl_dir_t *dd;
1116 	dsl_pool_t *dp;
1117 	dsl_dataset_t *ds;
1118 	zap_cursor_t zc;
1119 	zap_attribute_t *attr;
1120 	char *child;
1121 	uint64_t thisobj;
1122 	int err;
1123 
1124 	if (name == NULL)
1125 		name = spa_name(spa);
1126 	err = dsl_dir_open_spa(spa, name, FTAG, &dd, NULL);
1127 	if (err)
1128 		return (err);
1129 
1130 	/* Don't visit hidden ($MOS & $ORIGIN) objsets. */
1131 	if (dd->dd_myname[0] == '$') {
1132 		dsl_dir_close(dd, FTAG);
1133 		return (0);
1134 	}
1135 
1136 	thisobj = dd->dd_phys->dd_head_dataset_obj;
1137 	attr = kmem_alloc(sizeof (zap_attribute_t), KM_SLEEP);
1138 	dp = dd->dd_pool;
1139 
1140 	/*
1141 	 * Iterate over all children.
1142 	 */
1143 	if (flags & DS_FIND_CHILDREN) {
1144 		for (zap_cursor_init(&zc, dp->dp_meta_objset,
1145 		    dd->dd_phys->dd_child_dir_zapobj);
1146 		    zap_cursor_retrieve(&zc, attr) == 0;
1147 		    (void) zap_cursor_advance(&zc)) {
1148 			ASSERT(attr->za_integer_length == sizeof (uint64_t));
1149 			ASSERT(attr->za_num_integers == 1);
1150 
1151 			child = kmem_alloc(MAXPATHLEN, KM_SLEEP);
1152 			(void) strcpy(child, name);
1153 			(void) strcat(child, "/");
1154 			(void) strcat(child, attr->za_name);
1155 			err = dmu_objset_find_spa(spa, child, func, arg, flags);
1156 			kmem_free(child, MAXPATHLEN);
1157 			if (err)
1158 				break;
1159 		}
1160 		zap_cursor_fini(&zc);
1161 
1162 		if (err) {
1163 			dsl_dir_close(dd, FTAG);
1164 			kmem_free(attr, sizeof (zap_attribute_t));
1165 			return (err);
1166 		}
1167 	}
1168 
1169 	/*
1170 	 * Iterate over all snapshots.
1171 	 */
1172 	if (flags & DS_FIND_SNAPSHOTS) {
1173 		if (!dsl_pool_sync_context(dp))
1174 			rw_enter(&dp->dp_config_rwlock, RW_READER);
1175 		err = dsl_dataset_hold_obj(dp, thisobj, FTAG, &ds);
1176 		if (!dsl_pool_sync_context(dp))
1177 			rw_exit(&dp->dp_config_rwlock);
1178 
1179 		if (err == 0) {
1180 			uint64_t snapobj = ds->ds_phys->ds_snapnames_zapobj;
1181 			dsl_dataset_rele(ds, FTAG);
1182 
1183 			for (zap_cursor_init(&zc, dp->dp_meta_objset, snapobj);
1184 			    zap_cursor_retrieve(&zc, attr) == 0;
1185 			    (void) zap_cursor_advance(&zc)) {
1186 				ASSERT(attr->za_integer_length ==
1187 				    sizeof (uint64_t));
1188 				ASSERT(attr->za_num_integers == 1);
1189 
1190 				child = kmem_alloc(MAXPATHLEN, KM_SLEEP);
1191 				(void) strcpy(child, name);
1192 				(void) strcat(child, "@");
1193 				(void) strcat(child, attr->za_name);
1194 				err = func(spa, attr->za_first_integer,
1195 				    child, arg);
1196 				kmem_free(child, MAXPATHLEN);
1197 				if (err)
1198 					break;
1199 			}
1200 			zap_cursor_fini(&zc);
1201 		}
1202 	}
1203 
1204 	dsl_dir_close(dd, FTAG);
1205 	kmem_free(attr, sizeof (zap_attribute_t));
1206 
1207 	if (err)
1208 		return (err);
1209 
1210 	/*
1211 	 * Apply to self if appropriate.
1212 	 */
1213 	err = func(spa, thisobj, name, arg);
1214 	return (err);
1215 }
1216 
1217 void
1218 dmu_objset_set_user(objset_t *os, void *user_ptr)
1219 {
1220 	ASSERT(MUTEX_HELD(&os->os->os_user_ptr_lock));
1221 	os->os->os_user_ptr = user_ptr;
1222 }
1223 
1224 void *
1225 dmu_objset_get_user(objset_t *os)
1226 {
1227 	ASSERT(MUTEX_HELD(&os->os->os_user_ptr_lock));
1228 	return (os->os->os_user_ptr);
1229 }
1230