xref: /illumos-gate/usr/src/uts/common/fs/zfs/dsl_pool.c (revision 86714001)
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 (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
23  * Copyright (c) 2011, 2017 by Delphix. All rights reserved.
24  * Copyright (c) 2013 Steven Hartland. All rights reserved.
25  * Copyright (c) 2014 Spectra Logic Corporation, All rights reserved.
26  * Copyright (c) 2014 Integros [integros.com]
27  * Copyright 2016 Nexenta Systems, Inc.  All rights reserved.
28  */
29 
30 #include <sys/dsl_pool.h>
31 #include <sys/dsl_dataset.h>
32 #include <sys/dsl_prop.h>
33 #include <sys/dsl_dir.h>
34 #include <sys/dsl_synctask.h>
35 #include <sys/dsl_scan.h>
36 #include <sys/dnode.h>
37 #include <sys/dmu_tx.h>
38 #include <sys/dmu_objset.h>
39 #include <sys/arc.h>
40 #include <sys/zap.h>
41 #include <sys/zio.h>
42 #include <sys/zfs_context.h>
43 #include <sys/fs/zfs.h>
44 #include <sys/zfs_znode.h>
45 #include <sys/spa_impl.h>
46 #include <sys/dsl_deadlist.h>
47 #include <sys/vdev_impl.h>
48 #include <sys/metaslab_impl.h>
49 #include <sys/bptree.h>
50 #include <sys/zfeature.h>
51 #include <sys/zil_impl.h>
52 #include <sys/dsl_userhold.h>
53 
54 /*
55  * ZFS Write Throttle
56  * ------------------
57  *
58  * ZFS must limit the rate of incoming writes to the rate at which it is able
59  * to sync data modifications to the backend storage. Throttling by too much
60  * creates an artificial limit; throttling by too little can only be sustained
61  * for short periods and would lead to highly lumpy performance. On a per-pool
62  * basis, ZFS tracks the amount of modified (dirty) data. As operations change
63  * data, the amount of dirty data increases; as ZFS syncs out data, the amount
64  * of dirty data decreases. When the amount of dirty data exceeds a
65  * predetermined threshold further modifications are blocked until the amount
66  * of dirty data decreases (as data is synced out).
67  *
68  * The limit on dirty data is tunable, and should be adjusted according to
69  * both the IO capacity and available memory of the system. The larger the
70  * window, the more ZFS is able to aggregate and amortize metadata (and data)
71  * changes. However, memory is a limited resource, and allowing for more dirty
72  * data comes at the cost of keeping other useful data in memory (for example
73  * ZFS data cached by the ARC).
74  *
75  * Implementation
76  *
77  * As buffers are modified dsl_pool_willuse_space() increments both the per-
78  * txg (dp_dirty_pertxg[]) and poolwide (dp_dirty_total) accounting of
79  * dirty space used; dsl_pool_dirty_space() decrements those values as data
80  * is synced out from dsl_pool_sync(). While only the poolwide value is
81  * relevant, the per-txg value is useful for debugging. The tunable
82  * zfs_dirty_data_max determines the dirty space limit. Once that value is
83  * exceeded, new writes are halted until space frees up.
84  *
85  * The zfs_dirty_data_sync tunable dictates the threshold at which we
86  * ensure that there is a txg syncing (see the comment in txg.c for a full
87  * description of transaction group stages).
88  *
89  * The IO scheduler uses both the dirty space limit and current amount of
90  * dirty data as inputs. Those values affect the number of concurrent IOs ZFS
91  * issues. See the comment in vdev_queue.c for details of the IO scheduler.
92  *
93  * The delay is also calculated based on the amount of dirty data.  See the
94  * comment above dmu_tx_delay() for details.
95  */
96 
97 /*
98  * zfs_dirty_data_max will be set to zfs_dirty_data_max_percent% of all memory,
99  * capped at zfs_dirty_data_max_max.  It can also be overridden in /etc/system.
100  */
101 uint64_t zfs_dirty_data_max;
102 uint64_t zfs_dirty_data_max_max = 4ULL * 1024 * 1024 * 1024;
103 int zfs_dirty_data_max_percent = 10;
104 
105 /*
106  * If there is at least this much dirty data, push out a txg.
107  */
108 uint64_t zfs_dirty_data_sync = 64 * 1024 * 1024;
109 
110 /*
111  * Once there is this amount of dirty data, the dmu_tx_delay() will kick in
112  * and delay each transaction.
113  * This value should be >= zfs_vdev_async_write_active_max_dirty_percent.
114  */
115 int zfs_delay_min_dirty_percent = 60;
116 
117 /*
118  * This controls how quickly the delay approaches infinity.
119  * Larger values cause it to delay more for a given amount of dirty data.
120  * Therefore larger values will cause there to be less dirty data for a
121  * given throughput.
122  *
123  * For the smoothest delay, this value should be about 1 billion divided
124  * by the maximum number of operations per second.  This will smoothly
125  * handle between 10x and 1/10th this number.
126  *
127  * Note: zfs_delay_scale * zfs_dirty_data_max must be < 2^64, due to the
128  * multiply in dmu_tx_delay().
129  */
130 uint64_t zfs_delay_scale = 1000 * 1000 * 1000 / 2000;
131 
132 /*
133  * This determines the number of threads used by the dp_sync_taskq.
134  */
135 int zfs_sync_taskq_batch_pct = 75;
136 
137 /*
138  * These tunables determine the behavior of how zil_itxg_clean() is
139  * called via zil_clean() in the context of spa_sync(). When an itxg
140  * list needs to be cleaned, TQ_NOSLEEP will be used when dispatching.
141  * If the dispatch fails, the call to zil_itxg_clean() will occur
142  * synchronously in the context of spa_sync(), which can negatively
143  * impact the performance of spa_sync() (e.g. in the case of the itxg
144  * list having a large number of itxs that needs to be cleaned).
145  *
146  * Thus, these tunables can be used to manipulate the behavior of the
147  * taskq used by zil_clean(); they determine the number of taskq entries
148  * that are pre-populated when the taskq is first created (via the
149  * "zfs_zil_clean_taskq_minalloc" tunable) and the maximum number of
150  * taskq entries that are cached after an on-demand allocation (via the
151  * "zfs_zil_clean_taskq_maxalloc").
152  *
153  * The idea being, we want to try reasonably hard to ensure there will
154  * already be a taskq entry pre-allocated by the time that it is needed
155  * by zil_clean(). This way, we can avoid the possibility of an
156  * on-demand allocation of a new taskq entry from failing, which would
157  * result in zil_itxg_clean() being called synchronously from zil_clean()
158  * (which can adversely affect performance of spa_sync()).
159  *
160  * Additionally, the number of threads used by the taskq can be
161  * configured via the "zfs_zil_clean_taskq_nthr_pct" tunable.
162  */
163 int zfs_zil_clean_taskq_nthr_pct = 100;
164 int zfs_zil_clean_taskq_minalloc = 1024;
165 int zfs_zil_clean_taskq_maxalloc = 1024 * 1024;
166 
167 int
168 dsl_pool_open_special_dir(dsl_pool_t *dp, const char *name, dsl_dir_t **ddp)
169 {
170 	uint64_t obj;
171 	int err;
172 
173 	err = zap_lookup(dp->dp_meta_objset,
174 	    dsl_dir_phys(dp->dp_root_dir)->dd_child_dir_zapobj,
175 	    name, sizeof (obj), 1, &obj);
176 	if (err)
177 		return (err);
178 
179 	return (dsl_dir_hold_obj(dp, obj, name, dp, ddp));
180 }
181 
182 static dsl_pool_t *
183 dsl_pool_open_impl(spa_t *spa, uint64_t txg)
184 {
185 	dsl_pool_t *dp;
186 	blkptr_t *bp = spa_get_rootblkptr(spa);
187 
188 	dp = kmem_zalloc(sizeof (dsl_pool_t), KM_SLEEP);
189 	dp->dp_spa = spa;
190 	dp->dp_meta_rootbp = *bp;
191 	rrw_init(&dp->dp_config_rwlock, B_TRUE);
192 	txg_init(dp, txg);
193 
194 	txg_list_create(&dp->dp_dirty_datasets, spa,
195 	    offsetof(dsl_dataset_t, ds_dirty_link));
196 	txg_list_create(&dp->dp_dirty_zilogs, spa,
197 	    offsetof(zilog_t, zl_dirty_link));
198 	txg_list_create(&dp->dp_dirty_dirs, spa,
199 	    offsetof(dsl_dir_t, dd_dirty_link));
200 	txg_list_create(&dp->dp_sync_tasks, spa,
201 	    offsetof(dsl_sync_task_t, dst_node));
202 	txg_list_create(&dp->dp_early_sync_tasks, spa,
203 	    offsetof(dsl_sync_task_t, dst_node));
204 
205 	dp->dp_sync_taskq = taskq_create("dp_sync_taskq",
206 	    zfs_sync_taskq_batch_pct, minclsyspri, 1, INT_MAX,
207 	    TASKQ_THREADS_CPU_PCT);
208 
209 	dp->dp_zil_clean_taskq = taskq_create("dp_zil_clean_taskq",
210 	    zfs_zil_clean_taskq_nthr_pct, minclsyspri,
211 	    zfs_zil_clean_taskq_minalloc,
212 	    zfs_zil_clean_taskq_maxalloc,
213 	    TASKQ_PREPOPULATE | TASKQ_THREADS_CPU_PCT);
214 
215 	mutex_init(&dp->dp_lock, NULL, MUTEX_DEFAULT, NULL);
216 	cv_init(&dp->dp_spaceavail_cv, NULL, CV_DEFAULT, NULL);
217 
218 	dp->dp_vnrele_taskq = taskq_create("zfs_vn_rele_taskq", 1, minclsyspri,
219 	    1, 4, 0);
220 
221 	return (dp);
222 }
223 
224 int
225 dsl_pool_init(spa_t *spa, uint64_t txg, dsl_pool_t **dpp)
226 {
227 	int err;
228 	dsl_pool_t *dp = dsl_pool_open_impl(spa, txg);
229 
230 	err = dmu_objset_open_impl(spa, NULL, &dp->dp_meta_rootbp,
231 	    &dp->dp_meta_objset);
232 	if (err != 0)
233 		dsl_pool_close(dp);
234 	else
235 		*dpp = dp;
236 
237 	return (err);
238 }
239 
240 int
241 dsl_pool_open(dsl_pool_t *dp)
242 {
243 	int err;
244 	dsl_dir_t *dd;
245 	dsl_dataset_t *ds;
246 	uint64_t obj;
247 
248 	rrw_enter(&dp->dp_config_rwlock, RW_WRITER, FTAG);
249 	err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
250 	    DMU_POOL_ROOT_DATASET, sizeof (uint64_t), 1,
251 	    &dp->dp_root_dir_obj);
252 	if (err)
253 		goto out;
254 
255 	err = dsl_dir_hold_obj(dp, dp->dp_root_dir_obj,
256 	    NULL, dp, &dp->dp_root_dir);
257 	if (err)
258 		goto out;
259 
260 	err = dsl_pool_open_special_dir(dp, MOS_DIR_NAME, &dp->dp_mos_dir);
261 	if (err)
262 		goto out;
263 
264 	if (spa_version(dp->dp_spa) >= SPA_VERSION_ORIGIN) {
265 		err = dsl_pool_open_special_dir(dp, ORIGIN_DIR_NAME, &dd);
266 		if (err)
267 			goto out;
268 		err = dsl_dataset_hold_obj(dp,
269 		    dsl_dir_phys(dd)->dd_head_dataset_obj, FTAG, &ds);
270 		if (err == 0) {
271 			err = dsl_dataset_hold_obj(dp,
272 			    dsl_dataset_phys(ds)->ds_prev_snap_obj, dp,
273 			    &dp->dp_origin_snap);
274 			dsl_dataset_rele(ds, FTAG);
275 		}
276 		dsl_dir_rele(dd, dp);
277 		if (err)
278 			goto out;
279 	}
280 
281 	if (spa_version(dp->dp_spa) >= SPA_VERSION_DEADLISTS) {
282 		err = dsl_pool_open_special_dir(dp, FREE_DIR_NAME,
283 		    &dp->dp_free_dir);
284 		if (err)
285 			goto out;
286 
287 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
288 		    DMU_POOL_FREE_BPOBJ, sizeof (uint64_t), 1, &obj);
289 		if (err)
290 			goto out;
291 		VERIFY0(bpobj_open(&dp->dp_free_bpobj,
292 		    dp->dp_meta_objset, obj));
293 	}
294 
295 	if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_OBSOLETE_COUNTS)) {
296 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
297 		    DMU_POOL_OBSOLETE_BPOBJ, sizeof (uint64_t), 1, &obj);
298 		if (err == 0) {
299 			VERIFY0(bpobj_open(&dp->dp_obsolete_bpobj,
300 			    dp->dp_meta_objset, obj));
301 		} else if (err == ENOENT) {
302 			/*
303 			 * We might not have created the remap bpobj yet.
304 			 */
305 			err = 0;
306 		} else {
307 			goto out;
308 		}
309 	}
310 
311 	/*
312 	 * Note: errors ignored, because the these special dirs, used for
313 	 * space accounting, are only created on demand.
314 	 */
315 	(void) dsl_pool_open_special_dir(dp, LEAK_DIR_NAME,
316 	    &dp->dp_leak_dir);
317 
318 	if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_ASYNC_DESTROY)) {
319 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
320 		    DMU_POOL_BPTREE_OBJ, sizeof (uint64_t), 1,
321 		    &dp->dp_bptree_obj);
322 		if (err != 0)
323 			goto out;
324 	}
325 
326 	if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_EMPTY_BPOBJ)) {
327 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
328 		    DMU_POOL_EMPTY_BPOBJ, sizeof (uint64_t), 1,
329 		    &dp->dp_empty_bpobj);
330 		if (err != 0)
331 			goto out;
332 	}
333 
334 	err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
335 	    DMU_POOL_TMP_USERREFS, sizeof (uint64_t), 1,
336 	    &dp->dp_tmp_userrefs_obj);
337 	if (err == ENOENT)
338 		err = 0;
339 	if (err)
340 		goto out;
341 
342 	err = dsl_scan_init(dp, dp->dp_tx.tx_open_txg);
343 
344 out:
345 	rrw_exit(&dp->dp_config_rwlock, FTAG);
346 	return (err);
347 }
348 
349 void
350 dsl_pool_close(dsl_pool_t *dp)
351 {
352 	/*
353 	 * Drop our references from dsl_pool_open().
354 	 *
355 	 * Since we held the origin_snap from "syncing" context (which
356 	 * includes pool-opening context), it actually only got a "ref"
357 	 * and not a hold, so just drop that here.
358 	 */
359 	if (dp->dp_origin_snap != NULL)
360 		dsl_dataset_rele(dp->dp_origin_snap, dp);
361 	if (dp->dp_mos_dir != NULL)
362 		dsl_dir_rele(dp->dp_mos_dir, dp);
363 	if (dp->dp_free_dir != NULL)
364 		dsl_dir_rele(dp->dp_free_dir, dp);
365 	if (dp->dp_leak_dir != NULL)
366 		dsl_dir_rele(dp->dp_leak_dir, dp);
367 	if (dp->dp_root_dir != NULL)
368 		dsl_dir_rele(dp->dp_root_dir, dp);
369 
370 	bpobj_close(&dp->dp_free_bpobj);
371 	bpobj_close(&dp->dp_obsolete_bpobj);
372 
373 	/* undo the dmu_objset_open_impl(mos) from dsl_pool_open() */
374 	if (dp->dp_meta_objset != NULL)
375 		dmu_objset_evict(dp->dp_meta_objset);
376 
377 	txg_list_destroy(&dp->dp_dirty_datasets);
378 	txg_list_destroy(&dp->dp_dirty_zilogs);
379 	txg_list_destroy(&dp->dp_sync_tasks);
380 	txg_list_destroy(&dp->dp_early_sync_tasks);
381 	txg_list_destroy(&dp->dp_dirty_dirs);
382 
383 	taskq_destroy(dp->dp_zil_clean_taskq);
384 	taskq_destroy(dp->dp_sync_taskq);
385 
386 	/*
387 	 * We can't set retry to TRUE since we're explicitly specifying
388 	 * a spa to flush. This is good enough; any missed buffers for
389 	 * this spa won't cause trouble, and they'll eventually fall
390 	 * out of the ARC just like any other unused buffer.
391 	 */
392 	arc_flush(dp->dp_spa, FALSE);
393 
394 	txg_fini(dp);
395 	dsl_scan_fini(dp);
396 	dmu_buf_user_evict_wait();
397 
398 	rrw_destroy(&dp->dp_config_rwlock);
399 	mutex_destroy(&dp->dp_lock);
400 	taskq_destroy(dp->dp_vnrele_taskq);
401 	if (dp->dp_blkstats != NULL)
402 		kmem_free(dp->dp_blkstats, sizeof (zfs_all_blkstats_t));
403 	kmem_free(dp, sizeof (dsl_pool_t));
404 }
405 
406 void
407 dsl_pool_create_obsolete_bpobj(dsl_pool_t *dp, dmu_tx_t *tx)
408 {
409 	uint64_t obj;
410 	/*
411 	 * Currently, we only create the obsolete_bpobj where there are
412 	 * indirect vdevs with referenced mappings.
413 	 */
414 	ASSERT(spa_feature_is_active(dp->dp_spa, SPA_FEATURE_DEVICE_REMOVAL));
415 	/* create and open the obsolete_bpobj */
416 	obj = bpobj_alloc(dp->dp_meta_objset, SPA_OLD_MAXBLOCKSIZE, tx);
417 	VERIFY0(bpobj_open(&dp->dp_obsolete_bpobj, dp->dp_meta_objset, obj));
418 	VERIFY0(zap_add(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
419 	    DMU_POOL_OBSOLETE_BPOBJ, sizeof (uint64_t), 1, &obj, tx));
420 	spa_feature_incr(dp->dp_spa, SPA_FEATURE_OBSOLETE_COUNTS, tx);
421 }
422 
423 void
424 dsl_pool_destroy_obsolete_bpobj(dsl_pool_t *dp, dmu_tx_t *tx)
425 {
426 	spa_feature_decr(dp->dp_spa, SPA_FEATURE_OBSOLETE_COUNTS, tx);
427 	VERIFY0(zap_remove(dp->dp_meta_objset,
428 	    DMU_POOL_DIRECTORY_OBJECT,
429 	    DMU_POOL_OBSOLETE_BPOBJ, tx));
430 	bpobj_free(dp->dp_meta_objset,
431 	    dp->dp_obsolete_bpobj.bpo_object, tx);
432 	bpobj_close(&dp->dp_obsolete_bpobj);
433 }
434 
435 dsl_pool_t *
436 dsl_pool_create(spa_t *spa, nvlist_t *zplprops, uint64_t txg)
437 {
438 	int err;
439 	dsl_pool_t *dp = dsl_pool_open_impl(spa, txg);
440 	dmu_tx_t *tx = dmu_tx_create_assigned(dp, txg);
441 	dsl_dataset_t *ds;
442 	uint64_t obj;
443 
444 	rrw_enter(&dp->dp_config_rwlock, RW_WRITER, FTAG);
445 
446 	/* create and open the MOS (meta-objset) */
447 	dp->dp_meta_objset = dmu_objset_create_impl(spa,
448 	    NULL, &dp->dp_meta_rootbp, DMU_OST_META, tx);
449 
450 	/* create the pool directory */
451 	err = zap_create_claim(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
452 	    DMU_OT_OBJECT_DIRECTORY, DMU_OT_NONE, 0, tx);
453 	ASSERT0(err);
454 
455 	/* Initialize scan structures */
456 	VERIFY0(dsl_scan_init(dp, txg));
457 
458 	/* create and open the root dir */
459 	dp->dp_root_dir_obj = dsl_dir_create_sync(dp, NULL, NULL, tx);
460 	VERIFY0(dsl_dir_hold_obj(dp, dp->dp_root_dir_obj,
461 	    NULL, dp, &dp->dp_root_dir));
462 
463 	/* create and open the meta-objset dir */
464 	(void) dsl_dir_create_sync(dp, dp->dp_root_dir, MOS_DIR_NAME, tx);
465 	VERIFY0(dsl_pool_open_special_dir(dp,
466 	    MOS_DIR_NAME, &dp->dp_mos_dir));
467 
468 	if (spa_version(spa) >= SPA_VERSION_DEADLISTS) {
469 		/* create and open the free dir */
470 		(void) dsl_dir_create_sync(dp, dp->dp_root_dir,
471 		    FREE_DIR_NAME, tx);
472 		VERIFY0(dsl_pool_open_special_dir(dp,
473 		    FREE_DIR_NAME, &dp->dp_free_dir));
474 
475 		/* create and open the free_bplist */
476 		obj = bpobj_alloc(dp->dp_meta_objset, SPA_OLD_MAXBLOCKSIZE, tx);
477 		VERIFY(zap_add(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
478 		    DMU_POOL_FREE_BPOBJ, sizeof (uint64_t), 1, &obj, tx) == 0);
479 		VERIFY0(bpobj_open(&dp->dp_free_bpobj,
480 		    dp->dp_meta_objset, obj));
481 	}
482 
483 	if (spa_version(spa) >= SPA_VERSION_DSL_SCRUB)
484 		dsl_pool_create_origin(dp, tx);
485 
486 	/* create the root dataset */
487 	obj = dsl_dataset_create_sync_dd(dp->dp_root_dir, NULL, 0, tx);
488 
489 	/* create the root objset */
490 	VERIFY0(dsl_dataset_hold_obj(dp, obj, FTAG, &ds));
491 #ifdef _KERNEL
492 	{
493 		objset_t *os;
494 		rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
495 		os = dmu_objset_create_impl(dp->dp_spa, ds,
496 		    dsl_dataset_get_blkptr(ds), DMU_OST_ZFS, tx);
497 		rrw_exit(&ds->ds_bp_rwlock, FTAG);
498 		zfs_create_fs(os, kcred, zplprops, tx);
499 	}
500 #endif
501 	dsl_dataset_rele(ds, FTAG);
502 
503 	dmu_tx_commit(tx);
504 
505 	rrw_exit(&dp->dp_config_rwlock, FTAG);
506 
507 	return (dp);
508 }
509 
510 /*
511  * Account for the meta-objset space in its placeholder dsl_dir.
512  */
513 void
514 dsl_pool_mos_diduse_space(dsl_pool_t *dp,
515     int64_t used, int64_t comp, int64_t uncomp)
516 {
517 	ASSERT3U(comp, ==, uncomp); /* it's all metadata */
518 	mutex_enter(&dp->dp_lock);
519 	dp->dp_mos_used_delta += used;
520 	dp->dp_mos_compressed_delta += comp;
521 	dp->dp_mos_uncompressed_delta += uncomp;
522 	mutex_exit(&dp->dp_lock);
523 }
524 
525 static void
526 dsl_pool_sync_mos(dsl_pool_t *dp, dmu_tx_t *tx)
527 {
528 	zio_t *zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
529 	dmu_objset_sync(dp->dp_meta_objset, zio, tx);
530 	VERIFY0(zio_wait(zio));
531 	dprintf_bp(&dp->dp_meta_rootbp, "meta objset rootbp is %s", "");
532 	spa_set_rootblkptr(dp->dp_spa, &dp->dp_meta_rootbp);
533 }
534 
535 static void
536 dsl_pool_dirty_delta(dsl_pool_t *dp, int64_t delta)
537 {
538 	ASSERT(MUTEX_HELD(&dp->dp_lock));
539 
540 	if (delta < 0)
541 		ASSERT3U(-delta, <=, dp->dp_dirty_total);
542 
543 	dp->dp_dirty_total += delta;
544 
545 	/*
546 	 * Note: we signal even when increasing dp_dirty_total.
547 	 * This ensures forward progress -- each thread wakes the next waiter.
548 	 */
549 	if (dp->dp_dirty_total < zfs_dirty_data_max)
550 		cv_signal(&dp->dp_spaceavail_cv);
551 }
552 
553 static boolean_t
554 dsl_early_sync_task_verify(dsl_pool_t *dp, uint64_t txg)
555 {
556 	spa_t *spa = dp->dp_spa;
557 	vdev_t *rvd = spa->spa_root_vdev;
558 
559 	for (uint64_t c = 0; c < rvd->vdev_children; c++) {
560 		vdev_t *vd = rvd->vdev_child[c];
561 		txg_list_t *tl = &vd->vdev_ms_list;
562 		metaslab_t *ms;
563 
564 		for (ms = txg_list_head(tl, TXG_CLEAN(txg)); ms;
565 		    ms = txg_list_next(tl, ms, TXG_CLEAN(txg))) {
566 			VERIFY(range_tree_is_empty(ms->ms_freeing));
567 			VERIFY(range_tree_is_empty(ms->ms_checkpointing));
568 		}
569 	}
570 
571 	return (B_TRUE);
572 }
573 
574 void
575 dsl_pool_sync(dsl_pool_t *dp, uint64_t txg)
576 {
577 	zio_t *zio;
578 	dmu_tx_t *tx;
579 	dsl_dir_t *dd;
580 	dsl_dataset_t *ds;
581 	objset_t *mos = dp->dp_meta_objset;
582 	list_t synced_datasets;
583 
584 	list_create(&synced_datasets, sizeof (dsl_dataset_t),
585 	    offsetof(dsl_dataset_t, ds_synced_link));
586 
587 	tx = dmu_tx_create_assigned(dp, txg);
588 
589 	/*
590 	 * Run all early sync tasks before writing out any dirty blocks.
591 	 * For more info on early sync tasks see block comment in
592 	 * dsl_early_sync_task().
593 	 */
594 	if (!txg_list_empty(&dp->dp_early_sync_tasks, txg)) {
595 		dsl_sync_task_t *dst;
596 
597 		ASSERT3U(spa_sync_pass(dp->dp_spa), ==, 1);
598 		while ((dst =
599 		    txg_list_remove(&dp->dp_early_sync_tasks, txg)) != NULL) {
600 			ASSERT(dsl_early_sync_task_verify(dp, txg));
601 			dsl_sync_task_sync(dst, tx);
602 		}
603 		ASSERT(dsl_early_sync_task_verify(dp, txg));
604 	}
605 
606 	/*
607 	 * Write out all dirty blocks of dirty datasets.
608 	 */
609 	zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
610 	while ((ds = txg_list_remove(&dp->dp_dirty_datasets, txg)) != NULL) {
611 		/*
612 		 * We must not sync any non-MOS datasets twice, because
613 		 * we may have taken a snapshot of them.  However, we
614 		 * may sync newly-created datasets on pass 2.
615 		 */
616 		ASSERT(!list_link_active(&ds->ds_synced_link));
617 		list_insert_tail(&synced_datasets, ds);
618 		dsl_dataset_sync(ds, zio, tx);
619 	}
620 	VERIFY0(zio_wait(zio));
621 
622 	/*
623 	 * We have written all of the accounted dirty data, so our
624 	 * dp_space_towrite should now be zero.  However, some seldom-used
625 	 * code paths do not adhere to this (e.g. dbuf_undirty(), also
626 	 * rounding error in dbuf_write_physdone).
627 	 * Shore up the accounting of any dirtied space now.
628 	 */
629 	dsl_pool_undirty_space(dp, dp->dp_dirty_pertxg[txg & TXG_MASK], txg);
630 
631 	/*
632 	 * Update the long range free counter after
633 	 * we're done syncing user data
634 	 */
635 	mutex_enter(&dp->dp_lock);
636 	ASSERT(spa_sync_pass(dp->dp_spa) == 1 ||
637 	    dp->dp_long_free_dirty_pertxg[txg & TXG_MASK] == 0);
638 	dp->dp_long_free_dirty_pertxg[txg & TXG_MASK] = 0;
639 	mutex_exit(&dp->dp_lock);
640 
641 	/*
642 	 * After the data blocks have been written (ensured by the zio_wait()
643 	 * above), update the user/group space accounting.  This happens
644 	 * in tasks dispatched to dp_sync_taskq, so wait for them before
645 	 * continuing.
646 	 */
647 	for (ds = list_head(&synced_datasets); ds != NULL;
648 	    ds = list_next(&synced_datasets, ds)) {
649 		dmu_objset_do_userquota_updates(ds->ds_objset, tx);
650 	}
651 	taskq_wait(dp->dp_sync_taskq);
652 
653 	/*
654 	 * Sync the datasets again to push out the changes due to
655 	 * userspace updates.  This must be done before we process the
656 	 * sync tasks, so that any snapshots will have the correct
657 	 * user accounting information (and we won't get confused
658 	 * about which blocks are part of the snapshot).
659 	 */
660 	zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
661 	while ((ds = txg_list_remove(&dp->dp_dirty_datasets, txg)) != NULL) {
662 		ASSERT(list_link_active(&ds->ds_synced_link));
663 		dmu_buf_rele(ds->ds_dbuf, ds);
664 		dsl_dataset_sync(ds, zio, tx);
665 	}
666 	VERIFY0(zio_wait(zio));
667 
668 	/*
669 	 * Now that the datasets have been completely synced, we can
670 	 * clean up our in-memory structures accumulated while syncing:
671 	 *
672 	 *  - move dead blocks from the pending deadlist to the on-disk deadlist
673 	 *  - release hold from dsl_dataset_dirty()
674 	 */
675 	while ((ds = list_remove_head(&synced_datasets)) != NULL) {
676 		dsl_dataset_sync_done(ds, tx);
677 	}
678 	while ((dd = txg_list_remove(&dp->dp_dirty_dirs, txg)) != NULL) {
679 		dsl_dir_sync(dd, tx);
680 	}
681 
682 	/*
683 	 * The MOS's space is accounted for in the pool/$MOS
684 	 * (dp_mos_dir).  We can't modify the mos while we're syncing
685 	 * it, so we remember the deltas and apply them here.
686 	 */
687 	if (dp->dp_mos_used_delta != 0 || dp->dp_mos_compressed_delta != 0 ||
688 	    dp->dp_mos_uncompressed_delta != 0) {
689 		dsl_dir_diduse_space(dp->dp_mos_dir, DD_USED_HEAD,
690 		    dp->dp_mos_used_delta,
691 		    dp->dp_mos_compressed_delta,
692 		    dp->dp_mos_uncompressed_delta, tx);
693 		dp->dp_mos_used_delta = 0;
694 		dp->dp_mos_compressed_delta = 0;
695 		dp->dp_mos_uncompressed_delta = 0;
696 	}
697 
698 	if (!multilist_is_empty(mos->os_dirty_dnodes[txg & TXG_MASK])) {
699 		dsl_pool_sync_mos(dp, tx);
700 	}
701 
702 	/*
703 	 * If we modify a dataset in the same txg that we want to destroy it,
704 	 * its dsl_dir's dd_dbuf will be dirty, and thus have a hold on it.
705 	 * dsl_dir_destroy_check() will fail if there are unexpected holds.
706 	 * Therefore, we want to sync the MOS (thus syncing the dd_dbuf
707 	 * and clearing the hold on it) before we process the sync_tasks.
708 	 * The MOS data dirtied by the sync_tasks will be synced on the next
709 	 * pass.
710 	 */
711 	if (!txg_list_empty(&dp->dp_sync_tasks, txg)) {
712 		dsl_sync_task_t *dst;
713 		/*
714 		 * No more sync tasks should have been added while we
715 		 * were syncing.
716 		 */
717 		ASSERT3U(spa_sync_pass(dp->dp_spa), ==, 1);
718 		while ((dst = txg_list_remove(&dp->dp_sync_tasks, txg)) != NULL)
719 			dsl_sync_task_sync(dst, tx);
720 	}
721 
722 	dmu_tx_commit(tx);
723 
724 	DTRACE_PROBE2(dsl_pool_sync__done, dsl_pool_t *dp, dp, uint64_t, txg);
725 }
726 
727 void
728 dsl_pool_sync_done(dsl_pool_t *dp, uint64_t txg)
729 {
730 	zilog_t *zilog;
731 
732 	while (zilog = txg_list_head(&dp->dp_dirty_zilogs, txg)) {
733 		dsl_dataset_t *ds = dmu_objset_ds(zilog->zl_os);
734 		/*
735 		 * We don't remove the zilog from the dp_dirty_zilogs
736 		 * list until after we've cleaned it. This ensures that
737 		 * callers of zilog_is_dirty() receive an accurate
738 		 * answer when they are racing with the spa sync thread.
739 		 */
740 		zil_clean(zilog, txg);
741 		(void) txg_list_remove_this(&dp->dp_dirty_zilogs, zilog, txg);
742 		ASSERT(!dmu_objset_is_dirty(zilog->zl_os, txg));
743 		dmu_buf_rele(ds->ds_dbuf, zilog);
744 	}
745 	ASSERT(!dmu_objset_is_dirty(dp->dp_meta_objset, txg));
746 }
747 
748 /*
749  * TRUE if the current thread is the tx_sync_thread or if we
750  * are being called from SPA context during pool initialization.
751  */
752 int
753 dsl_pool_sync_context(dsl_pool_t *dp)
754 {
755 	return (curthread == dp->dp_tx.tx_sync_thread ||
756 	    spa_is_initializing(dp->dp_spa) ||
757 	    taskq_member(dp->dp_sync_taskq, curthread));
758 }
759 
760 /*
761  * This function returns the amount of allocatable space in the pool
762  * minus whatever space is currently reserved by ZFS for specific
763  * purposes. Specifically:
764  *
765  * 1] Any reserved SLOP space
766  * 2] Any space used by the checkpoint
767  * 3] Any space used for deferred frees
768  *
769  * The latter 2 are especially important because they are needed to
770  * rectify the SPA's and DMU's different understanding of how much space
771  * is used. Now the DMU is aware of that extra space tracked by the SPA
772  * without having to maintain a separate special dir (e.g similar to
773  * $MOS, $FREEING, and $LEAKED).
774  *
775  * Note: By deferred frees here, we mean the frees that were deferred
776  * in spa_sync() after sync pass 1 (spa_deferred_bpobj), and not the
777  * segments placed in ms_defer trees during metaslab_sync_done().
778  */
779 uint64_t
780 dsl_pool_adjustedsize(dsl_pool_t *dp, zfs_space_check_t slop_policy)
781 {
782 	spa_t *spa = dp->dp_spa;
783 	uint64_t space, resv, adjustedsize;
784 	uint64_t spa_deferred_frees =
785 	    spa->spa_deferred_bpobj.bpo_phys->bpo_bytes;
786 
787 	space = spa_get_dspace(spa)
788 	    - spa_get_checkpoint_space(spa) - spa_deferred_frees;
789 	resv = spa_get_slop_space(spa);
790 
791 	switch (slop_policy) {
792 	case ZFS_SPACE_CHECK_NORMAL:
793 		break;
794 	case ZFS_SPACE_CHECK_RESERVED:
795 		resv >>= 1;
796 		break;
797 	case ZFS_SPACE_CHECK_EXTRA_RESERVED:
798 		resv >>= 2;
799 		break;
800 	case ZFS_SPACE_CHECK_NONE:
801 		resv = 0;
802 		break;
803 	default:
804 		panic("invalid slop policy value: %d", slop_policy);
805 		break;
806 	}
807 	adjustedsize = (space >= resv) ? (space - resv) : 0;
808 
809 	return (adjustedsize);
810 }
811 
812 uint64_t
813 dsl_pool_unreserved_space(dsl_pool_t *dp, zfs_space_check_t slop_policy)
814 {
815 	uint64_t poolsize = dsl_pool_adjustedsize(dp, slop_policy);
816 	uint64_t deferred =
817 	    metaslab_class_get_deferred(spa_normal_class(dp->dp_spa));
818 	uint64_t quota = (poolsize >= deferred) ? (poolsize - deferred) : 0;
819 	return (quota);
820 }
821 
822 boolean_t
823 dsl_pool_need_dirty_delay(dsl_pool_t *dp)
824 {
825 	uint64_t delay_min_bytes =
826 	    zfs_dirty_data_max * zfs_delay_min_dirty_percent / 100;
827 	boolean_t rv;
828 
829 	mutex_enter(&dp->dp_lock);
830 	if (dp->dp_dirty_total > zfs_dirty_data_sync)
831 		txg_kick(dp);
832 	rv = (dp->dp_dirty_total > delay_min_bytes);
833 	mutex_exit(&dp->dp_lock);
834 	return (rv);
835 }
836 
837 void
838 dsl_pool_dirty_space(dsl_pool_t *dp, int64_t space, dmu_tx_t *tx)
839 {
840 	if (space > 0) {
841 		mutex_enter(&dp->dp_lock);
842 		dp->dp_dirty_pertxg[tx->tx_txg & TXG_MASK] += space;
843 		dsl_pool_dirty_delta(dp, space);
844 		mutex_exit(&dp->dp_lock);
845 	}
846 }
847 
848 void
849 dsl_pool_undirty_space(dsl_pool_t *dp, int64_t space, uint64_t txg)
850 {
851 	ASSERT3S(space, >=, 0);
852 	if (space == 0)
853 		return;
854 	mutex_enter(&dp->dp_lock);
855 	if (dp->dp_dirty_pertxg[txg & TXG_MASK] < space) {
856 		/* XXX writing something we didn't dirty? */
857 		space = dp->dp_dirty_pertxg[txg & TXG_MASK];
858 	}
859 	ASSERT3U(dp->dp_dirty_pertxg[txg & TXG_MASK], >=, space);
860 	dp->dp_dirty_pertxg[txg & TXG_MASK] -= space;
861 	ASSERT3U(dp->dp_dirty_total, >=, space);
862 	dsl_pool_dirty_delta(dp, -space);
863 	mutex_exit(&dp->dp_lock);
864 }
865 
866 /* ARGSUSED */
867 static int
868 upgrade_clones_cb(dsl_pool_t *dp, dsl_dataset_t *hds, void *arg)
869 {
870 	dmu_tx_t *tx = arg;
871 	dsl_dataset_t *ds, *prev = NULL;
872 	int err;
873 
874 	err = dsl_dataset_hold_obj(dp, hds->ds_object, FTAG, &ds);
875 	if (err)
876 		return (err);
877 
878 	while (dsl_dataset_phys(ds)->ds_prev_snap_obj != 0) {
879 		err = dsl_dataset_hold_obj(dp,
880 		    dsl_dataset_phys(ds)->ds_prev_snap_obj, FTAG, &prev);
881 		if (err) {
882 			dsl_dataset_rele(ds, FTAG);
883 			return (err);
884 		}
885 
886 		if (dsl_dataset_phys(prev)->ds_next_snap_obj != ds->ds_object)
887 			break;
888 		dsl_dataset_rele(ds, FTAG);
889 		ds = prev;
890 		prev = NULL;
891 	}
892 
893 	if (prev == NULL) {
894 		prev = dp->dp_origin_snap;
895 
896 		/*
897 		 * The $ORIGIN can't have any data, or the accounting
898 		 * will be wrong.
899 		 */
900 		rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
901 		ASSERT0(dsl_dataset_phys(prev)->ds_bp.blk_birth);
902 		rrw_exit(&ds->ds_bp_rwlock, FTAG);
903 
904 		/* The origin doesn't get attached to itself */
905 		if (ds->ds_object == prev->ds_object) {
906 			dsl_dataset_rele(ds, FTAG);
907 			return (0);
908 		}
909 
910 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
911 		dsl_dataset_phys(ds)->ds_prev_snap_obj = prev->ds_object;
912 		dsl_dataset_phys(ds)->ds_prev_snap_txg =
913 		    dsl_dataset_phys(prev)->ds_creation_txg;
914 
915 		dmu_buf_will_dirty(ds->ds_dir->dd_dbuf, tx);
916 		dsl_dir_phys(ds->ds_dir)->dd_origin_obj = prev->ds_object;
917 
918 		dmu_buf_will_dirty(prev->ds_dbuf, tx);
919 		dsl_dataset_phys(prev)->ds_num_children++;
920 
921 		if (dsl_dataset_phys(ds)->ds_next_snap_obj == 0) {
922 			ASSERT(ds->ds_prev == NULL);
923 			VERIFY0(dsl_dataset_hold_obj(dp,
924 			    dsl_dataset_phys(ds)->ds_prev_snap_obj,
925 			    ds, &ds->ds_prev));
926 		}
927 	}
928 
929 	ASSERT3U(dsl_dir_phys(ds->ds_dir)->dd_origin_obj, ==, prev->ds_object);
930 	ASSERT3U(dsl_dataset_phys(ds)->ds_prev_snap_obj, ==, prev->ds_object);
931 
932 	if (dsl_dataset_phys(prev)->ds_next_clones_obj == 0) {
933 		dmu_buf_will_dirty(prev->ds_dbuf, tx);
934 		dsl_dataset_phys(prev)->ds_next_clones_obj =
935 		    zap_create(dp->dp_meta_objset,
936 		    DMU_OT_NEXT_CLONES, DMU_OT_NONE, 0, tx);
937 	}
938 	VERIFY0(zap_add_int(dp->dp_meta_objset,
939 	    dsl_dataset_phys(prev)->ds_next_clones_obj, ds->ds_object, tx));
940 
941 	dsl_dataset_rele(ds, FTAG);
942 	if (prev != dp->dp_origin_snap)
943 		dsl_dataset_rele(prev, FTAG);
944 	return (0);
945 }
946 
947 void
948 dsl_pool_upgrade_clones(dsl_pool_t *dp, dmu_tx_t *tx)
949 {
950 	ASSERT(dmu_tx_is_syncing(tx));
951 	ASSERT(dp->dp_origin_snap != NULL);
952 
953 	VERIFY0(dmu_objset_find_dp(dp, dp->dp_root_dir_obj, upgrade_clones_cb,
954 	    tx, DS_FIND_CHILDREN | DS_FIND_SERIALIZE));
955 }
956 
957 /* ARGSUSED */
958 static int
959 upgrade_dir_clones_cb(dsl_pool_t *dp, dsl_dataset_t *ds, void *arg)
960 {
961 	dmu_tx_t *tx = arg;
962 	objset_t *mos = dp->dp_meta_objset;
963 
964 	if (dsl_dir_phys(ds->ds_dir)->dd_origin_obj != 0) {
965 		dsl_dataset_t *origin;
966 
967 		VERIFY0(dsl_dataset_hold_obj(dp,
968 		    dsl_dir_phys(ds->ds_dir)->dd_origin_obj, FTAG, &origin));
969 
970 		if (dsl_dir_phys(origin->ds_dir)->dd_clones == 0) {
971 			dmu_buf_will_dirty(origin->ds_dir->dd_dbuf, tx);
972 			dsl_dir_phys(origin->ds_dir)->dd_clones =
973 			    zap_create(mos, DMU_OT_DSL_CLONES, DMU_OT_NONE,
974 			    0, tx);
975 		}
976 
977 		VERIFY0(zap_add_int(dp->dp_meta_objset,
978 		    dsl_dir_phys(origin->ds_dir)->dd_clones,
979 		    ds->ds_object, tx));
980 
981 		dsl_dataset_rele(origin, FTAG);
982 	}
983 	return (0);
984 }
985 
986 void
987 dsl_pool_upgrade_dir_clones(dsl_pool_t *dp, dmu_tx_t *tx)
988 {
989 	ASSERT(dmu_tx_is_syncing(tx));
990 	uint64_t obj;
991 
992 	(void) dsl_dir_create_sync(dp, dp->dp_root_dir, FREE_DIR_NAME, tx);
993 	VERIFY0(dsl_pool_open_special_dir(dp,
994 	    FREE_DIR_NAME, &dp->dp_free_dir));
995 
996 	/*
997 	 * We can't use bpobj_alloc(), because spa_version() still
998 	 * returns the old version, and we need a new-version bpobj with
999 	 * subobj support.  So call dmu_object_alloc() directly.
1000 	 */
1001 	obj = dmu_object_alloc(dp->dp_meta_objset, DMU_OT_BPOBJ,
1002 	    SPA_OLD_MAXBLOCKSIZE, DMU_OT_BPOBJ_HDR, sizeof (bpobj_phys_t), tx);
1003 	VERIFY0(zap_add(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
1004 	    DMU_POOL_FREE_BPOBJ, sizeof (uint64_t), 1, &obj, tx));
1005 	VERIFY0(bpobj_open(&dp->dp_free_bpobj, dp->dp_meta_objset, obj));
1006 
1007 	VERIFY0(dmu_objset_find_dp(dp, dp->dp_root_dir_obj,
1008 	    upgrade_dir_clones_cb, tx, DS_FIND_CHILDREN | DS_FIND_SERIALIZE));
1009 }
1010 
1011 void
1012 dsl_pool_create_origin(dsl_pool_t *dp, dmu_tx_t *tx)
1013 {
1014 	uint64_t dsobj;
1015 	dsl_dataset_t *ds;
1016 
1017 	ASSERT(dmu_tx_is_syncing(tx));
1018 	ASSERT(dp->dp_origin_snap == NULL);
1019 	ASSERT(rrw_held(&dp->dp_config_rwlock, RW_WRITER));
1020 
1021 	/* create the origin dir, ds, & snap-ds */
1022 	dsobj = dsl_dataset_create_sync(dp->dp_root_dir, ORIGIN_DIR_NAME,
1023 	    NULL, 0, kcred, tx);
1024 	VERIFY0(dsl_dataset_hold_obj(dp, dsobj, FTAG, &ds));
1025 	dsl_dataset_snapshot_sync_impl(ds, ORIGIN_DIR_NAME, tx);
1026 	VERIFY0(dsl_dataset_hold_obj(dp, dsl_dataset_phys(ds)->ds_prev_snap_obj,
1027 	    dp, &dp->dp_origin_snap));
1028 	dsl_dataset_rele(ds, FTAG);
1029 }
1030 
1031 taskq_t *
1032 dsl_pool_vnrele_taskq(dsl_pool_t *dp)
1033 {
1034 	return (dp->dp_vnrele_taskq);
1035 }
1036 
1037 /*
1038  * Walk through the pool-wide zap object of temporary snapshot user holds
1039  * and release them.
1040  */
1041 void
1042 dsl_pool_clean_tmp_userrefs(dsl_pool_t *dp)
1043 {
1044 	zap_attribute_t za;
1045 	zap_cursor_t zc;
1046 	objset_t *mos = dp->dp_meta_objset;
1047 	uint64_t zapobj = dp->dp_tmp_userrefs_obj;
1048 	nvlist_t *holds;
1049 
1050 	if (zapobj == 0)
1051 		return;
1052 	ASSERT(spa_version(dp->dp_spa) >= SPA_VERSION_USERREFS);
1053 
1054 	holds = fnvlist_alloc();
1055 
1056 	for (zap_cursor_init(&zc, mos, zapobj);
1057 	    zap_cursor_retrieve(&zc, &za) == 0;
1058 	    zap_cursor_advance(&zc)) {
1059 		char *htag;
1060 		nvlist_t *tags;
1061 
1062 		htag = strchr(za.za_name, '-');
1063 		*htag = '\0';
1064 		++htag;
1065 		if (nvlist_lookup_nvlist(holds, za.za_name, &tags) != 0) {
1066 			tags = fnvlist_alloc();
1067 			fnvlist_add_boolean(tags, htag);
1068 			fnvlist_add_nvlist(holds, za.za_name, tags);
1069 			fnvlist_free(tags);
1070 		} else {
1071 			fnvlist_add_boolean(tags, htag);
1072 		}
1073 	}
1074 	dsl_dataset_user_release_tmp(dp, holds);
1075 	fnvlist_free(holds);
1076 	zap_cursor_fini(&zc);
1077 }
1078 
1079 /*
1080  * Create the pool-wide zap object for storing temporary snapshot holds.
1081  */
1082 void
1083 dsl_pool_user_hold_create_obj(dsl_pool_t *dp, dmu_tx_t *tx)
1084 {
1085 	objset_t *mos = dp->dp_meta_objset;
1086 
1087 	ASSERT(dp->dp_tmp_userrefs_obj == 0);
1088 	ASSERT(dmu_tx_is_syncing(tx));
1089 
1090 	dp->dp_tmp_userrefs_obj = zap_create_link(mos, DMU_OT_USERREFS,
1091 	    DMU_POOL_DIRECTORY_OBJECT, DMU_POOL_TMP_USERREFS, tx);
1092 }
1093 
1094 static int
1095 dsl_pool_user_hold_rele_impl(dsl_pool_t *dp, uint64_t dsobj,
1096     const char *tag, uint64_t now, dmu_tx_t *tx, boolean_t holding)
1097 {
1098 	objset_t *mos = dp->dp_meta_objset;
1099 	uint64_t zapobj = dp->dp_tmp_userrefs_obj;
1100 	char *name;
1101 	int error;
1102 
1103 	ASSERT(spa_version(dp->dp_spa) >= SPA_VERSION_USERREFS);
1104 	ASSERT(dmu_tx_is_syncing(tx));
1105 
1106 	/*
1107 	 * If the pool was created prior to SPA_VERSION_USERREFS, the
1108 	 * zap object for temporary holds might not exist yet.
1109 	 */
1110 	if (zapobj == 0) {
1111 		if (holding) {
1112 			dsl_pool_user_hold_create_obj(dp, tx);
1113 			zapobj = dp->dp_tmp_userrefs_obj;
1114 		} else {
1115 			return (SET_ERROR(ENOENT));
1116 		}
1117 	}
1118 
1119 	name = kmem_asprintf("%llx-%s", (u_longlong_t)dsobj, tag);
1120 	if (holding)
1121 		error = zap_add(mos, zapobj, name, 8, 1, &now, tx);
1122 	else
1123 		error = zap_remove(mos, zapobj, name, tx);
1124 	strfree(name);
1125 
1126 	return (error);
1127 }
1128 
1129 /*
1130  * Add a temporary hold for the given dataset object and tag.
1131  */
1132 int
1133 dsl_pool_user_hold(dsl_pool_t *dp, uint64_t dsobj, const char *tag,
1134     uint64_t now, dmu_tx_t *tx)
1135 {
1136 	return (dsl_pool_user_hold_rele_impl(dp, dsobj, tag, now, tx, B_TRUE));
1137 }
1138 
1139 /*
1140  * Release a temporary hold for the given dataset object and tag.
1141  */
1142 int
1143 dsl_pool_user_release(dsl_pool_t *dp, uint64_t dsobj, const char *tag,
1144     dmu_tx_t *tx)
1145 {
1146 	return (dsl_pool_user_hold_rele_impl(dp, dsobj, tag, NULL,
1147 	    tx, B_FALSE));
1148 }
1149 
1150 /*
1151  * DSL Pool Configuration Lock
1152  *
1153  * The dp_config_rwlock protects against changes to DSL state (e.g. dataset
1154  * creation / destruction / rename / property setting).  It must be held for
1155  * read to hold a dataset or dsl_dir.  I.e. you must call
1156  * dsl_pool_config_enter() or dsl_pool_hold() before calling
1157  * dsl_{dataset,dir}_hold{_obj}.  In most circumstances, the dp_config_rwlock
1158  * must be held continuously until all datasets and dsl_dirs are released.
1159  *
1160  * The only exception to this rule is that if a "long hold" is placed on
1161  * a dataset, then the dp_config_rwlock may be dropped while the dataset
1162  * is still held.  The long hold will prevent the dataset from being
1163  * destroyed -- the destroy will fail with EBUSY.  A long hold can be
1164  * obtained by calling dsl_dataset_long_hold(), or by "owning" a dataset
1165  * (by calling dsl_{dataset,objset}_{try}own{_obj}).
1166  *
1167  * Legitimate long-holders (including owners) should be long-running, cancelable
1168  * tasks that should cause "zfs destroy" to fail.  This includes DMU
1169  * consumers (i.e. a ZPL filesystem being mounted or ZVOL being open),
1170  * "zfs send", and "zfs diff".  There are several other long-holders whose
1171  * uses are suboptimal (e.g. "zfs promote", and zil_suspend()).
1172  *
1173  * The usual formula for long-holding would be:
1174  * dsl_pool_hold()
1175  * dsl_dataset_hold()
1176  * ... perform checks ...
1177  * dsl_dataset_long_hold()
1178  * dsl_pool_rele()
1179  * ... perform long-running task ...
1180  * dsl_dataset_long_rele()
1181  * dsl_dataset_rele()
1182  *
1183  * Note that when the long hold is released, the dataset is still held but
1184  * the pool is not held.  The dataset may change arbitrarily during this time
1185  * (e.g. it could be destroyed).  Therefore you shouldn't do anything to the
1186  * dataset except release it.
1187  *
1188  * User-initiated operations (e.g. ioctls, zfs_ioc_*()) are either read-only
1189  * or modifying operations.
1190  *
1191  * Modifying operations should generally use dsl_sync_task().  The synctask
1192  * infrastructure enforces proper locking strategy with respect to the
1193  * dp_config_rwlock.  See the comment above dsl_sync_task() for details.
1194  *
1195  * Read-only operations will manually hold the pool, then the dataset, obtain
1196  * information from the dataset, then release the pool and dataset.
1197  * dmu_objset_{hold,rele}() are convenience routines that also do the pool
1198  * hold/rele.
1199  */
1200 
1201 int
1202 dsl_pool_hold(const char *name, void *tag, dsl_pool_t **dp)
1203 {
1204 	spa_t *spa;
1205 	int error;
1206 
1207 	error = spa_open(name, &spa, tag);
1208 	if (error == 0) {
1209 		*dp = spa_get_dsl(spa);
1210 		dsl_pool_config_enter(*dp, tag);
1211 	}
1212 	return (error);
1213 }
1214 
1215 void
1216 dsl_pool_rele(dsl_pool_t *dp, void *tag)
1217 {
1218 	dsl_pool_config_exit(dp, tag);
1219 	spa_close(dp->dp_spa, tag);
1220 }
1221 
1222 void
1223 dsl_pool_config_enter(dsl_pool_t *dp, void *tag)
1224 {
1225 	/*
1226 	 * We use a "reentrant" reader-writer lock, but not reentrantly.
1227 	 *
1228 	 * The rrwlock can (with the track_all flag) track all reading threads,
1229 	 * which is very useful for debugging which code path failed to release
1230 	 * the lock, and for verifying that the *current* thread does hold
1231 	 * the lock.
1232 	 *
1233 	 * (Unlike a rwlock, which knows that N threads hold it for
1234 	 * read, but not *which* threads, so rw_held(RW_READER) returns TRUE
1235 	 * if any thread holds it for read, even if this thread doesn't).
1236 	 */
1237 	ASSERT(!rrw_held(&dp->dp_config_rwlock, RW_READER));
1238 	rrw_enter(&dp->dp_config_rwlock, RW_READER, tag);
1239 }
1240 
1241 void
1242 dsl_pool_config_enter_prio(dsl_pool_t *dp, void *tag)
1243 {
1244 	ASSERT(!rrw_held(&dp->dp_config_rwlock, RW_READER));
1245 	rrw_enter_read_prio(&dp->dp_config_rwlock, tag);
1246 }
1247 
1248 void
1249 dsl_pool_config_exit(dsl_pool_t *dp, void *tag)
1250 {
1251 	rrw_exit(&dp->dp_config_rwlock, tag);
1252 }
1253 
1254 boolean_t
1255 dsl_pool_config_held(dsl_pool_t *dp)
1256 {
1257 	return (RRW_LOCK_HELD(&dp->dp_config_rwlock));
1258 }
1259 
1260 boolean_t
1261 dsl_pool_config_held_writer(dsl_pool_t *dp)
1262 {
1263 	return (RRW_WRITE_HELD(&dp->dp_config_rwlock));
1264 }
1265