module/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 https://opensource.org/licenses/CDDL-1.0.
  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, 2020 by Delphix. All rights reserved.
  24  * Copyright (c) 2013 by Saso Kiselkov. All rights reserved.
  25  * Copyright (c) 2013, Joyent, Inc. All rights reserved.
  26  * Copyright (c) 2016, Nexenta Systems, Inc. All rights reserved.
  27  * Copyright (c) 2015 by Chunwei Chen. All rights reserved.
  28  * Copyright (c) 2019 Datto Inc.
  29  * Copyright (c) 2019, Klara Inc.
  30  * Copyright (c) 2019, Allan Jude
  31  * Copyright (c) 2022 Hewlett Packard Enterprise Development LP.
  32  * Copyright (c) 2021, 2022 by Pawel Jakub Dawidek
  33  */
  34
  35 #include <sys/dmu.h>
  36 #include <sys/dmu_impl.h>
  37 #include <sys/dmu_tx.h>
  38 #include <sys/dbuf.h>
  39 #include <sys/dnode.h>
  40 #include <sys/zfs_context.h>
  41 #include <sys/dmu_objset.h>
  42 #include <sys/dmu_traverse.h>
  43 #include <sys/dsl_dataset.h>
  44 #include <sys/dsl_dir.h>
  45 #include <sys/dsl_pool.h>
  46 #include <sys/dsl_synctask.h>
  47 #include <sys/dsl_prop.h>
  48 #include <sys/dmu_zfetch.h>
  49 #include <sys/zfs_ioctl.h>
  50 #include <sys/zap.h>
  51 #include <sys/zio_checksum.h>
  52 #include <sys/zio_compress.h>
  53 #include <sys/sa.h>
  54 #include <sys/zfeature.h>
  55 #include <sys/abd.h>
  56 #include <sys/brt.h>
  57 #include <sys/trace_zfs.h>
  58 #include <sys/zfs_racct.h>
  59 #include <sys/zfs_rlock.h>
  60 #ifdef _KERNEL
  61 #include <sys/vmsystm.h>
  62 #include <sys/zfs_znode.h>
  63 #endif
  64
  65 /*
  66  * Enable/disable nopwrite feature.
  67  */
  68 static int zfs_nopwrite_enabled = 1;
  69
  70 /*
  71  * Tunable to control percentage of dirtied L1 blocks from frees allowed into
  72  * one TXG. After this threshold is crossed, additional dirty blocks from frees
  73  * will wait until the next TXG.
  74  * A value of zero will disable this throttle.
  75  */
  76 static uint_t zfs_per_txg_dirty_frees_percent = 30;
  77
  78 /*
  79  * Enable/disable forcing txg sync when dirty checking for holes with lseek().
  80  * By default this is enabled to ensure accurate hole reporting, it can result
  81  * in a significant performance penalty for lseek(SEEK_HOLE) heavy workloads.
  82  * Disabling this option will result in holes never being reported in dirty
  83  * files which is always safe.
  84  */
  85 static int zfs_dmu_offset_next_sync = 1;
  86
  87 /*
  88  * Limit the amount we can prefetch with one call to this amount.  This
  89  * helps to limit the amount of memory that can be used by prefetching.
  90  * Larger objects should be prefetched a bit at a time.
  91  */
  92 #ifdef _ILP32
  93 uint_t dmu_prefetch_max = 8 * 1024 * 1024;
  94 #else
  95 uint_t dmu_prefetch_max = 8 * SPA_MAXBLOCKSIZE;
  96 #endif
  97
  98 const dmu_object_type_info_t dmu_ot[DMU_OT_NUMTYPES] = {
  99         {DMU_BSWAP_UINT8,  TRUE,  FALSE, FALSE, "unallocated"           },
 100         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "object directory"      },
 101         {DMU_BSWAP_UINT64, TRUE,  TRUE,  FALSE, "object array"          },
 102         {DMU_BSWAP_UINT8,  TRUE,  FALSE, FALSE, "packed nvlist"         },
 103         {DMU_BSWAP_UINT64, TRUE,  FALSE, FALSE, "packed nvlist size"    },
 104         {DMU_BSWAP_UINT64, TRUE,  FALSE, FALSE, "bpobj"                 },
 105         {DMU_BSWAP_UINT64, TRUE,  FALSE, FALSE, "bpobj header"          },
 106         {DMU_BSWAP_UINT64, TRUE,  FALSE, FALSE, "SPA space map header"  },
 107         {DMU_BSWAP_UINT64, TRUE,  FALSE, FALSE, "SPA space map"         },
 108         {DMU_BSWAP_UINT64, TRUE,  FALSE, TRUE,  "ZIL intent log"        },
 109         {DMU_BSWAP_DNODE,  TRUE,  FALSE, TRUE,  "DMU dnode"             },
 110         {DMU_BSWAP_OBJSET, TRUE,  TRUE,  FALSE, "DMU objset"            },
 111         {DMU_BSWAP_UINT64, TRUE,  TRUE,  FALSE, "DSL directory"         },
 112         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "DSL directory child map"},
 113         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "DSL dataset snap map"  },
 114         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "DSL props"             },
 115         {DMU_BSWAP_UINT64, TRUE,  TRUE,  FALSE, "DSL dataset"           },
 116         {DMU_BSWAP_ZNODE,  TRUE,  FALSE, FALSE, "ZFS znode"             },
 117         {DMU_BSWAP_OLDACL, TRUE,  FALSE, TRUE,  "ZFS V0 ACL"            },
 118         {DMU_BSWAP_UINT8,  FALSE, FALSE, TRUE,  "ZFS plain file"        },
 119         {DMU_BSWAP_ZAP,    TRUE,  FALSE, TRUE,  "ZFS directory"         },
 120         {DMU_BSWAP_ZAP,    TRUE,  FALSE, FALSE, "ZFS master node"       },
 121         {DMU_BSWAP_ZAP,    TRUE,  FALSE, TRUE,  "ZFS delete queue"      },
 122         {DMU_BSWAP_UINT8,  FALSE, FALSE, TRUE,  "zvol object"           },
 123         {DMU_BSWAP_ZAP,    TRUE,  FALSE, FALSE, "zvol prop"             },
 124         {DMU_BSWAP_UINT8,  FALSE, FALSE, TRUE,  "other uint8[]"         },
 125         {DMU_BSWAP_UINT64, FALSE, FALSE, TRUE,  "other uint64[]"        },
 126         {DMU_BSWAP_ZAP,    TRUE,  FALSE, FALSE, "other ZAP"             },
 127         {DMU_BSWAP_ZAP,    TRUE,  FALSE, FALSE, "persistent error log"  },
 128         {DMU_BSWAP_UINT8,  TRUE,  FALSE, FALSE, "SPA history"           },
 129         {DMU_BSWAP_UINT64, TRUE,  FALSE, FALSE, "SPA history offsets"   },
 130         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "Pool properties"       },
 131         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "DSL permissions"       },
 132         {DMU_BSWAP_ACL,    TRUE,  FALSE, TRUE,  "ZFS ACL"               },
 133         {DMU_BSWAP_UINT8,  TRUE,  FALSE, TRUE,  "ZFS SYSACL"            },
 134         {DMU_BSWAP_UINT8,  TRUE,  FALSE, TRUE,  "FUID table"            },
 135         {DMU_BSWAP_UINT64, TRUE,  FALSE, FALSE, "FUID table size"       },
 136         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "DSL dataset next clones"},
 137         {DMU_BSWAP_ZAP,    TRUE,  FALSE, FALSE, "scan work queue"       },
 138         {DMU_BSWAP_ZAP,    TRUE,  FALSE, TRUE,  "ZFS user/group/project used" },
 139         {DMU_BSWAP_ZAP,    TRUE,  FALSE, TRUE,  "ZFS user/group/project quota"},
 140         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "snapshot refcount tags"},
 141         {DMU_BSWAP_ZAP,    TRUE,  FALSE, FALSE, "DDT ZAP algorithm"     },
 142         {DMU_BSWAP_ZAP,    TRUE,  FALSE, FALSE, "DDT statistics"        },
 143         {DMU_BSWAP_UINT8,  TRUE,  FALSE, TRUE,  "System attributes"     },
 144         {DMU_BSWAP_ZAP,    TRUE,  FALSE, TRUE,  "SA master node"        },
 145         {DMU_BSWAP_ZAP,    TRUE,  FALSE, TRUE,  "SA attr registration"  },
 146         {DMU_BSWAP_ZAP,    TRUE,  FALSE, TRUE,  "SA attr layouts"       },
 147         {DMU_BSWAP_ZAP,    TRUE,  FALSE, FALSE, "scan translations"     },
 148         {DMU_BSWAP_UINT8,  FALSE, FALSE, TRUE,  "deduplicated block"    },
 149         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "DSL deadlist map"      },
 150         {DMU_BSWAP_UINT64, TRUE,  TRUE,  FALSE, "DSL deadlist map hdr"  },
 151         {DMU_BSWAP_ZAP,    TRUE,  TRUE,  FALSE, "DSL dir clones"        },
 152         {DMU_BSWAP_UINT64, TRUE,  FALSE, FALSE, "bpobj subobj"          }
 153 };
 154
 155 dmu_object_byteswap_info_t dmu_ot_byteswap[DMU_BSWAP_NUMFUNCS] = {
 156         {       byteswap_uint8_array,   "uint8"         },
 157         {       byteswap_uint16_array,  "uint16"        },
 158         {       byteswap_uint32_array,  "uint32"        },
 159         {       byteswap_uint64_array,  "uint64"        },
 160         {       zap_byteswap,           "zap"           },
 161         {       dnode_buf_byteswap,     "dnode"         },
 162         {       dmu_objset_byteswap,    "objset"        },
 163         {       zfs_znode_byteswap,     "znode"         },
 164         {       zfs_oldacl_byteswap,    "oldacl"        },
 165         {       zfs_acl_byteswap,       "acl"           }
 166 };
 167
 168 int
 169 dmu_buf_hold_noread_by_dnode(dnode_t *dn, uint64_t offset,
 170     const void *tag, dmu_buf_t **dbp)
 171 {
 172         uint64_t blkid;
 173         dmu_buf_impl_t *db;
 174
 175         rw_enter(&dn->dn_struct_rwlock, RW_READER);
 176         blkid = dbuf_whichblock(dn, 0, offset);
 177         db = dbuf_hold(dn, blkid, tag);
 178         rw_exit(&dn->dn_struct_rwlock);
 179
 180         if (db == NULL) {
 181                 *dbp = NULL;
 182                 return (SET_ERROR(EIO));
 183         }
 184
 185         *dbp = &db->db;
 186         return (0);
 187 }
 188
 189 int
 190 dmu_buf_hold_noread(objset_t *os, uint64_t object, uint64_t offset,
 191     const void *tag, dmu_buf_t **dbp)
 192 {
 193         dnode_t *dn;
 194         uint64_t blkid;
 195         dmu_buf_impl_t *db;
 196         int err;
 197
 198         err = dnode_hold(os, object, FTAG, &dn);
 199         if (err)
 200                 return (err);
 201         rw_enter(&dn->dn_struct_rwlock, RW_READER);
 202         blkid = dbuf_whichblock(dn, 0, offset);
 203         db = dbuf_hold(dn, blkid, tag);
 204         rw_exit(&dn->dn_struct_rwlock);
 205         dnode_rele(dn, FTAG);
 206
 207         if (db == NULL) {
 208                 *dbp = NULL;
 209                 return (SET_ERROR(EIO));
 210         }
 211
 212         *dbp = &db->db;
 213         return (err);
 214 }
 215
 216 int
 217 dmu_buf_hold_by_dnode(dnode_t *dn, uint64_t offset,
 218     const void *tag, dmu_buf_t **dbp, int flags)
 219 {
 220         int err;
 221         int db_flags = DB_RF_CANFAIL;
 222
 223         if (flags & DMU_READ_NO_PREFETCH)
 224                 db_flags |= DB_RF_NOPREFETCH;
 225         if (flags & DMU_READ_NO_DECRYPT)
 226                 db_flags |= DB_RF_NO_DECRYPT;
 227
 228         err = dmu_buf_hold_noread_by_dnode(dn, offset, tag, dbp);
 229         if (err == 0) {
 230                 dmu_buf_impl_t *db = (dmu_buf_impl_t *)(*dbp);
 231                 err = dbuf_read(db, NULL, db_flags);
 232                 if (err != 0) {
 233                         dbuf_rele(db, tag);
 234                         *dbp = NULL;
 235                 }
 236         }
 237
 238         return (err);
 239 }
 240
 241 int
 242 dmu_buf_hold(objset_t *os, uint64_t object, uint64_t offset,
 243     const void *tag, dmu_buf_t **dbp, int flags)
 244 {
 245         int err;
 246         int db_flags = DB_RF_CANFAIL;
 247
 248         if (flags & DMU_READ_NO_PREFETCH)
 249                 db_flags |= DB_RF_NOPREFETCH;
 250         if (flags & DMU_READ_NO_DECRYPT)
 251                 db_flags |= DB_RF_NO_DECRYPT;
 252
 253         err = dmu_buf_hold_noread(os, object, offset, tag, dbp);
 254         if (err == 0) {
 255                 dmu_buf_impl_t *db = (dmu_buf_impl_t *)(*dbp);
 256                 err = dbuf_read(db, NULL, db_flags);
 257                 if (err != 0) {
 258                         dbuf_rele(db, tag);
 259                         *dbp = NULL;
 260                 }
 261         }
 262
 263         return (err);
 264 }
 265
 266 int
 267 dmu_bonus_max(void)
 268 {
 269         return (DN_OLD_MAX_BONUSLEN);
 270 }
 271
 272 int
 273 dmu_set_bonus(dmu_buf_t *db_fake, int newsize, dmu_tx_t *tx)
 274 {
 275         dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
 276         dnode_t *dn;
 277         int error;
 278
 279         DB_DNODE_ENTER(db);
 280         dn = DB_DNODE(db);
 281
 282         if (dn->dn_bonus != db) {
 283                 error = SET_ERROR(EINVAL);
 284         } else if (newsize < 0 || newsize > db_fake->db_size) {
 285                 error = SET_ERROR(EINVAL);
 286         } else {
 287                 dnode_setbonuslen(dn, newsize, tx);
 288                 error = 0;
 289         }
 290
 291         DB_DNODE_EXIT(db);
 292         return (error);
 293 }
 294
 295 int
 296 dmu_set_bonustype(dmu_buf_t *db_fake, dmu_object_type_t type, dmu_tx_t *tx)
 297 {
 298         dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
 299         dnode_t *dn;
 300         int error;
 301
 302         DB_DNODE_ENTER(db);
 303         dn = DB_DNODE(db);
 304
 305         if (!DMU_OT_IS_VALID(type)) {
 306                 error = SET_ERROR(EINVAL);
 307         } else if (dn->dn_bonus != db) {
 308                 error = SET_ERROR(EINVAL);
 309         } else {
 310                 dnode_setbonus_type(dn, type, tx);
 311                 error = 0;
 312         }
 313
 314         DB_DNODE_EXIT(db);
 315         return (error);
 316 }
 317
 318 dmu_object_type_t
 319 dmu_get_bonustype(dmu_buf_t *db_fake)
 320 {
 321         dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
 322         dnode_t *dn;
 323         dmu_object_type_t type;
 324
 325         DB_DNODE_ENTER(db);
 326         dn = DB_DNODE(db);
 327         type = dn->dn_bonustype;
 328         DB_DNODE_EXIT(db);
 329
 330         return (type);
 331 }
 332
 333 int
 334 dmu_rm_spill(objset_t *os, uint64_t object, dmu_tx_t *tx)
 335 {
 336         dnode_t *dn;
 337         int error;
 338
 339         error = dnode_hold(os, object, FTAG, &dn);
 340         dbuf_rm_spill(dn, tx);
 341         rw_enter(&dn->dn_struct_rwlock, RW_WRITER);
 342         dnode_rm_spill(dn, tx);
 343         rw_exit(&dn->dn_struct_rwlock);
 344         dnode_rele(dn, FTAG);
 345         return (error);
 346 }
 347
 348 /*
 349  * Lookup and hold the bonus buffer for the provided dnode.  If the dnode
 350  * has not yet been allocated a new bonus dbuf a will be allocated.
 351  * Returns ENOENT, EIO, or 0.
 352  */
 353 int dmu_bonus_hold_by_dnode(dnode_t *dn, const void *tag, dmu_buf_t **dbp,
 354     uint32_t flags)
 355 {
 356         dmu_buf_impl_t *db;
 357         int error;
 358         uint32_t db_flags = DB_RF_MUST_SUCCEED;
 359
 360         if (flags & DMU_READ_NO_PREFETCH)
 361                 db_flags |= DB_RF_NOPREFETCH;
 362         if (flags & DMU_READ_NO_DECRYPT)
 363                 db_flags |= DB_RF_NO_DECRYPT;
 364
 365         rw_enter(&dn->dn_struct_rwlock, RW_READER);
 366         if (dn->dn_bonus == NULL) {
 367                 if (!rw_tryupgrade(&dn->dn_struct_rwlock)) {
 368                         rw_exit(&dn->dn_struct_rwlock);
 369                         rw_enter(&dn->dn_struct_rwlock, RW_WRITER);
 370                 }
 371                 if (dn->dn_bonus == NULL)
 372                         dbuf_create_bonus(dn);
 373         }
 374         db = dn->dn_bonus;
 375
 376         /* as long as the bonus buf is held, the dnode will be held */
 377         if (zfs_refcount_add(&db->db_holds, tag) == 1) {
 378                 VERIFY(dnode_add_ref(dn, db));
 379                 atomic_inc_32(&dn->dn_dbufs_count);
 380         }
 381
 382         /*
 383          * Wait to drop dn_struct_rwlock until after adding the bonus dbuf's
 384          * hold and incrementing the dbuf count to ensure that dnode_move() sees
 385          * a dnode hold for every dbuf.
 386          */
 387         rw_exit(&dn->dn_struct_rwlock);
 388
 389         error = dbuf_read(db, NULL, db_flags);
 390         if (error) {
 391                 dnode_evict_bonus(dn);
 392                 dbuf_rele(db, tag);
 393                 *dbp = NULL;
 394                 return (error);
 395         }
 396
 397         *dbp = &db->db;
 398         return (0);
 399 }
 400
 401 int
 402 dmu_bonus_hold(objset_t *os, uint64_t object, const void *tag, dmu_buf_t **dbp)
 403 {
 404         dnode_t *dn;
 405         int error;
 406
 407         error = dnode_hold(os, object, FTAG, &dn);
 408         if (error)
 409                 return (error);
 410
 411         error = dmu_bonus_hold_by_dnode(dn, tag, dbp, DMU_READ_NO_PREFETCH);
 412         dnode_rele(dn, FTAG);
 413
 414         return (error);
 415 }
 416
 417 /*
 418  * returns ENOENT, EIO, or 0.
 419  *
 420  * This interface will allocate a blank spill dbuf when a spill blk
 421  * doesn't already exist on the dnode.
 422  *
 423  * if you only want to find an already existing spill db, then
 424  * dmu_spill_hold_existing() should be used.
 425  */
 426 int
 427 dmu_spill_hold_by_dnode(dnode_t *dn, uint32_t flags, const void *tag,
 428     dmu_buf_t **dbp)
 429 {
 430         dmu_buf_impl_t *db = NULL;
 431         int err;
 432
 433         if ((flags & DB_RF_HAVESTRUCT) == 0)
 434                 rw_enter(&dn->dn_struct_rwlock, RW_READER);
 435
 436         db = dbuf_hold(dn, DMU_SPILL_BLKID, tag);
 437
 438         if ((flags & DB_RF_HAVESTRUCT) == 0)
 439                 rw_exit(&dn->dn_struct_rwlock);
 440
 441         if (db == NULL) {
 442                 *dbp = NULL;
 443                 return (SET_ERROR(EIO));
 444         }
 445         err = dbuf_read(db, NULL, flags);
 446         if (err == 0)
 447                 *dbp = &db->db;
 448         else {
 449                 dbuf_rele(db, tag);
 450                 *dbp = NULL;
 451         }
 452         return (err);
 453 }
 454
 455 int
 456 dmu_spill_hold_existing(dmu_buf_t *bonus, const void *tag, dmu_buf_t **dbp)
 457 {
 458         dmu_buf_impl_t *db = (dmu_buf_impl_t *)bonus;
 459         dnode_t *dn;
 460         int err;
 461
 462         DB_DNODE_ENTER(db);
 463         dn = DB_DNODE(db);
 464
 465         if (spa_version(dn->dn_objset->os_spa) < SPA_VERSION_SA) {
 466                 err = SET_ERROR(EINVAL);
 467         } else {
 468                 rw_enter(&dn->dn_struct_rwlock, RW_READER);
 469
 470                 if (!dn->dn_have_spill) {
 471                         err = SET_ERROR(ENOENT);
 472                 } else {
 473                         err = dmu_spill_hold_by_dnode(dn,
 474                             DB_RF_HAVESTRUCT | DB_RF_CANFAIL, tag, dbp);
 475                 }
 476
 477                 rw_exit(&dn->dn_struct_rwlock);
 478         }
 479
 480         DB_DNODE_EXIT(db);
 481         return (err);
 482 }
 483
 484 int
 485 dmu_spill_hold_by_bonus(dmu_buf_t *bonus, uint32_t flags, const void *tag,
 486     dmu_buf_t **dbp)
 487 {
 488         dmu_buf_impl_t *db = (dmu_buf_impl_t *)bonus;
 489         dnode_t *dn;
 490         int err;
 491         uint32_t db_flags = DB_RF_CANFAIL;
 492
 493         if (flags & DMU_READ_NO_DECRYPT)
 494                 db_flags |= DB_RF_NO_DECRYPT;
 495
 496         DB_DNODE_ENTER(db);
 497         dn = DB_DNODE(db);
 498         err = dmu_spill_hold_by_dnode(dn, db_flags, tag, dbp);
 499         DB_DNODE_EXIT(db);
 500
 501         return (err);
 502 }
 503
 504 /*
 505  * Note: longer-term, we should modify all of the dmu_buf_*() interfaces
 506  * to take a held dnode rather than <os, object> -- the lookup is wasteful,
 507  * and can induce severe lock contention when writing to several files
 508  * whose dnodes are in the same block.
 509  */
 510 int
 511 dmu_buf_hold_array_by_dnode(dnode_t *dn, uint64_t offset, uint64_t length,
 512     boolean_t read, const void *tag, int *numbufsp, dmu_buf_t ***dbpp,
 513     uint32_t flags)
 514 {
 515         dmu_buf_t **dbp;
 516         zstream_t *zs = NULL;
 517         uint64_t blkid, nblks, i;
 518         uint32_t dbuf_flags;
 519         int err;
 520         zio_t *zio = NULL;
 521         boolean_t missed = B_FALSE;
 522
 523         ASSERT(!read || length <= DMU_MAX_ACCESS);
 524
 525         /*
 526          * Note: We directly notify the prefetch code of this read, so that
 527          * we can tell it about the multi-block read.  dbuf_read() only knows
 528          * about the one block it is accessing.
 529          */
 530         dbuf_flags = DB_RF_CANFAIL | DB_RF_NEVERWAIT | DB_RF_HAVESTRUCT |
 531             DB_RF_NOPREFETCH;
 532
 533         if ((flags & DMU_READ_NO_DECRYPT) != 0)
 534                 dbuf_flags |= DB_RF_NO_DECRYPT;
 535
 536         rw_enter(&dn->dn_struct_rwlock, RW_READER);
 537         if (dn->dn_datablkshift) {
 538                 int blkshift = dn->dn_datablkshift;
 539                 nblks = (P2ROUNDUP(offset + length, 1ULL << blkshift) -
 540                     P2ALIGN(offset, 1ULL << blkshift)) >> blkshift;
 541         } else {
 542                 if (offset + length > dn->dn_datablksz) {
 543                         zfs_panic_recover("zfs: accessing past end of object "
 544                             "%llx/%llx (size=%u access=%llu+%llu)",
 545                             (longlong_t)dn->dn_objset->
 546                             os_dsl_dataset->ds_object,
 547                             (longlong_t)dn->dn_object, dn->dn_datablksz,
 548                             (longlong_t)offset, (longlong_t)length);
 549                         rw_exit(&dn->dn_struct_rwlock);
 550                         return (SET_ERROR(EIO));
 551                 }
 552                 nblks = 1;
 553         }
 554         dbp = kmem_zalloc(sizeof (dmu_buf_t *) * nblks, KM_SLEEP);
 555
 556         if (read)
 557                 zio = zio_root(dn->dn_objset->os_spa, NULL, NULL,
 558                     ZIO_FLAG_CANFAIL);
 559         blkid = dbuf_whichblock(dn, 0, offset);
 560         if ((flags & DMU_READ_NO_PREFETCH) == 0) {
 561                 /*
 562                  * Prepare the zfetch before initiating the demand reads, so
 563                  * that if multiple threads block on same indirect block, we
 564                  * base predictions on the original less racy request order.
 565                  */
 566                 zs = dmu_zfetch_prepare(&dn->dn_zfetch, blkid, nblks, read,
 567                     B_TRUE);
 568         }
 569         for (i = 0; i < nblks; i++) {
 570                 dmu_buf_impl_t *db = dbuf_hold(dn, blkid + i, tag);
 571                 if (db == NULL) {
 572                         if (zs) {
 573                                 dmu_zfetch_run(&dn->dn_zfetch, zs, missed,
 574                                     B_TRUE);
 575                         }
 576                         rw_exit(&dn->dn_struct_rwlock);
 577                         dmu_buf_rele_array(dbp, nblks, tag);
 578                         if (read)
 579                                 zio_nowait(zio);
 580                         return (SET_ERROR(EIO));
 581                 }
 582
 583                 /*
 584                  * Initiate async demand data read.
 585                  * We check the db_state after calling dbuf_read() because
 586                  * (1) dbuf_read() may change the state to CACHED due to a
 587                  * hit in the ARC, and (2) on a cache miss, a child will
 588                  * have been added to "zio" but not yet completed, so the
 589                  * state will not yet be CACHED.
 590                  */
 591                 if (read) {
 592                         if (i == nblks - 1 && blkid + i < dn->dn_maxblkid &&
 593                             offset + length < db->db.db_offset +
 594                             db->db.db_size) {
 595                                 if (offset <= db->db.db_offset)
 596                                         dbuf_flags |= DB_RF_PARTIAL_FIRST;
 597                                 else
 598                                         dbuf_flags |= DB_RF_PARTIAL_MORE;
 599                         }
 600                         (void) dbuf_read(db, zio, dbuf_flags);
 601                         if (db->db_state != DB_CACHED)
 602                                 missed = B_TRUE;
 603                 }
 604                 dbp[i] = &db->db;
 605         }
 606
 607         if (!read)
 608                 zfs_racct_write(length, nblks);
 609
 610         if (zs)
 611                 dmu_zfetch_run(&dn->dn_zfetch, zs, missed, B_TRUE);
 612         rw_exit(&dn->dn_struct_rwlock);
 613
 614         if (read) {
 615                 /* wait for async read i/o */
 616                 err = zio_wait(zio);
 617                 if (err) {
 618                         dmu_buf_rele_array(dbp, nblks, tag);
 619                         return (err);
 620                 }
 621
 622                 /* wait for other io to complete */
 623                 for (i = 0; i < nblks; i++) {
 624                         dmu_buf_impl_t *db = (dmu_buf_impl_t *)dbp[i];
 625                         mutex_enter(&db->db_mtx);
 626                         while (db->db_state == DB_READ ||
 627                             db->db_state == DB_FILL)
 628                                 cv_wait(&db->db_changed, &db->db_mtx);
 629                         if (db->db_state == DB_UNCACHED)
 630                                 err = SET_ERROR(EIO);
 631                         mutex_exit(&db->db_mtx);
 632                         if (err) {
 633                                 dmu_buf_rele_array(dbp, nblks, tag);
 634                                 return (err);
 635                         }
 636                 }
 637         }
 638
 639         *numbufsp = nblks;
 640         *dbpp = dbp;
 641         return (0);
 642 }
 643
 644 int
 645 dmu_buf_hold_array(objset_t *os, uint64_t object, uint64_t offset,
 646     uint64_t length, int read, const void *tag, int *numbufsp,
 647     dmu_buf_t ***dbpp)
 648 {
 649         dnode_t *dn;
 650         int err;
 651
 652         err = dnode_hold(os, object, FTAG, &dn);
 653         if (err)
 654                 return (err);
 655
 656         err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,
 657             numbufsp, dbpp, DMU_READ_PREFETCH);
 658
 659         dnode_rele(dn, FTAG);
 660
 661         return (err);
 662 }
 663
 664 int
 665 dmu_buf_hold_array_by_bonus(dmu_buf_t *db_fake, uint64_t offset,
 666     uint64_t length, boolean_t read, const void *tag, int *numbufsp,
 667     dmu_buf_t ***dbpp)
 668 {
 669         dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
 670         dnode_t *dn;
 671         int err;
 672
 673         DB_DNODE_ENTER(db);
 674         dn = DB_DNODE(db);
 675         err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,
 676             numbufsp, dbpp, DMU_READ_PREFETCH);
 677         DB_DNODE_EXIT(db);
 678
 679         return (err);
 680 }
 681
 682 void
 683 dmu_buf_rele_array(dmu_buf_t **dbp_fake, int numbufs, const void *tag)
 684 {
 685         int i;
 686         dmu_buf_impl_t **dbp = (dmu_buf_impl_t **)dbp_fake;
 687
 688         if (numbufs == 0)
 689                 return;
 690
 691         for (i = 0; i < numbufs; i++) {
 692                 if (dbp[i])
 693                         dbuf_rele(dbp[i], tag);
 694         }
 695
 696         kmem_free(dbp, sizeof (dmu_buf_t *) * numbufs);
 697 }
 698
 699 /*
 700  * Issue prefetch I/Os for the given blocks.  If level is greater than 0, the
 701  * indirect blocks prefetched will be those that point to the blocks containing
 702  * the data starting at offset, and continuing to offset + len.  If the range
 703  * it too long, prefetch the first dmu_prefetch_max bytes as requested, while
 704  * for the rest only a higher level, also fitting within dmu_prefetch_max.  It
 705  * should primarily help random reads, since for long sequential reads there is
 706  * a speculative prefetcher.
 707  *
 708  * Note that if the indirect blocks above the blocks being prefetched are not
 709  * in cache, they will be asynchronously read in.  Dnode read by dnode_hold()
 710  * is currently synchronous.
 711  */
 712 void
 713 dmu_prefetch(objset_t *os, uint64_t object, int64_t level, uint64_t offset,
 714     uint64_t len, zio_priority_t pri)
 715 {
 716         dnode_t *dn;
 717
 718         if (dmu_prefetch_max == 0 || len == 0) {
 719                 dmu_prefetch_dnode(os, object, pri);
 720                 return;
 721         }
 722
 723         if (dnode_hold(os, object, FTAG, &dn) != 0)
 724                 return;
 725
 726         dmu_prefetch_by_dnode(dn, level, offset, len, pri);
 727
 728         dnode_rele(dn, FTAG);
 729 }
 730
 731 void
 732 dmu_prefetch_by_dnode(dnode_t *dn, int64_t level, uint64_t offset,
 733     uint64_t len, zio_priority_t pri)
 734 {
 735         int64_t level2 = level;
 736         uint64_t start, end, start2, end2;
 737
 738         /*
 739          * Depending on len we may do two prefetches: blocks [start, end) at
 740          * level, and following blocks [start2, end2) at higher level2.
 741          */
 742         rw_enter(&dn->dn_struct_rwlock, RW_READER);
 743         if (dn->dn_datablkshift != 0) {
 744                 /*
 745                  * The object has multiple blocks.  Calculate the full range
 746                  * of blocks [start, end2) and then split it into two parts,
 747                  * so that the first [start, end) fits into dmu_prefetch_max.
 748                  */
 749                 start = dbuf_whichblock(dn, level, offset);
 750                 end2 = dbuf_whichblock(dn, level, offset + len - 1) + 1;
 751                 uint8_t ibs = dn->dn_indblkshift;
 752                 uint8_t bs = (level == 0) ? dn->dn_datablkshift : ibs;
 753                 uint_t limit = P2ROUNDUP(dmu_prefetch_max, 1 << bs) >> bs;
 754                 start2 = end = MIN(end2, start + limit);
 755
 756                 /*
 757                  * Find level2 where [start2, end2) fits into dmu_prefetch_max.
 758                  */
 759                 uint8_t ibps = ibs - SPA_BLKPTRSHIFT;
 760                 limit = P2ROUNDUP(dmu_prefetch_max, 1 << ibs) >> ibs;
 761                 do {
 762                         level2++;
 763                         start2 = P2ROUNDUP(start2, 1 << ibps) >> ibps;
 764                         end2 = P2ROUNDUP(end2, 1 << ibps) >> ibps;
 765                 } while (end2 - start2 > limit);
 766         } else {
 767                 /* There is only one block.  Prefetch it or nothing. */
 768                 start = start2 = end2 = 0;
 769                 end = start + (level == 0 && offset < dn->dn_datablksz);
 770         }
 771
 772         for (uint64_t i = start; i < end; i++)
 773                 dbuf_prefetch(dn, level, i, pri, 0);
 774         for (uint64_t i = start2; i < end2; i++)
 775                 dbuf_prefetch(dn, level2, i, pri, 0);
 776         rw_exit(&dn->dn_struct_rwlock);
 777 }
 778
 779 /*
 780  * Issue prefetch I/Os for the given object's dnode.
 781  */
 782 void
 783 dmu_prefetch_dnode(objset_t *os, uint64_t object, zio_priority_t pri)
 784 {
 785         if (object == 0 || object >= DN_MAX_OBJECT)
 786                 return;
 787
 788         dnode_t *dn = DMU_META_DNODE(os);
 789         rw_enter(&dn->dn_struct_rwlock, RW_READER);
 790         uint64_t blkid = dbuf_whichblock(dn, 0, object * sizeof (dnode_phys_t));
 791         dbuf_prefetch(dn, 0, blkid, pri, 0);
 792         rw_exit(&dn->dn_struct_rwlock);
 793 }
 794
 795 /*
 796  * Get the next "chunk" of file data to free.  We traverse the file from
 797  * the end so that the file gets shorter over time (if we crashes in the
 798  * middle, this will leave us in a better state).  We find allocated file
 799  * data by simply searching the allocated level 1 indirects.
 800  *
 801  * On input, *start should be the first offset that does not need to be
 802  * freed (e.g. "offset + length").  On return, *start will be the first
 803  * offset that should be freed and l1blks is set to the number of level 1
 804  * indirect blocks found within the chunk.
 805  */
 806 static int
 807 get_next_chunk(dnode_t *dn, uint64_t *start, uint64_t minimum, uint64_t *l1blks)
 808 {
 809         uint64_t blks;
 810         uint64_t maxblks = DMU_MAX_ACCESS >> (dn->dn_indblkshift + 1);
 811         /* bytes of data covered by a level-1 indirect block */
 812         uint64_t iblkrange = (uint64_t)dn->dn_datablksz *
 813             EPB(dn->dn_indblkshift, SPA_BLKPTRSHIFT);
 814
 815         ASSERT3U(minimum, <=, *start);
 816
 817         /*
 818          * Check if we can free the entire range assuming that all of the
 819          * L1 blocks in this range have data. If we can, we use this
 820          * worst case value as an estimate so we can avoid having to look
 821          * at the object's actual data.
 822          */
 823         uint64_t total_l1blks =
 824             (roundup(*start, iblkrange) - (minimum / iblkrange * iblkrange)) /
 825             iblkrange;
 826         if (total_l1blks <= maxblks) {
 827                 *l1blks = total_l1blks;
 828                 *start = minimum;
 829                 return (0);
 830         }
 831         ASSERT(ISP2(iblkrange));
 832
 833         for (blks = 0; *start > minimum && blks < maxblks; blks++) {
 834                 int err;
 835
 836                 /*
 837                  * dnode_next_offset(BACKWARDS) will find an allocated L1
 838                  * indirect block at or before the input offset.  We must
 839                  * decrement *start so that it is at the end of the region
 840                  * to search.
 841                  */
 842                 (*start)--;
 843
 844                 err = dnode_next_offset(dn,
 845                     DNODE_FIND_BACKWARDS, start, 2, 1, 0);
 846
 847                 /* if there are no indirect blocks before start, we are done */
 848                 if (err == ESRCH) {
 849                         *start = minimum;
 850                         break;
 851                 } else if (err != 0) {
 852                         *l1blks = blks;
 853                         return (err);
 854                 }
 855
 856                 /* set start to the beginning of this L1 indirect */
 857                 *start = P2ALIGN(*start, iblkrange);
 858         }
 859         if (*start < minimum)
 860                 *start = minimum;
 861         *l1blks = blks;
 862
 863         return (0);
 864 }
 865
 866 /*
 867  * If this objset is of type OST_ZFS return true if vfs's unmounted flag is set,
 868  * otherwise return false.
 869  * Used below in dmu_free_long_range_impl() to enable abort when unmounting
 870  */
 871 static boolean_t
 872 dmu_objset_zfs_unmounting(objset_t *os)
 873 {
 874 #ifdef _KERNEL
 875         if (dmu_objset_type(os) == DMU_OST_ZFS)
 876                 return (zfs_get_vfs_flag_unmounted(os));
 877 #else
 878         (void) os;
 879 #endif
 880         return (B_FALSE);
 881 }
 882
 883 static int
 884 dmu_free_long_range_impl(objset_t *os, dnode_t *dn, uint64_t offset,
 885     uint64_t length)
 886 {
 887         uint64_t object_size;
 888         int err;
 889         uint64_t dirty_frees_threshold;
 890         dsl_pool_t *dp = dmu_objset_pool(os);
 891
 892         if (dn == NULL)
 893                 return (SET_ERROR(EINVAL));
 894
 895         object_size = (dn->dn_maxblkid + 1) * dn->dn_datablksz;
 896         if (offset >= object_size)
 897                 return (0);
 898
 899         if (zfs_per_txg_dirty_frees_percent <= 100)
 900                 dirty_frees_threshold =
 901                     zfs_per_txg_dirty_frees_percent * zfs_dirty_data_max / 100;
 902         else
 903                 dirty_frees_threshold = zfs_dirty_data_max / 20;
 904
 905         if (length == DMU_OBJECT_END || offset + length > object_size)
 906                 length = object_size - offset;
 907
 908         while (length != 0) {
 909                 uint64_t chunk_end, chunk_begin, chunk_len;
 910                 uint64_t l1blks;
 911                 dmu_tx_t *tx;
 912
 913                 if (dmu_objset_zfs_unmounting(dn->dn_objset))
 914                         return (SET_ERROR(EINTR));
 915
 916                 chunk_end = chunk_begin = offset + length;
 917
 918                 /* move chunk_begin backwards to the beginning of this chunk */
 919                 err = get_next_chunk(dn, &chunk_begin, offset, &l1blks);
 920                 if (err)
 921                         return (err);
 922                 ASSERT3U(chunk_begin, >=, offset);
 923                 ASSERT3U(chunk_begin, <=, chunk_end);
 924
 925                 chunk_len = chunk_end - chunk_begin;
 926
 927                 tx = dmu_tx_create(os);
 928                 dmu_tx_hold_free(tx, dn->dn_object, chunk_begin, chunk_len);
 929
 930                 /*
 931                  * Mark this transaction as typically resulting in a net
 932                  * reduction in space used.
 933                  */
 934                 dmu_tx_mark_netfree(tx);
 935                 err = dmu_tx_assign(tx, TXG_WAIT);
 936                 if (err) {
 937                         dmu_tx_abort(tx);
 938                         return (err);
 939                 }
 940
 941                 uint64_t txg = dmu_tx_get_txg(tx);
 942
 943                 mutex_enter(&dp->dp_lock);
 944                 uint64_t long_free_dirty =
 945                     dp->dp_long_free_dirty_pertxg[txg & TXG_MASK];
 946                 mutex_exit(&dp->dp_lock);
 947
 948                 /*
 949                  * To avoid filling up a TXG with just frees, wait for
 950                  * the next TXG to open before freeing more chunks if
 951                  * we have reached the threshold of frees.
 952                  */
 953                 if (dirty_frees_threshold != 0 &&
 954                     long_free_dirty >= dirty_frees_threshold) {
 955                         DMU_TX_STAT_BUMP(dmu_tx_dirty_frees_delay);
 956                         dmu_tx_commit(tx);
 957                         txg_wait_open(dp, 0, B_TRUE);
 958                         continue;
 959                 }
 960
 961                 /*
 962                  * In order to prevent unnecessary write throttling, for each
 963                  * TXG, we track the cumulative size of L1 blocks being dirtied
 964                  * in dnode_free_range() below. We compare this number to a
 965                  * tunable threshold, past which we prevent new L1 dirty freeing
 966                  * blocks from being added into the open TXG. See
 967                  * dmu_free_long_range_impl() for details. The threshold
 968                  * prevents write throttle activation due to dirty freeing L1
 969                  * blocks taking up a large percentage of zfs_dirty_data_max.
 970                  */
 971                 mutex_enter(&dp->dp_lock);
 972                 dp->dp_long_free_dirty_pertxg[txg & TXG_MASK] +=
 973                     l1blks << dn->dn_indblkshift;
 974                 mutex_exit(&dp->dp_lock);
 975                 DTRACE_PROBE3(free__long__range,
 976                     uint64_t, long_free_dirty, uint64_t, chunk_len,
 977                     uint64_t, txg);
 978                 dnode_free_range(dn, chunk_begin, chunk_len, tx);
 979
 980                 dmu_tx_commit(tx);
 981
 982                 length -= chunk_len;
 983         }
 984         return (0);
 985 }
 986
 987 int
 988 dmu_free_long_range(objset_t *os, uint64_t object,
 989     uint64_t offset, uint64_t length)
 990 {
 991         dnode_t *dn;
 992         int err;
 993
 994         err = dnode_hold(os, object, FTAG, &dn);
 995         if (err != 0)
 996                 return (err);
 997         err = dmu_free_long_range_impl(os, dn, offset, length);
 998
 999         /*
1000          * It is important to zero out the maxblkid when freeing the entire
1001          * file, so that (a) subsequent calls to dmu_free_long_range_impl()
1002          * will take the fast path, and (b) dnode_reallocate() can verify
1003          * that the entire file has been freed.
1004          */
1005         if (err == 0 && offset == 0 && length == DMU_OBJECT_END)
1006                 dn->dn_maxblkid = 0;
1007
1008         dnode_rele(dn, FTAG);
1009         return (err);
1010 }
1011
1012 int
1013 dmu_free_long_object(objset_t *os, uint64_t object)
1014 {
1015         dmu_tx_t *tx;
1016         int err;
1017
1018         err = dmu_free_long_range(os, object, 0, DMU_OBJECT_END);
1019         if (err != 0)
1020                 return (err);
1021
1022         tx = dmu_tx_create(os);
1023         dmu_tx_hold_bonus(tx, object);
1024         dmu_tx_hold_free(tx, object, 0, DMU_OBJECT_END);
1025         dmu_tx_mark_netfree(tx);
1026         err = dmu_tx_assign(tx, TXG_WAIT);
1027         if (err == 0) {
1028                 err = dmu_object_free(os, object, tx);
1029                 dmu_tx_commit(tx);
1030         } else {
1031                 dmu_tx_abort(tx);
1032         }
1033
1034         return (err);
1035 }
1036
1037 int
1038 dmu_free_range(objset_t *os, uint64_t object, uint64_t offset,
1039     uint64_t size, dmu_tx_t *tx)
1040 {
1041         dnode_t *dn;
1042         int err = dnode_hold(os, object, FTAG, &dn);
1043         if (err)
1044                 return (err);
1045         ASSERT(offset < UINT64_MAX);
1046         ASSERT(size == DMU_OBJECT_END || size <= UINT64_MAX - offset);
1047         dnode_free_range(dn, offset, size, tx);
1048         dnode_rele(dn, FTAG);
1049         return (0);
1050 }
1051
1052 static int
1053 dmu_read_impl(dnode_t *dn, uint64_t offset, uint64_t size,
1054     void *buf, uint32_t flags)
1055 {
1056         dmu_buf_t **dbp;
1057         int numbufs, err = 0;
1058
1059         /*
1060          * Deal with odd block sizes, where there can't be data past the first
1061          * block.  If we ever do the tail block optimization, we will need to
1062          * handle that here as well.
1063          */
1064         if (dn->dn_maxblkid == 0) {
1065                 uint64_t newsz = offset > dn->dn_datablksz ? 0 :
1066                     MIN(size, dn->dn_datablksz - offset);
1067                 memset((char *)buf + newsz, 0, size - newsz);
1068                 size = newsz;
1069         }
1070
1071         while (size > 0) {
1072                 uint64_t mylen = MIN(size, DMU_MAX_ACCESS / 2);
1073                 int i;
1074
1075                 /*
1076                  * NB: we could do this block-at-a-time, but it's nice
1077                  * to be reading in parallel.
1078                  */
1079                 err = dmu_buf_hold_array_by_dnode(dn, offset, mylen,
1080                     TRUE, FTAG, &numbufs, &dbp, flags);
1081                 if (err)
1082                         break;
1083
1084                 for (i = 0; i < numbufs; i++) {
1085                         uint64_t tocpy;
1086                         int64_t bufoff;
1087                         dmu_buf_t *db = dbp[i];
1088
1089                         ASSERT(size > 0);
1090
1091                         bufoff = offset - db->db_offset;
1092                         tocpy = MIN(db->db_size - bufoff, size);
1093
1094                         (void) memcpy(buf, (char *)db->db_data + bufoff, tocpy);
1095
1096                         offset += tocpy;
1097                         size -= tocpy;
1098                         buf = (char *)buf + tocpy;
1099                 }
1100                 dmu_buf_rele_array(dbp, numbufs, FTAG);
1101         }
1102         return (err);
1103 }
1104
1105 int
1106 dmu_read(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
1107     void *buf, uint32_t flags)
1108 {
1109         dnode_t *dn;
1110         int err;
1111
1112         err = dnode_hold(os, object, FTAG, &dn);
1113         if (err != 0)
1114                 return (err);
1115
1116         err = dmu_read_impl(dn, offset, size, buf, flags);
1117         dnode_rele(dn, FTAG);
1118         return (err);
1119 }
1120
1121 int
1122 dmu_read_by_dnode(dnode_t *dn, uint64_t offset, uint64_t size, void *buf,
1123     uint32_t flags)
1124 {
1125         return (dmu_read_impl(dn, offset, size, buf, flags));
1126 }
1127
1128 static void
1129 dmu_write_impl(dmu_buf_t **dbp, int numbufs, uint64_t offset, uint64_t size,
1130     const void *buf, dmu_tx_t *tx)
1131 {
1132         int i;
1133
1134         for (i = 0; i < numbufs; i++) {
1135                 uint64_t tocpy;
1136                 int64_t bufoff;
1137                 dmu_buf_t *db = dbp[i];
1138
1139                 ASSERT(size > 0);
1140
1141                 bufoff = offset - db->db_offset;
1142                 tocpy = MIN(db->db_size - bufoff, size);
1143
1144                 ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);
1145
1146                 if (tocpy == db->db_size)
1147                         dmu_buf_will_fill(db, tx, B_FALSE);
1148                 else
1149                         dmu_buf_will_dirty(db, tx);
1150
1151                 (void) memcpy((char *)db->db_data + bufoff, buf, tocpy);
1152
1153                 if (tocpy == db->db_size)
1154                         dmu_buf_fill_done(db, tx, B_FALSE);
1155
1156                 offset += tocpy;
1157                 size -= tocpy;
1158                 buf = (char *)buf + tocpy;
1159         }
1160 }
1161
1162 void
1163 dmu_write(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
1164     const void *buf, dmu_tx_t *tx)
1165 {
1166         dmu_buf_t **dbp;
1167         int numbufs;
1168
1169         if (size == 0)
1170                 return;
1171
1172         VERIFY0(dmu_buf_hold_array(os, object, offset, size,
1173             FALSE, FTAG, &numbufs, &dbp));
1174         dmu_write_impl(dbp, numbufs, offset, size, buf, tx);
1175         dmu_buf_rele_array(dbp, numbufs, FTAG);
1176 }
1177
1178 /*
1179  * Note: Lustre is an external consumer of this interface.
1180  */
1181 void
1182 dmu_write_by_dnode(dnode_t *dn, uint64_t offset, uint64_t size,
1183     const void *buf, dmu_tx_t *tx)
1184 {
1185         dmu_buf_t **dbp;
1186         int numbufs;
1187
1188         if (size == 0)
1189                 return;
1190
1191         VERIFY0(dmu_buf_hold_array_by_dnode(dn, offset, size,
1192             FALSE, FTAG, &numbufs, &dbp, DMU_READ_PREFETCH));
1193         dmu_write_impl(dbp, numbufs, offset, size, buf, tx);
1194         dmu_buf_rele_array(dbp, numbufs, FTAG);
1195 }
1196
1197 void
1198 dmu_prealloc(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
1199     dmu_tx_t *tx)
1200 {
1201         dmu_buf_t **dbp;
1202         int numbufs, i;
1203
1204         if (size == 0)
1205                 return;
1206
1207         VERIFY(0 == dmu_buf_hold_array(os, object, offset, size,
1208             FALSE, FTAG, &numbufs, &dbp));
1209
1210         for (i = 0; i < numbufs; i++) {
1211                 dmu_buf_t *db = dbp[i];
1212
1213                 dmu_buf_will_not_fill(db, tx);
1214         }
1215         dmu_buf_rele_array(dbp, numbufs, FTAG);
1216 }
1217
1218 void
1219 dmu_write_embedded(objset_t *os, uint64_t object, uint64_t offset,
1220     void *data, uint8_t etype, uint8_t comp, int uncompressed_size,
1221     int compressed_size, int byteorder, dmu_tx_t *tx)
1222 {
1223         dmu_buf_t *db;
1224
1225         ASSERT3U(etype, <, NUM_BP_EMBEDDED_TYPES);
1226         ASSERT3U(comp, <, ZIO_COMPRESS_FUNCTIONS);
1227         VERIFY0(dmu_buf_hold_noread(os, object, offset,
1228             FTAG, &db));
1229
1230         dmu_buf_write_embedded(db,
1231             data, (bp_embedded_type_t)etype, (enum zio_compress)comp,
1232             uncompressed_size, compressed_size, byteorder, tx);
1233
1234         dmu_buf_rele(db, FTAG);
1235 }
1236
1237 void
1238 dmu_redact(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,
1239     dmu_tx_t *tx)
1240 {
1241         int numbufs, i;
1242         dmu_buf_t **dbp;
1243
1244         VERIFY0(dmu_buf_hold_array(os, object, offset, size, FALSE, FTAG,
1245             &numbufs, &dbp));
1246         for (i = 0; i < numbufs; i++)
1247                 dmu_buf_redact(dbp[i], tx);
1248         dmu_buf_rele_array(dbp, numbufs, FTAG);
1249 }
1250
1251 #ifdef _KERNEL
1252 int
1253 dmu_read_uio_dnode(dnode_t *dn, zfs_uio_t *uio, uint64_t size)
1254 {
1255         dmu_buf_t **dbp;
1256         int numbufs, i, err;
1257
1258         /*
1259          * NB: we could do this block-at-a-time, but it's nice
1260          * to be reading in parallel.
1261          */
1262         err = dmu_buf_hold_array_by_dnode(dn, zfs_uio_offset(uio), size,
1263             TRUE, FTAG, &numbufs, &dbp, 0);
1264         if (err)
1265                 return (err);
1266
1267         for (i = 0; i < numbufs; i++) {
1268                 uint64_t tocpy;
1269                 int64_t bufoff;
1270                 dmu_buf_t *db = dbp[i];
1271
1272                 ASSERT(size > 0);
1273
1274                 bufoff = zfs_uio_offset(uio) - db->db_offset;
1275                 tocpy = MIN(db->db_size - bufoff, size);
1276
1277                 err = zfs_uio_fault_move((char *)db->db_data + bufoff, tocpy,
1278                     UIO_READ, uio);
1279
1280                 if (err)
1281                         break;
1282
1283                 size -= tocpy;
1284         }
1285         dmu_buf_rele_array(dbp, numbufs, FTAG);
1286
1287         return (err);
1288 }
1289
1290 /*
1291  * Read 'size' bytes into the uio buffer.
1292  * From object zdb->db_object.
1293  * Starting at zfs_uio_offset(uio).
1294  *
1295  * If the caller already has a dbuf in the target object
1296  * (e.g. its bonus buffer), this routine is faster than dmu_read_uio(),
1297  * because we don't have to find the dnode_t for the object.
1298  */
1299 int
1300 dmu_read_uio_dbuf(dmu_buf_t *zdb, zfs_uio_t *uio, uint64_t size)
1301 {
1302         dmu_buf_impl_t *db = (dmu_buf_impl_t *)zdb;
1303         dnode_t *dn;
1304         int err;
1305
1306         if (size == 0)
1307                 return (0);
1308
1309         DB_DNODE_ENTER(db);
1310         dn = DB_DNODE(db);
1311         err = dmu_read_uio_dnode(dn, uio, size);
1312         DB_DNODE_EXIT(db);
1313
1314         return (err);
1315 }
1316
1317 /*
1318  * Read 'size' bytes into the uio buffer.
1319  * From the specified object
1320  * Starting at offset zfs_uio_offset(uio).
1321  */
1322 int
1323 dmu_read_uio(objset_t *os, uint64_t object, zfs_uio_t *uio, uint64_t size)
1324 {
1325         dnode_t *dn;
1326         int err;
1327
1328         if (size == 0)
1329                 return (0);
1330
1331         err = dnode_hold(os, object, FTAG, &dn);
1332         if (err)
1333                 return (err);
1334
1335         err = dmu_read_uio_dnode(dn, uio, size);
1336
1337         dnode_rele(dn, FTAG);
1338
1339         return (err);
1340 }
1341
1342 int
1343 dmu_write_uio_dnode(dnode_t *dn, zfs_uio_t *uio, uint64_t size, dmu_tx_t *tx)
1344 {
1345         dmu_buf_t **dbp;
1346         int numbufs;
1347         int err = 0;
1348         int i;
1349
1350         err = dmu_buf_hold_array_by_dnode(dn, zfs_uio_offset(uio), size,
1351             FALSE, FTAG, &numbufs, &dbp, DMU_READ_PREFETCH);
1352         if (err)
1353                 return (err);
1354
1355         for (i = 0; i < numbufs; i++) {
1356                 uint64_t tocpy;
1357                 int64_t bufoff;
1358                 dmu_buf_t *db = dbp[i];
1359
1360                 ASSERT(size > 0);
1361
1362                 offset_t off = zfs_uio_offset(uio);
1363                 bufoff = off - db->db_offset;
1364                 tocpy = MIN(db->db_size - bufoff, size);
1365
1366                 ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);
1367
1368                 if (tocpy == db->db_size)
1369                         dmu_buf_will_fill(db, tx, B_TRUE);
1370                 else
1371                         dmu_buf_will_dirty(db, tx);
1372
1373                 err = zfs_uio_fault_move((char *)db->db_data + bufoff,
1374                     tocpy, UIO_WRITE, uio);
1375
1376                 if (tocpy == db->db_size && dmu_buf_fill_done(db, tx, err)) {
1377                         /* The fill was reverted.  Undo any uio progress. */
1378                         zfs_uio_advance(uio, off - zfs_uio_offset(uio));
1379                 }
1380
1381                 if (err)
1382                         break;
1383
1384                 size -= tocpy;
1385         }
1386
1387         dmu_buf_rele_array(dbp, numbufs, FTAG);
1388         return (err);
1389 }
1390
1391 /*
1392  * Write 'size' bytes from the uio buffer.
1393  * To object zdb->db_object.
1394  * Starting at offset zfs_uio_offset(uio).
1395  *
1396  * If the caller already has a dbuf in the target object
1397  * (e.g. its bonus buffer), this routine is faster than dmu_write_uio(),
1398  * because we don't have to find the dnode_t for the object.
1399  */
1400 int
1401 dmu_write_uio_dbuf(dmu_buf_t *zdb, zfs_uio_t *uio, uint64_t size,
1402     dmu_tx_t *tx)
1403 {
1404         dmu_buf_impl_t *db = (dmu_buf_impl_t *)zdb;
1405         dnode_t *dn;
1406         int err;
1407
1408         if (size == 0)
1409                 return (0);
1410
1411         DB_DNODE_ENTER(db);
1412         dn = DB_DNODE(db);
1413         err = dmu_write_uio_dnode(dn, uio, size, tx);
1414         DB_DNODE_EXIT(db);
1415
1416         return (err);
1417 }
1418
1419 /*
1420  * Write 'size' bytes from the uio buffer.
1421  * To the specified object.
1422  * Starting at offset zfs_uio_offset(uio).
1423  */
1424 int
1425 dmu_write_uio(objset_t *os, uint64_t object, zfs_uio_t *uio, uint64_t size,
1426     dmu_tx_t *tx)
1427 {
1428         dnode_t *dn;
1429         int err;
1430
1431         if (size == 0)
1432                 return (0);
1433
1434         err = dnode_hold(os, object, FTAG, &dn);
1435         if (err)
1436                 return (err);
1437
1438         err = dmu_write_uio_dnode(dn, uio, size, tx);
1439
1440         dnode_rele(dn, FTAG);
1441
1442         return (err);
1443 }
1444 #endif /* _KERNEL */
1445
1446 /*
1447  * Allocate a loaned anonymous arc buffer.
1448  */
1449 arc_buf_t *
1450 dmu_request_arcbuf(dmu_buf_t *handle, int size)
1451 {
1452         dmu_buf_impl_t *db = (dmu_buf_impl_t *)handle;
1453
1454         return (arc_loan_buf(db->db_objset->os_spa, B_FALSE, size));
1455 }
1456
1457 /*
1458  * Free a loaned arc buffer.
1459  */
1460 void
1461 dmu_return_arcbuf(arc_buf_t *buf)
1462 {
1463         arc_return_buf(buf, FTAG);
1464         arc_buf_destroy(buf, FTAG);
1465 }
1466
1467 /*
1468  * A "lightweight" write is faster than a regular write (e.g.
1469  * dmu_write_by_dnode() or dmu_assign_arcbuf_by_dnode()), because it avoids the
1470  * CPU cost of creating a dmu_buf_impl_t and arc_buf_[hdr_]_t.  However, the
1471  * data can not be read or overwritten until the transaction's txg has been
1472  * synced.  This makes it appropriate for workloads that are known to be
1473  * (temporarily) write-only, like "zfs receive".
1474  *
1475  * A single block is written, starting at the specified offset in bytes.  If
1476  * the call is successful, it returns 0 and the provided abd has been
1477  * consumed (the caller should not free it).
1478  */
1479 int
1480 dmu_lightweight_write_by_dnode(dnode_t *dn, uint64_t offset, abd_t *abd,
1481     const zio_prop_t *zp, zio_flag_t flags, dmu_tx_t *tx)
1482 {
1483         dbuf_dirty_record_t *dr =
1484             dbuf_dirty_lightweight(dn, dbuf_whichblock(dn, 0, offset), tx);
1485         if (dr == NULL)
1486                 return (SET_ERROR(EIO));
1487         dr->dt.dll.dr_abd = abd;
1488         dr->dt.dll.dr_props = *zp;
1489         dr->dt.dll.dr_flags = flags;
1490         return (0);
1491 }
1492
1493 /*
1494  * When possible directly assign passed loaned arc buffer to a dbuf.
1495  * If this is not possible copy the contents of passed arc buf via
1496  * dmu_write().
1497  */
1498 int
1499 dmu_assign_arcbuf_by_dnode(dnode_t *dn, uint64_t offset, arc_buf_t *buf,
1500     dmu_tx_t *tx)
1501 {
1502         dmu_buf_impl_t *db;
1503         objset_t *os = dn->dn_objset;
1504         uint64_t object = dn->dn_object;
1505         uint32_t blksz = (uint32_t)arc_buf_lsize(buf);
1506         uint64_t blkid;
1507
1508         rw_enter(&dn->dn_struct_rwlock, RW_READER);
1509         blkid = dbuf_whichblock(dn, 0, offset);
1510         db = dbuf_hold(dn, blkid, FTAG);
1511         rw_exit(&dn->dn_struct_rwlock);
1512         if (db == NULL)
1513                 return (SET_ERROR(EIO));
1514
1515         /*
1516          * We can only assign if the offset is aligned and the arc buf is the
1517          * same size as the dbuf.
1518          */
1519         if (offset == db->db.db_offset && blksz == db->db.db_size) {
1520                 zfs_racct_write(blksz, 1);
1521                 dbuf_assign_arcbuf(db, buf, tx);
1522                 dbuf_rele(db, FTAG);
1523         } else {
1524                 /* compressed bufs must always be assignable to their dbuf */
1525                 ASSERT3U(arc_get_compression(buf), ==, ZIO_COMPRESS_OFF);
1526                 ASSERT(!(buf->b_flags & ARC_BUF_FLAG_COMPRESSED));
1527
1528                 dbuf_rele(db, FTAG);
1529                 dmu_write(os, object, offset, blksz, buf->b_data, tx);
1530                 dmu_return_arcbuf(buf);
1531         }
1532
1533         return (0);
1534 }
1535
1536 int
1537 dmu_assign_arcbuf_by_dbuf(dmu_buf_t *handle, uint64_t offset, arc_buf_t *buf,
1538     dmu_tx_t *tx)
1539 {
1540         int err;
1541         dmu_buf_impl_t *dbuf = (dmu_buf_impl_t *)handle;
1542
1543         DB_DNODE_ENTER(dbuf);
1544         err = dmu_assign_arcbuf_by_dnode(DB_DNODE(dbuf), offset, buf, tx);
1545         DB_DNODE_EXIT(dbuf);
1546
1547         return (err);
1548 }
1549
1550 typedef struct {
1551         dbuf_dirty_record_t     *dsa_dr;
1552         dmu_sync_cb_t           *dsa_done;
1553         zgd_t                   *dsa_zgd;
1554         dmu_tx_t                *dsa_tx;
1555 } dmu_sync_arg_t;
1556
1557 static void
1558 dmu_sync_ready(zio_t *zio, arc_buf_t *buf, void *varg)
1559 {
1560         (void) buf;
1561         dmu_sync_arg_t *dsa = varg;
1562         dmu_buf_t *db = dsa->dsa_zgd->zgd_db;
1563         blkptr_t *bp = zio->io_bp;
1564
1565         if (zio->io_error == 0) {
1566                 if (BP_IS_HOLE(bp)) {
1567                         /*
1568                          * A block of zeros may compress to a hole, but the
1569                          * block size still needs to be known for replay.
1570                          */
1571                         BP_SET_LSIZE(bp, db->db_size);
1572                 } else if (!BP_IS_EMBEDDED(bp)) {
1573                         ASSERT(BP_GET_LEVEL(bp) == 0);
1574                         BP_SET_FILL(bp, 1);
1575                 }
1576         }
1577 }
1578
1579 static void
1580 dmu_sync_late_arrival_ready(zio_t *zio)
1581 {
1582         dmu_sync_ready(zio, NULL, zio->io_private);
1583 }
1584
1585 static void
1586 dmu_sync_done(zio_t *zio, arc_buf_t *buf, void *varg)
1587 {
1588         (void) buf;
1589         dmu_sync_arg_t *dsa = varg;
1590         dbuf_dirty_record_t *dr = dsa->dsa_dr;
1591         dmu_buf_impl_t *db = dr->dr_dbuf;
1592         zgd_t *zgd = dsa->dsa_zgd;
1593
1594         /*
1595          * Record the vdev(s) backing this blkptr so they can be flushed after
1596          * the writes for the lwb have completed.
1597          */
1598         if (zio->io_error == 0) {
1599                 zil_lwb_add_block(zgd->zgd_lwb, zgd->zgd_bp);
1600         }
1601
1602         mutex_enter(&db->db_mtx);
1603         ASSERT(dr->dt.dl.dr_override_state == DR_IN_DMU_SYNC);
1604         if (zio->io_error == 0) {
1605                 dr->dt.dl.dr_nopwrite = !!(zio->io_flags & ZIO_FLAG_NOPWRITE);
1606                 if (dr->dt.dl.dr_nopwrite) {
1607                         blkptr_t *bp = zio->io_bp;
1608                         blkptr_t *bp_orig = &zio->io_bp_orig;
1609                         uint8_t chksum = BP_GET_CHECKSUM(bp_orig);
1610
1611                         ASSERT(BP_EQUAL(bp, bp_orig));
1612                         VERIFY(BP_EQUAL(bp, db->db_blkptr));
1613                         ASSERT(zio->io_prop.zp_compress != ZIO_COMPRESS_OFF);
1614                         VERIFY(zio_checksum_table[chksum].ci_flags &
1615                             ZCHECKSUM_FLAG_NOPWRITE);
1616                 }
1617                 dr->dt.dl.dr_overridden_by = *zio->io_bp;
1618                 dr->dt.dl.dr_override_state = DR_OVERRIDDEN;
1619                 dr->dt.dl.dr_copies = zio->io_prop.zp_copies;
1620
1621                 /*
1622                  * Old style holes are filled with all zeros, whereas
1623                  * new-style holes maintain their lsize, type, level,
1624                  * and birth time (see zio_write_compress). While we
1625                  * need to reset the BP_SET_LSIZE() call that happened
1626                  * in dmu_sync_ready for old style holes, we do *not*
1627                  * want to wipe out the information contained in new
1628                  * style holes. Thus, only zero out the block pointer if
1629                  * it's an old style hole.
1630                  */
1631                 if (BP_IS_HOLE(&dr->dt.dl.dr_overridden_by) &&
1632                     BP_GET_LOGICAL_BIRTH(&dr->dt.dl.dr_overridden_by) == 0)
1633                         BP_ZERO(&dr->dt.dl.dr_overridden_by);
1634         } else {
1635                 dr->dt.dl.dr_override_state = DR_NOT_OVERRIDDEN;
1636         }
1637         cv_broadcast(&db->db_changed);
1638         mutex_exit(&db->db_mtx);
1639
1640         dsa->dsa_done(dsa->dsa_zgd, zio->io_error);
1641
1642         kmem_free(dsa, sizeof (*dsa));
1643 }
1644
1645 static void
1646 dmu_sync_late_arrival_done(zio_t *zio)
1647 {
1648         blkptr_t *bp = zio->io_bp;
1649         dmu_sync_arg_t *dsa = zio->io_private;
1650         zgd_t *zgd = dsa->dsa_zgd;
1651
1652         if (zio->io_error == 0) {
1653                 /*
1654                  * Record the vdev(s) backing this blkptr so they can be
1655                  * flushed after the writes for the lwb have completed.
1656                  */
1657                 zil_lwb_add_block(zgd->zgd_lwb, zgd->zgd_bp);
1658
1659                 if (!BP_IS_HOLE(bp)) {
1660                         blkptr_t *bp_orig __maybe_unused = &zio->io_bp_orig;
1661                         ASSERT(!(zio->io_flags & ZIO_FLAG_NOPWRITE));
1662                         ASSERT(BP_IS_HOLE(bp_orig) || !BP_EQUAL(bp, bp_orig));
1663                         ASSERT(BP_GET_LOGICAL_BIRTH(zio->io_bp) == zio->io_txg);
1664                         ASSERT(zio->io_txg > spa_syncing_txg(zio->io_spa));
1665                         zio_free(zio->io_spa, zio->io_txg, zio->io_bp);
1666                 }
1667         }
1668
1669         dmu_tx_commit(dsa->dsa_tx);
1670
1671         dsa->dsa_done(dsa->dsa_zgd, zio->io_error);
1672
1673         abd_free(zio->io_abd);
1674         kmem_free(dsa, sizeof (*dsa));
1675 }
1676
1677 static int
1678 dmu_sync_late_arrival(zio_t *pio, objset_t *os, dmu_sync_cb_t *done, zgd_t *zgd,
1679     zio_prop_t *zp, zbookmark_phys_t *zb)
1680 {
1681         dmu_sync_arg_t *dsa;
1682         dmu_tx_t *tx;
1683         int error;
1684
1685         error = dbuf_read((dmu_buf_impl_t *)zgd->zgd_db, NULL,
1686             DB_RF_CANFAIL | DB_RF_NOPREFETCH);
1687         if (error != 0)
1688                 return (error);
1689
1690         tx = dmu_tx_create(os);
1691         dmu_tx_hold_space(tx, zgd->zgd_db->db_size);
1692         /*
1693          * This transaction does not produce any dirty data or log blocks, so
1694          * it should not be throttled.  All other cases wait for TXG sync, by
1695          * which time the log block we are writing will be obsolete, so we can
1696          * skip waiting and just return error here instead.
1697          */
1698         if (dmu_tx_assign(tx, TXG_NOWAIT | TXG_NOTHROTTLE) != 0) {
1699                 dmu_tx_abort(tx);
1700                 /* Make zl_get_data do txg_waited_synced() */
1701                 return (SET_ERROR(EIO));
1702         }
1703
1704         /*
1705          * In order to prevent the zgd's lwb from being free'd prior to
1706          * dmu_sync_late_arrival_done() being called, we have to ensure
1707          * the lwb's "max txg" takes this tx's txg into account.
1708          */
1709         zil_lwb_add_txg(zgd->zgd_lwb, dmu_tx_get_txg(tx));
1710
1711         dsa = kmem_alloc(sizeof (dmu_sync_arg_t), KM_SLEEP);
1712         dsa->dsa_dr = NULL;
1713         dsa->dsa_done = done;
1714         dsa->dsa_zgd = zgd;
1715         dsa->dsa_tx = tx;
1716
1717         /*
1718          * Since we are currently syncing this txg, it's nontrivial to
1719          * determine what BP to nopwrite against, so we disable nopwrite.
1720          *
1721          * When syncing, the db_blkptr is initially the BP of the previous
1722          * txg.  We can not nopwrite against it because it will be changed
1723          * (this is similar to the non-late-arrival case where the dbuf is
1724          * dirty in a future txg).
1725          *
1726          * Then dbuf_write_ready() sets bp_blkptr to the location we will write.
1727          * We can not nopwrite against it because although the BP will not
1728          * (typically) be changed, the data has not yet been persisted to this
1729          * location.
1730          *
1731          * Finally, when dbuf_write_done() is called, it is theoretically
1732          * possible to always nopwrite, because the data that was written in
1733          * this txg is the same data that we are trying to write.  However we
1734          * would need to check that this dbuf is not dirty in any future
1735          * txg's (as we do in the normal dmu_sync() path). For simplicity, we
1736          * don't nopwrite in this case.
1737          */
1738         zp->zp_nopwrite = B_FALSE;
1739
1740         zio_nowait(zio_write(pio, os->os_spa, dmu_tx_get_txg(tx), zgd->zgd_bp,
1741             abd_get_from_buf(zgd->zgd_db->db_data, zgd->zgd_db->db_size),
1742             zgd->zgd_db->db_size, zgd->zgd_db->db_size, zp,
1743             dmu_sync_late_arrival_ready, NULL, dmu_sync_late_arrival_done,
1744             dsa, ZIO_PRIORITY_SYNC_WRITE, ZIO_FLAG_CANFAIL, zb));
1745
1746         return (0);
1747 }
1748
1749 /*
1750  * Intent log support: sync the block associated with db to disk.
1751  * N.B. and XXX: the caller is responsible for making sure that the
1752  * data isn't changing while dmu_sync() is writing it.
1753  *
1754  * Return values:
1755  *
1756  *      EEXIST: this txg has already been synced, so there's nothing to do.
1757  *              The caller should not log the write.
1758  *
1759  *      ENOENT: the block was dbuf_free_range()'d, so there's nothing to do.
1760  *              The caller should not log the write.
1761  *
1762  *      EALREADY: this block is already in the process of being synced.
1763  *              The caller should track its progress (somehow).
1764  *
1765  *      EIO: could not do the I/O.
1766  *              The caller should do a txg_wait_synced().
1767  *
1768  *      0: the I/O has been initiated.
1769  *              The caller should log this blkptr in the done callback.
1770  *              It is possible that the I/O will fail, in which case
1771  *              the error will be reported to the done callback and
1772  *              propagated to pio from zio_done().
1773  */
1774 int
1775 dmu_sync(zio_t *pio, uint64_t txg, dmu_sync_cb_t *done, zgd_t *zgd)
1776 {
1777         dmu_buf_impl_t *db = (dmu_buf_impl_t *)zgd->zgd_db;
1778         objset_t *os = db->db_objset;
1779         dsl_dataset_t *ds = os->os_dsl_dataset;
1780         dbuf_dirty_record_t *dr, *dr_next;
1781         dmu_sync_arg_t *dsa;
1782         zbookmark_phys_t zb;
1783         zio_prop_t zp;
1784         dnode_t *dn;
1785
1786         ASSERT(pio != NULL);
1787         ASSERT(txg != 0);
1788
1789         SET_BOOKMARK(&zb, ds->ds_object,
1790             db->db.db_object, db->db_level, db->db_blkid);
1791
1792         DB_DNODE_ENTER(db);
1793         dn = DB_DNODE(db);
1794         dmu_write_policy(os, dn, db->db_level, WP_DMU_SYNC, &zp);
1795         DB_DNODE_EXIT(db);
1796
1797         /*
1798          * If we're frozen (running ziltest), we always need to generate a bp.
1799          */
1800         if (txg > spa_freeze_txg(os->os_spa))
1801                 return (dmu_sync_late_arrival(pio, os, done, zgd, &zp, &zb));
1802
1803         /*
1804          * Grabbing db_mtx now provides a barrier between dbuf_sync_leaf()
1805          * and us.  If we determine that this txg is not yet syncing,
1806          * but it begins to sync a moment later, that's OK because the
1807          * sync thread will block in dbuf_sync_leaf() until we drop db_mtx.
1808          */
1809         mutex_enter(&db->db_mtx);
1810
1811         if (txg <= spa_last_synced_txg(os->os_spa)) {
1812                 /*
1813                  * This txg has already synced.  There's nothing to do.
1814                  */
1815                 mutex_exit(&db->db_mtx);
1816                 return (SET_ERROR(EEXIST));
1817         }
1818
1819         if (txg <= spa_syncing_txg(os->os_spa)) {
1820                 /*
1821                  * This txg is currently syncing, so we can't mess with
1822                  * the dirty record anymore; just write a new log block.
1823                  */
1824                 mutex_exit(&db->db_mtx);
1825                 return (dmu_sync_late_arrival(pio, os, done, zgd, &zp, &zb));
1826         }
1827
1828         dr = dbuf_find_dirty_eq(db, txg);
1829
1830         if (dr == NULL) {
1831                 /*
1832                  * There's no dr for this dbuf, so it must have been freed.
1833                  * There's no need to log writes to freed blocks, so we're done.
1834                  */
1835                 mutex_exit(&db->db_mtx);
1836                 return (SET_ERROR(ENOENT));
1837         }
1838
1839         dr_next = list_next(&db->db_dirty_records, dr);
1840         ASSERT(dr_next == NULL || dr_next->dr_txg < txg);
1841
1842         if (db->db_blkptr != NULL) {
1843                 /*
1844                  * We need to fill in zgd_bp with the current blkptr so that
1845                  * the nopwrite code can check if we're writing the same
1846                  * data that's already on disk.  We can only nopwrite if we
1847                  * are sure that after making the copy, db_blkptr will not
1848                  * change until our i/o completes.  We ensure this by
1849                  * holding the db_mtx, and only allowing nopwrite if the
1850                  * block is not already dirty (see below).  This is verified
1851                  * by dmu_sync_done(), which VERIFYs that the db_blkptr has
1852                  * not changed.
1853                  */
1854                 *zgd->zgd_bp = *db->db_blkptr;
1855         }
1856
1857         /*
1858          * Assume the on-disk data is X, the current syncing data (in
1859          * txg - 1) is Y, and the current in-memory data is Z (currently
1860          * in dmu_sync).
1861          *
1862          * We usually want to perform a nopwrite if X and Z are the
1863          * same.  However, if Y is different (i.e. the BP is going to
1864          * change before this write takes effect), then a nopwrite will
1865          * be incorrect - we would override with X, which could have
1866          * been freed when Y was written.
1867          *
1868          * (Note that this is not a concern when we are nop-writing from
1869          * syncing context, because X and Y must be identical, because
1870          * all previous txgs have been synced.)
1871          *
1872          * Therefore, we disable nopwrite if the current BP could change
1873          * before this TXG.  There are two ways it could change: by
1874          * being dirty (dr_next is non-NULL), or by being freed
1875          * (dnode_block_freed()).  This behavior is verified by
1876          * zio_done(), which VERIFYs that the override BP is identical
1877          * to the on-disk BP.
1878          */
1879         DB_DNODE_ENTER(db);
1880         dn = DB_DNODE(db);
1881         if (dr_next != NULL || dnode_block_freed(dn, db->db_blkid))
1882                 zp.zp_nopwrite = B_FALSE;
1883         DB_DNODE_EXIT(db);
1884
1885         ASSERT(dr->dr_txg == txg);
1886         if (dr->dt.dl.dr_override_state == DR_IN_DMU_SYNC ||
1887             dr->dt.dl.dr_override_state == DR_OVERRIDDEN) {
1888                 /*
1889                  * We have already issued a sync write for this buffer,
1890                  * or this buffer has already been synced.  It could not
1891                  * have been dirtied since, or we would have cleared the state.
1892                  */
1893                 mutex_exit(&db->db_mtx);
1894                 return (SET_ERROR(EALREADY));
1895         }
1896
1897         ASSERT(dr->dt.dl.dr_override_state == DR_NOT_OVERRIDDEN);
1898         dr->dt.dl.dr_override_state = DR_IN_DMU_SYNC;
1899         mutex_exit(&db->db_mtx);
1900
1901         dsa = kmem_alloc(sizeof (dmu_sync_arg_t), KM_SLEEP);
1902         dsa->dsa_dr = dr;
1903         dsa->dsa_done = done;
1904         dsa->dsa_zgd = zgd;
1905         dsa->dsa_tx = NULL;
1906
1907         zio_nowait(arc_write(pio, os->os_spa, txg, zgd->zgd_bp,
1908             dr->dt.dl.dr_data, !DBUF_IS_CACHEABLE(db), dbuf_is_l2cacheable(db),
1909             &zp, dmu_sync_ready, NULL, dmu_sync_done, dsa,
1910             ZIO_PRIORITY_SYNC_WRITE, ZIO_FLAG_CANFAIL, &zb));
1911
1912         return (0);
1913 }
1914
1915 int
1916 dmu_object_set_nlevels(objset_t *os, uint64_t object, int nlevels, dmu_tx_t *tx)
1917 {
1918         dnode_t *dn;
1919         int err;
1920
1921         err = dnode_hold(os, object, FTAG, &dn);
1922         if (err)
1923                 return (err);
1924         err = dnode_set_nlevels(dn, nlevels, tx);
1925         dnode_rele(dn, FTAG);
1926         return (err);
1927 }
1928
1929 int
1930 dmu_object_set_blocksize(objset_t *os, uint64_t object, uint64_t size, int ibs,
1931     dmu_tx_t *tx)
1932 {
1933         dnode_t *dn;
1934         int err;
1935
1936         err = dnode_hold(os, object, FTAG, &dn);
1937         if (err)
1938                 return (err);
1939         err = dnode_set_blksz(dn, size, ibs, tx);
1940         dnode_rele(dn, FTAG);
1941         return (err);
1942 }
1943
1944 int
1945 dmu_object_set_maxblkid(objset_t *os, uint64_t object, uint64_t maxblkid,
1946     dmu_tx_t *tx)
1947 {
1948         dnode_t *dn;
1949         int err;
1950
1951         err = dnode_hold(os, object, FTAG, &dn);
1952         if (err)
1953                 return (err);
1954         rw_enter(&dn->dn_struct_rwlock, RW_WRITER);
1955         dnode_new_blkid(dn, maxblkid, tx, B_FALSE, B_TRUE);
1956         rw_exit(&dn->dn_struct_rwlock);
1957         dnode_rele(dn, FTAG);
1958         return (0);
1959 }
1960
1961 void
1962 dmu_object_set_checksum(objset_t *os, uint64_t object, uint8_t checksum,
1963     dmu_tx_t *tx)
1964 {
1965         dnode_t *dn;
1966
1967         /*
1968          * Send streams include each object's checksum function.  This
1969          * check ensures that the receiving system can understand the
1970          * checksum function transmitted.
1971          */
1972         ASSERT3U(checksum, <, ZIO_CHECKSUM_LEGACY_FUNCTIONS);
1973
1974         VERIFY0(dnode_hold(os, object, FTAG, &dn));
1975         ASSERT3U(checksum, <, ZIO_CHECKSUM_FUNCTIONS);
1976         dn->dn_checksum = checksum;
1977         dnode_setdirty(dn, tx);
1978         dnode_rele(dn, FTAG);
1979 }
1980
1981 void
1982 dmu_object_set_compress(objset_t *os, uint64_t object, uint8_t compress,
1983     dmu_tx_t *tx)
1984 {
1985         dnode_t *dn;
1986
1987         /*
1988          * Send streams include each object's compression function.  This
1989          * check ensures that the receiving system can understand the
1990          * compression function transmitted.
1991          */
1992         ASSERT3U(compress, <, ZIO_COMPRESS_LEGACY_FUNCTIONS);
1993
1994         VERIFY0(dnode_hold(os, object, FTAG, &dn));
1995         dn->dn_compress = compress;
1996         dnode_setdirty(dn, tx);
1997         dnode_rele(dn, FTAG);
1998 }
1999
2000 /*
2001  * When the "redundant_metadata" property is set to "most", only indirect
2002  * blocks of this level and higher will have an additional ditto block.
2003  */
2004 static const int zfs_redundant_metadata_most_ditto_level = 2;
2005
2006 void
2007 dmu_write_policy(objset_t *os, dnode_t *dn, int level, int wp, zio_prop_t *zp)
2008 {
2009         dmu_object_type_t type = dn ? dn->dn_type : DMU_OT_OBJSET;
2010         boolean_t ismd = (level > 0 || DMU_OT_IS_METADATA(type) ||
2011             (wp & WP_SPILL));
2012         enum zio_checksum checksum = os->os_checksum;
2013         enum zio_compress compress = os->os_compress;
2014         uint8_t complevel = os->os_complevel;
2015         enum zio_checksum dedup_checksum = os->os_dedup_checksum;
2016         boolean_t dedup = B_FALSE;
2017         boolean_t nopwrite = B_FALSE;
2018         boolean_t dedup_verify = os->os_dedup_verify;
2019         boolean_t encrypt = B_FALSE;
2020         int copies = os->os_copies;
2021
2022         /*
2023          * We maintain different write policies for each of the following
2024          * types of data:
2025          *       1. metadata
2026          *       2. preallocated blocks (i.e. level-0 blocks of a dump device)
2027          *       3. all other level 0 blocks
2028          */
2029         if (ismd) {
2030                 /*
2031                  * XXX -- we should design a compression algorithm
2032                  * that specializes in arrays of bps.
2033                  */
2034                 compress = zio_compress_select(os->os_spa,
2035                     ZIO_COMPRESS_ON, ZIO_COMPRESS_ON);
2036
2037                 /*
2038                  * Metadata always gets checksummed.  If the data
2039                  * checksum is multi-bit correctable, and it's not a
2040                  * ZBT-style checksum, then it's suitable for metadata
2041                  * as well.  Otherwise, the metadata checksum defaults
2042                  * to fletcher4.
2043                  */
2044                 if (!(zio_checksum_table[checksum].ci_flags &
2045                     ZCHECKSUM_FLAG_METADATA) ||
2046                     (zio_checksum_table[checksum].ci_flags &
2047                     ZCHECKSUM_FLAG_EMBEDDED))
2048                         checksum = ZIO_CHECKSUM_FLETCHER_4;
2049
2050                 switch (os->os_redundant_metadata) {
2051                 case ZFS_REDUNDANT_METADATA_ALL:
2052                         copies++;
2053                         break;
2054                 case ZFS_REDUNDANT_METADATA_MOST:
2055                         if (level >= zfs_redundant_metadata_most_ditto_level ||
2056                             DMU_OT_IS_METADATA(type) || (wp & WP_SPILL))
2057                                 copies++;
2058                         break;
2059                 case ZFS_REDUNDANT_METADATA_SOME:
2060                         if (DMU_OT_IS_CRITICAL(type))
2061                                 copies++;
2062                         break;
2063                 case ZFS_REDUNDANT_METADATA_NONE:
2064                         break;
2065                 }
2066         } else if (wp & WP_NOFILL) {
2067                 ASSERT(level == 0);
2068
2069                 /*
2070                  * If we're writing preallocated blocks, we aren't actually
2071                  * writing them so don't set any policy properties.  These
2072                  * blocks are currently only used by an external subsystem
2073                  * outside of zfs (i.e. dump) and not written by the zio
2074                  * pipeline.
2075                  */
2076                 compress = ZIO_COMPRESS_OFF;
2077                 checksum = ZIO_CHECKSUM_OFF;
2078         } else {
2079                 compress = zio_compress_select(os->os_spa, dn->dn_compress,
2080                     compress);
2081                 complevel = zio_complevel_select(os->os_spa, compress,
2082                     complevel, complevel);
2083
2084                 checksum = (dedup_checksum == ZIO_CHECKSUM_OFF) ?
2085                     zio_checksum_select(dn->dn_checksum, checksum) :
2086                     dedup_checksum;
2087
2088                 /*
2089                  * Determine dedup setting.  If we are in dmu_sync(),
2090                  * we won't actually dedup now because that's all
2091                  * done in syncing context; but we do want to use the
2092                  * dedup checksum.  If the checksum is not strong
2093                  * enough to ensure unique signatures, force
2094                  * dedup_verify.
2095                  */
2096                 if (dedup_checksum != ZIO_CHECKSUM_OFF) {
2097                         dedup = (wp & WP_DMU_SYNC) ? B_FALSE : B_TRUE;
2098                         if (!(zio_checksum_table[checksum].ci_flags &
2099                             ZCHECKSUM_FLAG_DEDUP))
2100                                 dedup_verify = B_TRUE;
2101                 }
2102
2103                 /*
2104                  * Enable nopwrite if we have secure enough checksum
2105                  * algorithm (see comment in zio_nop_write) and
2106                  * compression is enabled.  We don't enable nopwrite if
2107                  * dedup is enabled as the two features are mutually
2108                  * exclusive.
2109                  */
2110                 nopwrite = (!dedup && (zio_checksum_table[checksum].ci_flags &
2111                     ZCHECKSUM_FLAG_NOPWRITE) &&
2112                     compress != ZIO_COMPRESS_OFF && zfs_nopwrite_enabled);
2113         }
2114
2115         /*
2116          * All objects in an encrypted objset are protected from modification
2117          * via a MAC. Encrypted objects store their IV and salt in the last DVA
2118          * in the bp, so we cannot use all copies. Encrypted objects are also
2119          * not subject to nopwrite since writing the same data will still
2120          * result in a new ciphertext. Only encrypted blocks can be dedup'd
2121          * to avoid ambiguity in the dedup code since the DDT does not store
2122          * object types.
2123          */
2124         if (os->os_encrypted && (wp & WP_NOFILL) == 0) {
2125                 encrypt = B_TRUE;
2126
2127                 if (DMU_OT_IS_ENCRYPTED(type)) {
2128                         copies = MIN(copies, SPA_DVAS_PER_BP - 1);
2129                         nopwrite = B_FALSE;
2130                 } else {
2131                         dedup = B_FALSE;
2132                 }
2133
2134                 if (level <= 0 &&
2135                     (type == DMU_OT_DNODE || type == DMU_OT_OBJSET)) {
2136                         compress = ZIO_COMPRESS_EMPTY;
2137                 }
2138         }
2139
2140         zp->zp_compress = compress;
2141         zp->zp_complevel = complevel;
2142         zp->zp_checksum = checksum;
2143         zp->zp_type = (wp & WP_SPILL) ? dn->dn_bonustype : type;
2144         zp->zp_level = level;
2145         zp->zp_copies = MIN(copies, spa_max_replication(os->os_spa));
2146         zp->zp_dedup = dedup;
2147         zp->zp_dedup_verify = dedup && dedup_verify;
2148         zp->zp_nopwrite = nopwrite;
2149         zp->zp_encrypt = encrypt;
2150         zp->zp_byteorder = ZFS_HOST_BYTEORDER;
2151         memset(zp->zp_salt, 0, ZIO_DATA_SALT_LEN);
2152         memset(zp->zp_iv, 0, ZIO_DATA_IV_LEN);
2153         memset(zp->zp_mac, 0, ZIO_DATA_MAC_LEN);
2154         zp->zp_zpl_smallblk = DMU_OT_IS_FILE(zp->zp_type) ?
2155             os->os_zpl_special_smallblock : 0;
2156
2157         ASSERT3U(zp->zp_compress, !=, ZIO_COMPRESS_INHERIT);
2158 }
2159
2160 /*
2161  * Reports the location of data and holes in an object.  In order to
2162  * accurately report holes all dirty data must be synced to disk.  This
2163  * causes extremely poor performance when seeking for holes in a dirty file.
2164  * As a compromise, only provide hole data when the dnode is clean.  When
2165  * a dnode is dirty report the dnode as having no holes by returning EBUSY
2166  * which is always safe to do.
2167  */
2168 int
2169 dmu_offset_next(objset_t *os, uint64_t object, boolean_t hole, uint64_t *off)
2170 {
2171         dnode_t *dn;
2172         int restarted = 0, err;
2173
2174 restart:
2175         err = dnode_hold(os, object, FTAG, &dn);
2176         if (err)
2177                 return (err);
2178
2179         rw_enter(&dn->dn_struct_rwlock, RW_READER);
2180
2181         if (dnode_is_dirty(dn)) {
2182                 /*
2183                  * If the zfs_dmu_offset_next_sync module option is enabled
2184                  * then hole reporting has been requested.  Dirty dnodes
2185                  * must be synced to disk to accurately report holes.
2186                  *
2187                  * Provided a RL_READER rangelock spanning 0-UINT64_MAX is
2188                  * held by the caller only a single restart will be required.
2189                  * We tolerate callers which do not hold the rangelock by
2190                  * returning EBUSY and not reporting holes after one restart.
2191                  */
2192                 if (zfs_dmu_offset_next_sync) {
2193                         rw_exit(&dn->dn_struct_rwlock);
2194                         dnode_rele(dn, FTAG);
2195
2196                         if (restarted)
2197                                 return (SET_ERROR(EBUSY));
2198
2199                         txg_wait_synced(dmu_objset_pool(os), 0);
2200                         restarted = 1;
2201                         goto restart;
2202                 }
2203
2204                 err = SET_ERROR(EBUSY);
2205         } else {
2206                 err = dnode_next_offset(dn, DNODE_FIND_HAVELOCK |
2207                     (hole ? DNODE_FIND_HOLE : 0), off, 1, 1, 0);
2208         }
2209
2210         rw_exit(&dn->dn_struct_rwlock);
2211         dnode_rele(dn, FTAG);
2212
2213         return (err);
2214 }
2215
2216 int
2217 dmu_read_l0_bps(objset_t *os, uint64_t object, uint64_t offset, uint64_t length,
2218     blkptr_t *bps, size_t *nbpsp)
2219 {
2220         dmu_buf_t **dbp, *dbuf;
2221         dmu_buf_impl_t *db;
2222         blkptr_t *bp;
2223         int error, numbufs;
2224
2225         error = dmu_buf_hold_array(os, object, offset, length, FALSE, FTAG,
2226             &numbufs, &dbp);
2227         if (error != 0) {
2228                 if (error == ESRCH) {
2229                         error = SET_ERROR(ENXIO);
2230                 }
2231                 return (error);
2232         }
2233
2234         ASSERT3U(numbufs, <=, *nbpsp);
2235
2236         for (int i = 0; i < numbufs; i++) {
2237                 dbuf = dbp[i];
2238                 db = (dmu_buf_impl_t *)dbuf;
2239
2240                 mutex_enter(&db->db_mtx);
2241
2242                 if (!list_is_empty(&db->db_dirty_records)) {
2243                         dbuf_dirty_record_t *dr;
2244
2245                         dr = list_head(&db->db_dirty_records);
2246                         if (dr->dt.dl.dr_brtwrite) {
2247                                 /*
2248                                  * This is very special case where we clone a
2249                                  * block and in the same transaction group we
2250                                  * read its BP (most likely to clone the clone).
2251                                  */
2252                                 bp = &dr->dt.dl.dr_overridden_by;
2253                         } else {
2254                                 /*
2255                                  * The block was modified in the same
2256                                  * transaction group.
2257                                  */
2258                                 mutex_exit(&db->db_mtx);
2259                                 error = SET_ERROR(EAGAIN);
2260                                 goto out;
2261                         }
2262                 } else {
2263                         bp = db->db_blkptr;
2264                 }
2265
2266                 mutex_exit(&db->db_mtx);
2267
2268                 if (bp == NULL) {
2269                         /*
2270                          * The file size was increased, but the block was never
2271                          * written, otherwise we would either have the block
2272                          * pointer or the dirty record and would not get here.
2273                          * It is effectively a hole, so report it as such.
2274                          */
2275                         BP_ZERO(&bps[i]);
2276                         continue;
2277                 }
2278                 /*
2279                  * Make sure we clone only data blocks.
2280                  */
2281                 if (BP_IS_METADATA(bp) && !BP_IS_HOLE(bp)) {
2282                         error = SET_ERROR(EINVAL);
2283                         goto out;
2284                 }
2285
2286                 /*
2287                  * If the block was allocated in transaction group that is not
2288                  * yet synced, we could clone it, but we couldn't write this
2289                  * operation into ZIL, or it may be impossible to replay, since
2290                  * the block may appear not yet allocated at that point.
2291                  */
2292                 if (BP_GET_BIRTH(bp) > spa_freeze_txg(os->os_spa)) {
2293                         error = SET_ERROR(EINVAL);
2294                         goto out;
2295                 }
2296                 if (BP_GET_BIRTH(bp) > spa_last_synced_txg(os->os_spa)) {
2297                         error = SET_ERROR(EAGAIN);
2298                         goto out;
2299                 }
2300
2301                 bps[i] = *bp;
2302         }
2303
2304         *nbpsp = numbufs;
2305 out:
2306         dmu_buf_rele_array(dbp, numbufs, FTAG);
2307
2308         return (error);
2309 }
2310
2311 int
2312 dmu_brt_clone(objset_t *os, uint64_t object, uint64_t offset, uint64_t length,
2313     dmu_tx_t *tx, const blkptr_t *bps, size_t nbps)
2314 {
2315         spa_t *spa;
2316         dmu_buf_t **dbp, *dbuf;
2317         dmu_buf_impl_t *db;
2318         struct dirty_leaf *dl;
2319         dbuf_dirty_record_t *dr;
2320         const blkptr_t *bp;
2321         int error = 0, i, numbufs;
2322
2323         spa = os->os_spa;
2324
2325         VERIFY0(dmu_buf_hold_array(os, object, offset, length, FALSE, FTAG,
2326             &numbufs, &dbp));
2327         ASSERT3U(nbps, ==, numbufs);
2328
2329         /*
2330          * Before we start cloning make sure that the dbufs sizes match new BPs
2331          * sizes. If they don't, that's a no-go, as we are not able to shrink
2332          * dbufs.
2333          */
2334         for (i = 0; i < numbufs; i++) {
2335                 dbuf = dbp[i];
2336                 db = (dmu_buf_impl_t *)dbuf;
2337                 bp = &bps[i];
2338
2339                 ASSERT0(db->db_level);
2340                 ASSERT(db->db_blkid != DMU_BONUS_BLKID);
2341                 ASSERT(db->db_blkid != DMU_SPILL_BLKID);
2342
2343                 if (!BP_IS_HOLE(bp) && BP_GET_LSIZE(bp) != dbuf->db_size) {
2344                         error = SET_ERROR(EXDEV);
2345                         goto out;
2346                 }
2347         }
2348
2349         for (i = 0; i < numbufs; i++) {
2350                 dbuf = dbp[i];
2351                 db = (dmu_buf_impl_t *)dbuf;
2352                 bp = &bps[i];
2353
2354                 ASSERT0(db->db_level);
2355                 ASSERT(db->db_blkid != DMU_BONUS_BLKID);
2356                 ASSERT(db->db_blkid != DMU_SPILL_BLKID);
2357                 ASSERT(BP_IS_HOLE(bp) || dbuf->db_size == BP_GET_LSIZE(bp));
2358
2359                 dmu_buf_will_clone(dbuf, tx);
2360
2361                 mutex_enter(&db->db_mtx);
2362
2363                 dr = list_head(&db->db_dirty_records);
2364                 VERIFY(dr != NULL);
2365                 ASSERT3U(dr->dr_txg, ==, tx->tx_txg);
2366                 dl = &dr->dt.dl;
2367                 dl->dr_overridden_by = *bp;
2368                 if (!BP_IS_HOLE(bp) || BP_GET_LOGICAL_BIRTH(bp) != 0) {
2369                         if (!BP_IS_EMBEDDED(bp)) {
2370                                 BP_SET_BIRTH(&dl->dr_overridden_by, dr->dr_txg,
2371                                     BP_GET_BIRTH(bp));
2372                         } else {
2373                                 BP_SET_LOGICAL_BIRTH(&dl->dr_overridden_by,
2374                                     dr->dr_txg);
2375                         }
2376                 }
2377                 dl->dr_brtwrite = B_TRUE;
2378                 dl->dr_override_state = DR_OVERRIDDEN;
2379
2380                 mutex_exit(&db->db_mtx);
2381
2382                 /*
2383                  * When data in embedded into BP there is no need to create
2384                  * BRT entry as there is no data block. Just copy the BP as
2385                  * it contains the data.
2386                  */
2387                 if (!BP_IS_HOLE(bp) && !BP_IS_EMBEDDED(bp)) {
2388                         brt_pending_add(spa, bp, tx);
2389                 }
2390         }
2391 out:
2392         dmu_buf_rele_array(dbp, numbufs, FTAG);
2393
2394         return (error);
2395 }
2396
2397 void
2398 __dmu_object_info_from_dnode(dnode_t *dn, dmu_object_info_t *doi)
2399 {
2400         dnode_phys_t *dnp = dn->dn_phys;
2401
2402         doi->doi_data_block_size = dn->dn_datablksz;
2403         doi->doi_metadata_block_size = dn->dn_indblkshift ?
2404             1ULL << dn->dn_indblkshift : 0;
2405         doi->doi_type = dn->dn_type;
2406         doi->doi_bonus_type = dn->dn_bonustype;
2407         doi->doi_bonus_size = dn->dn_bonuslen;
2408         doi->doi_dnodesize = dn->dn_num_slots << DNODE_SHIFT;
2409         doi->doi_indirection = dn->dn_nlevels;
2410         doi->doi_checksum = dn->dn_checksum;
2411         doi->doi_compress = dn->dn_compress;
2412         doi->doi_nblkptr = dn->dn_nblkptr;
2413         doi->doi_physical_blocks_512 = (DN_USED_BYTES(dnp) + 256) >> 9;
2414         doi->doi_max_offset = (dn->dn_maxblkid + 1) * dn->dn_datablksz;
2415         doi->doi_fill_count = 0;
2416         for (int i = 0; i < dnp->dn_nblkptr; i++)
2417                 doi->doi_fill_count += BP_GET_FILL(&dnp->dn_blkptr[i]);
2418 }
2419
2420 void
2421 dmu_object_info_from_dnode(dnode_t *dn, dmu_object_info_t *doi)
2422 {
2423         rw_enter(&dn->dn_struct_rwlock, RW_READER);
2424         mutex_enter(&dn->dn_mtx);
2425
2426         __dmu_object_info_from_dnode(dn, doi);
2427
2428         mutex_exit(&dn->dn_mtx);
2429         rw_exit(&dn->dn_struct_rwlock);
2430 }
2431
2432 /*
2433  * Get information on a DMU object.
2434  * If doi is NULL, just indicates whether the object exists.
2435  */
2436 int
2437 dmu_object_info(objset_t *os, uint64_t object, dmu_object_info_t *doi)
2438 {
2439         dnode_t *dn;
2440         int err = dnode_hold(os, object, FTAG, &dn);
2441
2442         if (err)
2443                 return (err);
2444
2445         if (doi != NULL)
2446                 dmu_object_info_from_dnode(dn, doi);
2447
2448         dnode_rele(dn, FTAG);
2449         return (0);
2450 }
2451
2452 /*
2453  * As above, but faster; can be used when you have a held dbuf in hand.
2454  */
2455 void
2456 dmu_object_info_from_db(dmu_buf_t *db_fake, dmu_object_info_t *doi)
2457 {
2458         dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
2459
2460         DB_DNODE_ENTER(db);
2461         dmu_object_info_from_dnode(DB_DNODE(db), doi);
2462         DB_DNODE_EXIT(db);
2463 }
2464
2465 /*
2466  * Faster still when you only care about the size.
2467  */
2468 void
2469 dmu_object_size_from_db(dmu_buf_t *db_fake, uint32_t *blksize,
2470     u_longlong_t *nblk512)
2471 {
2472         dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
2473         dnode_t *dn;
2474
2475         DB_DNODE_ENTER(db);
2476         dn = DB_DNODE(db);
2477
2478         *blksize = dn->dn_datablksz;
2479         /* add in number of slots used for the dnode itself */
2480         *nblk512 = ((DN_USED_BYTES(dn->dn_phys) + SPA_MINBLOCKSIZE/2) >>
2481             SPA_MINBLOCKSHIFT) + dn->dn_num_slots;
2482         DB_DNODE_EXIT(db);
2483 }
2484
2485 void
2486 dmu_object_dnsize_from_db(dmu_buf_t *db_fake, int *dnsize)
2487 {
2488         dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;
2489         dnode_t *dn;
2490
2491         DB_DNODE_ENTER(db);
2492         dn = DB_DNODE(db);
2493         *dnsize = dn->dn_num_slots << DNODE_SHIFT;
2494         DB_DNODE_EXIT(db);
2495 }
2496
2497 void
2498 byteswap_uint64_array(void *vbuf, size_t size)
2499 {
2500         uint64_t *buf = vbuf;
2501         size_t count = size >> 3;
2502         int i;
2503
2504         ASSERT((size & 7) == 0);
2505
2506         for (i = 0; i < count; i++)
2507                 buf[i] = BSWAP_64(buf[i]);
2508 }
2509
2510 void
2511 byteswap_uint32_array(void *vbuf, size_t size)
2512 {
2513         uint32_t *buf = vbuf;
2514         size_t count = size >> 2;
2515         int i;
2516
2517         ASSERT((size & 3) == 0);
2518
2519         for (i = 0; i < count; i++)
2520                 buf[i] = BSWAP_32(buf[i]);
2521 }
2522
2523 void
2524 byteswap_uint16_array(void *vbuf, size_t size)
2525 {
2526         uint16_t *buf = vbuf;
2527         size_t count = size >> 1;
2528         int i;
2529
2530         ASSERT((size & 1) == 0);
2531
2532         for (i = 0; i < count; i++)
2533                 buf[i] = BSWAP_16(buf[i]);
2534 }
2535
2536 void
2537 byteswap_uint8_array(void *vbuf, size_t size)
2538 {
2539         (void) vbuf, (void) size;
2540 }
2541
2542 void
2543 dmu_init(void)
2544 {
2545         abd_init();
2546         zfs_dbgmsg_init();
2547         sa_cache_init();
2548         dmu_objset_init();
2549         dnode_init();
2550         zfetch_init();
2551         dmu_tx_init();
2552         l2arc_init();
2553         arc_init();
2554         dbuf_init();
2555 }
2556
2557 void
2558 dmu_fini(void)
2559 {
2560         arc_fini(); /* arc depends on l2arc, so arc must go first */
2561         l2arc_fini();
2562         dmu_tx_fini();
2563         zfetch_fini();
2564         dbuf_fini();
2565         dnode_fini();
2566         dmu_objset_fini();
2567         sa_cache_fini();
2568         zfs_dbgmsg_fini();
2569         abd_fini();
2570 }
2571
2572 EXPORT_SYMBOL(dmu_bonus_hold);
2573 EXPORT_SYMBOL(dmu_bonus_hold_by_dnode);
2574 EXPORT_SYMBOL(dmu_buf_hold_array_by_bonus);
2575 EXPORT_SYMBOL(dmu_buf_rele_array);
2576 EXPORT_SYMBOL(dmu_prefetch);
2577 EXPORT_SYMBOL(dmu_prefetch_by_dnode);
2578 EXPORT_SYMBOL(dmu_prefetch_dnode);
2579 EXPORT_SYMBOL(dmu_free_range);
2580 EXPORT_SYMBOL(dmu_free_long_range);
2581 EXPORT_SYMBOL(dmu_free_long_object);
2582 EXPORT_SYMBOL(dmu_read);
2583 EXPORT_SYMBOL(dmu_read_by_dnode);
2584 EXPORT_SYMBOL(dmu_write);
2585 EXPORT_SYMBOL(dmu_write_by_dnode);
2586 EXPORT_SYMBOL(dmu_prealloc);
2587 EXPORT_SYMBOL(dmu_object_info);
2588 EXPORT_SYMBOL(dmu_object_info_from_dnode);
2589 EXPORT_SYMBOL(dmu_object_info_from_db);
2590 EXPORT_SYMBOL(dmu_object_size_from_db);
2591 EXPORT_SYMBOL(dmu_object_dnsize_from_db);
2592 EXPORT_SYMBOL(dmu_object_set_nlevels);
2593 EXPORT_SYMBOL(dmu_object_set_blocksize);
2594 EXPORT_SYMBOL(dmu_object_set_maxblkid);
2595 EXPORT_SYMBOL(dmu_object_set_checksum);
2596 EXPORT_SYMBOL(dmu_object_set_compress);
2597 EXPORT_SYMBOL(dmu_offset_next);
2598 EXPORT_SYMBOL(dmu_write_policy);
2599 EXPORT_SYMBOL(dmu_sync);
2600 EXPORT_SYMBOL(dmu_request_arcbuf);
2601 EXPORT_SYMBOL(dmu_return_arcbuf);
2602 EXPORT_SYMBOL(dmu_assign_arcbuf_by_dnode);
2603 EXPORT_SYMBOL(dmu_assign_arcbuf_by_dbuf);
2604 EXPORT_SYMBOL(dmu_buf_hold);
2605 EXPORT_SYMBOL(dmu_ot);
2606
2607 ZFS_MODULE_PARAM(zfs, zfs_, nopwrite_enabled, INT, ZMOD_RW,
2608         "Enable NOP writes");
2609
2610 ZFS_MODULE_PARAM(zfs, zfs_, per_txg_dirty_frees_percent, UINT, ZMOD_RW,
2611         "Percentage of dirtied blocks from frees in one TXG");
2612
2613 ZFS_MODULE_PARAM(zfs, zfs_, dmu_offset_next_sync, INT, ZMOD_RW,
2614         "Enable forcing txg sync to find holes");
2615
2616 /* CSTYLED */
2617 ZFS_MODULE_PARAM(zfs, , dmu_prefetch_max, UINT, ZMOD_RW,
2618         "Limit one prefetch call to this size");