xref: /illumos-gate/usr/src/uts/common/fs/zfs/dmu_objset.c (revision 6142aa70a0f83a099e021c8378a46f4f8f374bb9)
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 2010 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 #include <sys/cred.h>
27 #include <sys/zfs_context.h>
28 #include <sys/dmu_objset.h>
29 #include <sys/dsl_dir.h>
30 #include <sys/dsl_dataset.h>
31 #include <sys/dsl_prop.h>
32 #include <sys/dsl_pool.h>
33 #include <sys/dsl_synctask.h>
34 #include <sys/dsl_deleg.h>
35 #include <sys/dnode.h>
36 #include <sys/dbuf.h>
37 #include <sys/zvol.h>
38 #include <sys/dmu_tx.h>
39 #include <sys/zap.h>
40 #include <sys/zil.h>
41 #include <sys/dmu_impl.h>
42 #include <sys/zfs_ioctl.h>
43 #include <sys/sunddi.h>
44 
45 spa_t *
46 dmu_objset_spa(objset_t *os)
47 {
48 	return (os->os_spa);
49 }
50 
51 zilog_t *
52 dmu_objset_zil(objset_t *os)
53 {
54 	return (os->os_zil);
55 }
56 
57 dsl_pool_t *
58 dmu_objset_pool(objset_t *os)
59 {
60 	dsl_dataset_t *ds;
61 
62 	if ((ds = os->os_dsl_dataset) != NULL && ds->ds_dir)
63 		return (ds->ds_dir->dd_pool);
64 	else
65 		return (spa_get_dsl(os->os_spa));
66 }
67 
68 dsl_dataset_t *
69 dmu_objset_ds(objset_t *os)
70 {
71 	return (os->os_dsl_dataset);
72 }
73 
74 dmu_objset_type_t
75 dmu_objset_type(objset_t *os)
76 {
77 	return (os->os_phys->os_type);
78 }
79 
80 void
81 dmu_objset_name(objset_t *os, char *buf)
82 {
83 	dsl_dataset_name(os->os_dsl_dataset, buf);
84 }
85 
86 uint64_t
87 dmu_objset_id(objset_t *os)
88 {
89 	dsl_dataset_t *ds = os->os_dsl_dataset;
90 
91 	return (ds ? ds->ds_object : 0);
92 }
93 
94 uint64_t
95 dmu_objset_logbias(objset_t *os)
96 {
97 	return (os->os_logbias);
98 }
99 
100 static void
101 checksum_changed_cb(void *arg, uint64_t newval)
102 {
103 	objset_t *os = arg;
104 
105 	/*
106 	 * Inheritance should have been done by now.
107 	 */
108 	ASSERT(newval != ZIO_CHECKSUM_INHERIT);
109 
110 	os->os_checksum = zio_checksum_select(newval, ZIO_CHECKSUM_ON_VALUE);
111 }
112 
113 static void
114 compression_changed_cb(void *arg, uint64_t newval)
115 {
116 	objset_t *os = arg;
117 
118 	/*
119 	 * Inheritance and range checking should have been done by now.
120 	 */
121 	ASSERT(newval != ZIO_COMPRESS_INHERIT);
122 
123 	os->os_compress = zio_compress_select(newval, ZIO_COMPRESS_ON_VALUE);
124 }
125 
126 static void
127 copies_changed_cb(void *arg, uint64_t newval)
128 {
129 	objset_t *os = arg;
130 
131 	/*
132 	 * Inheritance and range checking should have been done by now.
133 	 */
134 	ASSERT(newval > 0);
135 	ASSERT(newval <= spa_max_replication(os->os_spa));
136 
137 	os->os_copies = newval;
138 }
139 
140 static void
141 dedup_changed_cb(void *arg, uint64_t newval)
142 {
143 	objset_t *os = arg;
144 	spa_t *spa = os->os_spa;
145 	enum zio_checksum checksum;
146 
147 	/*
148 	 * Inheritance should have been done by now.
149 	 */
150 	ASSERT(newval != ZIO_CHECKSUM_INHERIT);
151 
152 	checksum = zio_checksum_dedup_select(spa, newval, ZIO_CHECKSUM_OFF);
153 
154 	os->os_dedup_checksum = checksum & ZIO_CHECKSUM_MASK;
155 	os->os_dedup_verify = !!(checksum & ZIO_CHECKSUM_VERIFY);
156 }
157 
158 static void
159 primary_cache_changed_cb(void *arg, uint64_t newval)
160 {
161 	objset_t *os = arg;
162 
163 	/*
164 	 * Inheritance and range checking should have been done by now.
165 	 */
166 	ASSERT(newval == ZFS_CACHE_ALL || newval == ZFS_CACHE_NONE ||
167 	    newval == ZFS_CACHE_METADATA);
168 
169 	os->os_primary_cache = newval;
170 }
171 
172 static void
173 secondary_cache_changed_cb(void *arg, uint64_t newval)
174 {
175 	objset_t *os = arg;
176 
177 	/*
178 	 * Inheritance and range checking should have been done by now.
179 	 */
180 	ASSERT(newval == ZFS_CACHE_ALL || newval == ZFS_CACHE_NONE ||
181 	    newval == ZFS_CACHE_METADATA);
182 
183 	os->os_secondary_cache = newval;
184 }
185 
186 static void
187 logbias_changed_cb(void *arg, uint64_t newval)
188 {
189 	objset_t *os = arg;
190 
191 	ASSERT(newval == ZFS_LOGBIAS_LATENCY ||
192 	    newval == ZFS_LOGBIAS_THROUGHPUT);
193 	os->os_logbias = newval;
194 	if (os->os_zil)
195 		zil_set_logbias(os->os_zil, newval);
196 }
197 
198 void
199 dmu_objset_byteswap(void *buf, size_t size)
200 {
201 	objset_phys_t *osp = buf;
202 
203 	ASSERT(size == OBJSET_OLD_PHYS_SIZE || size == sizeof (objset_phys_t));
204 	dnode_byteswap(&osp->os_meta_dnode);
205 	byteswap_uint64_array(&osp->os_zil_header, sizeof (zil_header_t));
206 	osp->os_type = BSWAP_64(osp->os_type);
207 	osp->os_flags = BSWAP_64(osp->os_flags);
208 	if (size == sizeof (objset_phys_t)) {
209 		dnode_byteswap(&osp->os_userused_dnode);
210 		dnode_byteswap(&osp->os_groupused_dnode);
211 	}
212 }
213 
214 int
215 dmu_objset_open_impl(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
216     objset_t **osp)
217 {
218 	objset_t *os;
219 	int i, err;
220 
221 	ASSERT(ds == NULL || MUTEX_HELD(&ds->ds_opening_lock));
222 
223 	os = kmem_zalloc(sizeof (objset_t), KM_SLEEP);
224 	os->os_dsl_dataset = ds;
225 	os->os_spa = spa;
226 	os->os_rootbp = bp;
227 	if (!BP_IS_HOLE(os->os_rootbp)) {
228 		uint32_t aflags = ARC_WAIT;
229 		zbookmark_t zb;
230 		SET_BOOKMARK(&zb, ds ? ds->ds_object : DMU_META_OBJSET,
231 		    ZB_ROOT_OBJECT, ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
232 
233 		if (DMU_OS_IS_L2CACHEABLE(os))
234 			aflags |= ARC_L2CACHE;
235 
236 		dprintf_bp(os->os_rootbp, "reading %s", "");
237 		/*
238 		 * NB: when bprewrite scrub can change the bp,
239 		 * and this is called from dmu_objset_open_ds_os, the bp
240 		 * could change, and we'll need a lock.
241 		 */
242 		err = arc_read_nolock(NULL, spa, os->os_rootbp,
243 		    arc_getbuf_func, &os->os_phys_buf,
244 		    ZIO_PRIORITY_SYNC_READ, ZIO_FLAG_CANFAIL, &aflags, &zb);
245 		if (err) {
246 			kmem_free(os, sizeof (objset_t));
247 			/* convert checksum errors into IO errors */
248 			if (err == ECKSUM)
249 				err = EIO;
250 			return (err);
251 		}
252 
253 		/* Increase the blocksize if we are permitted. */
254 		if (spa_version(spa) >= SPA_VERSION_USERSPACE &&
255 		    arc_buf_size(os->os_phys_buf) < sizeof (objset_phys_t)) {
256 			arc_buf_t *buf = arc_buf_alloc(spa,
257 			    sizeof (objset_phys_t), &os->os_phys_buf,
258 			    ARC_BUFC_METADATA);
259 			bzero(buf->b_data, sizeof (objset_phys_t));
260 			bcopy(os->os_phys_buf->b_data, buf->b_data,
261 			    arc_buf_size(os->os_phys_buf));
262 			(void) arc_buf_remove_ref(os->os_phys_buf,
263 			    &os->os_phys_buf);
264 			os->os_phys_buf = buf;
265 		}
266 
267 		os->os_phys = os->os_phys_buf->b_data;
268 		os->os_flags = os->os_phys->os_flags;
269 	} else {
270 		int size = spa_version(spa) >= SPA_VERSION_USERSPACE ?
271 		    sizeof (objset_phys_t) : OBJSET_OLD_PHYS_SIZE;
272 		os->os_phys_buf = arc_buf_alloc(spa, size,
273 		    &os->os_phys_buf, ARC_BUFC_METADATA);
274 		os->os_phys = os->os_phys_buf->b_data;
275 		bzero(os->os_phys, size);
276 	}
277 
278 	/*
279 	 * Note: the changed_cb will be called once before the register
280 	 * func returns, thus changing the checksum/compression from the
281 	 * default (fletcher2/off).  Snapshots don't need to know about
282 	 * checksum/compression/copies.
283 	 */
284 	if (ds) {
285 		err = dsl_prop_register(ds, "primarycache",
286 		    primary_cache_changed_cb, os);
287 		if (err == 0)
288 			err = dsl_prop_register(ds, "secondarycache",
289 			    secondary_cache_changed_cb, os);
290 		if (!dsl_dataset_is_snapshot(ds)) {
291 			if (err == 0)
292 				err = dsl_prop_register(ds, "checksum",
293 				    checksum_changed_cb, os);
294 			if (err == 0)
295 				err = dsl_prop_register(ds, "compression",
296 				    compression_changed_cb, os);
297 			if (err == 0)
298 				err = dsl_prop_register(ds, "copies",
299 				    copies_changed_cb, os);
300 			if (err == 0)
301 				err = dsl_prop_register(ds, "dedup",
302 				    dedup_changed_cb, os);
303 			if (err == 0)
304 				err = dsl_prop_register(ds, "logbias",
305 				    logbias_changed_cb, os);
306 		}
307 		if (err) {
308 			VERIFY(arc_buf_remove_ref(os->os_phys_buf,
309 			    &os->os_phys_buf) == 1);
310 			kmem_free(os, sizeof (objset_t));
311 			return (err);
312 		}
313 	} else if (ds == NULL) {
314 		/* It's the meta-objset. */
315 		os->os_checksum = ZIO_CHECKSUM_FLETCHER_4;
316 		os->os_compress = ZIO_COMPRESS_LZJB;
317 		os->os_copies = spa_max_replication(spa);
318 		os->os_dedup_checksum = ZIO_CHECKSUM_OFF;
319 		os->os_dedup_verify = 0;
320 		os->os_logbias = 0;
321 		os->os_primary_cache = ZFS_CACHE_ALL;
322 		os->os_secondary_cache = ZFS_CACHE_ALL;
323 	}
324 
325 	os->os_zil_header = os->os_phys->os_zil_header;
326 	os->os_zil = zil_alloc(os, &os->os_zil_header);
327 
328 	for (i = 0; i < TXG_SIZE; i++) {
329 		list_create(&os->os_dirty_dnodes[i], sizeof (dnode_t),
330 		    offsetof(dnode_t, dn_dirty_link[i]));
331 		list_create(&os->os_free_dnodes[i], sizeof (dnode_t),
332 		    offsetof(dnode_t, dn_dirty_link[i]));
333 	}
334 	list_create(&os->os_dnodes, sizeof (dnode_t),
335 	    offsetof(dnode_t, dn_link));
336 	list_create(&os->os_downgraded_dbufs, sizeof (dmu_buf_impl_t),
337 	    offsetof(dmu_buf_impl_t, db_link));
338 
339 	mutex_init(&os->os_lock, NULL, MUTEX_DEFAULT, NULL);
340 	mutex_init(&os->os_obj_lock, NULL, MUTEX_DEFAULT, NULL);
341 	mutex_init(&os->os_user_ptr_lock, NULL, MUTEX_DEFAULT, NULL);
342 
343 	os->os_meta_dnode = dnode_special_open(os,
344 	    &os->os_phys->os_meta_dnode, DMU_META_DNODE_OBJECT);
345 	if (arc_buf_size(os->os_phys_buf) >= sizeof (objset_phys_t)) {
346 		os->os_userused_dnode = dnode_special_open(os,
347 		    &os->os_phys->os_userused_dnode, DMU_USERUSED_OBJECT);
348 		os->os_groupused_dnode = dnode_special_open(os,
349 		    &os->os_phys->os_groupused_dnode, DMU_GROUPUSED_OBJECT);
350 	}
351 
352 	/*
353 	 * We should be the only thread trying to do this because we
354 	 * have ds_opening_lock
355 	 */
356 	if (ds) {
357 		mutex_enter(&ds->ds_lock);
358 		ASSERT(ds->ds_objset == NULL);
359 		ds->ds_objset = os;
360 		mutex_exit(&ds->ds_lock);
361 	}
362 
363 	*osp = os;
364 	return (0);
365 }
366 
367 int
368 dmu_objset_from_ds(dsl_dataset_t *ds, objset_t **osp)
369 {
370 	int err = 0;
371 
372 	mutex_enter(&ds->ds_opening_lock);
373 	*osp = ds->ds_objset;
374 	if (*osp == NULL) {
375 		err = dmu_objset_open_impl(dsl_dataset_get_spa(ds),
376 		    ds, &ds->ds_phys->ds_bp, osp);
377 	}
378 	mutex_exit(&ds->ds_opening_lock);
379 	return (err);
380 }
381 
382 /* called from zpl */
383 int
384 dmu_objset_hold(const char *name, void *tag, objset_t **osp)
385 {
386 	dsl_dataset_t *ds;
387 	int err;
388 
389 	err = dsl_dataset_hold(name, tag, &ds);
390 	if (err)
391 		return (err);
392 
393 	err = dmu_objset_from_ds(ds, osp);
394 	if (err)
395 		dsl_dataset_rele(ds, tag);
396 
397 	return (err);
398 }
399 
400 /* called from zpl */
401 int
402 dmu_objset_own(const char *name, dmu_objset_type_t type,
403     boolean_t readonly, void *tag, objset_t **osp)
404 {
405 	dsl_dataset_t *ds;
406 	int err;
407 
408 	err = dsl_dataset_own(name, B_FALSE, tag, &ds);
409 	if (err)
410 		return (err);
411 
412 	err = dmu_objset_from_ds(ds, osp);
413 	if (err) {
414 		dsl_dataset_disown(ds, tag);
415 	} else if (type != DMU_OST_ANY && type != (*osp)->os_phys->os_type) {
416 		dmu_objset_disown(*osp, tag);
417 		return (EINVAL);
418 	} else if (!readonly && dsl_dataset_is_snapshot(ds)) {
419 		dmu_objset_disown(*osp, tag);
420 		return (EROFS);
421 	}
422 	return (err);
423 }
424 
425 void
426 dmu_objset_rele(objset_t *os, void *tag)
427 {
428 	dsl_dataset_rele(os->os_dsl_dataset, tag);
429 }
430 
431 void
432 dmu_objset_disown(objset_t *os, void *tag)
433 {
434 	dsl_dataset_disown(os->os_dsl_dataset, tag);
435 }
436 
437 int
438 dmu_objset_evict_dbufs(objset_t *os)
439 {
440 	dnode_t *dn;
441 
442 	mutex_enter(&os->os_lock);
443 
444 	/* process the mdn last, since the other dnodes have holds on it */
445 	list_remove(&os->os_dnodes, os->os_meta_dnode);
446 	list_insert_tail(&os->os_dnodes, os->os_meta_dnode);
447 
448 	/*
449 	 * Find the first dnode with holds.  We have to do this dance
450 	 * because dnode_add_ref() only works if you already have a
451 	 * hold.  If there are no holds then it has no dbufs so OK to
452 	 * skip.
453 	 */
454 	for (dn = list_head(&os->os_dnodes);
455 	    dn && !dnode_add_ref(dn, FTAG);
456 	    dn = list_next(&os->os_dnodes, dn))
457 		continue;
458 
459 	while (dn) {
460 		dnode_t *next_dn = dn;
461 
462 		do {
463 			next_dn = list_next(&os->os_dnodes, next_dn);
464 		} while (next_dn && !dnode_add_ref(next_dn, FTAG));
465 
466 		mutex_exit(&os->os_lock);
467 		dnode_evict_dbufs(dn);
468 		dnode_rele(dn, FTAG);
469 		mutex_enter(&os->os_lock);
470 		dn = next_dn;
471 	}
472 	mutex_exit(&os->os_lock);
473 	return (list_head(&os->os_dnodes) != os->os_meta_dnode);
474 }
475 
476 void
477 dmu_objset_evict(objset_t *os)
478 {
479 	dsl_dataset_t *ds = os->os_dsl_dataset;
480 
481 	for (int t = 0; t < TXG_SIZE; t++)
482 		ASSERT(!dmu_objset_is_dirty(os, t));
483 
484 	if (ds) {
485 		if (!dsl_dataset_is_snapshot(ds)) {
486 			VERIFY(0 == dsl_prop_unregister(ds, "checksum",
487 			    checksum_changed_cb, os));
488 			VERIFY(0 == dsl_prop_unregister(ds, "compression",
489 			    compression_changed_cb, os));
490 			VERIFY(0 == dsl_prop_unregister(ds, "copies",
491 			    copies_changed_cb, os));
492 			VERIFY(0 == dsl_prop_unregister(ds, "dedup",
493 			    dedup_changed_cb, os));
494 			VERIFY(0 == dsl_prop_unregister(ds, "logbias",
495 			    logbias_changed_cb, os));
496 		}
497 		VERIFY(0 == dsl_prop_unregister(ds, "primarycache",
498 		    primary_cache_changed_cb, os));
499 		VERIFY(0 == dsl_prop_unregister(ds, "secondarycache",
500 		    secondary_cache_changed_cb, os));
501 	}
502 
503 	/*
504 	 * We should need only a single pass over the dnode list, since
505 	 * nothing can be added to the list at this point.
506 	 */
507 	(void) dmu_objset_evict_dbufs(os);
508 
509 	dnode_special_close(os->os_meta_dnode);
510 	if (os->os_userused_dnode) {
511 		dnode_special_close(os->os_userused_dnode);
512 		dnode_special_close(os->os_groupused_dnode);
513 	}
514 	zil_free(os->os_zil);
515 
516 	ASSERT3P(list_head(&os->os_dnodes), ==, NULL);
517 
518 	VERIFY(arc_buf_remove_ref(os->os_phys_buf, &os->os_phys_buf) == 1);
519 	mutex_destroy(&os->os_lock);
520 	mutex_destroy(&os->os_obj_lock);
521 	mutex_destroy(&os->os_user_ptr_lock);
522 	kmem_free(os, sizeof (objset_t));
523 }
524 
525 timestruc_t
526 dmu_objset_snap_cmtime(objset_t *os)
527 {
528 	return (dsl_dir_snap_cmtime(os->os_dsl_dataset->ds_dir));
529 }
530 
531 /* called from dsl for meta-objset */
532 objset_t *
533 dmu_objset_create_impl(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
534     dmu_objset_type_t type, dmu_tx_t *tx)
535 {
536 	objset_t *os;
537 	dnode_t *mdn;
538 
539 	ASSERT(dmu_tx_is_syncing(tx));
540 	if (ds)
541 		mutex_enter(&ds->ds_opening_lock);
542 	VERIFY(0 == dmu_objset_open_impl(spa, ds, bp, &os));
543 	if (ds)
544 		mutex_exit(&ds->ds_opening_lock);
545 	mdn = os->os_meta_dnode;
546 
547 	dnode_allocate(mdn, DMU_OT_DNODE, 1 << DNODE_BLOCK_SHIFT,
548 	    DN_MAX_INDBLKSHIFT, DMU_OT_NONE, 0, tx);
549 
550 	/*
551 	 * We don't want to have to increase the meta-dnode's nlevels
552 	 * later, because then we could do it in quescing context while
553 	 * we are also accessing it in open context.
554 	 *
555 	 * This precaution is not necessary for the MOS (ds == NULL),
556 	 * because the MOS is only updated in syncing context.
557 	 * This is most fortunate: the MOS is the only objset that
558 	 * needs to be synced multiple times as spa_sync() iterates
559 	 * to convergence, so minimizing its dn_nlevels matters.
560 	 */
561 	if (ds != NULL) {
562 		int levels = 1;
563 
564 		/*
565 		 * Determine the number of levels necessary for the meta-dnode
566 		 * to contain DN_MAX_OBJECT dnodes.
567 		 */
568 		while ((uint64_t)mdn->dn_nblkptr << (mdn->dn_datablkshift +
569 		    (levels - 1) * (mdn->dn_indblkshift - SPA_BLKPTRSHIFT)) <
570 		    DN_MAX_OBJECT * sizeof (dnode_phys_t))
571 			levels++;
572 
573 		mdn->dn_next_nlevels[tx->tx_txg & TXG_MASK] =
574 		    mdn->dn_nlevels = levels;
575 	}
576 
577 	ASSERT(type != DMU_OST_NONE);
578 	ASSERT(type != DMU_OST_ANY);
579 	ASSERT(type < DMU_OST_NUMTYPES);
580 	os->os_phys->os_type = type;
581 	if (dmu_objset_userused_enabled(os)) {
582 		os->os_phys->os_flags |= OBJSET_FLAG_USERACCOUNTING_COMPLETE;
583 		os->os_flags = os->os_phys->os_flags;
584 	}
585 
586 	dsl_dataset_dirty(ds, tx);
587 
588 	return (os);
589 }
590 
591 struct oscarg {
592 	void (*userfunc)(objset_t *os, void *arg, cred_t *cr, dmu_tx_t *tx);
593 	void *userarg;
594 	dsl_dataset_t *clone_origin;
595 	const char *lastname;
596 	dmu_objset_type_t type;
597 	uint64_t flags;
598 };
599 
600 /*ARGSUSED*/
601 static int
602 dmu_objset_create_check(void *arg1, void *arg2, dmu_tx_t *tx)
603 {
604 	dsl_dir_t *dd = arg1;
605 	struct oscarg *oa = arg2;
606 	objset_t *mos = dd->dd_pool->dp_meta_objset;
607 	int err;
608 	uint64_t ddobj;
609 
610 	err = zap_lookup(mos, dd->dd_phys->dd_child_dir_zapobj,
611 	    oa->lastname, sizeof (uint64_t), 1, &ddobj);
612 	if (err != ENOENT)
613 		return (err ? err : EEXIST);
614 
615 	if (oa->clone_origin != NULL) {
616 		/* You can't clone across pools. */
617 		if (oa->clone_origin->ds_dir->dd_pool != dd->dd_pool)
618 			return (EXDEV);
619 
620 		/* You can only clone snapshots, not the head datasets. */
621 		if (!dsl_dataset_is_snapshot(oa->clone_origin))
622 			return (EINVAL);
623 	}
624 
625 	return (0);
626 }
627 
628 static void
629 dmu_objset_create_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
630 {
631 	dsl_dir_t *dd = arg1;
632 	struct oscarg *oa = arg2;
633 	uint64_t dsobj;
634 
635 	ASSERT(dmu_tx_is_syncing(tx));
636 
637 	dsobj = dsl_dataset_create_sync(dd, oa->lastname,
638 	    oa->clone_origin, oa->flags, cr, tx);
639 
640 	if (oa->clone_origin == NULL) {
641 		dsl_dataset_t *ds;
642 		blkptr_t *bp;
643 		objset_t *os;
644 
645 		VERIFY(0 == dsl_dataset_hold_obj(dd->dd_pool, dsobj,
646 		    FTAG, &ds));
647 		bp = dsl_dataset_get_blkptr(ds);
648 		ASSERT(BP_IS_HOLE(bp));
649 
650 		os = dmu_objset_create_impl(dsl_dataset_get_spa(ds),
651 		    ds, bp, oa->type, tx);
652 
653 		if (oa->userfunc)
654 			oa->userfunc(os, oa->userarg, cr, tx);
655 		dsl_dataset_rele(ds, FTAG);
656 	}
657 
658 	spa_history_internal_log(LOG_DS_CREATE, dd->dd_pool->dp_spa,
659 	    tx, cr, "dataset = %llu", dsobj);
660 }
661 
662 int
663 dmu_objset_create(const char *name, dmu_objset_type_t type, uint64_t flags,
664     void (*func)(objset_t *os, void *arg, cred_t *cr, dmu_tx_t *tx), void *arg)
665 {
666 	dsl_dir_t *pdd;
667 	const char *tail;
668 	int err = 0;
669 	struct oscarg oa = { 0 };
670 
671 	ASSERT(strchr(name, '@') == NULL);
672 	err = dsl_dir_open(name, FTAG, &pdd, &tail);
673 	if (err)
674 		return (err);
675 	if (tail == NULL) {
676 		dsl_dir_close(pdd, FTAG);
677 		return (EEXIST);
678 	}
679 
680 	oa.userfunc = func;
681 	oa.userarg = arg;
682 	oa.lastname = tail;
683 	oa.type = type;
684 	oa.flags = flags;
685 
686 	err = dsl_sync_task_do(pdd->dd_pool, dmu_objset_create_check,
687 	    dmu_objset_create_sync, pdd, &oa, 5);
688 	dsl_dir_close(pdd, FTAG);
689 	return (err);
690 }
691 
692 int
693 dmu_objset_clone(const char *name, dsl_dataset_t *clone_origin, uint64_t flags)
694 {
695 	dsl_dir_t *pdd;
696 	const char *tail;
697 	int err = 0;
698 	struct oscarg oa = { 0 };
699 
700 	ASSERT(strchr(name, '@') == NULL);
701 	err = dsl_dir_open(name, FTAG, &pdd, &tail);
702 	if (err)
703 		return (err);
704 	if (tail == NULL) {
705 		dsl_dir_close(pdd, FTAG);
706 		return (EEXIST);
707 	}
708 
709 	oa.lastname = tail;
710 	oa.clone_origin = clone_origin;
711 	oa.flags = flags;
712 
713 	err = dsl_sync_task_do(pdd->dd_pool, dmu_objset_create_check,
714 	    dmu_objset_create_sync, pdd, &oa, 5);
715 	dsl_dir_close(pdd, FTAG);
716 	return (err);
717 }
718 
719 int
720 dmu_objset_destroy(const char *name, boolean_t defer)
721 {
722 	dsl_dataset_t *ds;
723 	int error;
724 
725 	/*
726 	 * dsl_dataset_destroy() can free any claimed-but-unplayed
727 	 * intent log, but if there is an active log, it has blocks that
728 	 * are allocated, but may not yet be reflected in the on-disk
729 	 * structure.  Only the ZIL knows how to free them, so we have
730 	 * to call into it here.
731 	 */
732 	error = dsl_dataset_own(name, B_TRUE, FTAG, &ds);
733 	if (error == 0) {
734 		objset_t *os;
735 		if (dmu_objset_from_ds(ds, &os) == 0)
736 			zil_destroy(dmu_objset_zil(os), B_FALSE);
737 		error = dsl_dataset_destroy(ds, FTAG, defer);
738 		/* dsl_dataset_destroy() closes the ds. */
739 	}
740 
741 	return (error);
742 }
743 
744 struct snaparg {
745 	dsl_sync_task_group_t *dstg;
746 	char *snapname;
747 	char failed[MAXPATHLEN];
748 	boolean_t recursive;
749 	nvlist_t *props;
750 };
751 
752 static int
753 snapshot_check(void *arg1, void *arg2, dmu_tx_t *tx)
754 {
755 	objset_t *os = arg1;
756 	struct snaparg *sn = arg2;
757 
758 	/* The props have already been checked by zfs_check_userprops(). */
759 
760 	return (dsl_dataset_snapshot_check(os->os_dsl_dataset,
761 	    sn->snapname, tx));
762 }
763 
764 static void
765 snapshot_sync(void *arg1, void *arg2, cred_t *cr, dmu_tx_t *tx)
766 {
767 	objset_t *os = arg1;
768 	dsl_dataset_t *ds = os->os_dsl_dataset;
769 	struct snaparg *sn = arg2;
770 
771 	dsl_dataset_snapshot_sync(ds, sn->snapname, cr, tx);
772 
773 	if (sn->props) {
774 		dsl_props_arg_t pa;
775 		pa.pa_props = sn->props;
776 		pa.pa_source = ZPROP_SRC_LOCAL;
777 		dsl_props_set_sync(ds->ds_prev, &pa, cr, tx);
778 	}
779 }
780 
781 static int
782 dmu_objset_snapshot_one(const char *name, void *arg)
783 {
784 	struct snaparg *sn = arg;
785 	objset_t *os;
786 	int err;
787 	char *cp;
788 
789 	/*
790 	 * If the objset starts with a '%', then ignore it unless it was
791 	 * explicitly named (ie, not recursive).  These hidden datasets
792 	 * are always inconsistent, and by not opening them here, we can
793 	 * avoid a race with dsl_dir_destroy_check().
794 	 */
795 	cp = strrchr(name, '/');
796 	if (cp && cp[1] == '%' && sn->recursive)
797 		return (0);
798 
799 	(void) strcpy(sn->failed, name);
800 
801 	/*
802 	 * Check permissions if we are doing a recursive snapshot.  The
803 	 * permission checks for the starting dataset have already been
804 	 * performed in zfs_secpolicy_snapshot()
805 	 */
806 	if (sn->recursive && (err = zfs_secpolicy_snapshot_perms(name, CRED())))
807 		return (err);
808 
809 	err = dmu_objset_hold(name, sn, &os);
810 	if (err != 0)
811 		return (err);
812 
813 	/*
814 	 * If the objset is in an inconsistent state (eg, in the process
815 	 * of being destroyed), don't snapshot it.  As with %hidden
816 	 * datasets, we return EBUSY if this name was explicitly
817 	 * requested (ie, not recursive), and otherwise ignore it.
818 	 */
819 	if (os->os_dsl_dataset->ds_phys->ds_flags & DS_FLAG_INCONSISTENT) {
820 		dmu_objset_rele(os, sn);
821 		return (sn->recursive ? 0 : EBUSY);
822 	}
823 
824 	/*
825 	 * NB: we need to wait for all in-flight changes to get to disk,
826 	 * so that we snapshot those changes.  zil_suspend does this as
827 	 * a side effect.
828 	 */
829 	err = zil_suspend(dmu_objset_zil(os));
830 	if (err == 0) {
831 		dsl_sync_task_create(sn->dstg, snapshot_check,
832 		    snapshot_sync, os, sn, 3);
833 	} else {
834 		dmu_objset_rele(os, sn);
835 	}
836 
837 	return (err);
838 }
839 
840 int
841 dmu_objset_snapshot(char *fsname, char *snapname,
842     nvlist_t *props, boolean_t recursive)
843 {
844 	dsl_sync_task_t *dst;
845 	struct snaparg sn;
846 	spa_t *spa;
847 	int err;
848 
849 	(void) strcpy(sn.failed, fsname);
850 
851 	err = spa_open(fsname, &spa, FTAG);
852 	if (err)
853 		return (err);
854 
855 	sn.dstg = dsl_sync_task_group_create(spa_get_dsl(spa));
856 	sn.snapname = snapname;
857 	sn.props = props;
858 	sn.recursive = recursive;
859 
860 	if (recursive) {
861 		err = dmu_objset_find(fsname,
862 		    dmu_objset_snapshot_one, &sn, DS_FIND_CHILDREN);
863 	} else {
864 		err = dmu_objset_snapshot_one(fsname, &sn);
865 	}
866 
867 	if (err == 0)
868 		err = dsl_sync_task_group_wait(sn.dstg);
869 
870 	for (dst = list_head(&sn.dstg->dstg_tasks); dst;
871 	    dst = list_next(&sn.dstg->dstg_tasks, dst)) {
872 		objset_t *os = dst->dst_arg1;
873 		dsl_dataset_t *ds = os->os_dsl_dataset;
874 		if (dst->dst_err)
875 			dsl_dataset_name(ds, sn.failed);
876 		zil_resume(dmu_objset_zil(os));
877 		dmu_objset_rele(os, &sn);
878 	}
879 
880 	if (err)
881 		(void) strcpy(fsname, sn.failed);
882 	dsl_sync_task_group_destroy(sn.dstg);
883 	spa_close(spa, FTAG);
884 	return (err);
885 }
886 
887 static void
888 dmu_objset_sync_dnodes(list_t *list, list_t *newlist, dmu_tx_t *tx)
889 {
890 	dnode_t *dn;
891 
892 	while (dn = list_head(list)) {
893 		ASSERT(dn->dn_object != DMU_META_DNODE_OBJECT);
894 		ASSERT(dn->dn_dbuf->db_data_pending);
895 		/*
896 		 * Initialize dn_zio outside dnode_sync() because the
897 		 * meta-dnode needs to set it ouside dnode_sync().
898 		 */
899 		dn->dn_zio = dn->dn_dbuf->db_data_pending->dr_zio;
900 		ASSERT(dn->dn_zio);
901 
902 		ASSERT3U(dn->dn_nlevels, <=, DN_MAX_LEVELS);
903 		list_remove(list, dn);
904 
905 		if (newlist) {
906 			(void) dnode_add_ref(dn, newlist);
907 			list_insert_tail(newlist, dn);
908 		}
909 
910 		dnode_sync(dn, tx);
911 	}
912 }
913 
914 /* ARGSUSED */
915 static void
916 dmu_objset_write_ready(zio_t *zio, arc_buf_t *abuf, void *arg)
917 {
918 	blkptr_t *bp = zio->io_bp;
919 	objset_t *os = arg;
920 	dnode_phys_t *dnp = &os->os_phys->os_meta_dnode;
921 
922 	ASSERT(bp == os->os_rootbp);
923 	ASSERT(BP_GET_TYPE(bp) == DMU_OT_OBJSET);
924 	ASSERT(BP_GET_LEVEL(bp) == 0);
925 
926 	/*
927 	 * Update rootbp fill count: it should be the number of objects
928 	 * allocated in the object set (not counting the "special"
929 	 * objects that are stored in the objset_phys_t -- the meta
930 	 * dnode and user/group accounting objects).
931 	 */
932 	bp->blk_fill = 0;
933 	for (int i = 0; i < dnp->dn_nblkptr; i++)
934 		bp->blk_fill += dnp->dn_blkptr[i].blk_fill;
935 }
936 
937 /* ARGSUSED */
938 static void
939 dmu_objset_write_done(zio_t *zio, arc_buf_t *abuf, void *arg)
940 {
941 	blkptr_t *bp = zio->io_bp;
942 	blkptr_t *bp_orig = &zio->io_bp_orig;
943 	objset_t *os = arg;
944 
945 	if (zio->io_flags & ZIO_FLAG_IO_REWRITE) {
946 		ASSERT(BP_EQUAL(bp, bp_orig));
947 	} else {
948 		dsl_dataset_t *ds = os->os_dsl_dataset;
949 		dmu_tx_t *tx = os->os_synctx;
950 
951 		(void) dsl_dataset_block_kill(ds, bp_orig, tx, B_TRUE);
952 		dsl_dataset_block_born(ds, bp, tx);
953 	}
954 }
955 
956 /* called from dsl */
957 void
958 dmu_objset_sync(objset_t *os, zio_t *pio, dmu_tx_t *tx)
959 {
960 	int txgoff;
961 	zbookmark_t zb;
962 	zio_prop_t zp;
963 	zio_t *zio;
964 	list_t *list;
965 	list_t *newlist = NULL;
966 	dbuf_dirty_record_t *dr;
967 
968 	dprintf_ds(os->os_dsl_dataset, "txg=%llu\n", tx->tx_txg);
969 
970 	ASSERT(dmu_tx_is_syncing(tx));
971 	/* XXX the write_done callback should really give us the tx... */
972 	os->os_synctx = tx;
973 
974 	if (os->os_dsl_dataset == NULL) {
975 		/*
976 		 * This is the MOS.  If we have upgraded,
977 		 * spa_max_replication() could change, so reset
978 		 * os_copies here.
979 		 */
980 		os->os_copies = spa_max_replication(os->os_spa);
981 	}
982 
983 	/*
984 	 * Create the root block IO
985 	 */
986 	arc_release(os->os_phys_buf, &os->os_phys_buf);
987 
988 	SET_BOOKMARK(&zb, os->os_dsl_dataset ?
989 	    os->os_dsl_dataset->ds_object : DMU_META_OBJSET,
990 	    ZB_ROOT_OBJECT, ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
991 
992 	dmu_write_policy(os, NULL, 0, 0, &zp);
993 
994 	zio = arc_write(pio, os->os_spa, tx->tx_txg,
995 	    os->os_rootbp, os->os_phys_buf, DMU_OS_IS_L2CACHEABLE(os), &zp,
996 	    dmu_objset_write_ready, dmu_objset_write_done, os,
997 	    ZIO_PRIORITY_ASYNC_WRITE, ZIO_FLAG_MUSTSUCCEED, &zb);
998 
999 	/*
1000 	 * Sync special dnodes - the parent IO for the sync is the root block
1001 	 */
1002 	os->os_meta_dnode->dn_zio = zio;
1003 	dnode_sync(os->os_meta_dnode, tx);
1004 
1005 	os->os_phys->os_flags = os->os_flags;
1006 
1007 	if (os->os_userused_dnode &&
1008 	    os->os_userused_dnode->dn_type != DMU_OT_NONE) {
1009 		os->os_userused_dnode->dn_zio = zio;
1010 		dnode_sync(os->os_userused_dnode, tx);
1011 		os->os_groupused_dnode->dn_zio = zio;
1012 		dnode_sync(os->os_groupused_dnode, tx);
1013 	}
1014 
1015 	txgoff = tx->tx_txg & TXG_MASK;
1016 
1017 	if (dmu_objset_userused_enabled(os)) {
1018 		newlist = &os->os_synced_dnodes;
1019 		/*
1020 		 * We must create the list here because it uses the
1021 		 * dn_dirty_link[] of this txg.
1022 		 */
1023 		list_create(newlist, sizeof (dnode_t),
1024 		    offsetof(dnode_t, dn_dirty_link[txgoff]));
1025 	}
1026 
1027 	dmu_objset_sync_dnodes(&os->os_free_dnodes[txgoff], newlist, tx);
1028 	dmu_objset_sync_dnodes(&os->os_dirty_dnodes[txgoff], newlist, tx);
1029 
1030 	list = &os->os_meta_dnode->dn_dirty_records[txgoff];
1031 	while (dr = list_head(list)) {
1032 		ASSERT(dr->dr_dbuf->db_level == 0);
1033 		list_remove(list, dr);
1034 		if (dr->dr_zio)
1035 			zio_nowait(dr->dr_zio);
1036 	}
1037 	/*
1038 	 * Free intent log blocks up to this tx.
1039 	 */
1040 	zil_sync(os->os_zil, tx);
1041 	os->os_phys->os_zil_header = os->os_zil_header;
1042 	zio_nowait(zio);
1043 }
1044 
1045 boolean_t
1046 dmu_objset_is_dirty(objset_t *os, uint64_t txg)
1047 {
1048 	return (!list_is_empty(&os->os_dirty_dnodes[txg & TXG_MASK]) ||
1049 	    !list_is_empty(&os->os_free_dnodes[txg & TXG_MASK]));
1050 }
1051 
1052 static objset_used_cb_t *used_cbs[DMU_OST_NUMTYPES];
1053 
1054 void
1055 dmu_objset_register_type(dmu_objset_type_t ost, objset_used_cb_t *cb)
1056 {
1057 	used_cbs[ost] = cb;
1058 }
1059 
1060 boolean_t
1061 dmu_objset_userused_enabled(objset_t *os)
1062 {
1063 	return (spa_version(os->os_spa) >= SPA_VERSION_USERSPACE &&
1064 	    used_cbs[os->os_phys->os_type] &&
1065 	    os->os_userused_dnode);
1066 }
1067 
1068 static void
1069 do_userquota_callback(objset_t *os, dnode_phys_t *dnp,
1070     boolean_t subtract, dmu_tx_t *tx)
1071 {
1072 	static const char zerobuf[DN_MAX_BONUSLEN] = {0};
1073 	uint64_t user, group;
1074 
1075 	ASSERT(dnp->dn_type != 0 ||
1076 	    (bcmp(DN_BONUS(dnp), zerobuf, DN_MAX_BONUSLEN) == 0 &&
1077 	    DN_USED_BYTES(dnp) == 0));
1078 
1079 	if ((dnp->dn_flags & DNODE_FLAG_USERUSED_ACCOUNTED) &&
1080 	    0 == used_cbs[os->os_phys->os_type](dnp->dn_bonustype,
1081 	    DN_BONUS(dnp), &user, &group)) {
1082 		int64_t delta = DNODE_SIZE + DN_USED_BYTES(dnp);
1083 		if (subtract)
1084 			delta = -delta;
1085 		VERIFY3U(0, ==, zap_increment_int(os, DMU_USERUSED_OBJECT,
1086 		    user, delta, tx));
1087 		VERIFY3U(0, ==, zap_increment_int(os, DMU_GROUPUSED_OBJECT,
1088 		    group, delta, tx));
1089 	}
1090 }
1091 
1092 void
1093 dmu_objset_do_userquota_callbacks(objset_t *os, dmu_tx_t *tx)
1094 {
1095 	dnode_t *dn;
1096 	list_t *list = &os->os_synced_dnodes;
1097 
1098 	ASSERT(list_head(list) == NULL || dmu_objset_userused_enabled(os));
1099 
1100 	while (dn = list_head(list)) {
1101 		ASSERT(!DMU_OBJECT_IS_SPECIAL(dn->dn_object));
1102 		ASSERT(dn->dn_oldphys);
1103 		ASSERT(dn->dn_phys->dn_type == DMU_OT_NONE ||
1104 		    dn->dn_phys->dn_flags &
1105 		    DNODE_FLAG_USERUSED_ACCOUNTED);
1106 
1107 		/* Allocate the user/groupused objects if necessary. */
1108 		if (os->os_userused_dnode->dn_type == DMU_OT_NONE) {
1109 			VERIFY(0 == zap_create_claim(os,
1110 			    DMU_USERUSED_OBJECT,
1111 			    DMU_OT_USERGROUP_USED, DMU_OT_NONE, 0, tx));
1112 			VERIFY(0 == zap_create_claim(os,
1113 			    DMU_GROUPUSED_OBJECT,
1114 			    DMU_OT_USERGROUP_USED, DMU_OT_NONE, 0, tx));
1115 		}
1116 
1117 		/*
1118 		 * We intentionally modify the zap object even if the
1119 		 * net delta (due to phys-oldphys) is zero.  Otherwise
1120 		 * the block of the zap obj could be shared between
1121 		 * datasets but need to be different between them after
1122 		 * a bprewrite.
1123 		 */
1124 		do_userquota_callback(os, dn->dn_oldphys, B_TRUE, tx);
1125 		do_userquota_callback(os, dn->dn_phys, B_FALSE, tx);
1126 
1127 		/*
1128 		 * The mutex is needed here for interlock with dnode_allocate.
1129 		 */
1130 		mutex_enter(&dn->dn_mtx);
1131 		zio_buf_free(dn->dn_oldphys, sizeof (dnode_phys_t));
1132 		dn->dn_oldphys = NULL;
1133 		mutex_exit(&dn->dn_mtx);
1134 
1135 		list_remove(list, dn);
1136 		dnode_rele(dn, list);
1137 	}
1138 }
1139 
1140 boolean_t
1141 dmu_objset_userspace_present(objset_t *os)
1142 {
1143 	return (os->os_phys->os_flags &
1144 	    OBJSET_FLAG_USERACCOUNTING_COMPLETE);
1145 }
1146 
1147 int
1148 dmu_objset_userspace_upgrade(objset_t *os)
1149 {
1150 	uint64_t obj;
1151 	int err = 0;
1152 
1153 	if (dmu_objset_userspace_present(os))
1154 		return (0);
1155 	if (!dmu_objset_userused_enabled(os))
1156 		return (ENOTSUP);
1157 	if (dmu_objset_is_snapshot(os))
1158 		return (EINVAL);
1159 
1160 	/*
1161 	 * We simply need to mark every object dirty, so that it will be
1162 	 * synced out and now accounted.  If this is called
1163 	 * concurrently, or if we already did some work before crashing,
1164 	 * that's fine, since we track each object's accounted state
1165 	 * independently.
1166 	 */
1167 
1168 	for (obj = 0; err == 0; err = dmu_object_next(os, &obj, FALSE, 0)) {
1169 		dmu_tx_t *tx;
1170 		dmu_buf_t *db;
1171 		int objerr;
1172 
1173 		if (issig(JUSTLOOKING) && issig(FORREAL))
1174 			return (EINTR);
1175 
1176 		objerr = dmu_bonus_hold(os, obj, FTAG, &db);
1177 		if (objerr)
1178 			continue;
1179 		tx = dmu_tx_create(os);
1180 		dmu_tx_hold_bonus(tx, obj);
1181 		objerr = dmu_tx_assign(tx, TXG_WAIT);
1182 		if (objerr) {
1183 			dmu_tx_abort(tx);
1184 			continue;
1185 		}
1186 		dmu_buf_will_dirty(db, tx);
1187 		dmu_buf_rele(db, FTAG);
1188 		dmu_tx_commit(tx);
1189 	}
1190 
1191 	os->os_flags |= OBJSET_FLAG_USERACCOUNTING_COMPLETE;
1192 	txg_wait_synced(dmu_objset_pool(os), 0);
1193 	return (0);
1194 }
1195 
1196 void
1197 dmu_objset_space(objset_t *os, uint64_t *refdbytesp, uint64_t *availbytesp,
1198     uint64_t *usedobjsp, uint64_t *availobjsp)
1199 {
1200 	dsl_dataset_space(os->os_dsl_dataset, refdbytesp, availbytesp,
1201 	    usedobjsp, availobjsp);
1202 }
1203 
1204 uint64_t
1205 dmu_objset_fsid_guid(objset_t *os)
1206 {
1207 	return (dsl_dataset_fsid_guid(os->os_dsl_dataset));
1208 }
1209 
1210 void
1211 dmu_objset_fast_stat(objset_t *os, dmu_objset_stats_t *stat)
1212 {
1213 	stat->dds_type = os->os_phys->os_type;
1214 	if (os->os_dsl_dataset)
1215 		dsl_dataset_fast_stat(os->os_dsl_dataset, stat);
1216 }
1217 
1218 void
1219 dmu_objset_stats(objset_t *os, nvlist_t *nv)
1220 {
1221 	ASSERT(os->os_dsl_dataset ||
1222 	    os->os_phys->os_type == DMU_OST_META);
1223 
1224 	if (os->os_dsl_dataset != NULL)
1225 		dsl_dataset_stats(os->os_dsl_dataset, nv);
1226 
1227 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_TYPE,
1228 	    os->os_phys->os_type);
1229 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USERACCOUNTING,
1230 	    dmu_objset_userspace_present(os));
1231 }
1232 
1233 int
1234 dmu_objset_is_snapshot(objset_t *os)
1235 {
1236 	if (os->os_dsl_dataset != NULL)
1237 		return (dsl_dataset_is_snapshot(os->os_dsl_dataset));
1238 	else
1239 		return (B_FALSE);
1240 }
1241 
1242 int
1243 dmu_snapshot_realname(objset_t *os, char *name, char *real, int maxlen,
1244     boolean_t *conflict)
1245 {
1246 	dsl_dataset_t *ds = os->os_dsl_dataset;
1247 	uint64_t ignored;
1248 
1249 	if (ds->ds_phys->ds_snapnames_zapobj == 0)
1250 		return (ENOENT);
1251 
1252 	return (zap_lookup_norm(ds->ds_dir->dd_pool->dp_meta_objset,
1253 	    ds->ds_phys->ds_snapnames_zapobj, name, 8, 1, &ignored, MT_FIRST,
1254 	    real, maxlen, conflict));
1255 }
1256 
1257 int
1258 dmu_snapshot_list_next(objset_t *os, int namelen, char *name,
1259     uint64_t *idp, uint64_t *offp, boolean_t *case_conflict)
1260 {
1261 	dsl_dataset_t *ds = os->os_dsl_dataset;
1262 	zap_cursor_t cursor;
1263 	zap_attribute_t attr;
1264 
1265 	if (ds->ds_phys->ds_snapnames_zapobj == 0)
1266 		return (ENOENT);
1267 
1268 	zap_cursor_init_serialized(&cursor,
1269 	    ds->ds_dir->dd_pool->dp_meta_objset,
1270 	    ds->ds_phys->ds_snapnames_zapobj, *offp);
1271 
1272 	if (zap_cursor_retrieve(&cursor, &attr) != 0) {
1273 		zap_cursor_fini(&cursor);
1274 		return (ENOENT);
1275 	}
1276 
1277 	if (strlen(attr.za_name) + 1 > namelen) {
1278 		zap_cursor_fini(&cursor);
1279 		return (ENAMETOOLONG);
1280 	}
1281 
1282 	(void) strcpy(name, attr.za_name);
1283 	if (idp)
1284 		*idp = attr.za_first_integer;
1285 	if (case_conflict)
1286 		*case_conflict = attr.za_normalization_conflict;
1287 	zap_cursor_advance(&cursor);
1288 	*offp = zap_cursor_serialize(&cursor);
1289 	zap_cursor_fini(&cursor);
1290 
1291 	return (0);
1292 }
1293 
1294 int
1295 dmu_dir_list_next(objset_t *os, int namelen, char *name,
1296     uint64_t *idp, uint64_t *offp)
1297 {
1298 	dsl_dir_t *dd = os->os_dsl_dataset->ds_dir;
1299 	zap_cursor_t cursor;
1300 	zap_attribute_t attr;
1301 
1302 	/* there is no next dir on a snapshot! */
1303 	if (os->os_dsl_dataset->ds_object !=
1304 	    dd->dd_phys->dd_head_dataset_obj)
1305 		return (ENOENT);
1306 
1307 	zap_cursor_init_serialized(&cursor,
1308 	    dd->dd_pool->dp_meta_objset,
1309 	    dd->dd_phys->dd_child_dir_zapobj, *offp);
1310 
1311 	if (zap_cursor_retrieve(&cursor, &attr) != 0) {
1312 		zap_cursor_fini(&cursor);
1313 		return (ENOENT);
1314 	}
1315 
1316 	if (strlen(attr.za_name) + 1 > namelen) {
1317 		zap_cursor_fini(&cursor);
1318 		return (ENAMETOOLONG);
1319 	}
1320 
1321 	(void) strcpy(name, attr.za_name);
1322 	if (idp)
1323 		*idp = attr.za_first_integer;
1324 	zap_cursor_advance(&cursor);
1325 	*offp = zap_cursor_serialize(&cursor);
1326 	zap_cursor_fini(&cursor);
1327 
1328 	return (0);
1329 }
1330 
1331 struct findarg {
1332 	int (*func)(const char *, void *);
1333 	void *arg;
1334 };
1335 
1336 /* ARGSUSED */
1337 static int
1338 findfunc(spa_t *spa, uint64_t dsobj, const char *dsname, void *arg)
1339 {
1340 	struct findarg *fa = arg;
1341 	return (fa->func(dsname, fa->arg));
1342 }
1343 
1344 /*
1345  * Find all objsets under name, and for each, call 'func(child_name, arg)'.
1346  * Perhaps change all callers to use dmu_objset_find_spa()?
1347  */
1348 int
1349 dmu_objset_find(char *name, int func(const char *, void *), void *arg,
1350     int flags)
1351 {
1352 	struct findarg fa;
1353 	fa.func = func;
1354 	fa.arg = arg;
1355 	return (dmu_objset_find_spa(NULL, name, findfunc, &fa, flags));
1356 }
1357 
1358 /*
1359  * Find all objsets under name, call func on each
1360  */
1361 int
1362 dmu_objset_find_spa(spa_t *spa, const char *name,
1363     int func(spa_t *, uint64_t, const char *, void *), void *arg, int flags)
1364 {
1365 	dsl_dir_t *dd;
1366 	dsl_pool_t *dp;
1367 	dsl_dataset_t *ds;
1368 	zap_cursor_t zc;
1369 	zap_attribute_t *attr;
1370 	char *child;
1371 	uint64_t thisobj;
1372 	int err;
1373 
1374 	if (name == NULL)
1375 		name = spa_name(spa);
1376 	err = dsl_dir_open_spa(spa, name, FTAG, &dd, NULL);
1377 	if (err)
1378 		return (err);
1379 
1380 	/* Don't visit hidden ($MOS & $ORIGIN) objsets. */
1381 	if (dd->dd_myname[0] == '$') {
1382 		dsl_dir_close(dd, FTAG);
1383 		return (0);
1384 	}
1385 
1386 	thisobj = dd->dd_phys->dd_head_dataset_obj;
1387 	attr = kmem_alloc(sizeof (zap_attribute_t), KM_SLEEP);
1388 	dp = dd->dd_pool;
1389 
1390 	/*
1391 	 * Iterate over all children.
1392 	 */
1393 	if (flags & DS_FIND_CHILDREN) {
1394 		for (zap_cursor_init(&zc, dp->dp_meta_objset,
1395 		    dd->dd_phys->dd_child_dir_zapobj);
1396 		    zap_cursor_retrieve(&zc, attr) == 0;
1397 		    (void) zap_cursor_advance(&zc)) {
1398 			ASSERT(attr->za_integer_length == sizeof (uint64_t));
1399 			ASSERT(attr->za_num_integers == 1);
1400 
1401 			child = kmem_asprintf("%s/%s", name, attr->za_name);
1402 			err = dmu_objset_find_spa(spa, child, func, arg, flags);
1403 			strfree(child);
1404 			if (err)
1405 				break;
1406 		}
1407 		zap_cursor_fini(&zc);
1408 
1409 		if (err) {
1410 			dsl_dir_close(dd, FTAG);
1411 			kmem_free(attr, sizeof (zap_attribute_t));
1412 			return (err);
1413 		}
1414 	}
1415 
1416 	/*
1417 	 * Iterate over all snapshots.
1418 	 */
1419 	if (flags & DS_FIND_SNAPSHOTS) {
1420 		if (!dsl_pool_sync_context(dp))
1421 			rw_enter(&dp->dp_config_rwlock, RW_READER);
1422 		err = dsl_dataset_hold_obj(dp, thisobj, FTAG, &ds);
1423 		if (!dsl_pool_sync_context(dp))
1424 			rw_exit(&dp->dp_config_rwlock);
1425 
1426 		if (err == 0) {
1427 			uint64_t snapobj = ds->ds_phys->ds_snapnames_zapobj;
1428 			dsl_dataset_rele(ds, FTAG);
1429 
1430 			for (zap_cursor_init(&zc, dp->dp_meta_objset, snapobj);
1431 			    zap_cursor_retrieve(&zc, attr) == 0;
1432 			    (void) zap_cursor_advance(&zc)) {
1433 				ASSERT(attr->za_integer_length ==
1434 				    sizeof (uint64_t));
1435 				ASSERT(attr->za_num_integers == 1);
1436 
1437 				child = kmem_asprintf("%s@%s",
1438 				    name, attr->za_name);
1439 				err = func(spa, attr->za_first_integer,
1440 				    child, arg);
1441 				strfree(child);
1442 				if (err)
1443 					break;
1444 			}
1445 			zap_cursor_fini(&zc);
1446 		}
1447 	}
1448 
1449 	dsl_dir_close(dd, FTAG);
1450 	kmem_free(attr, sizeof (zap_attribute_t));
1451 
1452 	if (err)
1453 		return (err);
1454 
1455 	/*
1456 	 * Apply to self if appropriate.
1457 	 */
1458 	err = func(spa, thisobj, name, arg);
1459 	return (err);
1460 }
1461 
1462 /* ARGSUSED */
1463 int
1464 dmu_objset_prefetch(const char *name, void *arg)
1465 {
1466 	dsl_dataset_t *ds;
1467 
1468 	if (dsl_dataset_hold(name, FTAG, &ds))
1469 		return (0);
1470 
1471 	if (!BP_IS_HOLE(&ds->ds_phys->ds_bp)) {
1472 		mutex_enter(&ds->ds_opening_lock);
1473 		if (ds->ds_objset == NULL) {
1474 			uint32_t aflags = ARC_NOWAIT | ARC_PREFETCH;
1475 			zbookmark_t zb;
1476 
1477 			SET_BOOKMARK(&zb, ds->ds_object, ZB_ROOT_OBJECT,
1478 			    ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
1479 
1480 			(void) arc_read_nolock(NULL, dsl_dataset_get_spa(ds),
1481 			    &ds->ds_phys->ds_bp, NULL, NULL,
1482 			    ZIO_PRIORITY_ASYNC_READ,
1483 			    ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE,
1484 			    &aflags, &zb);
1485 		}
1486 		mutex_exit(&ds->ds_opening_lock);
1487 	}
1488 
1489 	dsl_dataset_rele(ds, FTAG);
1490 	return (0);
1491 }
1492 
1493 void
1494 dmu_objset_set_user(objset_t *os, void *user_ptr)
1495 {
1496 	ASSERT(MUTEX_HELD(&os->os_user_ptr_lock));
1497 	os->os_user_ptr = user_ptr;
1498 }
1499 
1500 void *
1501 dmu_objset_get_user(objset_t *os)
1502 {
1503 	ASSERT(MUTEX_HELD(&os->os_user_ptr_lock));
1504 	return (os->os_user_ptr);
1505 }
1506