Author: mav
Date: Wed Oct 14 11:36:55 2015
New Revision: 289312
URL: https://svnweb.freebsd.org/changeset/base/289312

Log:
  2605 want to resume interrupted zfs send
  
  Reviewed by: George Wilson <[email protected]>
  Reviewed by: Paul Dagnelie <[email protected]>
  Reviewed by: Richard Elling <[email protected]>
  Reviewed by: Xin Li <[email protected]>
  Reviewed by: Arne Jansen <[email protected]>
  Approved by: Dan McDonald <[email protected]>
  Author: Matthew Ahrens <[email protected]>
  
  illumos/illumos-gate@9c3fd1216fa7fb02cfbc78a2518a686d54b48ab8

Modified:
  vendor-sys/illumos/dist/common/zfs/zfs_prop.c
  vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_objset.c
  vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_send.c
  vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_traverse.c
  vendor-sys/illumos/dist/uts/common/fs/zfs/dsl_dataset.c
  vendor-sys/illumos/dist/uts/common/fs/zfs/dsl_destroy.c
  vendor-sys/illumos/dist/uts/common/fs/zfs/sys/dmu_impl.h
  vendor-sys/illumos/dist/uts/common/fs/zfs/sys/dmu_send.h
  vendor-sys/illumos/dist/uts/common/fs/zfs/sys/dmu_traverse.h
  vendor-sys/illumos/dist/uts/common/fs/zfs/sys/dsl_dataset.h
  vendor-sys/illumos/dist/uts/common/fs/zfs/sys/zfs_ioctl.h
  vendor-sys/illumos/dist/uts/common/fs/zfs/zfs_ioctl.c
  vendor-sys/illumos/dist/uts/common/sys/fs/zfs.h

Changes in other areas also in this revision:
Modified:
  vendor/illumos/dist/cmd/zfs/zfs_main.c
  vendor/illumos/dist/cmd/zstreamdump/zstreamdump.c
  vendor/illumos/dist/lib/libzfs/common/libzfs.h
  vendor/illumos/dist/lib/libzfs/common/libzfs_dataset.c
  vendor/illumos/dist/lib/libzfs/common/libzfs_mount.c
  vendor/illumos/dist/lib/libzfs/common/libzfs_sendrecv.c
  vendor/illumos/dist/lib/libzfs_core/common/libzfs_core.c
  vendor/illumos/dist/lib/libzfs_core/common/libzfs_core.h

Modified: vendor-sys/illumos/dist/common/zfs/zfs_prop.c
==============================================================================
--- vendor-sys/illumos/dist/common/zfs/zfs_prop.c       Wed Oct 14 11:33:07 
2015        (r289311)
+++ vendor-sys/illumos/dist/common/zfs/zfs_prop.c       Wed Oct 14 11:36:55 
2015        (r289312)
@@ -342,6 +342,10 @@ zfs_prop_init(void)
        zprop_register_string(ZFS_PROP_MLSLABEL, "mlslabel",
            ZFS_MLSLABEL_DEFAULT, PROP_INHERIT, ZFS_TYPE_DATASET,
            "<sensitivity label>", "MLSLABEL");
