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