Merge pull request ceph#61937 from vshankar/wip-file-block-diff

mds/client: file blockdiff support

Author: kotreshhr

src/client/Client.cc

@@ -9878,6 +9878,144 @@ int Client::readdirplus_r(dir_result_t *d, struct dirent *de,
   return 0;
 }
 
+static void cleanup_state(Client *client, struct scan_state_t *sst)
+{
+  if (sst->fd1 != -1) {
+    client->_close(sst->fd1);
+  }
+  if (sst->fd2 != -1) {
+    client->_close(sst->fd2);
+  }
+  delete sst;
+}
+
+int Client::file_blockdiff_init_state(const char* path1, const char* path2,
+				      const UserPerm &perms, struct scan_state_t **state)
+{
+  ldout(cct, 20) << __func__ << dendl;
+
+  InodeRef inode1, inode2;
+  scan_state_t *sst = new scan_state_t();
+  sst->fd1 = sst->fd2 = -1;
+
+  /*
+   * lets have a constraint that both snapshot paths should be
+   * present - otherwise the caller should do a full copy or a
+   * delete on the path.
+   */
+  int r = open(path1, O_RDONLY, perms, 0);
+  if (r < 0) {
+    return r;
+  }
+  sst->fd1 = r;
+
+  r = open(path2, O_RDONLY, perms, 0);
+  if (r < 0) {
+    cleanup_state(this, sst);
+    return r;
+  }
+  sst->fd2 = r;
+
+  std::unique_lock lock(client_lock);
+  r = get_fd_inode(sst->fd1, &inode1);
+  if (r < 0) {
+    cleanup_state(this, sst);
+    return r;
+  }
+  r = get_fd_inode(sst->fd2, &inode2);
+  if (r < 0) {
+    cleanup_state(this, sst);
+    return r;
+  }
+
+  ldout(cct, 20) << __func__ << ": (snapid1, ino1, size)=(" << inode1->snapid
+		 << "," << std::hex << inode1->ino << std::dec << ","
+		 << inode1->size <<")" << " (snapid2, ino2, size)=("
+		 << inode2->snapid << "," << std::hex << inode2->ino << std::dec
+		 << "," << inode2->size << ")" << dendl;
+  if (inode1->ino != inode2->ino) {
+    cleanup_state(this, sst);
+    return -EINVAL;
+  }
+
+  sst->index = 0;
+  *state = sst;
+  return 0;
+}
+
+int Client::file_blockdiff_finish(struct scan_state_t *state)
+{
+  std::unique_lock lock(client_lock);
+
+  _close(state->fd1);
+  _close(state->fd2);
+  delete state;
+  return 0;
+}
+
+int Client::file_blockdiff(struct scan_state_t *state, const UserPerm &perms,
+			   std::vector<std::pair<uint64_t,uint64_t>> *blocks)
+{
+  RWRef_t mref_reader(mount_state, CLIENT_MOUNTING);
+  if (!mref_reader.is_state_satisfied()) {
+    return -ENOTCONN;
+  }
+
+  ldout(cct, 20) << __func__ << dendl;
+
+  InodeRef inode1;
+  InodeRef inode2;
+
+  std::unique_lock lock(client_lock);
+
+  int r = get_fd_inode(state->fd1, &inode1);
+  if (r < 0) {
+    return r;
+  }
+  r = get_fd_inode(state->fd2, &inode2);
+  if (r < 0) {
+    return r;
+  }
+
+  ceph_assert(inode1->ino == inode2->ino);
+
+  MetaRequest *req = new MetaRequest(CEPH_MDS_OP_FILE_BLOCKDIFF);
+
+  filepath path1, path2;
+  inode1->make_nosnap_relative_path(path1);
+  req->set_filepath(path1);
+
+  inode2->make_nosnap_relative_path(path2);
+  req->set_filepath2(path2);
+  req->set_inode(inode2.get());
+
+  req->head.args.blockdiff.scan_idx = state->index;
+  req->head.args.blockdiff.max_objects =
+    cct->_conf.get_val<uint64_t>("client_file_blockdiff_max_concurrent_object_scans");
+
+  bufferlist bl;
+  r = make_request(req, perms, nullptr, nullptr, -1, &bl);
+  ldout(cct, 10) << __func__ << ": result=" << r << dendl;
+
+  if (r < 0) {
+    return r;
+  }
+
+  BlockDiff block_diff;
+  auto p = bl.cbegin();
+  decode(block_diff, p);
+
+  ldout(cct, 10) << __func__ << ": block_diff=" << block_diff << dendl;
+  if (!block_diff.blocks.empty()) {
+    for (auto &block : block_diff.blocks) {
+      blocks->emplace_back(std::make_pair(block.first, block.second));
+    }
+  }
+
+  state->index = block_diff.scan_idx;
+  return block_diff.rval;
+}
+
 int Client::readdir_snapdiff(dir_result_t* d1, snapid_t snap2,
                              struct dirent* out_de,
                              snapid_t* out_snap)

src/client/Client.h

@@ -147,6 +147,12 @@ class ceph_lock_state_t;
 // ========================================================
 // client interface
 
