OS-1566 filesystem limits for ZFS datasets

*** 641,650 ****
--- 641,663 ----
                          return (EINVAL);
                  if (rbsa->origin->ds_phys->ds_guid != rbsa->fromguid)
                          return (ENODEV);
          }
  
+         /*
+          * Check filesystem and snapshot limits before receiving. We'll recheck
+          * again at the end, but might as well abort before receiving if we're
+          * already over the limit.
+          */
+         err = dsl_dir_fscount_check(dd, 1, NULL, rbsa->cr);
+         if (err != 0)
+                 return (err);
+ 
+         err = dsl_snapcount_check(dd, 1, NULL, rbsa->cr);
+         if (err != 0)
+                 return (err);
+ 
          return (0);
  }
  
  static void
  recv_new_sync(void *arg1, void *arg2, dmu_tx_t *tx)
*** 666,675 ****
--- 679,692 ----
          }
  
          spa_history_log_internal_ds(rbsa->ds, "receive new", tx, "");
  }
  
+ /*
+  * Note that we do not check the file system limit with dsl_dir_fscount_check
+  * because the temporary %clones don't count against that limit.
+  */
  /* ARGSUSED */
  static int
  recv_existing_check(void *arg1, void *arg2, dmu_tx_t *tx)
  {
          dsl_dataset_t *ds = arg1;
*** 723,732 ****
--- 740,754 ----
                  }
          } else {
                  /* if full, most recent snapshot must be $ORIGIN */
                  if (ds->ds_phys->ds_prev_snap_txg >= TXG_INITIAL)
                          return (ENODEV);
+ 
+                 /* Check snapshot limit before receiving */
+                 err = dsl_snapcount_check(ds->ds_dir, 1, NULL, rbsa->cr);
+                 if (err != 0)
+                         return (err);
          }
  
          /* temporary clone name must not exist */
          err = zap_lookup(ds->ds_dir->dd_pool->dp_meta_objset,
              ds->ds_dir->dd_phys->dd_child_dir_zapobj,
*** 1545,1571 ****
  
  struct recvendsyncarg {
          char *tosnap;
          uint64_t creation_time;
          uint64_t toguid;
  };
  
  static int
  recv_end_check(void *arg1, void *arg2, dmu_tx_t *tx)
  {
          dsl_dataset_t *ds = arg1;
          struct recvendsyncarg *resa = arg2;
  
!         return (dsl_dataset_snapshot_check(ds, resa->tosnap, tx));
  }
  
  static void
  recv_end_sync(void *arg1, void *arg2, dmu_tx_t *tx)
  {
          dsl_dataset_t *ds = arg1;
          struct recvendsyncarg *resa = arg2;
  
          dsl_dataset_snapshot_sync(ds, resa->tosnap, tx);
  
          /* set snapshot's creation time and guid */
          dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
          ds->ds_prev->ds_phys->ds_creation_time = resa->creation_time;
--- 1567,1609 ----
  
  struct recvendsyncarg {
          char *tosnap;
          uint64_t creation_time;
          uint64_t toguid;
+         boolean_t is_new;
+         cred_t *cr;
  };
  
  static int
  recv_end_check(void *arg1, void *arg2, dmu_tx_t *tx)
  {
          dsl_dataset_t *ds = arg1;
          struct recvendsyncarg *resa = arg2;
  
!         if (resa->is_new) {
!                 /* re-check the filesystem limit now that recv is complete */
!                 int err;
! 
!                 err = dsl_dir_fscount_check(ds->ds_dir, 1, NULL, resa->cr);
!                 if (err != 0)
!                         return (err);
!         }
! 
!         return (dsl_dataset_snapshot_check(ds, resa->tosnap, 1, tx, resa->cr));
  }
  
  static void
  recv_end_sync(void *arg1, void *arg2, dmu_tx_t *tx)
  {
          dsl_dataset_t *ds = arg1;
          struct recvendsyncarg *resa = arg2;
  
+         if (resa->is_new) {
+                 /* update the filesystem counts */
+                 dsl_dir_fscount_adjust(ds->ds_dir->dd_parent, tx, 1, B_TRUE);
+         }
+ 
          dsl_dataset_snapshot_sync(ds, resa->tosnap, tx);
  
          /* set snapshot's creation time and guid */
          dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
          ds->ds_prev->ds_phys->ds_creation_time = resa->creation_time;
*** 1622,1631 ****
--- 1660,1671 ----
          }
  
          resa.creation_time = drc->drc_drrb->drr_creation_time;
          resa.toguid = drc->drc_drrb->drr_toguid;
          resa.tosnap = drc->drc_tosnap;
+         resa.is_new = B_FALSE;
+         resa.cr = CRED();
  
          err = dsl_sync_task_do(ds->ds_dir->dd_pool,
              recv_end_check, recv_end_sync, ds, &resa, 3);
          if (err) {
                  /* swap back */
*** 1657,1666 ****
--- 1697,1708 ----
          txg_wait_synced(ds->ds_dir->dd_pool, 0);
  
          resa.creation_time = drc->drc_drrb->drr_creation_time;
          resa.toguid = drc->drc_drrb->drr_toguid;
          resa.tosnap = drc->drc_tosnap;
+         resa.is_new = B_TRUE;
+         resa.cr = CRED();
  
          err = dsl_sync_task_do(ds->ds_dir->dd_pool,
              recv_end_check, recv_end_sync, ds, &resa, 3);
          if (err) {
                  /* clean up the fs we just recv'd into */