xref: /freebsd/sys/contrib/openzfs/module/zfs/dmu_objset.c (revision a0b956f5ac5e0941f9e74e24c1c53e05ad061a38)
1 /*
2  * CDDL HEADER START
3  *
4  * The contents of this file are subject to the terms of the
5  * Common Development and Distribution License (the "License").
6  * You may not use this file except in compliance with the License.
7  *
8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9  * or http://www.opensolaris.org/os/licensing.
10  * See the License for the specific language governing permissions
11  * and limitations under the License.
12  *
13  * When distributing Covered Code, include this CDDL HEADER in each
14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15  * If applicable, add the following below this CDDL HEADER, with the
16  * fields enclosed by brackets "[]" replaced with your own identifying
17  * information: Portions Copyright [yyyy] [name of copyright owner]
18  *
19  * CDDL HEADER END
20  */
21 
22 /*
23  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Copyright (c) 2012, 2020 by Delphix. All rights reserved.
25  * Copyright (c) 2013 by Saso Kiselkov. All rights reserved.
26  * Copyright (c) 2013, Joyent, Inc. All rights reserved.
27  * Copyright (c) 2014 Spectra Logic Corporation, All rights reserved.
28  * Copyright (c) 2015, STRATO AG, Inc. All rights reserved.
29  * Copyright (c) 2016 Actifio, Inc. All rights reserved.
30  * Copyright 2017 Nexenta Systems, Inc.
31  * Copyright (c) 2017 Open-E, Inc. All Rights Reserved.
32  * Copyright (c) 2018, loli10K <ezomori.nozomu@gmail.com>. All rights reserved.
33  * Copyright (c) 2019, Klara Inc.
34  * Copyright (c) 2019, Allan Jude
35  */
36 
37 /* Portions Copyright 2010 Robert Milkowski */
38 
39 #include <sys/cred.h>
40 #include <sys/zfs_context.h>
41 #include <sys/dmu_objset.h>
42 #include <sys/dsl_dir.h>
43 #include <sys/dsl_dataset.h>
44 #include <sys/dsl_prop.h>
45 #include <sys/dsl_pool.h>
46 #include <sys/dsl_synctask.h>
47 #include <sys/dsl_deleg.h>
48 #include <sys/dnode.h>
49 #include <sys/dbuf.h>
50 #include <sys/zvol.h>
51 #include <sys/dmu_tx.h>
52 #include <sys/zap.h>
53 #include <sys/zil.h>
54 #include <sys/dmu_impl.h>
55 #include <sys/zfs_ioctl.h>
56 #include <sys/sa.h>
57 #include <sys/zfs_onexit.h>
58 #include <sys/dsl_destroy.h>
59 #include <sys/vdev.h>
60 #include <sys/zfeature.h>
61 #include <sys/policy.h>
62 #include <sys/spa_impl.h>
63 #include <sys/dmu_recv.h>
64 #include <sys/zfs_project.h>
65 #include "zfs_namecheck.h"
66 #include <sys/vdev_impl.h>
67 #include <sys/arc.h>
68 
69 /*
70  * Needed to close a window in dnode_move() that allows the objset to be freed
71  * before it can be safely accessed.
72  */
73 krwlock_t os_lock;
74 
75 /*
76  * Tunable to overwrite the maximum number of threads for the parallelization
77  * of dmu_objset_find_dp, needed to speed up the import of pools with many
78  * datasets.
79  * Default is 4 times the number of leaf vdevs.
80  */
81 static const int dmu_find_threads = 0;
82 
83 /*
84  * Backfill lower metadnode objects after this many have been freed.
85  * Backfilling negatively impacts object creation rates, so only do it
86  * if there are enough holes to fill.
87  */
88 static const int dmu_rescan_dnode_threshold = 1 << DN_MAX_INDBLKSHIFT;
89 
90 static const char *upgrade_tag = "upgrade_tag";
91 
92 static void dmu_objset_find_dp_cb(void *arg);
93 
94 static void dmu_objset_upgrade(objset_t *os, dmu_objset_upgrade_cb_t cb);
95 static void dmu_objset_upgrade_stop(objset_t *os);
96 
97 void
98 dmu_objset_init(void)
99 {
100 	rw_init(&os_lock, NULL, RW_DEFAULT, NULL);
101 }
102 
103 void
104 dmu_objset_fini(void)
105 {
106 	rw_destroy(&os_lock);
107 }
108 
109 spa_t *
110 dmu_objset_spa(objset_t *os)
111 {
112 	return (os->os_spa);
113 }
114 
115 zilog_t *
116 dmu_objset_zil(objset_t *os)
117 {
118 	return (os->os_zil);
119 }
120 
121 dsl_pool_t *
122 dmu_objset_pool(objset_t *os)
123 {
124 	dsl_dataset_t *ds;
125 
126 	if ((ds = os->os_dsl_dataset) != NULL && ds->ds_dir)
127 		return (ds->ds_dir->dd_pool);
128 	else
129 		return (spa_get_dsl(os->os_spa));
130 }
131 
132 dsl_dataset_t *
133 dmu_objset_ds(objset_t *os)
134 {
135 	return (os->os_dsl_dataset);
136 }
137 
138 dmu_objset_type_t
139 dmu_objset_type(objset_t *os)
140 {
141 	return (os->os_phys->os_type);
142 }
143 
144 void
145 dmu_objset_name(objset_t *os, char *buf)
146 {
147 	dsl_dataset_name(os->os_dsl_dataset, buf);
148 }
149 
150 uint64_t
151 dmu_objset_id(objset_t *os)
152 {
153 	dsl_dataset_t *ds = os->os_dsl_dataset;
154 
155 	return (ds ? ds->ds_object : 0);
156 }
157 
158 uint64_t
159 dmu_objset_dnodesize(objset_t *os)
160 {
161 	return (os->os_dnodesize);
162 }
163 
164 zfs_sync_type_t
165 dmu_objset_syncprop(objset_t *os)
166 {
167 	return (os->os_sync);
168 }
169 
170 zfs_logbias_op_t
171 dmu_objset_logbias(objset_t *os)
172 {
173 	return (os->os_logbias);
174 }
175 
176 static void
177 checksum_changed_cb(void *arg, uint64_t newval)
178 {
179 	objset_t *os = arg;
180 
181 	/*
182 	 * Inheritance should have been done by now.
183 	 */
184 	ASSERT(newval != ZIO_CHECKSUM_INHERIT);
185 
186 	os->os_checksum = zio_checksum_select(newval, ZIO_CHECKSUM_ON_VALUE);
187 }
188 
189 static void
190 compression_changed_cb(void *arg, uint64_t newval)
191 {
192 	objset_t *os = arg;
193 
194 	/*
195 	 * Inheritance and range checking should have been done by now.
196 	 */
197 	ASSERT(newval != ZIO_COMPRESS_INHERIT);
198 
199 	os->os_compress = zio_compress_select(os->os_spa,
200 	    ZIO_COMPRESS_ALGO(newval), ZIO_COMPRESS_ON);
201 	os->os_complevel = zio_complevel_select(os->os_spa, os->os_compress,
202 	    ZIO_COMPRESS_LEVEL(newval), ZIO_COMPLEVEL_DEFAULT);
203 }
204 
205 static void
206 copies_changed_cb(void *arg, uint64_t newval)
207 {
208 	objset_t *os = arg;
209 
210 	/*
211 	 * Inheritance and range checking should have been done by now.
212 	 */
213 	ASSERT(newval > 0);
214 	ASSERT(newval <= spa_max_replication(os->os_spa));
215 
216 	os->os_copies = newval;
217 }
218 
219 static void
220 dedup_changed_cb(void *arg, uint64_t newval)
221 {
222 	objset_t *os = arg;
223 	spa_t *spa = os->os_spa;
224 	enum zio_checksum checksum;
225 
226 	/*
227 	 * Inheritance should have been done by now.
228 	 */
229 	ASSERT(newval != ZIO_CHECKSUM_INHERIT);
230 
231 	checksum = zio_checksum_dedup_select(spa, newval, ZIO_CHECKSUM_OFF);
232 
233 	os->os_dedup_checksum = checksum & ZIO_CHECKSUM_MASK;
234 	os->os_dedup_verify = !!(checksum & ZIO_CHECKSUM_VERIFY);
235 }
236 
237 static void
238 primary_cache_changed_cb(void *arg, uint64_t newval)
239 {
240 	objset_t *os = arg;
241 
242 	/*
243 	 * Inheritance and range checking should have been done by now.
244 	 */
245 	ASSERT(newval == ZFS_CACHE_ALL || newval == ZFS_CACHE_NONE ||
246 	    newval == ZFS_CACHE_METADATA);
247 
248 	os->os_primary_cache = newval;
249 }
250 
251 static void
252 secondary_cache_changed_cb(void *arg, uint64_t newval)
253 {
254 	objset_t *os = arg;
255 
256 	/*
257 	 * Inheritance and range checking should have been done by now.
258 	 */
259 	ASSERT(newval == ZFS_CACHE_ALL || newval == ZFS_CACHE_NONE ||
260 	    newval == ZFS_CACHE_METADATA);
261 
262 	os->os_secondary_cache = newval;
263 }
264 
265 static void
266 sync_changed_cb(void *arg, uint64_t newval)
267 {
268 	objset_t *os = arg;
269 
270 	/*
271 	 * Inheritance and range checking should have been done by now.
272 	 */
273 	ASSERT(newval == ZFS_SYNC_STANDARD || newval == ZFS_SYNC_ALWAYS ||
274 	    newval == ZFS_SYNC_DISABLED);
275 
276 	os->os_sync = newval;
277 	if (os->os_zil)
278 		zil_set_sync(os->os_zil, newval);
279 }
280 
281 static void
282 redundant_metadata_changed_cb(void *arg, uint64_t newval)
283 {
284 	objset_t *os = arg;
285 
286 	/*
287 	 * Inheritance and range checking should have been done by now.
288 	 */
289 	ASSERT(newval == ZFS_REDUNDANT_METADATA_ALL ||
290 	    newval == ZFS_REDUNDANT_METADATA_MOST);
291 
292 	os->os_redundant_metadata = newval;
293 }
294 
295 static void
296 dnodesize_changed_cb(void *arg, uint64_t newval)
297 {
298 	objset_t *os = arg;
299 
300 	switch (newval) {
301 	case ZFS_DNSIZE_LEGACY:
302 		os->os_dnodesize = DNODE_MIN_SIZE;
303 		break;
304 	case ZFS_DNSIZE_AUTO:
305 		/*
306 		 * Choose a dnode size that will work well for most
307 		 * workloads if the user specified "auto". Future code
308 		 * improvements could dynamically select a dnode size
309 		 * based on observed workload patterns.
310 		 */
311 		os->os_dnodesize = DNODE_MIN_SIZE * 2;
312 		break;
313 	case ZFS_DNSIZE_1K:
314 	case ZFS_DNSIZE_2K:
315 	case ZFS_DNSIZE_4K:
316 	case ZFS_DNSIZE_8K:
317 	case ZFS_DNSIZE_16K:
318 		os->os_dnodesize = newval;
319 		break;
320 	}
321 }
322 
323 static void
324 smallblk_changed_cb(void *arg, uint64_t newval)
325 {
326 	objset_t *os = arg;
327 
328 	/*
329 	 * Inheritance and range checking should have been done by now.
330 	 */
331 	ASSERT(newval <= SPA_MAXBLOCKSIZE);
332 	ASSERT(ISP2(newval));
333 
334 	os->os_zpl_special_smallblock = newval;
335 }
336 
337 static void
338 logbias_changed_cb(void *arg, uint64_t newval)
339 {
340 	objset_t *os = arg;
341 
342 	ASSERT(newval == ZFS_LOGBIAS_LATENCY ||
343 	    newval == ZFS_LOGBIAS_THROUGHPUT);
344 	os->os_logbias = newval;
345 	if (os->os_zil)
346 		zil_set_logbias(os->os_zil, newval);
347 }
348 
349 static void
350 recordsize_changed_cb(void *arg, uint64_t newval)
351 {
352 	objset_t *os = arg;
353 
354 	os->os_recordsize = newval;
355 }
356 
357 void
358 dmu_objset_byteswap(void *buf, size_t size)
359 {
360 	objset_phys_t *osp = buf;
361 
362 	ASSERT(size == OBJSET_PHYS_SIZE_V1 || size == OBJSET_PHYS_SIZE_V2 ||
363 	    size == sizeof (objset_phys_t));
364 	dnode_byteswap(&osp->os_meta_dnode);
365 	byteswap_uint64_array(&osp->os_zil_header, sizeof (zil_header_t));
366 	osp->os_type = BSWAP_64(osp->os_type);
367 	osp->os_flags = BSWAP_64(osp->os_flags);
368 	if (size >= OBJSET_PHYS_SIZE_V2) {
369 		dnode_byteswap(&osp->os_userused_dnode);
370 		dnode_byteswap(&osp->os_groupused_dnode);
371 		if (size >= sizeof (objset_phys_t))
372 			dnode_byteswap(&osp->os_projectused_dnode);
373 	}
374 }
375 
376 /*
377  * The hash is a CRC-based hash of the objset_t pointer and the object number.
378  */
379 static uint64_t
380 dnode_hash(const objset_t *os, uint64_t obj)
381 {
382 	uintptr_t osv = (uintptr_t)os;
383 	uint64_t crc = -1ULL;
384 
385 	ASSERT(zfs_crc64_table[128] == ZFS_CRC64_POLY);
386 	/*
387 	 * The low 6 bits of the pointer don't have much entropy, because
388 	 * the objset_t is larger than 2^6 bytes long.
389 	 */
390 	crc = (crc >> 8) ^ zfs_crc64_table[(crc ^ (osv >> 6)) & 0xFF];
391 	crc = (crc >> 8) ^ zfs_crc64_table[(crc ^ (obj >> 0)) & 0xFF];
392 	crc = (crc >> 8) ^ zfs_crc64_table[(crc ^ (obj >> 8)) & 0xFF];
393 	crc = (crc >> 8) ^ zfs_crc64_table[(crc ^ (obj >> 16)) & 0xFF];
394 
395 	crc ^= (osv>>14) ^ (obj>>24);
396 
397 	return (crc);
398 }
399 
400 static unsigned int
401 dnode_multilist_index_func(multilist_t *ml, void *obj)
402 {
403 	dnode_t *dn = obj;
404 
405 	/*
406 	 * The low order bits of the hash value are thought to be
407 	 * distributed evenly. Otherwise, in the case that the multilist
408 	 * has a power of two number of sublists, each sublists' usage
409 	 * would not be evenly distributed. In this context full 64bit
410 	 * division would be a waste of time, so limit it to 32 bits.
411 	 */
412 	return ((unsigned int)dnode_hash(dn->dn_objset, dn->dn_object) %
413 	    multilist_get_num_sublists(ml));
414 }
415 
416 static inline boolean_t
417 dmu_os_is_l2cacheable(objset_t *os)
418 {
419 	vdev_t *vd = NULL;
420 	zfs_cache_type_t cache = os->os_secondary_cache;
421 	blkptr_t *bp = os->os_rootbp;
422 
423 	if (bp != NULL && !BP_IS_HOLE(bp)) {
424 		uint64_t vdev = DVA_GET_VDEV(bp->blk_dva);
425 		vdev_t *rvd = os->os_spa->spa_root_vdev;
426 
427 		if (vdev < rvd->vdev_children)
428 			vd = rvd->vdev_child[vdev];
429 
430 		if (cache == ZFS_CACHE_ALL || cache == ZFS_CACHE_METADATA) {
431 			if (vd == NULL)
432 				return (B_TRUE);
433 
434 			if ((vd->vdev_alloc_bias != VDEV_BIAS_SPECIAL &&
435 			    vd->vdev_alloc_bias != VDEV_BIAS_DEDUP) ||
436 			    l2arc_exclude_special == 0)
437 				return (B_TRUE);
438 		}
439 	}
440 
441 	return (B_FALSE);
442 }
443 
444 /*
445  * Instantiates the objset_t in-memory structure corresponding to the
446  * objset_phys_t that's pointed to by the specified blkptr_t.
447  */
448 int
449 dmu_objset_open_impl(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
450     objset_t **osp)
451 {
452 	objset_t *os;
453 	int i, err;
454 
455 	ASSERT(ds == NULL || MUTEX_HELD(&ds->ds_opening_lock));
456 	ASSERT(!BP_IS_REDACTED(bp));
457 
458 	/*
459 	 * We need the pool config lock to get properties.
460 	 */
461 	ASSERT(ds == NULL || dsl_pool_config_held(ds->ds_dir->dd_pool));
462 
463 	/*
464 	 * The $ORIGIN dataset (if it exists) doesn't have an associated
465 	 * objset, so there's no reason to open it. The $ORIGIN dataset
466 	 * will not exist on pools older than SPA_VERSION_ORIGIN.
467 	 */
468 	if (ds != NULL && spa_get_dsl(spa) != NULL &&
469 	    spa_get_dsl(spa)->dp_origin_snap != NULL) {
470 		ASSERT3P(ds->ds_dir, !=,
471 		    spa_get_dsl(spa)->dp_origin_snap->ds_dir);
472 	}
473 
474 	os = kmem_zalloc(sizeof (objset_t), KM_SLEEP);
475 	os->os_dsl_dataset = ds;
476 	os->os_spa = spa;
477 	os->os_rootbp = bp;
478 	if (!BP_IS_HOLE(os->os_rootbp)) {
479 		arc_flags_t aflags = ARC_FLAG_WAIT;
480 		zbookmark_phys_t zb;
481 		int size;
482 		enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;
483 		SET_BOOKMARK(&zb, ds ? ds->ds_object : DMU_META_OBJSET,
484 		    ZB_ROOT_OBJECT, ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
485 
486 		if (dmu_os_is_l2cacheable(os))
487 			aflags |= ARC_FLAG_L2CACHE;
488 
489 		if (ds != NULL && ds->ds_dir->dd_crypto_obj != 0) {
490 			ASSERT3U(BP_GET_COMPRESS(bp), ==, ZIO_COMPRESS_OFF);
491 			ASSERT(BP_IS_AUTHENTICATED(bp));
492 			zio_flags |= ZIO_FLAG_RAW;
493 		}
494 
495 		dprintf_bp(os->os_rootbp, "reading %s", "");
496 		err = arc_read(NULL, spa, os->os_rootbp,
497 		    arc_getbuf_func, &os->os_phys_buf,
498 		    ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);
499 		if (err != 0) {
500 			kmem_free(os, sizeof (objset_t));
501 			/* convert checksum errors into IO errors */
502 			if (err == ECKSUM)
503 				err = SET_ERROR(EIO);
504 			return (err);
505 		}
506 
507 		if (spa_version(spa) < SPA_VERSION_USERSPACE)
508 			size = OBJSET_PHYS_SIZE_V1;
509 		else if (!spa_feature_is_enabled(spa,
510 		    SPA_FEATURE_PROJECT_QUOTA))
511 			size = OBJSET_PHYS_SIZE_V2;
512 		else
513 			size = sizeof (objset_phys_t);
514 
515 		/* Increase the blocksize if we are permitted. */
516 		if (arc_buf_size(os->os_phys_buf) < size) {
517 			arc_buf_t *buf = arc_alloc_buf(spa, &os->os_phys_buf,
518 			    ARC_BUFC_METADATA, size);
519 			memset(buf->b_data, 0, size);
520 			memcpy(buf->b_data, os->os_phys_buf->b_data,
521 			    arc_buf_size(os->os_phys_buf));
522 			arc_buf_destroy(os->os_phys_buf, &os->os_phys_buf);
523 			os->os_phys_buf = buf;
524 		}
525 
526 		os->os_phys = os->os_phys_buf->b_data;
527 		os->os_flags = os->os_phys->os_flags;
528 	} else {
529 		int size = spa_version(spa) >= SPA_VERSION_USERSPACE ?
530 		    sizeof (objset_phys_t) : OBJSET_PHYS_SIZE_V1;
531 		os->os_phys_buf = arc_alloc_buf(spa, &os->os_phys_buf,
532 		    ARC_BUFC_METADATA, size);
533 		os->os_phys = os->os_phys_buf->b_data;
534 		memset(os->os_phys, 0, size);
535 	}
536 	/*
537 	 * These properties will be filled in by the logic in zfs_get_zplprop()
538 	 * when they are queried for the first time.
539 	 */
540 	os->os_version = OBJSET_PROP_UNINITIALIZED;
541 	os->os_normalization = OBJSET_PROP_UNINITIALIZED;
542 	os->os_utf8only = OBJSET_PROP_UNINITIALIZED;
543 	os->os_casesensitivity = OBJSET_PROP_UNINITIALIZED;
544 
545 	/*
546 	 * Note: the changed_cb will be called once before the register
547 	 * func returns, thus changing the checksum/compression from the
548 	 * default (fletcher2/off).  Snapshots don't need to know about
549 	 * checksum/compression/copies.
550 	 */
551 	if (ds != NULL) {
552 		os->os_encrypted = (ds->ds_dir->dd_crypto_obj != 0);
553 
554 		err = dsl_prop_register(ds,
555 		    zfs_prop_to_name(ZFS_PROP_PRIMARYCACHE),
556 		    primary_cache_changed_cb, os);
557 		if (err == 0) {
558 			err = dsl_prop_register(ds,
559 			    zfs_prop_to_name(ZFS_PROP_SECONDARYCACHE),
560 			    secondary_cache_changed_cb, os);
561 		}
562 		if (!ds->ds_is_snapshot) {
563 			if (err == 0) {
564 				err = dsl_prop_register(ds,
565 				    zfs_prop_to_name(ZFS_PROP_CHECKSUM),
566 				    checksum_changed_cb, os);
567 			}
568 			if (err == 0) {
569 				err = dsl_prop_register(ds,
570 				    zfs_prop_to_name(ZFS_PROP_COMPRESSION),
571 				    compression_changed_cb, os);
572 			}
573 			if (err == 0) {
574 				err = dsl_prop_register(ds,
575 				    zfs_prop_to_name(ZFS_PROP_COPIES),
576 				    copies_changed_cb, os);
577 			}
578 			if (err == 0) {
579 				err = dsl_prop_register(ds,
580 				    zfs_prop_to_name(ZFS_PROP_DEDUP),
581 				    dedup_changed_cb, os);
582 			}
583 			if (err == 0) {
584 				err = dsl_prop_register(ds,
585 				    zfs_prop_to_name(ZFS_PROP_LOGBIAS),
586 				    logbias_changed_cb, os);
587 			}
588 			if (err == 0) {
589 				err = dsl_prop_register(ds,
590 				    zfs_prop_to_name(ZFS_PROP_SYNC),
591 				    sync_changed_cb, os);
592 			}
593 			if (err == 0) {
594 				err = dsl_prop_register(ds,
595 				    zfs_prop_to_name(
596 				    ZFS_PROP_REDUNDANT_METADATA),
597 				    redundant_metadata_changed_cb, os);
598 			}
599 			if (err == 0) {
600 				err = dsl_prop_register(ds,
601 				    zfs_prop_to_name(ZFS_PROP_RECORDSIZE),
602 				    recordsize_changed_cb, os);
603 			}
604 			if (err == 0) {
605 				err = dsl_prop_register(ds,
606 				    zfs_prop_to_name(ZFS_PROP_DNODESIZE),
607 				    dnodesize_changed_cb, os);
608 			}
609 			if (err == 0) {
610 				err = dsl_prop_register(ds,
611 				    zfs_prop_to_name(
612 				    ZFS_PROP_SPECIAL_SMALL_BLOCKS),
613 				    smallblk_changed_cb, os);
614 			}
615 		}
616 		if (err != 0) {
617 			arc_buf_destroy(os->os_phys_buf, &os->os_phys_buf);
618 			kmem_free(os, sizeof (objset_t));
619 			return (err);
620 		}
621 	} else {
622 		/* It's the meta-objset. */
623 		os->os_checksum = ZIO_CHECKSUM_FLETCHER_4;
624 		os->os_compress = ZIO_COMPRESS_ON;
625 		os->os_complevel = ZIO_COMPLEVEL_DEFAULT;
626 		os->os_encrypted = B_FALSE;
627 		os->os_copies = spa_max_replication(spa);
628 		os->os_dedup_checksum = ZIO_CHECKSUM_OFF;
629 		os->os_dedup_verify = B_FALSE;
630 		os->os_logbias = ZFS_LOGBIAS_LATENCY;
631 		os->os_sync = ZFS_SYNC_STANDARD;
632 		os->os_primary_cache = ZFS_CACHE_ALL;
633 		os->os_secondary_cache = ZFS_CACHE_ALL;
634 		os->os_dnodesize = DNODE_MIN_SIZE;
635 	}
636 
637 	if (ds == NULL || !ds->ds_is_snapshot)
638 		os->os_zil_header = os->os_phys->os_zil_header;
639 	os->os_zil = zil_alloc(os, &os->os_zil_header);
640 
641 	for (i = 0; i < TXG_SIZE; i++) {
642 		multilist_create(&os->os_dirty_dnodes[i], sizeof (dnode_t),
643 		    offsetof(dnode_t, dn_dirty_link[i]),
644 		    dnode_multilist_index_func);
645 	}
646 	list_create(&os->os_dnodes, sizeof (dnode_t),
647 	    offsetof(dnode_t, dn_link));
648 	list_create(&os->os_downgraded_dbufs, sizeof (dmu_buf_impl_t),
649 	    offsetof(dmu_buf_impl_t, db_link));
650 
651 	list_link_init(&os->os_evicting_node);
652 
653 	mutex_init(&os->os_lock, NULL, MUTEX_DEFAULT, NULL);
654 	mutex_init(&os->os_userused_lock, NULL, MUTEX_DEFAULT, NULL);
655 	mutex_init(&os->os_obj_lock, NULL, MUTEX_DEFAULT, NULL);
656 	mutex_init(&os->os_user_ptr_lock, NULL, MUTEX_DEFAULT, NULL);
657 	os->os_obj_next_percpu_len = boot_ncpus;
658 	os->os_obj_next_percpu = kmem_zalloc(os->os_obj_next_percpu_len *
659 	    sizeof (os->os_obj_next_percpu[0]), KM_SLEEP);
660 
661 	dnode_special_open(os, &os->os_phys->os_meta_dnode,
662 	    DMU_META_DNODE_OBJECT, &os->os_meta_dnode);
663 	if (OBJSET_BUF_HAS_USERUSED(os->os_phys_buf)) {
664 		dnode_special_open(os, &os->os_phys->os_userused_dnode,
665 		    DMU_USERUSED_OBJECT, &os->os_userused_dnode);
666 		dnode_special_open(os, &os->os_phys->os_groupused_dnode,
667 		    DMU_GROUPUSED_OBJECT, &os->os_groupused_dnode);
668 		if (OBJSET_BUF_HAS_PROJECTUSED(os->os_phys_buf))
669 			dnode_special_open(os,
670 			    &os->os_phys->os_projectused_dnode,
671 			    DMU_PROJECTUSED_OBJECT, &os->os_projectused_dnode);
672 	}
673 
674 	mutex_init(&os->os_upgrade_lock, NULL, MUTEX_DEFAULT, NULL);
675 
676 	*osp = os;
677 	return (0);
678 }
679 
680 int
681 dmu_objset_from_ds(dsl_dataset_t *ds, objset_t **osp)
682 {
683 	int err = 0;
684 
685 	/*
686 	 * We need the pool_config lock to manipulate the dsl_dataset_t.
687 	 * Even if the dataset is long-held, we need the pool_config lock
688 	 * to open the objset, as it needs to get properties.
689 	 */
690 	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
691 
692 	mutex_enter(&ds->ds_opening_lock);
693 	if (ds->ds_objset == NULL) {
694 		objset_t *os;
695 		rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
696 		err = dmu_objset_open_impl(dsl_dataset_get_spa(ds),
697 		    ds, dsl_dataset_get_blkptr(ds), &os);
698 		rrw_exit(&ds->ds_bp_rwlock, FTAG);
699 
700 		if (err == 0) {
701 			mutex_enter(&ds->ds_lock);
702 			ASSERT(ds->ds_objset == NULL);
703 			ds->ds_objset = os;
704 			mutex_exit(&ds->ds_lock);
705 		}
706 	}
707 	*osp = ds->ds_objset;
708 	mutex_exit(&ds->ds_opening_lock);
709 	return (err);
710 }
711 
712 /*
713  * Holds the pool while the objset is held.  Therefore only one objset
714  * can be held at a time.
715  */
716 int
717 dmu_objset_hold_flags(const char *name, boolean_t decrypt, const void *tag,
718     objset_t **osp)
719 {
720 	dsl_pool_t *dp;
721 	dsl_dataset_t *ds;
722 	int err;
723 	ds_hold_flags_t flags;
724 
725 	flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
726 	err = dsl_pool_hold(name, tag, &dp);
727 	if (err != 0)
728 		return (err);
729 	err = dsl_dataset_hold_flags(dp, name, flags, tag, &ds);
730 	if (err != 0) {
731 		dsl_pool_rele(dp, tag);
732 		return (err);
733 	}
734 
735 	err = dmu_objset_from_ds(ds, osp);
736 	if (err != 0) {
737 		dsl_dataset_rele(ds, tag);
738 		dsl_pool_rele(dp, tag);
739 	}
740 
741 	return (err);
742 }
743 
744 int
745 dmu_objset_hold(const char *name, const void *tag, objset_t **osp)
746 {
747 	return (dmu_objset_hold_flags(name, B_FALSE, tag, osp));
748 }
749 
750 static int
751 dmu_objset_own_impl(dsl_dataset_t *ds, dmu_objset_type_t type,
752     boolean_t readonly, boolean_t decrypt, const void *tag, objset_t **osp)
753 {
754 	(void) tag;
755 
756 	int err = dmu_objset_from_ds(ds, osp);
757 	if (err != 0) {
758 		return (err);
759 	} else if (type != DMU_OST_ANY && type != (*osp)->os_phys->os_type) {
760 		return (SET_ERROR(EINVAL));
761 	} else if (!readonly && dsl_dataset_is_snapshot(ds)) {
762 		return (SET_ERROR(EROFS));
763 	} else if (!readonly && decrypt &&
764 	    dsl_dir_incompatible_encryption_version(ds->ds_dir)) {
765 		return (SET_ERROR(EROFS));
766 	}
767 
768 	/* if we are decrypting, we can now check MACs in os->os_phys_buf */
769 	if (decrypt && arc_is_unauthenticated((*osp)->os_phys_buf)) {
770 		zbookmark_phys_t zb;
771 
772 		SET_BOOKMARK(&zb, ds->ds_object, ZB_ROOT_OBJECT,
773 		    ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
774 		err = arc_untransform((*osp)->os_phys_buf, (*osp)->os_spa,
775 		    &zb, B_FALSE);
776 		if (err != 0)
777 			return (err);
778 
779 		ASSERT0(arc_is_unauthenticated((*osp)->os_phys_buf));
780 	}
781 
782 	return (0);
783 }
784 
785 /*
786  * dsl_pool must not be held when this is called.
787  * Upon successful return, there will be a longhold on the dataset,
788  * and the dsl_pool will not be held.
789  */
790 int
791 dmu_objset_own(const char *name, dmu_objset_type_t type,
792     boolean_t readonly, boolean_t decrypt, const void *tag, objset_t **osp)
793 {
794 	dsl_pool_t *dp;
795 	dsl_dataset_t *ds;
796 	int err;
797 	ds_hold_flags_t flags;
798 
799 	flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
800 	err = dsl_pool_hold(name, FTAG, &dp);
801 	if (err != 0)
802 		return (err);
803 	err = dsl_dataset_own(dp, name, flags, tag, &ds);
804 	if (err != 0) {
805 		dsl_pool_rele(dp, FTAG);
806 		return (err);
807 	}
808 	err = dmu_objset_own_impl(ds, type, readonly, decrypt, tag, osp);
809 	if (err != 0) {
810 		dsl_dataset_disown(ds, flags, tag);
811 		dsl_pool_rele(dp, FTAG);
812 		return (err);
813 	}
814 
815 	/*
816 	 * User accounting requires the dataset to be decrypted and rw.
817 	 * We also don't begin user accounting during claiming to help
818 	 * speed up pool import times and to keep this txg reserved
819 	 * completely for recovery work.
820 	 */
821 	if (!readonly && !dp->dp_spa->spa_claiming &&
822 	    (ds->ds_dir->dd_crypto_obj == 0 || decrypt)) {
823 		if (dmu_objset_userobjspace_upgradable(*osp) ||
824 		    dmu_objset_projectquota_upgradable(*osp)) {
825 			dmu_objset_id_quota_upgrade(*osp);
826 		} else if (dmu_objset_userused_enabled(*osp)) {
827 			dmu_objset_userspace_upgrade(*osp);
828 		}
829 	}
830 
831 	dsl_pool_rele(dp, FTAG);
832 	return (0);
833 }
834 
835 int
836 dmu_objset_own_obj(dsl_pool_t *dp, uint64_t obj, dmu_objset_type_t type,
837     boolean_t readonly, boolean_t decrypt, const void *tag, objset_t **osp)
838 {
839 	dsl_dataset_t *ds;
840 	int err;
841 	ds_hold_flags_t flags;
842 
843 	flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
844 	err = dsl_dataset_own_obj(dp, obj, flags, tag, &ds);
845 	if (err != 0)
846 		return (err);
847 
848 	err = dmu_objset_own_impl(ds, type, readonly, decrypt, tag, osp);
849 	if (err != 0) {
850 		dsl_dataset_disown(ds, flags, tag);
851 		return (err);
852 	}
853 
854 	return (0);
855 }
856 
857 void
858 dmu_objset_rele_flags(objset_t *os, boolean_t decrypt, const void *tag)
859 {
860 	ds_hold_flags_t flags;
861 	dsl_pool_t *dp = dmu_objset_pool(os);
862 
863 	flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
864 	dsl_dataset_rele_flags(os->os_dsl_dataset, flags, tag);
865 	dsl_pool_rele(dp, tag);
866 }
867 
868 void
869 dmu_objset_rele(objset_t *os, const void *tag)
870 {
871 	dmu_objset_rele_flags(os, B_FALSE, tag);
872 }
873 
874 /*
875  * When we are called, os MUST refer to an objset associated with a dataset
876  * that is owned by 'tag'; that is, is held and long held by 'tag' and ds_owner
877  * == tag.  We will then release and reacquire ownership of the dataset while
878  * holding the pool config_rwlock to avoid intervening namespace or ownership
879  * changes may occur.
880  *
881  * This exists solely to accommodate zfs_ioc_userspace_upgrade()'s desire to
882  * release the hold on its dataset and acquire a new one on the dataset of the
883  * same name so that it can be partially torn down and reconstructed.
884  */
885 void
886 dmu_objset_refresh_ownership(dsl_dataset_t *ds, dsl_dataset_t **newds,
887     boolean_t decrypt, const void *tag)
888 {
889 	dsl_pool_t *dp;
890 	char name[ZFS_MAX_DATASET_NAME_LEN];
891 	ds_hold_flags_t flags;
892 
893 	flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
894 	VERIFY3P(ds, !=, NULL);
895 	VERIFY3P(ds->ds_owner, ==, tag);
896 	VERIFY(dsl_dataset_long_held(ds));
897 
898 	dsl_dataset_name(ds, name);
899 	dp = ds->ds_dir->dd_pool;
900 	dsl_pool_config_enter(dp, FTAG);
901 	dsl_dataset_disown(ds, flags, tag);
902 	VERIFY0(dsl_dataset_own(dp, name, flags, tag, newds));
903 	dsl_pool_config_exit(dp, FTAG);
904 }
905 
906 void
907 dmu_objset_disown(objset_t *os, boolean_t decrypt, const void *tag)
908 {
909 	ds_hold_flags_t flags;
910 
911 	flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
912 	/*
913 	 * Stop upgrading thread
914 	 */
915 	dmu_objset_upgrade_stop(os);
916 	dsl_dataset_disown(os->os_dsl_dataset, flags, tag);
917 }
918 
919 void
920 dmu_objset_evict_dbufs(objset_t *os)
921 {
922 	dnode_t *dn_marker;
923 	dnode_t *dn;
924 
925 	dn_marker = kmem_alloc(sizeof (dnode_t), KM_SLEEP);
926 
927 	mutex_enter(&os->os_lock);
928 	dn = list_head(&os->os_dnodes);
929 	while (dn != NULL) {
930 		/*
931 		 * Skip dnodes without holds.  We have to do this dance
932 		 * because dnode_add_ref() only works if there is already a
933 		 * hold.  If the dnode has no holds, then it has no dbufs.
934 		 */
935 		if (dnode_add_ref(dn, FTAG)) {
936 			list_insert_after(&os->os_dnodes, dn, dn_marker);
937 			mutex_exit(&os->os_lock);
938 
939 			dnode_evict_dbufs(dn);
940 			dnode_rele(dn, FTAG);
941 
942 			mutex_enter(&os->os_lock);
943 			dn = list_next(&os->os_dnodes, dn_marker);
944 			list_remove(&os->os_dnodes, dn_marker);
945 		} else {
946 			dn = list_next(&os->os_dnodes, dn);
947 		}
948 	}
949 	mutex_exit(&os->os_lock);
950 
951 	kmem_free(dn_marker, sizeof (dnode_t));
952 
953 	if (DMU_USERUSED_DNODE(os) != NULL) {
954 		if (DMU_PROJECTUSED_DNODE(os) != NULL)
955 			dnode_evict_dbufs(DMU_PROJECTUSED_DNODE(os));
956 		dnode_evict_dbufs(DMU_GROUPUSED_DNODE(os));
957 		dnode_evict_dbufs(DMU_USERUSED_DNODE(os));
958 	}
959 	dnode_evict_dbufs(DMU_META_DNODE(os));
960 }
961 
962 /*
963  * Objset eviction processing is split into into two pieces.
964  * The first marks the objset as evicting, evicts any dbufs that
965  * have a refcount of zero, and then queues up the objset for the
966  * second phase of eviction.  Once os->os_dnodes has been cleared by
967  * dnode_buf_pageout()->dnode_destroy(), the second phase is executed.
968  * The second phase closes the special dnodes, dequeues the objset from
969  * the list of those undergoing eviction, and finally frees the objset.
970  *
971  * NOTE: Due to asynchronous eviction processing (invocation of
972  *       dnode_buf_pageout()), it is possible for the meta dnode for the
973  *       objset to have no holds even though os->os_dnodes is not empty.
974  */
975 void
976 dmu_objset_evict(objset_t *os)
977 {
978 	dsl_dataset_t *ds = os->os_dsl_dataset;
979 
980 	for (int t = 0; t < TXG_SIZE; t++)
981 		ASSERT(!dmu_objset_is_dirty(os, t));
982 
983 	if (ds)
984 		dsl_prop_unregister_all(ds, os);
985 
986 	if (os->os_sa)
987 		sa_tear_down(os);
988 
989 	dmu_objset_evict_dbufs(os);
990 
991 	mutex_enter(&os->os_lock);
992 	spa_evicting_os_register(os->os_spa, os);
993 	if (list_is_empty(&os->os_dnodes)) {
994 		mutex_exit(&os->os_lock);
995 		dmu_objset_evict_done(os);
996 	} else {
997 		mutex_exit(&os->os_lock);
998 	}
999 
1000 
1001 }
1002 
1003 void
1004 dmu_objset_evict_done(objset_t *os)
1005 {
1006 	ASSERT3P(list_head(&os->os_dnodes), ==, NULL);
1007 
1008 	dnode_special_close(&os->os_meta_dnode);
1009 	if (DMU_USERUSED_DNODE(os)) {
1010 		if (DMU_PROJECTUSED_DNODE(os))
1011 			dnode_special_close(&os->os_projectused_dnode);
1012 		dnode_special_close(&os->os_userused_dnode);
1013 		dnode_special_close(&os->os_groupused_dnode);
1014 	}
1015 	zil_free(os->os_zil);
1016 
1017 	arc_buf_destroy(os->os_phys_buf, &os->os_phys_buf);
1018 
1019 	/*
1020 	 * This is a barrier to prevent the objset from going away in
1021 	 * dnode_move() until we can safely ensure that the objset is still in
1022 	 * use. We consider the objset valid before the barrier and invalid
1023 	 * after the barrier.
1024 	 */
1025 	rw_enter(&os_lock, RW_READER);
1026 	rw_exit(&os_lock);
1027 
1028 	kmem_free(os->os_obj_next_percpu,
1029 	    os->os_obj_next_percpu_len * sizeof (os->os_obj_next_percpu[0]));
1030 
1031 	mutex_destroy(&os->os_lock);
1032 	mutex_destroy(&os->os_userused_lock);
1033 	mutex_destroy(&os->os_obj_lock);
1034 	mutex_destroy(&os->os_user_ptr_lock);
1035 	mutex_destroy(&os->os_upgrade_lock);
1036 	for (int i = 0; i < TXG_SIZE; i++)
1037 		multilist_destroy(&os->os_dirty_dnodes[i]);
1038 	spa_evicting_os_deregister(os->os_spa, os);
1039 	kmem_free(os, sizeof (objset_t));
1040 }
1041 
1042 inode_timespec_t
1043 dmu_objset_snap_cmtime(objset_t *os)
1044 {
1045 	return (dsl_dir_snap_cmtime(os->os_dsl_dataset->ds_dir));
1046 }
1047 
1048 objset_t *
1049 dmu_objset_create_impl_dnstats(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
1050     dmu_objset_type_t type, int levels, int blksz, int ibs, dmu_tx_t *tx)
1051 {
1052 	objset_t *os;
1053 	dnode_t *mdn;
1054 
1055 	ASSERT(dmu_tx_is_syncing(tx));
1056 
1057 	if (blksz == 0)
1058 		blksz = DNODE_BLOCK_SIZE;
1059 	if (ibs == 0)
1060 		ibs = DN_MAX_INDBLKSHIFT;
1061 
1062 	if (ds != NULL)
1063 		VERIFY0(dmu_objset_from_ds(ds, &os));
1064 	else
1065 		VERIFY0(dmu_objset_open_impl(spa, NULL, bp, &os));
1066 
1067 	mdn = DMU_META_DNODE(os);
1068 
1069 	dnode_allocate(mdn, DMU_OT_DNODE, blksz, ibs, DMU_OT_NONE, 0,
1070 	    DNODE_MIN_SLOTS, tx);
1071 
1072 	/*
1073 	 * We don't want to have to increase the meta-dnode's nlevels
1074 	 * later, because then we could do it in quiescing context while
1075 	 * we are also accessing it in open context.
1076 	 *
1077 	 * This precaution is not necessary for the MOS (ds == NULL),
1078 	 * because the MOS is only updated in syncing context.
1079 	 * This is most fortunate: the MOS is the only objset that
1080 	 * needs to be synced multiple times as spa_sync() iterates
1081 	 * to convergence, so minimizing its dn_nlevels matters.
1082 	 */
1083 	if (ds != NULL) {
1084 		if (levels == 0) {
1085 			levels = 1;
1086 
1087 			/*
1088 			 * Determine the number of levels necessary for the
1089 			 * meta-dnode to contain DN_MAX_OBJECT dnodes.  Note
1090 			 * that in order to ensure that we do not overflow
1091 			 * 64 bits, there has to be a nlevels that gives us a
1092 			 * number of blocks > DN_MAX_OBJECT but < 2^64.
1093 			 * Therefore, (mdn->dn_indblkshift - SPA_BLKPTRSHIFT)
1094 			 * (10) must be less than (64 - log2(DN_MAX_OBJECT))
1095 			 * (16).
1096 			 */
1097 			while ((uint64_t)mdn->dn_nblkptr <<
1098 			    (mdn->dn_datablkshift - DNODE_SHIFT + (levels - 1) *
1099 			    (mdn->dn_indblkshift - SPA_BLKPTRSHIFT)) <
1100 			    DN_MAX_OBJECT)
1101 				levels++;
1102 		}
1103 
1104 		mdn->dn_next_nlevels[tx->tx_txg & TXG_MASK] =
1105 		    mdn->dn_nlevels = levels;
1106 	}
1107 
1108 	ASSERT(type != DMU_OST_NONE);
1109 	ASSERT(type != DMU_OST_ANY);
1110 	ASSERT(type < DMU_OST_NUMTYPES);
1111 	os->os_phys->os_type = type;
1112 
1113 	/*
1114 	 * Enable user accounting if it is enabled and this is not an
1115 	 * encrypted receive.
1116 	 */
1117 	if (dmu_objset_userused_enabled(os) &&
1118 	    (!os->os_encrypted || !dmu_objset_is_receiving(os))) {
1119 		os->os_phys->os_flags |= OBJSET_FLAG_USERACCOUNTING_COMPLETE;
1120 		if (dmu_objset_userobjused_enabled(os)) {
1121 			ds->ds_feature_activation[
1122 			    SPA_FEATURE_USEROBJ_ACCOUNTING] = (void *)B_TRUE;
1123 			os->os_phys->os_flags |=
1124 			    OBJSET_FLAG_USEROBJACCOUNTING_COMPLETE;
1125 		}
1126 		if (dmu_objset_projectquota_enabled(os)) {
1127 			ds->ds_feature_activation[
1128 			    SPA_FEATURE_PROJECT_QUOTA] = (void *)B_TRUE;
1129 			os->os_phys->os_flags |=
1130 			    OBJSET_FLAG_PROJECTQUOTA_COMPLETE;
1131 		}
1132 		os->os_flags = os->os_phys->os_flags;
1133 	}
1134 
1135 	dsl_dataset_dirty(ds, tx);
1136 
1137 	return (os);
1138 }
1139 
1140 /* called from dsl for meta-objset */
1141 objset_t *
1142 dmu_objset_create_impl(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
1143     dmu_objset_type_t type, dmu_tx_t *tx)
1144 {
1145 	return (dmu_objset_create_impl_dnstats(spa, ds, bp, type, 0, 0, 0, tx));
1146 }
1147 
1148 typedef struct dmu_objset_create_arg {
1149 	const char *doca_name;
1150 	cred_t *doca_cred;
1151 	proc_t *doca_proc;
1152 	void (*doca_userfunc)(objset_t *os, void *arg,
1153 	    cred_t *cr, dmu_tx_t *tx);
1154 	void *doca_userarg;
1155 	dmu_objset_type_t doca_type;
1156 	uint64_t doca_flags;
1157 	dsl_crypto_params_t *doca_dcp;
1158 } dmu_objset_create_arg_t;
1159 
1160 static int
1161 dmu_objset_create_check(void *arg, dmu_tx_t *tx)
1162 {
1163 	dmu_objset_create_arg_t *doca = arg;
1164 	dsl_pool_t *dp = dmu_tx_pool(tx);
1165 	dsl_dir_t *pdd;
1166 	dsl_dataset_t *parentds;
1167 	objset_t *parentos;
1168 	const char *tail;
1169 	int error;
1170 
1171 	if (strchr(doca->doca_name, '@') != NULL)
1172 		return (SET_ERROR(EINVAL));
1173 
1174 	if (strlen(doca->doca_name) >= ZFS_MAX_DATASET_NAME_LEN)
1175 		return (SET_ERROR(ENAMETOOLONG));
1176 
1177 	if (dataset_nestcheck(doca->doca_name) != 0)
1178 		return (SET_ERROR(ENAMETOOLONG));
1179 
1180 	error = dsl_dir_hold(dp, doca->doca_name, FTAG, &pdd, &tail);
1181 	if (error != 0)
1182 		return (error);
1183 	if (tail == NULL) {
1184 		dsl_dir_rele(pdd, FTAG);
1185 		return (SET_ERROR(EEXIST));
1186 	}
1187 
1188 	error = dmu_objset_create_crypt_check(pdd, doca->doca_dcp, NULL);
1189 	if (error != 0) {
1190 		dsl_dir_rele(pdd, FTAG);
1191 		return (error);
1192 	}
1193 
1194 	error = dsl_fs_ss_limit_check(pdd, 1, ZFS_PROP_FILESYSTEM_LIMIT, NULL,
1195 	    doca->doca_cred, doca->doca_proc);
1196 	if (error != 0) {
1197 		dsl_dir_rele(pdd, FTAG);
1198 		return (error);
1199 	}
1200 
1201 	/* can't create below anything but filesystems (eg. no ZVOLs) */
1202 	error = dsl_dataset_hold_obj(pdd->dd_pool,
1203 	    dsl_dir_phys(pdd)->dd_head_dataset_obj, FTAG, &parentds);
1204 	if (error != 0) {
1205 		dsl_dir_rele(pdd, FTAG);
1206 		return (error);
1207 	}
1208 	error = dmu_objset_from_ds(parentds, &parentos);
1209 	if (error != 0) {
1210 		dsl_dataset_rele(parentds, FTAG);
1211 		dsl_dir_rele(pdd, FTAG);
1212 		return (error);
1213 	}
1214 	if (dmu_objset_type(parentos) != DMU_OST_ZFS) {
1215 		dsl_dataset_rele(parentds, FTAG);
1216 		dsl_dir_rele(pdd, FTAG);
1217 		return (SET_ERROR(ZFS_ERR_WRONG_PARENT));
1218 	}
1219 	dsl_dataset_rele(parentds, FTAG);
1220 	dsl_dir_rele(pdd, FTAG);
1221 
1222 	return (error);
1223 }
1224 
1225 static void
1226 dmu_objset_create_sync(void *arg, dmu_tx_t *tx)
1227 {
1228 	dmu_objset_create_arg_t *doca = arg;
1229 	dsl_pool_t *dp = dmu_tx_pool(tx);
1230 	spa_t *spa = dp->dp_spa;
1231 	dsl_dir_t *pdd;
1232 	const char *tail;
1233 	dsl_dataset_t *ds;
1234 	uint64_t obj;
1235 	blkptr_t *bp;
1236 	objset_t *os;
1237 	zio_t *rzio;
1238 
1239 	VERIFY0(dsl_dir_hold(dp, doca->doca_name, FTAG, &pdd, &tail));
1240 
1241 	obj = dsl_dataset_create_sync(pdd, tail, NULL, doca->doca_flags,
1242 	    doca->doca_cred, doca->doca_dcp, tx);
1243 
1244 	VERIFY0(dsl_dataset_hold_obj_flags(pdd->dd_pool, obj,
1245 	    DS_HOLD_FLAG_DECRYPT, FTAG, &ds));
1246 	rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
1247 	bp = dsl_dataset_get_blkptr(ds);
1248 	os = dmu_objset_create_impl(spa, ds, bp, doca->doca_type, tx);
1249 	rrw_exit(&ds->ds_bp_rwlock, FTAG);
1250 
1251 	if (doca->doca_userfunc != NULL) {
1252 		doca->doca_userfunc(os, doca->doca_userarg,
1253 		    doca->doca_cred, tx);
1254 	}
1255 
1256 	/*
1257 	 * The doca_userfunc() may write out some data that needs to be
1258 	 * encrypted if the dataset is encrypted (specifically the root
1259 	 * directory).  This data must be written out before the encryption
1260 	 * key mapping is removed by dsl_dataset_rele_flags().  Force the
1261 	 * I/O to occur immediately by invoking the relevant sections of
1262 	 * dsl_pool_sync().
1263 	 */
1264 	if (os->os_encrypted) {
1265 		dsl_dataset_t *tmpds = NULL;
1266 		boolean_t need_sync_done = B_FALSE;
1267 
1268 		mutex_enter(&ds->ds_lock);
1269 		ds->ds_owner = FTAG;
1270 		mutex_exit(&ds->ds_lock);
1271 
1272 		rzio = zio_root(spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
1273 		tmpds = txg_list_remove_this(&dp->dp_dirty_datasets, ds,
1274 		    tx->tx_txg);
1275 		if (tmpds != NULL) {
1276 			dsl_dataset_sync(ds, rzio, tx);
1277 			need_sync_done = B_TRUE;
1278 		}
1279 		VERIFY0(zio_wait(rzio));
1280 
1281 		dmu_objset_sync_done(os, tx);
1282 		taskq_wait(dp->dp_sync_taskq);
1283 		if (txg_list_member(&dp->dp_dirty_datasets, ds, tx->tx_txg)) {
1284 			ASSERT3P(ds->ds_key_mapping, !=, NULL);
1285 			key_mapping_rele(spa, ds->ds_key_mapping, ds);
1286 		}
1287 
1288 		rzio = zio_root(spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
1289 		tmpds = txg_list_remove_this(&dp->dp_dirty_datasets, ds,
1290 		    tx->tx_txg);
1291 		if (tmpds != NULL) {
1292 			dmu_buf_rele(ds->ds_dbuf, ds);
1293 			dsl_dataset_sync(ds, rzio, tx);
1294 		}
1295 		VERIFY0(zio_wait(rzio));
1296 
1297 		if (need_sync_done) {
1298 			ASSERT3P(ds->ds_key_mapping, !=, NULL);
1299 			key_mapping_rele(spa, ds->ds_key_mapping, ds);
1300 			dsl_dataset_sync_done(ds, tx);
1301 		}
1302 
1303 		mutex_enter(&ds->ds_lock);
1304 		ds->ds_owner = NULL;
1305 		mutex_exit(&ds->ds_lock);
1306 	}
1307 
1308 	spa_history_log_internal_ds(ds, "create", tx, " ");
1309 
1310 	dsl_dataset_rele_flags(ds, DS_HOLD_FLAG_DECRYPT, FTAG);
1311 	dsl_dir_rele(pdd, FTAG);
1312 }
1313 
1314 int
1315 dmu_objset_create(const char *name, dmu_objset_type_t type, uint64_t flags,
1316     dsl_crypto_params_t *dcp, dmu_objset_create_sync_func_t func, void *arg)
1317 {
1318 	dmu_objset_create_arg_t doca;
1319 	dsl_crypto_params_t tmp_dcp = { 0 };
1320 
1321 	doca.doca_name = name;
1322 	doca.doca_cred = CRED();
1323 	doca.doca_proc = curproc;
1324 	doca.doca_flags = flags;
1325 	doca.doca_userfunc = func;
1326 	doca.doca_userarg = arg;
1327 	doca.doca_type = type;
1328 
1329 	/*
1330 	 * Some callers (mostly for testing) do not provide a dcp on their
1331 	 * own but various code inside the sync task will require it to be
1332 	 * allocated. Rather than adding NULL checks throughout this code
1333 	 * or adding dummy dcp's to all of the callers we simply create a
1334 	 * dummy one here and use that. This zero dcp will have the same
1335 	 * effect as asking for inheritance of all encryption params.
1336 	 */
1337 	doca.doca_dcp = (dcp != NULL) ? dcp : &tmp_dcp;
1338 
1339 	int rv = dsl_sync_task(name,
1340 	    dmu_objset_create_check, dmu_objset_create_sync, &doca,
1341 	    6, ZFS_SPACE_CHECK_NORMAL);
1342 
1343 	if (rv == 0)
1344 		zvol_create_minor(name);
1345 	return (rv);
1346 }
1347 
1348 typedef struct dmu_objset_clone_arg {
1349 	const char *doca_clone;
1350 	const char *doca_origin;
1351 	cred_t *doca_cred;
1352 	proc_t *doca_proc;
1353 } dmu_objset_clone_arg_t;
1354 
1355 static int
1356 dmu_objset_clone_check(void *arg, dmu_tx_t *tx)
1357 {
1358 	dmu_objset_clone_arg_t *doca = arg;
1359 	dsl_dir_t *pdd;
1360 	const char *tail;
1361 	int error;
1362 	dsl_dataset_t *origin;
1363 	dsl_pool_t *dp = dmu_tx_pool(tx);
1364 
1365 	if (strchr(doca->doca_clone, '@') != NULL)
1366 		return (SET_ERROR(EINVAL));
1367 
1368 	if (strlen(doca->doca_clone) >= ZFS_MAX_DATASET_NAME_LEN)
1369 		return (SET_ERROR(ENAMETOOLONG));
1370 
1371 	error = dsl_dir_hold(dp, doca->doca_clone, FTAG, &pdd, &tail);
1372 	if (error != 0)
1373 		return (error);
1374 	if (tail == NULL) {
1375 		dsl_dir_rele(pdd, FTAG);
1376 		return (SET_ERROR(EEXIST));
1377 	}
1378 
1379 	error = dsl_fs_ss_limit_check(pdd, 1, ZFS_PROP_FILESYSTEM_LIMIT, NULL,
1380 	    doca->doca_cred, doca->doca_proc);
1381 	if (error != 0) {
1382 		dsl_dir_rele(pdd, FTAG);
1383 		return (SET_ERROR(EDQUOT));
1384 	}
1385 
1386 	error = dsl_dataset_hold(dp, doca->doca_origin, FTAG, &origin);
1387 	if (error != 0) {
1388 		dsl_dir_rele(pdd, FTAG);
1389 		return (error);
1390 	}
1391 
1392 	/* You can only clone snapshots, not the head datasets. */
1393 	if (!origin->ds_is_snapshot) {
1394 		dsl_dataset_rele(origin, FTAG);
1395 		dsl_dir_rele(pdd, FTAG);
1396 		return (SET_ERROR(EINVAL));
1397 	}
1398 
1399 	dsl_dataset_rele(origin, FTAG);
1400 	dsl_dir_rele(pdd, FTAG);
1401 
1402 	return (0);
1403 }
1404 
1405 static void
1406 dmu_objset_clone_sync(void *arg, dmu_tx_t *tx)
1407 {
1408 	dmu_objset_clone_arg_t *doca = arg;
1409 	dsl_pool_t *dp = dmu_tx_pool(tx);
1410 	dsl_dir_t *pdd;
1411 	const char *tail;
1412 	dsl_dataset_t *origin, *ds;
1413 	uint64_t obj;
1414 	char namebuf[ZFS_MAX_DATASET_NAME_LEN];
1415 
1416 	VERIFY0(dsl_dir_hold(dp, doca->doca_clone, FTAG, &pdd, &tail));
1417 	VERIFY0(dsl_dataset_hold(dp, doca->doca_origin, FTAG, &origin));
1418 
1419 	obj = dsl_dataset_create_sync(pdd, tail, origin, 0,
1420 	    doca->doca_cred, NULL, tx);
1421 
1422 	VERIFY0(dsl_dataset_hold_obj(pdd->dd_pool, obj, FTAG, &ds));
1423 	dsl_dataset_name(origin, namebuf);
1424 	spa_history_log_internal_ds(ds, "clone", tx,
1425 	    "origin=%s (%llu)", namebuf, (u_longlong_t)origin->ds_object);
1426 	dsl_dataset_rele(ds, FTAG);
1427 	dsl_dataset_rele(origin, FTAG);
1428 	dsl_dir_rele(pdd, FTAG);
1429 }
1430 
1431 int
1432 dmu_objset_clone(const char *clone, const char *origin)
1433 {
1434 	dmu_objset_clone_arg_t doca;
1435 
1436 	doca.doca_clone = clone;
1437 	doca.doca_origin = origin;
1438 	doca.doca_cred = CRED();
1439 	doca.doca_proc = curproc;
1440 
1441 	int rv = dsl_sync_task(clone,
1442 	    dmu_objset_clone_check, dmu_objset_clone_sync, &doca,
1443 	    6, ZFS_SPACE_CHECK_NORMAL);
1444 
1445 	if (rv == 0)
1446 		zvol_create_minor(clone);
1447 
1448 	return (rv);
1449 }
1450 
1451 int
1452 dmu_objset_snapshot_one(const char *fsname, const char *snapname)
1453 {
1454 	int err;
1455 	char *longsnap = kmem_asprintf("%s@%s", fsname, snapname);
1456 	nvlist_t *snaps = fnvlist_alloc();
1457 
1458 	fnvlist_add_boolean(snaps, longsnap);
1459 	kmem_strfree(longsnap);
1460 	err = dsl_dataset_snapshot(snaps, NULL, NULL);
1461 	fnvlist_free(snaps);
1462 	return (err);
1463 }
1464 
1465 static void
1466 dmu_objset_upgrade_task_cb(void *data)
1467 {
1468 	objset_t *os = data;
1469 
1470 	mutex_enter(&os->os_upgrade_lock);
1471 	os->os_upgrade_status = EINTR;
1472 	if (!os->os_upgrade_exit) {
1473 		int status;
1474 
1475 		mutex_exit(&os->os_upgrade_lock);
1476 
1477 		status = os->os_upgrade_cb(os);
1478 
1479 		mutex_enter(&os->os_upgrade_lock);
1480 
1481 		os->os_upgrade_status = status;
1482 	}
1483 	os->os_upgrade_exit = B_TRUE;
1484 	os->os_upgrade_id = 0;
1485 	mutex_exit(&os->os_upgrade_lock);
1486 	dsl_dataset_long_rele(dmu_objset_ds(os), upgrade_tag);
1487 }
1488 
1489 static void
1490 dmu_objset_upgrade(objset_t *os, dmu_objset_upgrade_cb_t cb)
1491 {
1492 	if (os->os_upgrade_id != 0)
1493 		return;
1494 
1495 	ASSERT(dsl_pool_config_held(dmu_objset_pool(os)));
1496 	dsl_dataset_long_hold(dmu_objset_ds(os), upgrade_tag);
1497 
1498 	mutex_enter(&os->os_upgrade_lock);
1499 	if (os->os_upgrade_id == 0 && os->os_upgrade_status == 0) {
1500 		os->os_upgrade_exit = B_FALSE;
1501 		os->os_upgrade_cb = cb;
1502 		os->os_upgrade_id = taskq_dispatch(
1503 		    os->os_spa->spa_upgrade_taskq,
1504 		    dmu_objset_upgrade_task_cb, os, TQ_SLEEP);
1505 		if (os->os_upgrade_id == TASKQID_INVALID) {
1506 			dsl_dataset_long_rele(dmu_objset_ds(os), upgrade_tag);
1507 			os->os_upgrade_status = ENOMEM;
1508 		}
1509 	} else {
1510 		dsl_dataset_long_rele(dmu_objset_ds(os), upgrade_tag);
1511 	}
1512 	mutex_exit(&os->os_upgrade_lock);
1513 }
1514 
1515 static void
1516 dmu_objset_upgrade_stop(objset_t *os)
1517 {
1518 	mutex_enter(&os->os_upgrade_lock);
1519 	os->os_upgrade_exit = B_TRUE;
1520 	if (os->os_upgrade_id != 0) {
1521 		taskqid_t id = os->os_upgrade_id;
1522 
1523 		os->os_upgrade_id = 0;
1524 		mutex_exit(&os->os_upgrade_lock);
1525 
1526 		if ((taskq_cancel_id(os->os_spa->spa_upgrade_taskq, id)) == 0) {
1527 			dsl_dataset_long_rele(dmu_objset_ds(os), upgrade_tag);
1528 		}
1529 		txg_wait_synced(os->os_spa->spa_dsl_pool, 0);
1530 	} else {
1531 		mutex_exit(&os->os_upgrade_lock);
1532 	}
1533 }
1534 
1535 static void
1536 dmu_objset_sync_dnodes(multilist_sublist_t *list, dmu_tx_t *tx)
1537 {
1538 	dnode_t *dn;
1539 
1540 	while ((dn = multilist_sublist_head(list)) != NULL) {
1541 		ASSERT(dn->dn_object != DMU_META_DNODE_OBJECT);
1542 		ASSERT(dn->dn_dbuf->db_data_pending);
1543 		/*
1544 		 * Initialize dn_zio outside dnode_sync() because the
1545 		 * meta-dnode needs to set it outside dnode_sync().
1546 		 */
1547 		dn->dn_zio = dn->dn_dbuf->db_data_pending->dr_zio;
1548 		ASSERT(dn->dn_zio);
1549 
1550 		ASSERT3U(dn->dn_nlevels, <=, DN_MAX_LEVELS);
1551 		multilist_sublist_remove(list, dn);
1552 
1553 		/*
1554 		 * See the comment above dnode_rele_task() for an explanation
1555 		 * of why this dnode hold is always needed (even when not
1556 		 * doing user accounting).
1557 		 */
1558 		multilist_t *newlist = &dn->dn_objset->os_synced_dnodes;
1559 		(void) dnode_add_ref(dn, newlist);
1560 		multilist_insert(newlist, dn);
1561 
1562 		dnode_sync(dn, tx);
1563 	}
1564 }
1565 
1566 static void
1567 dmu_objset_write_ready(zio_t *zio, arc_buf_t *abuf, void *arg)
1568 {
1569 	(void) abuf;
1570 	blkptr_t *bp = zio->io_bp;
1571 	objset_t *os = arg;
1572 	dnode_phys_t *dnp = &os->os_phys->os_meta_dnode;
1573 	uint64_t fill = 0;
1574 
1575 	ASSERT(!BP_IS_EMBEDDED(bp));
1576 	ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_OBJSET);
1577 	ASSERT0(BP_GET_LEVEL(bp));
1578 
1579 	/*
1580 	 * Update rootbp fill count: it should be the number of objects
1581 	 * allocated in the object set (not counting the "special"
1582 	 * objects that are stored in the objset_phys_t -- the meta
1583 	 * dnode and user/group/project accounting objects).
1584 	 */
1585 	for (int i = 0; i < dnp->dn_nblkptr; i++)
1586 		fill += BP_GET_FILL(&dnp->dn_blkptr[i]);
1587 
1588 	BP_SET_FILL(bp, fill);
1589 
1590 	if (os->os_dsl_dataset != NULL)
1591 		rrw_enter(&os->os_dsl_dataset->ds_bp_rwlock, RW_WRITER, FTAG);
1592 	*os->os_rootbp = *bp;
1593 	if (os->os_dsl_dataset != NULL)
1594 		rrw_exit(&os->os_dsl_dataset->ds_bp_rwlock, FTAG);
1595 }
1596 
1597 static void
1598 dmu_objset_write_done(zio_t *zio, arc_buf_t *abuf, void *arg)
1599 {
1600 	(void) abuf;
1601 	blkptr_t *bp = zio->io_bp;
1602 	blkptr_t *bp_orig = &zio->io_bp_orig;
1603 	objset_t *os = arg;
1604 
1605 	if (zio->io_flags & ZIO_FLAG_IO_REWRITE) {
1606 		ASSERT(BP_EQUAL(bp, bp_orig));
1607 	} else {
1608 		dsl_dataset_t *ds = os->os_dsl_dataset;
1609 		dmu_tx_t *tx = os->os_synctx;
1610 
1611 		(void) dsl_dataset_block_kill(ds, bp_orig, tx, B_TRUE);
1612 		dsl_dataset_block_born(ds, bp, tx);
1613 	}
1614 	kmem_free(bp, sizeof (*bp));
1615 }
1616 
1617 typedef struct sync_dnodes_arg {
1618 	multilist_t *sda_list;
1619 	int sda_sublist_idx;
1620 	multilist_t *sda_newlist;
1621 	dmu_tx_t *sda_tx;
1622 } sync_dnodes_arg_t;
1623 
1624 static void
1625 sync_dnodes_task(void *arg)
1626 {
1627 	sync_dnodes_arg_t *sda = arg;
1628 
1629 	multilist_sublist_t *ms =
1630 	    multilist_sublist_lock(sda->sda_list, sda->sda_sublist_idx);
1631 
1632 	dmu_objset_sync_dnodes(ms, sda->sda_tx);
1633 
1634 	multilist_sublist_unlock(ms);
1635 
1636 	kmem_free(sda, sizeof (*sda));
1637 }
1638 
1639 
1640 /* called from dsl */
1641 void
1642 dmu_objset_sync(objset_t *os, zio_t *pio, dmu_tx_t *tx)
1643 {
1644 	int txgoff;
1645 	zbookmark_phys_t zb;
1646 	zio_prop_t zp;
1647 	zio_t *zio;
1648 	list_t *list;
1649 	dbuf_dirty_record_t *dr;
1650 	int num_sublists;
1651 	multilist_t *ml;
1652 	blkptr_t *blkptr_copy = kmem_alloc(sizeof (*os->os_rootbp), KM_SLEEP);
1653 	*blkptr_copy = *os->os_rootbp;
1654 
1655 	dprintf_ds(os->os_dsl_dataset, "txg=%llu\n", (u_longlong_t)tx->tx_txg);
1656 
1657 	ASSERT(dmu_tx_is_syncing(tx));
1658 	/* XXX the write_done callback should really give us the tx... */
1659 	os->os_synctx = tx;
1660 
1661 	if (os->os_dsl_dataset == NULL) {
1662 		/*
1663 		 * This is the MOS.  If we have upgraded,
1664 		 * spa_max_replication() could change, so reset
1665 		 * os_copies here.
1666 		 */
1667 		os->os_copies = spa_max_replication(os->os_spa);
1668 	}
1669 
1670 	/*
1671 	 * Create the root block IO
1672 	 */
1673 	SET_BOOKMARK(&zb, os->os_dsl_dataset ?
1674 	    os->os_dsl_dataset->ds_object : DMU_META_OBJSET,
1675 	    ZB_ROOT_OBJECT, ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
1676 	arc_release(os->os_phys_buf, &os->os_phys_buf);
1677 
1678 	dmu_write_policy(os, NULL, 0, 0, &zp);
1679 
1680 	/*
1681 	 * If we are either claiming the ZIL or doing a raw receive, write
1682 	 * out the os_phys_buf raw. Neither of these actions will effect the
1683 	 * MAC at this point.
1684 	 */
1685 	if (os->os_raw_receive ||
1686 	    os->os_next_write_raw[tx->tx_txg & TXG_MASK]) {
1687 		ASSERT(os->os_encrypted);
1688 		arc_convert_to_raw(os->os_phys_buf,
1689 		    os->os_dsl_dataset->ds_object, ZFS_HOST_BYTEORDER,
1690 		    DMU_OT_OBJSET, NULL, NULL, NULL);
1691 	}
1692 
1693 	zio = arc_write(pio, os->os_spa, tx->tx_txg,
1694 	    blkptr_copy, os->os_phys_buf, dmu_os_is_l2cacheable(os),
1695 	    &zp, dmu_objset_write_ready, NULL, NULL, dmu_objset_write_done,
1696 	    os, ZIO_PRIORITY_ASYNC_WRITE, ZIO_FLAG_MUSTSUCCEED, &zb);
1697 
1698 	/*
1699 	 * In the codepath dsl_dataset_sync()->dmu_objset_sync() we cannot
1700 	 * rely on the zio above completing and calling back
1701 	 * dmu_objset_write_done()->dsl_dataset_block_born() before
1702 	 * dsl_dataset_sync() actually activates feature flags near its end.
1703 	 * Decide here if any features need to be activated, before
1704 	 * dsl_dataset_sync() completes its run.
1705 	 */
1706 	dsl_dataset_feature_set_activation(blkptr_copy, os->os_dsl_dataset);
1707 
1708 	/*
1709 	 * Sync special dnodes - the parent IO for the sync is the root block
1710 	 */
1711 	DMU_META_DNODE(os)->dn_zio = zio;
1712 	dnode_sync(DMU_META_DNODE(os), tx);
1713 
1714 	os->os_phys->os_flags = os->os_flags;
1715 
1716 	if (DMU_USERUSED_DNODE(os) &&
1717 	    DMU_USERUSED_DNODE(os)->dn_type != DMU_OT_NONE) {
1718 		DMU_USERUSED_DNODE(os)->dn_zio = zio;
1719 		dnode_sync(DMU_USERUSED_DNODE(os), tx);
1720 		DMU_GROUPUSED_DNODE(os)->dn_zio = zio;
1721 		dnode_sync(DMU_GROUPUSED_DNODE(os), tx);
1722 	}
1723 
1724 	if (DMU_PROJECTUSED_DNODE(os) &&
1725 	    DMU_PROJECTUSED_DNODE(os)->dn_type != DMU_OT_NONE) {
1726 		DMU_PROJECTUSED_DNODE(os)->dn_zio = zio;
1727 		dnode_sync(DMU_PROJECTUSED_DNODE(os), tx);
1728 	}
1729 
1730 	txgoff = tx->tx_txg & TXG_MASK;
1731 
1732 	/*
1733 	 * We must create the list here because it uses the
1734 	 * dn_dirty_link[] of this txg.  But it may already
1735 	 * exist because we call dsl_dataset_sync() twice per txg.
1736 	 */
1737 	if (os->os_synced_dnodes.ml_sublists == NULL) {
1738 		multilist_create(&os->os_synced_dnodes, sizeof (dnode_t),
1739 		    offsetof(dnode_t, dn_dirty_link[txgoff]),
1740 		    dnode_multilist_index_func);
1741 	} else {
1742 		ASSERT3U(os->os_synced_dnodes.ml_offset, ==,
1743 		    offsetof(dnode_t, dn_dirty_link[txgoff]));
1744 	}
1745 
1746 	ml = &os->os_dirty_dnodes[txgoff];
1747 	num_sublists = multilist_get_num_sublists(ml);
1748 	for (int i = 0; i < num_sublists; i++) {
1749 		if (multilist_sublist_is_empty_idx(ml, i))
1750 			continue;
1751 		sync_dnodes_arg_t *sda = kmem_alloc(sizeof (*sda), KM_SLEEP);
1752 		sda->sda_list = ml;
1753 		sda->sda_sublist_idx = i;
1754 		sda->sda_tx = tx;
1755 		(void) taskq_dispatch(dmu_objset_pool(os)->dp_sync_taskq,
1756 		    sync_dnodes_task, sda, 0);
1757 		/* callback frees sda */
1758 	}
1759 	taskq_wait(dmu_objset_pool(os)->dp_sync_taskq);
1760 
1761 	list = &DMU_META_DNODE(os)->dn_dirty_records[txgoff];
1762 	while ((dr = list_head(list)) != NULL) {
1763 		ASSERT0(dr->dr_dbuf->db_level);
1764 		list_remove(list, dr);
1765 		zio_nowait(dr->dr_zio);
1766 	}
1767 
1768 	/* Enable dnode backfill if enough objects have been freed. */
1769 	if (os->os_freed_dnodes >= dmu_rescan_dnode_threshold) {
1770 		os->os_rescan_dnodes = B_TRUE;
1771 		os->os_freed_dnodes = 0;
1772 	}
1773 
1774 	/*
1775 	 * Free intent log blocks up to this tx.
1776 	 */
1777 	zil_sync(os->os_zil, tx);
1778 	os->os_phys->os_zil_header = os->os_zil_header;
1779 	zio_nowait(zio);
1780 }
1781 
1782 boolean_t
1783 dmu_objset_is_dirty(objset_t *os, uint64_t txg)
1784 {
1785 	return (!multilist_is_empty(&os->os_dirty_dnodes[txg & TXG_MASK]));
1786 }
1787 
1788 static file_info_cb_t *file_cbs[DMU_OST_NUMTYPES];
1789 
1790 void
1791 dmu_objset_register_type(dmu_objset_type_t ost, file_info_cb_t *cb)
1792 {
1793 	file_cbs[ost] = cb;
1794 }
1795 
1796 int
1797 dmu_get_file_info(objset_t *os, dmu_object_type_t bonustype, const void *data,
1798     zfs_file_info_t *zfi)
1799 {
1800 	file_info_cb_t *cb = file_cbs[os->os_phys->os_type];
1801 	if (cb == NULL)
1802 		return (EINVAL);
1803 	return (cb(bonustype, data, zfi));
1804 }
1805 
1806 boolean_t
1807 dmu_objset_userused_enabled(objset_t *os)
1808 {
1809 	return (spa_version(os->os_spa) >= SPA_VERSION_USERSPACE &&
1810 	    file_cbs[os->os_phys->os_type] != NULL &&
1811 	    DMU_USERUSED_DNODE(os) != NULL);
1812 }
1813 
1814 boolean_t
1815 dmu_objset_userobjused_enabled(objset_t *os)
1816 {
1817 	return (dmu_objset_userused_enabled(os) &&
1818 	    spa_feature_is_enabled(os->os_spa, SPA_FEATURE_USEROBJ_ACCOUNTING));
1819 }
1820 
1821 boolean_t
1822 dmu_objset_projectquota_enabled(objset_t *os)
1823 {
1824 	return (file_cbs[os->os_phys->os_type] != NULL &&
1825 	    DMU_PROJECTUSED_DNODE(os) != NULL &&
1826 	    spa_feature_is_enabled(os->os_spa, SPA_FEATURE_PROJECT_QUOTA));
1827 }
1828 
1829 typedef struct userquota_node {
1830 	/* must be in the first filed, see userquota_update_cache() */
1831 	char		uqn_id[20 + DMU_OBJACCT_PREFIX_LEN];
1832 	int64_t		uqn_delta;
1833 	avl_node_t	uqn_node;
1834 } userquota_node_t;
1835 
1836 typedef struct userquota_cache {
1837 	avl_tree_t uqc_user_deltas;
1838 	avl_tree_t uqc_group_deltas;
1839 	avl_tree_t uqc_project_deltas;
1840 } userquota_cache_t;
1841 
1842 static int
1843 userquota_compare(const void *l, const void *r)
1844 {
1845 	const userquota_node_t *luqn = l;
1846 	const userquota_node_t *ruqn = r;
1847 	int rv;
1848 
1849 	/*
1850 	 * NB: can only access uqn_id because userquota_update_cache() doesn't
1851 	 * pass in an entire userquota_node_t.
1852 	 */
1853 	rv = strcmp(luqn->uqn_id, ruqn->uqn_id);
1854 
1855 	return (TREE_ISIGN(rv));
1856 }
1857 
1858 static void
1859 do_userquota_cacheflush(objset_t *os, userquota_cache_t *cache, dmu_tx_t *tx)
1860 {
1861 	void *cookie;
1862 	userquota_node_t *uqn;
1863 
1864 	ASSERT(dmu_tx_is_syncing(tx));
1865 
1866 	cookie = NULL;
1867 	while ((uqn = avl_destroy_nodes(&cache->uqc_user_deltas,
1868 	    &cookie)) != NULL) {
1869 		/*
1870 		 * os_userused_lock protects against concurrent calls to
1871 		 * zap_increment_int().  It's needed because zap_increment_int()
1872 		 * is not thread-safe (i.e. not atomic).
1873 		 */
1874 		mutex_enter(&os->os_userused_lock);
1875 		VERIFY0(zap_increment(os, DMU_USERUSED_OBJECT,
1876 		    uqn->uqn_id, uqn->uqn_delta, tx));
1877 		mutex_exit(&os->os_userused_lock);
1878 		kmem_free(uqn, sizeof (*uqn));
1879 	}
1880 	avl_destroy(&cache->uqc_user_deltas);
1881 
1882 	cookie = NULL;
1883 	while ((uqn = avl_destroy_nodes(&cache->uqc_group_deltas,
1884 	    &cookie)) != NULL) {
1885 		mutex_enter(&os->os_userused_lock);
1886 		VERIFY0(zap_increment(os, DMU_GROUPUSED_OBJECT,
1887 		    uqn->uqn_id, uqn->uqn_delta, tx));
1888 		mutex_exit(&os->os_userused_lock);
1889 		kmem_free(uqn, sizeof (*uqn));
1890 	}
1891 	avl_destroy(&cache->uqc_group_deltas);
1892 
1893 	if (dmu_objset_projectquota_enabled(os)) {
1894 		cookie = NULL;
1895 		while ((uqn = avl_destroy_nodes(&cache->uqc_project_deltas,
1896 		    &cookie)) != NULL) {
1897 			mutex_enter(&os->os_userused_lock);
1898 			VERIFY0(zap_increment(os, DMU_PROJECTUSED_OBJECT,
1899 			    uqn->uqn_id, uqn->uqn_delta, tx));
1900 			mutex_exit(&os->os_userused_lock);
1901 			kmem_free(uqn, sizeof (*uqn));
1902 		}
1903 		avl_destroy(&cache->uqc_project_deltas);
1904 	}
1905 }
1906 
1907 static void
1908 userquota_update_cache(avl_tree_t *avl, const char *id, int64_t delta)
1909 {
1910 	userquota_node_t *uqn;
1911 	avl_index_t idx;
1912 
1913 	ASSERT(strlen(id) < sizeof (uqn->uqn_id));
1914 	/*
1915 	 * Use id directly for searching because uqn_id is the first field of
1916 	 * userquota_node_t and fields after uqn_id won't be accessed in
1917 	 * avl_find().
1918 	 */
1919 	uqn = avl_find(avl, (const void *)id, &idx);
1920 	if (uqn == NULL) {
1921 		uqn = kmem_zalloc(sizeof (*uqn), KM_SLEEP);
1922 		strlcpy(uqn->uqn_id, id, sizeof (uqn->uqn_id));
1923 		avl_insert(avl, uqn, idx);
1924 	}
1925 	uqn->uqn_delta += delta;
1926 }
1927 
1928 static void
1929 do_userquota_update(objset_t *os, userquota_cache_t *cache, uint64_t used,
1930     uint64_t flags, uint64_t user, uint64_t group, uint64_t project,
1931     boolean_t subtract)
1932 {
1933 	if (flags & DNODE_FLAG_USERUSED_ACCOUNTED) {
1934 		int64_t delta = DNODE_MIN_SIZE + used;
1935 		char name[20];
1936 
1937 		if (subtract)
1938 			delta = -delta;
1939 
1940 		(void) snprintf(name, sizeof (name), "%llx", (longlong_t)user);
1941 		userquota_update_cache(&cache->uqc_user_deltas, name, delta);
1942 
1943 		(void) snprintf(name, sizeof (name), "%llx", (longlong_t)group);
1944 		userquota_update_cache(&cache->uqc_group_deltas, name, delta);
1945 
1946 		if (dmu_objset_projectquota_enabled(os)) {
1947 			(void) snprintf(name, sizeof (name), "%llx",
1948 			    (longlong_t)project);
1949 			userquota_update_cache(&cache->uqc_project_deltas,
1950 			    name, delta);
1951 		}
1952 	}
1953 }
1954 
1955 static void
1956 do_userobjquota_update(objset_t *os, userquota_cache_t *cache, uint64_t flags,
1957     uint64_t user, uint64_t group, uint64_t project, boolean_t subtract)
1958 {
1959 	if (flags & DNODE_FLAG_USEROBJUSED_ACCOUNTED) {
1960 		char name[20 + DMU_OBJACCT_PREFIX_LEN];
1961 		int delta = subtract ? -1 : 1;
1962 
1963 		(void) snprintf(name, sizeof (name), DMU_OBJACCT_PREFIX "%llx",
1964 		    (longlong_t)user);
1965 		userquota_update_cache(&cache->uqc_user_deltas, name, delta);
1966 
1967 		(void) snprintf(name, sizeof (name), DMU_OBJACCT_PREFIX "%llx",
1968 		    (longlong_t)group);
1969 		userquota_update_cache(&cache->uqc_group_deltas, name, delta);
1970 
1971 		if (dmu_objset_projectquota_enabled(os)) {
1972 			(void) snprintf(name, sizeof (name),
1973 			    DMU_OBJACCT_PREFIX "%llx", (longlong_t)project);
1974 			userquota_update_cache(&cache->uqc_project_deltas,
1975 			    name, delta);
1976 		}
1977 	}
1978 }
1979 
1980 typedef struct userquota_updates_arg {
1981 	objset_t *uua_os;
1982 	int uua_sublist_idx;
1983 	dmu_tx_t *uua_tx;
1984 } userquota_updates_arg_t;
1985 
1986 static void
1987 userquota_updates_task(void *arg)
1988 {
1989 	userquota_updates_arg_t *uua = arg;
1990 	objset_t *os = uua->uua_os;
1991 	dmu_tx_t *tx = uua->uua_tx;
1992 	dnode_t *dn;
1993 	userquota_cache_t cache = { { 0 } };
1994 
1995 	multilist_sublist_t *list =
1996 	    multilist_sublist_lock(&os->os_synced_dnodes, uua->uua_sublist_idx);
1997 
1998 	ASSERT(multilist_sublist_head(list) == NULL ||
1999 	    dmu_objset_userused_enabled(os));
2000 	avl_create(&cache.uqc_user_deltas, userquota_compare,
2001 	    sizeof (userquota_node_t), offsetof(userquota_node_t, uqn_node));
2002 	avl_create(&cache.uqc_group_deltas, userquota_compare,
2003 	    sizeof (userquota_node_t), offsetof(userquota_node_t, uqn_node));
2004 	if (dmu_objset_projectquota_enabled(os))
2005 		avl_create(&cache.uqc_project_deltas, userquota_compare,
2006 		    sizeof (userquota_node_t), offsetof(userquota_node_t,
2007 		    uqn_node));
2008 
2009 	while ((dn = multilist_sublist_head(list)) != NULL) {
2010 		int flags;
2011 		ASSERT(!DMU_OBJECT_IS_SPECIAL(dn->dn_object));
2012 		ASSERT(dn->dn_phys->dn_type == DMU_OT_NONE ||
2013 		    dn->dn_phys->dn_flags &
2014 		    DNODE_FLAG_USERUSED_ACCOUNTED);
2015 
2016 		flags = dn->dn_id_flags;
2017 		ASSERT(flags);
2018 		if (flags & DN_ID_OLD_EXIST)  {
2019 			do_userquota_update(os, &cache, dn->dn_oldused,
2020 			    dn->dn_oldflags, dn->dn_olduid, dn->dn_oldgid,
2021 			    dn->dn_oldprojid, B_TRUE);
2022 			do_userobjquota_update(os, &cache, dn->dn_oldflags,
2023 			    dn->dn_olduid, dn->dn_oldgid,
2024 			    dn->dn_oldprojid, B_TRUE);
2025 		}
2026 		if (flags & DN_ID_NEW_EXIST) {
2027 			do_userquota_update(os, &cache,
2028 			    DN_USED_BYTES(dn->dn_phys), dn->dn_phys->dn_flags,
2029 			    dn->dn_newuid, dn->dn_newgid,
2030 			    dn->dn_newprojid, B_FALSE);
2031 			do_userobjquota_update(os, &cache,
2032 			    dn->dn_phys->dn_flags, dn->dn_newuid, dn->dn_newgid,
2033 			    dn->dn_newprojid, B_FALSE);
2034 		}
2035 
2036 		mutex_enter(&dn->dn_mtx);
2037 		dn->dn_oldused = 0;
2038 		dn->dn_oldflags = 0;
2039 		if (dn->dn_id_flags & DN_ID_NEW_EXIST) {
2040 			dn->dn_olduid = dn->dn_newuid;
2041 			dn->dn_oldgid = dn->dn_newgid;
2042 			dn->dn_oldprojid = dn->dn_newprojid;
2043 			dn->dn_id_flags |= DN_ID_OLD_EXIST;
2044 			if (dn->dn_bonuslen == 0)
2045 				dn->dn_id_flags |= DN_ID_CHKED_SPILL;
2046 			else
2047 				dn->dn_id_flags |= DN_ID_CHKED_BONUS;
2048 		}
2049 		dn->dn_id_flags &= ~(DN_ID_NEW_EXIST);
2050 		mutex_exit(&dn->dn_mtx);
2051 
2052 		multilist_sublist_remove(list, dn);
2053 		dnode_rele(dn, &os->os_synced_dnodes);
2054 	}
2055 	do_userquota_cacheflush(os, &cache, tx);
2056 	multilist_sublist_unlock(list);
2057 	kmem_free(uua, sizeof (*uua));
2058 }
2059 
2060 /*
2061  * Release dnode holds from dmu_objset_sync_dnodes().  When the dnode is being
2062  * synced (i.e. we have issued the zio's for blocks in the dnode), it can't be
2063  * evicted because the block containing the dnode can't be evicted until it is
2064  * written out.  However, this hold is necessary to prevent the dnode_t from
2065  * being moved (via dnode_move()) while it's still referenced by
2066  * dbuf_dirty_record_t:dr_dnode.  And dr_dnode is needed for
2067  * dirty_lightweight_leaf-type dirty records.
2068  *
2069  * If we are doing user-object accounting, the dnode_rele() happens from
2070  * userquota_updates_task() instead.
2071  */
2072 static void
2073 dnode_rele_task(void *arg)
2074 {
2075 	userquota_updates_arg_t *uua = arg;
2076 	objset_t *os = uua->uua_os;
2077 
2078 	multilist_sublist_t *list =
2079 	    multilist_sublist_lock(&os->os_synced_dnodes, uua->uua_sublist_idx);
2080 
2081 	dnode_t *dn;
2082 	while ((dn = multilist_sublist_head(list)) != NULL) {
2083 		multilist_sublist_remove(list, dn);
2084 		dnode_rele(dn, &os->os_synced_dnodes);
2085 	}
2086 	multilist_sublist_unlock(list);
2087 	kmem_free(uua, sizeof (*uua));
2088 }
2089 
2090 /*
2091  * Return TRUE if userquota updates are needed.
2092  */
2093 static boolean_t
2094 dmu_objset_do_userquota_updates_prep(objset_t *os, dmu_tx_t *tx)
2095 {
2096 	if (!dmu_objset_userused_enabled(os))
2097 		return (B_FALSE);
2098 
2099 	/*
2100 	 * If this is a raw receive just return and handle accounting
2101 	 * later when we have the keys loaded. We also don't do user
2102 	 * accounting during claiming since the datasets are not owned
2103 	 * for the duration of claiming and this txg should only be
2104 	 * used for recovery.
2105 	 */
2106 	if (os->os_encrypted && dmu_objset_is_receiving(os))
2107 		return (B_FALSE);
2108 
2109 	if (tx->tx_txg <= os->os_spa->spa_claim_max_txg)
2110 		return (B_FALSE);
2111 
2112 	/* Allocate the user/group/project used objects if necessary. */
2113 	if (DMU_USERUSED_DNODE(os)->dn_type == DMU_OT_NONE) {
2114 		VERIFY0(zap_create_claim(os,
2115 		    DMU_USERUSED_OBJECT,
2116 		    DMU_OT_USERGROUP_USED, DMU_OT_NONE, 0, tx));
2117 		VERIFY0(zap_create_claim(os,
2118 		    DMU_GROUPUSED_OBJECT,
2119 		    DMU_OT_USERGROUP_USED, DMU_OT_NONE, 0, tx));
2120 	}
2121 
2122 	if (dmu_objset_projectquota_enabled(os) &&
2123 	    DMU_PROJECTUSED_DNODE(os)->dn_type == DMU_OT_NONE) {
2124 		VERIFY0(zap_create_claim(os, DMU_PROJECTUSED_OBJECT,
2125 		    DMU_OT_USERGROUP_USED, DMU_OT_NONE, 0, tx));
2126 	}
2127 	return (B_TRUE);
2128 }
2129 
2130 /*
2131  * Dispatch taskq tasks to dp_sync_taskq to update the user accounting, and
2132  * also release the holds on the dnodes from dmu_objset_sync_dnodes().
2133  * The caller must taskq_wait(dp_sync_taskq).
2134  */
2135 void
2136 dmu_objset_sync_done(objset_t *os, dmu_tx_t *tx)
2137 {
2138 	boolean_t need_userquota = dmu_objset_do_userquota_updates_prep(os, tx);
2139 
2140 	int num_sublists = multilist_get_num_sublists(&os->os_synced_dnodes);
2141 	for (int i = 0; i < num_sublists; i++) {
2142 		userquota_updates_arg_t *uua =
2143 		    kmem_alloc(sizeof (*uua), KM_SLEEP);
2144 		uua->uua_os = os;
2145 		uua->uua_sublist_idx = i;
2146 		uua->uua_tx = tx;
2147 
2148 		/*
2149 		 * If we don't need to update userquotas, use
2150 		 * dnode_rele_task() to call dnode_rele()
2151 		 */
2152 		(void) taskq_dispatch(dmu_objset_pool(os)->dp_sync_taskq,
2153 		    need_userquota ? userquota_updates_task : dnode_rele_task,
2154 		    uua, 0);
2155 		/* callback frees uua */
2156 	}
2157 }
2158 
2159 
2160 /*
2161  * Returns a pointer to data to find uid/gid from
2162  *
2163  * If a dirty record for transaction group that is syncing can't
2164  * be found then NULL is returned.  In the NULL case it is assumed
2165  * the uid/gid aren't changing.
2166  */
2167 static void *
2168 dmu_objset_userquota_find_data(dmu_buf_impl_t *db, dmu_tx_t *tx)
2169 {
2170 	dbuf_dirty_record_t *dr;
2171 	void *data;
2172 
2173 	if (db->db_dirtycnt == 0)
2174 		return (db->db.db_data);  /* Nothing is changing */
2175 
2176 	dr = dbuf_find_dirty_eq(db, tx->tx_txg);
2177 
2178 	if (dr == NULL) {
2179 		data = NULL;
2180 	} else {
2181 		if (dr->dr_dnode->dn_bonuslen == 0 &&
2182 		    dr->dr_dbuf->db_blkid == DMU_SPILL_BLKID)
2183 			data = dr->dt.dl.dr_data->b_data;
2184 		else
2185 			data = dr->dt.dl.dr_data;
2186 	}
2187 
2188 	return (data);
2189 }
2190 
2191 void
2192 dmu_objset_userquota_get_ids(dnode_t *dn, boolean_t before, dmu_tx_t *tx)
2193 {
2194 	objset_t *os = dn->dn_objset;
2195 	void *data = NULL;
2196 	dmu_buf_impl_t *db = NULL;
2197 	int flags = dn->dn_id_flags;
2198 	int error;
2199 	boolean_t have_spill = B_FALSE;
2200 
2201 	if (!dmu_objset_userused_enabled(dn->dn_objset))
2202 		return;
2203 
2204 	/*
2205 	 * Raw receives introduce a problem with user accounting. Raw
2206 	 * receives cannot update the user accounting info because the
2207 	 * user ids and the sizes are encrypted. To guarantee that we
2208 	 * never end up with bad user accounting, we simply disable it
2209 	 * during raw receives. We also disable this for normal receives
2210 	 * so that an incremental raw receive may be done on top of an
2211 	 * existing non-raw receive.
2212 	 */
2213 	if (os->os_encrypted && dmu_objset_is_receiving(os))
2214 		return;
2215 
2216 	if (before && (flags & (DN_ID_CHKED_BONUS|DN_ID_OLD_EXIST|
2217 	    DN_ID_CHKED_SPILL)))
2218 		return;
2219 
2220 	if (before && dn->dn_bonuslen != 0)
2221 		data = DN_BONUS(dn->dn_phys);
2222 	else if (!before && dn->dn_bonuslen != 0) {
2223 		if (dn->dn_bonus) {
2224 			db = dn->dn_bonus;
2225 			mutex_enter(&db->db_mtx);
2226 			data = dmu_objset_userquota_find_data(db, tx);
2227 		} else {
2228 			data = DN_BONUS(dn->dn_phys);
2229 		}
2230 	} else if (dn->dn_bonuslen == 0 && dn->dn_bonustype == DMU_OT_SA) {
2231 			int rf = 0;
2232 
2233 			if (RW_WRITE_HELD(&dn->dn_struct_rwlock))
2234 				rf |= DB_RF_HAVESTRUCT;
2235 			error = dmu_spill_hold_by_dnode(dn,
2236 			    rf | DB_RF_MUST_SUCCEED,
2237 			    FTAG, (dmu_buf_t **)&db);
2238 			ASSERT(error == 0);
2239 			mutex_enter(&db->db_mtx);
2240 			data = (before) ? db->db.db_data :
2241 			    dmu_objset_userquota_find_data(db, tx);
2242 			have_spill = B_TRUE;
2243 	} else {
2244 		mutex_enter(&dn->dn_mtx);
2245 		dn->dn_id_flags |= DN_ID_CHKED_BONUS;
2246 		mutex_exit(&dn->dn_mtx);
2247 		return;
2248 	}
2249 
2250 	/*
2251 	 * Must always call the callback in case the object
2252 	 * type has changed and that type isn't an object type to track
2253 	 */
2254 	zfs_file_info_t zfi;
2255 	error = file_cbs[os->os_phys->os_type](dn->dn_bonustype, data, &zfi);
2256 
2257 	if (before) {
2258 		ASSERT(data);
2259 		dn->dn_olduid = zfi.zfi_user;
2260 		dn->dn_oldgid = zfi.zfi_group;
2261 		dn->dn_oldprojid = zfi.zfi_project;
2262 	} else if (data) {
2263 		dn->dn_newuid = zfi.zfi_user;
2264 		dn->dn_newgid = zfi.zfi_group;
2265 		dn->dn_newprojid = zfi.zfi_project;
2266 	}
2267 
2268 	/*
2269 	 * Preserve existing uid/gid when the callback can't determine
2270 	 * what the new uid/gid are and the callback returned EEXIST.
2271 	 * The EEXIST error tells us to just use the existing uid/gid.
2272 	 * If we don't know what the old values are then just assign
2273 	 * them to 0, since that is a new file  being created.
2274 	 */
2275 	if (!before && data == NULL && error == EEXIST) {
2276 		if (flags & DN_ID_OLD_EXIST) {
2277 			dn->dn_newuid = dn->dn_olduid;
2278 			dn->dn_newgid = dn->dn_oldgid;
2279 			dn->dn_newprojid = dn->dn_oldprojid;
2280 		} else {
2281 			dn->dn_newuid = 0;
2282 			dn->dn_newgid = 0;
2283 			dn->dn_newprojid = ZFS_DEFAULT_PROJID;
2284 		}
2285 		error = 0;
2286 	}
2287 
2288 	if (db)
2289 		mutex_exit(&db->db_mtx);
2290 
2291 	mutex_enter(&dn->dn_mtx);
2292 	if (error == 0 && before)
2293 		dn->dn_id_flags |= DN_ID_OLD_EXIST;
2294 	if (error == 0 && !before)
2295 		dn->dn_id_flags |= DN_ID_NEW_EXIST;
2296 
2297 	if (have_spill) {
2298 		dn->dn_id_flags |= DN_ID_CHKED_SPILL;
2299 	} else {
2300 		dn->dn_id_flags |= DN_ID_CHKED_BONUS;
2301 	}
2302 	mutex_exit(&dn->dn_mtx);
2303 	if (have_spill)
2304 		dmu_buf_rele((dmu_buf_t *)db, FTAG);
2305 }
2306 
2307 boolean_t
2308 dmu_objset_userspace_present(objset_t *os)
2309 {
2310 	return (os->os_phys->os_flags &
2311 	    OBJSET_FLAG_USERACCOUNTING_COMPLETE);
2312 }
2313 
2314 boolean_t
2315 dmu_objset_userobjspace_present(objset_t *os)
2316 {
2317 	return (os->os_phys->os_flags &
2318 	    OBJSET_FLAG_USEROBJACCOUNTING_COMPLETE);
2319 }
2320 
2321 boolean_t
2322 dmu_objset_projectquota_present(objset_t *os)
2323 {
2324 	return (os->os_phys->os_flags &
2325 	    OBJSET_FLAG_PROJECTQUOTA_COMPLETE);
2326 }
2327 
2328 static int
2329 dmu_objset_space_upgrade(objset_t *os)
2330 {
2331 	uint64_t obj;
2332 	int err = 0;
2333 
2334 	/*
2335 	 * We simply need to mark every object dirty, so that it will be
2336 	 * synced out and now accounted.  If this is called
2337 	 * concurrently, or if we already did some work before crashing,
2338 	 * that's fine, since we track each object's accounted state
2339 	 * independently.
2340 	 */
2341 
2342 	for (obj = 0; err == 0; err = dmu_object_next(os, &obj, FALSE, 0)) {
2343 		dmu_tx_t *tx;
2344 		dmu_buf_t *db;
2345 		int objerr;
2346 
2347 		mutex_enter(&os->os_upgrade_lock);
2348 		if (os->os_upgrade_exit)
2349 			err = SET_ERROR(EINTR);
2350 		mutex_exit(&os->os_upgrade_lock);
2351 		if (err != 0)
2352 			return (err);
2353 
2354 		if (issig(JUSTLOOKING) && issig(FORREAL))
2355 			return (SET_ERROR(EINTR));
2356 
2357 		objerr = dmu_bonus_hold(os, obj, FTAG, &db);
2358 		if (objerr != 0)
2359 			continue;
2360 		tx = dmu_tx_create(os);
2361 		dmu_tx_hold_bonus(tx, obj);
2362 		objerr = dmu_tx_assign(tx, TXG_WAIT);
2363 		if (objerr != 0) {
2364 			dmu_buf_rele(db, FTAG);
2365 			dmu_tx_abort(tx);
2366 			continue;
2367 		}
2368 		dmu_buf_will_dirty(db, tx);
2369 		dmu_buf_rele(db, FTAG);
2370 		dmu_tx_commit(tx);
2371 	}
2372 	return (0);
2373 }
2374 
2375 static int
2376 dmu_objset_userspace_upgrade_cb(objset_t *os)
2377 {
2378 	int err = 0;
2379 
2380 	if (dmu_objset_userspace_present(os))
2381 		return (0);
2382 	if (dmu_objset_is_snapshot(os))
2383 		return (SET_ERROR(EINVAL));
2384 	if (!dmu_objset_userused_enabled(os))
2385 		return (SET_ERROR(ENOTSUP));
2386 
2387 	err = dmu_objset_space_upgrade(os);
2388 	if (err)
2389 		return (err);
2390 
2391 	os->os_flags |= OBJSET_FLAG_USERACCOUNTING_COMPLETE;
2392 	txg_wait_synced(dmu_objset_pool(os), 0);
2393 	return (0);
2394 }
2395 
2396 void
2397 dmu_objset_userspace_upgrade(objset_t *os)
2398 {
2399 	dmu_objset_upgrade(os, dmu_objset_userspace_upgrade_cb);
2400 }
2401 
2402 static int
2403 dmu_objset_id_quota_upgrade_cb(objset_t *os)
2404 {
2405 	int err = 0;
2406 
2407 	if (dmu_objset_userobjspace_present(os) &&
2408 	    dmu_objset_projectquota_present(os))
2409 		return (0);
2410 	if (dmu_objset_is_snapshot(os))
2411 		return (SET_ERROR(EINVAL));
2412 	if (!dmu_objset_userused_enabled(os))
2413 		return (SET_ERROR(ENOTSUP));
2414 	if (!dmu_objset_projectquota_enabled(os) &&
2415 	    dmu_objset_userobjspace_present(os))
2416 		return (SET_ERROR(ENOTSUP));
2417 
2418 	if (dmu_objset_userobjused_enabled(os))
2419 		dmu_objset_ds(os)->ds_feature_activation[
2420 		    SPA_FEATURE_USEROBJ_ACCOUNTING] = (void *)B_TRUE;
2421 	if (dmu_objset_projectquota_enabled(os))
2422 		dmu_objset_ds(os)->ds_feature_activation[
2423 		    SPA_FEATURE_PROJECT_QUOTA] = (void *)B_TRUE;
2424 
2425 	err = dmu_objset_space_upgrade(os);
2426 	if (err)
2427 		return (err);
2428 
2429 	os->os_flags |= OBJSET_FLAG_USERACCOUNTING_COMPLETE;
2430 	if (dmu_objset_userobjused_enabled(os))
2431 		os->os_flags |= OBJSET_FLAG_USEROBJACCOUNTING_COMPLETE;
2432 	if (dmu_objset_projectquota_enabled(os))
2433 		os->os_flags |= OBJSET_FLAG_PROJECTQUOTA_COMPLETE;
2434 
2435 	txg_wait_synced(dmu_objset_pool(os), 0);
2436 	return (0);
2437 }
2438 
2439 void
2440 dmu_objset_id_quota_upgrade(objset_t *os)
2441 {
2442 	dmu_objset_upgrade(os, dmu_objset_id_quota_upgrade_cb);
2443 }
2444 
2445 boolean_t
2446 dmu_objset_userobjspace_upgradable(objset_t *os)
2447 {
2448 	return (dmu_objset_type(os) == DMU_OST_ZFS &&
2449 	    !dmu_objset_is_snapshot(os) &&
2450 	    dmu_objset_userobjused_enabled(os) &&
2451 	    !dmu_objset_userobjspace_present(os) &&
2452 	    spa_writeable(dmu_objset_spa(os)));
2453 }
2454 
2455 boolean_t
2456 dmu_objset_projectquota_upgradable(objset_t *os)
2457 {
2458 	return (dmu_objset_type(os) == DMU_OST_ZFS &&
2459 	    !dmu_objset_is_snapshot(os) &&
2460 	    dmu_objset_projectquota_enabled(os) &&
2461 	    !dmu_objset_projectquota_present(os) &&
2462 	    spa_writeable(dmu_objset_spa(os)));
2463 }
2464 
2465 void
2466 dmu_objset_space(objset_t *os, uint64_t *refdbytesp, uint64_t *availbytesp,
2467     uint64_t *usedobjsp, uint64_t *availobjsp)
2468 {
2469 	dsl_dataset_space(os->os_dsl_dataset, refdbytesp, availbytesp,
2470 	    usedobjsp, availobjsp);
2471 }
2472 
2473 uint64_t
2474 dmu_objset_fsid_guid(objset_t *os)
2475 {
2476 	return (dsl_dataset_fsid_guid(os->os_dsl_dataset));
2477 }
2478 
2479 void
2480 dmu_objset_fast_stat(objset_t *os, dmu_objset_stats_t *stat)
2481 {
2482 	stat->dds_type = os->os_phys->os_type;
2483 	if (os->os_dsl_dataset)
2484 		dsl_dataset_fast_stat(os->os_dsl_dataset, stat);
2485 }
2486 
2487 void
2488 dmu_objset_stats(objset_t *os, nvlist_t *nv)
2489 {
2490 	ASSERT(os->os_dsl_dataset ||
2491 	    os->os_phys->os_type == DMU_OST_META);
2492 
2493 	if (os->os_dsl_dataset != NULL)
2494 		dsl_dataset_stats(os->os_dsl_dataset, nv);
2495 
2496 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_TYPE,
2497 	    os->os_phys->os_type);
2498 	dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USERACCOUNTING,
2499 	    dmu_objset_userspace_present(os));
2500 }
2501 
2502 int
2503 dmu_objset_is_snapshot(objset_t *os)
2504 {
2505 	if (os->os_dsl_dataset != NULL)
2506 		return (os->os_dsl_dataset->ds_is_snapshot);
2507 	else
2508 		return (B_FALSE);
2509 }
2510 
2511 int
2512 dmu_snapshot_realname(objset_t *os, const char *name, char *real, int maxlen,
2513     boolean_t *conflict)
2514 {
2515 	dsl_dataset_t *ds = os->os_dsl_dataset;
2516 	uint64_t ignored;
2517 
2518 	if (dsl_dataset_phys(ds)->ds_snapnames_zapobj == 0)
2519 		return (SET_ERROR(ENOENT));
2520 
2521 	return (zap_lookup_norm(ds->ds_dir->dd_pool->dp_meta_objset,
2522 	    dsl_dataset_phys(ds)->ds_snapnames_zapobj, name, 8, 1, &ignored,
2523 	    MT_NORMALIZE, real, maxlen, conflict));
2524 }
2525 
2526 int
2527 dmu_snapshot_list_next(objset_t *os, int namelen, char *name,
2528     uint64_t *idp, uint64_t *offp, boolean_t *case_conflict)
2529 {
2530 	dsl_dataset_t *ds = os->os_dsl_dataset;
2531 	zap_cursor_t cursor;
2532 	zap_attribute_t attr;
2533 
2534 	ASSERT(dsl_pool_config_held(dmu_objset_pool(os)));
2535 
2536 	if (dsl_dataset_phys(ds)->ds_snapnames_zapobj == 0)
2537 		return (SET_ERROR(ENOENT));
2538 
2539 	zap_cursor_init_serialized(&cursor,
2540 	    ds->ds_dir->dd_pool->dp_meta_objset,
2541 	    dsl_dataset_phys(ds)->ds_snapnames_zapobj, *offp);
2542 
2543 	if (zap_cursor_retrieve(&cursor, &attr) != 0) {
2544 		zap_cursor_fini(&cursor);
2545 		return (SET_ERROR(ENOENT));
2546 	}
2547 
2548 	if (strlen(attr.za_name) + 1 > namelen) {
2549 		zap_cursor_fini(&cursor);
2550 		return (SET_ERROR(ENAMETOOLONG));
2551 	}
2552 
2553 	(void) strlcpy(name, attr.za_name, namelen);
2554 	if (idp)
2555 		*idp = attr.za_first_integer;
2556 	if (case_conflict)
2557 		*case_conflict = attr.za_normalization_conflict;
2558 	zap_cursor_advance(&cursor);
2559 	*offp = zap_cursor_serialize(&cursor);
2560 	zap_cursor_fini(&cursor);
2561 
2562 	return (0);
2563 }
2564 
2565 int
2566 dmu_snapshot_lookup(objset_t *os, const char *name, uint64_t *value)
2567 {
2568 	return (dsl_dataset_snap_lookup(os->os_dsl_dataset, name, value));
2569 }
2570 
2571 int
2572 dmu_dir_list_next(objset_t *os, int namelen, char *name,
2573     uint64_t *idp, uint64_t *offp)
2574 {
2575 	dsl_dir_t *dd = os->os_dsl_dataset->ds_dir;
2576 	zap_cursor_t cursor;
2577 	zap_attribute_t attr;
2578 
2579 	/* there is no next dir on a snapshot! */
2580 	if (os->os_dsl_dataset->ds_object !=
2581 	    dsl_dir_phys(dd)->dd_head_dataset_obj)
2582 		return (SET_ERROR(ENOENT));
2583 
2584 	zap_cursor_init_serialized(&cursor,
2585 	    dd->dd_pool->dp_meta_objset,
2586 	    dsl_dir_phys(dd)->dd_child_dir_zapobj, *offp);
2587 
2588 	if (zap_cursor_retrieve(&cursor, &attr) != 0) {
2589 		zap_cursor_fini(&cursor);
2590 		return (SET_ERROR(ENOENT));
2591 	}
2592 
2593 	if (strlen(attr.za_name) + 1 > namelen) {
2594 		zap_cursor_fini(&cursor);
2595 		return (SET_ERROR(ENAMETOOLONG));
2596 	}
2597 
2598 	(void) strlcpy(name, attr.za_name, namelen);
2599 	if (idp)
2600 		*idp = attr.za_first_integer;
2601 	zap_cursor_advance(&cursor);
2602 	*offp = zap_cursor_serialize(&cursor);
2603 	zap_cursor_fini(&cursor);
2604 
2605 	return (0);
2606 }
2607 
2608 typedef struct dmu_objset_find_ctx {
2609 	taskq_t		*dc_tq;
2610 	dsl_pool_t	*dc_dp;
2611 	uint64_t	dc_ddobj;
2612 	char		*dc_ddname; /* last component of ddobj's name */
2613 	int		(*dc_func)(dsl_pool_t *, dsl_dataset_t *, void *);
2614 	void		*dc_arg;
2615 	int		dc_flags;
2616 	kmutex_t	*dc_error_lock;
2617 	int		*dc_error;
2618 } dmu_objset_find_ctx_t;
2619 
2620 static void
2621 dmu_objset_find_dp_impl(dmu_objset_find_ctx_t *dcp)
2622 {
2623 	dsl_pool_t *dp = dcp->dc_dp;
2624 	dsl_dir_t *dd;
2625 	dsl_dataset_t *ds;
2626 	zap_cursor_t zc;
2627 	zap_attribute_t *attr;
2628 	uint64_t thisobj;
2629 	int err = 0;
2630 
2631 	/* don't process if there already was an error */
2632 	if (*dcp->dc_error != 0)
2633 		goto out;
2634 
2635 	/*
2636 	 * Note: passing the name (dc_ddname) here is optional, but it
2637 	 * improves performance because we don't need to call
2638 	 * zap_value_search() to determine the name.
2639 	 */
2640 	err = dsl_dir_hold_obj(dp, dcp->dc_ddobj, dcp->dc_ddname, FTAG, &dd);
2641 	if (err != 0)
2642 		goto out;
2643 
2644 	/* Don't visit hidden ($MOS & $ORIGIN) objsets. */
2645 	if (dd->dd_myname[0] == '$') {
2646 		dsl_dir_rele(dd, FTAG);
2647 		goto out;
2648 	}
2649 
2650 	thisobj = dsl_dir_phys(dd)->dd_head_dataset_obj;
2651 	attr = kmem_alloc(sizeof (zap_attribute_t), KM_SLEEP);
2652 
2653 	/*
2654 	 * Iterate over all children.
2655 	 */
2656 	if (dcp->dc_flags & DS_FIND_CHILDREN) {
2657 		for (zap_cursor_init(&zc, dp->dp_meta_objset,
2658 		    dsl_dir_phys(dd)->dd_child_dir_zapobj);
2659 		    zap_cursor_retrieve(&zc, attr) == 0;
2660 		    (void) zap_cursor_advance(&zc)) {
2661 			ASSERT3U(attr->za_integer_length, ==,
2662 			    sizeof (uint64_t));
2663 			ASSERT3U(attr->za_num_integers, ==, 1);
2664 
2665 			dmu_objset_find_ctx_t *child_dcp =
2666 			    kmem_alloc(sizeof (*child_dcp), KM_SLEEP);
2667 			*child_dcp = *dcp;
2668 			child_dcp->dc_ddobj = attr->za_first_integer;
2669 			child_dcp->dc_ddname = spa_strdup(attr->za_name);
2670 			if (dcp->dc_tq != NULL)
2671 				(void) taskq_dispatch(dcp->dc_tq,
2672 				    dmu_objset_find_dp_cb, child_dcp, TQ_SLEEP);
2673 			else
2674 				dmu_objset_find_dp_impl(child_dcp);
2675 		}
2676 		zap_cursor_fini(&zc);
2677 	}
2678 
2679 	/*
2680 	 * Iterate over all snapshots.
2681 	 */
2682 	if (dcp->dc_flags & DS_FIND_SNAPSHOTS) {
2683 		dsl_dataset_t *ds;
2684 		err = dsl_dataset_hold_obj(dp, thisobj, FTAG, &ds);
2685 
2686 		if (err == 0) {
2687 			uint64_t snapobj;
2688 
2689 			snapobj = dsl_dataset_phys(ds)->ds_snapnames_zapobj;
2690 			dsl_dataset_rele(ds, FTAG);
2691 
2692 			for (zap_cursor_init(&zc, dp->dp_meta_objset, snapobj);
2693 			    zap_cursor_retrieve(&zc, attr) == 0;
2694 			    (void) zap_cursor_advance(&zc)) {
2695 				ASSERT3U(attr->za_integer_length, ==,
2696 				    sizeof (uint64_t));
2697 				ASSERT3U(attr->za_num_integers, ==, 1);
2698 
2699 				err = dsl_dataset_hold_obj(dp,
2700 				    attr->za_first_integer, FTAG, &ds);
2701 				if (err != 0)
2702 					break;
2703 				err = dcp->dc_func(dp, ds, dcp->dc_arg);
2704 				dsl_dataset_rele(ds, FTAG);
2705 				if (err != 0)
2706 					break;
2707 			}
2708 			zap_cursor_fini(&zc);
2709 		}
2710 	}
2711 
2712 	kmem_free(attr, sizeof (zap_attribute_t));
2713 
2714 	if (err != 0) {
2715 		dsl_dir_rele(dd, FTAG);
2716 		goto out;
2717 	}
2718 
2719 	/*
2720 	 * Apply to self.
2721 	 */
2722 	err = dsl_dataset_hold_obj(dp, thisobj, FTAG, &ds);
2723 
2724 	/*
2725 	 * Note: we hold the dir while calling dsl_dataset_hold_obj() so
2726 	 * that the dir will remain cached, and we won't have to re-instantiate
2727 	 * it (which could be expensive due to finding its name via
2728 	 * zap_value_search()).
2729 	 */
2730 	dsl_dir_rele(dd, FTAG);
2731 	if (err != 0)
2732 		goto out;
2733 	err = dcp->dc_func(dp, ds, dcp->dc_arg);
2734 	dsl_dataset_rele(ds, FTAG);
2735 
2736 out:
2737 	if (err != 0) {
2738 		mutex_enter(dcp->dc_error_lock);
2739 		/* only keep first error */
2740 		if (*dcp->dc_error == 0)
2741 			*dcp->dc_error = err;
2742 		mutex_exit(dcp->dc_error_lock);
2743 	}
2744 
2745 	if (dcp->dc_ddname != NULL)
2746 		spa_strfree(dcp->dc_ddname);
2747 	kmem_free(dcp, sizeof (*dcp));
2748 }
2749 
2750 static void
2751 dmu_objset_find_dp_cb(void *arg)
2752 {
2753 	dmu_objset_find_ctx_t *dcp = arg;
2754 	dsl_pool_t *dp = dcp->dc_dp;
2755 
2756 	/*
2757 	 * We need to get a pool_config_lock here, as there are several
2758 	 * assert(pool_config_held) down the stack. Getting a lock via
2759 	 * dsl_pool_config_enter is risky, as it might be stalled by a
2760 	 * pending writer. This would deadlock, as the write lock can
2761 	 * only be granted when our parent thread gives up the lock.
2762 	 * The _prio interface gives us priority over a pending writer.
2763 	 */
2764 	dsl_pool_config_enter_prio(dp, FTAG);
2765 
2766 	dmu_objset_find_dp_impl(dcp);
2767 
2768 	dsl_pool_config_exit(dp, FTAG);
2769 }
2770 
2771 /*
2772  * Find objsets under and including ddobj, call func(ds) on each.
2773  * The order for the enumeration is completely undefined.
2774  * func is called with dsl_pool_config held.
2775  */
2776 int
2777 dmu_objset_find_dp(dsl_pool_t *dp, uint64_t ddobj,
2778     int func(dsl_pool_t *, dsl_dataset_t *, void *), void *arg, int flags)
2779 {
2780 	int error = 0;
2781 	taskq_t *tq = NULL;
2782 	int ntasks;
2783 	dmu_objset_find_ctx_t *dcp;
2784 	kmutex_t err_lock;
2785 
2786 	mutex_init(&err_lock, NULL, MUTEX_DEFAULT, NULL);
2787 	dcp = kmem_alloc(sizeof (*dcp), KM_SLEEP);
2788 	dcp->dc_tq = NULL;
2789 	dcp->dc_dp = dp;
2790 	dcp->dc_ddobj = ddobj;
2791 	dcp->dc_ddname = NULL;
2792 	dcp->dc_func = func;
2793 	dcp->dc_arg = arg;
2794 	dcp->dc_flags = flags;
2795 	dcp->dc_error_lock = &err_lock;
2796 	dcp->dc_error = &error;
2797 
2798 	if ((flags & DS_FIND_SERIALIZE) || dsl_pool_config_held_writer(dp)) {
2799 		/*
2800 		 * In case a write lock is held we can't make use of
2801 		 * parallelism, as down the stack of the worker threads
2802 		 * the lock is asserted via dsl_pool_config_held.
2803 		 * In case of a read lock this is solved by getting a read
2804 		 * lock in each worker thread, which isn't possible in case
2805 		 * of a writer lock. So we fall back to the synchronous path
2806 		 * here.
2807 		 * In the future it might be possible to get some magic into
2808 		 * dsl_pool_config_held in a way that it returns true for
2809 		 * the worker threads so that a single lock held from this
2810 		 * thread suffices. For now, stay single threaded.
2811 		 */
2812 		dmu_objset_find_dp_impl(dcp);
2813 		mutex_destroy(&err_lock);
2814 
2815 		return (error);
2816 	}
2817 
2818 	ntasks = dmu_find_threads;
2819 	if (ntasks == 0)
2820 		ntasks = vdev_count_leaves(dp->dp_spa) * 4;
2821 	tq = taskq_create("dmu_objset_find", ntasks, maxclsyspri, ntasks,
2822 	    INT_MAX, 0);
2823 	if (tq == NULL) {
2824 		kmem_free(dcp, sizeof (*dcp));
2825 		mutex_destroy(&err_lock);
2826 
2827 		return (SET_ERROR(ENOMEM));
2828 	}
2829 	dcp->dc_tq = tq;
2830 
2831 	/* dcp will be freed by task */
2832 	(void) taskq_dispatch(tq, dmu_objset_find_dp_cb, dcp, TQ_SLEEP);
2833 
2834 	/*
2835 	 * PORTING: this code relies on the property of taskq_wait to wait
2836 	 * until no more tasks are queued and no more tasks are active. As
2837 	 * we always queue new tasks from within other tasks, task_wait
2838 	 * reliably waits for the full recursion to finish, even though we
2839 	 * enqueue new tasks after taskq_wait has been called.
2840 	 * On platforms other than illumos, taskq_wait may not have this
2841 	 * property.
2842 	 */
2843 	taskq_wait(tq);
2844 	taskq_destroy(tq);
2845 	mutex_destroy(&err_lock);
2846 
2847 	return (error);
2848 }
2849 
2850 /*
2851  * Find all objsets under name, and for each, call 'func(child_name, arg)'.
2852  * The dp_config_rwlock must not be held when this is called, and it
2853  * will not be held when the callback is called.
2854  * Therefore this function should only be used when the pool is not changing
2855  * (e.g. in syncing context), or the callback can deal with the possible races.
2856  */
2857 static int
2858 dmu_objset_find_impl(spa_t *spa, const char *name,
2859     int func(const char *, void *), void *arg, int flags)
2860 {
2861 	dsl_dir_t *dd;
2862 	dsl_pool_t *dp = spa_get_dsl(spa);
2863 	dsl_dataset_t *ds;
2864 	zap_cursor_t zc;
2865 	zap_attribute_t *attr;
2866 	char *child;
2867 	uint64_t thisobj;
2868 	int err;
2869 
2870 	dsl_pool_config_enter(dp, FTAG);
2871 
2872 	err = dsl_dir_hold(dp, name, FTAG, &dd, NULL);
2873 	if (err != 0) {
2874 		dsl_pool_config_exit(dp, FTAG);
2875 		return (err);
2876 	}
2877 
2878 	/* Don't visit hidden ($MOS & $ORIGIN) objsets. */
2879 	if (dd->dd_myname[0] == '$') {
2880 		dsl_dir_rele(dd, FTAG);
2881 		dsl_pool_config_exit(dp, FTAG);
2882 		return (0);
2883 	}
2884 
2885 	thisobj = dsl_dir_phys(dd)->dd_head_dataset_obj;
2886 	attr = kmem_alloc(sizeof (zap_attribute_t), KM_SLEEP);
2887 
2888 	/*
2889 	 * Iterate over all children.
2890 	 */
2891 	if (flags & DS_FIND_CHILDREN) {
2892 		for (zap_cursor_init(&zc, dp->dp_meta_objset,
2893 		    dsl_dir_phys(dd)->dd_child_dir_zapobj);
2894 		    zap_cursor_retrieve(&zc, attr) == 0;
2895 		    (void) zap_cursor_advance(&zc)) {
2896 			ASSERT3U(attr->za_integer_length, ==,
2897 			    sizeof (uint64_t));
2898 			ASSERT3U(attr->za_num_integers, ==, 1);
2899 
2900 			child = kmem_asprintf("%s/%s", name, attr->za_name);
2901 			dsl_pool_config_exit(dp, FTAG);
2902 			err = dmu_objset_find_impl(spa, child,
2903 			    func, arg, flags);
2904 			dsl_pool_config_enter(dp, FTAG);
2905 			kmem_strfree(child);
2906 			if (err != 0)
2907 				break;
2908 		}
2909 		zap_cursor_fini(&zc);
2910 
2911 		if (err != 0) {
2912 			dsl_dir_rele(dd, FTAG);
2913 			dsl_pool_config_exit(dp, FTAG);
2914 			kmem_free(attr, sizeof (zap_attribute_t));
2915 			return (err);
2916 		}
2917 	}
2918 
2919 	/*
2920 	 * Iterate over all snapshots.
2921 	 */
2922 	if (flags & DS_FIND_SNAPSHOTS) {
2923 		err = dsl_dataset_hold_obj(dp, thisobj, FTAG, &ds);
2924 
2925 		if (err == 0) {
2926 			uint64_t snapobj;
2927 
2928 			snapobj = dsl_dataset_phys(ds)->ds_snapnames_zapobj;
2929 			dsl_dataset_rele(ds, FTAG);
2930 
2931 			for (zap_cursor_init(&zc, dp->dp_meta_objset, snapobj);
2932 			    zap_cursor_retrieve(&zc, attr) == 0;
2933 			    (void) zap_cursor_advance(&zc)) {
2934 				ASSERT3U(attr->za_integer_length, ==,
2935 				    sizeof (uint64_t));
2936 				ASSERT3U(attr->za_num_integers, ==, 1);
2937 
2938 				child = kmem_asprintf("%s@%s",
2939 				    name, attr->za_name);
2940 				dsl_pool_config_exit(dp, FTAG);
2941 				err = func(child, arg);
2942 				dsl_pool_config_enter(dp, FTAG);
2943 				kmem_strfree(child);
2944 				if (err != 0)
2945 					break;
2946 			}
2947 			zap_cursor_fini(&zc);
2948 		}
2949 	}
2950 
2951 	dsl_dir_rele(dd, FTAG);
2952 	kmem_free(attr, sizeof (zap_attribute_t));
2953 	dsl_pool_config_exit(dp, FTAG);
2954 
2955 	if (err != 0)
2956 		return (err);
2957 
2958 	/* Apply to self. */
2959 	return (func(name, arg));
2960 }
2961 
2962 /*
2963  * See comment above dmu_objset_find_impl().
2964  */
2965 int
2966 dmu_objset_find(const char *name, int func(const char *, void *), void *arg,
2967     int flags)
2968 {
2969 	spa_t *spa;
2970 	int error;
2971 
2972 	error = spa_open(name, &spa, FTAG);
2973 	if (error != 0)
2974 		return (error);
2975 	error = dmu_objset_find_impl(spa, name, func, arg, flags);
2976 	spa_close(spa, FTAG);
2977 	return (error);
2978 }
2979 
2980 boolean_t
2981 dmu_objset_incompatible_encryption_version(objset_t *os)
2982 {
2983 	return (dsl_dir_incompatible_encryption_version(
2984 	    os->os_dsl_dataset->ds_dir));
2985 }
2986 
2987 void
2988 dmu_objset_set_user(objset_t *os, void *user_ptr)
2989 {
2990 	ASSERT(MUTEX_HELD(&os->os_user_ptr_lock));
2991 	os->os_user_ptr = user_ptr;
2992 }
2993 
2994 void *
2995 dmu_objset_get_user(objset_t *os)
2996 {
2997 	ASSERT(MUTEX_HELD(&os->os_user_ptr_lock));
2998 	return (os->os_user_ptr);
2999 }
3000 
3001 /*
3002  * Determine name of filesystem, given name of snapshot.
3003  * buf must be at least ZFS_MAX_DATASET_NAME_LEN bytes
3004  */
3005 int
3006 dmu_fsname(const char *snapname, char *buf)
3007 {
3008 	char *atp = strchr(snapname, '@');
3009 	if (atp == NULL)
3010 		return (SET_ERROR(EINVAL));
3011 	if (atp - snapname >= ZFS_MAX_DATASET_NAME_LEN)
3012 		return (SET_ERROR(ENAMETOOLONG));
3013 	(void) strlcpy(buf, snapname, atp - snapname + 1);
3014 	return (0);
3015 }
3016 
3017 /*
3018  * Call when we think we're going to write/free space in open context
3019  * to track the amount of dirty data in the open txg, which is also the
3020  * amount of memory that can not be evicted until this txg syncs.
3021  *
3022  * Note that there are two conditions where this can be called from
3023  * syncing context:
3024  *
3025  * [1] When we just created the dataset, in which case we go on with
3026  *     updating any accounting of dirty data as usual.
3027  * [2] When we are dirtying MOS data, in which case we only update the
3028  *     pool's accounting of dirty data.
3029  */
3030 void
3031 dmu_objset_willuse_space(objset_t *os, int64_t space, dmu_tx_t *tx)
3032 {
3033 	dsl_dataset_t *ds = os->os_dsl_dataset;
3034 	int64_t aspace = spa_get_worst_case_asize(os->os_spa, space);
3035 
3036 	if (ds != NULL) {
3037 		dsl_dir_willuse_space(ds->ds_dir, aspace, tx);
3038 	}
3039 
3040 	dsl_pool_dirty_space(dmu_tx_pool(tx), space, tx);
3041 }
3042 
3043 #if defined(_KERNEL)
3044 EXPORT_SYMBOL(dmu_objset_zil);
3045 EXPORT_SYMBOL(dmu_objset_pool);
3046 EXPORT_SYMBOL(dmu_objset_ds);
3047 EXPORT_SYMBOL(dmu_objset_type);
3048 EXPORT_SYMBOL(dmu_objset_name);
3049 EXPORT_SYMBOL(dmu_objset_hold);
3050 EXPORT_SYMBOL(dmu_objset_hold_flags);
3051 EXPORT_SYMBOL(dmu_objset_own);
3052 EXPORT_SYMBOL(dmu_objset_rele);
3053 EXPORT_SYMBOL(dmu_objset_rele_flags);
3054 EXPORT_SYMBOL(dmu_objset_disown);
3055 EXPORT_SYMBOL(dmu_objset_from_ds);
3056 EXPORT_SYMBOL(dmu_objset_create);
3057 EXPORT_SYMBOL(dmu_objset_clone);
3058 EXPORT_SYMBOL(dmu_objset_stats);
3059 EXPORT_SYMBOL(dmu_objset_fast_stat);
3060 EXPORT_SYMBOL(dmu_objset_spa);
3061 EXPORT_SYMBOL(dmu_objset_space);
3062 EXPORT_SYMBOL(dmu_objset_fsid_guid);
3063 EXPORT_SYMBOL(dmu_objset_find);
3064 EXPORT_SYMBOL(dmu_objset_byteswap);
3065 EXPORT_SYMBOL(dmu_objset_evict_dbufs);
3066 EXPORT_SYMBOL(dmu_objset_snap_cmtime);
3067 EXPORT_SYMBOL(dmu_objset_dnodesize);
3068 
3069 EXPORT_SYMBOL(dmu_objset_sync);
3070 EXPORT_SYMBOL(dmu_objset_is_dirty);
3071 EXPORT_SYMBOL(dmu_objset_create_impl_dnstats);
3072 EXPORT_SYMBOL(dmu_objset_create_impl);
3073 EXPORT_SYMBOL(dmu_objset_open_impl);
3074 EXPORT_SYMBOL(dmu_objset_evict);
3075 EXPORT_SYMBOL(dmu_objset_register_type);
3076 EXPORT_SYMBOL(dmu_objset_sync_done);
3077 EXPORT_SYMBOL(dmu_objset_userquota_get_ids);
3078 EXPORT_SYMBOL(dmu_objset_userused_enabled);
3079 EXPORT_SYMBOL(dmu_objset_userspace_upgrade);
3080 EXPORT_SYMBOL(dmu_objset_userspace_present);
3081 EXPORT_SYMBOL(dmu_objset_userobjused_enabled);
3082 EXPORT_SYMBOL(dmu_objset_userobjspace_upgradable);
3083 EXPORT_SYMBOL(dmu_objset_userobjspace_present);
3084 EXPORT_SYMBOL(dmu_objset_projectquota_enabled);
3085 EXPORT_SYMBOL(dmu_objset_projectquota_present);
3086 EXPORT_SYMBOL(dmu_objset_projectquota_upgradable);
3087 EXPORT_SYMBOL(dmu_objset_id_quota_upgrade);
3088 #endif
3089