]> CyberLeo.Net >> Repos - FreeBSD/stable/8.git/blob - sys/cddl/contrib/opensolaris/uts/common/fs/zfs/dmu.c
MFC r214378:
[FreeBSD/stable/8.git] / sys / cddl / contrib / opensolaris / uts / common / fs / zfs / dmu.c
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/dmu.h>
27 #include <sys/dmu_impl.h>
28 #include <sys/dmu_tx.h>
29 #include <sys/dbuf.h>
30 #include <sys/dnode.h>
31 #include <sys/zfs_context.h>
32 #include <sys/dmu_objset.h>
33 #include <sys/dmu_traverse.h>
34 #include <sys/dsl_dataset.h>
35 #include <sys/dsl_dir.h>
36 #include <sys/dsl_pool.h>
37 #include <sys/dsl_synctask.h>
38 #include <sys/dsl_prop.h>
39 #include <sys/dmu_zfetch.h>
40 #include <sys/zfs_ioctl.h>
41 #include <sys/zap.h>
42 #include <sys/zio_checksum.h>
43 #include <sys/zfs_znode.h>
44
45 const dmu_object_type_info_t dmu_ot[DMU_OT_NUMTYPES] = {
46         {       byteswap_uint8_array,   TRUE,   "unallocated"           },
47         {       zap_byteswap,           TRUE,   "object directory"      },
48         {       byteswap_uint64_array,  TRUE,   "object array"          },
49         {       byteswap_uint8_array,   TRUE,   "packed nvlist"         },
50         {       byteswap_uint64_array,  TRUE,   "packed nvlist size"    },
51         {       byteswap_uint64_array,  TRUE,   "bplist"                },
52         {       byteswap_uint64_array,  TRUE,   "bplist header"         },
53         {       byteswap_uint64_array,  TRUE,   "SPA space map header"  },
54         {       byteswap_uint64_array,  TRUE,   "SPA space map"         },
55         {       byteswap_uint64_array,  TRUE,   "ZIL intent log"        },
56         {       dnode_buf_byteswap,     TRUE,   "DMU dnode"             },
57         {       dmu_objset_byteswap,    TRUE,   "DMU objset"            },
58         {       byteswap_uint64_array,  TRUE,   "DSL directory"         },
59         {       zap_byteswap,           TRUE,   "DSL directory child map"},
60         {       zap_byteswap,           TRUE,   "DSL dataset snap map"  },
61         {       zap_byteswap,           TRUE,   "DSL props"             },
62         {       byteswap_uint64_array,  TRUE,   "DSL dataset"           },
63         {       zfs_znode_byteswap,     TRUE,   "ZFS znode"             },
64         {       zfs_oldacl_byteswap,    TRUE,   "ZFS V0 ACL"            },
65         {       byteswap_uint8_array,   FALSE,  "ZFS plain file"        },
66         {       zap_byteswap,           TRUE,   "ZFS directory"         },
67         {       zap_byteswap,           TRUE,   "ZFS master node"       },
68         {       zap_byteswap,           TRUE,   "ZFS delete queue"      },
69         {       byteswap_uint8_array,   FALSE,  "zvol object"           },
70         {       zap_byteswap,           TRUE,   "zvol prop"             },
71         {       byteswap_uint8_array,   FALSE,  "other uint8[]"         },
72         {       byteswap_uint64_array,  FALSE,  "other uint64[]"        },
73         {       zap_byteswap,           TRUE,   "other ZAP"             },
74         {       zap_byteswap,           TRUE,   "persistent error log"  },
75         {       byteswap_uint8_array,   TRUE,   "SPA history"           },
76         {       byteswap_uint64_array,  TRUE,   "SPA history offsets"   },
77         {       zap_byteswap,           TRUE,   "Pool properties"       },
78         {       zap_byteswap,           TRUE,   "DSL permissions"       },
79         {       zfs_acl_byteswap,       TRUE,   "ZFS ACL"               },
80         {       byteswap_uint8_array,   TRUE,   "ZFS SYSACL"            },
81         {       byteswap_uint8_array,   TRUE,   "FUID table"            },
82         {       byteswap_uint64_array,  TRUE,   "FUID table size"       },
83         {       zap_byteswap,           TRUE,   "DSL dataset next clones"},
84         {       zap_byteswap,           TRUE,   "scrub work queue"      },
85         {       zap_byteswap,           TRUE,   "ZFS user/group used"   },
86         {       zap_byteswap,           TRUE,   "ZFS user/group quota"  },
87 };
88
89 int
90 dmu_buf_hold(objset_t *os, uint64_t object, uint64_t offset,
91     void *tag, dmu_buf_t **dbp)
92 {
93         dnode_t *dn;
94         uint64_t blkid;
95         dmu_buf_impl_t *db;
96         int err;
97
98         err = dnode_hold(os->os, object, FTAG, &dn);
99         if (err)
100                 return (err);
101         blkid = dbuf_whichblock(dn, offset);
102         rw_enter(&dn->dn_struct_rwlock, RW_READER);
103         db = dbuf_hold(dn, blkid, tag);
104         rw_exit(&dn->dn_struct_rwlock);
105         if (db == NULL) {
106                 err = EIO;
107         } else {
108                 err = dbuf_read(db, NULL, DB_RF_CANFAIL);
109                 if (err) {
110                         dbuf_rele(db, tag);
111                         db = NULL;
112                 }
113         }
114
115         dnode_rele(dn, FTAG);
116         *dbp = &db->db;
117         return (err);
118 }
119
120 int
121 dmu_bonus_max(void)
122 {
123         return (DN_MAX_BONUSLEN);
124 }
125
126 int
127 dmu_set_bonus(dmu_buf_t *db, int newsize, dmu_tx_t *tx)
128 {
129         dnode_t *dn = ((dmu_buf_impl_t *)db)->db_dnode;
130
131         if (dn->dn_bonus != (dmu_buf_impl_t *)db)
132                 return (EINVAL);
133         if (newsize < 0 || newsize > db->db_size)
134                 return (EINVAL);
135         dnode_setbonuslen(dn, newsize, tx);
136         return (0);
137 }
138
139 /*
140  * returns ENOENT, EIO, or 0.
141  */
142 int
143 dmu_bonus_hold(objset_t *os, uint64_t object, void *tag, dmu_buf_t **dbp)
144 {
145         dnode_t *dn;
146         dmu_buf_impl_t *db;
147         int error;
148
149         error = dnode_hold(os->os, object, FTAG, &dn);
150         if (error)
151                 return (error);
152
153         rw_enter(&dn->dn_struct_rwlock, RW_READER);
154         if (dn->dn_bonus == NULL) {
155                 rw_exit(&dn->dn_struct_rwlock);
156                 rw_enter(&dn->dn_struct_rwlock, RW_WRITER);
157                 if (dn->dn_bonus == NULL)
158                         dbuf_create_bonus(dn);
159         }
160         db = dn->dn_bonus;
161         rw_exit(&dn->dn_struct_rwlock);
162
163         /* as long as the bonus buf is held, the dnode will be held */
164         if (refcount_add(&db->db_holds, tag) == 1)
165                 VERIFY(dnode_add_ref(dn, db));
166
167         dnode_rele(dn, FTAG);
168
169         VERIFY(0 == dbuf_read(db, NULL, DB_RF_MUST_SUCCEED));
170
171         *dbp = &db->db;
172         return (0);
173 }
174
175 /*
176  * Note: longer-term, we should modify all of the dmu_buf_*() interfaces
177  * to take a held dnode rather than <os, object> -- the lookup is wasteful,
178  * and can induce severe lock contention when writing to several files
179  * whose dnodes are in the same block.
180  */
181 static int
182 dmu_buf_hold_array_by_dnode(dnode_t *dn, uint64_t offset, uint64_t length,
183     int read, void *tag, int *numbufsp, dmu_buf_t ***dbpp, uint32_t flags)
184 {
185         dsl_pool_t *dp = NULL;
186         dmu_buf_t **dbp;
187         uint64_t blkid, nblks, i;
188         uint32_t dbuf_flags;
189         int err;
190         zio_t *zio;
191         hrtime_t start;
192
193         ASSERT(length <= DMU_MAX_ACCESS);
194
195         dbuf_flags = DB_RF_CANFAIL | DB_RF_NEVERWAIT | DB_RF_HAVESTRUCT;
196         if (flags & DMU_READ_NO_PREFETCH || length > zfetch_array_rd_sz)
197                 dbuf_flags |= DB_RF_NOPREFETCH;
198
199         rw_enter(&dn->dn_struct_rwlock, RW_READER);
200         if (dn->dn_datablkshift) {
201                 int blkshift = dn->dn_datablkshift;
202                 nblks = (P2ROUNDUP(offset+length, 1ULL<<blkshift) -
203                     P2ALIGN(offset, 1ULL<<blkshift)) >> blkshift;
204         } else {
205                 if (offset + length > dn->dn_datablksz) {
206                         zfs_panic_recover("zfs: accessing past end of object "
207                             "%llx/%llx (size=%u access=%llu+%llu)",
208                             (longlong_t)dn->dn_objset->
209                             os_dsl_dataset->ds_object,
210                             (longlong_t)dn->dn_object, dn->dn_datablksz,
211                             (longlong_t)offset, (longlong_t)length);
212                         rw_exit(&dn->dn_struct_rwlock);
213                         return (EIO);
214                 }
215                 nblks = 1;
216         }
217         dbp = kmem_zalloc(sizeof (dmu_buf_t *) * nblks, KM_SLEEP);
218
219         if (dn->dn_objset->os_dsl_dataset)
220                 dp = dn->dn_objset->os_dsl_dataset->ds_dir->dd_pool;
221         if (dp && dsl_pool_sync_context(dp))
222                 start = gethrtime();
223         zio = zio_root(dn->dn_objset->os_spa, NULL, NULL, ZIO_FLAG_CANFAIL);
224         blkid = dbuf_whichblock(dn, offset);
225         for (i = 0; i < nblks; i++) {
226                 dmu_buf_impl_t *db = dbuf_hold(dn, blkid+i, tag);
227                 if (db == NULL) {
228                         rw_exit(&dn->dn_struct_rwlock);
229                         dmu_buf_rele_array(dbp, nblks, tag);
230                         zio_nowait(zio);
231                         return (EIO);
232                 }
233                 /* initiate async i/o */
234                 if (read) {
235                         (void) dbuf_read(db, zio, dbuf_flags);
236                 }
237                 dbp[i] = &db->db;
238         }
239         rw_exit(&dn->dn_struct_rwlock);
240
241         /* wait for async i/o */
242         err = zio_wait(zio);
243         /* track read overhead when we are in sync context */
244         if (dp && dsl_pool_sync_context(dp))
245                 dp->dp_read_overhead += gethrtime() - start;
246         if (err) {
247                 dmu_buf_rele_array(dbp, nblks, tag);
248                 return (err);
249         }
250
251         /* wait for other io to complete */
252         if (read) {
253                 for (i = 0; i < nblks; i++) {
254                         dmu_buf_impl_t *db = (dmu_buf_impl_t *)dbp[i];
255                         mutex_enter(&db->db_mtx);
256                         while (db->db_state == DB_READ ||
257                             db->db_state == DB_FILL)
258                                 cv_wait(&db->db_changed, &db->db_mtx);
259                         if (db->db_state == DB_UNCACHED)
260                                 err = EIO;
261                         mutex_exit(&db->db_mtx);
262                         if (err) {
263                                 dmu_buf_rele_array(dbp, nblks, tag);
264                                 return (err);
265                         }
266                 }
267         }
268
269         *numbufsp = nblks;
270         *dbpp = dbp;
271         return (0);
272 }
273
274 static int
275 dmu_buf_hold_array(objset_t *os, uint64_t object, uint64_t offset,
276     uint64_t length, int read, void *tag, int *numbufsp, dmu_buf_t ***dbpp)
277 {
278         dnode_t *dn;
279         int err;
280
281         err = dnode_hold(os->os, object, FTAG, &dn);
282         if (err)
283                 return (err);
284
285         err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,
286             numbufsp, dbpp, DMU_READ_PREFETCH);
287
288         dnode_rele(dn, FTAG);
289
290         return (err);
291 }
292
293 int
294 dmu_buf_hold_array_by_bonus(dmu_buf_t *db, uint64_t offset,
295     uint64_t length, int read, void *tag, int *numbufsp, dmu_buf_t ***dbpp)
296 {
297         dnode_t *dn = ((dmu_buf_impl_t *)db)->db_dnode;
298         int err;
299
300         err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,
301             numbufsp, dbpp, DMU_READ_PREFETCH);
302
303         return (err);
304 }
305
306 void
307 dmu_buf_rele_array(dmu_buf_t **dbp_fake, int numbufs, void *tag)
308 {
309         int i;
310         dmu_buf_impl_t **dbp = (dmu_buf_impl_t **)dbp_fake;
311
312         if (numbufs == 0)
313                 return;
314
315         for (i = 0; i < numbufs; i++) {
316                 if (dbp[i])
317                         dbuf_rele(dbp[i], tag);
318         }
319
320         kmem_free(dbp, sizeof (dmu_buf_t *) * numbufs);
321 }
322
323 void
324 dmu_prefetch(objset_t *os, uint64_t object, uint64_t offset, uint64_t len)
325 {
326         dnode_t *dn;
327         uint64_t blkid;
328         int nblks, i, err;
329
330         if (zfs_prefetch_disable)
331                 return;
332
333         if (len == 0) {  /* they're interested in the bonus buffer */
334                 dn = os->os->os_meta_dnode;
335
336                 if (object == 0 || object >= DN_MAX_OBJECT)
337                         return;
338
339                 rw_enter(&dn->dn_struct_rwlock, RW_READER);
340                 blkid = dbuf_whichblock(dn, object * sizeof (dnode_phys_t));
341                 dbuf_prefetch(dn, blkid);
342                 rw_exit(&dn->dn_struct_rwlock);
343                 return;
344         }
345
346         /*
347          * XXX - Note, if the dnode for the requested object is not
348          * already cached, we will do a *synchronous* read in the
349          * dnode_hold() call.  The same is true for any indirects.
350          */
351         err = dnode_hold(os->os, object, FTAG, &dn);
352         if (err != 0)
353                 return;
354
355         rw_enter(&dn->dn_struct_rwlock, RW_READER);
356         if (dn->dn_datablkshift) {
357                 int blkshift = dn->dn_datablkshift;
358                 nblks = (P2ROUNDUP(offset+len, 1<<blkshift) -
359                     P2ALIGN(offset, 1<<blkshift)) >> blkshift;
360         } else {
361                 nblks = (offset < dn->dn_datablksz);
362         }
363
364         if (nblks != 0) {
365                 blkid = dbuf_whichblock(dn, offset);
366                 for (i = 0; i < nblks; i++)
367                         dbuf_prefetch(dn, blkid+i);
368         }
369
370         rw_exit(&dn->dn_struct_rwlock);
371
372         dnode_rele(dn, FTAG);
373 }
374
375 /*
376  * Get the next "chunk" of file data to free.  We traverse the file from
377  * the end so that the file gets shorter over time (if we crashes in the
378  * middle, this will leave us in a better state).  We find allocated file
379  * data by simply searching the allocated level 1 indirects.
380  */
381 static int
382 get_next_chunk(dnode_t *dn, uint64_t *start, uint64_t limit)
383 {
384         uint64_t len = *start - limit;
385         uint64_t blkcnt = 0;
386         uint64_t maxblks = DMU_MAX_ACCESS / (1ULL << (dn->dn_indblkshift + 1));
387         uint64_t iblkrange =
388             dn->dn_datablksz * EPB(dn->dn_indblkshift, SPA_BLKPTRSHIFT);
389
390         ASSERT(limit <= *start);
391
392         if (len <= iblkrange * maxblks) {
393                 *start = limit;
394                 return (0);
395         }
396         ASSERT(ISP2(iblkrange));
397
398         while (*start > limit && blkcnt < maxblks) {
399                 int err;
400
401                 /* find next allocated L1 indirect */
402                 err = dnode_next_offset(dn,
403                     DNODE_FIND_BACKWARDS, start, 2, 1, 0);
404
405                 /* if there are no more, then we are done */
406                 if (err == ESRCH) {
407                         *start = limit;
408                         return (0);
409                 } else if (err) {
410                         return (err);
411                 }
412                 blkcnt += 1;
413
414                 /* reset offset to end of "next" block back */
415                 *start = P2ALIGN(*start, iblkrange);
416                 if (*start <= limit)
417                         *start = limit;
418                 else
419                         *start -= 1;
420         }
421         return (0);
422 }
423
424 static int
425 dmu_free_long_range_impl(objset_t *os, dnode_t *dn, uint64_t offset,
426     uint64_t length, boolean_t free_dnode)
427 {
428         dmu_tx_t *tx;
429         uint64_t object_size, start, end, len;
430         boolean_t trunc = (length == DMU_OBJECT_END);
431         int align, err;
432
433         align = 1 << dn->dn_datablkshift;
434         ASSERT(align > 0);
435         object_size = align == 1 ? dn->dn_datablksz :
436             (dn->dn_maxblkid + 1) << dn->dn_datablkshift;
437
438         end = offset + length;
439         if (trunc || end > object_size)
440                 end = object_size;
441         if (end <= offset)
442                 return (0);
443         length = end - offset;
444
445         while (length) {
446                 start = end;
447                 /* assert(offset <= start) */
448                 err = get_next_chunk(dn, &start, offset);
449                 if (err)
450                         return (err);
451                 len = trunc ? DMU_OBJECT_END : end - start;
452
453                 tx = dmu_tx_create(os);
454                 dmu_tx_hold_free(tx, dn->dn_object, start, len);
455                 err = dmu_tx_assign(tx, TXG_WAIT);
456                 if (err) {
457                         dmu_tx_abort(tx);
458                         return (err);
459                 }
460
461                 dnode_free_range(dn, start, trunc ? -1 : len, tx);
462
463                 if (start == 0 && free_dnode) {
464                         ASSERT(trunc);
465                         dnode_free(dn, tx);
466                 }
467
468                 length -= end - start;
469
470                 dmu_tx_commit(tx);
471                 end = start;
472         }
473         return (0);
474 }
475
476 int
477 dmu_free_long_range(objset_t *os, uint64_t object,
478     uint64_t offset, uint64_t length)
479 {
480         dnode_t *dn;
481         int err;
482
483         err = dnode_hold(os->os, object, FTAG, &dn);
484         if (err != 0)
485                 return (err);
486         err = dmu_free_long_range_impl(os, dn, offset, length, FALSE);
487         dnode_rele(dn, FTAG);
488         return (err);
489 }
490
491 int
492 dmu_free_object(objset_t *os, uint64_t object)
493 {
494         dnode_t *dn;
495         dmu_tx_t *tx;
496         int err;
497
498         err = dnode_hold_impl(os->os, object, DNODE_MUST_BE_ALLOCATED,
499             FTAG, &dn);
500         if (err != 0)
501                 return (err);
502         if (dn->dn_nlevels == 1) {
503                 tx = dmu_tx_create(os);
504                 dmu_tx_hold_bonus(tx, object);
505                 dmu_tx_hold_free(tx, dn->dn_object, 0, DMU_OBJECT_END);
506                 err = dmu_tx_assign(tx, TXG_WAIT);
507                 if (err == 0) {
508                         dnode_free_range(dn, 0, DMU_OBJECT_END, tx);
509                         dnode_free(dn, tx);
510                         dmu_tx_commit(tx);
511                 } else {
512                         dmu_tx_abort(tx);
513                 }
514         } else {
515                 err = dmu_free_long_range_impl(os, dn, 0, DMU_OBJECT_END, TRUE);
516         }
517         dnode_rele(dn, FTAG);
518         return (err);
519 }
520
521 int
522 dmu_free_range(objset_t *os, uint64_t object, uint64_t offset,
523     uint64_t size, dmu_tx_t *tx)
524 {
525         dnode_t *dn;
526         int err = dnode_hold(os->os, object, FTAG, &dn);
527         if (err)
528                 return (err);
529         ASSERT(offset < UINT64_MAX);
530         ASSERT(size == -1ULL || size <= UINT64_MAX - offset);
531         dnode_free_range(dn, offset, size, tx);
532         dnode_rele(dn, FTAG);
533         return (0);
534 }
535
536 int
537 dmu_read(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
538     void *buf, uint32_t flags)
539 {
540         dnode_t *dn;
541         dmu_buf_t **dbp;
542         int numbufs, err;
543
544         err = dnode_hold(os->os, object, FTAG, &dn);
545         if (err)
546                 return (err);
547
548         /*
549          * Deal with odd block sizes, where there can't be data past the first
550          * block.  If we ever do the tail block optimization, we will need to
551          * handle that here as well.
552          */
553         if (dn->dn_maxblkid == 0) {
554                 int newsz = offset > dn->dn_datablksz ? 0 :
555                     MIN(size, dn->dn_datablksz - offset);
556                 bzero((char *)buf + newsz, size - newsz);
557                 size = newsz;
558         }
559
560         while (size > 0) {
561                 uint64_t mylen = MIN(size, DMU_MAX_ACCESS / 2);
562                 int i;
563
564                 /*
565                  * NB: we could do this block-at-a-time, but it's nice
566                  * to be reading in parallel.
567                  */
568                 err = dmu_buf_hold_array_by_dnode(dn, offset, mylen,
569                     TRUE, FTAG, &numbufs, &dbp, flags);
570                 if (err)
571                         break;
572
573                 for (i = 0; i < numbufs; i++) {
574                         int tocpy;
575                         int bufoff;
576                         dmu_buf_t *db = dbp[i];
577
578                         ASSERT(size > 0);
579
580                         bufoff = offset - db->db_offset;
581                         tocpy = (int)MIN(db->db_size - bufoff, size);
582
583                         bcopy((char *)db->db_data + bufoff, buf, tocpy);
584
585                         offset += tocpy;
586                         size -= tocpy;
587                         buf = (char *)buf + tocpy;
588                 }
589                 dmu_buf_rele_array(dbp, numbufs, FTAG);
590         }
591         dnode_rele(dn, FTAG);
592         return (err);
593 }
594
595 void
596 dmu_write(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
597     const void *buf, dmu_tx_t *tx)
598 {
599         dmu_buf_t **dbp;
600         int numbufs, i;
601
602         if (size == 0)
603                 return;
604
605         VERIFY(0 == dmu_buf_hold_array(os, object, offset, size,
606             FALSE, FTAG, &numbufs, &dbp));
607
608         for (i = 0; i < numbufs; i++) {
609                 int tocpy;
610                 int bufoff;
611                 dmu_buf_t *db = dbp[i];
612
613                 ASSERT(size > 0);
614
615                 bufoff = offset - db->db_offset;
616                 tocpy = (int)MIN(db->db_size - bufoff, size);
617
618                 ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);
619
620                 if (tocpy == db->db_size)
621                         dmu_buf_will_fill(db, tx);
622                 else
623                         dmu_buf_will_dirty(db, tx);
624
625                 bcopy(buf, (char *)db->db_data + bufoff, tocpy);
626
627                 if (tocpy == db->db_size)
628                         dmu_buf_fill_done(db, tx);
629
630                 offset += tocpy;
631                 size -= tocpy;
632                 buf = (char *)buf + tocpy;
633         }
634         dmu_buf_rele_array(dbp, numbufs, FTAG);
635 }
636
637 #ifdef _KERNEL
638 int
639 dmu_read_uio(objset_t *os, uint64_t object, uio_t *uio, uint64_t size)
640 {
641         dmu_buf_t **dbp;
642         int numbufs, i, err;
643
644         /*
645          * NB: we could do this block-at-a-time, but it's nice
646          * to be reading in parallel.
647          */
648         err = dmu_buf_hold_array(os, object, uio->uio_loffset, size, TRUE, FTAG,
649             &numbufs, &dbp);
650         if (err)
651                 return (err);
652
653         for (i = 0; i < numbufs; i++) {
654                 int tocpy;
655                 int bufoff;
656                 dmu_buf_t *db = dbp[i];
657
658                 ASSERT(size > 0);
659
660                 bufoff = uio->uio_loffset - db->db_offset;
661                 tocpy = (int)MIN(db->db_size - bufoff, size);
662
663                 err = uiomove((char *)db->db_data + bufoff, tocpy,
664                     UIO_READ, uio);
665                 if (err)
666                         break;
667
668                 size -= tocpy;
669         }
670         dmu_buf_rele_array(dbp, numbufs, FTAG);
671
672         return (err);
673 }
674
675 int
676 dmu_write_uio(objset_t *os, uint64_t object, uio_t *uio, uint64_t size,
677     dmu_tx_t *tx)
678 {
679         dmu_buf_t **dbp;
680         int numbufs, i;
681         int err = 0;
682
683         if (size == 0)
684                 return (0);
685
686         err = dmu_buf_hold_array(os, object, uio->uio_loffset, size,
687             FALSE, FTAG, &numbufs, &dbp);
688         if (err)
689                 return (err);
690
691         for (i = 0; i < numbufs; i++) {
692                 int tocpy;
693                 int bufoff;
694                 dmu_buf_t *db = dbp[i];
695
696                 ASSERT(size > 0);
697
698                 bufoff = uio->uio_loffset - db->db_offset;
699                 tocpy = (int)MIN(db->db_size - bufoff, size);
700
701                 ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);
702
703                 if (tocpy == db->db_size)
704                         dmu_buf_will_fill(db, tx);
705                 else
706                         dmu_buf_will_dirty(db, tx);
707
708                 /*
709                  * XXX uiomove could block forever (eg. nfs-backed
710                  * pages).  There needs to be a uiolockdown() function
711                  * to lock the pages in memory, so that uiomove won't
712                  * block.
713                  */
714                 err = uiomove((char *)db->db_data + bufoff, tocpy,
715                     UIO_WRITE, uio);
716
717                 if (tocpy == db->db_size)
718                         dmu_buf_fill_done(db, tx);
719
720                 if (err)
721                         break;
722
723                 size -= tocpy;
724         }
725         dmu_buf_rele_array(dbp, numbufs, FTAG);
726         return (err);
727 }
728
729 #ifndef __FreeBSD__
730 int
731 dmu_write_pages(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
732     page_t *pp, dmu_tx_t *tx)
733 {
734         dmu_buf_t **dbp;
735         int numbufs, i;
736         int err;
737
738         if (size == 0)
739                 return (0);
740
741         err = dmu_buf_hold_array(os, object, offset, size,
742             FALSE, FTAG, &numbufs, &dbp);
743         if (err)
744                 return (err);
745
746         for (i = 0; i < numbufs; i++) {
747                 int tocpy, copied, thiscpy;
748                 int bufoff;
749                 dmu_buf_t *db = dbp[i];
750                 caddr_t va;
751
752                 ASSERT(size > 0);
753                 ASSERT3U(db->db_size, >=, PAGESIZE);
754
755                 bufoff = offset - db->db_offset;
756                 tocpy = (int)MIN(db->db_size - bufoff, size);
757
758                 ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);
759
760                 if (tocpy == db->db_size)
761                         dmu_buf_will_fill(db, tx);
762                 else
763                         dmu_buf_will_dirty(db, tx);
764
765                 for (copied = 0; copied < tocpy; copied += PAGESIZE) {
766                         ASSERT3U(pp->p_offset, ==, db->db_offset + bufoff);
767                         thiscpy = MIN(PAGESIZE, tocpy - copied);
768                         va = zfs_map_page(pp, S_READ);
769                         bcopy(va, (char *)db->db_data + bufoff, thiscpy);
770                         zfs_unmap_page(pp, va);
771                         pp = pp->p_next;
772                         bufoff += PAGESIZE;
773                 }
774
775                 if (tocpy == db->db_size)
776                         dmu_buf_fill_done(db, tx);
777
778                 offset += tocpy;
779                 size -= tocpy;
780         }
781         dmu_buf_rele_array(dbp, numbufs, FTAG);
782         return (err);
783 }
784 #endif  /* !__FreeBSD__ */
785 #endif  /* _KERNEL */
786
787 /*
788  * Allocate a loaned anonymous arc buffer.
789  */
790 arc_buf_t *
791 dmu_request_arcbuf(dmu_buf_t *handle, int size)
792 {
793         dnode_t *dn = ((dmu_buf_impl_t *)handle)->db_dnode;
794
795         return (arc_loan_buf(dn->dn_objset->os_spa, size));
796 }
797
798 /*
799  * Free a loaned arc buffer.
800  */
801 void
802 dmu_return_arcbuf(arc_buf_t *buf)
803 {
804         arc_return_buf(buf, FTAG);
805         VERIFY(arc_buf_remove_ref(buf, FTAG) == 1);
806 }
807
808 /*
809  * When possible directly assign passed loaned arc buffer to a dbuf.
810  * If this is not possible copy the contents of passed arc buf via
811  * dmu_write().
812  */
813 void
814 dmu_assign_arcbuf(dmu_buf_t *handle, uint64_t offset, arc_buf_t *buf,
815     dmu_tx_t *tx)
816 {
817         dnode_t *dn = ((dmu_buf_impl_t *)handle)->db_dnode;
818         dmu_buf_impl_t *db;
819         uint32_t blksz = (uint32_t)arc_buf_size(buf);
820         uint64_t blkid;
821
822         rw_enter(&dn->dn_struct_rwlock, RW_READER);
823         blkid = dbuf_whichblock(dn, offset);
824         VERIFY((db = dbuf_hold(dn, blkid, FTAG)) != NULL);
825         rw_exit(&dn->dn_struct_rwlock);
826
827         if (offset == db->db.db_offset && blksz == db->db.db_size) {
828                 dbuf_assign_arcbuf(db, buf, tx);
829                 dbuf_rele(db, FTAG);
830         } else {
831                 dbuf_rele(db, FTAG);
832                 ASSERT(dn->dn_objset->os.os == dn->dn_objset);
833                 dmu_write(&dn->dn_objset->os, dn->dn_object, offset, blksz,
834                     buf->b_data, tx);
835                 dmu_return_arcbuf(buf);
836         }
837 }
838
839 typedef struct {
840         dbuf_dirty_record_t     *dr;
841         dmu_sync_cb_t           *done;
842         void                    *arg;
843 } dmu_sync_arg_t;
844
845 /* ARGSUSED */
846 static void
847 dmu_sync_ready(zio_t *zio, arc_buf_t *buf, void *varg)
848 {
849         blkptr_t *bp = zio->io_bp;
850         dmu_sync_arg_t *in = varg;
851         dbuf_dirty_record_t *dr = in->dr;
852         dmu_buf_impl_t *db = dr->dr_dbuf;
853
854         if (!BP_IS_HOLE(bp)) {
855                 ASSERT(BP_GET_TYPE(bp) == db->db_dnode->dn_type);
856                 ASSERT(BP_GET_LEVEL(bp) == 0);
857                 bp->blk_fill = 1;
858         } else {
859                 /*
860                  * dmu_sync() can compress a block of zeros to a null blkptr
861                  * but the block size still needs to be passed through to replay
862                  */
863                 BP_SET_LSIZE(bp, db->db.db_size);
864         }
865 }
866
867 /* ARGSUSED */
868 static void
869 dmu_sync_done(zio_t *zio, arc_buf_t *buf, void *varg)
870 {
871         dmu_sync_arg_t *in = varg;
872         dbuf_dirty_record_t *dr = in->dr;
873         dmu_buf_impl_t *db = dr->dr_dbuf;
874         dmu_sync_cb_t *done = in->done;
875
876         mutex_enter(&db->db_mtx);
877         ASSERT(dr->dt.dl.dr_override_state == DR_IN_DMU_SYNC);
878         dr->dt.dl.dr_overridden_by = *zio->io_bp; /* structure assignment */
879         if (BP_IS_HOLE(&dr->dt.dl.dr_overridden_by))
880                 BP_ZERO(&dr->dt.dl.dr_overridden_by);
881         dr->dt.dl.dr_override_state = DR_OVERRIDDEN;
882         cv_broadcast(&db->db_changed);
883         mutex_exit(&db->db_mtx);
884
885         if (done)
886                 done(&(db->db), in->arg);
887
888         kmem_free(in, sizeof (dmu_sync_arg_t));
889 }
890
891 /*
892  * Intent log support: sync the block associated with db to disk.
893  * N.B. and XXX: the caller is responsible for making sure that the
894  * data isn't changing while dmu_sync() is writing it.
895  *
896  * Return values:
897  *
898  *      EEXIST: this txg has already been synced, so there's nothing to to.
899  *              The caller should not log the write.
900  *
901  *      ENOENT: the block was dbuf_free_range()'d, so there's nothing to do.
902  *              The caller should not log the write.
903  *
904  *      EALREADY: this block is already in the process of being synced.
905  *              The caller should track its progress (somehow).
906  *
907  *      EINPROGRESS: the IO has been initiated.
908  *              The caller should log this blkptr in the callback.
909  *
910  *      0: completed.  Sets *bp to the blkptr just written.
911  *              The caller should log this blkptr immediately.
912  */
913 int
914 dmu_sync(zio_t *pio, dmu_buf_t *db_fake,
915     blkptr_t *bp, uint64_t txg, dmu_sync_cb_t *done, void *arg)
916 {
917         dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
918         objset_impl_t *os = db->db_objset;
919         dsl_pool_t *dp = os->os_dsl_dataset->ds_dir->dd_pool;
920         tx_state_t *tx = &dp->dp_tx;
921         dbuf_dirty_record_t *dr;
922         dmu_sync_arg_t *in;
923         zbookmark_t zb;
924         writeprops_t wp = { 0 };
925         zio_t *zio;
926         int err;
927
928         ASSERT(BP_IS_HOLE(bp));
929         ASSERT(txg != 0);
930
931         dprintf("dmu_sync txg=%llu, s,o,q %llu %llu %llu\n",
932             txg, tx->tx_synced_txg, tx->tx_open_txg, tx->tx_quiesced_txg);
933
934         /*
935          * XXX - would be nice if we could do this without suspending...
936          */
937         txg_suspend(dp);
938
939         /*
940          * If this txg already synced, there's nothing to do.
941          */
942         if (txg <= tx->tx_synced_txg) {
943                 txg_resume(dp);
944                 /*
945                  * If we're running ziltest, we need the blkptr regardless.
946                  */
947                 if (txg > spa_freeze_txg(dp->dp_spa)) {
948                         /* if db_blkptr == NULL, this was an empty write */
949                         if (db->db_blkptr)
950                                 *bp = *db->db_blkptr; /* structure assignment */
951                         return (0);
952                 }
953                 return (EEXIST);
954         }
955
956         mutex_enter(&db->db_mtx);
957
958         if (txg == tx->tx_syncing_txg) {
959                 while (db->db_data_pending) {
960                         /*
961                          * IO is in-progress.  Wait for it to finish.
962                          * XXX - would be nice to be able to somehow "attach"
963                          * this zio to the parent zio passed in.
964                          */
965                         cv_wait(&db->db_changed, &db->db_mtx);
966                         if (!db->db_data_pending &&
967                             db->db_blkptr && BP_IS_HOLE(db->db_blkptr)) {
968                                 /*
969                                  * IO was compressed away
970                                  */
971                                 *bp = *db->db_blkptr; /* structure assignment */
972                                 mutex_exit(&db->db_mtx);
973                                 txg_resume(dp);
974                                 return (0);
975                         }
976                         ASSERT(db->db_data_pending ||
977                             (db->db_blkptr && db->db_blkptr->blk_birth == txg));
978                 }
979
980                 if (db->db_blkptr && db->db_blkptr->blk_birth == txg) {
981                         /*
982                          * IO is already completed.
983                          */
984                         *bp = *db->db_blkptr; /* structure assignment */
985                         mutex_exit(&db->db_mtx);
986                         txg_resume(dp);
987                         return (0);
988                 }
989         }
990
991         dr = db->db_last_dirty;
992         while (dr && dr->dr_txg > txg)
993                 dr = dr->dr_next;
994         if (dr == NULL || dr->dr_txg < txg) {
995                 /*
996                  * This dbuf isn't dirty, must have been free_range'd.
997                  * There's no need to log writes to freed blocks, so we're done.
998                  */
999                 mutex_exit(&db->db_mtx);
1000                 txg_resume(dp);
1001                 return (ENOENT);
1002         }
1003
1004         ASSERT(dr->dr_txg == txg);
1005         if (dr->dt.dl.dr_override_state == DR_IN_DMU_SYNC) {
1006                 /*
1007                  * We have already issued a sync write for this buffer.
1008                  */
1009                 mutex_exit(&db->db_mtx);
1010                 txg_resume(dp);
1011                 return (EALREADY);
1012         } else if (dr->dt.dl.dr_override_state == DR_OVERRIDDEN) {
1013                 /*
1014                  * This buffer has already been synced.  It could not
1015                  * have been dirtied since, or we would have cleared the state.
1016                  */
1017                 *bp = dr->dt.dl.dr_overridden_by; /* structure assignment */
1018                 mutex_exit(&db->db_mtx);
1019                 txg_resume(dp);
1020                 return (0);
1021         }
1022
1023         dr->dt.dl.dr_override_state = DR_IN_DMU_SYNC;
1024         in = kmem_alloc(sizeof (dmu_sync_arg_t), KM_SLEEP);
1025         in->dr = dr;
1026         in->done = done;
1027         in->arg = arg;
1028         mutex_exit(&db->db_mtx);
1029         txg_resume(dp);
1030
1031         zb.zb_objset = os->os_dsl_dataset->ds_object;
1032         zb.zb_object = db->db.db_object;
1033         zb.zb_level = db->db_level;
1034         zb.zb_blkid = db->db_blkid;
1035
1036         wp.wp_type = db->db_dnode->dn_type;
1037         wp.wp_level = db->db_level;
1038         wp.wp_copies = os->os_copies;
1039         wp.wp_dnchecksum = db->db_dnode->dn_checksum;
1040         wp.wp_oschecksum = os->os_checksum;
1041         wp.wp_dncompress = db->db_dnode->dn_compress;
1042         wp.wp_oscompress = os->os_compress;
1043
1044         ASSERT(BP_IS_HOLE(bp));
1045
1046         zio = arc_write(pio, os->os_spa, &wp, DBUF_IS_L2CACHEABLE(db),
1047             txg, bp, dr->dt.dl.dr_data, dmu_sync_ready, dmu_sync_done, in,
1048             ZIO_PRIORITY_SYNC_WRITE, ZIO_FLAG_MUSTSUCCEED, &zb);
1049
1050         if (pio) {
1051                 zio_nowait(zio);
1052                 err = EINPROGRESS;
1053         } else {
1054                 err = zio_wait(zio);
1055                 ASSERT(err == 0);
1056         }
1057         return (err);
1058 }
1059
1060 int
1061 dmu_object_set_blocksize(objset_t *os, uint64_t object, uint64_t size, int ibs,
1062         dmu_tx_t *tx)
1063 {
1064         dnode_t *dn;
1065         int err;
1066
1067         err = dnode_hold(os->os, object, FTAG, &dn);
1068         if (err)
1069                 return (err);
1070         err = dnode_set_blksz(dn, size, ibs, tx);
1071         dnode_rele(dn, FTAG);
1072         return (err);
1073 }
1074
1075 void
1076 dmu_object_set_checksum(objset_t *os, uint64_t object, uint8_t checksum,
1077         dmu_tx_t *tx)
1078 {
1079         dnode_t *dn;
1080
1081         /* XXX assumes dnode_hold will not get an i/o error */
1082         (void) dnode_hold(os->os, object, FTAG, &dn);
1083         ASSERT(checksum < ZIO_CHECKSUM_FUNCTIONS);
1084         dn->dn_checksum = checksum;
1085         dnode_setdirty(dn, tx);
1086         dnode_rele(dn, FTAG);
1087 }
1088
1089 void
1090 dmu_object_set_compress(objset_t *os, uint64_t object, uint8_t compress,
1091         dmu_tx_t *tx)
1092 {
1093         dnode_t *dn;
1094
1095         /* XXX assumes dnode_hold will not get an i/o error */
1096         (void) dnode_hold(os->os, object, FTAG, &dn);
1097         ASSERT(compress < ZIO_COMPRESS_FUNCTIONS);
1098         dn->dn_compress = compress;
1099         dnode_setdirty(dn, tx);
1100         dnode_rele(dn, FTAG);
1101 }
1102
1103 int
1104 dmu_offset_next(objset_t *os, uint64_t object, boolean_t hole, uint64_t *off)
1105 {
1106         dnode_t *dn;
1107         int i, err;
1108
1109         err = dnode_hold(os->os, object, FTAG, &dn);
1110         if (err)
1111                 return (err);
1112         /*
1113          * Sync any current changes before
1114          * we go trundling through the block pointers.
1115          */
1116         for (i = 0; i < TXG_SIZE; i++) {
1117                 if (list_link_active(&dn->dn_dirty_link[i]))
1118                         break;
1119         }
1120         if (i != TXG_SIZE) {
1121                 dnode_rele(dn, FTAG);
1122                 txg_wait_synced(dmu_objset_pool(os), 0);
1123                 err = dnode_hold(os->os, object, FTAG, &dn);
1124                 if (err)
1125                         return (err);
1126         }
1127
1128         err = dnode_next_offset(dn, (hole ? DNODE_FIND_HOLE : 0), off, 1, 1, 0);
1129         dnode_rele(dn, FTAG);
1130
1131         return (err);
1132 }
1133
1134 void
1135 dmu_object_info_from_dnode(dnode_t *dn, dmu_object_info_t *doi)
1136 {
1137         rw_enter(&dn->dn_struct_rwlock, RW_READER);
1138         mutex_enter(&dn->dn_mtx);
1139
1140         doi->doi_data_block_size = dn->dn_datablksz;
1141         doi->doi_metadata_block_size = dn->dn_indblkshift ?
1142             1ULL << dn->dn_indblkshift : 0;
1143         doi->doi_indirection = dn->dn_nlevels;
1144         doi->doi_checksum = dn->dn_checksum;
1145         doi->doi_compress = dn->dn_compress;
1146         doi->doi_physical_blks = (DN_USED_BYTES(dn->dn_phys) +
1147             SPA_MINBLOCKSIZE/2) >> SPA_MINBLOCKSHIFT;
1148         doi->doi_max_block_offset = dn->dn_phys->dn_maxblkid;
1149         doi->doi_type = dn->dn_type;
1150         doi->doi_bonus_size = dn->dn_bonuslen;
1151         doi->doi_bonus_type = dn->dn_bonustype;
1152
1153         mutex_exit(&dn->dn_mtx);
1154         rw_exit(&dn->dn_struct_rwlock);
1155 }
1156
1157 /*
1158  * Get information on a DMU object.
1159  * If doi is NULL, just indicates whether the object exists.
1160  */
1161 int
1162 dmu_object_info(objset_t *os, uint64_t object, dmu_object_info_t *doi)
1163 {
1164         dnode_t *dn;
1165         int err = dnode_hold(os->os, object, FTAG, &dn);
1166
1167         if (err)
1168                 return (err);
1169
1170         if (doi != NULL)
1171                 dmu_object_info_from_dnode(dn, doi);
1172
1173         dnode_rele(dn, FTAG);
1174         return (0);
1175 }
1176
1177 /*
1178  * As above, but faster; can be used when you have a held dbuf in hand.
1179  */
1180 void
1181 dmu_object_info_from_db(dmu_buf_t *db, dmu_object_info_t *doi)
1182 {
1183         dmu_object_info_from_dnode(((dmu_buf_impl_t *)db)->db_dnode, doi);
1184 }
1185
1186 /*
1187  * Faster still when you only care about the size.
1188  * This is specifically optimized for zfs_getattr().
1189  */
1190 void
1191 dmu_object_size_from_db(dmu_buf_t *db, uint32_t *blksize, u_longlong_t *nblk512)
1192 {
1193         dnode_t *dn = ((dmu_buf_impl_t *)db)->db_dnode;
1194
1195         *blksize = dn->dn_datablksz;
1196         /* add 1 for dnode space */
1197         *nblk512 = ((DN_USED_BYTES(dn->dn_phys) + SPA_MINBLOCKSIZE/2) >>
1198             SPA_MINBLOCKSHIFT) + 1;
1199 }
1200
1201 void
1202 byteswap_uint64_array(void *vbuf, size_t size)
1203 {
1204         uint64_t *buf = vbuf;
1205         size_t count = size >> 3;
1206         int i;
1207
1208         ASSERT((size & 7) == 0);
1209
1210         for (i = 0; i < count; i++)
1211                 buf[i] = BSWAP_64(buf[i]);
1212 }
1213
1214 void
1215 byteswap_uint32_array(void *vbuf, size_t size)
1216 {
1217         uint32_t *buf = vbuf;
1218         size_t count = size >> 2;
1219         int i;
1220
1221         ASSERT((size & 3) == 0);
1222
1223         for (i = 0; i < count; i++)
1224                 buf[i] = BSWAP_32(buf[i]);
1225 }
1226
1227 void
1228 byteswap_uint16_array(void *vbuf, size_t size)
1229 {
1230         uint16_t *buf = vbuf;
1231         size_t count = size >> 1;
1232         int i;
1233
1234         ASSERT((size & 1) == 0);
1235
1236         for (i = 0; i < count; i++)
1237                 buf[i] = BSWAP_16(buf[i]);
1238 }
1239
1240 /* ARGSUSED */
1241 void
1242 byteswap_uint8_array(void *vbuf, size_t size)
1243 {
1244 }
1245
1246 void
1247 dmu_init(void)
1248 {
1249         dbuf_init();
1250         dnode_init();
1251         zfetch_init();
1252         arc_init();
1253         l2arc_init();
1254 }
1255
1256 void
1257 dmu_fini(void)
1258 {
1259         arc_fini();
1260         zfetch_fini();
1261         dnode_fini();
1262         dbuf_fini();
1263         l2arc_fini();
1264 }