xref: /illumos-gate/usr/src/uts/common/fs/zfs/dmu.c (revision b7edcb94)
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  */
25 /* Copyright (c) 2013 by Saso Kiselkov. All rights reserved. */
26 /* Copyright (c) 2013, Joyent, Inc. All rights reserved. */
27 /* Copyright 2016 Nexenta Systems, Inc. All rights reserved. */
28 
29 #include <sys/dmu.h>
30 #include <sys/dmu_impl.h>
31 #include <sys/dmu_tx.h>
32 #include <sys/dbuf.h>
33 #include <sys/dnode.h>
34 #include <sys/zfs_context.h>
35 #include <sys/dmu_objset.h>
36 #include <sys/dmu_traverse.h>
37 #include <sys/dsl_dataset.h>
38 #include <sys/dsl_dir.h>
39 #include <sys/dsl_pool.h>
40 #include <sys/dsl_synctask.h>
41 #include <sys/dsl_prop.h>
42 #include <sys/dmu_zfetch.h>
43 #include <sys/zfs_ioctl.h>
44 #include <sys/zap.h>
45 #include <sys/zio_checksum.h>
46 #include <sys/zio_compress.h>
47 #include <sys/sa.h>
48 #include <sys/zfeature.h>
49 #include <sys/abd.h>
50 #ifdef _KERNEL
51 #include <sys/vmsystm.h>
52 #include <sys/zfs_znode.h>
53 #endif
54 
55 /*
56  * Enable/disable nopwrite feature.
57  */
58 int zfs_nopwrite_enabled = 1;
59 
60 /*
61  * Tunable to control percentage of dirtied blocks from frees in one TXG.
62  * After this threshold is crossed, additional dirty blocks from frees
63  * wait until the next TXG.
64  * A value of zero will disable this throttle.
65  */
66 uint32_t zfs_per_txg_dirty_frees_percent = 30;
67 
68 const dmu_object_type_info_t dmu_ot[DMU_OT_NUMTYPES] = {
69 	{	DMU_BSWAP_UINT8,	TRUE,	"unallocated"		},
70 	{	DMU_BSWAP_ZAP,		TRUE,	"object directory"	},
71 	{	DMU_BSWAP_UINT64,	TRUE,	"object array"		},
72 	{	DMU_BSWAP_UINT8,	TRUE,	"packed nvlist"		},
73 	{	DMU_BSWAP_UINT64,	TRUE,	"packed nvlist size"	},
74 	{	DMU_BSWAP_UINT64,	TRUE,	"bpobj"			},
75 	{	DMU_BSWAP_UINT64,	TRUE,	"bpobj header"		},
76 	{	DMU_BSWAP_UINT64,	TRUE,	"SPA space map header"	},
77 	{	DMU_BSWAP_UINT64,	TRUE,	"SPA space map"		},
78 	{	DMU_BSWAP_UINT64,	TRUE,	"ZIL intent log"	},
79 	{	DMU_BSWAP_DNODE,	TRUE,	"DMU dnode"		},
80 	{	DMU_BSWAP_OBJSET,	TRUE,	"DMU objset"		},
81 	{	DMU_BSWAP_UINT64,	TRUE,	"DSL directory"		},
82 	{	DMU_BSWAP_ZAP,		TRUE,	"DSL directory child map"},
83 	{	DMU_BSWAP_ZAP,		TRUE,	"DSL dataset snap map"	},
84 	{	DMU_BSWAP_ZAP,		TRUE,	"DSL props"		},
85 	{	DMU_BSWAP_UINT64,	TRUE,	"DSL dataset"		},
86 	{	DMU_BSWAP_ZNODE,	TRUE,	"ZFS znode"		},
87 	{	DMU_BSWAP_OLDACL,	TRUE,	"ZFS V0 ACL"		},
88 	{	DMU_BSWAP_UINT8,	FALSE,	"ZFS plain file"	},
89 	{	DMU_BSWAP_ZAP,		TRUE,	"ZFS directory"		},
90 	{	DMU_BSWAP_ZAP,		TRUE,	"ZFS master node"	},
91 	{	DMU_BSWAP_ZAP,		TRUE,	"ZFS delete queue"	},
92 	{	DMU_BSWAP_UINT8,	FALSE,	"zvol object"		},
93 	{	DMU_BSWAP_ZAP,		TRUE,	"zvol prop"		},
94 	{	DMU_BSWAP_UINT8,	FALSE,	"other uint8[]"		},
95 	{	DMU_BSWAP_UINT64,	FALSE,	"other uint64[]"	},
96 	{	DMU_BSWAP_ZAP,		TRUE,	"other ZAP"		},
97 	{	DMU_BSWAP_ZAP,		TRUE,	"persistent error log"	},
98 	{	DMU_BSWAP_UINT8,	TRUE,	"SPA history"		},
99 	{	DMU_BSWAP_UINT64,	TRUE,	"SPA history offsets"	},
100 	{	DMU_BSWAP_ZAP,		TRUE,	"Pool properties"	},
101 	{	DMU_BSWAP_ZAP,		TRUE,	"DSL permissions"	},
102 	{	DMU_BSWAP_ACL,		TRUE,	"ZFS ACL"		},
103 	{	DMU_BSWAP_UINT8,	TRUE,	"ZFS SYSACL"		},
104 	{	DMU_BSWAP_UINT8,	TRUE,	"FUID table"		},
105 	{	DMU_BSWAP_UINT64,	TRUE,	"FUID table size"	},
106 	{	DMU_BSWAP_ZAP,		TRUE,	"DSL dataset next clones"},
107 	{	DMU_BSWAP_ZAP,		TRUE,	"scan work queue"	},
108 	{	DMU_BSWAP_ZAP,		TRUE,	"ZFS user/group used"	},
109 	{	DMU_BSWAP_ZAP,		TRUE,	"ZFS user/group quota"	},
110 	{	DMU_BSWAP_ZAP,		TRUE,	"snapshot refcount tags"},
111 	{	DMU_BSWAP_ZAP,		TRUE,	"DDT ZAP algorithm"	},
112 	{	DMU_BSWAP_ZAP,		TRUE,	"DDT statistics"	},
113 	{	DMU_BSWAP_UINT8,	TRUE,	"System attributes"	},
114 	{	DMU_BSWAP_ZAP,		TRUE,	"SA master node"	},
115 	{	DMU_BSWAP_ZAP,		TRUE,	"SA attr registration"	},
116 	{	DMU_BSWAP_ZAP,		TRUE,	"SA attr layouts"	},
117 	{	DMU_BSWAP_ZAP,		TRUE,	"scan translations"	},
118 	{	DMU_BSWAP_UINT8,	FALSE,	"deduplicated block"	},
119 	{	DMU_BSWAP_ZAP,		TRUE,	"DSL deadlist map"	},
120 	{	DMU_BSWAP_UINT64,	TRUE,	"DSL deadlist map hdr"	},
121 	{	DMU_BSWAP_ZAP,		TRUE,	"DSL dir clones"	},
122 	{	DMU_BSWAP_UINT64,	TRUE,	"bpobj subobj"		}
123 };
124 
125 const dmu_object_byteswap_info_t dmu_ot_byteswap[DMU_BSWAP_NUMFUNCS] = {
126 	{	byteswap_uint8_array,	"uint8"		},
127 	{	byteswap_uint16_array,	"uint16"	},
128 	{	byteswap_uint32_array,	"uint32"	},
129 	{	byteswap_uint64_array,	"uint64"	},
130 	{	zap_byteswap,		"zap"		},
131 	{	dnode_buf_byteswap,	"dnode"		},
132 	{	dmu_objset_byteswap,	"objset"	},
133 	{	zfs_znode_byteswap,	"znode"		},
134 	{	zfs_oldacl_byteswap,	"oldacl"	},
135 	{	zfs_acl_byteswap,	"acl"		}
136 };
137 
138 int
139 dmu_buf_hold_noread_by_dnode(dnode_t *dn, uint64_t offset,
140     void *tag, dmu_buf_t **dbp)
141 {
142 	uint64_t blkid;
143 	dmu_buf_impl_t *db;
144 
145 	blkid = dbuf_whichblock(dn, 0, offset);
146 	rw_enter(&dn->dn_struct_rwlock, RW_READER);
147 	db = dbuf_hold(dn, blkid, tag);
148 	rw_exit(&dn->dn_struct_rwlock);
149 
150 	if (db == NULL) {
151 		*dbp = NULL;
152 		return (SET_ERROR(EIO));
153 	}
154 
155 	*dbp = &db->db;
156 	return (0);
157 }
158 int
159 dmu_buf_hold_noread(objset_t *os, uint64_t object, uint64_t offset,
160     void *tag, dmu_buf_t **dbp)
161 {
162 	dnode_t *dn;
163 	uint64_t blkid;
164 	dmu_buf_impl_t *db;
165 	int err;
166 
167 	err = dnode_hold(os, object, FTAG, &dn);
168 	if (err)
169 		return (err);
170 	blkid = dbuf_whichblock(dn, 0, offset);
171 	rw_enter(&dn->dn_struct_rwlock, RW_READER);
172 	db = dbuf_hold(dn, blkid, tag);
173 	rw_exit(&dn->dn_struct_rwlock);
174 	dnode_rele(dn, FTAG);
175 
176 	if (db == NULL) {
177 		*dbp = NULL;
178 		return (SET_ERROR(EIO));
179 	}
180 
181 	*dbp = &db->db;
182 	return (err);
183 }
184 
185 int
186 dmu_buf_hold_by_dnode(dnode_t *dn, uint64_t offset,
187     void *tag, dmu_buf_t **dbp, int flags)
188 {
189 	int err;
190 	int db_flags = DB_RF_CANFAIL;
191 
192 	if (flags & DMU_READ_NO_PREFETCH)
193 		db_flags |= DB_RF_NOPREFETCH;
194 
195 	err = dmu_buf_hold_noread_by_dnode(dn, offset, tag, dbp);
196 	if (err == 0) {
197 		dmu_buf_impl_t *db = (dmu_buf_impl_t *)(*dbp);
198 		err = dbuf_read(db, NULL, db_flags);
199 		if (err != 0) {
200 			dbuf_rele(db, tag);
201 			*dbp = NULL;
202 		}
203 	}
204 
205 	return (err);
206 }
207 
208 int
209 dmu_buf_hold(objset_t *os, uint64_t object, uint64_t offset,
210     void *tag, dmu_buf_t **dbp, int flags)
211 {
212 	int err;
213 	int db_flags = DB_RF_CANFAIL;
214 
215 	if (flags & DMU_READ_NO_PREFETCH)
216 		db_flags |= DB_RF_NOPREFETCH;
217 
218 	err = dmu_buf_hold_noread(os, object, offset, tag, dbp);
219 	if (err == 0) {
220 		dmu_buf_impl_t *db = (dmu_buf_impl_t *)(*dbp);
221 		err = dbuf_read(db, NULL, db_flags);
222 		if (err != 0) {
223 			dbuf_rele(db, tag);
224 			*dbp = NULL;
225 		}
226 	}
227 
228 	return (err);
229 }
230 
231 int
232 dmu_bonus_max(void)
233 {
234 	return (DN_MAX_BONUSLEN);
235 }
236 
237 int
238 dmu_set_bonus(dmu_buf_t *db_fake, int newsize, dmu_tx_t *tx)
239 {
240 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
241 	dnode_t *dn;
242 	int error;
243 
244 	DB_DNODE_ENTER(db);
245 	dn = DB_DNODE(db);
246 
247 	if (dn->dn_bonus != db) {
248 		error = SET_ERROR(EINVAL);
249 	} else if (newsize < 0 || newsize > db_fake->db_size) {
250 		error = SET_ERROR(EINVAL);
251 	} else {
252 		dnode_setbonuslen(dn, newsize, tx);
253 		error = 0;
254 	}
255 
256 	DB_DNODE_EXIT(db);
257 	return (error);
258 }
259 
260 int
261 dmu_set_bonustype(dmu_buf_t *db_fake, dmu_object_type_t type, dmu_tx_t *tx)
262 {
263 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
264 	dnode_t *dn;
265 	int error;
266 
267 	DB_DNODE_ENTER(db);
268 	dn = DB_DNODE(db);
269 
270 	if (!DMU_OT_IS_VALID(type)) {
271 		error = SET_ERROR(EINVAL);
272 	} else if (dn->dn_bonus != db) {
273 		error = SET_ERROR(EINVAL);
274 	} else {
275 		dnode_setbonus_type(dn, type, tx);
276 		error = 0;
277 	}
278 
279 	DB_DNODE_EXIT(db);
280 	return (error);
281 }
282 
283 dmu_object_type_t
284 dmu_get_bonustype(dmu_buf_t *db_fake)
285 {
286 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
287 	dnode_t *dn;
288 	dmu_object_type_t type;
289 
290 	DB_DNODE_ENTER(db);
291 	dn = DB_DNODE(db);
292 	type = dn->dn_bonustype;
293 	DB_DNODE_EXIT(db);
294 
295 	return (type);
296 }
297 
298 int
299 dmu_rm_spill(objset_t *os, uint64_t object, dmu_tx_t *tx)
300 {
301 	dnode_t *dn;
302 	int error;
303 
304 	error = dnode_hold(os, object, FTAG, &dn);
305 	dbuf_rm_spill(dn, tx);
306 	rw_enter(&dn->dn_struct_rwlock, RW_WRITER);
307 	dnode_rm_spill(dn, tx);
308 	rw_exit(&dn->dn_struct_rwlock);
309 	dnode_rele(dn, FTAG);
310 	return (error);
311 }
312 
313 /*
314  * returns ENOENT, EIO, or 0.
315  */
316 int
317 dmu_bonus_hold(objset_t *os, uint64_t object, void *tag, dmu_buf_t **dbp)
318 {
319 	dnode_t *dn;
320 	dmu_buf_impl_t *db;
321 	int error;
322 
323 	error = dnode_hold(os, object, FTAG, &dn);
324 	if (error)
325 		return (error);
326 
327 	rw_enter(&dn->dn_struct_rwlock, RW_READER);
328 	if (dn->dn_bonus == NULL) {
329 		rw_exit(&dn->dn_struct_rwlock);
330 		rw_enter(&dn->dn_struct_rwlock, RW_WRITER);
331 		if (dn->dn_bonus == NULL)
332 			dbuf_create_bonus(dn);
333 	}
334 	db = dn->dn_bonus;
335 
336 	/* as long as the bonus buf is held, the dnode will be held */
337 	if (refcount_add(&db->db_holds, tag) == 1) {
338 		VERIFY(dnode_add_ref(dn, db));
339 		atomic_inc_32(&dn->dn_dbufs_count);
340 	}
341 
342 	/*
343 	 * Wait to drop dn_struct_rwlock until after adding the bonus dbuf's
344 	 * hold and incrementing the dbuf count to ensure that dnode_move() sees
345 	 * a dnode hold for every dbuf.
346 	 */
347 	rw_exit(&dn->dn_struct_rwlock);
348 
349 	dnode_rele(dn, FTAG);
350 
351 	VERIFY(0 == dbuf_read(db, NULL, DB_RF_MUST_SUCCEED | DB_RF_NOPREFETCH));
352 
353 	*dbp = &db->db;
354 	return (0);
355 }
356 
357 /*
358  * returns ENOENT, EIO, or 0.
359  *
360  * This interface will allocate a blank spill dbuf when a spill blk
361  * doesn't already exist on the dnode.
362  *
363  * if you only want to find an already existing spill db, then
364  * dmu_spill_hold_existing() should be used.
365  */
366 int
367 dmu_spill_hold_by_dnode(dnode_t *dn, uint32_t flags, void *tag, dmu_buf_t **dbp)
368 {
369 	dmu_buf_impl_t *db = NULL;
370 	int err;
371 
372 	if ((flags & DB_RF_HAVESTRUCT) == 0)
373 		rw_enter(&dn->dn_struct_rwlock, RW_READER);
374 
375 	db = dbuf_hold(dn, DMU_SPILL_BLKID, tag);
376 
377 	if ((flags & DB_RF_HAVESTRUCT) == 0)
378 		rw_exit(&dn->dn_struct_rwlock);
379 
380 	ASSERT(db != NULL);
381 	err = dbuf_read(db, NULL, flags);
382 	if (err == 0)
383 		*dbp = &db->db;
384 	else
385 		dbuf_rele(db, tag);
386 	return (err);
387 }
388 
389 int
390 dmu_spill_hold_existing(dmu_buf_t *bonus, void *tag, dmu_buf_t **dbp)
391 {
392 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)bonus;
393 	dnode_t *dn;
394 	int err;
395 
396 	DB_DNODE_ENTER(db);
397 	dn = DB_DNODE(db);
398 
399 	if (spa_version(dn->dn_objset->os_spa) < SPA_VERSION_SA) {
400 		err = SET_ERROR(EINVAL);
401 	} else {
402 		rw_enter(&dn->dn_struct_rwlock, RW_READER);
403 
404 		if (!dn->dn_have_spill) {
405 			err = SET_ERROR(ENOENT);
406 		} else {
407 			err = dmu_spill_hold_by_dnode(dn,
408 			    DB_RF_HAVESTRUCT | DB_RF_CANFAIL, tag, dbp);
409 		}
410 
411 		rw_exit(&dn->dn_struct_rwlock);
412 	}
413 
414 	DB_DNODE_EXIT(db);
415 	return (err);
416 }
417 
418 int
419 dmu_spill_hold_by_bonus(dmu_buf_t *bonus, void *tag, dmu_buf_t **dbp)
420 {
421 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)bonus;
422 	dnode_t *dn;
423 	int err;
424 
425 	DB_DNODE_ENTER(db);
426 	dn = DB_DNODE(db);
427 	err = dmu_spill_hold_by_dnode(dn, DB_RF_CANFAIL, tag, dbp);
428 	DB_DNODE_EXIT(db);
429 
430 	return (err);
431 }
432 
433 /*
434  * Note: longer-term, we should modify all of the dmu_buf_*() interfaces
435  * to take a held dnode rather than <os, object> -- the lookup is wasteful,
436  * and can induce severe lock contention when writing to several files
437  * whose dnodes are in the same block.
438  */
439 static int
440 dmu_buf_hold_array_by_dnode(dnode_t *dn, uint64_t offset, uint64_t length,
441     boolean_t read, void *tag, int *numbufsp, dmu_buf_t ***dbpp, uint32_t flags)
442 {
443 	dmu_buf_t **dbp;
444 	uint64_t blkid, nblks, i;
445 	uint32_t dbuf_flags;
446 	int err;
447 	zio_t *zio;
448 
449 	ASSERT(length <= DMU_MAX_ACCESS);
450 
451 	/*
452 	 * Note: We directly notify the prefetch code of this read, so that
453 	 * we can tell it about the multi-block read.  dbuf_read() only knows
454 	 * about the one block it is accessing.
455 	 */
456 	dbuf_flags = DB_RF_CANFAIL | DB_RF_NEVERWAIT | DB_RF_HAVESTRUCT |
457 	    DB_RF_NOPREFETCH;
458 
459 	rw_enter(&dn->dn_struct_rwlock, RW_READER);
460 	if (dn->dn_datablkshift) {
461 		int blkshift = dn->dn_datablkshift;
462 		nblks = (P2ROUNDUP(offset + length, 1ULL << blkshift) -
463 		    P2ALIGN(offset, 1ULL << blkshift)) >> blkshift;
464 	} else {
465 		if (offset + length > dn->dn_datablksz) {
466 			zfs_panic_recover("zfs: accessing past end of object "
467 			    "%llx/%llx (size=%u access=%llu+%llu)",
468 			    (longlong_t)dn->dn_objset->
469 			    os_dsl_dataset->ds_object,
470 			    (longlong_t)dn->dn_object, dn->dn_datablksz,
471 			    (longlong_t)offset, (longlong_t)length);
472 			rw_exit(&dn->dn_struct_rwlock);
473 			return (SET_ERROR(EIO));
474 		}
475 		nblks = 1;
476 	}
477 	dbp = kmem_zalloc(sizeof (dmu_buf_t *) * nblks, KM_SLEEP);
478 
479 	zio = zio_root(dn->dn_objset->os_spa, NULL, NULL, ZIO_FLAG_CANFAIL);
480 	blkid = dbuf_whichblock(dn, 0, offset);
481 	for (i = 0; i < nblks; i++) {
482 		dmu_buf_impl_t *db = dbuf_hold(dn, blkid + i, tag);
483 		if (db == NULL) {
484 			rw_exit(&dn->dn_struct_rwlock);
485 			dmu_buf_rele_array(dbp, nblks, tag);
486 			zio_nowait(zio);
487 			return (SET_ERROR(EIO));
488 		}
489 
490 		/* initiate async i/o */
491 		if (read)
492 			(void) dbuf_read(db, zio, dbuf_flags);
493 		dbp[i] = &db->db;
494 	}
495 
496 	if ((flags & DMU_READ_NO_PREFETCH) == 0 &&
497 	    DNODE_META_IS_CACHEABLE(dn) && length <= zfetch_array_rd_sz) {
498 		dmu_zfetch(&dn->dn_zfetch, blkid, nblks,
499 		    read && DNODE_IS_CACHEABLE(dn));
500 	}
501 	rw_exit(&dn->dn_struct_rwlock);
502 
503 	/* wait for async i/o */
504 	err = zio_wait(zio);
505 	if (err) {
506 		dmu_buf_rele_array(dbp, nblks, tag);
507 		return (err);
508 	}
509 
510 	/* wait for other io to complete */
511 	if (read) {
512 		for (i = 0; i < nblks; i++) {
513 			dmu_buf_impl_t *db = (dmu_buf_impl_t *)dbp[i];
514 			mutex_enter(&db->db_mtx);
515 			while (db->db_state == DB_READ ||
516 			    db->db_state == DB_FILL)
517 				cv_wait(&db->db_changed, &db->db_mtx);
518 			if (db->db_state == DB_UNCACHED)
519 				err = SET_ERROR(EIO);
520 			mutex_exit(&db->db_mtx);
521 			if (err) {
522 				dmu_buf_rele_array(dbp, nblks, tag);
523 				return (err);
524 			}
525 		}
526 	}
527 
528 	*numbufsp = nblks;
529 	*dbpp = dbp;
530 	return (0);
531 }
532 
533 static int
534 dmu_buf_hold_array(objset_t *os, uint64_t object, uint64_t offset,
535     uint64_t length, int read, void *tag, int *numbufsp, dmu_buf_t ***dbpp)
536 {
537 	dnode_t *dn;
538 	int err;
539 
540 	err = dnode_hold(os, object, FTAG, &dn);
541 	if (err)
542 		return (err);
543 
544 	err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,
545 	    numbufsp, dbpp, DMU_READ_PREFETCH);
546 
547 	dnode_rele(dn, FTAG);
548 
549 	return (err);
550 }
551 
552 int
553 dmu_buf_hold_array_by_bonus(dmu_buf_t *db_fake, uint64_t offset,
554     uint64_t length, boolean_t read, void *tag, int *numbufsp,
555     dmu_buf_t ***dbpp)
556 {
557 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
558 	dnode_t *dn;
559 	int err;
560 
561 	DB_DNODE_ENTER(db);
562 	dn = DB_DNODE(db);
563 	err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,
564 	    numbufsp, dbpp, DMU_READ_PREFETCH);
565 	DB_DNODE_EXIT(db);
566 
567 	return (err);
568 }
569 
570 void
571 dmu_buf_rele_array(dmu_buf_t **dbp_fake, int numbufs, void *tag)
572 {
573 	int i;
574 	dmu_buf_impl_t **dbp = (dmu_buf_impl_t **)dbp_fake;
575 
576 	if (numbufs == 0)
577 		return;
578 
579 	for (i = 0; i < numbufs; i++) {
580 		if (dbp[i])
581 			dbuf_rele(dbp[i], tag);
582 	}
583 
584 	kmem_free(dbp, sizeof (dmu_buf_t *) * numbufs);
585 }
586 
587 /*
588  * Issue prefetch i/os for the given blocks.  If level is greater than 0, the
589  * indirect blocks prefeteched will be those that point to the blocks containing
590  * the data starting at offset, and continuing to offset + len.
591  *
592  * Note that if the indirect blocks above the blocks being prefetched are not in
593  * cache, they will be asychronously read in.
594  */
595 void
596 dmu_prefetch(objset_t *os, uint64_t object, int64_t level, uint64_t offset,
597     uint64_t len, zio_priority_t pri)
598 {
599 	dnode_t *dn;
600 	uint64_t blkid;
601 	int nblks, err;
602 
603 	if (len == 0) {  /* they're interested in the bonus buffer */
604 		dn = DMU_META_DNODE(os);
605 
606 		if (object == 0 || object >= DN_MAX_OBJECT)
607 			return;
608 
609 		rw_enter(&dn->dn_struct_rwlock, RW_READER);
610 		blkid = dbuf_whichblock(dn, level,
611 		    object * sizeof (dnode_phys_t));
612 		dbuf_prefetch(dn, level, blkid, pri, 0);
613 		rw_exit(&dn->dn_struct_rwlock);
614 		return;
615 	}
616 
617 	/*
618 	 * XXX - Note, if the dnode for the requested object is not
619 	 * already cached, we will do a *synchronous* read in the
620 	 * dnode_hold() call.  The same is true for any indirects.
621 	 */
622 	err = dnode_hold(os, object, FTAG, &dn);
623 	if (err != 0)
624 		return;
625 
626 	rw_enter(&dn->dn_struct_rwlock, RW_READER);
627 	/*
628 	 * offset + len - 1 is the last byte we want to prefetch for, and offset
629 	 * is the first.  Then dbuf_whichblk(dn, level, off + len - 1) is the
630 	 * last block we want to prefetch, and dbuf_whichblock(dn, level,
631 	 * offset)  is the first.  Then the number we need to prefetch is the
632 	 * last - first + 1.
633 	 */
634 	if (level > 0 || dn->dn_datablkshift != 0) {
635 		nblks = dbuf_whichblock(dn, level, offset + len - 1) -
636 		    dbuf_whichblock(dn, level, offset) + 1;
637 	} else {
638 		nblks = (offset < dn->dn_datablksz);
639 	}
640 
641 	if (nblks != 0) {
642 		blkid = dbuf_whichblock(dn, level, offset);
643 		for (int i = 0; i < nblks; i++)
644 			dbuf_prefetch(dn, level, blkid + i, pri, 0);
645 	}
646 
647 	rw_exit(&dn->dn_struct_rwlock);
648 
649 	dnode_rele(dn, FTAG);
650 }
651 
652 /*
653  * Get the next "chunk" of file data to free.  We traverse the file from
654  * the end so that the file gets shorter over time (if we crashes in the
655  * middle, this will leave us in a better state).  We find allocated file
656  * data by simply searching the allocated level 1 indirects.
657  *
658  * On input, *start should be the first offset that does not need to be
659  * freed (e.g. "offset + length").  On return, *start will be the first
660  * offset that should be freed.
661  */
662 static int
663 get_next_chunk(dnode_t *dn, uint64_t *start, uint64_t minimum)
664 {
665 	uint64_t maxblks = DMU_MAX_ACCESS >> (dn->dn_indblkshift + 1);
666 	/* bytes of data covered by a level-1 indirect block */
667 	uint64_t iblkrange =
668 	    dn->dn_datablksz * EPB(dn->dn_indblkshift, SPA_BLKPTRSHIFT);
669 
670 	ASSERT3U(minimum, <=, *start);
671 
672 	if (*start - minimum <= iblkrange * maxblks) {
673 		*start = minimum;
674 		return (0);
675 	}
676 	ASSERT(ISP2(iblkrange));
677 
678 	for (uint64_t blks = 0; *start > minimum && blks < maxblks; blks++) {
679 		int err;
680 
681 		/*
682 		 * dnode_next_offset(BACKWARDS) will find an allocated L1
683 		 * indirect block at or before the input offset.  We must
684 		 * decrement *start so that it is at the end of the region
685 		 * to search.
686 		 */
687 		(*start)--;
688 		err = dnode_next_offset(dn,
689 		    DNODE_FIND_BACKWARDS, start, 2, 1, 0);
690 
691 		/* if there are no indirect blocks before start, we are done */
692 		if (err == ESRCH) {
693 			*start = minimum;
694 			break;
695 		} else if (err != 0) {
696 			return (err);
697 		}
698 
699 		/* set start to the beginning of this L1 indirect */
700 		*start = P2ALIGN(*start, iblkrange);
701 	}
702 	if (*start < minimum)
703 		*start = minimum;
704 	return (0);
705 }
706 
707 /*
708  * If this objset is of type OST_ZFS return true if vfs's unmounted flag is set,
709  * otherwise return false.
710  * Used below in dmu_free_long_range_impl() to enable abort when unmounting
711  */
712 /*ARGSUSED*/
713 static boolean_t
714 dmu_objset_zfs_unmounting(objset_t *os)
715 {
716 #ifdef _KERNEL
717 	if (dmu_objset_type(os) == DMU_OST_ZFS)
718 		return (zfs_get_vfs_flag_unmounted(os));
719 #endif
720 	return (B_FALSE);
721 }
722 
723 static int
724 dmu_free_long_range_impl(objset_t *os, dnode_t *dn, uint64_t offset,
725     uint64_t length)
726 {
727 	uint64_t object_size = (dn->dn_maxblkid + 1) * dn->dn_datablksz;
728 	int err;
729 	uint64_t dirty_frees_threshold;
730 	dsl_pool_t *dp = dmu_objset_pool(os);
731 
732 	if (offset >= object_size)
733 		return (0);
734 
735 	if (zfs_per_txg_dirty_frees_percent <= 100)
736 		dirty_frees_threshold =
737 		    zfs_per_txg_dirty_frees_percent * zfs_dirty_data_max / 100;
738 	else
739 		dirty_frees_threshold = zfs_dirty_data_max / 4;
740 
741 	if (length == DMU_OBJECT_END || offset + length > object_size)
742 		length = object_size - offset;
743 
744 	while (length != 0) {
745 		uint64_t chunk_end, chunk_begin, chunk_len;
746 		uint64_t long_free_dirty_all_txgs = 0;
747 		dmu_tx_t *tx;
748 
749 		if (dmu_objset_zfs_unmounting(dn->dn_objset))
750 			return (SET_ERROR(EINTR));
751 
752 		chunk_end = chunk_begin = offset + length;
753 
754 		/* move chunk_begin backwards to the beginning of this chunk */
755 		err = get_next_chunk(dn, &chunk_begin, offset);
756 		if (err)
757 			return (err);
758 		ASSERT3U(chunk_begin, >=, offset);
759 		ASSERT3U(chunk_begin, <=, chunk_end);
760 
761 		chunk_len = chunk_end - chunk_begin;
762 
763 		mutex_enter(&dp->dp_lock);
764 		for (int t = 0; t < TXG_SIZE; t++) {
765 			long_free_dirty_all_txgs +=
766 			    dp->dp_long_free_dirty_pertxg[t];
767 		}
768 		mutex_exit(&dp->dp_lock);
769 
770 		/*
771 		 * To avoid filling up a TXG with just frees wait for
772 		 * the next TXG to open before freeing more chunks if
773 		 * we have reached the threshold of frees
774 		 */
775 		if (dirty_frees_threshold != 0 &&
776 		    long_free_dirty_all_txgs >= dirty_frees_threshold) {
777 			txg_wait_open(dp, 0);
778 			continue;
779 		}
780 
781 		tx = dmu_tx_create(os);
782 		dmu_tx_hold_free(tx, dn->dn_object, chunk_begin, chunk_len);
783 
784 		/*
785 		 * Mark this transaction as typically resulting in a net
786 		 * reduction in space used.
787 		 */
788 		dmu_tx_mark_netfree(tx);
789 		err = dmu_tx_assign(tx, TXG_WAIT);
790 		if (err) {
791 			dmu_tx_abort(tx);
792 			return (err);
793 		}
794 
795 		mutex_enter(&dp->dp_lock);
796 		dp->dp_long_free_dirty_pertxg[dmu_tx_get_txg(tx) & TXG_MASK] +=
797 		    chunk_len;
798 		mutex_exit(&dp->dp_lock);
799 		DTRACE_PROBE3(free__long__range,
800 		    uint64_t, long_free_dirty_all_txgs, uint64_t, chunk_len,
801 		    uint64_t, dmu_tx_get_txg(tx));
802 		dnode_free_range(dn, chunk_begin, chunk_len, tx);
803 		dmu_tx_commit(tx);
804 
805 		length -= chunk_len;
806 	}
807 	return (0);
808 }
809 
810 int
811 dmu_free_long_range(objset_t *os, uint64_t object,
812     uint64_t offset, uint64_t length)
813 {
814 	dnode_t *dn;
815 	int err;
816 
817 	err = dnode_hold(os, object, FTAG, &dn);
818 	if (err != 0)
819 		return (err);
820 	err = dmu_free_long_range_impl(os, dn, offset, length);
821 
822 	/*
823 	 * It is important to zero out the maxblkid when freeing the entire
824 	 * file, so that (a) subsequent calls to dmu_free_long_range_impl()
825 	 * will take the fast path, and (b) dnode_reallocate() can verify
826 	 * that the entire file has been freed.
827 	 */
828 	if (err == 0 && offset == 0 && length == DMU_OBJECT_END)
829 		dn->dn_maxblkid = 0;
830 
831 	dnode_rele(dn, FTAG);
832 	return (err);
833 }
834 
835 int
836 dmu_free_long_object(objset_t *os, uint64_t object)
837 {
838 	dmu_tx_t *tx;
839 	int err;
840 
841 	err = dmu_free_long_range(os, object, 0, DMU_OBJECT_END);
842 	if (err != 0)
843 		return (err);
844 
845 	tx = dmu_tx_create(os);
846 	dmu_tx_hold_bonus(tx, object);
847 	dmu_tx_hold_free(tx, object, 0, DMU_OBJECT_END);
848 	dmu_tx_mark_netfree(tx);
849 	err = dmu_tx_assign(tx, TXG_WAIT);
850 	if (err == 0) {
851 		err = dmu_object_free(os, object, tx);
852 		dmu_tx_commit(tx);
853 	} else {
854 		dmu_tx_abort(tx);
855 	}
856 
857 	return (err);
858 }
859 
860 int
861 dmu_free_range(objset_t *os, uint64_t object, uint64_t offset,
862     uint64_t size, dmu_tx_t *tx)
863 {
864 	dnode_t *dn;
865 	int err = dnode_hold(os, object, FTAG, &dn);
866 	if (err)
867 		return (err);
868 	ASSERT(offset < UINT64_MAX);
869 	ASSERT(size == -1ULL || size <= UINT64_MAX - offset);
870 	dnode_free_range(dn, offset, size, tx);
871 	dnode_rele(dn, FTAG);
872 	return (0);
873 }
874 
875 static int
876 dmu_read_impl(dnode_t *dn, uint64_t offset, uint64_t size,
877     void *buf, uint32_t flags)
878 {
879 	dmu_buf_t **dbp;
880 	int numbufs, err = 0;
881 
882 	/*
883 	 * Deal with odd block sizes, where there can't be data past the first
884 	 * block.  If we ever do the tail block optimization, we will need to
885 	 * handle that here as well.
886 	 */
887 	if (dn->dn_maxblkid == 0) {
888 		int newsz = offset > dn->dn_datablksz ? 0 :
889 		    MIN(size, dn->dn_datablksz - offset);
890 		bzero((char *)buf + newsz, size - newsz);
891 		size = newsz;
892 	}
893 
894 	while (size > 0) {
895 		uint64_t mylen = MIN(size, DMU_MAX_ACCESS / 2);
896 		int i;
897 
898 		/*
899 		 * NB: we could do this block-at-a-time, but it's nice
900 		 * to be reading in parallel.
901 		 */
902 		err = dmu_buf_hold_array_by_dnode(dn, offset, mylen,
903 		    TRUE, FTAG, &numbufs, &dbp, flags);
904 		if (err)
905 			break;
906 
907 		for (i = 0; i < numbufs; i++) {
908 			int tocpy;
909 			int bufoff;
910 			dmu_buf_t *db = dbp[i];
911 
912 			ASSERT(size > 0);
913 
914 			bufoff = offset - db->db_offset;
915 			tocpy = (int)MIN(db->db_size - bufoff, size);
916 
917 			bcopy((char *)db->db_data + bufoff, buf, tocpy);
918 
919 			offset += tocpy;
920 			size -= tocpy;
921 			buf = (char *)buf + tocpy;
922 		}
923 		dmu_buf_rele_array(dbp, numbufs, FTAG);
924 	}
925 	return (err);
926 }
927 
928 int
929 dmu_read(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
930     void *buf, uint32_t flags)
931 {
932 	dnode_t *dn;
933 	int err;
934 
935 	err = dnode_hold(os, object, FTAG, &dn);
936 	if (err != 0)
937 		return (err);
938 
939 	err = dmu_read_impl(dn, offset, size, buf, flags);
940 	dnode_rele(dn, FTAG);
941 	return (err);
942 }
943 
944 int
945 dmu_read_by_dnode(dnode_t *dn, uint64_t offset, uint64_t size, void *buf,
946     uint32_t flags)
947 {
948 	return (dmu_read_impl(dn, offset, size, buf, flags));
949 }
950 
951 static void
952 dmu_write_impl(dmu_buf_t **dbp, int numbufs, uint64_t offset, uint64_t size,
953     const void *buf, dmu_tx_t *tx)
954 {
955 	int i;
956 
957 	for (i = 0; i < numbufs; i++) {
958 		int tocpy;
959 		int bufoff;
960 		dmu_buf_t *db = dbp[i];
961 
962 		ASSERT(size > 0);
963 
964 		bufoff = offset - db->db_offset;
965 		tocpy = (int)MIN(db->db_size - bufoff, size);
966 
967 		ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);
968 
969 		if (tocpy == db->db_size)
970 			dmu_buf_will_fill(db, tx);
971 		else
972 			dmu_buf_will_dirty(db, tx);
973 
974 		bcopy(buf, (char *)db->db_data + bufoff, tocpy);
975 
976 		if (tocpy == db->db_size)
977 			dmu_buf_fill_done(db, tx);
978 
979 		offset += tocpy;
980 		size -= tocpy;
981 		buf = (char *)buf + tocpy;
982 	}
983 }
984 
985 void
986 dmu_write(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
987     const void *buf, dmu_tx_t *tx)
988 {
989 	dmu_buf_t **dbp;
990 	int numbufs;
991 
992 	if (size == 0)
993 		return;
994 
995 	VERIFY0(dmu_buf_hold_array(os, object, offset, size,
996 	    FALSE, FTAG, &numbufs, &dbp));
997 	dmu_write_impl(dbp, numbufs, offset, size, buf, tx);
998 	dmu_buf_rele_array(dbp, numbufs, FTAG);
999 }
1000 
1001 void
1002 dmu_write_by_dnode(dnode_t *dn, uint64_t offset, uint64_t size,
1003     const void *buf, dmu_tx_t *tx)
1004 {
1005 	dmu_buf_t **dbp;
1006 	int numbufs;
1007 
1008 	if (size == 0)
1009 		return;
1010 
1011 	VERIFY0(dmu_buf_hold_array_by_dnode(dn, offset, size,
1012 	    FALSE, FTAG, &numbufs, &dbp, DMU_READ_PREFETCH));
1013 	dmu_write_impl(dbp, numbufs, offset, size, buf, tx);
1014 	dmu_buf_rele_array(dbp, numbufs, FTAG);
1015 }
1016 
1017 void
1018 dmu_prealloc(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
1019     dmu_tx_t *tx)
1020 {
1021 	dmu_buf_t **dbp;
1022 	int numbufs, i;
1023 
1024 	if (size == 0)
1025 		return;
1026 
1027 	VERIFY(0 == dmu_buf_hold_array(os, object, offset, size,
1028 	    FALSE, FTAG, &numbufs, &dbp));
1029 
1030 	for (i = 0; i < numbufs; i++) {
1031 		dmu_buf_t *db = dbp[i];
1032 
1033 		dmu_buf_will_not_fill(db, tx);
1034 	}
1035 	dmu_buf_rele_array(dbp, numbufs, FTAG);
1036 }
1037 
1038 void
1039 dmu_write_embedded(objset_t *os, uint64_t object, uint64_t offset,
1040     void *data, uint8_t etype, uint8_t comp, int uncompressed_size,
1041     int compressed_size, int byteorder, dmu_tx_t *tx)
1042 {
1043 	dmu_buf_t *db;
1044 
1045 	ASSERT3U(etype, <, NUM_BP_EMBEDDED_TYPES);
1046 	ASSERT3U(comp, <, ZIO_COMPRESS_FUNCTIONS);
1047 	VERIFY0(dmu_buf_hold_noread(os, object, offset,
1048 	    FTAG, &db));
1049 
1050 	dmu_buf_write_embedded(db,
1051 	    data, (bp_embedded_type_t)etype, (enum zio_compress)comp,
1052 	    uncompressed_size, compressed_size, byteorder, tx);
1053 
1054 	dmu_buf_rele(db, FTAG);
1055 }
1056 
1057 /*
1058  * DMU support for xuio
1059  */
1060 kstat_t *xuio_ksp = NULL;
1061 
1062 int
1063 dmu_xuio_init(xuio_t *xuio, int nblk)
1064 {
1065 	dmu_xuio_t *priv;
1066 	uio_t *uio = &xuio->xu_uio;
1067 
1068 	uio->uio_iovcnt = nblk;
1069 	uio->uio_iov = kmem_zalloc(nblk * sizeof (iovec_t), KM_SLEEP);
1070 
1071 	priv = kmem_zalloc(sizeof (dmu_xuio_t), KM_SLEEP);
1072 	priv->cnt = nblk;
1073 	priv->bufs = kmem_zalloc(nblk * sizeof (arc_buf_t *), KM_SLEEP);
1074 	priv->iovp = uio->uio_iov;
1075 	XUIO_XUZC_PRIV(xuio) = priv;
1076 
1077 	if (XUIO_XUZC_RW(xuio) == UIO_READ)
1078 		XUIOSTAT_INCR(xuiostat_onloan_rbuf, nblk);
1079 	else
1080 		XUIOSTAT_INCR(xuiostat_onloan_wbuf, nblk);
1081 
1082 	return (0);
1083 }
1084 
1085 void
1086 dmu_xuio_fini(xuio_t *xuio)
1087 {
1088 	dmu_xuio_t *priv = XUIO_XUZC_PRIV(xuio);
1089 	int nblk = priv->cnt;
1090 
1091 	kmem_free(priv->iovp, nblk * sizeof (iovec_t));
1092 	kmem_free(priv->bufs, nblk * sizeof (arc_buf_t *));
1093 	kmem_free(priv, sizeof (dmu_xuio_t));
1094 
1095 	if (XUIO_XUZC_RW(xuio) == UIO_READ)
1096 		XUIOSTAT_INCR(xuiostat_onloan_rbuf, -nblk);
1097 	else
1098 		XUIOSTAT_INCR(xuiostat_onloan_wbuf, -nblk);
1099 }
1100 
1101 /*
1102  * Initialize iov[priv->next] and priv->bufs[priv->next] with { off, n, abuf }
1103  * and increase priv->next by 1.
1104  */
1105 int
1106 dmu_xuio_add(xuio_t *xuio, arc_buf_t *abuf, offset_t off, size_t n)
1107 {
1108 	struct iovec *iov;
1109 	uio_t *uio = &xuio->xu_uio;
1110 	dmu_xuio_t *priv = XUIO_XUZC_PRIV(xuio);
1111 	int i = priv->next++;
1112 
1113 	ASSERT(i < priv->cnt);
1114 	ASSERT(off + n <= arc_buf_lsize(abuf));
1115 	iov = uio->uio_iov + i;
1116 	iov->iov_base = (char *)abuf->b_data + off;
1117 	iov->iov_len = n;
1118 	priv->bufs[i] = abuf;
1119 	return (0);
1120 }
1121 
1122 int
1123 dmu_xuio_cnt(xuio_t *xuio)
1124 {
1125 	dmu_xuio_t *priv = XUIO_XUZC_PRIV(xuio);
1126 	return (priv->cnt);
1127 }
1128 
1129 arc_buf_t *
1130 dmu_xuio_arcbuf(xuio_t *xuio, int i)
1131 {
1132 	dmu_xuio_t *priv = XUIO_XUZC_PRIV(xuio);
1133 
1134 	ASSERT(i < priv->cnt);
1135 	return (priv->bufs[i]);
1136 }
1137 
1138 void
1139 dmu_xuio_clear(xuio_t *xuio, int i)
1140 {
1141 	dmu_xuio_t *priv = XUIO_XUZC_PRIV(xuio);
1142 
1143 	ASSERT(i < priv->cnt);
1144 	priv->bufs[i] = NULL;
1145 }
1146 
1147 static void
1148 xuio_stat_init(void)
1149 {
1150 	xuio_ksp = kstat_create("zfs", 0, "xuio_stats", "misc",
1151 	    KSTAT_TYPE_NAMED, sizeof (xuio_stats) / sizeof (kstat_named_t),
1152 	    KSTAT_FLAG_VIRTUAL);
1153 	if (xuio_ksp != NULL) {
1154 		xuio_ksp->ks_data = &xuio_stats;
1155 		kstat_install(xuio_ksp);
1156 	}
1157 }
1158 
1159 static void
1160 xuio_stat_fini(void)
1161 {
1162 	if (xuio_ksp != NULL) {
1163 		kstat_delete(xuio_ksp);
1164 		xuio_ksp = NULL;
1165 	}
1166 }
1167 
1168 void
1169 xuio_stat_wbuf_copied(void)
1170 {
1171 	XUIOSTAT_BUMP(xuiostat_wbuf_copied);
1172 }
1173 
1174 void
1175 xuio_stat_wbuf_nocopy(void)
1176 {
1177 	XUIOSTAT_BUMP(xuiostat_wbuf_nocopy);
1178 }
1179 
1180 #ifdef _KERNEL
1181 static int
1182 dmu_read_uio_dnode(dnode_t *dn, uio_t *uio, uint64_t size)
1183 {
1184 	dmu_buf_t **dbp;
1185 	int numbufs, i, err;
1186 	xuio_t *xuio = NULL;
1187 
1188 	/*
1189 	 * NB: we could do this block-at-a-time, but it's nice
1190 	 * to be reading in parallel.
1191 	 */
1192 	err = dmu_buf_hold_array_by_dnode(dn, uio->uio_loffset, size,
1193 	    TRUE, FTAG, &numbufs, &dbp, 0);
1194 	if (err)
1195 		return (err);
1196 
1197 	if (uio->uio_extflg == UIO_XUIO)
1198 		xuio = (xuio_t *)uio;
1199 
1200 	for (i = 0; i < numbufs; i++) {
1201 		int tocpy;
1202 		int bufoff;
1203 		dmu_buf_t *db = dbp[i];
1204 
1205 		ASSERT(size > 0);
1206 
1207 		bufoff = uio->uio_loffset - db->db_offset;
1208 		tocpy = (int)MIN(db->db_size - bufoff, size);
1209 
1210 		if (xuio) {
1211 			dmu_buf_impl_t *dbi = (dmu_buf_impl_t *)db;
1212 			arc_buf_t *dbuf_abuf = dbi->db_buf;
1213 			arc_buf_t *abuf = dbuf_loan_arcbuf(dbi);
1214 			err = dmu_xuio_add(xuio, abuf, bufoff, tocpy);
1215 			if (!err) {
1216 				uio->uio_resid -= tocpy;
1217 				uio->uio_loffset += tocpy;
1218 			}
1219 
1220 			if (abuf == dbuf_abuf)
1221 				XUIOSTAT_BUMP(xuiostat_rbuf_nocopy);
1222 			else
1223 				XUIOSTAT_BUMP(xuiostat_rbuf_copied);
1224 		} else {
1225 			err = uiomove((char *)db->db_data + bufoff, tocpy,
1226 			    UIO_READ, uio);
1227 		}
1228 		if (err)
1229 			break;
1230 
1231 		size -= tocpy;
1232 	}
1233 	dmu_buf_rele_array(dbp, numbufs, FTAG);
1234 
1235 	return (err);
1236 }
1237 
1238 /*
1239  * Read 'size' bytes into the uio buffer.
1240  * From object zdb->db_object.
1241  * Starting at offset uio->uio_loffset.
1242  *
1243  * If the caller already has a dbuf in the target object
1244  * (e.g. its bonus buffer), this routine is faster than dmu_read_uio(),
1245  * because we don't have to find the dnode_t for the object.
1246  */
1247 int
1248 dmu_read_uio_dbuf(dmu_buf_t *zdb, uio_t *uio, uint64_t size)
1249 {
1250 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)zdb;
1251 	dnode_t *dn;
1252 	int err;
1253 
1254 	if (size == 0)
1255 		return (0);
1256 
1257 	DB_DNODE_ENTER(db);
1258 	dn = DB_DNODE(db);
1259 	err = dmu_read_uio_dnode(dn, uio, size);
1260 	DB_DNODE_EXIT(db);
1261 
1262 	return (err);
1263 }
1264 
1265 /*
1266  * Read 'size' bytes into the uio buffer.
1267  * From the specified object
1268  * Starting at offset uio->uio_loffset.
1269  */
1270 int
1271 dmu_read_uio(objset_t *os, uint64_t object, uio_t *uio, uint64_t size)
1272 {
1273 	dnode_t *dn;
1274 	int err;
1275 
1276 	if (size == 0)
1277 		return (0);
1278 
1279 	err = dnode_hold(os, object, FTAG, &dn);
1280 	if (err)
1281 		return (err);
1282 
1283 	err = dmu_read_uio_dnode(dn, uio, size);
1284 
1285 	dnode_rele(dn, FTAG);
1286 
1287 	return (err);
1288 }
1289 
1290 static int
1291 dmu_write_uio_dnode(dnode_t *dn, uio_t *uio, uint64_t size, dmu_tx_t *tx)
1292 {
1293 	dmu_buf_t **dbp;
1294 	int numbufs;
1295 	int err = 0;
1296 	int i;
1297 
1298 	err = dmu_buf_hold_array_by_dnode(dn, uio->uio_loffset, size,
1299 	    FALSE, FTAG, &numbufs, &dbp, DMU_READ_PREFETCH);
1300 	if (err)
1301 		return (err);
1302 
1303 	for (i = 0; i < numbufs; i++) {
1304 		int tocpy;
1305 		int bufoff;
1306 		dmu_buf_t *db = dbp[i];
1307 
1308 		ASSERT(size > 0);
1309 
1310 		bufoff = uio->uio_loffset - db->db_offset;
1311 		tocpy = (int)MIN(db->db_size - bufoff, size);
1312 
1313 		ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);
1314 
1315 		if (tocpy == db->db_size)
1316 			dmu_buf_will_fill(db, tx);
1317 		else
1318 			dmu_buf_will_dirty(db, tx);
1319 
1320 		/*
1321 		 * XXX uiomove could block forever (eg. nfs-backed
1322 		 * pages).  There needs to be a uiolockdown() function
1323 		 * to lock the pages in memory, so that uiomove won't
1324 		 * block.
1325 		 */
1326 		err = uiomove((char *)db->db_data + bufoff, tocpy,
1327 		    UIO_WRITE, uio);
1328 
1329 		if (tocpy == db->db_size)
1330 			dmu_buf_fill_done(db, tx);
1331 
1332 		if (err)
1333 			break;
1334 
1335 		size -= tocpy;
1336 	}
1337 
1338 	dmu_buf_rele_array(dbp, numbufs, FTAG);
1339 	return (err);
1340 }
1341 
1342 /*
1343  * Write 'size' bytes from the uio buffer.
1344  * To object zdb->db_object.
1345  * Starting at offset uio->uio_loffset.
1346  *
1347  * If the caller already has a dbuf in the target object
1348  * (e.g. its bonus buffer), this routine is faster than dmu_write_uio(),
1349  * because we don't have to find the dnode_t for the object.
1350  */
1351 int
1352 dmu_write_uio_dbuf(dmu_buf_t *zdb, uio_t *uio, uint64_t size,
1353     dmu_tx_t *tx)
1354 {
1355 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)zdb;
1356 	dnode_t *dn;
1357 	int err;
1358 
1359 	if (size == 0)
1360 		return (0);
1361 
1362 	DB_DNODE_ENTER(db);
1363 	dn = DB_DNODE(db);
1364 	err = dmu_write_uio_dnode(dn, uio, size, tx);
1365 	DB_DNODE_EXIT(db);
1366 
1367 	return (err);
1368 }
1369 
1370 /*
1371  * Write 'size' bytes from the uio buffer.
1372  * To the specified object.
1373  * Starting at offset uio->uio_loffset.
1374  */
1375 int
1376 dmu_write_uio(objset_t *os, uint64_t object, uio_t *uio, uint64_t size,
1377     dmu_tx_t *tx)
1378 {
1379 	dnode_t *dn;
1380 	int err;
1381 
1382 	if (size == 0)
1383 		return (0);
1384 
1385 	err = dnode_hold(os, object, FTAG, &dn);
1386 	if (err)
1387 		return (err);
1388 
1389 	err = dmu_write_uio_dnode(dn, uio, size, tx);
1390 
1391 	dnode_rele(dn, FTAG);
1392 
1393 	return (err);
1394 }
1395 
1396 int
1397 dmu_write_pages(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
1398     page_t *pp, dmu_tx_t *tx)
1399 {
1400 	dmu_buf_t **dbp;
1401 	int numbufs, i;
1402 	int err;
1403 
1404 	if (size == 0)
1405 		return (0);
1406 
1407 	err = dmu_buf_hold_array(os, object, offset, size,
1408 	    FALSE, FTAG, &numbufs, &dbp);
1409 	if (err)
1410 		return (err);
1411 
1412 	for (i = 0; i < numbufs; i++) {
1413 		int tocpy, copied, thiscpy;
1414 		int bufoff;
1415 		dmu_buf_t *db = dbp[i];
1416 		caddr_t va;
1417 
1418 		ASSERT(size > 0);
1419 		ASSERT3U(db->db_size, >=, PAGESIZE);
1420 
1421 		bufoff = offset - db->db_offset;
1422 		tocpy = (int)MIN(db->db_size - bufoff, size);
1423 
1424 		ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);
1425 
1426 		if (tocpy == db->db_size)
1427 			dmu_buf_will_fill(db, tx);
1428 		else
1429 			dmu_buf_will_dirty(db, tx);
1430 
1431 		for (copied = 0; copied < tocpy; copied += PAGESIZE) {
1432 			ASSERT3U(pp->p_offset, ==, db->db_offset + bufoff);
1433 			thiscpy = MIN(PAGESIZE, tocpy - copied);
1434 			va = zfs_map_page(pp, S_READ);
1435 			bcopy(va, (char *)db->db_data + bufoff, thiscpy);
1436 			zfs_unmap_page(pp, va);
1437 			pp = pp->p_next;
1438 			bufoff += PAGESIZE;
1439 		}
1440 
1441 		if (tocpy == db->db_size)
1442 			dmu_buf_fill_done(db, tx);
1443 
1444 		offset += tocpy;
1445 		size -= tocpy;
1446 	}
1447 	dmu_buf_rele_array(dbp, numbufs, FTAG);
1448 	return (err);
1449 }
1450 #endif
1451 
1452 /*
1453  * Allocate a loaned anonymous arc buffer.
1454  */
1455 arc_buf_t *
1456 dmu_request_arcbuf(dmu_buf_t *handle, int size)
1457 {
1458 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)handle;
1459 
1460 	return (arc_loan_buf(db->db_objset->os_spa, B_FALSE, size));
1461 }
1462 
1463 /*
1464  * Free a loaned arc buffer.
1465  */
1466 void
1467 dmu_return_arcbuf(arc_buf_t *buf)
1468 {
1469 	arc_return_buf(buf, FTAG);
1470 	arc_buf_destroy(buf, FTAG);
1471 }
1472 
1473 /*
1474  * When possible directly assign passed loaned arc buffer to a dbuf.
1475  * If this is not possible copy the contents of passed arc buf via
1476  * dmu_write().
1477  */
1478 void
1479 dmu_assign_arcbuf(dmu_buf_t *handle, uint64_t offset, arc_buf_t *buf,
1480     dmu_tx_t *tx)
1481 {
1482 	dmu_buf_impl_t *dbuf = (dmu_buf_impl_t *)handle;
1483 	dnode_t *dn;
1484 	dmu_buf_impl_t *db;
1485 	uint32_t blksz = (uint32_t)arc_buf_lsize(buf);
1486 	uint64_t blkid;
1487 
1488 	DB_DNODE_ENTER(dbuf);
1489 	dn = DB_DNODE(dbuf);
1490 	rw_enter(&dn->dn_struct_rwlock, RW_READER);
1491 	blkid = dbuf_whichblock(dn, 0, offset);
1492 	VERIFY((db = dbuf_hold(dn, blkid, FTAG)) != NULL);
1493 	rw_exit(&dn->dn_struct_rwlock);
1494 	DB_DNODE_EXIT(dbuf);
1495 
1496 	/*
1497 	 * We can only assign if the offset is aligned, the arc buf is the
1498 	 * same size as the dbuf, and the dbuf is not metadata.
1499 	 */
1500 	if (offset == db->db.db_offset && blksz == db->db.db_size) {
1501 		dbuf_assign_arcbuf(db, buf, tx);
1502 		dbuf_rele(db, FTAG);
1503 	} else {
1504 		objset_t *os;
1505 		uint64_t object;
1506 
1507 		/* compressed bufs must always be assignable to their dbuf */
1508 		ASSERT3U(arc_get_compression(buf), ==, ZIO_COMPRESS_OFF);
1509 		ASSERT(!(buf->b_flags & ARC_BUF_FLAG_COMPRESSED));
1510 
1511 		DB_DNODE_ENTER(dbuf);
1512 		dn = DB_DNODE(dbuf);
1513 		os = dn->dn_objset;
1514 		object = dn->dn_object;
1515 		DB_DNODE_EXIT(dbuf);
1516 
1517 		dbuf_rele(db, FTAG);
1518 		dmu_write(os, object, offset, blksz, buf->b_data, tx);
1519 		dmu_return_arcbuf(buf);
1520 		XUIOSTAT_BUMP(xuiostat_wbuf_copied);
1521 	}
1522 }
1523 
1524 typedef struct {
1525 	dbuf_dirty_record_t	*dsa_dr;
1526 	dmu_sync_cb_t		*dsa_done;
1527 	zgd_t			*dsa_zgd;
1528 	dmu_tx_t		*dsa_tx;
1529 } dmu_sync_arg_t;
1530 
1531 /* ARGSUSED */
1532 static void
1533 dmu_sync_ready(zio_t *zio, arc_buf_t *buf, void *varg)
1534 {
1535 	dmu_sync_arg_t *dsa = varg;
1536 	dmu_buf_t *db = dsa->dsa_zgd->zgd_db;
1537 	blkptr_t *bp = zio->io_bp;
1538 
1539 	if (zio->io_error == 0) {
1540 		if (BP_IS_HOLE(bp)) {
1541 			/*
1542 			 * A block of zeros may compress to a hole, but the
1543 			 * block size still needs to be known for replay.
1544 			 */
1545 			BP_SET_LSIZE(bp, db->db_size);
1546 		} else if (!BP_IS_EMBEDDED(bp)) {
1547 			ASSERT(BP_GET_LEVEL(bp) == 0);
1548 			bp->blk_fill = 1;
1549 		}
1550 	}
1551 }
1552 
1553 static void
1554 dmu_sync_late_arrival_ready(zio_t *zio)
1555 {
1556 	dmu_sync_ready(zio, NULL, zio->io_private);
1557 }
1558 
1559 /* ARGSUSED */
1560 static void
1561 dmu_sync_done(zio_t *zio, arc_buf_t *buf, void *varg)
1562 {
1563 	dmu_sync_arg_t *dsa = varg;
1564 	dbuf_dirty_record_t *dr = dsa->dsa_dr;
1565 	dmu_buf_impl_t *db = dr->dr_dbuf;
1566 
1567 	mutex_enter(&db->db_mtx);
1568 	ASSERT(dr->dt.dl.dr_override_state == DR_IN_DMU_SYNC);
1569 	if (zio->io_error == 0) {
1570 		dr->dt.dl.dr_nopwrite = !!(zio->io_flags & ZIO_FLAG_NOPWRITE);
1571 		if (dr->dt.dl.dr_nopwrite) {
1572 			blkptr_t *bp = zio->io_bp;
1573 			blkptr_t *bp_orig = &zio->io_bp_orig;
1574 			uint8_t chksum = BP_GET_CHECKSUM(bp_orig);
1575 
1576 			ASSERT(BP_EQUAL(bp, bp_orig));
1577 			VERIFY(BP_EQUAL(bp, db->db_blkptr));
1578 			ASSERT(zio->io_prop.zp_compress != ZIO_COMPRESS_OFF);
1579 			ASSERT(zio_checksum_table[chksum].ci_flags &
1580 			    ZCHECKSUM_FLAG_NOPWRITE);
1581 		}
1582 		dr->dt.dl.dr_overridden_by = *zio->io_bp;
1583 		dr->dt.dl.dr_override_state = DR_OVERRIDDEN;
1584 		dr->dt.dl.dr_copies = zio->io_prop.zp_copies;
1585 
1586 		/*
1587 		 * Old style holes are filled with all zeros, whereas
1588 		 * new-style holes maintain their lsize, type, level,
1589 		 * and birth time (see zio_write_compress). While we
1590 		 * need to reset the BP_SET_LSIZE() call that happened
1591 		 * in dmu_sync_ready for old style holes, we do *not*
1592 		 * want to wipe out the information contained in new
1593 		 * style holes. Thus, only zero out the block pointer if
1594 		 * it's an old style hole.
1595 		 */
1596 		if (BP_IS_HOLE(&dr->dt.dl.dr_overridden_by) &&
1597 		    dr->dt.dl.dr_overridden_by.blk_birth == 0)
1598 			BP_ZERO(&dr->dt.dl.dr_overridden_by);
1599 	} else {
1600 		dr->dt.dl.dr_override_state = DR_NOT_OVERRIDDEN;
1601 	}
1602 	cv_broadcast(&db->db_changed);
1603 	mutex_exit(&db->db_mtx);
1604 
1605 	dsa->dsa_done(dsa->dsa_zgd, zio->io_error);
1606 
1607 	kmem_free(dsa, sizeof (*dsa));
1608 }
1609 
1610 static void
1611 dmu_sync_late_arrival_done(zio_t *zio)
1612 {
1613 	blkptr_t *bp = zio->io_bp;
1614 	dmu_sync_arg_t *dsa = zio->io_private;
1615 	blkptr_t *bp_orig = &zio->io_bp_orig;
1616 
1617 	if (zio->io_error == 0 && !BP_IS_HOLE(bp)) {
1618 		ASSERT(!(zio->io_flags & ZIO_FLAG_NOPWRITE));
1619 		ASSERT(BP_IS_HOLE(bp_orig) || !BP_EQUAL(bp, bp_orig));
1620 		ASSERT(zio->io_bp->blk_birth == zio->io_txg);
1621 		ASSERT(zio->io_txg > spa_syncing_txg(zio->io_spa));
1622 		zio_free(zio->io_spa, zio->io_txg, zio->io_bp);
1623 	}
1624 
1625 	dmu_tx_commit(dsa->dsa_tx);
1626 
1627 	dsa->dsa_done(dsa->dsa_zgd, zio->io_error);
1628 
1629 	abd_put(zio->io_abd);
1630 	kmem_free(dsa, sizeof (*dsa));
1631 }
1632 
1633 static int
1634 dmu_sync_late_arrival(zio_t *pio, objset_t *os, dmu_sync_cb_t *done, zgd_t *zgd,
1635     zio_prop_t *zp, zbookmark_phys_t *zb)
1636 {
1637 	dmu_sync_arg_t *dsa;
1638 	dmu_tx_t *tx;
1639 
1640 	tx = dmu_tx_create(os);
1641 	dmu_tx_hold_space(tx, zgd->zgd_db->db_size);
1642 	if (dmu_tx_assign(tx, TXG_WAIT) != 0) {
1643 		dmu_tx_abort(tx);
1644 		/* Make zl_get_data do txg_waited_synced() */
1645 		return (SET_ERROR(EIO));
1646 	}
1647 
1648 	dsa = kmem_alloc(sizeof (dmu_sync_arg_t), KM_SLEEP);
1649 	dsa->dsa_dr = NULL;
1650 	dsa->dsa_done = done;
1651 	dsa->dsa_zgd = zgd;
1652 	dsa->dsa_tx = tx;
1653 
1654 	/*
1655 	 * Since we are currently syncing this txg, it's nontrivial to
1656 	 * determine what BP to nopwrite against, so we disable nopwrite.
1657 	 *
1658 	 * When syncing, the db_blkptr is initially the BP of the previous
1659 	 * txg.  We can not nopwrite against it because it will be changed
1660 	 * (this is similar to the non-late-arrival case where the dbuf is
1661 	 * dirty in a future txg).
1662 	 *
1663 	 * Then dbuf_write_ready() sets bp_blkptr to the location we will write.
1664 	 * We can not nopwrite against it because although the BP will not
1665 	 * (typically) be changed, the data has not yet been persisted to this
1666 	 * location.
1667 	 *
1668 	 * Finally, when dbuf_write_done() is called, it is theoretically
1669 	 * possible to always nopwrite, because the data that was written in
1670 	 * this txg is the same data that we are trying to write.  However we
1671 	 * would need to check that this dbuf is not dirty in any future
1672 	 * txg's (as we do in the normal dmu_sync() path). For simplicity, we
1673 	 * don't nopwrite in this case.
1674 	 */
1675 	zp->zp_nopwrite = B_FALSE;
1676 
1677 	zio_nowait(zio_write(pio, os->os_spa, dmu_tx_get_txg(tx), zgd->zgd_bp,
1678 	    abd_get_from_buf(zgd->zgd_db->db_data, zgd->zgd_db->db_size),
1679 	    zgd->zgd_db->db_size, zgd->zgd_db->db_size, zp,
1680 	    dmu_sync_late_arrival_ready, NULL, NULL, dmu_sync_late_arrival_done,
1681 	    dsa, ZIO_PRIORITY_SYNC_WRITE, ZIO_FLAG_CANFAIL, zb));
1682 
1683 	return (0);
1684 }
1685 
1686 /*
1687  * Intent log support: sync the block associated with db to disk.
1688  * N.B. and XXX: the caller is responsible for making sure that the
1689  * data isn't changing while dmu_sync() is writing it.
1690  *
1691  * Return values:
1692  *
1693  *	EEXIST: this txg has already been synced, so there's nothing to do.
1694  *		The caller should not log the write.
1695  *
1696  *	ENOENT: the block was dbuf_free_range()'d, so there's nothing to do.
1697  *		The caller should not log the write.
1698  *
1699  *	EALREADY: this block is already in the process of being synced.
1700  *		The caller should track its progress (somehow).
1701  *
1702  *	EIO: could not do the I/O.
1703  *		The caller should do a txg_wait_synced().
1704  *
1705  *	0: the I/O has been initiated.
1706  *		The caller should log this blkptr in the done callback.
1707  *		It is possible that the I/O will fail, in which case
1708  *		the error will be reported to the done callback and
1709  *		propagated to pio from zio_done().
1710  */
1711 int
1712 dmu_sync(zio_t *pio, uint64_t txg, dmu_sync_cb_t *done, zgd_t *zgd)
1713 {
1714 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)zgd->zgd_db;
1715 	objset_t *os = db->db_objset;
1716 	dsl_dataset_t *ds = os->os_dsl_dataset;
1717 	dbuf_dirty_record_t *dr;
1718 	dmu_sync_arg_t *dsa;
1719 	zbookmark_phys_t zb;
1720 	zio_prop_t zp;
1721 	dnode_t *dn;
1722 
1723 	ASSERT(pio != NULL);
1724 	ASSERT(txg != 0);
1725 
1726 	SET_BOOKMARK(&zb, ds->ds_object,
1727 	    db->db.db_object, db->db_level, db->db_blkid);
1728 
1729 	DB_DNODE_ENTER(db);
1730 	dn = DB_DNODE(db);
1731 	dmu_write_policy(os, dn, db->db_level, WP_DMU_SYNC, &zp);
1732 	DB_DNODE_EXIT(db);
1733 
1734 	/*
1735 	 * If we're frozen (running ziltest), we always need to generate a bp.
1736 	 */
1737 	if (txg > spa_freeze_txg(os->os_spa))
1738 		return (dmu_sync_late_arrival(pio, os, done, zgd, &zp, &zb));
1739 
1740 	/*
1741 	 * Grabbing db_mtx now provides a barrier between dbuf_sync_leaf()
1742 	 * and us.  If we determine that this txg is not yet syncing,
1743 	 * but it begins to sync a moment later, that's OK because the
1744 	 * sync thread will block in dbuf_sync_leaf() until we drop db_mtx.
1745 	 */
1746 	mutex_enter(&db->db_mtx);
1747 
1748 	if (txg <= spa_last_synced_txg(os->os_spa)) {
1749 		/*
1750 		 * This txg has already synced.  There's nothing to do.
1751 		 */
1752 		mutex_exit(&db->db_mtx);
1753 		return (SET_ERROR(EEXIST));
1754 	}
1755 
1756 	if (txg <= spa_syncing_txg(os->os_spa)) {
1757 		/*
1758 		 * This txg is currently syncing, so we can't mess with
1759 		 * the dirty record anymore; just write a new log block.
1760 		 */
1761 		mutex_exit(&db->db_mtx);
1762 		return (dmu_sync_late_arrival(pio, os, done, zgd, &zp, &zb));
1763 	}
1764 
1765 	dr = db->db_last_dirty;
1766 	while (dr && dr->dr_txg != txg)
1767 		dr = dr->dr_next;
1768 
1769 	if (dr == NULL) {
1770 		/*
1771 		 * There's no dr for this dbuf, so it must have been freed.
1772 		 * There's no need to log writes to freed blocks, so we're done.
1773 		 */
1774 		mutex_exit(&db->db_mtx);
1775 		return (SET_ERROR(ENOENT));
1776 	}
1777 
1778 	ASSERT(dr->dr_next == NULL || dr->dr_next->dr_txg < txg);
1779 
1780 	if (db->db_blkptr != NULL) {
1781 		/*
1782 		 * We need to fill in zgd_bp with the current blkptr so that
1783 		 * the nopwrite code can check if we're writing the same
1784 		 * data that's already on disk.  We can only nopwrite if we
1785 		 * are sure that after making the copy, db_blkptr will not
1786 		 * change until our i/o completes.  We ensure this by
1787 		 * holding the db_mtx, and only allowing nopwrite if the
1788 		 * block is not already dirty (see below).  This is verified
1789 		 * by dmu_sync_done(), which VERIFYs that the db_blkptr has
1790 		 * not changed.
1791 		 */
1792 		*zgd->zgd_bp = *db->db_blkptr;
1793 	}
1794 
1795 	/*
1796 	 * Assume the on-disk data is X, the current syncing data (in
1797 	 * txg - 1) is Y, and the current in-memory data is Z (currently
1798 	 * in dmu_sync).
1799 	 *
1800 	 * We usually want to perform a nopwrite if X and Z are the
1801 	 * same.  However, if Y is different (i.e. the BP is going to
1802 	 * change before this write takes effect), then a nopwrite will
1803 	 * be incorrect - we would override with X, which could have
1804 	 * been freed when Y was written.
1805 	 *
1806 	 * (Note that this is not a concern when we are nop-writing from
1807 	 * syncing context, because X and Y must be identical, because
1808 	 * all previous txgs have been synced.)
1809 	 *
1810 	 * Therefore, we disable nopwrite if the current BP could change
1811 	 * before this TXG.  There are two ways it could change: by
1812 	 * being dirty (dr_next is non-NULL), or by being freed
1813 	 * (dnode_block_freed()).  This behavior is verified by
1814 	 * zio_done(), which VERIFYs that the override BP is identical
1815 	 * to the on-disk BP.
1816 	 */
1817 	DB_DNODE_ENTER(db);
1818 	dn = DB_DNODE(db);
1819 	if (dr->dr_next != NULL || dnode_block_freed(dn, db->db_blkid))
1820 		zp.zp_nopwrite = B_FALSE;
1821 	DB_DNODE_EXIT(db);
1822 
1823 	ASSERT(dr->dr_txg == txg);
1824 	if (dr->dt.dl.dr_override_state == DR_IN_DMU_SYNC ||
1825 	    dr->dt.dl.dr_override_state == DR_OVERRIDDEN) {
1826 		/*
1827 		 * We have already issued a sync write for this buffer,
1828 		 * or this buffer has already been synced.  It could not
1829 		 * have been dirtied since, or we would have cleared the state.
1830 		 */
1831 		mutex_exit(&db->db_mtx);
1832 		return (SET_ERROR(EALREADY));
1833 	}
1834 
1835 	ASSERT(dr->dt.dl.dr_override_state == DR_NOT_OVERRIDDEN);
1836 	dr->dt.dl.dr_override_state = DR_IN_DMU_SYNC;
1837 	mutex_exit(&db->db_mtx);
1838 
1839 	dsa = kmem_alloc(sizeof (dmu_sync_arg_t), KM_SLEEP);
1840 	dsa->dsa_dr = dr;
1841 	dsa->dsa_done = done;
1842 	dsa->dsa_zgd = zgd;
1843 	dsa->dsa_tx = NULL;
1844 
1845 	zio_nowait(arc_write(pio, os->os_spa, txg,
1846 	    zgd->zgd_bp, dr->dt.dl.dr_data, DBUF_IS_L2CACHEABLE(db),
1847 	    &zp, dmu_sync_ready, NULL, NULL, dmu_sync_done, dsa,
1848 	    ZIO_PRIORITY_SYNC_WRITE, ZIO_FLAG_CANFAIL, &zb));
1849 
1850 	return (0);
1851 }
1852 
1853 int
1854 dmu_object_set_blocksize(objset_t *os, uint64_t object, uint64_t size, int ibs,
1855     dmu_tx_t *tx)
1856 {
1857 	dnode_t *dn;
1858 	int err;
1859 
1860 	err = dnode_hold(os, object, FTAG, &dn);
1861 	if (err)
1862 		return (err);
1863 	err = dnode_set_blksz(dn, size, ibs, tx);
1864 	dnode_rele(dn, FTAG);
1865 	return (err);
1866 }
1867 
1868 void
1869 dmu_object_set_checksum(objset_t *os, uint64_t object, uint8_t checksum,
1870     dmu_tx_t *tx)
1871 {
1872 	dnode_t *dn;
1873 
1874 	/*
1875 	 * Send streams include each object's checksum function.  This
1876 	 * check ensures that the receiving system can understand the
1877 	 * checksum function transmitted.
1878 	 */
1879 	ASSERT3U(checksum, <, ZIO_CHECKSUM_LEGACY_FUNCTIONS);
1880 
1881 	VERIFY0(dnode_hold(os, object, FTAG, &dn));
1882 	ASSERT3U(checksum, <, ZIO_CHECKSUM_FUNCTIONS);
1883 	dn->dn_checksum = checksum;
1884 	dnode_setdirty(dn, tx);
1885 	dnode_rele(dn, FTAG);
1886 }
1887 
1888 void
1889 dmu_object_set_compress(objset_t *os, uint64_t object, uint8_t compress,
1890     dmu_tx_t *tx)
1891 {
1892 	dnode_t *dn;
1893 
1894 	/*
1895 	 * Send streams include each object's compression function.  This
1896 	 * check ensures that the receiving system can understand the
1897 	 * compression function transmitted.
1898 	 */
1899 	ASSERT3U(compress, <, ZIO_COMPRESS_LEGACY_FUNCTIONS);
1900 
1901 	VERIFY0(dnode_hold(os, object, FTAG, &dn));
1902 	dn->dn_compress = compress;
1903 	dnode_setdirty(dn, tx);
1904 	dnode_rele(dn, FTAG);
1905 }
1906 
1907 int zfs_mdcomp_disable = 0;
1908 
1909 /*
1910  * When the "redundant_metadata" property is set to "most", only indirect
1911  * blocks of this level and higher will have an additional ditto block.
1912  */
1913 int zfs_redundant_metadata_most_ditto_level = 2;
1914 
1915 void
1916 dmu_write_policy(objset_t *os, dnode_t *dn, int level, int wp, zio_prop_t *zp)
1917 {
1918 	dmu_object_type_t type = dn ? dn->dn_type : DMU_OT_OBJSET;
1919 	boolean_t ismd = (level > 0 || DMU_OT_IS_METADATA(type) ||
1920 	    (wp & WP_SPILL));
1921 	enum zio_checksum checksum = os->os_checksum;
1922 	enum zio_compress compress = os->os_compress;
1923 	enum zio_checksum dedup_checksum = os->os_dedup_checksum;
1924 	boolean_t dedup = B_FALSE;
1925 	boolean_t nopwrite = B_FALSE;
1926 	boolean_t dedup_verify = os->os_dedup_verify;
1927 	int copies = os->os_copies;
1928 
1929 	/*
1930 	 * We maintain different write policies for each of the following
1931 	 * types of data:
1932 	 *	 1. metadata
1933 	 *	 2. preallocated blocks (i.e. level-0 blocks of a dump device)
1934 	 *	 3. all other level 0 blocks
1935 	 */
1936 	if (ismd) {
1937 		if (zfs_mdcomp_disable) {
1938 			compress = ZIO_COMPRESS_EMPTY;
1939 		} else {
1940 			/*
1941 			 * XXX -- we should design a compression algorithm
1942 			 * that specializes in arrays of bps.
1943 			 */
1944 			compress = zio_compress_select(os->os_spa,
1945 			    ZIO_COMPRESS_ON, ZIO_COMPRESS_ON);
1946 		}
1947 
1948 		/*
1949 		 * Metadata always gets checksummed.  If the data
1950 		 * checksum is multi-bit correctable, and it's not a
1951 		 * ZBT-style checksum, then it's suitable for metadata
1952 		 * as well.  Otherwise, the metadata checksum defaults
1953 		 * to fletcher4.
1954 		 */
1955 		if (!(zio_checksum_table[checksum].ci_flags &
1956 		    ZCHECKSUM_FLAG_METADATA) ||
1957 		    (zio_checksum_table[checksum].ci_flags &
1958 		    ZCHECKSUM_FLAG_EMBEDDED))
1959 			checksum = ZIO_CHECKSUM_FLETCHER_4;
1960 
1961 		if (os->os_redundant_metadata == ZFS_REDUNDANT_METADATA_ALL ||
1962 		    (os->os_redundant_metadata ==
1963 		    ZFS_REDUNDANT_METADATA_MOST &&
1964 		    (level >= zfs_redundant_metadata_most_ditto_level ||
1965 		    DMU_OT_IS_METADATA(type) || (wp & WP_SPILL))))
1966 			copies++;
1967 	} else if (wp & WP_NOFILL) {
1968 		ASSERT(level == 0);
1969 
1970 		/*
1971 		 * If we're writing preallocated blocks, we aren't actually
1972 		 * writing them so don't set any policy properties.  These
1973 		 * blocks are currently only used by an external subsystem
1974 		 * outside of zfs (i.e. dump) and not written by the zio
1975 		 * pipeline.
1976 		 */
1977 		compress = ZIO_COMPRESS_OFF;
1978 		checksum = ZIO_CHECKSUM_NOPARITY;
1979 	} else {
1980 		compress = zio_compress_select(os->os_spa, dn->dn_compress,
1981 		    compress);
1982 
1983 		checksum = (dedup_checksum == ZIO_CHECKSUM_OFF) ?
1984 		    zio_checksum_select(dn->dn_checksum, checksum) :
1985 		    dedup_checksum;
1986 
1987 		/*
1988 		 * Determine dedup setting.  If we are in dmu_sync(),
1989 		 * we won't actually dedup now because that's all
1990 		 * done in syncing context; but we do want to use the
1991 		 * dedup checkum.  If the checksum is not strong
1992 		 * enough to ensure unique signatures, force
1993 		 * dedup_verify.
1994 		 */
1995 		if (dedup_checksum != ZIO_CHECKSUM_OFF) {
1996 			dedup = (wp & WP_DMU_SYNC) ? B_FALSE : B_TRUE;
1997 			if (!(zio_checksum_table[checksum].ci_flags &
1998 			    ZCHECKSUM_FLAG_DEDUP))
1999 				dedup_verify = B_TRUE;
2000 		}
2001 
2002 		/*
2003 		 * Enable nopwrite if we have secure enough checksum
2004 		 * algorithm (see comment in zio_nop_write) and
2005 		 * compression is enabled.  We don't enable nopwrite if
2006 		 * dedup is enabled as the two features are mutually
2007 		 * exclusive.
2008 		 */
2009 		nopwrite = (!dedup && (zio_checksum_table[checksum].ci_flags &
2010 		    ZCHECKSUM_FLAG_NOPWRITE) &&
2011 		    compress != ZIO_COMPRESS_OFF && zfs_nopwrite_enabled);
2012 	}
2013 
2014 	zp->zp_checksum = checksum;
2015 	zp->zp_compress = compress;
2016 	ASSERT3U(zp->zp_compress, !=, ZIO_COMPRESS_INHERIT);
2017 
2018 	zp->zp_type = (wp & WP_SPILL) ? dn->dn_bonustype : type;
2019 	zp->zp_level = level;
2020 	zp->zp_copies = MIN(copies, spa_max_replication(os->os_spa));
2021 	zp->zp_dedup = dedup;
2022 	zp->zp_dedup_verify = dedup && dedup_verify;
2023 	zp->zp_nopwrite = nopwrite;
2024 }
2025 
2026 int
2027 dmu_offset_next(objset_t *os, uint64_t object, boolean_t hole, uint64_t *off)
2028 {
2029 	dnode_t *dn;
2030 	int err;
2031 
2032 	/*
2033 	 * Sync any current changes before
2034 	 * we go trundling through the block pointers.
2035 	 */
2036 	err = dmu_object_wait_synced(os, object);
2037 	if (err) {
2038 		return (err);
2039 	}
2040 
2041 	err = dnode_hold(os, object, FTAG, &dn);
2042 	if (err) {
2043 		return (err);
2044 	}
2045 
2046 	err = dnode_next_offset(dn, (hole ? DNODE_FIND_HOLE : 0), off, 1, 1, 0);
2047 	dnode_rele(dn, FTAG);
2048 
2049 	return (err);
2050 }
2051 
2052 /*
2053  * Given the ZFS object, if it contains any dirty nodes
2054  * this function flushes all dirty blocks to disk. This
2055  * ensures the DMU object info is updated. A more efficient
2056  * future version might just find the TXG with the maximum
2057  * ID and wait for that to be synced.
2058  */
2059 int
2060 dmu_object_wait_synced(objset_t *os, uint64_t object)
2061 {
2062 	dnode_t *dn;
2063 	int error, i;
2064 
2065 	error = dnode_hold(os, object, FTAG, &dn);
2066 	if (error) {
2067 		return (error);
2068 	}
2069 
2070 	for (i = 0; i < TXG_SIZE; i++) {
2071 		if (list_link_active(&dn->dn_dirty_link[i])) {
2072 			break;
2073 		}
2074 	}
2075 	dnode_rele(dn, FTAG);
2076 	if (i != TXG_SIZE) {
2077 		txg_wait_synced(dmu_objset_pool(os), 0);
2078 	}
2079 
2080 	return (0);
2081 }
2082 
2083 void
2084 dmu_object_info_from_dnode(dnode_t *dn, dmu_object_info_t *doi)
2085 {
2086 	dnode_phys_t *dnp;
2087 
2088 	rw_enter(&dn->dn_struct_rwlock, RW_READER);
2089 	mutex_enter(&dn->dn_mtx);
2090 
2091 	dnp = dn->dn_phys;
2092 
2093 	doi->doi_data_block_size = dn->dn_datablksz;
2094 	doi->doi_metadata_block_size = dn->dn_indblkshift ?
2095 	    1ULL << dn->dn_indblkshift : 0;
2096 	doi->doi_type = dn->dn_type;
2097 	doi->doi_bonus_type = dn->dn_bonustype;
2098 	doi->doi_bonus_size = dn->dn_bonuslen;
2099 	doi->doi_indirection = dn->dn_nlevels;
2100 	doi->doi_checksum = dn->dn_checksum;
2101 	doi->doi_compress = dn->dn_compress;
2102 	doi->doi_nblkptr = dn->dn_nblkptr;
2103 	doi->doi_physical_blocks_512 = (DN_USED_BYTES(dnp) + 256) >> 9;
2104 	doi->doi_max_offset = (dn->dn_maxblkid + 1) * dn->dn_datablksz;
2105 	doi->doi_fill_count = 0;
2106 	for (int i = 0; i < dnp->dn_nblkptr; i++)
2107 		doi->doi_fill_count += BP_GET_FILL(&dnp->dn_blkptr[i]);
2108 
2109 	mutex_exit(&dn->dn_mtx);
2110 	rw_exit(&dn->dn_struct_rwlock);
2111 }
2112 
2113 /*
2114  * Get information on a DMU object.
2115  * If doi is NULL, just indicates whether the object exists.
2116  */
2117 int
2118 dmu_object_info(objset_t *os, uint64_t object, dmu_object_info_t *doi)
2119 {
2120 	dnode_t *dn;
2121 	int err = dnode_hold(os, object, FTAG, &dn);
2122 
2123 	if (err)
2124 		return (err);
2125 
2126 	if (doi != NULL)
2127 		dmu_object_info_from_dnode(dn, doi);
2128 
2129 	dnode_rele(dn, FTAG);
2130 	return (0);
2131 }
2132 
2133 /*
2134  * As above, but faster; can be used when you have a held dbuf in hand.
2135  */
2136 void
2137 dmu_object_info_from_db(dmu_buf_t *db_fake, dmu_object_info_t *doi)
2138 {
2139 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
2140 
2141 	DB_DNODE_ENTER(db);
2142 	dmu_object_info_from_dnode(DB_DNODE(db), doi);
2143 	DB_DNODE_EXIT(db);
2144 }
2145 
2146 /*
2147  * Faster still when you only care about the size.
2148  * This is specifically optimized for zfs_getattr().
2149  */
2150 void
2151 dmu_object_size_from_db(dmu_buf_t *db_fake, uint32_t *blksize,
2152     u_longlong_t *nblk512)
2153 {
2154 	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
2155 	dnode_t *dn;
2156 
2157 	DB_DNODE_ENTER(db);
2158 	dn = DB_DNODE(db);
2159 
2160 	*blksize = dn->dn_datablksz;
2161 	/* add 1 for dnode space */
2162 	*nblk512 = ((DN_USED_BYTES(dn->dn_phys) + SPA_MINBLOCKSIZE/2) >>
2163 	    SPA_MINBLOCKSHIFT) + 1;
2164 	DB_DNODE_EXIT(db);
2165 }
2166 
2167 void
2168 byteswap_uint64_array(void *vbuf, size_t size)
2169 {
2170 	uint64_t *buf = vbuf;
2171 	size_t count = size >> 3;
2172 	int i;
2173 
2174 	ASSERT((size & 7) == 0);
2175 
2176 	for (i = 0; i < count; i++)
2177 		buf[i] = BSWAP_64(buf[i]);
2178 }
2179 
2180 void
2181 byteswap_uint32_array(void *vbuf, size_t size)
2182 {
2183 	uint32_t *buf = vbuf;
2184 	size_t count = size >> 2;
2185 	int i;
2186 
2187 	ASSERT((size & 3) == 0);
2188 
2189 	for (i = 0; i < count; i++)
2190 		buf[i] = BSWAP_32(buf[i]);
2191 }
2192 
2193 void
2194 byteswap_uint16_array(void *vbuf, size_t size)
2195 {
2196 	uint16_t *buf = vbuf;
2197 	size_t count = size >> 1;
2198 	int i;
2199 
2200 	ASSERT((size & 1) == 0);
2201 
2202 	for (i = 0; i < count; i++)
2203 		buf[i] = BSWAP_16(buf[i]);
2204 }
2205 
2206 /* ARGSUSED */
2207 void
2208 byteswap_uint8_array(void *vbuf, size_t size)
2209 {
2210 }
2211 
2212 void
2213 dmu_init(void)
2214 {
2215 	abd_init();
2216 	zfs_dbgmsg_init();
2217 	sa_cache_init();
2218 	xuio_stat_init();
2219 	dmu_objset_init();
2220 	dnode_init();
2221 	zfetch_init();
2222 	l2arc_init();
2223 	arc_init();
2224 	dbuf_init();
2225 }
2226 
2227 void
2228 dmu_fini(void)
2229 {
2230 	arc_fini(); /* arc depends on l2arc, so arc must go first */
2231 	l2arc_fini();
2232 	zfetch_fini();
2233 	dbuf_fini();
2234 	dnode_fini();
2235 	dmu_objset_fini();
2236 	xuio_stat_fini();
2237 	sa_cache_fini();
2238 	zfs_dbgmsg_fini();
2239 	abd_fini();
2240 }
2241