kclient: mds requests in terms of dentries, not paths

Generate mds request paths when request is sent, so that it reflects
the current state of mds capabilities.  Eventually we can then make
the relative paths based on the current capabilities for that mds..

9 files changed, 253 insertions, 402 deletions

diff --git a/src/include/ceph_fs.h b/src/include/ceph_fs.h
index b17d12e8424..5c2f269f1a8 100644
--- a/src/include/ceph_fs.h
+++ b/src/include/ceph_fs.h
@@ -734,6 +734,52 @@ static inline const char *ceph_mds_op_name(int op)
 	}
 }
 
+union ceph_mds_request_args {
+	struct {
+		__le32 mask;
+	} __attribute__ ((packed)) stat;
+	struct {
+		__le32 mask;
+	} __attribute__ ((packed)) fstat;
+	struct {
+		__le32 frag;
+	} __attribute__ ((packed)) readdir;
+	struct {
+		struct ceph_timespec mtime;
+		struct ceph_timespec atime;
+		struct ceph_timespec ctime;
+		__le32 mask;
+	} __attribute__ ((packed)) utime;
+	struct {
+		__le32 mode;
+	} __attribute__ ((packed)) chmod;
+	struct {
+		__le32 uid;
+		__le32 gid;
+		__le32 mask;
+	} __attribute__ ((packed)) chown;
+	struct {
+		__le32 mode;
+		__le32 rdev;
+	} __attribute__ ((packed)) mknod;
+	struct {
+		__le32 mode;
+	} __attribute__ ((packed)) mkdir;
+	struct {
+		__le32 flags;
+		__le32 mode;
+	} __attribute__ ((packed)) open;
+	struct {
+		__le64 length, old_length;
+	} __attribute__ ((packed)) truncate;
+	struct {
+		__le32 flags;
+	} __attribute__ ((packed)) setxattr;
+	struct {
+		struct ceph_file_layout layout;
+	} __attribute__ ((packed)) setlayout;
+} __attribute__ ((packed));
+
 struct ceph_mds_request_head {
 	ceph_tid_t tid, oldest_client_tid;
 	ceph_epoch_t mdsmap_epoch; /* on client */
@@ -743,52 +789,7 @@ struct ceph_mds_request_head {
 	__le32 op;
 	__le32 caller_uid, caller_gid;
 	__le64 ino;    /* use this ino for openc, mkdir, mknod, etc. */
-
-	union {
-		struct {
-			__le32 mask;
-		} __attribute__ ((packed)) stat;
-		struct {
-			__le32 mask;
-		} __attribute__ ((packed)) fstat;
-		struct {
-			__le32 frag;
-		} __attribute__ ((packed)) readdir;
-		struct {
-			struct ceph_timespec mtime;
-			struct ceph_timespec atime;
-			struct ceph_timespec ctime;
-			__le32 mask;
-		} __attribute__ ((packed)) utime;
-		struct {
-			__le32 mode;
-		} __attribute__ ((packed)) chmod;
-		struct {
-			__le32 uid;
-			__le32 gid;
-			__le32 mask;
-		} __attribute__ ((packed)) chown;
-		struct {
-			__le32 mode;
-			__le32 rdev;
-		} __attribute__ ((packed)) mknod;
-		struct {
-			__le32 mode;
-		} __attribute__ ((packed)) mkdir;
-		struct {
-			__le32 flags;
-			__le32 mode;
-		} __attribute__ ((packed)) open;
-		struct {
-			__le64 length, old_length;
-		} __attribute__ ((packed)) truncate;
-		struct {
-			__le32 flags;
-		} __attribute__ ((packed)) setxattr;
-		struct {
-			struct ceph_file_layout layout;
-		} __attribute__ ((packed)) setlayout;
-	} __attribute__ ((packed)) args;
+	union ceph_mds_request_args args;
 } __attribute__ ((packed));
 
 /* masks for utimes() */
diff --git a/src/kernel/dir.c b/src/kernel/dir.c
index 85292110279..636fed7c02a 100644
--- a/src/kernel/dir.c
+++ b/src/kernel/dir.c
@@ -36,7 +36,8 @@ static int ceph_dentry_revalidate(struct dentry *dentry, struct nameidata *nd);
  * encode hidden .snap dirs as a double /, i.e.
  *   foo/.snap/bar -> foo//bar
  */
-char *ceph_build_path(struct dentry *dentry, int *plen, u64 *base, int min)
+char *ceph_build_path(struct dentry *dentry, int *plen, u64 *base,
+		      int mds)
 {
 	struct dentry *temp;
 	char *path;
@@ -149,10 +150,6 @@ nextfrag:
 	/* do we have the correct frag content buffered? */
 	if (fi->frag != frag || fi->last_readdir == NULL) {
 		struct ceph_mds_request *req;
-		struct ceph_mds_request_head *rhead;
-		char *path;
-		int pathlen;
-		u64 pathbase;
 		int op = ceph_snap(inode) == CEPH_SNAPDIR ?
 			CEPH_MDS_OP_LSSNAP : CEPH_MDS_OP_READDIR;
 
@@ -165,19 +162,15 @@ nextfrag:
 
 		dout(10, "readdir querying mds for ino %llx.%llx frag %x\n",
 		     ceph_vinop(inode), frag);
-		path = ceph_build_path(filp->f_dentry, &pathlen, &pathbase, 1);
-		req = ceph_mdsc_create_request(mdsc, op,
-					       pathbase, path, 0, NULL,
-					       filp->f_dentry, USE_AUTH_MDS);
-		kfree(path);
+		req = ceph_mdsc_create_request(mdsc, op, filp->f_dentry, NULL,
+					       NULL, NULL, USE_AUTH_MDS);
 		if (IS_ERR(req))
 			return PTR_ERR(req);
 		/* hints to request -> mds selection code */
 		req->r_direct_mode = USE_AUTH_MDS;
 		req->r_direct_hash = frag_value(frag);
 		req->r_direct_is_hash = true;
-		rhead = req->r_request->front.iov_base;
-		rhead->args.readdir.frag = cpu_to_le32(frag);
+		req->r_args.readdir.frag = cpu_to_le32(frag);
 		err = ceph_mdsc_do_request(mdsc, NULL, req);
 		if (err < 0) {
 			ceph_mdsc_put_request(req);
@@ -337,9 +330,9 @@ struct dentry *ceph_finish_lookup(struct ceph_mds_request *req,
 			ceph_init_dentry(dentry);
 			if (dentry->d_inode) {
 				d_drop(dentry);
-				req->r_last_dentry = d_alloc(dentry->d_parent,
-							     &dentry->d_name);
-				d_rehash(req->r_last_dentry);
+				req->r_dentry = d_alloc(dentry->d_parent,
+							&dentry->d_name);
+				d_rehash(req->r_dentry);
 			} else {
 				d_add(dentry, NULL);
 			}
@@ -348,8 +341,8 @@ struct dentry *ceph_finish_lookup(struct ceph_mds_request *req,
 	}
 	if (err)
 		dentry = ERR_PTR(err);
-	else if (dentry != req->r_last_dentry)
-		dentry = dget(req->r_last_dentry);   /* we got spliced */
+	else if (dentry != req->r_dentry)
+		dentry = dget(req->r_dentry);   /* we got spliced */
 	else
 		dentry = NULL;
 	return dentry;
@@ -365,10 +358,7 @@ struct dentry *ceph_do_lookup(struct super_block *sb, struct dentry *dentry,
 {
 	struct ceph_client *client = ceph_sb_to_client(sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
-	char *path;
-	int pathlen;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *rhead;
 	struct inode *dir = dentry->d_parent->d_inode;
 	int err;
 
@@ -396,29 +386,12 @@ struct dentry *ceph_do_lookup(struct super_block *sb, struct dentry *dentry,
 	}
 
 	dout(10, "do_lookup %p mask %d\n", dentry, mask);
-	if (on_inode) {
-		/* stat ino directly */
-		WARN_ON(ceph_snap(dentry->d_inode) != CEPH_NOSNAP);
-		req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LSTAT,
-					       ceph_ino(dentry->d_inode), NULL,
-					       0, NULL,
-					       dentry, USE_CAP_MDS);
-	} else {
-		/* build path */
-		u64 pathbase;
-		path = ceph_build_path(dentry, &pathlen, &pathbase, 1);
-		if (IS_ERR(path))
-			return ERR_PTR(PTR_ERR(path));
-		req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LSTAT,
-					       pathbase, path, 0, NULL,
-					       dentry, USE_ANY_MDS);
-		kfree(path);
-	}
+	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LSTAT,
+				       dentry, NULL, NULL, NULL,
+				       on_inode ? USE_CAP_MDS:USE_ANY_MDS);
 	if (IS_ERR(req))
 		return ERR_PTR(PTR_ERR(req));
-	rhead = req->r_request->front.iov_base;
-	rhead->args.stat.mask = cpu_to_le32(mask);
-	req->r_last_dentry = dget(dentry); /* try to use this in fill_trace */
+	req->r_args.stat.mask = cpu_to_le32(mask);
 	req->r_locked_dir = dentry->d_parent->d_inode;  /* by the VFS */
 	err = ceph_mdsc_do_request(mdsc, NULL, req);
 	dentry = ceph_finish_lookup(req, dentry, err);
@@ -462,10 +435,6 @@ static int ceph_mknod(struct inode *dir, struct dentry *dentry,
 	struct ceph_client *client = ceph_sb_to_client(dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *rhead;
-	char *path;
-	int pathlen;
-	u64 pathbase;
 	int err;
 
 	if (ceph_snap(dir) != CEPH_NOSNAP)
@@ -473,21 +442,15 @@ static int ceph_mknod(struct inode *dir, struct dentry *dentry,
 
 	dout(5, "mknod in dir %p dentry %p mode 0%o rdev %d\n",
 	     dir, dentry, mode, rdev);
-	path = ceph_build_path(dentry, &pathlen, &pathbase, 1);
-	if (IS_ERR(path))
-		return PTR_ERR(path);
-	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_MKNOD,
-				       pathbase, path, 0, NULL,
-				       dentry, USE_AUTH_MDS);
-	kfree(path);
+	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_MKNOD, dentry, NULL,
+				       NULL, NULL, USE_AUTH_MDS);
 	if (IS_ERR(req)) {
 		d_drop(dentry);
 		return PTR_ERR(req);
 	}
 	req->r_locked_dir = dir;
-	rhead = req->r_request->front.iov_base;
-	rhead->args.mknod.mode = cpu_to_le32(mode);
-	rhead->args.mknod.rdev = cpu_to_le32(rdev);
+	req->r_args.mknod.mode = cpu_to_le32(mode);
+	req->r_args.mknod.rdev = cpu_to_le32(rdev);
 	if ((ceph_caps_issued(ceph_inode(dir)) & CEPH_CAP_FILE_EXCL) == 0)
 		ceph_release_caps(dir, CEPH_CAP_FILE_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, dir, req);
@@ -542,22 +505,14 @@ static int ceph_symlink(struct inode *dir, struct dentry *dentry,
 	struct ceph_client *client = ceph_sb_to_client(dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
-	char *path;
-	int pathlen;
-	u64 pathbase;
 	int err;
 
 	if (ceph_snap(dir) != CEPH_NOSNAP)
 		return -EROFS;
 
 	dout(5, "symlink in dir %p dentry %p to '%s'\n", dir, dentry, dest);
-	path = ceph_build_path(dentry, &pathlen, &pathbase, 1);
-	if (IS_ERR(path))
-		return PTR_ERR(path);
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_SYMLINK,
-				       pathbase, path, 0, dest,
-				       dentry, USE_AUTH_MDS);
-	kfree(path);
+				       dentry, NULL, NULL, dest, USE_AUTH_MDS);
 	if (IS_ERR(req)) {
 		d_drop(dentry);
 		return PTR_ERR(req);
@@ -577,10 +532,6 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, int mode)
 	struct ceph_client *client = ceph_sb_to_client(dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *rhead;
-	char *path;
-	int pathlen;
-	u64 pathbase;
 	int err;
 	char *snap = NULL;
 	int snaplen;
@@ -601,24 +552,17 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, int mode)
 	} else {
 		dout(5, "mkdir dir %p dn %p mode 0%o\n", dir, dentry, mode);
 	}
-	path = ceph_build_path(pathdentry, &pathlen, &pathbase, 1);
+	req = ceph_mdsc_create_request(mdsc, op, pathdentry, NULL, 
+				       NULL, NULL, USE_AUTH_MDS);
 	if (pathdentry != dentry)
 		dput(pathdentry);
-	if (IS_ERR(path))
-		return PTR_ERR(path);
-	req = ceph_mdsc_create_request(mdsc, op,
-				       pathbase, path, 0, snap,
-				       dentry, USE_AUTH_MDS);
-	kfree(path);
 	if (IS_ERR(req)) {
 		d_drop(dentry);
 		return PTR_ERR(req);
 	}
 
-	req->r_last_dentry = dget(dentry); /* use this dentry in fill_trace */
 	req->r_locked_dir = dir;
-	rhead = req->r_request->front.iov_base;
-	rhead->args.mkdir.mode = cpu_to_le32(mode);
+	req->r_args.mkdir.mode = cpu_to_le32(mode);
 
 	if ((ceph_caps_issued(ceph_inode(dir)) & CEPH_CAP_FILE_EXCL) == 0)
 		ceph_release_caps(dir, CEPH_CAP_FILE_RDCACHE);
@@ -635,9 +579,6 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
 	struct ceph_client *client = ceph_sb_to_client(dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
-	char *oldpath, *path;
-	int oldpathlen, pathlen;
-	u64 oldpathbase, pathbase;
 	int err;
 
 	if (ceph_snap(dir) != CEPH_NOSNAP)
@@ -645,26 +586,13 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
 
 	dout(5, "link in dir %p old_dentry %p dentry %p\n", dir,
 	     old_dentry, dentry);
-	oldpath = ceph_build_path(old_dentry, &oldpathlen, &oldpathbase, 1);
-	if (IS_ERR(oldpath))
-		return PTR_ERR(oldpath);
-	path = ceph_build_path(dentry, &pathlen, &pathbase, 1);
-	if (IS_ERR(path)) {
-		kfree(oldpath);
-		return PTR_ERR(path);
-	}
-	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LINK,
-				       pathbase, path,
-				       oldpathbase, oldpath,
-				       dentry, USE_AUTH_MDS);
-	kfree(oldpath);
-	kfree(path);
+	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LINK, dentry,
+				       old_dentry, NULL, NULL, USE_AUTH_MDS);
 	if (IS_ERR(req)) {
 		d_drop(dentry);
 		return PTR_ERR(req);
 	}
 
-	req->r_last_dentry = dget(dentry); /* use this dentry in fill_trace */
 	req->r_locked_dir = old_dentry->d_inode;
 
 	if ((ceph_caps_issued(ceph_inode(dir)) & CEPH_CAP_FILE_EXCL) == 0)
@@ -693,9 +621,6 @@ static int ceph_unlink(struct inode *dir, struct dentry *dentry)
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct inode *inode = dentry->d_inode;
 	struct ceph_mds_request *req;
-	char *path;
-	int pathlen;
-	u64 pathbase;
 	char *snap = NULL;
 	int snaplen;
 	int err;
@@ -718,15 +643,10 @@ static int ceph_unlink(struct inode *dir, struct dentry *dentry)
 		dout(5, "unlink/rmdir dir %p dn %p inode %p\n",
 		     dir, dentry, inode);
 	}
-	path = ceph_build_path(pathdentry, &pathlen, &pathbase, 1);
+	req = ceph_mdsc_create_request(mdsc, op, pathdentry, NULL, NULL, NULL,
+				       USE_AUTH_MDS);
 	if (pathdentry != dentry)
 		dput(pathdentry);
-	if (IS_ERR(path))
-		return PTR_ERR(path);
-	req = ceph_mdsc_create_request(mdsc, op,
-				       pathbase, path, 0, snap,
-				       dentry, USE_AUTH_MDS);
-	kfree(path);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
 
@@ -747,9 +667,6 @@ static int ceph_rename(struct inode *old_dir, struct dentry *old_dentry,
 	struct ceph_client *client = ceph_sb_to_client(old_dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
-	char *oldpath, *newpath;
-	int oldpathlen, newpathlen;
-	u64 oldpathbase, newpathbase;
 	int err;
 
 	if (ceph_snap(old_dir) != ceph_snap(new_dir))
@@ -760,24 +677,11 @@ static int ceph_rename(struct inode *old_dir, struct dentry *old_dentry,
 
 	dout(5, "dir_rename in dir %p dentry %p to dir %p dentry %p\n",
 	     old_dir, old_dentry, new_dir, new_dentry);
-	oldpath = ceph_build_path(old_dentry, &oldpathlen, &oldpathbase, 1);
-	if (IS_ERR(oldpath))
-		return PTR_ERR(oldpath);
-	newpath = ceph_build_path(new_dentry, &newpathlen, &newpathbase, 1);
-	if (IS_ERR(newpath)) {
-		kfree(oldpath);
-		return PTR_ERR(newpath);
-	}
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_RENAME,
-				       oldpathbase, oldpath,
-				       newpathbase, newpath,
-				       new_dentry, USE_AUTH_MDS);
-	kfree(oldpath);
-	kfree(newpath);
+				       new_dentry, old_dentry, NULL, NULL,
+				       USE_AUTH_MDS);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
-	req->r_old_dentry = dget(old_dentry);
-	req->r_last_dentry = dget(new_dentry);
 	req->r_locked_dir = new_dir;
 	if ((ceph_caps_issued(ceph_inode(old_dir)) & CEPH_CAP_FILE_EXCL) == 0)
 		ceph_release_caps(old_dir, CEPH_CAP_FILE_RDCACHE);
diff --git a/src/kernel/export.c b/src/kernel/export.c
index 956d2bc1c59..5c9783d39b6 100644
--- a/src/kernel/export.c
+++ b/src/kernel/export.c
@@ -79,9 +79,10 @@ static struct dentry *__fh_to_dentry(struct super_block *sb,
 
 	inode = ceph_find_inode(sb, vino);
 	if (!inode) {
+#warning fixme
+		/*
 		struct ceph_mds_request *req;
 		derr(10, "fh_to_dentry %llx.%x -- no inode\n", vino.ino, hash);
-
 		req = ceph_mdsc_create_request(mdsc,
 					       CEPH_MDS_OP_FINDINODE,
 					       len, (char *)fh, 0, NULL,
@@ -90,7 +91,7 @@ static struct dentry *__fh_to_dentry(struct super_block *sb,
 			return ERR_PTR(PTR_ERR(req));
 		err = ceph_mdsc_do_request(mdsc, NULL, req);
 		ceph_mdsc_put_request(req);
-
+		*/
 		inode = ceph_find_inode(sb, vino);
 		if (!inode)
 			return ERR_PTR(err ? err : -ESTALE);
diff --git a/src/kernel/file.c b/src/kernel/file.c
index 16aa67cbf2d..5bb93b612a0 100644
--- a/src/kernel/file.c
+++ b/src/kernel/file.c
@@ -25,11 +25,7 @@ prepare_open_request(struct super_block *sb, struct dentry *dentry,
 {
 	struct ceph_client *client = ceph_sb_to_client(sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
-	u64 pathbase;
-	char *path;
-	int pathlen;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *rhead;
 	int want_auth = USE_ANY_MDS;
 
 	if (flags & (O_WRONLY|O_RDWR|O_CREAT|O_TRUNC))
@@ -37,13 +33,8 @@ prepare_open_request(struct super_block *sb, struct dentry *dentry,
 
 	dout(5, "prepare_open_request dentry %p name '%s' flags %d\n", dentry,
 	     dentry->d_name.name, flags);
-	path = ceph_build_path(dentry, &pathlen, &pathbase, 0);
-	if (IS_ERR(path))
-		return ERR_PTR(PTR_ERR(path));
-	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_OPEN, pathbase, path,
-				       0, NULL,
-				       dentry, want_auth);
-	kfree(path);
+	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_OPEN, dentry, NULL,
+				       NULL, NULL, want_auth);
 	if (IS_ERR(req))
 		goto out;
 	req->r_expected_cap = kmalloc(sizeof(struct ceph_cap), GFP_NOFS);
@@ -52,10 +43,8 @@ prepare_open_request(struct super_block *sb, struct dentry *dentry,
 		return ERR_PTR(-ENOMEM);
 	}
 	req->r_fmode = ceph_flags_to_mode(flags);
-
-	rhead = req->r_request->front.iov_base;
-	rhead->args.open.flags = cpu_to_le32(flags);
-	rhead->args.open.mode = cpu_to_le32(create_mode);
+	req->r_args.open.flags = cpu_to_le32(flags);
+	req->r_args.open.mode = cpu_to_le32(create_mode);
 out:
 	return req;
 }
@@ -183,7 +172,6 @@ struct dentry *ceph_lookup_open(struct inode *dir, struct dentry *dentry,
 	if ((flags & O_CREAT) &&
 	    (ceph_caps_issued(ceph_inode(dir)) & CEPH_CAP_FILE_EXCL) == 0)
 		ceph_release_caps(dir, CEPH_CAP_FILE_RDCACHE);
-	req->r_last_dentry = dget(dentry); /* use this dentry in fill_trace */
 	req->r_locked_dir = dir;           /* caller holds dir->i_mutex */
 	err = ceph_mdsc_do_request(mdsc, parent_inode, req);
 	dentry = ceph_finish_lookup(req, dentry, err);
diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index dc1a9949c3c..3a1139c91de 100644
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -914,23 +914,23 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 		 *  - only if there is no existing dn, and
 		 *  - only if parent is correct
 		 */
-		if (d == rinfo->trace_numd-1 && req->r_last_dentry) {
-			if (!dn && req->r_last_dentry->d_parent == parent) {
-				dn = req->r_last_dentry;
+		if (d == rinfo->trace_numd-1 && req->r_dentry) {
+			if (!dn && req->r_dentry->d_parent == parent) {
+				dn = req->r_dentry;
 				dout(10, "fill_trace provided dn %p '%.*s'\n",
 				     dn, dn->d_name.len, dn->d_name.name);
 				ceph_init_dentry(dn);  /* just in case */
-			} else if (dn == req->r_last_dentry) {
+			} else if (dn == req->r_dentry) {
 				dout(10, "fill_trace matches provided dn %p\n",
 				     dn);
-				dput(req->r_last_dentry);
+				dput(req->r_dentry);
 			} else {
 				dout(10, "fill_trace NOT using provided dn %p "
-				     "(parent %p)\n", req->r_last_dentry,
-				     req->r_last_dentry->d_parent);
-				dput(req->r_last_dentry);
+				     "(parent %p)\n", req->r_dentry,
+				     req->r_dentry->d_parent);
+				dput(req->r_dentry);
 			}
-			req->r_last_dentry = NULL;
+			req->r_dentry = NULL;
 		}
 
 		if (!dn) {
@@ -1094,11 +1094,11 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 		 * we couldn't take i_mutex for this dir, so do not
 		 * lookup or relink any existing dentry.
 		 */
-		if (d == rinfo->trace_numd-1 && req->r_last_dentry) {
-			dn = req->r_last_dentry;
+		if (d == rinfo->trace_numd-1 && req->r_dentry) {
+			dn = req->r_dentry;
 			dout(10, "fill_trace using provided dn %p\n", dn);
 			ceph_init_dentry(dn);
-			req->r_last_dentry = NULL;
+			req->r_dentry = NULL;
 		}
 
 		/* null dentry? */
@@ -1154,9 +1154,9 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 		dput(parent);
 
 	dout(10, "fill_trace done err=%d, last dn %p in %p\n", err, dn, in);
-	if (req->r_last_dentry)
-		dput(req->r_last_dentry);
-	req->r_last_dentry = dn;
+	if (req->r_dentry)
+		dput(req->r_dentry);
+	req->r_dentry = dn;
 	if (req->r_last_inode)
 		iput(req->r_last_inode);
 	req->r_last_inode = in;
@@ -1171,7 +1171,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 			     struct ceph_mds_session *session)
 {
-	struct dentry *parent = req->r_last_dentry;
+	struct dentry *parent = req->r_dentry;
 	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
 	struct qstr dname;
 	struct dentry *dn;
@@ -1468,30 +1468,17 @@ static struct ceph_mds_request *prepare_setattr(struct ceph_mds_client *mdsc,
 						struct dentry *dentry,
 						int ia_valid, int op)
 {
-	char *path;
-	int pathlen;
-	struct ceph_mds_request *req;
-	u64 pathbase;
 	int issued = ceph_caps_issued(ceph_inode(dentry->d_inode));
+	int mode = USE_ANY_MDS;
 
 	if ((ia_valid & ATTR_FILE) ||
-	    (issued & (CEPH_CAP_FILE_WR|CEPH_CAP_FILE_WRBUFFER))) {
-		dout(5, "prepare_setattr dentry %p (inode %llx.%llx)\n", dentry,
-		     ceph_vinop(dentry->d_inode));
-		req = ceph_mdsc_create_request(mdsc, op,
-					       ceph_ino(dentry->d_inode),
-					       "", 0, NULL,
-					       dentry, USE_CAP_MDS);
-	} else {
-		dout(5, "prepare_setattr dentry %p (full path)\n", dentry);
-		path = ceph_build_path(dentry, &pathlen, &pathbase, 0);
-		if (IS_ERR(path))
-			return ERR_PTR(PTR_ERR(path));
-		req = ceph_mdsc_create_request(mdsc, op, pathbase, path, 0,
-					       NULL, dentry, USE_ANY_MDS);
-		kfree(path);
-	}
-	return req;
+	    (issued & (CEPH_CAP_FILE_WR|CEPH_CAP_FILE_WRBUFFER)))
+		mode = USE_CAP_MDS;
+
+	dout(5, "prepare_setattr dentry %p (inode %llx.%llx)\n", dentry,
+	     ceph_vinop(dentry->d_inode));
+	return ceph_mdsc_create_request(mdsc, op, dentry, NULL,
+					NULL, NULL, mode);
 }
 
 static int ceph_setattr_chown(struct dentry *dentry, struct iattr *attr)
@@ -1503,7 +1490,6 @@ static int ceph_setattr_chown(struct dentry *dentry, struct iattr *attr)
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	const unsigned int ia_valid = attr->ia_valid;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *reqh;
 	int err;
 	int mask = 0;
 
@@ -1524,16 +1510,15 @@ static int ceph_setattr_chown(struct dentry *dentry, struct iattr *attr)
 	req = prepare_setattr(mdsc, dentry, ia_valid, CEPH_MDS_OP_LCHOWN);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
-	reqh = req->r_request->front.iov_base;
 	if (ia_valid & ATTR_UID) {
-		reqh->args.chown.uid = cpu_to_le32(attr->ia_uid);
+		req->r_args.chown.uid = cpu_to_le32(attr->ia_uid);
 		mask |= CEPH_CHOWN_UID;
 	}
 	if (ia_valid & ATTR_GID) {
-		reqh->args.chown.gid = cpu_to_le32(attr->ia_gid);
+		req->r_args.chown.gid = cpu_to_le32(attr->ia_gid);
 		mask |= CEPH_CHOWN_GID;
 	}
-	reqh->args.chown.mask = cpu_to_le32(mask);
+	req->r_args.chown.mask = cpu_to_le32(mask);
 	ceph_release_caps(inode, CEPH_CAP_AUTH_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, parent_inode, req);
 	ceph_mdsc_put_request(req);
@@ -1549,7 +1534,6 @@ static int ceph_setattr_chmod(struct dentry *dentry, struct iattr *attr)
 	struct ceph_client *client = ceph_sb_to_client(inode->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *reqh;
 	int err;
 
 	spin_lock(&inode->i_lock);
@@ -1566,8 +1550,7 @@ static int ceph_setattr_chmod(struct dentry *dentry, struct iattr *attr)
 	req = prepare_setattr(mdsc, dentry, attr->ia_valid, CEPH_MDS_OP_LCHMOD);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
-	reqh = req->r_request->front.iov_base;
-	reqh->args.chmod.mode = cpu_to_le32(attr->ia_mode);
+	req->r_args.chmod.mode = cpu_to_le32(attr->ia_mode);
 	ceph_release_caps(inode, CEPH_CAP_AUTH_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, parent_inode, req);
 	ceph_mdsc_put_request(req);
@@ -1584,7 +1567,6 @@ static int ceph_setattr_time(struct dentry *dentry, struct iattr *attr)
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	const unsigned int ia_valid = attr->ia_valid;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *reqh;
 	int err;
 
 	/* if i hold CAP_EXCL, i can change [am]time any way i like */
@@ -1628,15 +1610,14 @@ static int ceph_setattr_time(struct dentry *dentry, struct iattr *attr)
 	req = prepare_setattr(mdsc, dentry, ia_valid, CEPH_MDS_OP_LUTIME);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
-	reqh = req->r_request->front.iov_base;
-	ceph_encode_timespec(&reqh->args.utime.mtime, &attr->ia_mtime);
-	ceph_encode_timespec(&reqh->args.utime.atime, &attr->ia_atime);
+	ceph_encode_timespec(&req->r_args.utime.mtime, &attr->ia_mtime);
+	ceph_encode_timespec(&req->r_args.utime.atime, &attr->ia_atime);
 
-	reqh->args.utime.mask = 0;
+	req->r_args.utime.mask = 0;
 	if (ia_valid & ATTR_ATIME)
-		reqh->args.utime.mask |= cpu_to_le32(CEPH_UTIME_ATIME);
+		req->r_args.utime.mask |= cpu_to_le32(CEPH_UTIME_ATIME);
 	if (ia_valid & ATTR_MTIME)
-		reqh->args.utime.mask |= cpu_to_le32(CEPH_UTIME_MTIME);
+		req->r_args.utime.mask |= cpu_to_le32(CEPH_UTIME_MTIME);
 
 	ceph_release_caps(inode, CEPH_CAP_FILE_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, parent_inode, req);
@@ -1654,7 +1635,6 @@ static int ceph_setattr_size(struct dentry *dentry, struct iattr *attr)
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	const unsigned int ia_valid = attr->ia_valid;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *reqh;
 	int err;
 
 	dout(10, "truncate: ia_size %d i_size %d\n",
@@ -1680,9 +1660,8 @@ static int ceph_setattr_size(struct dentry *dentry, struct iattr *attr)
 	req = prepare_setattr(mdsc, dentry, ia_valid, CEPH_MDS_OP_LTRUNCATE);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
-	reqh = req->r_request->front.iov_base;
-	reqh->args.truncate.length = cpu_to_le64(attr->ia_size);
-	reqh->args.truncate.old_length = cpu_to_le64(inode->i_size);
+	req->r_args.truncate.length = cpu_to_le64(attr->ia_size);
+	req->r_args.truncate.old_length = cpu_to_le64(inode->i_size);
 	//ceph_release_caps(inode, CEPH_CAP_FILE_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, parent_inode, req);
 	ceph_mdsc_put_request(req);
@@ -2051,10 +2030,6 @@ int ceph_setxattr(struct dentry *dentry, const char *name,
 	struct inode *parent_inode = dentry->d_parent->d_inode;
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *rhead;
-	char *path;
-	int pathlen;
-	u64 pathbase;
 	int err;
 	int i, nr_pages;
 	struct page **pages = NULL;
@@ -2090,18 +2065,12 @@ int ceph_setxattr(struct dentry *dentry, const char *name,
 	}
 
 	/* do request */
-	path = ceph_build_path(dentry, &pathlen, &pathbase, 0);
-	if (IS_ERR(path))
-		return PTR_ERR(path);
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LSETXATTR,
-				       pathbase, path, 0, name,
-				       dentry, USE_AUTH_MDS);
-	kfree(path);
+				       dentry, NULL, NULL, NULL, USE_AUTH_MDS);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
 
-	rhead = req->r_request->front.iov_base;
-	rhead->args.setxattr.flags = cpu_to_le32(flags);
+	req->r_args.setxattr.flags = cpu_to_le32(flags);
 
 	req->r_request->pages = pages;
 	req->r_request->nr_pages = nr_pages;
@@ -2128,9 +2097,6 @@ int ceph_removexattr(struct dentry *dentry, const char *name)
 	struct inode *inode = dentry->d_inode;
 	struct inode *parent_inode = dentry->d_parent->d_inode;
 	struct ceph_mds_request *req;
-	char *path;
-	int pathlen;
-	u64 pathbase;
 	int err;
 
 	if (ceph_snap(inode) != CEPH_NOSNAP)
@@ -2143,13 +2109,8 @@ int ceph_removexattr(struct dentry *dentry, const char *name)
 	if (_ceph_match_vir_xattr(name) != NULL)
 		return -EOPNOTSUPP;
 
-	path = ceph_build_path(dentry, &pathlen, &pathbase, 0);
-	if (IS_ERR(path))
-		return PTR_ERR(path);
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LRMXATTR,
-				       pathbase, path, 0, name,
-				       dentry, USE_AUTH_MDS);
-	kfree(path);
+				       dentry, NULL, NULL, NULL, USE_AUTH_MDS);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
 
diff --git a/src/kernel/ioctl.c b/src/kernel/ioctl.c
index dda200ebb17..ee0ffd3b076 100644
--- a/src/kernel/ioctl.c
+++ b/src/kernel/ioctl.c
@@ -30,11 +30,7 @@ static long ceph_ioctl_set_layout(struct file *file, void __user *arg)
 	struct inode *inode = file->f_dentry->d_inode;
 	struct inode *parent_inode = file->f_dentry->d_parent->d_inode;
 	struct ceph_mds_client *mdsc = &ceph_sb_to_client(inode->i_sb)->mdsc;
-	char *path;
-	int pathlen;
 	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *reqh;
-	u64 pathbase;
 	struct ceph_file_layout layout;
 	int err;
 
@@ -42,16 +38,10 @@ static long ceph_ioctl_set_layout(struct file *file, void __user *arg)
 	if (copy_from_user(&layout, arg, sizeof(layout)))
 		return -EFAULT;
 
-	/* set */
-	path = ceph_build_path(file->f_dentry, &pathlen, &pathbase, 0);
-	if (IS_ERR(path))
-		return PTR_ERR(path);
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LSETLAYOUT,
-				       pathbase, path, 0, NULL,
-				       file->f_dentry, USE_ANY_MDS);
-	kfree(path);
-	reqh = req->r_request->front.iov_base;
-	reqh->args.setlayout.layout = layout;
+				       file->f_dentry, NULL, NULL, NULL,
+				       USE_AUTH_MDS);
+	req->r_args.setlayout.layout = layout;
 	ceph_release_caps(inode, CEPH_CAP_FILE_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, parent_inode, req);
 	ceph_mdsc_put_request(req);
diff --git a/src/kernel/mds_client.c b/src/kernel/mds_client.c
index 6cb37ccdfc1..7753a59505e 100644
--- a/src/kernel/mds_client.c
+++ b/src/kernel/mds_client.c
@@ -396,12 +396,10 @@ void ceph_mdsc_put_request(struct ceph_mds_request *req)
 			ceph_msg_put(req->r_reply);
 			destroy_reply_info(&req->r_reply_info);
 		}
-		if (req->r_direct_dentry)
-			dput(req->r_direct_dentry);
 		if (req->r_last_inode)
 			iput(req->r_last_inode);
-		if (req->r_last_dentry)
-			dput(req->r_last_dentry);
+		if (req->r_dentry)
+			dput(req->r_dentry);
 		if (req->r_old_dentry)
 			dput(req->r_old_dentry);
 		kfree(req->r_expected_cap);
@@ -426,24 +424,6 @@ static struct ceph_mds_request *__lookup_request(struct ceph_mds_client *mdsc,
 }
 
 /*
- * allocate and initialize a new request.  mostly zeroed.
- */
-static struct ceph_mds_request *new_request(struct ceph_msg *msg)
-{
-	struct ceph_mds_request *req;
-
-	req = kzalloc(sizeof(*req), GFP_NOFS);
-	req->r_request = msg;
-	req->r_started = jiffies;
-	req->r_resend_mds = -1;
-	req->r_fmode = -1;
-	atomic_set(&req->r_ref, 1);  /* one for request_tree, one for caller */
-	init_completion(&req->r_completion);
-	init_completion(&req->r_safe_completion);
-	return req;
-}
-
-/*
  * Register an in-flight request, and assign a tid in msg request header.
  *
  * Called under mdsc->mutex.
@@ -452,10 +432,9 @@ static void __register_request(struct ceph_mds_client *mdsc,
 			       struct inode *listener,
 			       struct ceph_mds_request *req)
 {
-	struct ceph_mds_request_head *head = req->r_request->front.iov_base;
 	struct ceph_inode_info *ci;
+
 	req->r_tid = ++mdsc->last_tid;
-	head->tid = cpu_to_le64(req->r_tid);
 	dout(30, "__register_request %p tid %lld\n", req, req->r_tid);
 	ceph_mdsc_get_request(req);
 	radix_tree_insert(&mdsc->request_tree, req->r_tid, (void *)req);
@@ -503,7 +482,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
 	int mds = -1;
 	u32 hash = req->r_direct_hash;
 	bool is_hash = req->r_direct_is_hash;
-	struct dentry *dentry = req->r_direct_dentry;
+	struct dentry *dentry = req->r_dentry;
 	struct ceph_inode_info *ci;
 	int mode = req->r_direct_mode;
 
@@ -865,82 +844,33 @@ void ceph_mdsc_flushed_all_caps(struct ceph_mds_client *mdsc,
 }
 
 
-
 /*
- * create an mds request and message.
- *
- * slight hacky weirdness: if op is a FINDINODE, ino1 is the _length_
- * of path1, and path1 isn't null terminated (it's an nfs filehandle
- * fragment).  path2 is not used in that case.
+ * Create an mds request.
  */
 struct ceph_mds_request *
 ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op,
-			 u64 ino1, const char *path1,
-			 u64 ino2, const char *path2,
-			 struct dentry *ref, int mode)
+			 struct dentry *dentry, struct dentry *old_dentry,
+			 const char *path1, const char *path2, int mode)
 {
-	struct ceph_msg *msg;
-	struct ceph_mds_request *req;
-	struct ceph_mds_request_head *head;
-	void *p, *end;
-	int pathlen;
-
-	if (op == CEPH_MDS_OP_FINDINODE) {
-		pathlen = sizeof(u32) + ino1*sizeof(struct ceph_inopath_item);
-	} else {
-		pathlen = 2*(sizeof(ino1) + sizeof(u32));
-		if (path1)
-			pathlen += strlen(path1);
-		if (path2)
-			pathlen += strlen(path2);
-	}
+	struct ceph_mds_request *req = kzalloc(sizeof(*req), GFP_NOFS);
 
-	msg = ceph_msg_new(CEPH_MSG_CLIENT_REQUEST,
-			   sizeof(struct ceph_mds_request_head) + pathlen,
-			   0, 0, NULL);
-	if (IS_ERR(msg))
-		return ERR_PTR(PTR_ERR(msg));
-	req = new_request(msg);
-	if (IS_ERR(req)) {
-		ceph_msg_put(msg);
-		return req;
-	}
-	head = msg->front.iov_base;
-	p = msg->front.iov_base + sizeof(*head);
-	end = msg->front.iov_base + msg->front.iov_len;
+	if (!req)
+		return ERR_PTR(-ENOMEM);
+	req->r_started = jiffies;
+	req->r_resend_mds = -1;
+	req->r_fmode = -1;
+	atomic_set(&req->r_ref, 1);  /* one for request_tree, one for caller */
+	init_completion(&req->r_completion);
+	init_completion(&req->r_safe_completion);
 
-	/* dentry used to direct mds request? */
-	req->r_direct_dentry = dget(ref);
+	req->r_op = op;
+	if (dentry)
+		req->r_dentry = dget(dentry);
+	if (old_dentry)
+		req->r_old_dentry = dget(old_dentry);
+	req->r_path1 = path1;
+	req->r_path2 = path2;	
 	req->r_direct_mode = mode;
-
-	/* tid, oldest_client_tid, retry_attempt set later. */
-	head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
-	head->num_fwd = 0;
-	head->mds_wants_replica_in_dirino = 0;
-	head->op = cpu_to_le32(op);
-	head->caller_uid = cpu_to_le32(current->fsuid);
-	head->caller_gid = cpu_to_le32(current->fsgid);
-	memset(&head->args, 0, sizeof(head->args));
-
-	/* encode paths */
-	if (op == CEPH_MDS_OP_FINDINODE) {
-		ceph_encode_32(&p, ino1);
-		memcpy(p, path1, ino1 * sizeof(struct ceph_inopath_item));
-		p += ino1 * sizeof(struct ceph_inopath_item);
-	} else {
-		ceph_encode_filepath(&p, end, ino1, path1);
-		ceph_encode_filepath(&p, end, ino2, path2);
-		if (path1)
-			dout(10, "create_request path1 %llx/%s\n",
-			     ino1, path1);
-		if (path2)
-			dout(10, "create_request path2 %llx/%s\n",
-			     ino2, path2);
-	}
-	dout_flag(10, DOUT_MASK_PROTOCOL, "create_request op %d=%s -> %p\n", op,
-	     ceph_mds_op_name(op), req);
-
-	BUG_ON(p != end);
 	return req;
 }
 
@@ -961,19 +891,89 @@ static u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
 /*
  * called under mdsc->mutex
  */
-static void __prepare_send_request(struct ceph_mds_client *mdsc,
-				   struct ceph_mds_request *req)
+static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
+					       struct ceph_mds_request *req,
+					       int mds)
 {
-	struct ceph_mds_request_head *rhead;
+	struct ceph_msg *msg;
+	struct ceph_mds_request_head *head;
+	const char *path1 = req->r_path1;
+	const char *path2 = req->r_path2;
+	u64 ino1 = 1, ino2 = 0;
+	int pathlen1 = 0, pathlen2 = 0;
+	void *p, *end;
+
+	if (path1)
+		pathlen1 = strlen(path1);
+	else
+		path1 = ceph_build_path(req->r_dentry, &pathlen1, &ino1, mds);
+	if (path2)
+		pathlen2 = strlen(path2);
+	else if (req->r_old_dentry)
+		path2 = ceph_build_path(req->r_old_dentry, &pathlen2, &ino2,
+					mds);
+
+	msg = ceph_msg_new(CEPH_MSG_CLIENT_REQUEST,
+			   sizeof(*head) + pathlen1 + pathlen2 +
+			   2*(sizeof(u32)+sizeof(u64)), 0, 0, NULL);
+	if (IS_ERR(msg))
+		goto out;
+
+	head = msg->front.iov_base;
+	p = msg->front.iov_base + sizeof(*head);
+	end = msg->front.iov_base + msg->front.iov_len;
+
+	head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
+	head->num_fwd = 0;
+	head->mds_wants_replica_in_dirino = 0;
+	head->op = cpu_to_le32(req->r_op);
+	head->caller_uid = cpu_to_le32(current->fsuid);
+	head->caller_gid = cpu_to_le32(current->fsgid);
+	head->args = req->r_args;
+
+	ceph_encode_filepath(&p, end, ino1, path1);
+	ceph_encode_filepath(&p, end, ino2, path2);
+	if (path1)
+		dout(10, "create_request_message path1 %llx/%s\n",
+		     ino1, path1);
+	if (path2)
+		dout(10, "create_request_message path2 %llx/%s\n",
+		     ino2, path2);
+
+ 	BUG_ON(p != end);
+
+out:
+	return msg;
+}
 
-	/* if there are other references on this message, e.g., if we are
-	 * told to forward it and the previous copy is still in flight, dup
-	 * it. */
-	req->r_request = ceph_msg_maybe_dup(req->r_request);
+/*
+ * called under mdsc->mutex
+ */
+static int __prepare_send_request(struct ceph_mds_client *mdsc,
+				  struct ceph_mds_request *req,
+				  int mds)
+{
+	struct ceph_mds_request_head *rhead;
+	struct ceph_msg *msg;
 
 	req->r_attempts++;
+	dout(10, "prepare_send_request %p tid %lld %s (attempt %d)\n", req,
+	     req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);
+
+	if (req->r_request) {
+		ceph_msg_put(req->r_request);
+		req->r_request = NULL;
+	}
+	msg = create_request_message(mdsc, req, mds);
+	if (IS_ERR(msg)) {
+		req->r_reply = ERR_PTR(PTR_ERR(msg));
+		complete(&req->r_completion);
+		return -PTR_ERR(msg);
+	}
+	req->r_request = msg;
 
-	rhead = req->r_request->front.iov_base;
+	rhead = msg->front.iov_base;
+	rhead->tid = cpu_to_le64(req->r_tid);
 	rhead->retry_attempt = cpu_to_le32(req->r_attempts - 1);
 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
 	rhead->num_fwd = cpu_to_le32(req->r_num_fwd);
@@ -982,6 +982,7 @@ static void __prepare_send_request(struct ceph_mds_client *mdsc,
 		rhead->ino = cpu_to_le64(ceph_ino(req->r_last_inode));
 	else
 		rhead->ino = 0;
+	return 0;
 }
 
 /*
@@ -1035,14 +1036,14 @@ static int __do_request(struct ceph_mds_client *mdsc,
 	if (req->r_request_started == 0)   /* note request start time */
 		req->r_request_started = jiffies;
 
-	__prepare_send_request(mdsc, req);
-
-	mutex_unlock(&mdsc->mutex);
-	ceph_msg_get(req->r_request);
-	ceph_send_msg_mds(mdsc, req->r_request, mds);
-	mutex_lock(&mdsc->mutex);
+	err = __prepare_send_request(mdsc, req, mds);
+	if (!err) {
+		mutex_unlock(&mdsc->mutex);
+		ceph_msg_get(req->r_request);
+		ceph_send_msg_mds(mdsc, req->r_request, mds);
+		mutex_lock(&mdsc->mutex);
+	}
 
-	err = 0;
 out_session:
 	ceph_put_mds_session(session);
 out:
@@ -1462,15 +1463,18 @@ static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
 {
 	struct list_head *p, *n;
 	struct ceph_mds_request *req;
+	int err;
 
 	dout(10, "replay_unsafe_requests mds%d\n", session->s_mds);
 
 	mutex_lock(&mdsc->mutex);
 	list_for_each_safe(p, n, &session->s_unsafe) {
 		req = list_entry(p, struct ceph_mds_request, r_unsafe_item);
-		__prepare_send_request(mdsc, req);
-		ceph_msg_get(req->r_request);
-		ceph_send_msg_mds(mdsc, req->r_request, session->s_mds);
+		err = __prepare_send_request(mdsc, req, session->s_mds);
+		if (!err) {
+			ceph_msg_get(req->r_request);
+			ceph_send_msg_mds(mdsc, req->r_request, session->s_mds);
+		}
 	}
 	mutex_unlock(&mdsc->mutex);
 }
diff --git a/src/kernel/mds_client.h b/src/kernel/mds_client.h
index b65bb7a8506..697bd6bcfd6 100644
--- a/src/kernel/mds_client.h
+++ b/src/kernel/mds_client.h
@@ -138,7 +138,14 @@ enum {
  * an in-flight mds request
  */
 struct ceph_mds_request {
-	u64             r_tid;      /* transaction id */
+	u64 r_tid;                   /* transaction id */
+
+	int r_op;
+	struct dentry *r_dentry;
+	struct dentry *r_old_dentry; /* rename from or link from */
+	const char *r_path1, *r_path2;
+	union ceph_mds_request_args r_args;
+
 	struct ceph_msg  *r_request;  /* original request */
 	struct ceph_msg  *r_reply;
 	struct ceph_mds_reply_info_parsed r_reply_info;
@@ -152,7 +159,6 @@ struct ceph_mds_request {
 	struct list_head r_wait;
 
 	/* for choosing which mds to send this request to */
-	struct dentry *r_direct_dentry;
 	int r_direct_mode;
 	u32 r_direct_hash;      /* choose dir frag based on this dentry hash */
 	bool r_direct_is_hash;  /* true if r_direct_hash is valid */
@@ -164,8 +170,6 @@ struct ceph_mds_request {
 	 * mds response.  also used to feed a VFS-provided dentry into
 	 * the reply handler */
 	struct inode     *r_last_inode;
-	struct dentry    *r_last_dentry;
-	struct dentry    *r_old_dentry;   /* for rename */
 	struct ceph_cap  *r_expected_cap; /* preallocate cap if we expect one */
 	int               r_fmode;        /* file mode, if expecting cap */
 	struct ceph_mds_session *r_session;
@@ -267,9 +271,9 @@ extern void ceph_mdsc_lease_release(struct ceph_mds_client *mdsc,
 
 extern struct ceph_mds_request *
 ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op,
-			 u64 ino1, const char *path1,
-			 u64 ino2, const char *path2,
-			 struct dentry *ref, int want_auth);
+			 struct dentry *dentry, struct dentry *old_dentry,
+			 const char *path1, const char *path2,
+			 int mode);
 extern int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 				struct inode *listener,
 				struct ceph_mds_request *req);
diff --git a/src/kernel/super.c b/src/kernel/super.c
index 5e2d4415d3f..e5205218356 100644
--- a/src/kernel/super.c
+++ b/src/kernel/super.c
@@ -720,24 +720,22 @@ static struct dentry *open_root_dentry(struct ceph_client *client,
 {
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req = NULL;
-	struct ceph_mds_request_head *reqhead;
 	int err;
 	struct dentry *root;
 
 	/* open dir */
 	dout(30, "open_root_inode opening '%s'\n", path);
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LSTAT,
-				       1, path, 0, NULL,
-				       NULL, USE_ANY_MDS);
+				       NULL, NULL, path, NULL,
+				       USE_ANY_MDS);
 	if (IS_ERR(req))
 		return ERR_PTR(PTR_ERR(req));
 	req->r_started = started;
 	req->r_timeout = client->mount_args.mount_timeout * HZ;
-	reqhead = req->r_request->front.iov_base;
-	reqhead->args.stat.mask = cpu_to_le32(CEPH_STAT_CAP_INODE);
+	req->r_args.stat.mask = cpu_to_le32(CEPH_STAT_CAP_INODE);
 	err = ceph_mdsc_do_request(mdsc, NULL, req);
 	if (err == 0) {
-		root = req->r_last_dentry;
+		root = req->r_dentry;
 		dget(root);
 		dout(30, "open_root_inode success, root dentry is %p\n", root);
 	} else