xref: /titanic_51/usr/src/uts/common/fs/zfs/dsl_pool.c (revision 37714ae43602c675f9dc59b070bfdf9fa702872c)
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 2009 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 #include <sys/dsl_pool.h>
27 #include <sys/dsl_dataset.h>
28 #include <sys/dsl_dir.h>
29 #include <sys/dsl_synctask.h>
30 #include <sys/dmu_tx.h>
31 #include <sys/dmu_objset.h>
32 #include <sys/arc.h>
33 #include <sys/zap.h>
34 #include <sys/zio.h>
35 #include <sys/zfs_context.h>
36 #include <sys/fs/zfs.h>
37 #include <sys/zfs_znode.h>
38 #include <sys/spa_impl.h>
39 
40 int zfs_no_write_throttle = 0;
41 int zfs_write_limit_shift = 3;			/* 1/8th of physical memory */
42 int zfs_txg_synctime = 5;			/* target secs to sync a txg */
43 
44 uint64_t zfs_write_limit_min = 32 << 20;	/* min write limit is 32MB */
45 uint64_t zfs_write_limit_max = 0;		/* max data payload per txg */
46 uint64_t zfs_write_limit_inflated = 0;
47 uint64_t zfs_write_limit_override = 0;
48 
49 kmutex_t zfs_write_limit_lock;
50 
51 static pgcnt_t old_physmem = 0;
52 
53 static int
54 dsl_pool_open_special_dir(dsl_pool_t *dp, const char *name, dsl_dir_t **ddp)
55 {
56 	uint64_t obj;
57 	int err;
58 
59 	err = zap_lookup(dp->dp_meta_objset,
60 	    dp->dp_root_dir->dd_phys->dd_child_dir_zapobj,
61 	    name, sizeof (obj), 1, &obj);
62 	if (err)
63 		return (err);
64 
65 	return (dsl_dir_open_obj(dp, obj, name, dp, ddp));
66 }
67 
68 static dsl_pool_t *
69 dsl_pool_open_impl(spa_t *spa, uint64_t txg)
70 {
71 	dsl_pool_t *dp;
72 	blkptr_t *bp = spa_get_rootblkptr(spa);
73 
74 	dp = kmem_zalloc(sizeof (dsl_pool_t), KM_SLEEP);
75 	dp->dp_spa = spa;
76 	dp->dp_meta_rootbp = *bp;
77 	rw_init(&dp->dp_config_rwlock, NULL, RW_DEFAULT, NULL);
78 	dp->dp_write_limit = zfs_write_limit_min;
79 	txg_init(dp, txg);
80 
81 	txg_list_create(&dp->dp_dirty_datasets,
82 	    offsetof(dsl_dataset_t, ds_dirty_link));
83 	txg_list_create(&dp->dp_dirty_dirs,
84 	    offsetof(dsl_dir_t, dd_dirty_link));
85 	txg_list_create(&dp->dp_sync_tasks,
86 	    offsetof(dsl_sync_task_group_t, dstg_node));
87 	list_create(&dp->dp_synced_datasets, sizeof (dsl_dataset_t),
88 	    offsetof(dsl_dataset_t, ds_synced_link));
89 
90 	mutex_init(&dp->dp_lock, NULL, MUTEX_DEFAULT, NULL);
91 	mutex_init(&dp->dp_scrub_cancel_lock, NULL, MUTEX_DEFAULT, NULL);
92 
93 	dp->dp_vnrele_taskq = taskq_create("zfs_vn_rele_taskq", 1, minclsyspri,
94 	    1, 4, 0);
95 
96 	return (dp);
97 }
98 
99 int
100 dsl_pool_open(spa_t *spa, uint64_t txg, dsl_pool_t **dpp)
101 {
102 	int err;
103 	dsl_pool_t *dp = dsl_pool_open_impl(spa, txg);
104 	dsl_dir_t *dd;
105 	dsl_dataset_t *ds;
106 
107 	rw_enter(&dp->dp_config_rwlock, RW_WRITER);
108 	err = dmu_objset_open_impl(spa, NULL, &dp->dp_meta_rootbp,
109 	    &dp->dp_meta_objset);
110 	if (err)
111 		goto out;
112 
113 	err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
114 	    DMU_POOL_ROOT_DATASET, sizeof (uint64_t), 1,
115 	    &dp->dp_root_dir_obj);
116 	if (err)
117 		goto out;
118 
119 	err = dsl_dir_open_obj(dp, dp->dp_root_dir_obj,
120 	    NULL, dp, &dp->dp_root_dir);
121 	if (err)
122 		goto out;
123 
124 	err = dsl_pool_open_special_dir(dp, MOS_DIR_NAME, &dp->dp_mos_dir);
125 	if (err)
126 		goto out;
127 
128 	if (spa_version(spa) >= SPA_VERSION_ORIGIN) {
129 		err = dsl_pool_open_special_dir(dp, ORIGIN_DIR_NAME, &dd);
130 		if (err)
131 			goto out;
132 		err = dsl_dataset_hold_obj(dp, dd->dd_phys->dd_head_dataset_obj,
133 		    FTAG, &ds);
134 		if (err == 0) {
135 			err = dsl_dataset_hold_obj(dp,
136 			    ds->ds_phys->ds_prev_snap_obj, dp,
137 			    &dp->dp_origin_snap);
138 			dsl_dataset_rele(ds, FTAG);
139 		}
140 		dsl_dir_close(dd, dp);
141 		if (err)
142 			goto out;
143 	}
144 
145 	/* get scrub status */
146 	err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
147 	    DMU_POOL_SCRUB_FUNC, sizeof (uint32_t), 1,
148 	    &dp->dp_scrub_func);
149 	if (err == 0) {
150 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
151 		    DMU_POOL_SCRUB_QUEUE, sizeof (uint64_t), 1,
152 		    &dp->dp_scrub_queue_obj);
153 		if (err)
154 			goto out;
155 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
156 		    DMU_POOL_SCRUB_MIN_TXG, sizeof (uint64_t), 1,
157 		    &dp->dp_scrub_min_txg);
158 		if (err)
159 			goto out;
160 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
161 		    DMU_POOL_SCRUB_MAX_TXG, sizeof (uint64_t), 1,
162 		    &dp->dp_scrub_max_txg);
163 		if (err)
164 			goto out;
165 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
166 		    DMU_POOL_SCRUB_BOOKMARK, sizeof (uint64_t), 4,
167 		    &dp->dp_scrub_bookmark);
168 		if (err)
169 			goto out;
170 		err = zap_lookup(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
171 		    DMU_POOL_SCRUB_ERRORS, sizeof (uint64_t), 1,
172 		    &spa->spa_scrub_errors);
173 		if (err)
174 			goto out;
175 		if (spa_version(spa) < SPA_VERSION_DSL_SCRUB) {
176 			/*
177 			 * A new-type scrub was in progress on an old
178 			 * pool.  Restart from the beginning, since the
179 			 * old software may have changed the pool in the
180 			 * meantime.
181 			 */
182 			dsl_pool_scrub_restart(dp);
183 		}
184 	} else {
185 		/*
186 		 * It's OK if there is no scrub in progress (and if
187 		 * there was an I/O error, ignore it).
188 		 */
189 		err = 0;
190 	}
191 
192 out:
193 	rw_exit(&dp->dp_config_rwlock);
194 	if (err)
195 		dsl_pool_close(dp);
196 	else
197 		*dpp = dp;
198 
199 	return (err);
200 }
201 
202 void
203 dsl_pool_close(dsl_pool_t *dp)
204 {
205 	/* drop our references from dsl_pool_open() */
206 
207 	/*
208 	 * Since we held the origin_snap from "syncing" context (which
209 	 * includes pool-opening context), it actually only got a "ref"
210 	 * and not a hold, so just drop that here.
211 	 */
212 	if (dp->dp_origin_snap)
213 		dsl_dataset_drop_ref(dp->dp_origin_snap, dp);
214 	if (dp->dp_mos_dir)
215 		dsl_dir_close(dp->dp_mos_dir, dp);
216 	if (dp->dp_root_dir)
217 		dsl_dir_close(dp->dp_root_dir, dp);
218 
219 	/* undo the dmu_objset_open_impl(mos) from dsl_pool_open() */
220 	if (dp->dp_meta_objset)
221 		dmu_objset_evict(dp->dp_meta_objset);
222 
223 	txg_list_destroy(&dp->dp_dirty_datasets);
224 	txg_list_destroy(&dp->dp_dirty_dirs);
225 	list_destroy(&dp->dp_synced_datasets);
226 
227 	arc_flush(dp->dp_spa);
228 	txg_fini(dp);
229 	rw_destroy(&dp->dp_config_rwlock);
230 	mutex_destroy(&dp->dp_lock);
231 	mutex_destroy(&dp->dp_scrub_cancel_lock);
232 	taskq_destroy(dp->dp_vnrele_taskq);
233 	if (dp->dp_blkstats)
234 		kmem_free(dp->dp_blkstats, sizeof (zfs_all_blkstats_t));
235 	kmem_free(dp, sizeof (dsl_pool_t));
236 }
237 
238 dsl_pool_t *
239 dsl_pool_create(spa_t *spa, nvlist_t *zplprops, uint64_t txg)
240 {
241 	int err;
242 	dsl_pool_t *dp = dsl_pool_open_impl(spa, txg);
243 	dmu_tx_t *tx = dmu_tx_create_assigned(dp, txg);
244 	objset_t *os;
245 	dsl_dataset_t *ds;
246 	uint64_t dsobj;
247 
248 	/* create and open the MOS (meta-objset) */
249 	dp->dp_meta_objset = dmu_objset_create_impl(spa,
250 	    NULL, &dp->dp_meta_rootbp, DMU_OST_META, tx);
251 
252 	/* create the pool directory */
253 	err = zap_create_claim(dp->dp_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
254 	    DMU_OT_OBJECT_DIRECTORY, DMU_OT_NONE, 0, tx);
255 	ASSERT3U(err, ==, 0);
256 
257 	/* create and open the root dir */
258 	dp->dp_root_dir_obj = dsl_dir_create_sync(dp, NULL, NULL, tx);
259 	VERIFY(0 == dsl_dir_open_obj(dp, dp->dp_root_dir_obj,
260 	    NULL, dp, &dp->dp_root_dir));
261 
262 	/* create and open the meta-objset dir */
263 	(void) dsl_dir_create_sync(dp, dp->dp_root_dir, MOS_DIR_NAME, tx);
264 	VERIFY(0 == dsl_pool_open_special_dir(dp,
265 	    MOS_DIR_NAME, &dp->dp_mos_dir));
266 
267 	if (spa_version(spa) >= SPA_VERSION_DSL_SCRUB)
268 		dsl_pool_create_origin(dp, tx);
269 
270 	/* create the root dataset */
271 	dsobj = dsl_dataset_create_sync_dd(dp->dp_root_dir, NULL, 0, tx);
272 
273 	/* create the root objset */
274 	VERIFY(0 == dsl_dataset_hold_obj(dp, dsobj, FTAG, &ds));
275 	os = dmu_objset_create_impl(dp->dp_spa, ds,
276 	    dsl_dataset_get_blkptr(ds), DMU_OST_ZFS, tx);
277 #ifdef _KERNEL
278 	zfs_create_fs(os, kcred, zplprops, tx);
279 #endif
280 	dsl_dataset_rele(ds, FTAG);
281 
282 	dmu_tx_commit(tx);
283 
284 	return (dp);
285 }
286 
287 void
288 dsl_pool_sync(dsl_pool_t *dp, uint64_t txg)
289 {
290 	zio_t *zio;
291 	dmu_tx_t *tx;
292 	dsl_dir_t *dd;
293 	dsl_dataset_t *ds;
294 	dsl_sync_task_group_t *dstg;
295 	objset_t *mos = dp->dp_meta_objset;
296 	hrtime_t start, write_time;
297 	uint64_t data_written;
298 	int err;
299 
300 	tx = dmu_tx_create_assigned(dp, txg);
301 
302 	dp->dp_read_overhead = 0;
303 	start = gethrtime();
304 
305 	zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
306 	while (ds = txg_list_remove(&dp->dp_dirty_datasets, txg)) {
307 		/*
308 		 * We must not sync any non-MOS datasets twice, because
309 		 * we may have taken a snapshot of them.  However, we
310 		 * may sync newly-created datasets on pass 2.
311 		 */
312 		ASSERT(!list_link_active(&ds->ds_synced_link));
313 		list_insert_tail(&dp->dp_synced_datasets, ds);
314 		dsl_dataset_sync(ds, zio, tx);
315 	}
316 	DTRACE_PROBE(pool_sync__1setup);
317 	err = zio_wait(zio);
318 
319 	write_time = gethrtime() - start;
320 	ASSERT(err == 0);
321 	DTRACE_PROBE(pool_sync__2rootzio);
322 
323 	for (ds = list_head(&dp->dp_synced_datasets); ds;
324 	    ds = list_next(&dp->dp_synced_datasets, ds))
325 		dmu_objset_do_userquota_callbacks(ds->ds_objset, tx);
326 
327 	/*
328 	 * Sync the datasets again to push out the changes due to
329 	 * userquota updates.  This must be done before we process the
330 	 * sync tasks, because that could cause a snapshot of a dataset
331 	 * whose ds_bp will be rewritten when we do this 2nd sync.
332 	 */
333 	zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
334 	while (ds = txg_list_remove(&dp->dp_dirty_datasets, txg)) {
335 		ASSERT(list_link_active(&ds->ds_synced_link));
336 		dmu_buf_rele(ds->ds_dbuf, ds);
337 		dsl_dataset_sync(ds, zio, tx);
338 	}
339 	err = zio_wait(zio);
340 
341 	while (dstg = txg_list_remove(&dp->dp_sync_tasks, txg)) {
342 		/*
343 		 * No more sync tasks should have been added while we
344 		 * were syncing.
345 		 */
346 		ASSERT(spa_sync_pass(dp->dp_spa) == 1);
347 		dsl_sync_task_group_sync(dstg, tx);
348 	}
349 	DTRACE_PROBE(pool_sync__3task);
350 
351 	start = gethrtime();
352 	while (dd = txg_list_remove(&dp->dp_dirty_dirs, txg))
353 		dsl_dir_sync(dd, tx);
354 	write_time += gethrtime() - start;
355 
356 	if (spa_sync_pass(dp->dp_spa) == 1)
357 		dsl_pool_scrub_sync(dp, tx);
358 
359 	start = gethrtime();
360 	if (list_head(&mos->os_dirty_dnodes[txg & TXG_MASK]) != NULL ||
361 	    list_head(&mos->os_free_dnodes[txg & TXG_MASK]) != NULL) {
362 		zio = zio_root(dp->dp_spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
363 		dmu_objset_sync(mos, zio, tx);
364 		err = zio_wait(zio);
365 		ASSERT(err == 0);
366 		dprintf_bp(&dp->dp_meta_rootbp, "meta objset rootbp is %s", "");
367 		spa_set_rootblkptr(dp->dp_spa, &dp->dp_meta_rootbp);
368 	}
369 	write_time += gethrtime() - start;
370 	DTRACE_PROBE2(pool_sync__4io, hrtime_t, write_time,
371 	    hrtime_t, dp->dp_read_overhead);
372 	write_time -= dp->dp_read_overhead;
373 
374 	dmu_tx_commit(tx);
375 
376 	data_written = dp->dp_space_towrite[txg & TXG_MASK];
377 	dp->dp_space_towrite[txg & TXG_MASK] = 0;
378 	ASSERT(dp->dp_tempreserved[txg & TXG_MASK] == 0);
379 
380 	/*
381 	 * If the write limit max has not been explicitly set, set it
382 	 * to a fraction of available physical memory (default 1/8th).
383 	 * Note that we must inflate the limit because the spa
384 	 * inflates write sizes to account for data replication.
385 	 * Check this each sync phase to catch changing memory size.
386 	 */
387 	if (physmem != old_physmem && zfs_write_limit_shift) {
388 		mutex_enter(&zfs_write_limit_lock);
389 		old_physmem = physmem;
390 		zfs_write_limit_max = ptob(physmem) >> zfs_write_limit_shift;
391 		zfs_write_limit_inflated = MAX(zfs_write_limit_min,
392 		    spa_get_asize(dp->dp_spa, zfs_write_limit_max));
393 		mutex_exit(&zfs_write_limit_lock);
394 	}
395 
396 	/*
397 	 * Attempt to keep the sync time consistent by adjusting the
398 	 * amount of write traffic allowed into each transaction group.
399 	 * Weight the throughput calculation towards the current value:
400 	 * 	thru = 3/4 old_thru + 1/4 new_thru
401 	 */
402 	ASSERT(zfs_write_limit_min > 0);
403 	if (data_written > zfs_write_limit_min / 8 && write_time > 0) {
404 		uint64_t throughput = (data_written * NANOSEC) / write_time;
405 		if (dp->dp_throughput)
406 			dp->dp_throughput = throughput / 4 +
407 			    3 * dp->dp_throughput / 4;
408 		else
409 			dp->dp_throughput = throughput;
410 		dp->dp_write_limit = MIN(zfs_write_limit_inflated,
411 		    MAX(zfs_write_limit_min,
412 		    dp->dp_throughput * zfs_txg_synctime));
413 	}
414 }
415 
416 void
417 dsl_pool_zil_clean(dsl_pool_t *dp)
418 {
419 	dsl_dataset_t *ds;
420 
421 	while (ds = list_head(&dp->dp_synced_datasets)) {
422 		list_remove(&dp->dp_synced_datasets, ds);
423 		ASSERT(ds->ds_objset != NULL);
424 		zil_clean(ds->ds_objset->os_zil);
425 		dmu_buf_rele(ds->ds_dbuf, ds);
426 	}
427 }
428 
429 /*
430  * TRUE if the current thread is the tx_sync_thread or if we
431  * are being called from SPA context during pool initialization.
432  */
433 int
434 dsl_pool_sync_context(dsl_pool_t *dp)
435 {
436 	return (curthread == dp->dp_tx.tx_sync_thread ||
437 	    spa_get_dsl(dp->dp_spa) == NULL);
438 }
439 
440 uint64_t
441 dsl_pool_adjustedsize(dsl_pool_t *dp, boolean_t netfree)
442 {
443 	uint64_t space, resv;
444 
445 	/*
446 	 * Reserve about 1.6% (1/64), or at least 32MB, for allocation
447 	 * efficiency.
448 	 * XXX The intent log is not accounted for, so it must fit
449 	 * within this slop.
450 	 *
451 	 * If we're trying to assess whether it's OK to do a free,
452 	 * cut the reservation in half to allow forward progress
453 	 * (e.g. make it possible to rm(1) files from a full pool).
454 	 */
455 	space = spa_get_dspace(dp->dp_spa);
456 	resv = MAX(space >> 6, SPA_MINDEVSIZE >> 1);
457 	if (netfree)
458 		resv >>= 1;
459 
460 	return (space - resv);
461 }
462 
463 int
464 dsl_pool_tempreserve_space(dsl_pool_t *dp, uint64_t space, dmu_tx_t *tx)
465 {
466 	uint64_t reserved = 0;
467 	uint64_t write_limit = (zfs_write_limit_override ?
468 	    zfs_write_limit_override : dp->dp_write_limit);
469 
470 	if (zfs_no_write_throttle) {
471 		atomic_add_64(&dp->dp_tempreserved[tx->tx_txg & TXG_MASK],
472 		    space);
473 		return (0);
474 	}
475 
476 	/*
477 	 * Check to see if we have exceeded the maximum allowed IO for
478 	 * this transaction group.  We can do this without locks since
479 	 * a little slop here is ok.  Note that we do the reserved check
480 	 * with only half the requested reserve: this is because the
481 	 * reserve requests are worst-case, and we really don't want to
482 	 * throttle based off of worst-case estimates.
483 	 */
484 	if (write_limit > 0) {
485 		reserved = dp->dp_space_towrite[tx->tx_txg & TXG_MASK]
486 		    + dp->dp_tempreserved[tx->tx_txg & TXG_MASK] / 2;
487 
488 		if (reserved && reserved > write_limit)
489 			return (ERESTART);
490 	}
491 
492 	atomic_add_64(&dp->dp_tempreserved[tx->tx_txg & TXG_MASK], space);
493 
494 	/*
495 	 * If this transaction group is over 7/8ths capacity, delay
496 	 * the caller 1 clock tick.  This will slow down the "fill"
497 	 * rate until the sync process can catch up with us.
498 	 */
499 	if (reserved && reserved > (write_limit - (write_limit >> 3)))
500 		txg_delay(dp, tx->tx_txg, 1);
501 
502 	return (0);
503 }
504 
505 void
506 dsl_pool_tempreserve_clear(dsl_pool_t *dp, int64_t space, dmu_tx_t *tx)
507 {
508 	ASSERT(dp->dp_tempreserved[tx->tx_txg & TXG_MASK] >= space);
509 	atomic_add_64(&dp->dp_tempreserved[tx->tx_txg & TXG_MASK], -space);
510 }
511 
512 void
513 dsl_pool_memory_pressure(dsl_pool_t *dp)
514 {
515 	uint64_t space_inuse = 0;
516 	int i;
517 
518 	if (dp->dp_write_limit == zfs_write_limit_min)
519 		return;
520 
521 	for (i = 0; i < TXG_SIZE; i++) {
522 		space_inuse += dp->dp_space_towrite[i];
523 		space_inuse += dp->dp_tempreserved[i];
524 	}
525 	dp->dp_write_limit = MAX(zfs_write_limit_min,
526 	    MIN(dp->dp_write_limit, space_inuse / 4));
527 }
528 
529 void
530 dsl_pool_willuse_space(dsl_pool_t *dp, int64_t space, dmu_tx_t *tx)
531 {
532 	if (space > 0) {
533 		mutex_enter(&dp->dp_lock);
534 		dp->dp_space_towrite[tx->tx_txg & TXG_MASK] += space;
535 		mutex_exit(&dp->dp_lock);
536 	}
537 }
538 
539 /* ARGSUSED */
540 static int
541 upgrade_clones_cb(spa_t *spa, uint64_t dsobj, const char *dsname, void *arg)
542 {
543 	dmu_tx_t *tx = arg;
544 	dsl_dataset_t *ds, *prev = NULL;
545 	int err;
546 	dsl_pool_t *dp = spa_get_dsl(spa);
547 
548 	err = dsl_dataset_hold_obj(dp, dsobj, FTAG, &ds);
549 	if (err)
550 		return (err);
551 
552 	while (ds->ds_phys->ds_prev_snap_obj != 0) {
553 		err = dsl_dataset_hold_obj(dp, ds->ds_phys->ds_prev_snap_obj,
554 		    FTAG, &prev);
555 		if (err) {
556 			dsl_dataset_rele(ds, FTAG);
557 			return (err);
558 		}
559 
560 		if (prev->ds_phys->ds_next_snap_obj != ds->ds_object)
561 			break;
562 		dsl_dataset_rele(ds, FTAG);
563 		ds = prev;
564 		prev = NULL;
565 	}
566 
567 	if (prev == NULL) {
568 		prev = dp->dp_origin_snap;
569 
570 		/*
571 		 * The $ORIGIN can't have any data, or the accounting
572 		 * will be wrong.
573 		 */
574 		ASSERT(prev->ds_phys->ds_bp.blk_birth == 0);
575 
576 		/* The origin doesn't get attached to itself */
577 		if (ds->ds_object == prev->ds_object) {
578 			dsl_dataset_rele(ds, FTAG);
579 			return (0);
580 		}
581 
582 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
583 		ds->ds_phys->ds_prev_snap_obj = prev->ds_object;
584 		ds->ds_phys->ds_prev_snap_txg = prev->ds_phys->ds_creation_txg;
585 
586 		dmu_buf_will_dirty(ds->ds_dir->dd_dbuf, tx);
587 		ds->ds_dir->dd_phys->dd_origin_obj = prev->ds_object;
588 
589 		dmu_buf_will_dirty(prev->ds_dbuf, tx);
590 		prev->ds_phys->ds_num_children++;
591 
592 		if (ds->ds_phys->ds_next_snap_obj == 0) {
593 			ASSERT(ds->ds_prev == NULL);
594 			VERIFY(0 == dsl_dataset_hold_obj(dp,
595 			    ds->ds_phys->ds_prev_snap_obj, ds, &ds->ds_prev));
596 		}
597 	}
598 
599 	ASSERT(ds->ds_dir->dd_phys->dd_origin_obj == prev->ds_object);
600 	ASSERT(ds->ds_phys->ds_prev_snap_obj == prev->ds_object);
601 
602 	if (prev->ds_phys->ds_next_clones_obj == 0) {
603 		prev->ds_phys->ds_next_clones_obj =
604 		    zap_create(dp->dp_meta_objset,
605 		    DMU_OT_NEXT_CLONES, DMU_OT_NONE, 0, tx);
606 	}
607 	VERIFY(0 == zap_add_int(dp->dp_meta_objset,
608 	    prev->ds_phys->ds_next_clones_obj, ds->ds_object, tx));
609 
610 	dsl_dataset_rele(ds, FTAG);
611 	if (prev != dp->dp_origin_snap)
612 		dsl_dataset_rele(prev, FTAG);
613 	return (0);
614 }
615 
616 void
617 dsl_pool_upgrade_clones(dsl_pool_t *dp, dmu_tx_t *tx)
618 {
619 	ASSERT(dmu_tx_is_syncing(tx));
620 	ASSERT(dp->dp_origin_snap != NULL);
621 
622 	(void) dmu_objset_find_spa(dp->dp_spa, NULL, upgrade_clones_cb,
623 	    tx, DS_FIND_CHILDREN);
624 }
625 
626 void
627 dsl_pool_create_origin(dsl_pool_t *dp, dmu_tx_t *tx)
628 {
629 	uint64_t dsobj;
630 	dsl_dataset_t *ds;
631 
632 	ASSERT(dmu_tx_is_syncing(tx));
633 	ASSERT(dp->dp_origin_snap == NULL);
634 
635 	/* create the origin dir, ds, & snap-ds */
636 	rw_enter(&dp->dp_config_rwlock, RW_WRITER);
637 	dsobj = dsl_dataset_create_sync(dp->dp_root_dir, ORIGIN_DIR_NAME,
638 	    NULL, 0, kcred, tx);
639 	VERIFY(0 == dsl_dataset_hold_obj(dp, dsobj, FTAG, &ds));
640 	dsl_dataset_snapshot_sync(ds, ORIGIN_DIR_NAME, kcred, tx);
641 	VERIFY(0 == dsl_dataset_hold_obj(dp, ds->ds_phys->ds_prev_snap_obj,
642 	    dp, &dp->dp_origin_snap));
643 	dsl_dataset_rele(ds, FTAG);
644 	rw_exit(&dp->dp_config_rwlock);
645 }
646 
647 taskq_t *
648 dsl_pool_vnrele_taskq(dsl_pool_t *dp)
649 {
650 	return (dp->dp_vnrele_taskq);
651 }
652