OS-1566 filesystem limits for ZFS datasets

@@ -641,10 +641,25 @@
                         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.
+         */
+        if (dd->dd_parent != NULL) {
+                err = dsl_dir_fscount_check(dd->dd_parent, 1, NULL);
+                if (err != 0)
+                        return (err);
+        }
+
+        err = dsl_snapcount_check(dd, 1, NULL);
+        if (err != 0)
+                return (err);
+
         return (0);
 }
 
 static void
 recv_new_sync(void *arg1, void *arg2, dmu_tx_t *tx)

@@ -723,10 +738,15 @@
                 }
         } 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);
+                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,27 +1565,46 @@
 
 struct recvendsyncarg {
         char *tosnap;
         uint64_t creation_time;
         uint64_t toguid;
+        boolean_t is_new;
 };
 
 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));
+        if (resa->is_new) {
+                /* re-check the filesystem limit now that recv is complete */
+                dsl_dir_t *dd;
+                int err;
+
+                dd = ds->ds_dir;
+                if (dd->dd_parent != NULL) {
+                        err = dsl_dir_fscount_check(dd->dd_parent, 1, NULL);
+                        if (err != 0)
+                                return (err);
+                }
+        }
+
+        return (dsl_dataset_snapshot_check(ds, resa->tosnap, 1, tx));
 }
 
 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_FALSE,
+                    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,10 +1661,11 @@
         }
 
         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;
 
         err = dsl_sync_task_do(ds->ds_dir->dd_pool,
             recv_end_check, recv_end_sync, ds, &resa, 3);
         if (err) {
                 /* swap back */

@@ -1657,10 +1697,11 @@
         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;
 
         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 */