+struct scan_state_t {
+  int fd1;
+  int fd2;
+  uint64_t index;
+};
+
 struct dir_result_t {
   static const int SHIFT = 28;
   static const int64_t MASK = (1 << SHIFT) - 1;
@@ -368,6 +374,12 @@ class Client : public Dispatcher, public md_config_obs_t {
   int readdir_r(dir_result_t *dirp, struct dirent *de);
   int readdirplus_r(dir_result_t *dirp, struct dirent *de, struct ceph_statx *stx, unsigned want, unsigned flags, Inode **out);
 
+  int file_blockdiff_init_state(const char *path1, const char *path2,
+				const UserPerm &perms, struct scan_state_t **state);
+  int file_blockdiff(struct scan_state_t *state, const UserPerm &perms,
+		     std::vector<std::pair<uint64_t,uint64_t>> *blocks);
+  int file_blockdiff_finish(struct scan_state_t *state);
+
   /*
    * Get the next snapshot delta entry.
    *

src/common/ceph_strings.cc

@@ -328,6 +328,7 @@ const char *ceph_mds_op_name(int op)
 	case CEPH_MDS_OP_QUIESCE_INODE: return "quiesce_inode";
 	case CEPH_MDS_OP_LOCK_PATH: return "lock_path";
 	case CEPH_MDS_OP_UNINLINE_DATA: return "uninline_data";
+        case CEPH_MDS_OP_FILE_BLOCKDIFF: return "blockdiff";
 	}
 	return "???";
 }

src/common/options/mds-client.yaml.in

@@ -589,3 +589,12 @@ options:
   - mds_client
   flags:
   - runtime
+- name: client_file_blockdiff_max_concurrent_object_scans
+  type: uint
+  level: advanced
+  desc: maximum number of concurrent object scans
+  long_desc: Maximum number of concurrent listsnaps operations sent to RADOS.
+  default: 16
+  services:
+  - mds_client
+  min: 1

src/common/options/mds.yaml.in

@@ -1754,3 +1754,12 @@ options:
   services:
   - mds
   min: 4
+- name: mds_file_blockdiff_max_concurrent_object_scans
+  type: uint
+  level: advanced
+  desc: maximum number of concurrent object scans
+  long_desc: Maximum number of concurrent listsnaps operations sent to RADOS.
+  default: 16
+  services:
+  - mds
+  min: 1

src/include/ceph_fs.h

@@ -429,6 +429,7 @@ enum {
 	CEPH_MDS_OP_LSSNAP     = 0x00402,
 	CEPH_MDS_OP_RENAMESNAP = 0x01403,
 	CEPH_MDS_OP_READDIR_SNAPDIFF   = 0x01404,
+	CEPH_MDS_OP_FILE_BLOCKDIFF = 0x01405,
 
 	// internal op
 	CEPH_MDS_OP_FRAGMENTDIR= 0x01500,
@@ -651,6 +652,12 @@ union ceph_mds_request_args {
                 __le32 offset_hash;
 		__le64 snap_other;
 	} __attribute__ ((packed)) snapdiff;
+        struct {
+                // latest scan "pointer"
+                __le64 scan_idx;
+                // how many data objects to scan in one invocation (capped by the mds).
+                __le64 max_objects;
+        } __attribute__ ((packed)) blockdiff;
 } __attribute__ ((packed));
 
 #define CEPH_MDS_REQUEST_HEAD_VERSION	3

src/include/cephfs/libcephfs.h

@@ -648,6 +648,71 @@ struct ceph_snapdiff_info
                                    // doesn't exist in the second snapshot
 };
 
+struct ceph_file_blockdiff_result;
+
+// blockdiff stream handle
+struct ceph_file_blockdiff_info
+{
+  struct ceph_mount_info* cmount;
+  struct ceph_file_blockdiff_result* blockp;
+};
+
+// set of file block diff's
+struct cblock
+{
+  uint64_t offset;
+  uint64_t len;
+};
+struct ceph_file_blockdiff_changedblocks
+{
+  uint64_t num_blocks;
+  struct cblock *b;
+};
+
+/**
+ * Initialize blockdiff stream to get file block deltas.
+ *
+ * @param cmount the ceph mount handle to use for snapdiff retrieval.
+ * @param root_path  root path for snapshots-in-question
+ * @param rel_path subpath under the root to build delta for
+ * @param snap1 the first snapshot name
+ * @param snap2 the second snapshot name
+ * @param out_info resulting blockdiff stream handle to be used for blokdiff results
+                   retrieval via ceph_file_blockdiff().
+ * @returns 0 on success and negative error code otherwise
+ */
+int ceph_file_blockdiff_init(struct ceph_mount_info* cmount,
+                             const char* root_path,
+                             const char* rel_path,
+                             const char* snap1,
+                             const char* snap2,
+                             struct ceph_file_blockdiff_info* out_info);
+
+/**
+ * Get a set of file blockdiff's
+ *
+ * @param info blockdiff stream handle
+ * @param blocks next set of file blockdiff's (offset, length)
+ * @returns 0 or 1 on success and negative error code otherwise
+ */
+int ceph_file_blockdiff(struct ceph_file_blockdiff_info* info,
+                        struct ceph_file_blockdiff_changedblocks* blocks);
+/**
+ * Free blockdiff buffer
+ *
+ * @param blocks file block diff's from ceph_file_blockdiff()
+ * @returns None
+ */
+void ceph_free_file_blockdiff_buffer(struct ceph_file_blockdiff_changedblocks* blocks);
+
+/**
+ * Close blockdiff stream
+ *
+ * @param info blockdiff stream handle
+ * @returns 0 on success and negative error code otherwise
+ */
+int ceph_file_blockdiff_finish(struct ceph_file_blockdiff_info* info);
+
 /**
  * Opens snapdiff stream to get snapshots delta (aka snapdiff).
  *