6536 zfs send: want a way to disable sending of free records
Reviewed by: Alexander Stetsenko <astetsenko@racktopsystems.com>
Reviewed by: Kim Shrier <kshrier@racktopsystems.com>

@@ -22,10 +22,11 @@
  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
  * Copyright 2011 Nexenta Systems, Inc. All rights reserved.
  * Copyright (c) 2011, 2015 by Delphix. All rights reserved.
  * Copyright (c) 2014, Joyent, Inc. All rights reserved.
  * Copyright 2014 HybridCluster. All rights reserved.
+ * Copyright 2015 RackTop Systems.
  */
 
 #include <sys/dmu.h>
 #include <sys/dmu_impl.h>
 #include <sys/dmu_tx.h>

@@ -150,10 +151,17 @@
     uint64_t length)
 {
         struct drr_free *drrf = &(dsp->dsa_drr->drr_u.drr_free);
 
         /*
+         * Skip free records if asked not to send them.  The resulting
+         * stream cannot be received as a clone.
+         */
+        if (dsp->dsa_skip_free)
+                return (0);
+
+        /*
          * When we receive a free record, dbuf_free_range() assumes
          * that the receiving system doesn't have any dbufs in the range
          * being freed.  This is always true because there is a one-record
          * constraint: we only send one WRITE record for any given
          * object,offset.  We know that the one-record constraint is

@@ -344,10 +352,14 @@
 static int
 dump_freeobjects(dmu_sendarg_t *dsp, uint64_t firstobj, uint64_t numobjs)
 {
         struct drr_freeobjects *drrfo = &(dsp->dsa_drr->drr_u.drr_freeobjects);
 
+        /* See comment in dump_free(). */
+        if (dsp->dsa_skip_free)
+                return (0);
+
         /*
          * If there is a pending op, but it's not PENDING_FREEOBJECTS,
          * push it out, since free block aggregation can only be done for
          * blocks of the same type (i.e., DRR_FREE records can only be
          * aggregated with other DRR_FREE records.  DRR_FREEOBJECTS records

@@ -681,12 +693,12 @@
  * Note: Releases dp using the specified tag.
  */
 static int
 dmu_send_impl(void *tag, dsl_pool_t *dp, dsl_dataset_t *to_ds,
     zfs_bookmark_phys_t *ancestor_zb,
-    boolean_t is_clone, boolean_t embedok, boolean_t large_block_ok, int outfd,
-    uint64_t resumeobj, uint64_t resumeoff,
+    boolean_t is_clone, boolean_t embedok, boolean_t large_block_ok,
+    boolean_t skip_free, int outfd, uint64_t resumeobj, uint64_t resumeoff,
     vnode_t *vp, offset_t *off)
 {
         objset_t *os;
         dmu_replay_record_t *drr;
         dmu_sendarg_t *dsp;

@@ -743,10 +755,11 @@
         if (is_clone)
                 drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_CLONE;
         drr->drr_u.drr_begin.drr_toguid = dsl_dataset_phys(to_ds)->ds_guid;
         if (dsl_dataset_phys(to_ds)->ds_flags & DS_FLAG_CI_DATASET)
                 drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_CI_DATA;
+        if (!skip_free)
         drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_FREERECORDS;
 
         if (ancestor_zb != NULL) {
                 drr->drr_u.drr_begin.drr_fromguid =
                     ancestor_zb->zbm_guid;

@@ -769,10 +782,11 @@
         dsp->dsa_toguid = dsl_dataset_phys(to_ds)->ds_guid;
         dsp->dsa_pending_op = PENDING_NONE;
         dsp->dsa_featureflags = featureflags;
         dsp->dsa_resume_object = resumeobj;
         dsp->dsa_resume_offset = resumeoff;
+        dsp->dsa_skip_free = skip_free;
 
         mutex_enter(&to_ds->ds_sendstream_lock);
         list_insert_head(&to_ds->ds_sendstreams, dsp);
         mutex_exit(&to_ds->ds_sendstream_lock);
 

@@ -871,11 +885,11 @@
         return (err);
 }
 
 int
 dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
-    boolean_t embedok, boolean_t large_block_ok,
+    boolean_t embedok, boolean_t large_block_ok, boolean_t skip_free,
     int outfd, vnode_t *vp, offset_t *off)
 {
         dsl_pool_t *dp;
         dsl_dataset_t *ds;
         dsl_dataset_t *fromds = NULL;

@@ -908,23 +922,23 @@
                 zb.zbm_creation_txg = dsl_dataset_phys(fromds)->ds_creation_txg;
                 zb.zbm_guid = dsl_dataset_phys(fromds)->ds_guid;
                 is_clone = (fromds->ds_dir != ds->ds_dir);
                 dsl_dataset_rele(fromds, FTAG);
                 err = dmu_send_impl(FTAG, dp, ds, &zb, is_clone,
-                    embedok, large_block_ok, outfd, 0, 0, vp, off);
+                    embedok, large_block_ok, skip_free, outfd, 0, 0, vp, off);
         } else {
                 err = dmu_send_impl(FTAG, dp, ds, NULL, B_FALSE,
-                    embedok, large_block_ok, outfd, 0, 0, vp, off);
+                    embedok, large_block_ok, skip_free, outfd, 0, 0, vp, off);
         }
         dsl_dataset_rele(ds, FTAG);
         return (err);
 }
 
 int
 dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
-    boolean_t large_block_ok, int outfd, uint64_t resumeobj, uint64_t resumeoff,
-    vnode_t *vp, offset_t *off)
+    boolean_t large_block_ok, boolean_t skip_free, int outfd,
+    uint64_t resumeobj, uint64_t resumeoff, vnode_t *vp, offset_t *off)
 {
         dsl_pool_t *dp;
         dsl_dataset_t *ds;
         int err;
         boolean_t owned = B_FALSE;

@@ -987,15 +1001,15 @@
                         dsl_dataset_rele(ds, FTAG);
                         dsl_pool_rele(dp, FTAG);
                         return (err);
                 }
                 err = dmu_send_impl(FTAG, dp, ds, &zb, is_clone,
-                    embedok, large_block_ok,
+                    embedok, large_block_ok, skip_free,
                     outfd, resumeobj, resumeoff, vp, off);
         } else {
                 err = dmu_send_impl(FTAG, dp, ds, NULL, B_FALSE,
-                    embedok, large_block_ok,
+                    embedok, large_block_ok, skip_free,
                     outfd, resumeobj, resumeoff, vp, off);
         }
         if (owned)
                 dsl_dataset_disown(ds, FTAG);
         else