module/zfs/dmu_objset.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 /*
  23  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Copyright (c) 2012, 2020 by Delphix. All rights reserved.
  25  * Copyright (c) 2013 by Saso Kiselkov. All rights reserved.
  26  * Copyright (c) 2013, Joyent, Inc. All rights reserved.
  27  * Copyright (c) 2014 Spectra Logic Corporation, All rights reserved.
  28  * Copyright (c) 2015, STRATO AG, Inc. All rights reserved.
  29  * Copyright (c) 2016 Actifio, Inc. All rights reserved.
  30  * Copyright 2017 Nexenta Systems, Inc.
  31  * Copyright (c) 2017 Open-E, Inc. All Rights Reserved.
  32  * Copyright (c) 2018, loli10K <ezomori.nozomu@gmail.com>. All rights reserved.
  33  * Copyright (c) 2019, Klara Inc.
  34  * Copyright (c) 2019, Allan Jude
  35  * Copyright (c) 2022 Hewlett Packard Enterprise Development LP.
  36  */
  37
  38 /* Portions Copyright 2010 Robert Milkowski */
  39
  40 #include <sys/cred.h>
  41 #include <sys/zfs_context.h>
  42 #include <sys/dmu_objset.h>
  43 #include <sys/dsl_dir.h>
  44 #include <sys/dsl_dataset.h>
  45 #include <sys/dsl_prop.h>
  46 #include <sys/dsl_pool.h>
  47 #include <sys/dsl_synctask.h>
  48 #include <sys/dsl_deleg.h>
  49 #include <sys/dnode.h>
  50 #include <sys/dbuf.h>
  51 #include <sys/zvol.h>
  52 #include <sys/dmu_tx.h>
  53 #include <sys/zap.h>
  54 #include <sys/zil.h>
  55 #include <sys/dmu_impl.h>
  56 #include <sys/zfs_ioctl.h>
  57 #include <sys/sa.h>
  58 #include <sys/zfs_onexit.h>
  59 #include <sys/dsl_destroy.h>
  60 #include <sys/vdev.h>
  61 #include <sys/zfeature.h>
  62 #include <sys/policy.h>
  63 #include <sys/spa_impl.h>
  64 #include <sys/dmu_recv.h>
  65 #include <sys/zfs_project.h>
  66 #include "zfs_namecheck.h"
  67 #include <sys/vdev_impl.h>
  68 #include <sys/arc.h>
  69
  70 /*
  71  * Needed to close a window in dnode_move() that allows the objset to be freed
  72  * before it can be safely accessed.
  73  */
  74 krwlock_t os_lock;
  75
  76 /*
  77  * Tunable to overwrite the maximum number of threads for the parallelization
  78  * of dmu_objset_find_dp, needed to speed up the import of pools with many
  79  * datasets.
  80  * Default is 4 times the number of leaf vdevs.
  81  */
  82 static const int dmu_find_threads = 0;
  83
  84 /*
  85  * Backfill lower metadnode objects after this many have been freed.
  86  * Backfilling negatively impacts object creation rates, so only do it
  87  * if there are enough holes to fill.
  88  */
  89 static const int dmu_rescan_dnode_threshold = 1 << DN_MAX_INDBLKSHIFT;
  90
  91 static const char *upgrade_tag = "upgrade_tag";
  92
  93 static void dmu_objset_find_dp_cb(void *arg);
  94
  95 static void dmu_objset_upgrade(objset_t *os, dmu_objset_upgrade_cb_t cb);
  96 static void dmu_objset_upgrade_stop(objset_t *os);
  97
  98 void
  99 dmu_objset_init(void)
 100 {
 101         rw_init(&os_lock, NULL, RW_DEFAULT, NULL);
 102 }
 103
 104 void
 105 dmu_objset_fini(void)
 106 {
 107         rw_destroy(&os_lock);
 108 }
 109
 110 spa_t *
 111 dmu_objset_spa(objset_t *os)
 112 {
 113         return (os->os_spa);
 114 }
 115
 116 zilog_t *
 117 dmu_objset_zil(objset_t *os)
 118 {
 119         return (os->os_zil);
 120 }
 121
 122 dsl_pool_t *
 123 dmu_objset_pool(objset_t *os)
 124 {
 125         dsl_dataset_t *ds;
 126
 127         if ((ds = os->os_dsl_dataset) != NULL && ds->ds_dir)
 128                 return (ds->ds_dir->dd_pool);
 129         else
 130                 return (spa_get_dsl(os->os_spa));
 131 }
 132
 133 dsl_dataset_t *
 134 dmu_objset_ds(objset_t *os)
 135 {
 136         return (os->os_dsl_dataset);
 137 }
 138
 139 dmu_objset_type_t
 140 dmu_objset_type(objset_t *os)
 141 {
 142         return (os->os_phys->os_type);
 143 }
 144
 145 void
 146 dmu_objset_name(objset_t *os, char *buf)
 147 {
 148         dsl_dataset_name(os->os_dsl_dataset, buf);
 149 }
 150
 151 uint64_t
 152 dmu_objset_id(objset_t *os)
 153 {
 154         dsl_dataset_t *ds = os->os_dsl_dataset;
 155
 156         return (ds ? ds->ds_object : 0);
 157 }
 158
 159 uint64_t
 160 dmu_objset_dnodesize(objset_t *os)
 161 {
 162         return (os->os_dnodesize);
 163 }
 164
 165 zfs_sync_type_t
 166 dmu_objset_syncprop(objset_t *os)
 167 {
 168         return (os->os_sync);
 169 }
 170
 171 zfs_logbias_op_t
 172 dmu_objset_logbias(objset_t *os)
 173 {
 174         return (os->os_logbias);
 175 }
 176
 177 static void
 178 checksum_changed_cb(void *arg, uint64_t newval)
 179 {
 180         objset_t *os = arg;
 181
 182         /*
 183          * Inheritance should have been done by now.
 184          */
 185         ASSERT(newval != ZIO_CHECKSUM_INHERIT);
 186
 187         os->os_checksum = zio_checksum_select(newval, ZIO_CHECKSUM_ON_VALUE);
 188 }
 189
 190 static void
 191 compression_changed_cb(void *arg, uint64_t newval)
 192 {
 193         objset_t *os = arg;
 194
 195         /*
 196          * Inheritance and range checking should have been done by now.
 197          */
 198         ASSERT(newval != ZIO_COMPRESS_INHERIT);
 199
 200         os->os_compress = zio_compress_select(os->os_spa,
 201             ZIO_COMPRESS_ALGO(newval), ZIO_COMPRESS_ON);
 202         os->os_complevel = zio_complevel_select(os->os_spa, os->os_compress,
 203             ZIO_COMPRESS_LEVEL(newval), ZIO_COMPLEVEL_DEFAULT);
 204 }
 205
 206 static void
 207 copies_changed_cb(void *arg, uint64_t newval)
 208 {
 209         objset_t *os = arg;
 210
 211         /*
 212          * Inheritance and range checking should have been done by now.
 213          */
 214         ASSERT(newval > 0);
 215         ASSERT(newval <= spa_max_replication(os->os_spa));
 216
 217         os->os_copies = newval;
 218 }
 219
 220 static void
 221 dedup_changed_cb(void *arg, uint64_t newval)
 222 {
 223         objset_t *os = arg;
 224         spa_t *spa = os->os_spa;
 225         enum zio_checksum checksum;
 226
 227         /*
 228          * Inheritance should have been done by now.
 229          */
 230         ASSERT(newval != ZIO_CHECKSUM_INHERIT);
 231
 232         checksum = zio_checksum_dedup_select(spa, newval, ZIO_CHECKSUM_OFF);
 233
 234         os->os_dedup_checksum = checksum & ZIO_CHECKSUM_MASK;
 235         os->os_dedup_verify = !!(checksum & ZIO_CHECKSUM_VERIFY);
 236 }
 237
 238 static void
 239 primary_cache_changed_cb(void *arg, uint64_t newval)
 240 {
 241         objset_t *os = arg;
 242
 243         /*
 244          * Inheritance and range checking should have been done by now.
 245          */
 246         ASSERT(newval == ZFS_CACHE_ALL || newval == ZFS_CACHE_NONE ||
 247             newval == ZFS_CACHE_METADATA);
 248
 249         os->os_primary_cache = newval;
 250 }
 251
 252 static void
 253 secondary_cache_changed_cb(void *arg, uint64_t newval)
 254 {
 255         objset_t *os = arg;
 256
 257         /*
 258          * Inheritance and range checking should have been done by now.
 259          */
 260         ASSERT(newval == ZFS_CACHE_ALL || newval == ZFS_CACHE_NONE ||
 261             newval == ZFS_CACHE_METADATA);
 262
 263         os->os_secondary_cache = newval;
 264 }
 265
 266 static void
 267 sync_changed_cb(void *arg, uint64_t newval)
 268 {
 269         objset_t *os = arg;
 270
 271         /*
 272          * Inheritance and range checking should have been done by now.
 273          */
 274         ASSERT(newval == ZFS_SYNC_STANDARD || newval == ZFS_SYNC_ALWAYS ||
 275             newval == ZFS_SYNC_DISABLED);
 276
 277         os->os_sync = newval;
 278         if (os->os_zil)
 279                 zil_set_sync(os->os_zil, newval);
 280 }
 281
 282 static void
 283 redundant_metadata_changed_cb(void *arg, uint64_t newval)
 284 {
 285         objset_t *os = arg;
 286
 287         /*
 288          * Inheritance and range checking should have been done by now.
 289          */
 290         ASSERT(newval == ZFS_REDUNDANT_METADATA_ALL ||
 291             newval == ZFS_REDUNDANT_METADATA_MOST ||
 292             newval == ZFS_REDUNDANT_METADATA_SOME ||
 293             newval == ZFS_REDUNDANT_METADATA_NONE);
 294
 295         os->os_redundant_metadata = newval;
 296 }
 297
 298 static void
 299 dnodesize_changed_cb(void *arg, uint64_t newval)
 300 {
 301         objset_t *os = arg;
 302
 303         switch (newval) {
 304         case ZFS_DNSIZE_LEGACY:
 305                 os->os_dnodesize = DNODE_MIN_SIZE;
 306                 break;
 307         case ZFS_DNSIZE_AUTO:
 308                 /*
 309                  * Choose a dnode size that will work well for most
 310                  * workloads if the user specified "auto". Future code
 311                  * improvements could dynamically select a dnode size
 312                  * based on observed workload patterns.
 313                  */
 314                 os->os_dnodesize = DNODE_MIN_SIZE * 2;
 315                 break;
 316         case ZFS_DNSIZE_1K:
 317         case ZFS_DNSIZE_2K:
 318         case ZFS_DNSIZE_4K:
 319         case ZFS_DNSIZE_8K:
 320         case ZFS_DNSIZE_16K:
 321                 os->os_dnodesize = newval;
 322                 break;
 323         }
 324 }
 325
 326 static void
 327 smallblk_changed_cb(void *arg, uint64_t newval)
 328 {
 329         objset_t *os = arg;
 330
 331         /*
 332          * Inheritance and range checking should have been done by now.
 333          */
 334         ASSERT(newval <= SPA_MAXBLOCKSIZE);
 335         ASSERT(ISP2(newval));
 336
 337         os->os_zpl_special_smallblock = newval;
 338 }
 339
 340 static void
 341 logbias_changed_cb(void *arg, uint64_t newval)
 342 {
 343         objset_t *os = arg;
 344
 345         ASSERT(newval == ZFS_LOGBIAS_LATENCY ||
 346             newval == ZFS_LOGBIAS_THROUGHPUT);
 347         os->os_logbias = newval;
 348         if (os->os_zil)
 349                 zil_set_logbias(os->os_zil, newval);
 350 }
 351
 352 static void
 353 recordsize_changed_cb(void *arg, uint64_t newval)
 354 {
 355         objset_t *os = arg;
 356
 357         os->os_recordsize = newval;
 358 }
 359
 360 void
 361 dmu_objset_byteswap(void *buf, size_t size)
 362 {
 363         objset_phys_t *osp = buf;
 364
 365         ASSERT(size == OBJSET_PHYS_SIZE_V1 || size == OBJSET_PHYS_SIZE_V2 ||
 366             size == sizeof (objset_phys_t));
 367         dnode_byteswap(&osp->os_meta_dnode);
 368         byteswap_uint64_array(&osp->os_zil_header, sizeof (zil_header_t));
 369         osp->os_type = BSWAP_64(osp->os_type);
 370         osp->os_flags = BSWAP_64(osp->os_flags);
 371         if (size >= OBJSET_PHYS_SIZE_V2) {
 372                 dnode_byteswap(&osp->os_userused_dnode);
 373                 dnode_byteswap(&osp->os_groupused_dnode);
 374                 if (size >= sizeof (objset_phys_t))
 375                         dnode_byteswap(&osp->os_projectused_dnode);
 376         }
 377 }
 378
 379 /*
 380  * The hash is a CRC-based hash of the objset_t pointer and the object number.
 381  */
 382 static uint64_t
 383 dnode_hash(const objset_t *os, uint64_t obj)
 384 {
 385         uintptr_t osv = (uintptr_t)os;
 386         uint64_t crc = -1ULL;
 387
 388         ASSERT(zfs_crc64_table[128] == ZFS_CRC64_POLY);
 389         /*
 390          * The low 6 bits of the pointer don't have much entropy, because
 391          * the objset_t is larger than 2^6 bytes long.
 392          */
 393         crc = (crc >> 8) ^ zfs_crc64_table[(crc ^ (osv >> 6)) & 0xFF];
 394         crc = (crc >> 8) ^ zfs_crc64_table[(crc ^ (obj >> 0)) & 0xFF];
 395         crc = (crc >> 8) ^ zfs_crc64_table[(crc ^ (obj >> 8)) & 0xFF];
 396         crc = (crc >> 8) ^ zfs_crc64_table[(crc ^ (obj >> 16)) & 0xFF];
 397
 398         crc ^= (osv>>14) ^ (obj>>24);
 399
 400         return (crc);
 401 }
 402
 403 static unsigned int
 404 dnode_multilist_index_func(multilist_t *ml, void *obj)
 405 {
 406         dnode_t *dn = obj;
 407
 408         /*
 409          * The low order bits of the hash value are thought to be
 410          * distributed evenly. Otherwise, in the case that the multilist
 411          * has a power of two number of sublists, each sublists' usage
 412          * would not be evenly distributed. In this context full 64bit
 413          * division would be a waste of time, so limit it to 32 bits.
 414          */
 415         return ((unsigned int)dnode_hash(dn->dn_objset, dn->dn_object) %
 416             multilist_get_num_sublists(ml));
 417 }
 418
 419 static inline boolean_t
 420 dmu_os_is_l2cacheable(objset_t *os)
 421 {
 422         vdev_t *vd = NULL;
 423         zfs_cache_type_t cache = os->os_secondary_cache;
 424         blkptr_t *bp = os->os_rootbp;
 425
 426         if (bp != NULL && !BP_IS_HOLE(bp)) {
 427                 uint64_t vdev = DVA_GET_VDEV(bp->blk_dva);
 428                 vdev_t *rvd = os->os_spa->spa_root_vdev;
 429
 430                 if (vdev < rvd->vdev_children)
 431                         vd = rvd->vdev_child[vdev];
 432
 433                 if (cache == ZFS_CACHE_ALL || cache == ZFS_CACHE_METADATA) {
 434                         if (vd == NULL)
 435                                 return (B_TRUE);
 436
 437                         if ((vd->vdev_alloc_bias != VDEV_BIAS_SPECIAL &&
 438                             vd->vdev_alloc_bias != VDEV_BIAS_DEDUP) ||
 439                             l2arc_exclude_special == 0)
 440                                 return (B_TRUE);
 441                 }
 442         }
 443
 444         return (B_FALSE);
 445 }
 446
 447 /*
 448  * Instantiates the objset_t in-memory structure corresponding to the
 449  * objset_phys_t that's pointed to by the specified blkptr_t.
 450  */
 451 int
 452 dmu_objset_open_impl(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
 453     objset_t **osp)
 454 {
 455         objset_t *os;
 456         int i, err;
 457
 458         ASSERT(ds == NULL || MUTEX_HELD(&ds->ds_opening_lock));
 459         ASSERT(!BP_IS_REDACTED(bp));
 460
 461         /*
 462          * We need the pool config lock to get properties.
 463          */
 464         ASSERT(ds == NULL || dsl_pool_config_held(ds->ds_dir->dd_pool));
 465
 466         /*
 467          * The $ORIGIN dataset (if it exists) doesn't have an associated
 468          * objset, so there's no reason to open it. The $ORIGIN dataset
 469          * will not exist on pools older than SPA_VERSION_ORIGIN.
 470          */
 471         if (ds != NULL && spa_get_dsl(spa) != NULL &&
 472             spa_get_dsl(spa)->dp_origin_snap != NULL) {
 473                 ASSERT3P(ds->ds_dir, !=,
 474                     spa_get_dsl(spa)->dp_origin_snap->ds_dir);
 475         }
 476
 477         os = kmem_zalloc(sizeof (objset_t), KM_SLEEP);
 478         os->os_dsl_dataset = ds;
 479         os->os_spa = spa;
 480         os->os_rootbp = bp;
 481         if (!BP_IS_HOLE(os->os_rootbp)) {
 482                 arc_flags_t aflags = ARC_FLAG_WAIT;
 483                 zbookmark_phys_t zb;
 484                 int size;
 485                 zio_flag_t zio_flags = ZIO_FLAG_CANFAIL;
 486                 SET_BOOKMARK(&zb, ds ? ds->ds_object : DMU_META_OBJSET,
 487                     ZB_ROOT_OBJECT, ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
 488
 489                 if (dmu_os_is_l2cacheable(os))
 490                         aflags |= ARC_FLAG_L2CACHE;
 491
 492                 if (ds != NULL && ds->ds_dir->dd_crypto_obj != 0) {
 493                         ASSERT3U(BP_GET_COMPRESS(bp), ==, ZIO_COMPRESS_OFF);
 494                         ASSERT(BP_IS_AUTHENTICATED(bp));
 495                         zio_flags |= ZIO_FLAG_RAW;
 496                 }
 497
 498                 dprintf_bp(os->os_rootbp, "reading %s", "");
 499                 err = arc_read(NULL, spa, os->os_rootbp,
 500                     arc_getbuf_func, &os->os_phys_buf,
 501                     ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);
 502                 if (err != 0) {
 503                         kmem_free(os, sizeof (objset_t));
 504                         /* convert checksum errors into IO errors */
 505                         if (err == ECKSUM)
 506                                 err = SET_ERROR(EIO);
 507                         return (err);
 508                 }
 509
 510                 if (spa_version(spa) < SPA_VERSION_USERSPACE)
 511                         size = OBJSET_PHYS_SIZE_V1;
 512                 else if (!spa_feature_is_enabled(spa,
 513                     SPA_FEATURE_PROJECT_QUOTA))
 514                         size = OBJSET_PHYS_SIZE_V2;
 515                 else
 516                         size = sizeof (objset_phys_t);
 517
 518                 /* Increase the blocksize if we are permitted. */
 519                 if (arc_buf_size(os->os_phys_buf) < size) {
 520                         arc_buf_t *buf = arc_alloc_buf(spa, &os->os_phys_buf,
 521                             ARC_BUFC_METADATA, size);
 522                         memset(buf->b_data, 0, size);
 523                         memcpy(buf->b_data, os->os_phys_buf->b_data,
 524                             arc_buf_size(os->os_phys_buf));
 525                         arc_buf_destroy(os->os_phys_buf, &os->os_phys_buf);
 526                         os->os_phys_buf = buf;
 527                 }
 528
 529                 os->os_phys = os->os_phys_buf->b_data;
 530                 os->os_flags = os->os_phys->os_flags;
 531         } else {
 532                 int size = spa_version(spa) >= SPA_VERSION_USERSPACE ?
 533                     sizeof (objset_phys_t) : OBJSET_PHYS_SIZE_V1;
 534                 os->os_phys_buf = arc_alloc_buf(spa, &os->os_phys_buf,
 535                     ARC_BUFC_METADATA, size);
 536                 os->os_phys = os->os_phys_buf->b_data;
 537                 memset(os->os_phys, 0, size);
 538         }
 539         /*
 540          * These properties will be filled in by the logic in zfs_get_zplprop()
 541          * when they are queried for the first time.
 542          */
 543         os->os_version = OBJSET_PROP_UNINITIALIZED;
 544         os->os_normalization = OBJSET_PROP_UNINITIALIZED;
 545         os->os_utf8only = OBJSET_PROP_UNINITIALIZED;
 546         os->os_casesensitivity = OBJSET_PROP_UNINITIALIZED;
 547
 548         /*
 549          * Note: the changed_cb will be called once before the register
 550          * func returns, thus changing the checksum/compression from the
 551          * default (fletcher2/off).  Snapshots don't need to know about
 552          * checksum/compression/copies.
 553          */
 554         if (ds != NULL) {
 555                 os->os_encrypted = (ds->ds_dir->dd_crypto_obj != 0);
 556
 557                 err = dsl_prop_register(ds,
 558                     zfs_prop_to_name(ZFS_PROP_PRIMARYCACHE),
 559                     primary_cache_changed_cb, os);
 560                 if (err == 0) {
 561                         err = dsl_prop_register(ds,
 562                             zfs_prop_to_name(ZFS_PROP_SECONDARYCACHE),
 563                             secondary_cache_changed_cb, os);
 564                 }
 565                 if (!ds->ds_is_snapshot) {
 566                         if (err == 0) {
 567                                 err = dsl_prop_register(ds,
 568                                     zfs_prop_to_name(ZFS_PROP_CHECKSUM),
 569                                     checksum_changed_cb, os);
 570                         }
 571                         if (err == 0) {
 572                                 err = dsl_prop_register(ds,
 573                                     zfs_prop_to_name(ZFS_PROP_COMPRESSION),
 574                                     compression_changed_cb, os);
 575                         }
 576                         if (err == 0) {
 577                                 err = dsl_prop_register(ds,
 578                                     zfs_prop_to_name(ZFS_PROP_COPIES),
 579                                     copies_changed_cb, os);
 580                         }
 581                         if (err == 0) {
 582                                 err = dsl_prop_register(ds,
 583                                     zfs_prop_to_name(ZFS_PROP_DEDUP),
 584                                     dedup_changed_cb, os);
 585                         }
 586                         if (err == 0) {
 587                                 err = dsl_prop_register(ds,
 588                                     zfs_prop_to_name(ZFS_PROP_LOGBIAS),
 589                                     logbias_changed_cb, os);
 590                         }
 591                         if (err == 0) {
 592                                 err = dsl_prop_register(ds,
 593                                     zfs_prop_to_name(ZFS_PROP_SYNC),
 594                                     sync_changed_cb, os);
 595                         }
 596                         if (err == 0) {
 597                                 err = dsl_prop_register(ds,
 598                                     zfs_prop_to_name(
 599                                     ZFS_PROP_REDUNDANT_METADATA),
 600                                     redundant_metadata_changed_cb, os);
 601                         }
 602                         if (err == 0) {
 603                                 err = dsl_prop_register(ds,
 604                                     zfs_prop_to_name(ZFS_PROP_RECORDSIZE),
 605                                     recordsize_changed_cb, os);
 606                         }
 607                         if (err == 0) {
 608                                 err = dsl_prop_register(ds,
 609                                     zfs_prop_to_name(ZFS_PROP_DNODESIZE),
 610                                     dnodesize_changed_cb, os);
 611                         }
 612                         if (err == 0) {
 613                                 err = dsl_prop_register(ds,
 614                                     zfs_prop_to_name(
 615                                     ZFS_PROP_SPECIAL_SMALL_BLOCKS),
 616                                     smallblk_changed_cb, os);
 617                         }
 618                 }
 619                 if (err != 0) {
 620                         arc_buf_destroy(os->os_phys_buf, &os->os_phys_buf);
 621                         kmem_free(os, sizeof (objset_t));
 622                         return (err);
 623                 }
 624         } else {
 625                 /* It's the meta-objset. */
 626                 os->os_checksum = ZIO_CHECKSUM_FLETCHER_4;
 627                 os->os_compress = ZIO_COMPRESS_ON;
 628                 os->os_complevel = ZIO_COMPLEVEL_DEFAULT;
 629                 os->os_encrypted = B_FALSE;
 630                 os->os_copies = spa_max_replication(spa);
 631                 os->os_dedup_checksum = ZIO_CHECKSUM_OFF;
 632                 os->os_dedup_verify = B_FALSE;
 633                 os->os_logbias = ZFS_LOGBIAS_LATENCY;
 634                 os->os_sync = ZFS_SYNC_STANDARD;
 635                 os->os_primary_cache = ZFS_CACHE_ALL;
 636                 os->os_secondary_cache = ZFS_CACHE_ALL;
 637                 os->os_dnodesize = DNODE_MIN_SIZE;
 638         }
 639
 640         if (ds == NULL || !ds->ds_is_snapshot)
 641                 os->os_zil_header = os->os_phys->os_zil_header;
 642         os->os_zil = zil_alloc(os, &os->os_zil_header);
 643
 644         for (i = 0; i < TXG_SIZE; i++) {
 645                 multilist_create(&os->os_dirty_dnodes[i], sizeof (dnode_t),
 646                     offsetof(dnode_t, dn_dirty_link[i]),
 647                     dnode_multilist_index_func);
 648         }
 649         list_create(&os->os_dnodes, sizeof (dnode_t),
 650             offsetof(dnode_t, dn_link));
 651         list_create(&os->os_downgraded_dbufs, sizeof (dmu_buf_impl_t),
 652             offsetof(dmu_buf_impl_t, db_link));
 653
 654         list_link_init(&os->os_evicting_node);
 655
 656         mutex_init(&os->os_lock, NULL, MUTEX_DEFAULT, NULL);
 657         mutex_init(&os->os_userused_lock, NULL, MUTEX_DEFAULT, NULL);
 658         mutex_init(&os->os_obj_lock, NULL, MUTEX_DEFAULT, NULL);
 659         mutex_init(&os->os_user_ptr_lock, NULL, MUTEX_DEFAULT, NULL);
 660         os->os_obj_next_percpu_len = boot_ncpus;
 661         os->os_obj_next_percpu = kmem_zalloc(os->os_obj_next_percpu_len *
 662             sizeof (os->os_obj_next_percpu[0]), KM_SLEEP);
 663
 664         dnode_special_open(os, &os->os_phys->os_meta_dnode,
 665             DMU_META_DNODE_OBJECT, &os->os_meta_dnode);
 666         if (OBJSET_BUF_HAS_USERUSED(os->os_phys_buf)) {
 667                 dnode_special_open(os, &os->os_phys->os_userused_dnode,
 668                     DMU_USERUSED_OBJECT, &os->os_userused_dnode);
 669                 dnode_special_open(os, &os->os_phys->os_groupused_dnode,
 670                     DMU_GROUPUSED_OBJECT, &os->os_groupused_dnode);
 671                 if (OBJSET_BUF_HAS_PROJECTUSED(os->os_phys_buf))
 672                         dnode_special_open(os,
 673                             &os->os_phys->os_projectused_dnode,
 674                             DMU_PROJECTUSED_OBJECT, &os->os_projectused_dnode);
 675         }
 676
 677         mutex_init(&os->os_upgrade_lock, NULL, MUTEX_DEFAULT, NULL);
 678
 679         *osp = os;
 680         return (0);
 681 }
 682
 683 int
 684 dmu_objset_from_ds(dsl_dataset_t *ds, objset_t **osp)
 685 {
 686         int err = 0;
 687
 688         /*
 689          * We need the pool_config lock to manipulate the dsl_dataset_t.
 690          * Even if the dataset is long-held, we need the pool_config lock
 691          * to open the objset, as it needs to get properties.
 692          */
 693         ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
 694
 695         mutex_enter(&ds->ds_opening_lock);
 696         if (ds->ds_objset == NULL) {
 697                 objset_t *os;
 698                 rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
 699                 err = dmu_objset_open_impl(dsl_dataset_get_spa(ds),
 700                     ds, dsl_dataset_get_blkptr(ds), &os);
 701                 rrw_exit(&ds->ds_bp_rwlock, FTAG);
 702
 703                 if (err == 0) {
 704                         mutex_enter(&ds->ds_lock);
 705                         ASSERT(ds->ds_objset == NULL);
 706                         ds->ds_objset = os;
 707                         mutex_exit(&ds->ds_lock);
 708                 }
 709         }
 710         *osp = ds->ds_objset;
 711         mutex_exit(&ds->ds_opening_lock);
 712         return (err);
 713 }
 714
 715 /*
 716  * Holds the pool while the objset is held.  Therefore only one objset
 717  * can be held at a time.
 718  */
 719 int
 720 dmu_objset_hold_flags(const char *name, boolean_t decrypt, const void *tag,
 721     objset_t **osp)
 722 {
 723         dsl_pool_t *dp;
 724         dsl_dataset_t *ds;
 725         int err;
 726         ds_hold_flags_t flags;
 727
 728         flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
 729         err = dsl_pool_hold(name, tag, &dp);
 730         if (err != 0)
 731                 return (err);
 732         err = dsl_dataset_hold_flags(dp, name, flags, tag, &ds);
 733         if (err != 0) {
 734                 dsl_pool_rele(dp, tag);
 735                 return (err);
 736         }
 737
 738         err = dmu_objset_from_ds(ds, osp);
 739         if (err != 0) {
 740                 dsl_dataset_rele(ds, tag);
 741                 dsl_pool_rele(dp, tag);
 742         }
 743
 744         return (err);
 745 }
 746
 747 int
 748 dmu_objset_hold(const char *name, const void *tag, objset_t **osp)
 749 {
 750         return (dmu_objset_hold_flags(name, B_FALSE, tag, osp));
 751 }
 752
 753 static int
 754 dmu_objset_own_impl(dsl_dataset_t *ds, dmu_objset_type_t type,
 755     boolean_t readonly, boolean_t decrypt, const void *tag, objset_t **osp)
 756 {
 757         (void) tag;
 758
 759         int err = dmu_objset_from_ds(ds, osp);
 760         if (err != 0) {
 761                 return (err);
 762         } else if (type != DMU_OST_ANY && type != (*osp)->os_phys->os_type) {
 763                 return (SET_ERROR(EINVAL));
 764         } else if (!readonly && dsl_dataset_is_snapshot(ds)) {
 765                 return (SET_ERROR(EROFS));
 766         } else if (!readonly && decrypt &&
 767             dsl_dir_incompatible_encryption_version(ds->ds_dir)) {
 768                 return (SET_ERROR(EROFS));
 769         }
 770
 771         /* if we are decrypting, we can now check MACs in os->os_phys_buf */
 772         if (decrypt && arc_is_unauthenticated((*osp)->os_phys_buf)) {
 773                 zbookmark_phys_t zb;
 774
 775                 SET_BOOKMARK(&zb, ds->ds_object, ZB_ROOT_OBJECT,
 776                     ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
 777                 err = arc_untransform((*osp)->os_phys_buf, (*osp)->os_spa,
 778                     &zb, B_FALSE);
 779                 if (err != 0)
 780                         return (err);
 781
 782                 ASSERT0(arc_is_unauthenticated((*osp)->os_phys_buf));
 783         }
 784
 785         return (0);
 786 }
 787
 788 /*
 789  * dsl_pool must not be held when this is called.
 790  * Upon successful return, there will be a longhold on the dataset,
 791  * and the dsl_pool will not be held.
 792  */
 793 int
 794 dmu_objset_own(const char *name, dmu_objset_type_t type,
 795     boolean_t readonly, boolean_t decrypt, const void *tag, objset_t **osp)
 796 {
 797         dsl_pool_t *dp;
 798         dsl_dataset_t *ds;
 799         int err;
 800         ds_hold_flags_t flags;
 801
 802         flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
 803         err = dsl_pool_hold(name, FTAG, &dp);
 804         if (err != 0)
 805                 return (err);
 806         err = dsl_dataset_own(dp, name, flags, tag, &ds);
 807         if (err != 0) {
 808                 dsl_pool_rele(dp, FTAG);
 809                 return (err);
 810         }
 811         err = dmu_objset_own_impl(ds, type, readonly, decrypt, tag, osp);
 812         if (err != 0) {
 813                 dsl_dataset_disown(ds, flags, tag);
 814                 dsl_pool_rele(dp, FTAG);
 815                 return (err);
 816         }
 817
 818         /*
 819          * User accounting requires the dataset to be decrypted and rw.
 820          * We also don't begin user accounting during claiming to help
 821          * speed up pool import times and to keep this txg reserved
 822          * completely for recovery work.
 823          */
 824         if (!readonly && !dp->dp_spa->spa_claiming &&
 825             (ds->ds_dir->dd_crypto_obj == 0 || decrypt)) {
 826                 if (dmu_objset_userobjspace_upgradable(*osp) ||
 827                     dmu_objset_projectquota_upgradable(*osp)) {
 828                         dmu_objset_id_quota_upgrade(*osp);
 829                 } else if (dmu_objset_userused_enabled(*osp)) {
 830                         dmu_objset_userspace_upgrade(*osp);
 831                 }
 832         }
 833
 834         dsl_pool_rele(dp, FTAG);
 835         return (0);
 836 }
 837
 838 int
 839 dmu_objset_own_obj(dsl_pool_t *dp, uint64_t obj, dmu_objset_type_t type,
 840     boolean_t readonly, boolean_t decrypt, const void *tag, objset_t **osp)
 841 {
 842         dsl_dataset_t *ds;
 843         int err;
 844         ds_hold_flags_t flags;
 845
 846         flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
 847         err = dsl_dataset_own_obj(dp, obj, flags, tag, &ds);
 848         if (err != 0)
 849                 return (err);
 850
 851         err = dmu_objset_own_impl(ds, type, readonly, decrypt, tag, osp);
 852         if (err != 0) {
 853                 dsl_dataset_disown(ds, flags, tag);
 854                 return (err);
 855         }
 856
 857         return (0);
 858 }
 859
 860 void
 861 dmu_objset_rele_flags(objset_t *os, boolean_t decrypt, const void *tag)
 862 {
 863         ds_hold_flags_t flags;
 864         dsl_pool_t *dp = dmu_objset_pool(os);
 865
 866         flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
 867         dsl_dataset_rele_flags(os->os_dsl_dataset, flags, tag);
 868         dsl_pool_rele(dp, tag);
 869 }
 870
 871 void
 872 dmu_objset_rele(objset_t *os, const void *tag)
 873 {
 874         dmu_objset_rele_flags(os, B_FALSE, tag);
 875 }
 876
 877 /*
 878  * When we are called, os MUST refer to an objset associated with a dataset
 879  * that is owned by 'tag'; that is, is held and long held by 'tag' and ds_owner
 880  * == tag.  We will then release and reacquire ownership of the dataset while
 881  * holding the pool config_rwlock to avoid intervening namespace or ownership
 882  * changes may occur.
 883  *
 884  * This exists solely to accommodate zfs_ioc_userspace_upgrade()'s desire to
 885  * release the hold on its dataset and acquire a new one on the dataset of the
 886  * same name so that it can be partially torn down and reconstructed.
 887  */
 888 void
 889 dmu_objset_refresh_ownership(dsl_dataset_t *ds, dsl_dataset_t **newds,
 890     boolean_t decrypt, const void *tag)
 891 {
 892         dsl_pool_t *dp;
 893         char name[ZFS_MAX_DATASET_NAME_LEN];
 894         ds_hold_flags_t flags;
 895
 896         flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
 897         VERIFY3P(ds, !=, NULL);
 898         VERIFY3P(ds->ds_owner, ==, tag);
 899         VERIFY(dsl_dataset_long_held(ds));
 900
 901         dsl_dataset_name(ds, name);
 902         dp = ds->ds_dir->dd_pool;
 903         dsl_pool_config_enter(dp, FTAG);
 904         dsl_dataset_disown(ds, flags, tag);
 905         VERIFY0(dsl_dataset_own(dp, name, flags, tag, newds));
 906         dsl_pool_config_exit(dp, FTAG);
 907 }
 908
 909 void
 910 dmu_objset_disown(objset_t *os, boolean_t decrypt, const void *tag)
 911 {
 912         ds_hold_flags_t flags;
 913
 914         flags = (decrypt) ? DS_HOLD_FLAG_DECRYPT : DS_HOLD_FLAG_NONE;
 915         /*
 916          * Stop upgrading thread
 917          */
 918         dmu_objset_upgrade_stop(os);
 919         dsl_dataset_disown(os->os_dsl_dataset, flags, tag);
 920 }
 921
 922 void
 923 dmu_objset_evict_dbufs(objset_t *os)
 924 {
 925         dnode_t *dn_marker;
 926         dnode_t *dn;
 927
 928         dn_marker = kmem_alloc(sizeof (dnode_t), KM_SLEEP);
 929
 930         mutex_enter(&os->os_lock);
 931         dn = list_head(&os->os_dnodes);
 932         while (dn != NULL) {
 933                 /*
 934                  * Skip dnodes without holds.  We have to do this dance
 935                  * because dnode_add_ref() only works if there is already a
 936                  * hold.  If the dnode has no holds, then it has no dbufs.
 937                  */
 938                 if (dnode_add_ref(dn, FTAG)) {
 939                         list_insert_after(&os->os_dnodes, dn, dn_marker);
 940                         mutex_exit(&os->os_lock);
 941
 942                         dnode_evict_dbufs(dn);
 943                         dnode_rele(dn, FTAG);
 944
 945                         mutex_enter(&os->os_lock);
 946                         dn = list_next(&os->os_dnodes, dn_marker);
 947                         list_remove(&os->os_dnodes, dn_marker);
 948                 } else {
 949                         dn = list_next(&os->os_dnodes, dn);
 950                 }
 951         }
 952         mutex_exit(&os->os_lock);
 953
 954         kmem_free(dn_marker, sizeof (dnode_t));
 955
 956         if (DMU_USERUSED_DNODE(os) != NULL) {
 957                 if (DMU_PROJECTUSED_DNODE(os) != NULL)
 958                         dnode_evict_dbufs(DMU_PROJECTUSED_DNODE(os));
 959                 dnode_evict_dbufs(DMU_GROUPUSED_DNODE(os));
 960                 dnode_evict_dbufs(DMU_USERUSED_DNODE(os));
 961         }
 962         dnode_evict_dbufs(DMU_META_DNODE(os));
 963 }
 964
 965 /*
 966  * Objset eviction processing is split into into two pieces.
 967  * The first marks the objset as evicting, evicts any dbufs that
 968  * have a refcount of zero, and then queues up the objset for the
 969  * second phase of eviction.  Once os->os_dnodes has been cleared by
 970  * dnode_buf_pageout()->dnode_destroy(), the second phase is executed.
 971  * The second phase closes the special dnodes, dequeues the objset from
 972  * the list of those undergoing eviction, and finally frees the objset.
 973  *
 974  * NOTE: Due to asynchronous eviction processing (invocation of
 975  *       dnode_buf_pageout()), it is possible for the meta dnode for the
 976  *       objset to have no holds even though os->os_dnodes is not empty.
 977  */
 978 void
 979 dmu_objset_evict(objset_t *os)
 980 {
 981         dsl_dataset_t *ds = os->os_dsl_dataset;
 982
 983         for (int t = 0; t < TXG_SIZE; t++)
 984                 ASSERT(!dmu_objset_is_dirty(os, t));
 985
 986         if (ds)
 987                 dsl_prop_unregister_all(ds, os);
 988
 989         if (os->os_sa)
 990                 sa_tear_down(os);
 991
 992         dmu_objset_evict_dbufs(os);
 993
 994         mutex_enter(&os->os_lock);
 995         spa_evicting_os_register(os->os_spa, os);
 996         if (list_is_empty(&os->os_dnodes)) {
 997                 mutex_exit(&os->os_lock);
 998                 dmu_objset_evict_done(os);
 999         } else {
1000                 mutex_exit(&os->os_lock);
1001         }
1002
1003
1004 }
1005
1006 void
1007 dmu_objset_evict_done(objset_t *os)
1008 {
1009         ASSERT3P(list_head(&os->os_dnodes), ==, NULL);
1010
1011         dnode_special_close(&os->os_meta_dnode);
1012         if (DMU_USERUSED_DNODE(os)) {
1013                 if (DMU_PROJECTUSED_DNODE(os))
1014                         dnode_special_close(&os->os_projectused_dnode);
1015                 dnode_special_close(&os->os_userused_dnode);
1016                 dnode_special_close(&os->os_groupused_dnode);
1017         }
1018         zil_free(os->os_zil);
1019
1020         arc_buf_destroy(os->os_phys_buf, &os->os_phys_buf);
1021
1022         /*
1023          * This is a barrier to prevent the objset from going away in
1024          * dnode_move() until we can safely ensure that the objset is still in
1025          * use. We consider the objset valid before the barrier and invalid
1026          * after the barrier.
1027          */
1028         rw_enter(&os_lock, RW_READER);
1029         rw_exit(&os_lock);
1030
1031         kmem_free(os->os_obj_next_percpu,
1032             os->os_obj_next_percpu_len * sizeof (os->os_obj_next_percpu[0]));
1033
1034         mutex_destroy(&os->os_lock);
1035         mutex_destroy(&os->os_userused_lock);
1036         mutex_destroy(&os->os_obj_lock);
1037         mutex_destroy(&os->os_user_ptr_lock);
1038         mutex_destroy(&os->os_upgrade_lock);
1039         for (int i = 0; i < TXG_SIZE; i++)
1040                 multilist_destroy(&os->os_dirty_dnodes[i]);
1041         spa_evicting_os_deregister(os->os_spa, os);
1042         kmem_free(os, sizeof (objset_t));
1043 }
1044
1045 inode_timespec_t
1046 dmu_objset_snap_cmtime(objset_t *os)
1047 {
1048         return (dsl_dir_snap_cmtime(os->os_dsl_dataset->ds_dir));
1049 }
1050
1051 objset_t *
1052 dmu_objset_create_impl_dnstats(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
1053     dmu_objset_type_t type, int levels, int blksz, int ibs, dmu_tx_t *tx)
1054 {
1055         objset_t *os;
1056         dnode_t *mdn;
1057
1058         ASSERT(dmu_tx_is_syncing(tx));
1059
1060         if (blksz == 0)
1061                 blksz = DNODE_BLOCK_SIZE;
1062         if (ibs == 0)
1063                 ibs = DN_MAX_INDBLKSHIFT;
1064
1065         if (ds != NULL)
1066                 VERIFY0(dmu_objset_from_ds(ds, &os));
1067         else
1068                 VERIFY0(dmu_objset_open_impl(spa, NULL, bp, &os));
1069
1070         mdn = DMU_META_DNODE(os);
1071
1072         dnode_allocate(mdn, DMU_OT_DNODE, blksz, ibs, DMU_OT_NONE, 0,
1073             DNODE_MIN_SLOTS, tx);
1074
1075         /*
1076          * We don't want to have to increase the meta-dnode's nlevels
1077          * later, because then we could do it in quiescing context while
1078          * we are also accessing it in open context.
1079          *
1080          * This precaution is not necessary for the MOS (ds == NULL),
1081          * because the MOS is only updated in syncing context.
1082          * This is most fortunate: the MOS is the only objset that
1083          * needs to be synced multiple times as spa_sync() iterates
1084          * to convergence, so minimizing its dn_nlevels matters.
1085          */
1086         if (ds != NULL) {
1087                 if (levels == 0) {
1088                         levels = 1;
1089
1090                         /*
1091                          * Determine the number of levels necessary for the
1092                          * meta-dnode to contain DN_MAX_OBJECT dnodes.  Note
1093                          * that in order to ensure that we do not overflow
1094                          * 64 bits, there has to be a nlevels that gives us a
1095                          * number of blocks > DN_MAX_OBJECT but < 2^64.
1096                          * Therefore, (mdn->dn_indblkshift - SPA_BLKPTRSHIFT)
1097                          * (10) must be less than (64 - log2(DN_MAX_OBJECT))
1098                          * (16).
1099                          */
1100                         while ((uint64_t)mdn->dn_nblkptr <<
1101                             (mdn->dn_datablkshift - DNODE_SHIFT + (levels - 1) *
1102                             (mdn->dn_indblkshift - SPA_BLKPTRSHIFT)) <
1103                             DN_MAX_OBJECT)
1104                                 levels++;
1105                 }
1106
1107                 mdn->dn_next_nlevels[tx->tx_txg & TXG_MASK] =
1108                     mdn->dn_nlevels = levels;
1109         }
1110
1111         ASSERT(type != DMU_OST_NONE);
1112         ASSERT(type != DMU_OST_ANY);
1113         ASSERT(type < DMU_OST_NUMTYPES);
1114         os->os_phys->os_type = type;
1115
1116         /*
1117          * Enable user accounting if it is enabled and this is not an
1118          * encrypted receive.
1119          */
1120         if (dmu_objset_userused_enabled(os) &&
1121             (!os->os_encrypted || !dmu_objset_is_receiving(os))) {
1122                 os->os_phys->os_flags |= OBJSET_FLAG_USERACCOUNTING_COMPLETE;
1123                 if (dmu_objset_userobjused_enabled(os)) {
1124                         ds->ds_feature_activation[
1125                             SPA_FEATURE_USEROBJ_ACCOUNTING] = (void *)B_TRUE;
1126                         os->os_phys->os_flags |=
1127                             OBJSET_FLAG_USEROBJACCOUNTING_COMPLETE;
1128                 }
1129                 if (dmu_objset_projectquota_enabled(os)) {
1130                         ds->ds_feature_activation[
1131                             SPA_FEATURE_PROJECT_QUOTA] = (void *)B_TRUE;
1132                         os->os_phys->os_flags |=
1133                             OBJSET_FLAG_PROJECTQUOTA_COMPLETE;
1134                 }
1135                 os->os_flags = os->os_phys->os_flags;
1136         }
1137
1138         dsl_dataset_dirty(ds, tx);
1139
1140         return (os);
1141 }
1142
1143 /* called from dsl for meta-objset */
1144 objset_t *
1145 dmu_objset_create_impl(spa_t *spa, dsl_dataset_t *ds, blkptr_t *bp,
1146     dmu_objset_type_t type, dmu_tx_t *tx)
1147 {
1148         return (dmu_objset_create_impl_dnstats(spa, ds, bp, type, 0, 0, 0, tx));
1149 }
1150
1151 typedef struct dmu_objset_create_arg {
1152         const char *doca_name;
1153         cred_t *doca_cred;
1154         proc_t *doca_proc;
1155         void (*doca_userfunc)(objset_t *os, void *arg,
1156             cred_t *cr, dmu_tx_t *tx);
1157         void *doca_userarg;
1158         dmu_objset_type_t doca_type;
1159         uint64_t doca_flags;
1160         dsl_crypto_params_t *doca_dcp;
1161 } dmu_objset_create_arg_t;
1162
1163 static int
1164 dmu_objset_create_check(void *arg, dmu_tx_t *tx)
1165 {
1166         dmu_objset_create_arg_t *doca = arg;
1167         dsl_pool_t *dp = dmu_tx_pool(tx);
1168         dsl_dir_t *pdd;
1169         dsl_dataset_t *parentds;
1170         objset_t *parentos;
1171         const char *tail;
1172         int error;
1173
1174         if (strchr(doca->doca_name, '@') != NULL)
1175                 return (SET_ERROR(EINVAL));
1176
1177         if (strlen(doca->doca_name) >= ZFS_MAX_DATASET_NAME_LEN)
1178                 return (SET_ERROR(ENAMETOOLONG));
1179
1180         if (dataset_nestcheck(doca->doca_name) != 0)
1181                 return (SET_ERROR(ENAMETOOLONG));
1182
1183         error = dsl_dir_hold(dp, doca->doca_name, FTAG, &pdd, &tail);
1184         if (error != 0)
1185                 return (error);
1186         if (tail == NULL) {
1187                 dsl_dir_rele(pdd, FTAG);
1188                 return (SET_ERROR(EEXIST));
1189         }
1190
1191         error = dmu_objset_create_crypt_check(pdd, doca->doca_dcp, NULL);
1192         if (error != 0) {
1193                 dsl_dir_rele(pdd, FTAG);
1194                 return (error);
1195         }
1196
1197         error = dsl_fs_ss_limit_check(pdd, 1, ZFS_PROP_FILESYSTEM_LIMIT, NULL,
1198             doca->doca_cred, doca->doca_proc);
1199         if (error != 0) {
1200                 dsl_dir_rele(pdd, FTAG);
1201                 return (error);
1202         }
1203
1204         /* can't create below anything but filesystems (eg. no ZVOLs) */
1205         error = dsl_dataset_hold_obj(pdd->dd_pool,
1206             dsl_dir_phys(pdd)->dd_head_dataset_obj, FTAG, &parentds);
1207         if (error != 0) {
1208                 dsl_dir_rele(pdd, FTAG);
1209                 return (error);
1210         }
1211         error = dmu_objset_from_ds(parentds, &parentos);
1212         if (error != 0) {
1213                 dsl_dataset_rele(parentds, FTAG);
1214                 dsl_dir_rele(pdd, FTAG);
1215                 return (error);
1216         }
1217         if (dmu_objset_type(parentos) != DMU_OST_ZFS) {
1218                 dsl_dataset_rele(parentds, FTAG);
1219                 dsl_dir_rele(pdd, FTAG);
1220                 return (SET_ERROR(ZFS_ERR_WRONG_PARENT));
1221         }
1222         dsl_dataset_rele(parentds, FTAG);
1223         dsl_dir_rele(pdd, FTAG);
1224
1225         return (error);
1226 }
1227
1228 static void
1229 dmu_objset_create_sync(void *arg, dmu_tx_t *tx)
1230 {
1231         dmu_objset_create_arg_t *doca = arg;
1232         dsl_pool_t *dp = dmu_tx_pool(tx);
1233         spa_t *spa = dp->dp_spa;
1234         dsl_dir_t *pdd;
1235         const char *tail;
1236         dsl_dataset_t *ds;
1237         uint64_t obj;
1238         blkptr_t *bp;
1239         objset_t *os;
1240         zio_t *rzio;
1241
1242         VERIFY0(dsl_dir_hold(dp, doca->doca_name, FTAG, &pdd, &tail));
1243
1244         obj = dsl_dataset_create_sync(pdd, tail, NULL, doca->doca_flags,
1245             doca->doca_cred, doca->doca_dcp, tx);
1246
1247         VERIFY0(dsl_dataset_hold_obj_flags(pdd->dd_pool, obj,
1248             DS_HOLD_FLAG_DECRYPT, FTAG, &ds));
1249         rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
1250         bp = dsl_dataset_get_blkptr(ds);
1251         os = dmu_objset_create_impl(spa, ds, bp, doca->doca_type, tx);
1252         rrw_exit(&ds->ds_bp_rwlock, FTAG);
1253
1254         if (doca->doca_userfunc != NULL) {
1255                 doca->doca_userfunc(os, doca->doca_userarg,
1256                     doca->doca_cred, tx);
1257         }
1258
1259         /*
1260          * The doca_userfunc() may write out some data that needs to be
1261          * encrypted if the dataset is encrypted (specifically the root
1262          * directory).  This data must be written out before the encryption
1263          * key mapping is removed by dsl_dataset_rele_flags().  Force the
1264          * I/O to occur immediately by invoking the relevant sections of
1265          * dsl_pool_sync().
1266          */
1267         if (os->os_encrypted) {
1268                 dsl_dataset_t *tmpds = NULL;
1269                 boolean_t need_sync_done = B_FALSE;
1270
1271                 mutex_enter(&ds->ds_lock);
1272                 ds->ds_owner = FTAG;
1273                 mutex_exit(&ds->ds_lock);
1274
1275                 rzio = zio_root(spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
1276                 tmpds = txg_list_remove_this(&dp->dp_dirty_datasets, ds,
1277                     tx->tx_txg);
1278                 if (tmpds != NULL) {
1279                         dsl_dataset_sync(ds, rzio, tx);
1280                         need_sync_done = B_TRUE;
1281                 }
1282                 VERIFY0(zio_wait(rzio));
1283
1284                 dmu_objset_sync_done(os, tx);
1285                 taskq_wait(dp->dp_sync_taskq);
1286                 if (txg_list_member(&dp->dp_dirty_datasets, ds, tx->tx_txg)) {
1287                         ASSERT3P(ds->ds_key_mapping, !=, NULL);
1288                         key_mapping_rele(spa, ds->ds_key_mapping, ds);
1289                 }
1290
1291                 rzio = zio_root(spa, NULL, NULL, ZIO_FLAG_MUSTSUCCEED);
1292                 tmpds = txg_list_remove_this(&dp->dp_dirty_datasets, ds,
1293                     tx->tx_txg);
1294                 if (tmpds != NULL) {
1295                         dmu_buf_rele(ds->ds_dbuf, ds);
1296                         dsl_dataset_sync(ds, rzio, tx);
1297                 }
1298                 VERIFY0(zio_wait(rzio));
1299
1300                 if (need_sync_done) {
1301                         ASSERT3P(ds->ds_key_mapping, !=, NULL);
1302                         key_mapping_rele(spa, ds->ds_key_mapping, ds);
1303                         dsl_dataset_sync_done(ds, tx);
1304                 }
1305
1306                 mutex_enter(&ds->ds_lock);
1307                 ds->ds_owner = NULL;
1308                 mutex_exit(&ds->ds_lock);
1309         }
1310
1311         spa_history_log_internal_ds(ds, "create", tx, " ");
1312
1313         dsl_dataset_rele_flags(ds, DS_HOLD_FLAG_DECRYPT, FTAG);
1314         dsl_dir_rele(pdd, FTAG);
1315 }
1316
1317 int
1318 dmu_objset_create(const char *name, dmu_objset_type_t type, uint64_t flags,
1319     dsl_crypto_params_t *dcp, dmu_objset_create_sync_func_t func, void *arg)
1320 {
1321         dmu_objset_create_arg_t doca;
1322         dsl_crypto_params_t tmp_dcp = { 0 };
1323
1324         doca.doca_name = name;
1325         doca.doca_cred = CRED();
1326         doca.doca_proc = curproc;
1327         doca.doca_flags = flags;
1328         doca.doca_userfunc = func;
1329         doca.doca_userarg = arg;
1330         doca.doca_type = type;
1331
1332         /*
1333          * Some callers (mostly for testing) do not provide a dcp on their
1334          * own but various code inside the sync task will require it to be
1335          * allocated. Rather than adding NULL checks throughout this code
1336          * or adding dummy dcp's to all of the callers we simply create a
1337          * dummy one here and use that. This zero dcp will have the same
1338          * effect as asking for inheritance of all encryption params.
1339          */
1340         doca.doca_dcp = (dcp != NULL) ? dcp : &tmp_dcp;
1341
1342         int rv = dsl_sync_task(name,
1343             dmu_objset_create_check, dmu_objset_create_sync, &doca,
1344             6, ZFS_SPACE_CHECK_NORMAL);
1345
1346         if (rv == 0)
1347                 zvol_create_minor(name);
1348         return (rv);
1349 }
1350
1351 typedef struct dmu_objset_clone_arg {
1352         const char *doca_clone;
1353         const char *doca_origin;
1354         cred_t *doca_cred;
1355         proc_t *doca_proc;
1356 } dmu_objset_clone_arg_t;
1357
1358 static int
1359 dmu_objset_clone_check(void *arg, dmu_tx_t *tx)
1360 {
1361         dmu_objset_clone_arg_t *doca = arg;
1362         dsl_dir_t *pdd;
1363         const char *tail;
1364         int error;
1365         dsl_dataset_t *origin;
1366         dsl_pool_t *dp = dmu_tx_pool(tx);
1367
1368         if (strchr(doca->doca_clone, '@') != NULL)
1369                 return (SET_ERROR(EINVAL));
1370
1371         if (strlen(doca->doca_clone) >= ZFS_MAX_DATASET_NAME_LEN)
1372                 return (SET_ERROR(ENAMETOOLONG));
1373
1374         error = dsl_dir_hold(dp, doca->doca_clone, FTAG, &pdd, &tail);
1375         if (error != 0)
1376                 return (error);
1377         if (tail == NULL) {
1378                 dsl_dir_rele(pdd, FTAG);
1379                 return (SET_ERROR(EEXIST));
1380         }
1381
1382         error = dsl_fs_ss_limit_check(pdd, 1, ZFS_PROP_FILESYSTEM_LIMIT, NULL,
1383             doca->doca_cred, doca->doca_proc);
1384         if (error != 0) {
1385                 dsl_dir_rele(pdd, FTAG);
1386                 return (SET_ERROR(EDQUOT));
1387         }
1388
1389         error = dsl_dataset_hold(dp, doca->doca_origin, FTAG, &origin);
1390         if (error != 0) {
1391                 dsl_dir_rele(pdd, FTAG);
1392                 return (error);
1393         }
1394
1395         /* You can only clone snapshots, not the head datasets. */
1396         if (!origin->ds_is_snapshot) {
1397                 dsl_dataset_rele(origin, FTAG);
1398                 dsl_dir_rele(pdd, FTAG);
1399                 return (SET_ERROR(EINVAL));
1400         }
1401
1402         dsl_dataset_rele(origin, FTAG);
1403         dsl_dir_rele(pdd, FTAG);
1404
1405         return (0);
1406 }
1407
1408 static void
1409 dmu_objset_clone_sync(void *arg, dmu_tx_t *tx)
1410 {
1411         dmu_objset_clone_arg_t *doca = arg;
1412         dsl_pool_t *dp = dmu_tx_pool(tx);
1413         dsl_dir_t *pdd;
1414         const char *tail;
1415         dsl_dataset_t *origin, *ds;
1416         uint64_t obj;
1417         char namebuf[ZFS_MAX_DATASET_NAME_LEN];
1418
1419         VERIFY0(dsl_dir_hold(dp, doca->doca_clone, FTAG, &pdd, &tail));
1420         VERIFY0(dsl_dataset_hold(dp, doca->doca_origin, FTAG, &origin));
1421
1422         obj = dsl_dataset_create_sync(pdd, tail, origin, 0,
1423             doca->doca_cred, NULL, tx);
1424
1425         VERIFY0(dsl_dataset_hold_obj(pdd->dd_pool, obj, FTAG, &ds));
1426         dsl_dataset_name(origin, namebuf);
1427         spa_history_log_internal_ds(ds, "clone", tx,
1428             "origin=%s (%llu)", namebuf, (u_longlong_t)origin->ds_object);
1429         dsl_dataset_rele(ds, FTAG);
1430         dsl_dataset_rele(origin, FTAG);
1431         dsl_dir_rele(pdd, FTAG);
1432 }
1433
1434 int
1435 dmu_objset_clone(const char *clone, const char *origin)
1436 {
1437         dmu_objset_clone_arg_t doca;
1438
1439         doca.doca_clone = clone;
1440         doca.doca_origin = origin;
1441         doca.doca_cred = CRED();
1442         doca.doca_proc = curproc;
1443
1444         int rv = dsl_sync_task(clone,
1445             dmu_objset_clone_check, dmu_objset_clone_sync, &doca,
1446             6, ZFS_SPACE_CHECK_NORMAL);
1447
1448         if (rv == 0)
1449                 zvol_create_minor(clone);
1450
1451         return (rv);
1452 }
1453
1454 int
1455 dmu_objset_snapshot_one(const char *fsname, const char *snapname)
1456 {
1457         int err;
1458         char *longsnap = kmem_asprintf("%s@%s", fsname, snapname);
1459         nvlist_t *snaps = fnvlist_alloc();
1460
1461         fnvlist_add_boolean(snaps, longsnap);
1462         kmem_strfree(longsnap);
1463         err = dsl_dataset_snapshot(snaps, NULL, NULL);
1464         fnvlist_free(snaps);
1465         return (err);
1466 }
1467
1468 static void
1469 dmu_objset_upgrade_task_cb(void *data)
1470 {
1471         objset_t *os = data;
1472
1473         mutex_enter(&os->os_upgrade_lock);
1474         os->os_upgrade_status = EINTR;
1475         if (!os->os_upgrade_exit) {
1476                 int status;
1477
1478                 mutex_exit(&os->os_upgrade_lock);
1479
1480                 status = os->os_upgrade_cb(os);
1481
1482                 mutex_enter(&os->os_upgrade_lock);
1483
1484                 os->os_upgrade_status = status;
1485         }
1486         os->os_upgrade_exit = B_TRUE;
1487         os->os_upgrade_id = 0;
1488         mutex_exit(&os->os_upgrade_lock);
1489         dsl_dataset_long_rele(dmu_objset_ds(os), upgrade_tag);
1490 }
1491
1492 static void
1493 dmu_objset_upgrade(objset_t *os, dmu_objset_upgrade_cb_t cb)
1494 {
1495         if (os->os_upgrade_id != 0)
1496                 return;
1497
1498         ASSERT(dsl_pool_config_held(dmu_objset_pool(os)));
1499         dsl_dataset_long_hold(dmu_objset_ds(os), upgrade_tag);
1500
1501         mutex_enter(&os->os_upgrade_lock);
1502         if (os->os_upgrade_id == 0 && os->os_upgrade_status == 0) {
1503                 os->os_upgrade_exit = B_FALSE;
1504                 os->os_upgrade_cb = cb;
1505                 os->os_upgrade_id = taskq_dispatch(
1506                     os->os_spa->spa_upgrade_taskq,
1507                     dmu_objset_upgrade_task_cb, os, TQ_SLEEP);
1508                 if (os->os_upgrade_id == TASKQID_INVALID) {
1509                         dsl_dataset_long_rele(dmu_objset_ds(os), upgrade_tag);
1510                         os->os_upgrade_status = ENOMEM;
1511                 }
1512         } else {
1513                 dsl_dataset_long_rele(dmu_objset_ds(os), upgrade_tag);
1514         }
1515         mutex_exit(&os->os_upgrade_lock);
1516 }
1517
1518 static void
1519 dmu_objset_upgrade_stop(objset_t *os)
1520 {
1521         mutex_enter(&os->os_upgrade_lock);
1522         os->os_upgrade_exit = B_TRUE;
1523         if (os->os_upgrade_id != 0) {
1524                 taskqid_t id = os->os_upgrade_id;
1525
1526                 os->os_upgrade_id = 0;
1527                 mutex_exit(&os->os_upgrade_lock);
1528
1529                 if ((taskq_cancel_id(os->os_spa->spa_upgrade_taskq, id)) == 0) {
1530                         dsl_dataset_long_rele(dmu_objset_ds(os), upgrade_tag);
1531                 }
1532                 txg_wait_synced(os->os_spa->spa_dsl_pool, 0);
1533         } else {
1534                 mutex_exit(&os->os_upgrade_lock);
1535         }
1536 }
1537
1538 static void
1539 dmu_objset_sync_dnodes(multilist_sublist_t *list, dmu_tx_t *tx)
1540 {
1541         dnode_t *dn;
1542
1543         while ((dn = multilist_sublist_head(list)) != NULL) {
1544                 ASSERT(dn->dn_object != DMU_META_DNODE_OBJECT);
1545                 ASSERT(dn->dn_dbuf->db_data_pending);
1546                 /*
1547                  * Initialize dn_zio outside dnode_sync() because the
1548                  * meta-dnode needs to set it outside dnode_sync().
1549                  */
1550                 dn->dn_zio = dn->dn_dbuf->db_data_pending->dr_zio;
1551                 ASSERT(dn->dn_zio);
1552
1553                 ASSERT3U(dn->dn_nlevels, <=, DN_MAX_LEVELS);
1554                 multilist_sublist_remove(list, dn);
1555
1556                 /*
1557                  * See the comment above dnode_rele_task() for an explanation
1558                  * of why this dnode hold is always needed (even when not
1559                  * doing user accounting).
1560                  */
1561                 multilist_t *newlist = &dn->dn_objset->os_synced_dnodes;
1562                 (void) dnode_add_ref(dn, newlist);
1563                 multilist_insert(newlist, dn);
1564
1565                 dnode_sync(dn, tx);
1566         }
1567 }
1568
1569 static void
1570 dmu_objset_write_ready(zio_t *zio, arc_buf_t *abuf, void *arg)
1571 {
1572         (void) abuf;
1573         blkptr_t *bp = zio->io_bp;
1574         objset_t *os = arg;
1575         dnode_phys_t *dnp = &os->os_phys->os_meta_dnode;
1576         uint64_t fill = 0;
1577
1578         ASSERT(!BP_IS_EMBEDDED(bp));
1579         ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_OBJSET);
1580         ASSERT0(BP_GET_LEVEL(bp));
1581
1582         /*
1583          * Update rootbp fill count: it should be the number of objects
1584          * allocated in the object set (not counting the "special"
1585          * objects that are stored in the objset_phys_t -- the meta
1586          * dnode and user/group/project accounting objects).
1587          */
1588         for (int i = 0; i < dnp->dn_nblkptr; i++)
1589                 fill += BP_GET_FILL(&dnp->dn_blkptr[i]);
1590
1591         BP_SET_FILL(bp, fill);
1592
1593         if (os->os_dsl_dataset != NULL)
1594                 rrw_enter(&os->os_dsl_dataset->ds_bp_rwlock, RW_WRITER, FTAG);
1595         *os->os_rootbp = *bp;
1596         if (os->os_dsl_dataset != NULL)
1597                 rrw_exit(&os->os_dsl_dataset->ds_bp_rwlock, FTAG);
1598 }
1599
1600 static void
1601 dmu_objset_write_done(zio_t *zio, arc_buf_t *abuf, void *arg)
1602 {
1603         (void) abuf;
1604         blkptr_t *bp = zio->io_bp;
1605         blkptr_t *bp_orig = &zio->io_bp_orig;
1606         objset_t *os = arg;
1607
1608         if (zio->io_flags & ZIO_FLAG_IO_REWRITE) {
1609                 ASSERT(BP_EQUAL(bp, bp_orig));
1610         } else {
1611                 dsl_dataset_t *ds = os->os_dsl_dataset;
1612                 dmu_tx_t *tx = os->os_synctx;
1613
1614                 (void) dsl_dataset_block_kill(ds, bp_orig, tx, B_TRUE);
1615                 dsl_dataset_block_born(ds, bp, tx);
1616         }
1617         kmem_free(bp, sizeof (*bp));
1618 }
1619
1620 typedef struct sync_dnodes_arg {
1621         multilist_t *sda_list;
1622         int sda_sublist_idx;
1623         multilist_t *sda_newlist;
1624         dmu_tx_t *sda_tx;
1625 } sync_dnodes_arg_t;
1626
1627 static void
1628 sync_dnodes_task(void *arg)
1629 {
1630         sync_dnodes_arg_t *sda = arg;
1631
1632         multilist_sublist_t *ms =
1633             multilist_sublist_lock(sda->sda_list, sda->sda_sublist_idx);
1634
1635         dmu_objset_sync_dnodes(ms, sda->sda_tx);
1636
1637         multilist_sublist_unlock(ms);
1638
1639         kmem_free(sda, sizeof (*sda));
1640 }
1641
1642
1643 /* called from dsl */
1644 void
1645 dmu_objset_sync(objset_t *os, zio_t *pio, dmu_tx_t *tx)
1646 {
1647         int txgoff;
1648         zbookmark_phys_t zb;
1649         zio_prop_t zp;
1650         zio_t *zio;
1651         list_t *list;
1652         dbuf_dirty_record_t *dr;
1653         int num_sublists;
1654         multilist_t *ml;
1655         blkptr_t *blkptr_copy = kmem_alloc(sizeof (*os->os_rootbp), KM_SLEEP);
1656         *blkptr_copy = *os->os_rootbp;
1657
1658         dprintf_ds(os->os_dsl_dataset, "txg=%llu\n", (u_longlong_t)tx->tx_txg);
1659
1660         ASSERT(dmu_tx_is_syncing(tx));
1661         /* XXX the write_done callback should really give us the tx... */
1662         os->os_synctx = tx;
1663
1664         if (os->os_dsl_dataset == NULL) {
1665                 /*
1666                  * This is the MOS.  If we have upgraded,
1667                  * spa_max_replication() could change, so reset
1668                  * os_copies here.
1669                  */
1670                 os->os_copies = spa_max_replication(os->os_spa);
1671         }
1672
1673         /*
1674          * Create the root block IO
1675          */
1676         SET_BOOKMARK(&zb, os->os_dsl_dataset ?
1677             os->os_dsl_dataset->ds_object : DMU_META_OBJSET,
1678             ZB_ROOT_OBJECT, ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
1679         arc_release(os->os_phys_buf, &os->os_phys_buf);
1680
1681         dmu_write_policy(os, NULL, 0, 0, &zp);
1682
1683         /*
1684          * If we are either claiming the ZIL or doing a raw receive, write
1685          * out the os_phys_buf raw. Neither of these actions will effect the
1686          * MAC at this point.
1687          */
1688         if (os->os_raw_receive ||
1689             os->os_next_write_raw[tx->tx_txg & TXG_MASK]) {
1690                 ASSERT(os->os_encrypted);
1691                 arc_convert_to_raw(os->os_phys_buf,
1692                     os->os_dsl_dataset->ds_object, ZFS_HOST_BYTEORDER,
1693                     DMU_OT_OBJSET, NULL, NULL, NULL);
1694         }
1695
1696         zio = arc_write(pio, os->os_spa, tx->tx_txg,
1697             blkptr_copy, os->os_phys_buf, dmu_os_is_l2cacheable(os),
1698             &zp, dmu_objset_write_ready, NULL, NULL, dmu_objset_write_done,
1699             os, ZIO_PRIORITY_ASYNC_WRITE, ZIO_FLAG_MUSTSUCCEED, &zb);
1700
1701         /*
1702          * Sync special dnodes - the parent IO for the sync is the root block
1703          */
1704         DMU_META_DNODE(os)->dn_zio = zio;
1705         dnode_sync(DMU_META_DNODE(os), tx);
1706
1707         os->os_phys->os_flags = os->os_flags;
1708
1709         if (DMU_USERUSED_DNODE(os) &&
1710             DMU_USERUSED_DNODE(os)->dn_type != DMU_OT_NONE) {
1711                 DMU_USERUSED_DNODE(os)->dn_zio = zio;
1712                 dnode_sync(DMU_USERUSED_DNODE(os), tx);
1713                 DMU_GROUPUSED_DNODE(os)->dn_zio = zio;
1714                 dnode_sync(DMU_GROUPUSED_DNODE(os), tx);
1715         }
1716
1717         if (DMU_PROJECTUSED_DNODE(os) &&
1718             DMU_PROJECTUSED_DNODE(os)->dn_type != DMU_OT_NONE) {
1719                 DMU_PROJECTUSED_DNODE(os)->dn_zio = zio;
1720                 dnode_sync(DMU_PROJECTUSED_DNODE(os), tx);
1721         }
1722
1723         txgoff = tx->tx_txg & TXG_MASK;
1724
1725         /*
1726          * We must create the list here because it uses the
1727          * dn_dirty_link[] of this txg.  But it may already
1728          * exist because we call dsl_dataset_sync() twice per txg.
1729          */
1730         if (os->os_synced_dnodes.ml_sublists == NULL) {
1731                 multilist_create(&os->os_synced_dnodes, sizeof (dnode_t),
1732                     offsetof(dnode_t, dn_dirty_link[txgoff]),
1733                     dnode_multilist_index_func);
1734         } else {
1735                 ASSERT3U(os->os_synced_dnodes.ml_offset, ==,
1736                     offsetof(dnode_t, dn_dirty_link[txgoff]));
1737         }
1738
1739         ml = &os->os_dirty_dnodes[txgoff];
1740         num_sublists = multilist_get_num_sublists(ml);
1741         for (int i = 0; i < num_sublists; i++) {
1742                 if (multilist_sublist_is_empty_idx(ml, i))
1743                         continue;
1744                 sync_dnodes_arg_t *sda = kmem_alloc(sizeof (*sda), KM_SLEEP);
1745                 sda->sda_list = ml;
1746                 sda->sda_sublist_idx = i;
1747                 sda->sda_tx = tx;
1748                 (void) taskq_dispatch(dmu_objset_pool(os)->dp_sync_taskq,
1749                     sync_dnodes_task, sda, 0);
1750                 /* callback frees sda */
1751         }
1752         taskq_wait(dmu_objset_pool(os)->dp_sync_taskq);
1753
1754         list = &DMU_META_DNODE(os)->dn_dirty_records[txgoff];
1755         while ((dr = list_head(list)) != NULL) {
1756                 ASSERT0(dr->dr_dbuf->db_level);
1757                 list_remove(list, dr);
1758                 zio_nowait(dr->dr_zio);
1759         }
1760
1761         /* Enable dnode backfill if enough objects have been freed. */
1762         if (os->os_freed_dnodes >= dmu_rescan_dnode_threshold) {
1763                 os->os_rescan_dnodes = B_TRUE;
1764                 os->os_freed_dnodes = 0;
1765         }
1766
1767         /*
1768          * Free intent log blocks up to this tx.
1769          */
1770         zil_sync(os->os_zil, tx);
1771         os->os_phys->os_zil_header = os->os_zil_header;
1772         zio_nowait(zio);
1773 }
1774
1775 boolean_t
1776 dmu_objset_is_dirty(objset_t *os, uint64_t txg)
1777 {
1778         return (!multilist_is_empty(&os->os_dirty_dnodes[txg & TXG_MASK]));
1779 }
1780
1781 static file_info_cb_t *file_cbs[DMU_OST_NUMTYPES];
1782
1783 void
1784 dmu_objset_register_type(dmu_objset_type_t ost, file_info_cb_t *cb)
1785 {
1786         file_cbs[ost] = cb;
1787 }
1788
1789 int
1790 dmu_get_file_info(objset_t *os, dmu_object_type_t bonustype, const void *data,
1791     zfs_file_info_t *zfi)
1792 {
1793         file_info_cb_t *cb = file_cbs[os->os_phys->os_type];
1794         if (cb == NULL)
1795                 return (EINVAL);
1796         return (cb(bonustype, data, zfi));
1797 }
1798
1799 boolean_t
1800 dmu_objset_userused_enabled(objset_t *os)
1801 {
1802         return (spa_version(os->os_spa) >= SPA_VERSION_USERSPACE &&
1803             file_cbs[os->os_phys->os_type] != NULL &&
1804             DMU_USERUSED_DNODE(os) != NULL);
1805 }
1806
1807 boolean_t
1808 dmu_objset_userobjused_enabled(objset_t *os)
1809 {
1810         return (dmu_objset_userused_enabled(os) &&
1811             spa_feature_is_enabled(os->os_spa, SPA_FEATURE_USEROBJ_ACCOUNTING));
1812 }
1813
1814 boolean_t
1815 dmu_objset_projectquota_enabled(objset_t *os)
1816 {
1817         return (file_cbs[os->os_phys->os_type] != NULL &&
1818             DMU_PROJECTUSED_DNODE(os) != NULL &&
1819             spa_feature_is_enabled(os->os_spa, SPA_FEATURE_PROJECT_QUOTA));
1820 }
1821
1822 typedef struct userquota_node {
1823         /* must be in the first filed, see userquota_update_cache() */
1824         char            uqn_id[20 + DMU_OBJACCT_PREFIX_LEN];
1825         int64_t         uqn_delta;
1826         avl_node_t      uqn_node;
1827 } userquota_node_t;
1828
1829 typedef struct userquota_cache {
1830         avl_tree_t uqc_user_deltas;
1831         avl_tree_t uqc_group_deltas;
1832         avl_tree_t uqc_project_deltas;
1833 } userquota_cache_t;
1834
1835 static int
1836 userquota_compare(const void *l, const void *r)
1837 {
1838         const userquota_node_t *luqn = l;
1839         const userquota_node_t *ruqn = r;
1840         int rv;
1841
1842         /*
1843          * NB: can only access uqn_id because userquota_update_cache() doesn't
1844          * pass in an entire userquota_node_t.
1845          */
1846         rv = strcmp(luqn->uqn_id, ruqn->uqn_id);
1847
1848         return (TREE_ISIGN(rv));
1849 }
1850
1851 static void
1852 do_userquota_cacheflush(objset_t *os, userquota_cache_t *cache, dmu_tx_t *tx)
1853 {
1854         void *cookie;
1855         userquota_node_t *uqn;
1856
1857         ASSERT(dmu_tx_is_syncing(tx));
1858
1859         cookie = NULL;
1860         while ((uqn = avl_destroy_nodes(&cache->uqc_user_deltas,
1861             &cookie)) != NULL) {
1862                 /*
1863                  * os_userused_lock protects against concurrent calls to
1864                  * zap_increment_int().  It's needed because zap_increment_int()
1865                  * is not thread-safe (i.e. not atomic).
1866                  */
1867                 mutex_enter(&os->os_userused_lock);
1868                 VERIFY0(zap_increment(os, DMU_USERUSED_OBJECT,
1869                     uqn->uqn_id, uqn->uqn_delta, tx));
1870                 mutex_exit(&os->os_userused_lock);
1871                 kmem_free(uqn, sizeof (*uqn));
1872         }
1873         avl_destroy(&cache->uqc_user_deltas);
1874
1875         cookie = NULL;
1876         while ((uqn = avl_destroy_nodes(&cache->uqc_group_deltas,
1877             &cookie)) != NULL) {
1878                 mutex_enter(&os->os_userused_lock);
1879                 VERIFY0(zap_increment(os, DMU_GROUPUSED_OBJECT,
1880                     uqn->uqn_id, uqn->uqn_delta, tx));
1881                 mutex_exit(&os->os_userused_lock);
1882                 kmem_free(uqn, sizeof (*uqn));
1883         }
1884         avl_destroy(&cache->uqc_group_deltas);
1885
1886         if (dmu_objset_projectquota_enabled(os)) {
1887                 cookie = NULL;
1888                 while ((uqn = avl_destroy_nodes(&cache->uqc_project_deltas,
1889                     &cookie)) != NULL) {
1890                         mutex_enter(&os->os_userused_lock);
1891                         VERIFY0(zap_increment(os, DMU_PROJECTUSED_OBJECT,
1892                             uqn->uqn_id, uqn->uqn_delta, tx));
1893                         mutex_exit(&os->os_userused_lock);
1894                         kmem_free(uqn, sizeof (*uqn));
1895                 }
1896                 avl_destroy(&cache->uqc_project_deltas);
1897         }
1898 }
1899
1900 static void
1901 userquota_update_cache(avl_tree_t *avl, const char *id, int64_t delta)
1902 {
1903         userquota_node_t *uqn;
1904         avl_index_t idx;
1905
1906         ASSERT(strlen(id) < sizeof (uqn->uqn_id));
1907         /*
1908          * Use id directly for searching because uqn_id is the first field of
1909          * userquota_node_t and fields after uqn_id won't be accessed in
1910          * avl_find().
1911          */
1912         uqn = avl_find(avl, (const void *)id, &idx);
1913         if (uqn == NULL) {
1914                 uqn = kmem_zalloc(sizeof (*uqn), KM_SLEEP);
1915                 strlcpy(uqn->uqn_id, id, sizeof (uqn->uqn_id));
1916                 avl_insert(avl, uqn, idx);
1917         }
1918         uqn->uqn_delta += delta;
1919 }
1920
1921 static void
1922 do_userquota_update(objset_t *os, userquota_cache_t *cache, uint64_t used,
1923     uint64_t flags, uint64_t user, uint64_t group, uint64_t project,
1924     boolean_t subtract)
1925 {
1926         if (flags & DNODE_FLAG_USERUSED_ACCOUNTED) {
1927                 int64_t delta = DNODE_MIN_SIZE + used;
1928                 char name[20];
1929
1930                 if (subtract)
1931                         delta = -delta;
1932
1933                 (void) snprintf(name, sizeof (name), "%llx", (longlong_t)user);
1934                 userquota_update_cache(&cache->uqc_user_deltas, name, delta);
1935
1936                 (void) snprintf(name, sizeof (name), "%llx", (longlong_t)group);
1937                 userquota_update_cache(&cache->uqc_group_deltas, name, delta);
1938
1939                 if (dmu_objset_projectquota_enabled(os)) {
1940                         (void) snprintf(name, sizeof (name), "%llx",
1941                             (longlong_t)project);
1942                         userquota_update_cache(&cache->uqc_project_deltas,
1943                             name, delta);
1944                 }
1945         }
1946 }
1947
1948 static void
1949 do_userobjquota_update(objset_t *os, userquota_cache_t *cache, uint64_t flags,
1950     uint64_t user, uint64_t group, uint64_t project, boolean_t subtract)
1951 {
1952         if (flags & DNODE_FLAG_USEROBJUSED_ACCOUNTED) {
1953                 char name[20 + DMU_OBJACCT_PREFIX_LEN];
1954                 int delta = subtract ? -1 : 1;
1955
1956                 (void) snprintf(name, sizeof (name), DMU_OBJACCT_PREFIX "%llx",
1957                     (longlong_t)user);
1958                 userquota_update_cache(&cache->uqc_user_deltas, name, delta);
1959
1960                 (void) snprintf(name, sizeof (name), DMU_OBJACCT_PREFIX "%llx",
1961                     (longlong_t)group);
1962                 userquota_update_cache(&cache->uqc_group_deltas, name, delta);
1963
1964                 if (dmu_objset_projectquota_enabled(os)) {
1965                         (void) snprintf(name, sizeof (name),
1966                             DMU_OBJACCT_PREFIX "%llx", (longlong_t)project);
1967                         userquota_update_cache(&cache->uqc_project_deltas,
1968                             name, delta);
1969                 }
1970         }
1971 }
1972
1973 typedef struct userquota_updates_arg {
1974         objset_t *uua_os;
1975         int uua_sublist_idx;
1976         dmu_tx_t *uua_tx;
1977 } userquota_updates_arg_t;
1978
1979 static void
1980 userquota_updates_task(void *arg)
1981 {
1982         userquota_updates_arg_t *uua = arg;
1983         objset_t *os = uua->uua_os;
1984         dmu_tx_t *tx = uua->uua_tx;
1985         dnode_t *dn;
1986         userquota_cache_t cache = { { 0 } };
1987
1988         multilist_sublist_t *list =
1989             multilist_sublist_lock(&os->os_synced_dnodes, uua->uua_sublist_idx);
1990
1991         ASSERT(multilist_sublist_head(list) == NULL ||
1992             dmu_objset_userused_enabled(os));
1993         avl_create(&cache.uqc_user_deltas, userquota_compare,
1994             sizeof (userquota_node_t), offsetof(userquota_node_t, uqn_node));
1995         avl_create(&cache.uqc_group_deltas, userquota_compare,
1996             sizeof (userquota_node_t), offsetof(userquota_node_t, uqn_node));
1997         if (dmu_objset_projectquota_enabled(os))
1998                 avl_create(&cache.uqc_project_deltas, userquota_compare,
1999                     sizeof (userquota_node_t), offsetof(userquota_node_t,
2000                     uqn_node));
2001
2002         while ((dn = multilist_sublist_head(list)) != NULL) {
2003                 int flags;
2004                 ASSERT(!DMU_OBJECT_IS_SPECIAL(dn->dn_object));
2005                 ASSERT(dn->dn_phys->dn_type == DMU_OT_NONE ||
2006                     dn->dn_phys->dn_flags &
2007                     DNODE_FLAG_USERUSED_ACCOUNTED);
2008
2009                 flags = dn->dn_id_flags;
2010                 ASSERT(flags);
2011                 if (flags & DN_ID_OLD_EXIST)  {
2012                         do_userquota_update(os, &cache, dn->dn_oldused,
2013                             dn->dn_oldflags, dn->dn_olduid, dn->dn_oldgid,
2014                             dn->dn_oldprojid, B_TRUE);
2015                         do_userobjquota_update(os, &cache, dn->dn_oldflags,
2016                             dn->dn_olduid, dn->dn_oldgid,
2017                             dn->dn_oldprojid, B_TRUE);
2018                 }
2019                 if (flags & DN_ID_NEW_EXIST) {
2020                         do_userquota_update(os, &cache,
2021                             DN_USED_BYTES(dn->dn_phys), dn->dn_phys->dn_flags,
2022                             dn->dn_newuid, dn->dn_newgid,
2023                             dn->dn_newprojid, B_FALSE);
2024                         do_userobjquota_update(os, &cache,
2025                             dn->dn_phys->dn_flags, dn->dn_newuid, dn->dn_newgid,
2026                             dn->dn_newprojid, B_FALSE);
2027                 }
2028
2029                 mutex_enter(&dn->dn_mtx);
2030                 dn->dn_oldused = 0;
2031                 dn->dn_oldflags = 0;
2032                 if (dn->dn_id_flags & DN_ID_NEW_EXIST) {
2033                         dn->dn_olduid = dn->dn_newuid;
2034                         dn->dn_oldgid = dn->dn_newgid;
2035                         dn->dn_oldprojid = dn->dn_newprojid;
2036                         dn->dn_id_flags |= DN_ID_OLD_EXIST;
2037                         if (dn->dn_bonuslen == 0)
2038                                 dn->dn_id_flags |= DN_ID_CHKED_SPILL;
2039                         else
2040                                 dn->dn_id_flags |= DN_ID_CHKED_BONUS;
2041                 }
2042                 dn->dn_id_flags &= ~(DN_ID_NEW_EXIST);
2043                 mutex_exit(&dn->dn_mtx);
2044
2045                 multilist_sublist_remove(list, dn);
2046                 dnode_rele(dn, &os->os_synced_dnodes);
2047         }
2048         do_userquota_cacheflush(os, &cache, tx);
2049         multilist_sublist_unlock(list);
2050         kmem_free(uua, sizeof (*uua));
2051 }
2052
2053 /*
2054  * Release dnode holds from dmu_objset_sync_dnodes().  When the dnode is being
2055  * synced (i.e. we have issued the zio's for blocks in the dnode), it can't be
2056  * evicted because the block containing the dnode can't be evicted until it is
2057  * written out.  However, this hold is necessary to prevent the dnode_t from
2058  * being moved (via dnode_move()) while it's still referenced by
2059  * dbuf_dirty_record_t:dr_dnode.  And dr_dnode is needed for
2060  * dirty_lightweight_leaf-type dirty records.
2061  *
2062  * If we are doing user-object accounting, the dnode_rele() happens from
2063  * userquota_updates_task() instead.
2064  */
2065 static void
2066 dnode_rele_task(void *arg)
2067 {
2068         userquota_updates_arg_t *uua = arg;
2069         objset_t *os = uua->uua_os;
2070
2071         multilist_sublist_t *list =
2072             multilist_sublist_lock(&os->os_synced_dnodes, uua->uua_sublist_idx);
2073
2074         dnode_t *dn;
2075         while ((dn = multilist_sublist_head(list)) != NULL) {
2076                 multilist_sublist_remove(list, dn);
2077                 dnode_rele(dn, &os->os_synced_dnodes);
2078         }
2079         multilist_sublist_unlock(list);
2080         kmem_free(uua, sizeof (*uua));
2081 }
2082
2083 /*
2084  * Return TRUE if userquota updates are needed.
2085  */
2086 static boolean_t
2087 dmu_objset_do_userquota_updates_prep(objset_t *os, dmu_tx_t *tx)
2088 {
2089         if (!dmu_objset_userused_enabled(os))
2090                 return (B_FALSE);
2091
2092         /*
2093          * If this is a raw receive just return and handle accounting
2094          * later when we have the keys loaded. We also don't do user
2095          * accounting during claiming since the datasets are not owned
2096          * for the duration of claiming and this txg should only be
2097          * used for recovery.
2098          */
2099         if (os->os_encrypted && dmu_objset_is_receiving(os))
2100                 return (B_FALSE);
2101
2102         if (tx->tx_txg <= os->os_spa->spa_claim_max_txg)
2103                 return (B_FALSE);
2104
2105         /* Allocate the user/group/project used objects if necessary. */
2106         if (DMU_USERUSED_DNODE(os)->dn_type == DMU_OT_NONE) {
2107                 VERIFY0(zap_create_claim(os,
2108                     DMU_USERUSED_OBJECT,
2109                     DMU_OT_USERGROUP_USED, DMU_OT_NONE, 0, tx));
2110                 VERIFY0(zap_create_claim(os,
2111                     DMU_GROUPUSED_OBJECT,
2112                     DMU_OT_USERGROUP_USED, DMU_OT_NONE, 0, tx));
2113         }
2114
2115         if (dmu_objset_projectquota_enabled(os) &&
2116             DMU_PROJECTUSED_DNODE(os)->dn_type == DMU_OT_NONE) {
2117                 VERIFY0(zap_create_claim(os, DMU_PROJECTUSED_OBJECT,
2118                     DMU_OT_USERGROUP_USED, DMU_OT_NONE, 0, tx));
2119         }
2120         return (B_TRUE);
2121 }
2122
2123 /*
2124  * Dispatch taskq tasks to dp_sync_taskq to update the user accounting, and
2125  * also release the holds on the dnodes from dmu_objset_sync_dnodes().
2126  * The caller must taskq_wait(dp_sync_taskq).
2127  */
2128 void
2129 dmu_objset_sync_done(objset_t *os, dmu_tx_t *tx)
2130 {
2131         boolean_t need_userquota = dmu_objset_do_userquota_updates_prep(os, tx);
2132
2133         int num_sublists = multilist_get_num_sublists(&os->os_synced_dnodes);
2134         for (int i = 0; i < num_sublists; i++) {
2135                 userquota_updates_arg_t *uua =
2136                     kmem_alloc(sizeof (*uua), KM_SLEEP);
2137                 uua->uua_os = os;
2138                 uua->uua_sublist_idx = i;
2139                 uua->uua_tx = tx;
2140
2141                 /*
2142                  * If we don't need to update userquotas, use
2143                  * dnode_rele_task() to call dnode_rele()
2144                  */
2145                 (void) taskq_dispatch(dmu_objset_pool(os)->dp_sync_taskq,
2146                     need_userquota ? userquota_updates_task : dnode_rele_task,
2147                     uua, 0);
2148                 /* callback frees uua */
2149         }
2150 }
2151
2152
2153 /*
2154  * Returns a pointer to data to find uid/gid from
2155  *
2156  * If a dirty record for transaction group that is syncing can't
2157  * be found then NULL is returned.  In the NULL case it is assumed
2158  * the uid/gid aren't changing.
2159  */
2160 static void *
2161 dmu_objset_userquota_find_data(dmu_buf_impl_t *db, dmu_tx_t *tx)
2162 {
2163         dbuf_dirty_record_t *dr;
2164         void *data;
2165
2166         if (db->db_dirtycnt == 0)
2167                 return (db->db.db_data);  /* Nothing is changing */
2168
2169         dr = dbuf_find_dirty_eq(db, tx->tx_txg);
2170
2171         if (dr == NULL) {
2172                 data = NULL;
2173         } else {
2174                 if (dr->dr_dnode->dn_bonuslen == 0 &&
2175                     dr->dr_dbuf->db_blkid == DMU_SPILL_BLKID)
2176                         data = dr->dt.dl.dr_data->b_data;
2177                 else
2178                         data = dr->dt.dl.dr_data;
2179         }
2180
2181         return (data);
2182 }
2183
2184 void
2185 dmu_objset_userquota_get_ids(dnode_t *dn, boolean_t before, dmu_tx_t *tx)
2186 {
2187         objset_t *os = dn->dn_objset;
2188         void *data = NULL;
2189         dmu_buf_impl_t *db = NULL;
2190         int flags = dn->dn_id_flags;
2191         int error;
2192         boolean_t have_spill = B_FALSE;
2193
2194         if (!dmu_objset_userused_enabled(dn->dn_objset))
2195                 return;
2196
2197         /*
2198          * Raw receives introduce a problem with user accounting. Raw
2199          * receives cannot update the user accounting info because the
2200          * user ids and the sizes are encrypted. To guarantee that we
2201          * never end up with bad user accounting, we simply disable it
2202          * during raw receives. We also disable this for normal receives
2203          * so that an incremental raw receive may be done on top of an
2204          * existing non-raw receive.
2205          */
2206         if (os->os_encrypted && dmu_objset_is_receiving(os))
2207                 return;
2208
2209         if (before && (flags & (DN_ID_CHKED_BONUS|DN_ID_OLD_EXIST|
2210             DN_ID_CHKED_SPILL)))
2211                 return;
2212
2213         if (before && dn->dn_bonuslen != 0)
2214                 data = DN_BONUS(dn->dn_phys);
2215         else if (!before && dn->dn_bonuslen != 0) {
2216                 if (dn->dn_bonus) {
2217                         db = dn->dn_bonus;
2218                         mutex_enter(&db->db_mtx);
2219                         data = dmu_objset_userquota_find_data(db, tx);
2220                 } else {
2221                         data = DN_BONUS(dn->dn_phys);
2222                 }
2223         } else if (dn->dn_bonuslen == 0 && dn->dn_bonustype == DMU_OT_SA) {
2224                         int rf = 0;
2225
2226                         if (RW_WRITE_HELD(&dn->dn_struct_rwlock))
2227                                 rf |= DB_RF_HAVESTRUCT;
2228                         error = dmu_spill_hold_by_dnode(dn,
2229                             rf | DB_RF_MUST_SUCCEED,
2230                             FTAG, (dmu_buf_t **)&db);
2231                         ASSERT(error == 0);
2232                         mutex_enter(&db->db_mtx);
2233                         data = (before) ? db->db.db_data :
2234                             dmu_objset_userquota_find_data(db, tx);
2235                         have_spill = B_TRUE;
2236         } else {
2237                 mutex_enter(&dn->dn_mtx);
2238                 dn->dn_id_flags |= DN_ID_CHKED_BONUS;
2239                 mutex_exit(&dn->dn_mtx);
2240                 return;
2241         }
2242
2243         /*
2244          * Must always call the callback in case the object
2245          * type has changed and that type isn't an object type to track
2246          */
2247         zfs_file_info_t zfi;
2248         error = file_cbs[os->os_phys->os_type](dn->dn_bonustype, data, &zfi);
2249
2250         if (before) {
2251                 ASSERT(data);
2252                 dn->dn_olduid = zfi.zfi_user;
2253                 dn->dn_oldgid = zfi.zfi_group;
2254                 dn->dn_oldprojid = zfi.zfi_project;
2255         } else if (data) {
2256                 dn->dn_newuid = zfi.zfi_user;
2257                 dn->dn_newgid = zfi.zfi_group;
2258                 dn->dn_newprojid = zfi.zfi_project;
2259         }
2260
2261         /*
2262          * Preserve existing uid/gid when the callback can't determine
2263          * what the new uid/gid are and the callback returned EEXIST.
2264          * The EEXIST error tells us to just use the existing uid/gid.
2265          * If we don't know what the old values are then just assign
2266          * them to 0, since that is a new file  being created.
2267          */
2268         if (!before && data == NULL && error == EEXIST) {
2269                 if (flags & DN_ID_OLD_EXIST) {
2270                         dn->dn_newuid = dn->dn_olduid;
2271                         dn->dn_newgid = dn->dn_oldgid;
2272                         dn->dn_newprojid = dn->dn_oldprojid;
2273                 } else {
2274                         dn->dn_newuid = 0;
2275                         dn->dn_newgid = 0;
2276                         dn->dn_newprojid = ZFS_DEFAULT_PROJID;
2277                 }
2278                 error = 0;
2279         }
2280
2281         if (db)
2282                 mutex_exit(&db->db_mtx);
2283
2284         mutex_enter(&dn->dn_mtx);
2285         if (error == 0 && before)
2286                 dn->dn_id_flags |= DN_ID_OLD_EXIST;
2287         if (error == 0 && !before)
2288                 dn->dn_id_flags |= DN_ID_NEW_EXIST;
2289
2290         if (have_spill) {
2291                 dn->dn_id_flags |= DN_ID_CHKED_SPILL;
2292         } else {
2293                 dn->dn_id_flags |= DN_ID_CHKED_BONUS;
2294         }
2295         mutex_exit(&dn->dn_mtx);
2296         if (have_spill)
2297                 dmu_buf_rele((dmu_buf_t *)db, FTAG);
2298 }
2299
2300 boolean_t
2301 dmu_objset_userspace_present(objset_t *os)
2302 {
2303         return (os->os_phys->os_flags &
2304             OBJSET_FLAG_USERACCOUNTING_COMPLETE);
2305 }
2306
2307 boolean_t
2308 dmu_objset_userobjspace_present(objset_t *os)
2309 {
2310         return (os->os_phys->os_flags &
2311             OBJSET_FLAG_USEROBJACCOUNTING_COMPLETE);
2312 }
2313
2314 boolean_t
2315 dmu_objset_projectquota_present(objset_t *os)
2316 {
2317         return (os->os_phys->os_flags &
2318             OBJSET_FLAG_PROJECTQUOTA_COMPLETE);
2319 }
2320
2321 static int
2322 dmu_objset_space_upgrade(objset_t *os)
2323 {
2324         uint64_t obj;
2325         int err = 0;
2326
2327         /*
2328          * We simply need to mark every object dirty, so that it will be
2329          * synced out and now accounted.  If this is called
2330          * concurrently, or if we already did some work before crashing,
2331          * that's fine, since we track each object's accounted state
2332          * independently.
2333          */
2334
2335         for (obj = 0; err == 0; err = dmu_object_next(os, &obj, FALSE, 0)) {
2336                 dmu_tx_t *tx;
2337                 dmu_buf_t *db;
2338                 int objerr;
2339
2340                 mutex_enter(&os->os_upgrade_lock);
2341                 if (os->os_upgrade_exit)
2342                         err = SET_ERROR(EINTR);
2343                 mutex_exit(&os->os_upgrade_lock);
2344                 if (err != 0)
2345                         return (err);
2346
2347                 if (issig(JUSTLOOKING) && issig(FORREAL))
2348                         return (SET_ERROR(EINTR));
2349
2350                 objerr = dmu_bonus_hold(os, obj, FTAG, &db);
2351                 if (objerr != 0)
2352                         continue;
2353                 tx = dmu_tx_create(os);
2354                 dmu_tx_hold_bonus(tx, obj);
2355                 objerr = dmu_tx_assign(tx, TXG_WAIT);
2356                 if (objerr != 0) {
2357                         dmu_buf_rele(db, FTAG);
2358                         dmu_tx_abort(tx);
2359                         continue;
2360                 }
2361                 dmu_buf_will_dirty(db, tx);
2362                 dmu_buf_rele(db, FTAG);
2363                 dmu_tx_commit(tx);
2364         }
2365         return (0);
2366 }
2367
2368 static int
2369 dmu_objset_userspace_upgrade_cb(objset_t *os)
2370 {
2371         int err = 0;
2372
2373         if (dmu_objset_userspace_present(os))
2374                 return (0);
2375         if (dmu_objset_is_snapshot(os))
2376                 return (SET_ERROR(EINVAL));
2377         if (!dmu_objset_userused_enabled(os))
2378                 return (SET_ERROR(ENOTSUP));
2379
2380         err = dmu_objset_space_upgrade(os);
2381         if (err)
2382                 return (err);
2383
2384         os->os_flags |= OBJSET_FLAG_USERACCOUNTING_COMPLETE;
2385         txg_wait_synced(dmu_objset_pool(os), 0);
2386         return (0);
2387 }
2388
2389 void
2390 dmu_objset_userspace_upgrade(objset_t *os)
2391 {
2392         dmu_objset_upgrade(os, dmu_objset_userspace_upgrade_cb);
2393 }
2394
2395 static int
2396 dmu_objset_id_quota_upgrade_cb(objset_t *os)
2397 {
2398         int err = 0;
2399
2400         if (dmu_objset_userobjspace_present(os) &&
2401             dmu_objset_projectquota_present(os))
2402                 return (0);
2403         if (dmu_objset_is_snapshot(os))
2404                 return (SET_ERROR(EINVAL));
2405         if (!dmu_objset_userused_enabled(os))
2406                 return (SET_ERROR(ENOTSUP));
2407         if (!dmu_objset_projectquota_enabled(os) &&
2408             dmu_objset_userobjspace_present(os))
2409                 return (SET_ERROR(ENOTSUP));
2410
2411         if (dmu_objset_userobjused_enabled(os))
2412                 dmu_objset_ds(os)->ds_feature_activation[
2413                     SPA_FEATURE_USEROBJ_ACCOUNTING] = (void *)B_TRUE;
2414         if (dmu_objset_projectquota_enabled(os))
2415                 dmu_objset_ds(os)->ds_feature_activation[
2416                     SPA_FEATURE_PROJECT_QUOTA] = (void *)B_TRUE;
2417
2418         err = dmu_objset_space_upgrade(os);
2419         if (err)
2420                 return (err);
2421
2422         os->os_flags |= OBJSET_FLAG_USERACCOUNTING_COMPLETE;
2423         if (dmu_objset_userobjused_enabled(os))
2424                 os->os_flags |= OBJSET_FLAG_USEROBJACCOUNTING_COMPLETE;
2425         if (dmu_objset_projectquota_enabled(os))
2426                 os->os_flags |= OBJSET_FLAG_PROJECTQUOTA_COMPLETE;
2427
2428         txg_wait_synced(dmu_objset_pool(os), 0);
2429         return (0);
2430 }
2431
2432 void
2433 dmu_objset_id_quota_upgrade(objset_t *os)
2434 {
2435         dmu_objset_upgrade(os, dmu_objset_id_quota_upgrade_cb);
2436 }
2437
2438 boolean_t
2439 dmu_objset_userobjspace_upgradable(objset_t *os)
2440 {
2441         return (dmu_objset_type(os) == DMU_OST_ZFS &&
2442             !dmu_objset_is_snapshot(os) &&
2443             dmu_objset_userobjused_enabled(os) &&
2444             !dmu_objset_userobjspace_present(os) &&
2445             spa_writeable(dmu_objset_spa(os)));
2446 }
2447
2448 boolean_t
2449 dmu_objset_projectquota_upgradable(objset_t *os)
2450 {
2451         return (dmu_objset_type(os) == DMU_OST_ZFS &&
2452             !dmu_objset_is_snapshot(os) &&
2453             dmu_objset_projectquota_enabled(os) &&
2454             !dmu_objset_projectquota_present(os) &&
2455             spa_writeable(dmu_objset_spa(os)));
2456 }
2457
2458 void
2459 dmu_objset_space(objset_t *os, uint64_t *refdbytesp, uint64_t *availbytesp,
2460     uint64_t *usedobjsp, uint64_t *availobjsp)
2461 {
2462         dsl_dataset_space(os->os_dsl_dataset, refdbytesp, availbytesp,
2463             usedobjsp, availobjsp);
2464 }
2465
2466 uint64_t
2467 dmu_objset_fsid_guid(objset_t *os)
2468 {
2469         return (dsl_dataset_fsid_guid(os->os_dsl_dataset));
2470 }
2471
2472 void
2473 dmu_objset_fast_stat(objset_t *os, dmu_objset_stats_t *stat)
2474 {
2475         stat->dds_type = os->os_phys->os_type;
2476         if (os->os_dsl_dataset)
2477                 dsl_dataset_fast_stat(os->os_dsl_dataset, stat);
2478 }
2479
2480 void
2481 dmu_objset_stats(objset_t *os, nvlist_t *nv)
2482 {
2483         ASSERT(os->os_dsl_dataset ||
2484             os->os_phys->os_type == DMU_OST_META);
2485
2486         if (os->os_dsl_dataset != NULL)
2487                 dsl_dataset_stats(os->os_dsl_dataset, nv);
2488
2489         dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_TYPE,
2490             os->os_phys->os_type);
2491         dsl_prop_nvlist_add_uint64(nv, ZFS_PROP_USERACCOUNTING,
2492             dmu_objset_userspace_present(os));
2493 }
2494
2495 int
2496 dmu_objset_is_snapshot(objset_t *os)
2497 {
2498         if (os->os_dsl_dataset != NULL)
2499                 return (os->os_dsl_dataset->ds_is_snapshot);
2500         else
2501                 return (B_FALSE);
2502 }
2503
2504 int
2505 dmu_snapshot_realname(objset_t *os, const char *name, char *real, int maxlen,
2506     boolean_t *conflict)
2507 {
2508         dsl_dataset_t *ds = os->os_dsl_dataset;
2509         uint64_t ignored;
2510
2511         if (dsl_dataset_phys(ds)->ds_snapnames_zapobj == 0)
2512                 return (SET_ERROR(ENOENT));
2513
2514         return (zap_lookup_norm(ds->ds_dir->dd_pool->dp_meta_objset,
2515             dsl_dataset_phys(ds)->ds_snapnames_zapobj, name, 8, 1, &ignored,
2516             MT_NORMALIZE, real, maxlen, conflict));
2517 }
2518
2519 int
2520 dmu_snapshot_list_next(objset_t *os, int namelen, char *name,
2521     uint64_t *idp, uint64_t *offp, boolean_t *case_conflict)
2522 {
2523         dsl_dataset_t *ds = os->os_dsl_dataset;
2524         zap_cursor_t cursor;
2525         zap_attribute_t attr;
2526
2527         ASSERT(dsl_pool_config_held(dmu_objset_pool(os)));
2528
2529         if (dsl_dataset_phys(ds)->ds_snapnames_zapobj == 0)
2530                 return (SET_ERROR(ENOENT));
2531
2532         zap_cursor_init_serialized(&cursor,
2533             ds->ds_dir->dd_pool->dp_meta_objset,
2534             dsl_dataset_phys(ds)->ds_snapnames_zapobj, *offp);
2535
2536         if (zap_cursor_retrieve(&cursor, &attr) != 0) {
2537                 zap_cursor_fini(&cursor);
2538                 return (SET_ERROR(ENOENT));
2539         }
2540
2541         if (strlen(attr.za_name) + 1 > namelen) {
2542                 zap_cursor_fini(&cursor);
2543                 return (SET_ERROR(ENAMETOOLONG));
2544         }
2545
2546         (void) strlcpy(name, attr.za_name, namelen);
2547         if (idp)
2548                 *idp = attr.za_first_integer;
2549         if (case_conflict)
2550                 *case_conflict = attr.za_normalization_conflict;
2551         zap_cursor_advance(&cursor);
2552         *offp = zap_cursor_serialize(&cursor);
2553         zap_cursor_fini(&cursor);
2554
2555         return (0);
2556 }
2557
2558 int
2559 dmu_snapshot_lookup(objset_t *os, const char *name, uint64_t *value)
2560 {
2561         return (dsl_dataset_snap_lookup(os->os_dsl_dataset, name, value));
2562 }
2563
2564 int
2565 dmu_dir_list_next(objset_t *os, int namelen, char *name,
2566     uint64_t *idp, uint64_t *offp)
2567 {
2568         dsl_dir_t *dd = os->os_dsl_dataset->ds_dir;
2569         zap_cursor_t cursor;
2570         zap_attribute_t attr;
2571
2572         /* there is no next dir on a snapshot! */
2573         if (os->os_dsl_dataset->ds_object !=
2574             dsl_dir_phys(dd)->dd_head_dataset_obj)
2575                 return (SET_ERROR(ENOENT));
2576
2577         zap_cursor_init_serialized(&cursor,
2578             dd->dd_pool->dp_meta_objset,
2579             dsl_dir_phys(dd)->dd_child_dir_zapobj, *offp);
2580
2581         if (zap_cursor_retrieve(&cursor, &attr) != 0) {
2582                 zap_cursor_fini(&cursor);
2583                 return (SET_ERROR(ENOENT));
2584         }
2585
2586         if (strlen(attr.za_name) + 1 > namelen) {
2587                 zap_cursor_fini(&cursor);
2588                 return (SET_ERROR(ENAMETOOLONG));
2589         }
2590
2591         (void) strlcpy(name, attr.za_name, namelen);
2592         if (idp)
2593                 *idp = attr.za_first_integer;
2594         zap_cursor_advance(&cursor);
2595         *offp = zap_cursor_serialize(&cursor);
2596         zap_cursor_fini(&cursor);
2597
2598         return (0);
2599 }
2600
2601 typedef struct dmu_objset_find_ctx {
2602         taskq_t         *dc_tq;
2603         dsl_pool_t      *dc_dp;
2604         uint64_t        dc_ddobj;
2605         char            *dc_ddname; /* last component of ddobj's name */
2606         int             (*dc_func)(dsl_pool_t *, dsl_dataset_t *, void *);
2607         void            *dc_arg;
2608         int             dc_flags;
2609         kmutex_t        *dc_error_lock;
2610         int             *dc_error;
2611 } dmu_objset_find_ctx_t;
2612
2613 static void
2614 dmu_objset_find_dp_impl(dmu_objset_find_ctx_t *dcp)
2615 {
2616         dsl_pool_t *dp = dcp->dc_dp;
2617         dsl_dir_t *dd;
2618         dsl_dataset_t *ds;
2619         zap_cursor_t zc;
2620         zap_attribute_t *attr;
2621         uint64_t thisobj;
2622         int err = 0;
2623
2624         /* don't process if there already was an error */
2625         if (*dcp->dc_error != 0)
2626                 goto out;
2627
2628         /*
2629          * Note: passing the name (dc_ddname) here is optional, but it
2630          * improves performance because we don't need to call
2631          * zap_value_search() to determine the name.
2632          */
2633         err = dsl_dir_hold_obj(dp, dcp->dc_ddobj, dcp->dc_ddname, FTAG, &dd);
2634         if (err != 0)
2635                 goto out;
2636
2637         /* Don't visit hidden ($MOS & $ORIGIN) objsets. */
2638         if (dd->dd_myname[0] == '$') {
2639                 dsl_dir_rele(dd, FTAG);
2640                 goto out;
2641         }
2642
2643         thisobj = dsl_dir_phys(dd)->dd_head_dataset_obj;
2644         attr = kmem_alloc(sizeof (zap_attribute_t), KM_SLEEP);
2645
2646         /*
2647          * Iterate over all children.
2648          */
2649         if (dcp->dc_flags & DS_FIND_CHILDREN) {
2650                 for (zap_cursor_init(&zc, dp->dp_meta_objset,
2651                     dsl_dir_phys(dd)->dd_child_dir_zapobj);
2652                     zap_cursor_retrieve(&zc, attr) == 0;
2653                     (void) zap_cursor_advance(&zc)) {
2654                         ASSERT3U(attr->za_integer_length, ==,
2655                             sizeof (uint64_t));
2656                         ASSERT3U(attr->za_num_integers, ==, 1);
2657
2658                         dmu_objset_find_ctx_t *child_dcp =
2659                             kmem_alloc(sizeof (*child_dcp), KM_SLEEP);
2660                         *child_dcp = *dcp;
2661                         child_dcp->dc_ddobj = attr->za_first_integer;
2662                         child_dcp->dc_ddname = spa_strdup(attr->za_name);
2663                         if (dcp->dc_tq != NULL)
2664                                 (void) taskq_dispatch(dcp->dc_tq,
2665                                     dmu_objset_find_dp_cb, child_dcp, TQ_SLEEP);
2666                         else
2667                                 dmu_objset_find_dp_impl(child_dcp);
2668                 }
2669                 zap_cursor_fini(&zc);
2670         }
2671
2672         /*
2673          * Iterate over all snapshots.
2674          */
2675         if (dcp->dc_flags & DS_FIND_SNAPSHOTS) {
2676                 dsl_dataset_t *ds;
2677                 err = dsl_dataset_hold_obj(dp, thisobj, FTAG, &ds);
2678
2679                 if (err == 0) {
2680                         uint64_t snapobj;
2681
2682                         snapobj = dsl_dataset_phys(ds)->ds_snapnames_zapobj;
2683                         dsl_dataset_rele(ds, FTAG);
2684
2685                         for (zap_cursor_init(&zc, dp->dp_meta_objset, snapobj);
2686                             zap_cursor_retrieve(&zc, attr) == 0;
2687                             (void) zap_cursor_advance(&zc)) {
2688                                 ASSERT3U(attr->za_integer_length, ==,
2689                                     sizeof (uint64_t));
2690                                 ASSERT3U(attr->za_num_integers, ==, 1);
2691
2692                                 err = dsl_dataset_hold_obj(dp,
2693                                     attr->za_first_integer, FTAG, &ds);
2694                                 if (err != 0)
2695                                         break;
2696                                 err = dcp->dc_func(dp, ds, dcp->dc_arg);
2697                                 dsl_dataset_rele(ds, FTAG);
2698                                 if (err != 0)
2699                                         break;
2700                         }
2701                         zap_cursor_fini(&zc);
2702                 }
2703         }
2704
2705         kmem_free(attr, sizeof (zap_attribute_t));
2706
2707         if (err != 0) {
2708                 dsl_dir_rele(dd, FTAG);
2709                 goto out;
2710         }
2711
2712         /*
2713          * Apply to self.
2714          */
2715         err = dsl_dataset_hold_obj(dp, thisobj, FTAG, &ds);
2716
2717         /*
2718          * Note: we hold the dir while calling dsl_dataset_hold_obj() so
2719          * that the dir will remain cached, and we won't have to re-instantiate
2720          * it (which could be expensive due to finding its name via
2721          * zap_value_search()).
2722          */
2723         dsl_dir_rele(dd, FTAG);
2724         if (err != 0)
2725                 goto out;
2726         err = dcp->dc_func(dp, ds, dcp->dc_arg);
2727         dsl_dataset_rele(ds, FTAG);
2728
2729 out:
2730         if (err != 0) {
2731                 mutex_enter(dcp->dc_error_lock);
2732                 /* only keep first error */
2733                 if (*dcp->dc_error == 0)
2734                         *dcp->dc_error = err;
2735                 mutex_exit(dcp->dc_error_lock);
2736         }
2737
2738         if (dcp->dc_ddname != NULL)
2739                 spa_strfree(dcp->dc_ddname);
2740         kmem_free(dcp, sizeof (*dcp));
2741 }
2742
2743 static void
2744 dmu_objset_find_dp_cb(void *arg)
2745 {
2746         dmu_objset_find_ctx_t *dcp = arg;
2747         dsl_pool_t *dp = dcp->dc_dp;
2748
2749         /*
2750          * We need to get a pool_config_lock here, as there are several
2751          * assert(pool_config_held) down the stack. Getting a lock via
2752          * dsl_pool_config_enter is risky, as it might be stalled by a
2753          * pending writer. This would deadlock, as the write lock can
2754          * only be granted when our parent thread gives up the lock.
2755          * The _prio interface gives us priority over a pending writer.
2756          */
2757         dsl_pool_config_enter_prio(dp, FTAG);
2758
2759         dmu_objset_find_dp_impl(dcp);
2760
2761         dsl_pool_config_exit(dp, FTAG);
2762 }
2763
2764 /*
2765  * Find objsets under and including ddobj, call func(ds) on each.
2766  * The order for the enumeration is completely undefined.
2767  * func is called with dsl_pool_config held.
2768  */
2769 int
2770 dmu_objset_find_dp(dsl_pool_t *dp, uint64_t ddobj,
2771     int func(dsl_pool_t *, dsl_dataset_t *, void *), void *arg, int flags)
2772 {
2773         int error = 0;
2774         taskq_t *tq = NULL;
2775         int ntasks;
2776         dmu_objset_find_ctx_t *dcp;
2777         kmutex_t err_lock;
2778
2779         mutex_init(&err_lock, NULL, MUTEX_DEFAULT, NULL);
2780         dcp = kmem_alloc(sizeof (*dcp), KM_SLEEP);
2781         dcp->dc_tq = NULL;
2782         dcp->dc_dp = dp;
2783         dcp->dc_ddobj = ddobj;
2784         dcp->dc_ddname = NULL;
2785         dcp->dc_func = func;
2786         dcp->dc_arg = arg;
2787         dcp->dc_flags = flags;
2788         dcp->dc_error_lock = &err_lock;
2789         dcp->dc_error = &error;
2790
2791         if ((flags & DS_FIND_SERIALIZE) || dsl_pool_config_held_writer(dp)) {
2792                 /*
2793                  * In case a write lock is held we can't make use of
2794                  * parallelism, as down the stack of the worker threads
2795                  * the lock is asserted via dsl_pool_config_held.
2796                  * In case of a read lock this is solved by getting a read
2797                  * lock in each worker thread, which isn't possible in case
2798                  * of a writer lock. So we fall back to the synchronous path
2799                  * here.
2800                  * In the future it might be possible to get some magic into
2801                  * dsl_pool_config_held in a way that it returns true for
2802                  * the worker threads so that a single lock held from this
2803                  * thread suffices. For now, stay single threaded.
2804                  */
2805                 dmu_objset_find_dp_impl(dcp);
2806                 mutex_destroy(&err_lock);
2807
2808                 return (error);
2809         }
2810
2811         ntasks = dmu_find_threads;
2812         if (ntasks == 0)
2813                 ntasks = vdev_count_leaves(dp->dp_spa) * 4;
2814         tq = taskq_create("dmu_objset_find", ntasks, maxclsyspri, ntasks,
2815             INT_MAX, 0);
2816         if (tq == NULL) {
2817                 kmem_free(dcp, sizeof (*dcp));
2818                 mutex_destroy(&err_lock);
2819
2820                 return (SET_ERROR(ENOMEM));
2821         }
2822         dcp->dc_tq = tq;
2823
2824         /* dcp will be freed by task */
2825         (void) taskq_dispatch(tq, dmu_objset_find_dp_cb, dcp, TQ_SLEEP);
2826
2827         /*
2828          * PORTING: this code relies on the property of taskq_wait to wait
2829          * until no more tasks are queued and no more tasks are active. As
2830          * we always queue new tasks from within other tasks, task_wait
2831          * reliably waits for the full recursion to finish, even though we
2832          * enqueue new tasks after taskq_wait has been called.
2833          * On platforms other than illumos, taskq_wait may not have this
2834          * property.
2835          */
2836         taskq_wait(tq);
2837         taskq_destroy(tq);
2838         mutex_destroy(&err_lock);
2839
2840         return (error);
2841 }
2842
2843 /*
2844  * Find all objsets under name, and for each, call 'func(child_name, arg)'.
2845  * The dp_config_rwlock must not be held when this is called, and it
2846  * will not be held when the callback is called.
2847  * Therefore this function should only be used when the pool is not changing
2848  * (e.g. in syncing context), or the callback can deal with the possible races.
2849  */
2850 static int
2851 dmu_objset_find_impl(spa_t *spa, const char *name,
2852     int func(const char *, void *), void *arg, int flags)
2853 {
2854         dsl_dir_t *dd;
2855         dsl_pool_t *dp = spa_get_dsl(spa);
2856         dsl_dataset_t *ds;
2857         zap_cursor_t zc;
2858         zap_attribute_t *attr;
2859         char *child;
2860         uint64_t thisobj;
2861         int err;
2862
2863         dsl_pool_config_enter(dp, FTAG);
2864
2865         err = dsl_dir_hold(dp, name, FTAG, &dd, NULL);
2866         if (err != 0) {
2867                 dsl_pool_config_exit(dp, FTAG);
2868                 return (err);
2869         }
2870
2871         /* Don't visit hidden ($MOS & $ORIGIN) objsets. */
2872         if (dd->dd_myname[0] == '$') {
2873                 dsl_dir_rele(dd, FTAG);
2874                 dsl_pool_config_exit(dp, FTAG);
2875                 return (0);
2876         }
2877
2878         thisobj = dsl_dir_phys(dd)->dd_head_dataset_obj;
2879         attr = kmem_alloc(sizeof (zap_attribute_t), KM_SLEEP);
2880
2881         /*
2882          * Iterate over all children.
2883          */
2884         if (flags & DS_FIND_CHILDREN) {
2885                 for (zap_cursor_init(&zc, dp->dp_meta_objset,
2886                     dsl_dir_phys(dd)->dd_child_dir_zapobj);
2887                     zap_cursor_retrieve(&zc, attr) == 0;
2888                     (void) zap_cursor_advance(&zc)) {
2889                         ASSERT3U(attr->za_integer_length, ==,
2890                             sizeof (uint64_t));
2891                         ASSERT3U(attr->za_num_integers, ==, 1);
2892
2893                         child = kmem_asprintf("%s/%s", name, attr->za_name);
2894                         dsl_pool_config_exit(dp, FTAG);
2895                         err = dmu_objset_find_impl(spa, child,
2896                             func, arg, flags);
2897                         dsl_pool_config_enter(dp, FTAG);
2898                         kmem_strfree(child);
2899                         if (err != 0)
2900                                 break;
2901                 }
2902                 zap_cursor_fini(&zc);
2903
2904                 if (err != 0) {
2905                         dsl_dir_rele(dd, FTAG);
2906                         dsl_pool_config_exit(dp, FTAG);
2907                         kmem_free(attr, sizeof (zap_attribute_t));
2908                         return (err);
2909                 }
2910         }
2911
2912         /*
2913          * Iterate over all snapshots.
2914          */
2915         if (flags & DS_FIND_SNAPSHOTS) {
2916                 err = dsl_dataset_hold_obj(dp, thisobj, FTAG, &ds);
2917
2918                 if (err == 0) {
2919                         uint64_t snapobj;
2920
2921                         snapobj = dsl_dataset_phys(ds)->ds_snapnames_zapobj;
2922                         dsl_dataset_rele(ds, FTAG);
2923
2924                         for (zap_cursor_init(&zc, dp->dp_meta_objset, snapobj);
2925                             zap_cursor_retrieve(&zc, attr) == 0;
2926                             (void) zap_cursor_advance(&zc)) {
2927                                 ASSERT3U(attr->za_integer_length, ==,
2928                                     sizeof (uint64_t));
2929                                 ASSERT3U(attr->za_num_integers, ==, 1);
2930
2931                                 child = kmem_asprintf("%s@%s",
2932                                     name, attr->za_name);
2933                                 dsl_pool_config_exit(dp, FTAG);
2934                                 err = func(child, arg);
2935                                 dsl_pool_config_enter(dp, FTAG);
2936                                 kmem_strfree(child);
2937                                 if (err != 0)
2938                                         break;
2939                         }
2940                         zap_cursor_fini(&zc);
2941                 }
2942         }
2943
2944         dsl_dir_rele(dd, FTAG);
2945         kmem_free(attr, sizeof (zap_attribute_t));
2946         dsl_pool_config_exit(dp, FTAG);
2947
2948         if (err != 0)
2949                 return (err);
2950
2951         /* Apply to self. */
2952         return (func(name, arg));
2953 }
2954
2955 /*
2956  * See comment above dmu_objset_find_impl().
2957  */
2958 int
2959 dmu_objset_find(const char *name, int func(const char *, void *), void *arg,
2960     int flags)
2961 {
2962         spa_t *spa;
2963         int error;
2964
2965         error = spa_open(name, &spa, FTAG);
2966         if (error != 0)
2967                 return (error);
2968         error = dmu_objset_find_impl(spa, name, func, arg, flags);
2969         spa_close(spa, FTAG);
2970         return (error);
2971 }
2972
2973 boolean_t
2974 dmu_objset_incompatible_encryption_version(objset_t *os)
2975 {
2976         return (dsl_dir_incompatible_encryption_version(
2977             os->os_dsl_dataset->ds_dir));
2978 }
2979
2980 void
2981 dmu_objset_set_user(objset_t *os, void *user_ptr)
2982 {
2983         ASSERT(MUTEX_HELD(&os->os_user_ptr_lock));
2984         os->os_user_ptr = user_ptr;
2985 }
2986
2987 void *
2988 dmu_objset_get_user(objset_t *os)
2989 {
2990         ASSERT(MUTEX_HELD(&os->os_user_ptr_lock));
2991         return (os->os_user_ptr);
2992 }
2993
2994 /*
2995  * Determine name of filesystem, given name of snapshot.
2996  * buf must be at least ZFS_MAX_DATASET_NAME_LEN bytes
2997  */
2998 int
2999 dmu_fsname(const char *snapname, char *buf)
3000 {
3001         char *atp = strchr(snapname, '@');
3002         if (atp == NULL)
3003                 return (SET_ERROR(EINVAL));
3004         if (atp - snapname >= ZFS_MAX_DATASET_NAME_LEN)
3005                 return (SET_ERROR(ENAMETOOLONG));
3006         (void) strlcpy(buf, snapname, atp - snapname + 1);
3007         return (0);
3008 }
3009
3010 /*
3011  * Call when we think we're going to write/free space in open context
3012  * to track the amount of dirty data in the open txg, which is also the
3013  * amount of memory that can not be evicted until this txg syncs.
3014  *
3015  * Note that there are two conditions where this can be called from
3016  * syncing context:
3017  *
3018  * [1] When we just created the dataset, in which case we go on with
3019  *     updating any accounting of dirty data as usual.
3020  * [2] When we are dirtying MOS data, in which case we only update the
3021  *     pool's accounting of dirty data.
3022  */
3023 void
3024 dmu_objset_willuse_space(objset_t *os, int64_t space, dmu_tx_t *tx)
3025 {
3026         dsl_dataset_t *ds = os->os_dsl_dataset;
3027         int64_t aspace = spa_get_worst_case_asize(os->os_spa, space);
3028
3029         if (ds != NULL) {
3030                 dsl_dir_willuse_space(ds->ds_dir, aspace, tx);
3031         }
3032
3033         dsl_pool_dirty_space(dmu_tx_pool(tx), space, tx);
3034 }
3035
3036 #if defined(_KERNEL)
3037 EXPORT_SYMBOL(dmu_objset_zil);
3038 EXPORT_SYMBOL(dmu_objset_pool);
3039 EXPORT_SYMBOL(dmu_objset_ds);
3040 EXPORT_SYMBOL(dmu_objset_type);
3041 EXPORT_SYMBOL(dmu_objset_name);
3042 EXPORT_SYMBOL(dmu_objset_hold);
3043 EXPORT_SYMBOL(dmu_objset_hold_flags);
3044 EXPORT_SYMBOL(dmu_objset_own);
3045 EXPORT_SYMBOL(dmu_objset_rele);
3046 EXPORT_SYMBOL(dmu_objset_rele_flags);
3047 EXPORT_SYMBOL(dmu_objset_disown);
3048 EXPORT_SYMBOL(dmu_objset_from_ds);
3049 EXPORT_SYMBOL(dmu_objset_create);
3050 EXPORT_SYMBOL(dmu_objset_clone);
3051 EXPORT_SYMBOL(dmu_objset_stats);
3052 EXPORT_SYMBOL(dmu_objset_fast_stat);
3053 EXPORT_SYMBOL(dmu_objset_spa);
3054 EXPORT_SYMBOL(dmu_objset_space);
3055 EXPORT_SYMBOL(dmu_objset_fsid_guid);
3056 EXPORT_SYMBOL(dmu_objset_find);
3057 EXPORT_SYMBOL(dmu_objset_byteswap);
3058 EXPORT_SYMBOL(dmu_objset_evict_dbufs);
3059 EXPORT_SYMBOL(dmu_objset_snap_cmtime);
3060 EXPORT_SYMBOL(dmu_objset_dnodesize);
3061
3062 EXPORT_SYMBOL(dmu_objset_sync);
3063 EXPORT_SYMBOL(dmu_objset_is_dirty);
3064 EXPORT_SYMBOL(dmu_objset_create_impl_dnstats);
3065 EXPORT_SYMBOL(dmu_objset_create_impl);
3066 EXPORT_SYMBOL(dmu_objset_open_impl);
3067 EXPORT_SYMBOL(dmu_objset_evict);
3068 EXPORT_SYMBOL(dmu_objset_register_type);
3069 EXPORT_SYMBOL(dmu_objset_sync_done);
3070 EXPORT_SYMBOL(dmu_objset_userquota_get_ids);
3071 EXPORT_SYMBOL(dmu_objset_userused_enabled);
3072 EXPORT_SYMBOL(dmu_objset_userspace_upgrade);
3073 EXPORT_SYMBOL(dmu_objset_userspace_present);
3074 EXPORT_SYMBOL(dmu_objset_userobjused_enabled);
3075 EXPORT_SYMBOL(dmu_objset_userobjspace_upgradable);
3076 EXPORT_SYMBOL(dmu_objset_userobjspace_present);
3077 EXPORT_SYMBOL(dmu_objset_projectquota_enabled);
3078 EXPORT_SYMBOL(dmu_objset_projectquota_present);
3079 EXPORT_SYMBOL(dmu_objset_projectquota_upgradable);
3080 EXPORT_SYMBOL(dmu_objset_id_quota_upgrade);
3081 #endif