+       zprop_register_string(ZFS_PROP_RECEIVE_RESUME_TOKEN,
+           "receive_resume_token",
+           NULL, PROP_READONLY, ZFS_TYPE_FILESYSTEM | ZFS_TYPE_VOLUME,
+           "<string token>", "RESUMETOK");
 
        /* readonly number properties */
        zprop_register_number(ZFS_PROP_USED, "used", 0, PROP_READONLY,

Modified: vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_objset.c
==============================================================================
--- vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_objset.c      Wed Oct 14 
11:33:07 2015        (r289311)
+++ vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_objset.c      Wed Oct 14 
11:36:55 2015        (r289312)
@@ -362,6 +362,17 @@ dmu_objset_open_impl(spa_t *spa, dsl_dat
         * checksum/compression/copies.
         */
        if (ds != NULL) {
+               boolean_t needlock = B_FALSE;
+
+               /*
+                * Note: it's valid to open the objset if the dataset is
+                * long-held, in which case the pool_config lock will not
+                * be held.
+                */
+               if (!dsl_pool_config_held(dmu_objset_pool(os))) {
+                       needlock = B_TRUE;
+                       dsl_pool_config_enter(dmu_objset_pool(os), FTAG);
+               }
                err = dsl_prop_register(ds,
                    zfs_prop_to_name(ZFS_PROP_PRIMARYCACHE),
                    primary_cache_changed_cb, os);
@@ -413,6 +424,8 @@ dmu_objset_open_impl(spa_t *spa, dsl_dat
                                    recordsize_changed_cb, os);
                        }
                }
+               if (needlock)
+                       dsl_pool_config_exit(dmu_objset_pool(os), FTAG);
                if (err != 0) {
                        VERIFY(arc_buf_remove_ref(os->os_phys_buf,
                            &os->os_phys_buf));
@@ -469,6 +482,13 @@ dmu_objset_from_ds(dsl_dataset_t *ds, ob
 {
        int err = 0;
 
+       /*
+        * We shouldn't be doing anything with dsl_dataset_t's unless the
+        * pool_config lock is held, or the dataset is long-held.
+        */
+       ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool) ||
+           dsl_dataset_long_held(ds));
+
        mutex_enter(&ds->ds_opening_lock);
        if (ds->ds_objset == NULL) {
                objset_t *os;

Modified: vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_send.c
==============================================================================
--- vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_send.c        Wed Oct 14 
11:33:07 2015        (r289311)
+++ vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_send.c        Wed Oct 14 
11:36:55 2015        (r289312)
@@ -60,12 +60,14 @@ int zfs_send_queue_length = 16 * 1024 * 
 int zfs_recv_queue_length = 16 * 1024 * 1024;
 
 static char *dmu_recv_tag = "dmu_recv_tag";
-static const char *recv_clone_name = "%recv";
+const char *recv_clone_name = "%recv";
 
 #define        BP_SPAN(datablkszsec, indblkshift, level) \
        (((uint64_t)datablkszsec) << (SPA_MINBLOCKSHIFT + \
        (level) * (indblkshift - SPA_BLKPTRSHIFT)))
 
+static void byteswap_record(dmu_replay_record_t *drr);
+
 struct send_thread_arg {
        bqueue_t        q;
        dsl_dataset_t   *ds;            /* Dataset to traverse */
@@ -73,6 +75,7 @@ struct send_thread_arg {
        int             flags;          /* flags to pass to traverse_dataset */
        int             error_code;
        boolean_t       cancel;
+       zbookmark_phys_t resume;
 };
 
 struct send_block_record {
@@ -87,7 +90,7 @@ struct send_block_record {
 static int
 dump_bytes(dmu_sendarg_t *dsp, void *buf, int len)
 {
-       dsl_dataset_t *ds = dsp->dsa_os->os_dsl_dataset;
+       dsl_dataset_t *ds = dmu_objset_ds(dsp->dsa_os);
        ssize_t resid; /* have to get resid to get detailed errno */
        ASSERT0(len % 8);
 
@@ -145,7 +148,7 @@ dump_free(dmu_sendarg_t *dsp, uint64_t o
         * 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
+        * object,offset.  We know that the one-record constraint is
         * true because we always send data in increasing order by
         * object,offset.
         *
@@ -394,6 +397,19 @@ dump_dnode(dmu_sendarg_t *dsp, uint64_t 
 {
        struct drr_object *drro = &(dsp->dsa_drr->drr_u.drr_object);
 
+       if (object < dsp->dsa_resume_object) {
+               /*
+                * Note: when resuming, we will visit all the dnodes in
+                * the block of dnodes that we are resuming from.  In
+                * this case it's unnecessary to send the dnodes prior to
+                * the one we are resuming from.  We should be at most one
+                * block's worth of dnodes behind the resume point.
+                */
+               ASSERT3U(dsp->dsa_resume_object - object, <,
+                   1 << (DNODE_BLOCK_SHIFT - DNODE_SHIFT));
+               return (0);
+       }
+
        if (dnp == NULL || dnp->dn_type == DMU_OT_NONE)
                return (dump_freeobjects(dsp, object, 1));
 
@@ -474,6 +490,9 @@ send_cb(spa_t *spa, zilog_t *zilog, cons
        uint64_t record_size;
        int err = 0;
 
+       ASSERT(zb->zb_object == DMU_META_DNODE_OBJECT ||
+           zb->zb_object >= sta->resume.zb_object);
+
        if (sta->cancel)
                return (SET_ERROR(EINTR));
 
@@ -510,8 +529,10 @@ send_traverse_thread(void *arg)
        struct send_block_record *data;
 
        if (st_arg->ds != NULL) {
-               err = traverse_dataset(st_arg->ds, st_arg->fromtxg,
-                   st_arg->flags, send_cb, arg);
+               err = traverse_dataset_resume(st_arg->ds,
+                   st_arg->fromtxg, &st_arg->resume,
+                   st_arg->flags, send_cb, st_arg);
+
                if (err != EINTR)
                        st_arg->error_code = err;
        }
@@ -539,6 +560,9 @@ do_dump(dmu_sendarg_t *dsa, struct send_
 
        ASSERT3U(zb->zb_level, >=, 0);
 
+       ASSERT(zb->zb_object == DMU_META_DNODE_OBJECT ||
+           zb->zb_object >= dsa->dsa_resume_object);
+
        if (zb->zb_object != DMU_META_DNODE_OBJECT &&
            DMU_OBJECT_IS_SPECIAL(zb->zb_object)) {
                return (0);
@@ -599,6 +623,10 @@ do_dump(dmu_sendarg_t *dsa, struct send_
                uint64_t offset;
 
                ASSERT0(zb->zb_level);
+               ASSERT(zb->zb_object > dsa->dsa_resume_object ||
+                   (zb->zb_object == dsa->dsa_resume_object &&
+                   zb->zb_blkid * blksz >= dsa->dsa_resume_offset));
+
                if (arc_read(NULL, spa, bp, arc_getbuf_func, &abuf,
                    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL,
                    &aflags, zb) != 0) {
@@ -659,8 +687,10 @@ get_next_record(bqueue_t *bq, struct sen
  */
 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, vnode_t *vp, offset_t *off)
+    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,
+    vnode_t *vp, offset_t *off)
 {
        objset_t *os;
        dmu_replay_record_t *drr;
@@ -668,7 +698,7 @@ dmu_send_impl(void *tag, dsl_pool_t *dp,
        int err;
        uint64_t fromtxg = 0;
        uint64_t featureflags = 0;
-       struct send_thread_arg to_arg;
+       struct send_thread_arg to_arg = { 0 };
 
        err = dmu_objset_from_ds(to_ds, &os);
        if (err != 0) {
@@ -705,6 +735,10 @@ dmu_send_impl(void *tag, dsl_pool_t *dp,
                        featureflags |= DMU_BACKUP_FEATURE_EMBED_DATA_LZ4;
        }
 
+       if (resumeobj != 0 || resumeoff != 0) {
+               featureflags |= DMU_BACKUP_FEATURE_RESUMING;
+       }
+
        DMU_SET_FEATUREFLAGS(drr->drr_u.drr_begin.drr_versioninfo,
            featureflags);
 
@@ -740,6 +774,8 @@ dmu_send_impl(void *tag, dsl_pool_t *dp,
        dsp->dsa_pending_op = PENDING_NONE;
        dsp->dsa_incremental = (ancestor_zb != NULL);
        dsp->dsa_featureflags = featureflags;
+       dsp->dsa_resume_object = resumeobj;
+       dsp->dsa_resume_offset = resumeoff;
 
        mutex_enter(&to_ds->ds_sendstream_lock);
        list_insert_head(&to_ds->ds_sendstreams, dsp);
@@ -748,7 +784,27 @@ dmu_send_impl(void *tag, dsl_pool_t *dp,
        dsl_dataset_long_hold(to_ds, FTAG);
        dsl_pool_rele(dp, tag);
 
-       if (dump_record(dsp, NULL, 0) != 0) {
+       void *payload = NULL;
+       size_t payload_len = 0;
+       if (resumeobj != 0 || resumeoff != 0) {
+               dmu_object_info_t to_doi;
+               err = dmu_object_info(os, resumeobj, &to_doi);
+               if (err != 0)
+                       goto out;
+               SET_BOOKMARK(&to_arg.resume, to_ds->ds_object, resumeobj, 0,
+                   resumeoff / to_doi.doi_data_block_size);
+
+               nvlist_t *nvl = fnvlist_alloc();
+               fnvlist_add_uint64(nvl, "resume_object", resumeobj);
+               fnvlist_add_uint64(nvl, "resume_offset", resumeoff);
+               payload = fnvlist_pack(nvl, &payload_len);
+               drr->drr_payloadlen = payload_len;
+               fnvlist_free(nvl);
+       }
+
+       err = dump_record(dsp, payload, payload_len);
+       fnvlist_pack_free(payload, payload_len);
+       if (err != 0) {
                err = dsp->dsa_err;
                goto out;
        }
@@ -859,19 +915,19 @@ dmu_send_obj(const char *pool, uint64_t 
                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, vp, off);
+                   embedok, large_block_ok, outfd, 0, 0, vp, off);
        } else {
                err = dmu_send_impl(FTAG, dp, ds, NULL, B_FALSE,
-                   embedok, large_block_ok, outfd, vp, off);
+                   embedok, large_block_ok, 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, vnode_t *vp, offset_t *off)
+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)
 {
        dsl_pool_t *dp;
        dsl_dataset_t *ds;
@@ -938,10 +994,12 @@ dmu_send(const char *tosnap, const char 
                        return (err);
                }
                err = dmu_send_impl(FTAG, dp, ds, &zb, is_clone,
-                   embedok, large_block_ok, outfd, vp, off);
+                   embedok, large_block_ok,
+                   outfd, resumeobj, resumeoff, vp, off);
        } else {
                err = dmu_send_impl(FTAG, dp, ds, NULL, B_FALSE,
-                   embedok, large_block_ok, outfd, vp, off);
+                   embedok, large_block_ok,
+                   outfd, resumeobj, resumeoff, vp, off);
        }
        if (owned)
                dsl_dataset_disown(ds, FTAG);
@@ -1184,6 +1242,7 @@ dmu_recv_begin_check(void *arg, dmu_tx_t
 
        /* already checked */
        ASSERT3U(drrb->drr_magic, ==, DMU_BACKUP_MAGIC);
+       ASSERT(!(featureflags & DMU_BACKUP_FEATURE_RESUMING));
 
        if (DMU_GET_STREAM_HDRTYPE(drrb->drr_versioninfo) ==
            DMU_COMPOUNDSTREAM ||
@@ -1196,6 +1255,10 @@ dmu_recv_begin_check(void *arg, dmu_tx_t
            spa_version(dp->dp_spa) < SPA_VERSION_SA)
                return (SET_ERROR(ENOTSUP));
 
+       if (drba->drba_cookie->drc_resumable &&
+           !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_EXTENSIBLE_DATASET))
+               return (SET_ERROR(ENOTSUP));
+
        /*
         * The receiving code doesn't know how to translate a WRITE_EMBEDDED
         * record to a plan WRITE record, so the pool must have the
@@ -1299,15 +1362,16 @@ dmu_recv_begin_sync(void *arg, dmu_tx_t 
 {
        dmu_recv_begin_arg_t *drba = arg;
        dsl_pool_t *dp = dmu_tx_pool(tx);
+       objset_t *mos = dp->dp_meta_objset;
        struct drr_begin *drrb = drba->drba_cookie->drc_drrb;
        const char *tofs = drba->drba_cookie->drc_tofs;
        dsl_dataset_t *ds, *newds;
        uint64_t dsobj;
        int error;
-       uint64_t crflags;
+       uint64_t crflags = 0;
 
-       crflags = (drrb->drr_flags & DRR_FLAG_CI_DATA) ?
-           DS_FLAG_CI_DATASET : 0;
+       if (drrb->drr_flags & DRR_FLAG_CI_DATA)
+               crflags |= DS_FLAG_CI_DATASET;
 
        error = dsl_dataset_hold(dp, tofs, FTAG, &ds);
        if (error == 0) {
@@ -1345,6 +1409,31 @@ dmu_recv_begin_sync(void *arg, dmu_tx_t 
        }
        VERIFY0(dsl_dataset_own_obj(dp, dsobj, dmu_recv_tag, &newds));
 
+       if (drba->drba_cookie->drc_resumable) {
+               dsl_dataset_zapify(newds, tx);
+               if (drrb->drr_fromguid != 0) {
+                       VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_FROMGUID,
+                           8, 1, &drrb->drr_fromguid, tx));
+               }
+               VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_TOGUID,
+                   8, 1, &drrb->drr_toguid, tx));
+               VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_TONAME,
+                   1, strlen(drrb->drr_toname) + 1, drrb->drr_toname, tx));
+               uint64_t one = 1;
+               uint64_t zero = 0;
+               VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_OBJECT,
+                   8, 1, &one, tx));
+               VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_OFFSET,
+                   8, 1, &zero, tx));
+               VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_BYTES,
+                   8, 1, &zero, tx));
+               if (DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo) &
+                   DMU_BACKUP_FEATURE_EMBED_DATA) {
+                       VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_EMBEDOK,
+                           8, 1, &one, tx));
+               }
+       }
+
        dmu_buf_will_dirty(newds->ds_dbuf, tx);
        dsl_dataset_phys(newds)->ds_flags |= DS_FLAG_INCONSISTENT;
 
@@ -1362,56 +1451,192 @@ dmu_recv_begin_sync(void *arg, dmu_tx_t 
        spa_history_log_internal_ds(newds, "receive", tx, "");
 }
 
+static int
+dmu_recv_resume_begin_check(void *arg, dmu_tx_t *tx)
+{
+       dmu_recv_begin_arg_t *drba = arg;
+       dsl_pool_t *dp = dmu_tx_pool(tx);
+       struct drr_begin *drrb = drba->drba_cookie->drc_drrb;
+       int error;
+       uint64_t featureflags = DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo);
+       dsl_dataset_t *ds;
+       const char *tofs = drba->drba_cookie->drc_tofs;
+
+       /* already checked */
+       ASSERT3U(drrb->drr_magic, ==, DMU_BACKUP_MAGIC);
+       ASSERT(featureflags & DMU_BACKUP_FEATURE_RESUMING);
+
+       if (DMU_GET_STREAM_HDRTYPE(drrb->drr_versioninfo) ==
+           DMU_COMPOUNDSTREAM ||
+           drrb->drr_type >= DMU_OST_NUMTYPES)
+               return (SET_ERROR(EINVAL));
+
+       /* Verify pool version supports SA if SA_SPILL feature set */
+       if ((featureflags & DMU_BACKUP_FEATURE_SA_SPILL) &&
+           spa_version(dp->dp_spa) < SPA_VERSION_SA)
+               return (SET_ERROR(ENOTSUP));
+
+       /*
+        * The receiving code doesn't know how to translate a WRITE_EMBEDDED
+        * record to a plain WRITE record, so the pool must have the
+        * EMBEDDED_DATA feature enabled if the stream has WRITE_EMBEDDED
+        * records.  Same with WRITE_EMBEDDED records that use LZ4 compression.
+        */
+       if ((featureflags & DMU_BACKUP_FEATURE_EMBED_DATA) &&
+           !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_EMBEDDED_DATA))
+               return (SET_ERROR(ENOTSUP));
+       if ((featureflags & DMU_BACKUP_FEATURE_EMBED_DATA_LZ4) &&
+           !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_LZ4_COMPRESS))
+               return (SET_ERROR(ENOTSUP));
+
+       char recvname[ZFS_MAXNAMELEN];
+
+       (void) snprintf(recvname, sizeof (recvname), "%s/%s",
+           tofs, recv_clone_name);
+
+       if (dsl_dataset_hold(dp, recvname, FTAG, &ds) != 0) {
+               /* %recv does not exist; continue in tofs */
+               error = dsl_dataset_hold(dp, tofs, FTAG, &ds);
+               if (error != 0)
+                       return (error);
+       }
+
+       /* check that ds is marked inconsistent */
+       if (!DS_IS_INCONSISTENT(ds)) {
+               dsl_dataset_rele(ds, FTAG);
+               return (SET_ERROR(EINVAL));
+       }
+
+       /* check that there is resuming data, and that the toguid matches */
+       if (!dsl_dataset_is_zapified(ds)) {
+               dsl_dataset_rele(ds, FTAG);
+               return (SET_ERROR(EINVAL));
+       }
+       uint64_t val;
+       error = zap_lookup(dp->dp_meta_objset, ds->ds_object,
+           DS_FIELD_RESUME_TOGUID, sizeof (val), 1, &val);
+       if (error != 0 || drrb->drr_toguid != val) {
+               dsl_dataset_rele(ds, FTAG);
+               return (SET_ERROR(EINVAL));
+       }
+
+       /*
+        * Check if the receive is still running.  If so, it will be owned.
+        * Note that nothing else can own the dataset (e.g. after the receive
+        * fails) because it will be marked inconsistent.
+        */
+       if (dsl_dataset_has_owner(ds)) {
+               dsl_dataset_rele(ds, FTAG);
+               return (SET_ERROR(EBUSY));
+       }
+
+       /* There should not be any snapshots of this fs yet. */
+       if (ds->ds_prev != NULL && ds->ds_prev->ds_dir == ds->ds_dir) {
+               dsl_dataset_rele(ds, FTAG);
+               return (SET_ERROR(EINVAL));
+       }
+
+       /*
+        * Note: resume point will be checked when we process the first WRITE
+        * record.
+        */
+
+       /* check that the origin matches */
+       val = 0;
+       (void) zap_lookup(dp->dp_meta_objset, ds->ds_object,
+           DS_FIELD_RESUME_FROMGUID, sizeof (val), 1, &val);
+       if (drrb->drr_fromguid != val) {
+               dsl_dataset_rele(ds, FTAG);
+               return (SET_ERROR(EINVAL));
+       }
+
+       dsl_dataset_rele(ds, FTAG);
+       return (0);
+}
+
+static void
+dmu_recv_resume_begin_sync(void *arg, dmu_tx_t *tx)
+{
+       dmu_recv_begin_arg_t *drba = arg;
+       dsl_pool_t *dp = dmu_tx_pool(tx);
+       const char *tofs = drba->drba_cookie->drc_tofs;
+       dsl_dataset_t *ds;
+       uint64_t dsobj;
+       char recvname[ZFS_MAXNAMELEN];
+
+       (void) snprintf(recvname, sizeof (recvname), "%s/%s",
+           tofs, recv_clone_name);
+
+       if (dsl_dataset_hold(dp, recvname, FTAG, &ds) != 0) {
+               /* %recv does not exist; continue in tofs */
+               VERIFY0(dsl_dataset_hold(dp, tofs, FTAG, &ds));
+               drba->drba_cookie->drc_newfs = B_TRUE;
+       }
+
+       /* clear the inconsistent flag so that we can own it */
+       ASSERT(DS_IS_INCONSISTENT(ds));
+       dmu_buf_will_dirty(ds->ds_dbuf, tx);
+       dsl_dataset_phys(ds)->ds_flags &= ~DS_FLAG_INCONSISTENT;
+       dsobj = ds->ds_object;
+       dsl_dataset_rele(ds, FTAG);
+
+       VERIFY0(dsl_dataset_own_obj(dp, dsobj, dmu_recv_tag, &ds));
+
+       dmu_buf_will_dirty(ds->ds_dbuf, tx);
+       dsl_dataset_phys(ds)->ds_flags |= DS_FLAG_INCONSISTENT;
+
+       ASSERT(!BP_IS_HOLE(dsl_dataset_get_blkptr(ds)));
+
+       drba->drba_cookie->drc_ds = ds;
+
+       spa_history_log_internal_ds(ds, "resume receive", tx, "");
+}
+
 /*
  * NB: callers *MUST* call dmu_recv_stream() if dmu_recv_begin()
  * succeeds; otherwise we will leak the holds on the datasets.
  */
 int
-dmu_recv_begin(char *tofs, char *tosnap, struct drr_begin *drrb,
-    boolean_t force, char *origin, dmu_recv_cookie_t *drc)
+dmu_recv_begin(char *tofs, char *tosnap, dmu_replay_record_t *drr_begin,
+    boolean_t force, boolean_t resumable, char *origin, dmu_recv_cookie_t *drc)
 {
        dmu_recv_begin_arg_t drba = { 0 };
-       dmu_replay_record_t *drr;
 
        bzero(drc, sizeof (dmu_recv_cookie_t));
-       drc->drc_drrb = drrb;
+       drc->drc_drr_begin = drr_begin;
+       drc->drc_drrb = &drr_begin->drr_u.drr_begin;
        drc->drc_tosnap = tosnap;
        drc->drc_tofs = tofs;
        drc->drc_force = force;
+       drc->drc_resumable = resumable;
        drc->drc_cred = CRED();
 
-       if (drrb->drr_magic == BSWAP_64(DMU_BACKUP_MAGIC))
+       if (drc->drc_drrb->drr_magic == BSWAP_64(DMU_BACKUP_MAGIC)) {
                drc->drc_byteswap = B_TRUE;
-       else if (drrb->drr_magic != DMU_BACKUP_MAGIC)
-               return (SET_ERROR(EINVAL));
-
-       drr = kmem_zalloc(sizeof (dmu_replay_record_t), KM_SLEEP);
-       drr->drr_type = DRR_BEGIN;
-       drr->drr_u.drr_begin = *drc->drc_drrb;
-       if (drc->drc_byteswap) {
-               fletcher_4_incremental_byteswap(drr,
+               fletcher_4_incremental_byteswap(drr_begin,
                    sizeof (dmu_replay_record_t), &drc->drc_cksum);
-       } else {
-               fletcher_4_incremental_native(drr,
+               byteswap_record(drr_begin);
+       } else if (drc->drc_drrb->drr_magic == DMU_BACKUP_MAGIC) {
+               fletcher_4_incremental_native(drr_begin,
                    sizeof (dmu_replay_record_t), &drc->drc_cksum);
-       }
-       kmem_free(drr, sizeof (dmu_replay_record_t));
-
-       if (drc->drc_byteswap) {
-               drrb->drr_magic = BSWAP_64(drrb->drr_magic);
-               drrb->drr_versioninfo = BSWAP_64(drrb->drr_versioninfo);
-               drrb->drr_creation_time = BSWAP_64(drrb->drr_creation_time);
-               drrb->drr_type = BSWAP_32(drrb->drr_type);
-               drrb->drr_toguid = BSWAP_64(drrb->drr_toguid);
-               drrb->drr_fromguid = BSWAP_64(drrb->drr_fromguid);
+       } else {
+               return (SET_ERROR(EINVAL));
        }
 
        drba.drba_origin = origin;
        drba.drba_cookie = drc;
        drba.drba_cred = CRED();
 
-       return (dsl_sync_task(tofs, dmu_recv_begin_check, dmu_recv_begin_sync,
-           &drba, 5, ZFS_SPACE_CHECK_NORMAL));
+       if (DMU_GET_FEATUREFLAGS(drc->drc_drrb->drr_versioninfo) &
+           DMU_BACKUP_FEATURE_RESUMING) {
+               return (dsl_sync_task(tofs,
+                   dmu_recv_resume_begin_check, dmu_recv_resume_begin_sync,
+                   &drba, 5, ZFS_SPACE_CHECK_NORMAL));
+       } else  {
+               return (dsl_sync_task(tofs,
+                   dmu_recv_begin_check, dmu_recv_begin_sync,
+                   &drba, 5, ZFS_SPACE_CHECK_NORMAL));
+       }
 }
 
 struct receive_record_arg {
@@ -1423,6 +1648,7 @@ struct receive_record_arg {
         */
        arc_buf_t *write_buf;
        int payload_size;
+       uint64_t bytes_read; /* bytes read from stream when record created */
        boolean_t eos_marker; /* Marks the end of the stream */
        bqueue_node_t node;
 };
@@ -1431,6 +1657,7 @@ struct receive_writer_arg {
        objset_t *os;
        boolean_t byteswap;
        bqueue_t q;
+
        /*
         * These three args are used to signal to the main thread that we're
         * done.
@@ -1438,15 +1665,20 @@ struct receive_writer_arg {
        kmutex_t mutex;
        kcondvar_t cv;
        boolean_t done;
+
        int err;
        /* A map from guid to dataset to help handle dedup'd streams. */
        avl_tree_t *guid_to_ds_map;
+       boolean_t resumable;
+       uint64_t last_object, last_offset;
+       uint64_t bytes_read; /* bytes read when current record created */
 };
 
 struct receive_arg  {
        objset_t *os;
        vnode_t *vp; /* The vnode to read the stream from */
        uint64_t voff; /* The current offset in the stream */
+       uint64_t bytes_read;
        /*
         * A record that has had its payload read in, but hasn't yet been handed
         * off to the worker thread.
@@ -1518,14 +1750,21 @@ receive_read(struct receive_arg *ra, int
                    ra->voff, UIO_SYSSPACE, FAPPEND,
                    RLIM64_INFINITY, CRED(), &resid);
 
-               if (resid == len - done)
-                       ra->err = SET_ERROR(EINVAL);
+               if (resid == len - done) {
+                       /*
+                        * Note: ECKSUM indicates that the receive
+                        * was interrupted and can potentially be resumed.
+                        */
+                       ra->err = SET_ERROR(ECKSUM);
+               }
                ra->voff += len - done - resid;
                done = len - resid;
                if (ra->err != 0)
                        return (ra->err);
        }
 
+       ra->bytes_read += len;
+
        ASSERT3U(done, ==, len);
        return (0);
 }
@@ -1626,6 +1865,43 @@ deduce_nblkptr(dmu_object_type_t bonus_t
        }
 }
 
+static void
+save_resume_state(struct receive_writer_arg *rwa,
+    uint64_t object, uint64_t offset, dmu_tx_t *tx)
+{
+       int txgoff = dmu_tx_get_txg(tx) & TXG_MASK;
+
+       if (!rwa->resumable)
+               return;
+
+       /*
+        * We use ds_resume_bytes[] != 0 to indicate that we need to
+        * update this on disk, so it must not be 0.
+        */
+       ASSERT(rwa->bytes_read != 0);
+
+       /*
+        * We only resume from write records, which have a valid
+        * (non-meta-dnode) object number.
+        */
+       ASSERT(object != 0);
+
+       /*
+        * For resuming to work correctly, we must receive records in order,
+        * sorted by object,offset.  This is checked by the callers, but
+        * assert it here for good measure.
+        */
+       ASSERT3U(object, >=, rwa->os->os_dsl_dataset->ds_resume_object[txgoff]);
+       ASSERT(object != rwa->os->os_dsl_dataset->ds_resume_object[txgoff] ||
+           offset >= rwa->os->os_dsl_dataset->ds_resume_offset[txgoff]);
+       ASSERT3U(rwa->bytes_read, >=,
+           rwa->os->os_dsl_dataset->ds_resume_bytes[txgoff]);
+
+       rwa->os->os_dsl_dataset->ds_resume_object[txgoff] = object;
+       rwa->os->os_dsl_dataset->ds_resume_offset[txgoff] = offset;
+       rwa->os->os_dsl_dataset->ds_resume_bytes[txgoff] = rwa->bytes_read;
+}
+
 static int
 receive_object(struct receive_writer_arg *rwa, struct drr_object *drro,
     void *data)
@@ -1722,6 +1998,7 @@ receive_object(struct receive_writer_arg
                dmu_buf_rele(db, FTAG);
        }
        dmu_tx_commit(tx);
+
        return (0);
 }
 
@@ -1747,6 +2024,7 @@ receive_freeobjects(struct receive_write
                if (err != 0)
                        return (err);
        }
+
        return (0);
 }
 
@@ -1761,6 +2039,18 @@ receive_write(struct receive_writer_arg 
            !DMU_OT_IS_VALID(drrw->drr_type))
                return (SET_ERROR(EINVAL));
 
+       /*
+        * For resuming to work, records must be in increasing order
+        * by (object, offset).
+        */
+       if (drrw->drr_object < rwa->last_object ||
+           (drrw->drr_object == rwa->last_object &&
+           drrw->drr_offset < rwa->last_offset)) {
+               return (SET_ERROR(EINVAL));
+       }
+       rwa->last_object = drrw->drr_object;
+       rwa->last_offset = drrw->drr_offset;
+
        if (dmu_object_info(rwa->os, drrw->drr_object, NULL) != 0)
                return (SET_ERROR(EINVAL));
 
@@ -1784,8 +2074,17 @@ receive_write(struct receive_writer_arg 
        if (dmu_bonus_hold(rwa->os, drrw->drr_object, FTAG, &bonus) != 0)
                return (SET_ERROR(EINVAL));
        dmu_assign_arcbuf(bonus, drrw->drr_offset, abuf, tx);
+
+       /*
+        * Note: If the receive fails, we want the resume stream to start
+        * with the same record that we last successfully received (as opposed
+        * to the next record), so that we can verify that we are
+        * resuming from the correct location.
+        */
+       save_resume_state(rwa, drrw->drr_object, drrw->drr_offset, tx);
        dmu_tx_commit(tx);
        dmu_buf_rele(bonus, FTAG);
+
        return (0);
 }
 
@@ -1844,43 +2143,48 @@ receive_write_byref(struct receive_write
        dmu_write(rwa->os, drrwbr->drr_object,
            drrwbr->drr_offset, drrwbr->drr_length, dbp->db_data, tx);
        dmu_buf_rele(dbp, FTAG);
+
+       /* See comment in restore_write. */
+       save_resume_state(rwa, drrwbr->drr_object, drrwbr->drr_offset, tx);
        dmu_tx_commit(tx);
        return (0);
 }
 
 static int
 receive_write_embedded(struct receive_writer_arg *rwa,
-    struct drr_write_embedded *drrwnp, void *data)
+    struct drr_write_embedded *drrwe, void *data)
 {
        dmu_tx_t *tx;
        int err;
 
-       if (drrwnp->drr_offset + drrwnp->drr_length < drrwnp->drr_offset)
+       if (drrwe->drr_offset + drrwe->drr_length < drrwe->drr_offset)
                return (EINVAL);
 
-       if (drrwnp->drr_psize > BPE_PAYLOAD_SIZE)
+       if (drrwe->drr_psize > BPE_PAYLOAD_SIZE)
                return (EINVAL);
 
-       if (drrwnp->drr_etype >= NUM_BP_EMBEDDED_TYPES)
+       if (drrwe->drr_etype >= NUM_BP_EMBEDDED_TYPES)
                return (EINVAL);
-       if (drrwnp->drr_compression >= ZIO_COMPRESS_FUNCTIONS)
+       if (drrwe->drr_compression >= ZIO_COMPRESS_FUNCTIONS)
                return (EINVAL);
 
        tx = dmu_tx_create(rwa->os);
 
-       dmu_tx_hold_write(tx, drrwnp->drr_object,
-           drrwnp->drr_offset, drrwnp->drr_length);
+       dmu_tx_hold_write(tx, drrwe->drr_object,
+           drrwe->drr_offset, drrwe->drr_length);
        err = dmu_tx_assign(tx, TXG_WAIT);
        if (err != 0) {
                dmu_tx_abort(tx);
                return (err);
        }
 
-       dmu_write_embedded(rwa->os, drrwnp->drr_object,
-           drrwnp->drr_offset, data, drrwnp->drr_etype,
-           drrwnp->drr_compression, drrwnp->drr_lsize, drrwnp->drr_psize,
+       dmu_write_embedded(rwa->os, drrwe->drr_object,
+           drrwe->drr_offset, data, drrwe->drr_etype,
+           drrwe->drr_compression, drrwe->drr_lsize, drrwe->drr_psize,
            rwa->byteswap ^ ZFS_HOST_BYTEORDER, tx);
 
+       /* See comment in restore_write. */
+       save_resume_state(rwa, drrwe->drr_object, drrwe->drr_offset, tx);
        dmu_tx_commit(tx);
        return (0);
 }
@@ -1954,10 +2258,16 @@ receive_free(struct receive_writer_arg *
 static void
 dmu_recv_cleanup_ds(dmu_recv_cookie_t *drc)
 {
-       char name[MAXNAMELEN];
-       dsl_dataset_name(drc->drc_ds, name);
-       dsl_dataset_disown(drc->drc_ds, dmu_recv_tag);
-       (void) dsl_destroy_head(name);
+       if (drc->drc_resumable) {
+               /* wait for our resume state to be written to disk */
+               txg_wait_synced(drc->drc_ds->ds_dir->dd_pool, 0);
+               dsl_dataset_disown(drc->drc_ds, dmu_recv_tag);
+       } else {
+               char name[MAXNAMELEN];
+               dsl_dataset_name(drc->drc_ds, name);
+               dsl_dataset_disown(drc->drc_ds, dmu_recv_tag);
+               (void) dsl_destroy_head(name);
+       }
 }
 
 static void
@@ -1984,12 +2294,17 @@ receive_read_payload_and_next_header(str
 
        if (len != 0) {
                ASSERT3U(len, <=, SPA_MAXBLOCKSIZE);
-               ra->rrd->payload = buf;
-               ra->rrd->payload_size = len;
-               err = receive_read(ra, len, ra->rrd->payload);
+               err = receive_read(ra, len, buf);
                if (err != 0)
                        return (err);
-               receive_cksum(ra, len, ra->rrd->payload);
+               receive_cksum(ra, len, buf);
+
+               /* note: rrd is NULL when reading the begin record's payload */
+               if (ra->rrd != NULL) {
+                       ra->rrd->payload = buf;
+                       ra->rrd->payload_size = len;
+                       ra->rrd->bytes_read = ra->bytes_read;
+               }
        }
 
        ra->prev_cksum = ra->cksum;
@@ -1997,6 +2312,7 @@ receive_read_payload_and_next_header(str
        ra->next_rrd = kmem_zalloc(sizeof (*ra->next_rrd), KM_SLEEP);
        err = receive_read(ra, sizeof (ra->next_rrd->header),
            &ra->next_rrd->header);
+       ra->next_rrd->bytes_read = ra->bytes_read;
        if (err != 0) {
                kmem_free(ra->next_rrd, sizeof (*ra->next_rrd));
                ra->next_rrd = NULL;
@@ -2176,7 +2492,7 @@ receive_read_record(struct receive_arg *
        {
                struct drr_end *drre = &ra->rrd->header.drr_u.drr_end;
                if (!ZIO_CHECKSUM_EQUAL(ra->prev_cksum, drre->drr_checksum))
-                       return (SET_ERROR(EINVAL));
+                       return (SET_ERROR(ECKSUM));
                return (0);
        }
        case DRR_SPILL:
@@ -2203,6 +2519,10 @@ receive_process_record(struct receive_wr
 {
        int err;
 
+       /* Processing in order, therefore bytes_read should be increasing. */
+       ASSERT3U(rrd->bytes_read, >=, rwa->bytes_read);
+       rwa->bytes_read = rrd->bytes_read;
+
        switch (rrd->header.drr_type) {
        case DRR_OBJECT:
        {
@@ -2297,6 +2617,33 @@ receive_writer_thread(void *arg)
        mutex_exit(&rwa->mutex);
 }
 
+static int
+resume_check(struct receive_arg *ra, nvlist_t *begin_nvl)
+{
+       uint64_t val;
+       objset_t *mos = dmu_objset_pool(ra->os)->dp_meta_objset;
+       uint64_t dsobj = dmu_objset_id(ra->os);
+       uint64_t resume_obj, resume_off;
+
+       if (nvlist_lookup_uint64(begin_nvl,
+           "resume_object", &resume_obj) != 0 ||
+           nvlist_lookup_uint64(begin_nvl,
+           "resume_offset", &resume_off) != 0) {
+               return (SET_ERROR(EINVAL));
+       }
+       VERIFY0(zap_lookup(mos, dsobj,
+           DS_FIELD_RESUME_OBJECT, sizeof (val), 1, &val));
+       if (resume_obj != val)
+               return (SET_ERROR(EINVAL));
+       VERIFY0(zap_lookup(mos, dsobj,
+           DS_FIELD_RESUME_OFFSET, sizeof (val), 1, &val));
+       if (resume_off != val)
+               return (SET_ERROR(EINVAL));
+
+       return (0);
+}
+
+
 /*
  * Read in the stream's records, one by one, and apply them to the pool.  There
  * are two threads involved; the thread that calls this function will spin up a
@@ -2317,11 +2664,19 @@ dmu_recv_stream(dmu_recv_cookie_t *drc, 
        struct receive_arg ra = { 0 };
        struct receive_writer_arg rwa = { 0 };
        int featureflags;
+       nvlist_t *begin_nvl = NULL;
 
        ra.byteswap = drc->drc_byteswap;
        ra.cksum = drc->drc_cksum;
        ra.vp = vp;
        ra.voff = *voffp;
+
+       if (dsl_dataset_is_zapified(drc->drc_ds)) {
+               (void) zap_lookup(drc->drc_ds->ds_dir->dd_pool->dp_meta_objset,
+                   drc->drc_ds->ds_object, DS_FIELD_RESUME_BYTES,
+                   sizeof (ra.bytes_read), 1, &ra.bytes_read);
+       }
+
        list_create(&ra.ignore_obj_list, sizeof (struct receive_ign_obj_node),
            offsetof(struct receive_ign_obj_node, node));
 
@@ -2374,9 +2729,29 @@ dmu_recv_stream(dmu_recv_cookie_t *drc, 
                drc->drc_guid_to_ds_map = rwa.guid_to_ds_map;
        }
 
-       err = receive_read_payload_and_next_header(&ra, 0, NULL);
-       if (err)
+       uint32_t payloadlen = drc->drc_drr_begin->drr_payloadlen;
+       void *payload = NULL;
+       if (payloadlen != 0)
+               payload = kmem_alloc(payloadlen, KM_SLEEP);
+
+       err = receive_read_payload_and_next_header(&ra, payloadlen, payload);
+       if (err != 0) {
+               if (payloadlen != 0)
+                       kmem_free(payload, payloadlen);
                goto out;
+       }
+       if (payloadlen != 0) {
+               err = nvlist_unpack(payload, payloadlen, &begin_nvl, KM_SLEEP);
+               kmem_free(payload, payloadlen);
+               if (err != 0)
+                       goto out;
+       }
+
+       if (featureflags & DMU_BACKUP_FEATURE_RESUMING) {
+               err = resume_check(&ra, begin_nvl);
+               if (err != 0)
+                       goto out;
+       }
 
        (void) bqueue_init(&rwa.q, zfs_recv_queue_length,
            offsetof(struct receive_record_arg, node));
@@ -2384,6 +2759,7 @@ dmu_recv_stream(dmu_recv_cookie_t *drc, 
        mutex_init(&rwa.mutex, NULL, MUTEX_DEFAULT, NULL);
        rwa.os = ra.os;
        rwa.byteswap = drc->drc_byteswap;
+       rwa.resumable = drc->drc_resumable;
 
        (void) thread_create(NULL, 0, receive_writer_thread, &rwa, 0, curproc,
            TS_RUN, minclsyspri);
@@ -2442,13 +2818,15 @@ dmu_recv_stream(dmu_recv_cookie_t *drc, 
                err = rwa.err;
 
 out:
+       nvlist_free(begin_nvl);
        if ((featureflags & DMU_BACKUP_FEATURE_DEDUP) && (cleanup_fd != -1))
                zfs_onexit_fd_rele(cleanup_fd);
 
        if (err != 0) {
                /*
-                * destroy what we created, so we don't leave it in the
-                * inconsistent restoring state.
+                * Clean up references. If receive is not resumable,
+                * destroy what we created, so we don't leave it in
+                * the inconsistent state.
                 */
                dmu_recv_cleanup_ds(drc);
        }
@@ -2608,6 +2986,20 @@ dmu_recv_end_sync(void *arg, dmu_tx_t *t
 
                dmu_buf_will_dirty(ds->ds_dbuf, tx);
                dsl_dataset_phys(ds)->ds_flags &= ~DS_FLAG_INCONSISTENT;
+               if (dsl_dataset_has_resume_receive_state(ds)) {
+                       (void) zap_remove(dp->dp_meta_objset, ds->ds_object,
+                           DS_FIELD_RESUME_FROMGUID, tx);
+                       (void) zap_remove(dp->dp_meta_objset, ds->ds_object,
+                           DS_FIELD_RESUME_OBJECT, tx);
+                       (void) zap_remove(dp->dp_meta_objset, ds->ds_object,
+                           DS_FIELD_RESUME_OFFSET, tx);
+                       (void) zap_remove(dp->dp_meta_objset, ds->ds_object,
+                           DS_FIELD_RESUME_BYTES, tx);
+                       (void) zap_remove(dp->dp_meta_objset, ds->ds_object,
+                           DS_FIELD_RESUME_TOGUID, tx);
+                       (void) zap_remove(dp->dp_meta_objset, ds->ds_object,
+                           DS_FIELD_RESUME_TONAME, tx);
+               }
        }
        drc->drc_newsnapobj = dsl_dataset_phys(drc->drc_ds)->ds_prev_snap_obj;
        /*

Modified: vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_traverse.c
==============================================================================
--- vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_traverse.c    Wed Oct 14 
11:33:07 2015        (r289311)
+++ vendor-sys/illumos/dist/uts/common/fs/zfs/dmu_traverse.c    Wed Oct 14 
11:36:55 2015        (r289312)
@@ -47,6 +47,7 @@ typedef struct prefetch_data {
        int pd_flags;
        boolean_t pd_cancel;
        boolean_t pd_exited;
+       zbookmark_phys_t pd_resume;
 } prefetch_data_t;
 
 typedef struct traverse_data {
@@ -311,33 +312,30 @@ traverse_visitbp(traverse_data_t *td, co
                    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL, &flags, zb);
                if (err != 0)

*** DIFF OUTPUT TRUNCATED AT 1000 LINES ***
_______________________________________________
[email protected] mailing list
https://lists.freebsd.org/mailman/listinfo/svn-src-all
To unsubscribe, send any mail to "[email protected]"

Reply via email to