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