[PATCH 0/5] ceph: fix performance regression due to ceph_d_revalidate fixes

[PATCH 0/5] ceph: fix performance regression due to ceph_d_revalidate fixes

[Jeff Layton]

Zheng mentioned that he had gotten a report that testing with vdbench on
a v4.8 kernel showed a large increase in the number of GETATTR calls
being made. Since we had changed the d_revalidate code to use a GETATTR
call and not set r_locked_dir in the request, the dentry lease wasn't
getting updated from the traces, even when the MDS had granted one.

This set does a bit of cleanup and bugfixing in ceph_fill_trace, and
sets it up to update the dentry lease even when r_locked_dir isn't set
in the request.

I was able to reproduce this by hand, and this seems to fix the issue.

Jeff Layton (5):
  ceph: don't update_dentry_lease unless we actually got one
  ceph: vet the parent inode before updating dentry lease
  ceph: clean up r_aborted handling in ceph_fill_trace
  ceph: call update_dentry_lease even when r_locked dir is not set
  ceph: do a LOOKUP in d_revalidate instead of GETATTR

 fs/ceph/dir.c   |  4 ++--
 fs/ceph/inode.c | 51 ++++++++++++++++++++++++++++++++++-----------------
 2 files changed, 36 insertions(+), 19 deletions(-)

-- 
2.9.3

[PATCH 1/5] ceph: don't update_dentry_lease unless we actually got one

[Jeff Layton]

This if block updates the dentry lease even in the case where
the MDS didn't grant one.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 4926265f4223..ad1ca46a058d 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1319,8 +1319,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 				ceph_dir_clear_ordered(dir);
 				dout("d_delete %p\n", dn);
 				d_delete(dn);
-			} else {
-				if (have_lease && d_unhashed(dn))
+			} else if (have_lease) {
+				if (d_unhashed(dn))
 					d_add(dn, NULL);
 				update_dentry_lease(dn, rinfo->dlease,
 						    session,
-- 
2.9.3

[PATCH 2/5] ceph: vet the parent inode before updating dentry lease

[Jeff Layton]

In a later patch, we're going to need to allow ceph_fill_trace to
update the dentry's lease when the parent is not locked. This is
potentially racy though -- by the time we get around to processing the
trace, the parent may have already changed.

Change update_dentry_lease to take a ceph_vino pointer and use that to
ensure that the dentry's parent still matches it before updating the
lease.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 24 ++++++++++++++++--------
 1 file changed, 16 insertions(+), 8 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index ad1ca46a058d..2b4863e82c1d 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1016,7 +1016,8 @@ static int fill_inode(struct inode *inode, struct page *locked_page,
 static void update_dentry_lease(struct dentry *dentry,
 				struct ceph_mds_reply_lease *lease,
 				struct ceph_mds_session *session,
-				unsigned long from_time)
+				unsigned long from_time,
+				struct ceph_vino *dir_vino)
 {
 	struct ceph_dentry_info *di = ceph_dentry(dentry);
 	long unsigned duration = le32_to_cpu(lease->duration_ms);
@@ -1031,6 +1032,9 @@ static void update_dentry_lease(struct dentry *dentry,
 	/* make lease_rdcache_gen match directory */
 	dir = d_inode(dentry->d_parent);
 
+	if (!ceph_ino_compare(dir, dir_vino))
+		goto out_unlock;
+
 	/* only track leases on regular dentries */
 	if (ceph_snap(dir) != CEPH_NOSNAP)
 		goto out_unlock;
@@ -1264,10 +1268,12 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 		BUG_ON(!dn);
 		BUG_ON(!dir);
 		BUG_ON(d_inode(dn->d_parent) != dir);
-		BUG_ON(ceph_ino(dir) !=
-		       le64_to_cpu(rinfo->diri.in->ino));
-		BUG_ON(ceph_snap(dir) !=
-		       le64_to_cpu(rinfo->diri.in->snapid));
+
+		vino.ino = le64_to_cpu(rinfo->diri.in->ino);
+		vino.snap = le64_to_cpu(rinfo->diri.in->snapid);
+
+		BUG_ON(ceph_ino(dir) != vino.ino);
+		BUG_ON(ceph_snap(dir) != vino.snap);
 
 		/* do we have a lease on the whole dir? */
 		have_dir_cap =
@@ -1324,7 +1330,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 					d_add(dn, NULL);
 				update_dentry_lease(dn, rinfo->dlease,
 						    session,
-						    req->r_request_started);
+						    req->r_request_started,
+						    &vino);
 			}
 			goto done;
 		}
@@ -1349,7 +1356,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 
 		if (have_lease)
 			update_dentry_lease(dn, rinfo->dlease, session,
-					    req->r_request_started);
+					    req->r_request_started, &vino);
 		dout(" final dn %p\n", dn);
 	} else if (!req->r_aborted &&
 		   (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
@@ -1617,8 +1624,9 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 
 		ceph_dentry(dn)->offset = rde->offset;
 
+		vino = ceph_vino(d_inode(parent));
 		update_dentry_lease(dn, rde->lease, req->r_session,
-				    req->r_request_started);
+				    req->r_request_started, &vino);
 
 		if (err == 0 && skipped == 0 && cache_ctl.index >= 0) {
 			ret = fill_readdir_cache(d_inode(parent), dn,
-- 
2.9.3

[PATCH 3/5] ceph: clean up r_aborted handling in ceph_fill_trace

[Jeff Layton]

When r_aborted is true, then we can just "goto done".

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 2b4863e82c1d..b5d4594b434d 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1247,12 +1247,14 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 		}
 	}
 
+	if (req->r_aborted)
+		goto done;
+
 	/*
 	 * ignore null lease/binding on snapdir ENOENT, or else we
 	 * will have trouble splicing in the virtual snapdir later
 	 */
-	if (rinfo->head->is_dentry && !req->r_aborted &&
-	    req->r_locked_dir &&
+	if (rinfo->head->is_dentry && req->r_locked_dir &&
 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
 					       fsc->mount_options->snapdir_name,
 					       req->r_dentry->d_name.len))) {
@@ -1358,9 +1360,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 			update_dentry_lease(dn, rinfo->dlease, session,
 					    req->r_request_started, &vino);
 		dout(" final dn %p\n", dn);
-	} else if (!req->r_aborted &&
-		   (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
-		    req->r_op == CEPH_MDS_OP_MKSNAP)) {
+	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
+		   req->r_op == CEPH_MDS_OP_MKSNAP) {
 		struct dentry *dn = req->r_dentry;
 		struct inode *dir = req->r_locked_dir;
 
-- 
2.9.3

[PATCH 4/5] ceph: call update_dentry_lease even when r_locked dir is not set

[Jeff Layton]

We don't really require that the parent be locked in order to update the
lease on a dentry. Lease info is protected by the d_lock. In the event
that the parent is not locked in ceph_fill_trace, and we have both
parent and target info, go ahead and update the dentry lease.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index b5d4594b434d..bb2413d711ce 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1378,6 +1378,16 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 			goto done;
 		}
 		req->r_dentry = dn;  /* may have spliced */
+	} else if (rinfo->head->is_dentry && rinfo->head->is_target) {
+		if ((le32_to_cpu(rinfo->diri.in->cap.caps) & CEPH_CAP_FILE_SHARED) ||
+		    le32_to_cpu(rinfo->dlease->duration_ms)) {
+			vino.ino = le64_to_cpu(rinfo->diri.in->ino);
+			vino.snap = le64_to_cpu(rinfo->diri.in->snapid);
+			update_dentry_lease(req->r_dentry, rinfo->dlease,
+				session, req->r_request_started, &vino);
+		} else {
+			dout("%s: no dentry lease or dir cap\n", __func__);
+		}
 	}
 done:
 	dout("fill_trace done err=%d\n", err);
-- 
2.9.3

[PATCH 5/5] ceph: do a LOOKUP in d_revalidate instead of GETATTR

[Jeff Layton]

In commit c3f4688a08f (ceph: don't set req->r_locked_dir in
ceph_d_revalidate), we changed the code to do a GETATTR instead of a
LOOKUP as the parent info isn't strictly necessary to revalidate the
dentry. What we missed there though is that in order to update the lease
on the dentry after revalidating it, we _do_ need parent info.

Change ceph_d_revalidate back to doing a LOOKUP instead of a GETATTR so
that we can get the parent info in order to update the lease from
ceph_fill_trace. Note too that because we don't set r_locked_dir in this
codepath, we must also  remove the WARN_ON_ONCE from ceph_fill_trace or
it will pop here.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/dir.c   | 4 ++--
 fs/ceph/inode.c | 2 --
 2 files changed, 2 insertions(+), 4 deletions(-)

diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index d4385563b70a..33da70514a95 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -1237,11 +1237,11 @@ static int ceph_d_revalidate(struct dentry *dentry, unsigned int flags)
 			return -ECHILD;
 
 		op = ceph_snap(dir) == CEPH_SNAPDIR ?
-			CEPH_MDS_OP_LOOKUPSNAP : CEPH_MDS_OP_GETATTR;
+			CEPH_MDS_OP_LOOKUPSNAP : CEPH_MDS_OP_LOOKUP;
 		req = ceph_mdsc_create_request(mdsc, op, USE_ANY_MDS);
 		if (!IS_ERR(req)) {
 			req->r_dentry = dget(dentry);
-			req->r_num_caps = op == CEPH_MDS_OP_GETATTR ? 1 : 2;
+			req->r_num_caps = 2;
 
 			mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED;
 			if (ceph_security_xattr_wanted(dir))
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index bb2413d711ce..870a0cfb8647 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1175,8 +1175,6 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 					 &req->r_caps_reservation);
 			if (err < 0)
 				goto done;
-		} else {
-			WARN_ON_ONCE(1);
 		}
 
 		if (dir && req->r_op == CEPH_MDS_OP_LOOKUPNAME) {
-- 
2.9.3

[PATCH v2 00/13] ceph: fix performance regression due to ceph_d_revalidate fixes

[Jeff Layton]

v2: fix a hang that would occasionally occur under fsstress

This is the second version of this patchset, and is a bit more of an
overhaul. I revised the set because I was seeing fsstress hangs.

Zheng mentioned that he had gotten a report that testing with vdbench on
a v4.8 kernel showed a large increase in the number of GETATTR calls
being made. Since we had changed the d_revalidate code to use a GETATTR
call and not set r_locked_dir in the request, the dentry lease wasn't
getting updated from the traces, even when the MDS had granted one.

We need to use a LOOKUP when revalidating dentries so we can update
the dentry lease if things look ok. That currently depends however on
having r_locked_dir set to the parent directory.

In d_revalidate though, we may very well not hold the parent's i_rwsem.
Manipulating the dcache is not allowed without it, but we can update the
inode and the dentry leases.

The basic approach here is to separate r_locked_dir into an inode
pointer representing the parent, and a flag indicating whether we
know that it's locked (note that it _may_ still be locked even if
that flag isn't set).

In ceph_fill_trace, we can then do all of the parts that don't require a
locked parent whenever the parent pointer is set, but avoid the
dcache manipulation when it isn't.

With this, dentry leases are again being updated as a result of
d_revalidate lookups. The switch to using a flags fields instead
of bools also shrinks the request size slightly. If everyone hates
that, I can just plop in another bool, but I think this gives us
more room to expand with other flags as needed.

Jeff Layton (13):
  ceph: add new r_req_flags field to ceph_mds_request
  ceph: move r_aborted flag into r_req_flags
  ceph: move r_got_unsafe flag into r_req_flags
  ceph: move r_got_safe flag into r_req_flags
  ceph: move r_got_result into r_req_flags
  ceph: move r_did_prepopulate into r_req_flags
  ceph: remove "Debugging hook" from ceph_fill_trace
  ceph: don't need session pointer to ceph_fill_trace
  ceph: add a new flag to indicate whether parent is locked
  ceph: don't update_dentry_lease unless we actually got one
  ceph: vet the parent inode before updating dentry lease
  ceph: call update_dentry_lease even when r_locked dir is not set
  ceph: do a LOOKUP in d_revalidate instead of GETATTR

 fs/ceph/debugfs.c    |   2 +-
 fs/ceph/dir.c        |  30 +++++++++------
 fs/ceph/export.c     |   3 +-
 fs/ceph/file.c       |   3 +-
 fs/ceph/inode.c      | 104 ++++++++++++++++++++++-----------------------------
 fs/ceph/mds_client.c |  73 +++++++++++++++++++-----------------
 fs/ceph/mds_client.h |  15 +++++---
 fs/ceph/super.h      |   3 +-
 8 files changed, 118 insertions(+), 115 deletions(-)

-- 
2.9.3

[PATCH v2 01/13] ceph: add new r_req_flags field to ceph_mds_request

[Jeff Layton]

...and start moving bool flags into it.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/dir.c        | 2 +-
 fs/ceph/mds_client.c | 2 +-
 fs/ceph/mds_client.h | 4 +++-
 3 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index d4385563b70a..04fa4ae3deca 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -371,7 +371,7 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
 		/* hints to request -> mds selection code */
 		req->r_direct_mode = USE_AUTH_MDS;
 		req->r_direct_hash = ceph_frag_value(frag);
-		req->r_direct_is_hash = true;
+		set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
 		if (fi->last_name) {
 			req->r_path2 = kstrdup(fi->last_name, GFP_KERNEL);
 			if (!req->r_path2) {
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 176512960b14..1f2ef02832d9 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -705,7 +705,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
 	int mode = req->r_direct_mode;
 	int mds = -1;
 	u32 hash = req->r_direct_hash;
-	bool is_hash = req->r_direct_is_hash;
+	bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
 
 	/*
 	 * is there a specific mds we should try?  ignore hint if we have
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 3c6f77b7bb02..a58cacccc986 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -205,6 +205,9 @@ struct ceph_mds_request {
 	struct inode *r_locked_dir; /* dir (if any) i_mutex locked by vfs */
 	struct inode *r_target_inode;       /* resulting inode */
 
+#define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
+	unsigned long	r_req_flags;
+
 	struct mutex r_fill_mutex;
 
 	union ceph_mds_request_args r_args;
@@ -216,7 +219,6 @@ struct ceph_mds_request {
 	/* for choosing which mds to send this request to */
 	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 */
 
 	/* data payload is used for xattr ops */
 	struct ceph_pagelist *r_pagelist;
-- 
2.9.3

[PATCH v2 02/13] ceph: move r_aborted flag into r_req_flags

[Jeff Layton]

...and simplify the handling of it in ceph_fill_trace.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c      | 17 +++++++++--------
 fs/ceph/mds_client.c |  8 ++++----
 fs/ceph/mds_client.h |  2 +-
 3 files changed, 14 insertions(+), 13 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 4926265f4223..bd2e94a78057 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1233,8 +1233,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 
 		err = fill_inode(in, req->r_locked_page, &rinfo->targeti, NULL,
 				session, req->r_request_started,
-				(!req->r_aborted && rinfo->head->result == 0) ?
-				req->r_fmode : -1,
+				(!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
+				rinfo->head->result == 0) ?  req->r_fmode : -1,
 				&req->r_caps_reservation);
 		if (err < 0) {
 			pr_err("fill_inode badness %p %llx.%llx\n",
@@ -1243,12 +1243,14 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 		}
 	}
 
+	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
+		goto done;
+
 	/*
 	 * ignore null lease/binding on snapdir ENOENT, or else we
 	 * will have trouble splicing in the virtual snapdir later
 	 */
-	if (rinfo->head->is_dentry && !req->r_aborted &&
-	    req->r_locked_dir &&
+	if (rinfo->head->is_dentry && req->r_locked_dir &&
 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
 					       fsc->mount_options->snapdir_name,
 					       req->r_dentry->d_name.len))) {
@@ -1351,9 +1353,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 			update_dentry_lease(dn, rinfo->dlease, session,
 					    req->r_request_started);
 		dout(" final dn %p\n", dn);
-	} else if (!req->r_aborted &&
-		   (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
-		    req->r_op == CEPH_MDS_OP_MKSNAP)) {
+	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
+		   req->r_op == CEPH_MDS_OP_MKSNAP) {
 		struct dentry *dn = req->r_dentry;
 		struct inode *dir = req->r_locked_dir;
 
@@ -1478,7 +1479,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 	u32 fpos_offset;
 	struct ceph_readdir_cache_control cache_ctl = {};
 
-	if (req->r_aborted)
+	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
 		return readdir_prepopulate_inodes_only(req, session);
 
 	if (rinfo->hash_order && req->r_path2) {
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 1f2ef02832d9..a5156b6a0aed 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -2115,7 +2115,7 @@ static int __do_request(struct ceph_mds_client *mdsc,
 	int err = 0;
 
 	if (req->r_err || req->r_got_result) {
-		if (req->r_aborted)
+		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
 			__unregister_request(mdsc, req);
 		goto out;
 	}
@@ -2331,7 +2331,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 		 */
 		mutex_lock(&req->r_fill_mutex);
 		req->r_err = err;
-		req->r_aborted = true;
+		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
 		mutex_unlock(&req->r_fill_mutex);
 
 		if (req->r_locked_dir &&
@@ -2538,7 +2538,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	}
 out_err:
 	mutex_lock(&mdsc->mutex);
-	if (!req->r_aborted) {
+	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
 		if (err) {
 			req->r_err = err;
 		} else {
@@ -2587,7 +2587,7 @@ static void handle_forward(struct ceph_mds_client *mdsc,
 		goto out;  /* dup reply? */
 	}
 
-	if (req->r_aborted) {
+	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
 		dout("forward tid %llu aborted, unregistering\n", tid);
 		__unregister_request(mdsc, req);
 	} else if (fwd_seq <= req->r_num_fwd) {
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index a58cacccc986..3da20955d5e6 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -206,6 +206,7 @@ struct ceph_mds_request {
 	struct inode *r_target_inode;       /* resulting inode */
 
 #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
+#define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
 	unsigned long	r_req_flags;
 
 	struct mutex r_fill_mutex;
@@ -236,7 +237,6 @@ struct ceph_mds_request {
 	struct ceph_mds_reply_info_parsed r_reply_info;
 	struct page *r_locked_page;
 	int r_err;
-	bool r_aborted;
 
 	unsigned long r_timeout;  /* optional.  jiffies, 0 is "wait forever" */
 	unsigned long r_started;  /* start time to measure timeout against */
-- 
2.9.3

[PATCH v2 03/13] ceph: move r_got_unsafe flag into r_req_flags

[Jeff Layton]

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/debugfs.c    |  2 +-
 fs/ceph/mds_client.c | 23 +++++++++++++----------
 fs/ceph/mds_client.h |  3 ++-
 3 files changed, 16 insertions(+), 12 deletions(-)

diff --git a/fs/ceph/debugfs.c b/fs/ceph/debugfs.c
index 39ff678e567f..f2ae393e2c31 100644
--- a/fs/ceph/debugfs.c
+++ b/fs/ceph/debugfs.c
@@ -70,7 +70,7 @@ static int mdsc_show(struct seq_file *s, void *p)
 
 		seq_printf(s, "%s", ceph_mds_op_name(req->r_op));
 
-		if (req->r_got_unsafe)
+		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 			seq_puts(s, "\t(unsafe)");
 		else
 			seq_puts(s, "\t");
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index a5156b6a0aed..ab9b03be1b3f 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -644,13 +644,15 @@ static void __unregister_request(struct ceph_mds_client *mdsc,
 
 	erase_request(&mdsc->request_tree, req);
 
-	if (req->r_unsafe_dir && req->r_got_unsafe) {
+	if (req->r_unsafe_dir  &&
+	    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 		struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
 		spin_lock(&ci->i_unsafe_lock);
 		list_del_init(&req->r_unsafe_dir_item);
 		spin_unlock(&ci->i_unsafe_lock);
 	}
-	if (req->r_target_inode && req->r_got_unsafe) {
+	if (req->r_target_inode &&
+	    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
 		spin_lock(&ci->i_unsafe_lock);
 		list_del_init(&req->r_unsafe_target_item);
@@ -2042,7 +2044,7 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
 	dout("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_got_unsafe) {
+	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 		void *p;
 		/*
 		 * Replay.  Do not regenerate message (and rebuild
@@ -2091,7 +2093,7 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
 
 	rhead = msg->front.iov_base;
 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
-	if (req->r_got_unsafe)
+	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 		flags |= CEPH_MDS_FLAG_REPLAY;
 	if (req->r_locked_dir)
 		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
@@ -2245,7 +2247,7 @@ static void kick_requests(struct ceph_mds_client *mdsc, int mds)
 	while (p) {
 		req = rb_entry(p, struct ceph_mds_request, r_node);
 		p = rb_next(p);
-		if (req->r_got_unsafe)
+		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 			continue;
 		if (req->r_attempts > 0)
 			continue; /* only new requests */
@@ -2409,7 +2411,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	}
 
 	/* dup? */
-	if ((req->r_got_unsafe && !head->safe) ||
+	if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) ||
 	    (req->r_got_safe && head->safe)) {
 		pr_warn("got a dup %s reply on %llu from mds%d\n",
 			   head->safe ? "safe" : "unsafe", tid, mds);
@@ -2458,7 +2460,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 		req->r_got_safe = true;
 		__unregister_request(mdsc, req);
 
-		if (req->r_got_unsafe) {
+		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 			/*
 			 * We already handled the unsafe response, now do the
 			 * cleanup.  No need to examine the response; the MDS
@@ -2476,7 +2478,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 			goto out;
 		}
 	} else {
-		req->r_got_unsafe = true;
+		set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags);
 		list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
 		if (req->r_unsafe_dir) {
 			struct ceph_inode_info *ci =
@@ -2530,7 +2532,8 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	if (realm)
 		ceph_put_snap_realm(mdsc, realm);
 
-	if (err == 0 && req->r_got_unsafe && req->r_target_inode) {
+	if (err == 0 && req->r_target_inode &&
+	    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
 		spin_lock(&ci->i_unsafe_lock);
 		list_add_tail(&req->r_unsafe_target_item, &ci->i_unsafe_iops);
@@ -2762,7 +2765,7 @@ static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
 	while (p) {
 		req = rb_entry(p, struct ceph_mds_request, r_node);
 		p = rb_next(p);
-		if (req->r_got_unsafe)
+		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 			continue;
 		if (req->r_attempts == 0)
 			continue; /* only old requests */
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 3da20955d5e6..e56fc25ef1bf 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -207,6 +207,7 @@ struct ceph_mds_request {
 
 #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
 #define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
+#define CEPH_MDS_R_GOT_UNSAFE		(3) /* got an unsafe reply */
 	unsigned long	r_req_flags;
 
 	struct mutex r_fill_mutex;
@@ -264,7 +265,7 @@ struct ceph_mds_request {
 	ceph_mds_request_callback_t r_callback;
 	ceph_mds_request_wait_callback_t r_wait_for_completion;
 	struct list_head  r_unsafe_item;  /* per-session unsafe list item */
-	bool		  r_got_unsafe, r_got_safe, r_got_result;
+	bool		  r_got_safe, r_got_result;
 
 	bool              r_did_prepopulate;
 	long long	  r_dir_release_cnt;
-- 
2.9.3

[PATCH v2 04/13] ceph: move r_got_safe flag into r_req_flags

[Jeff Layton]

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/mds_client.c | 6 +++---
 fs/ceph/mds_client.h | 3 ++-
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index ab9b03be1b3f..68d9f172fd02 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -2412,13 +2412,13 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 
 	/* dup? */
 	if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) ||
-	    (req->r_got_safe && head->safe)) {
+	    (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags) && head->safe)) {
 		pr_warn("got a dup %s reply on %llu from mds%d\n",
 			   head->safe ? "safe" : "unsafe", tid, mds);
 		mutex_unlock(&mdsc->mutex);
 		goto out;
 	}
-	if (req->r_got_safe) {
+	if (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags)) {
 		pr_warn("got unsafe after safe on %llu from mds%d\n",
 			   tid, mds);
 		mutex_unlock(&mdsc->mutex);
@@ -2457,7 +2457,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 
 
 	if (head->safe) {
-		req->r_got_safe = true;
+		set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags);
 		__unregister_request(mdsc, req);
 
 		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index e56fc25ef1bf..f5d2b4d3a336 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -208,6 +208,7 @@ struct ceph_mds_request {
 #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
 #define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
 #define CEPH_MDS_R_GOT_UNSAFE		(3) /* got an unsafe reply */
+#define CEPH_MDS_R_GOT_SAFE		(4) /* got a safe reply */
 	unsigned long	r_req_flags;
 
 	struct mutex r_fill_mutex;
@@ -265,7 +266,7 @@ struct ceph_mds_request {
 	ceph_mds_request_callback_t r_callback;
 	ceph_mds_request_wait_callback_t r_wait_for_completion;
 	struct list_head  r_unsafe_item;  /* per-session unsafe list item */
-	bool		  r_got_safe, r_got_result;
+	bool		  r_got_result;
 
 	bool              r_did_prepopulate;
 	long long	  r_dir_release_cnt;
-- 
2.9.3

[PATCH v2 05/13] ceph: move r_got_result into r_req_flags

[Jeff Layton]

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/mds_client.c | 8 ++++----
 fs/ceph/mds_client.h | 2 +-
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 68d9f172fd02..fc7ae1e0f691 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -2116,7 +2116,7 @@ static int __do_request(struct ceph_mds_client *mdsc,
 	int mds = -1;
 	int err = 0;
 
-	if (req->r_err || req->r_got_result) {
+	if (req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
 		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
 			__unregister_request(mdsc, req);
 		goto out;
@@ -2321,7 +2321,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 	mutex_lock(&mdsc->mutex);
 
 	/* only abort if we didn't race with a real reply */
-	if (req->r_got_result) {
+	if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
 		err = le32_to_cpu(req->r_reply_info.head->result);
 	} else if (err < 0) {
 		dout("aborted request %lld with %d\n", req->r_tid, err);
@@ -2546,7 +2546,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 			req->r_err = err;
 		} else {
 			req->r_reply =  ceph_msg_get(msg);
-			req->r_got_result = true;
+			set_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
 		}
 	} else {
 		dout("reply arrived after request %lld was aborted\n", tid);
@@ -2600,7 +2600,7 @@ static void handle_forward(struct ceph_mds_client *mdsc,
 		/* resend. forward race not possible; mds would drop */
 		dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
 		BUG_ON(req->r_err);
-		BUG_ON(req->r_got_result);
+		BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags));
 		req->r_attempts = 0;
 		req->r_num_fwd = fwd_seq;
 		req->r_resend_mds = next_mds;
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index f5d2b4d3a336..bc36e2c4fe52 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -209,6 +209,7 @@ struct ceph_mds_request {
 #define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
 #define CEPH_MDS_R_GOT_UNSAFE		(3) /* got an unsafe reply */
 #define CEPH_MDS_R_GOT_SAFE		(4) /* got a safe reply */
+#define CEPH_MDS_R_GOT_RESULT		(5) /* got a result */
 	unsigned long	r_req_flags;
 
 	struct mutex r_fill_mutex;
@@ -266,7 +267,6 @@ struct ceph_mds_request {
 	ceph_mds_request_callback_t r_callback;
 	ceph_mds_request_wait_callback_t r_wait_for_completion;
 	struct list_head  r_unsafe_item;  /* per-session unsafe list item */
-	bool		  r_got_result;
 
 	bool              r_did_prepopulate;
 	long long	  r_dir_release_cnt;
-- 
2.9.3

[PATCH v2 06/13] ceph: move r_did_prepopulate into r_req_flags

[Jeff Layton]

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/dir.c        | 2 +-
 fs/ceph/inode.c      | 2 +-
 fs/ceph/mds_client.h | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index 04fa4ae3deca..fba974dac48b 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -417,7 +417,7 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
 		fi->frag = frag;
 		fi->last_readdir = req;
 
-		if (req->r_did_prepopulate) {
+		if (test_bit(CEPH_MDS_R_DID_PREPOPULATE, &req->r_req_flags)) {
 			fi->readdir_cache_idx = req->r_readdir_cache_idx;
 			if (fi->readdir_cache_idx < 0) {
 				/* preclude from marking dir ordered */
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index bd2e94a78057..31827f3dcef0 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1633,7 +1633,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 	}
 out:
 	if (err == 0 && skipped == 0) {
-		req->r_did_prepopulate = true;
+		set_bit(CEPH_MDS_R_DID_PREPOPULATE, &req->r_req_flags);
 		req->r_readdir_cache_idx = cache_ctl.index;
 	}
 	ceph_readdir_cache_release(&cache_ctl);
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index bc36e2c4fe52..409b0e3c3b7a 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -210,6 +210,7 @@ struct ceph_mds_request {
 #define CEPH_MDS_R_GOT_UNSAFE		(3) /* got an unsafe reply */
 #define CEPH_MDS_R_GOT_SAFE		(4) /* got a safe reply */
 #define CEPH_MDS_R_GOT_RESULT		(5) /* got a result */
+#define CEPH_MDS_R_DID_PREPOPULATE	(6) /* prepopulated readdir */
 	unsigned long	r_req_flags;
 
 	struct mutex r_fill_mutex;
@@ -268,7 +269,6 @@ struct ceph_mds_request {
 	ceph_mds_request_wait_callback_t r_wait_for_completion;
 	struct list_head  r_unsafe_item;  /* per-session unsafe list item */
 
-	bool              r_did_prepopulate;
 	long long	  r_dir_release_cnt;
 	long long	  r_dir_ordered_cnt;
 	int		  r_readdir_cache_idx;
-- 
2.9.3

[PATCH v2 07/13] ceph: remove "Debugging hook" from ceph_fill_trace

[Jeff Layton]

Keeping around commented out code is just asking for it to bitrot and
makes viewing the code under cscope more confusing.  If
we really need this, then we can revert this patch and put it under a
Kconfig option.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 34 ----------------------------------
 1 file changed, 34 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 31827f3dcef0..b93e8eb3c2ee 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1120,40 +1120,6 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 	dout("fill_trace %p is_dentry %d is_target %d\n", req,
 	     rinfo->head->is_dentry, rinfo->head->is_target);
 
-#if 0
-	/*
-	 * Debugging hook:
-	 *
-	 * If we resend completed ops to a recovering mds, we get no
-	 * trace.  Since that is very rare, pretend this is the case
-	 * to ensure the 'no trace' handlers in the callers behave.
-	 *
-	 * Fill in inodes unconditionally to avoid breaking cap
-	 * invariants.
-	 */
-	if (rinfo->head->op & CEPH_MDS_OP_WRITE) {
-		pr_info("fill_trace faking empty trace on %lld %s\n",
-			req->r_tid, ceph_mds_op_name(rinfo->head->op));
-		if (rinfo->head->is_dentry) {
-			rinfo->head->is_dentry = 0;
-			err = fill_inode(req->r_locked_dir,
-					 &rinfo->diri, rinfo->dirfrag,
-					 session, req->r_request_started, -1);
-		}
-		if (rinfo->head->is_target) {
-			rinfo->head->is_target = 0;
-			ininfo = rinfo->targeti.in;
-			vino.ino = le64_to_cpu(ininfo->ino);
-			vino.snap = le64_to_cpu(ininfo->snapid);
-			in = ceph_get_inode(sb, vino);
-			err = fill_inode(in, &rinfo->targeti, NULL,
-					 session, req->r_request_started,
-					 req->r_fmode);
-			iput(in);
-		}
-	}
-#endif
-
 	if (!rinfo->head->is_target && !rinfo->head->is_dentry) {
 		dout("fill_trace reply is empty!\n");
 		if (rinfo->head->result == 0 && req->r_locked_dir)
-- 
2.9.3

[PATCH v2 08/13] ceph: don't need session pointer to ceph_fill_trace

[Jeff Layton]

Just get it from r_session since that's what's always passed in.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c      | 4 ++--
 fs/ceph/mds_client.c | 2 +-
 fs/ceph/super.h      | 3 +--
 3 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index b93e8eb3c2ee..e54ba03d3df7 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1108,9 +1108,9 @@ static struct dentry *splice_dentry(struct dentry *dn, struct inode *in)
  *
  * Called with snap_rwsem (read).
  */
-int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
-		    struct ceph_mds_session *session)
+int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 {
+	struct ceph_mds_session *session = req->r_session;
 	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
 	struct inode *in = NULL;
 	struct ceph_vino vino;
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index fc7ae1e0f691..ccf75a3260e8 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -2518,7 +2518,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	/* insert trace into our cache */
 	mutex_lock(&req->r_fill_mutex);
 	current->journal_info = req;
-	err = ceph_fill_trace(mdsc->fsc->sb, req, req->r_session);
+	err = ceph_fill_trace(mdsc->fsc->sb, req);
 	if (err == 0) {
 		if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
 				    req->r_op == CEPH_MDS_OP_LSSNAP))
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 6477264bfc7e..950170136be9 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -764,8 +764,7 @@ extern void ceph_fill_file_time(struct inode *inode, int issued,
 				u64 time_warp_seq, struct timespec *ctime,
 				struct timespec *mtime, struct timespec *atime);
 extern int ceph_fill_trace(struct super_block *sb,
-			   struct ceph_mds_request *req,
-			   struct ceph_mds_session *session);
+			   struct ceph_mds_request *req);
 extern int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 				    struct ceph_mds_session *session);
 
-- 
2.9.3

[PATCH v2 09/13] ceph: add a new flag to indicate whether parent is locked

[Jeff Layton]

struct ceph_mds_request has an r_locked_dir pointer, which is set to
indicate the parent inode and that its i_rwsem is locked.  In some
critical places, we need to be able to indicate the parent inode to the
request handling code, even when its i_rwsem may not be locked.

Most of the code that operates on r_locked_dir doesn't require that the
i_rwsem be locked. We only really need it to handle manipulation of the
dcache. The rest (filling of the inode, updating dentry leases, etc.)
already has its own locking.

Add a new r_req_flags bit that indicates whether the parent is locked
when doing the request, and rename the pointer to "r_parent". For now,
all the places that set r_parent also set this flag, but that will
change in a later patch.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/dir.c        | 21 ++++++++++++++-------
 fs/ceph/export.c     |  3 ++-
 fs/ceph/file.c       |  3 ++-
 fs/ceph/inode.c      | 11 ++++++-----
 fs/ceph/mds_client.c | 24 ++++++++++++------------
 fs/ceph/mds_client.h |  3 ++-
 6 files changed, 38 insertions(+), 27 deletions(-)

diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index fba974dac48b..b889a3b555cf 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -752,7 +752,8 @@ static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
 		mask |= CEPH_CAP_XATTR_SHARED;
 	req->r_args.getattr.mask = cpu_to_le32(mask);
 
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	err = ceph_mdsc_do_request(mdsc, NULL, req);
 	err = ceph_handle_snapdir(req, dentry, err);
 	dentry = ceph_finish_lookup(req, dentry, err);
@@ -813,7 +814,8 @@ static int ceph_mknod(struct inode *dir, struct dentry *dentry,
 	}
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_args.mknod.mode = cpu_to_le32(mode);
 	req->r_args.mknod.rdev = cpu_to_le32(rdev);
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
@@ -864,7 +866,8 @@ static int ceph_symlink(struct inode *dir, struct dentry *dentry,
 		ceph_mdsc_put_request(req);
 		goto out;
 	}
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
@@ -913,7 +916,8 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, umode_t mode)
 
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_args.mkdir.mode = cpu_to_le32(mode);
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
@@ -957,7 +961,8 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
 	req->r_old_dentry = dget(old_dentry);
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
 	/* release LINK_SHARED on source inode (mds will lock it) */
@@ -1023,7 +1028,8 @@ static int ceph_unlink(struct inode *dir, struct dentry *dentry)
 	}
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
 	req->r_inode_drop = drop_caps_for_unlink(inode);
@@ -1066,7 +1072,8 @@ static int ceph_rename(struct inode *old_dir, struct dentry *old_dentry,
 	req->r_num_caps = 2;
 	req->r_old_dentry = dget(old_dentry);
 	req->r_old_dentry_dir = old_dir;
-	req->r_locked_dir = new_dir;
+	req->r_parent = new_dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_old_dentry_drop = CEPH_CAP_FILE_SHARED;
 	req->r_old_dentry_unless = CEPH_CAP_FILE_EXCL;
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
diff --git a/fs/ceph/export.c b/fs/ceph/export.c
index 180bbef760f2..e8f11fa565c5 100644
--- a/fs/ceph/export.c
+++ b/fs/ceph/export.c
@@ -207,7 +207,8 @@ static int ceph_get_name(struct dentry *parent, char *name,
 	req->r_inode = d_inode(child);
 	ihold(d_inode(child));
 	req->r_ino2 = ceph_vino(d_inode(parent));
-	req->r_locked_dir = d_inode(parent);
+	req->r_parent = d_inode(parent);
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_num_caps = 2;
 	err = ceph_mdsc_do_request(mdsc, NULL, req);
 
diff --git a/fs/ceph/file.c b/fs/ceph/file.c
index 19419eb87f38..a91239c4b31c 100644
--- a/fs/ceph/file.c
+++ b/fs/ceph/file.c
@@ -310,7 +310,8 @@ int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
                mask |= CEPH_CAP_XATTR_SHARED;
        req->r_args.open.mask = cpu_to_le32(mask);
 
-	req->r_locked_dir = dir;           /* caller holds dir->i_mutex */
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	err = ceph_mdsc_do_request(mdsc,
 				   (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
 				   req);
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index e54ba03d3df7..cb94ec7961d2 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1122,13 +1122,13 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 
 	if (!rinfo->head->is_target && !rinfo->head->is_dentry) {
 		dout("fill_trace reply is empty!\n");
-		if (rinfo->head->result == 0 && req->r_locked_dir)
+		if (rinfo->head->result == 0 && req->r_parent)
 			ceph_invalidate_dir_request(req);
 		return 0;
 	}
 
 	if (rinfo->head->is_dentry) {
-		struct inode *dir = req->r_locked_dir;
+		struct inode *dir = req->r_parent;
 
 		if (dir) {
 			err = fill_inode(dir, NULL,
@@ -1216,7 +1216,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 	 * ignore null lease/binding on snapdir ENOENT, or else we
 	 * will have trouble splicing in the virtual snapdir later
 	 */
-	if (rinfo->head->is_dentry && req->r_locked_dir &&
+	if (rinfo->head->is_dentry &&
+	    test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags) &&
 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
 					       fsc->mount_options->snapdir_name,
 					       req->r_dentry->d_name.len))) {
@@ -1225,7 +1226,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 		 * mknod symlink mkdir  : null -> new inode
 		 * unlink               : linked -> null
 		 */
-		struct inode *dir = req->r_locked_dir;
+		struct inode *dir = req->r_parent;
 		struct dentry *dn = req->r_dentry;
 		bool have_dir_cap, have_lease;
 
@@ -1322,7 +1323,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
 		   req->r_op == CEPH_MDS_OP_MKSNAP) {
 		struct dentry *dn = req->r_dentry;
-		struct inode *dir = req->r_locked_dir;
+		struct inode *dir = req->r_parent;
 
 		/* fill out a snapdir LOOKUPSNAP dentry */
 		BUG_ON(!dn);
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index ccf75a3260e8..52521f339745 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -547,8 +547,8 @@ void ceph_mdsc_release_request(struct kref *kref)
 		ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
 		iput(req->r_inode);
 	}
-	if (req->r_locked_dir)
-		ceph_put_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
+	if (req->r_parent)
+		ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
 	iput(req->r_target_inode);
 	if (req->r_dentry)
 		dput(req->r_dentry);
@@ -735,7 +735,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
 
 		rcu_read_lock();
 		parent = req->r_dentry->d_parent;
-		dir = req->r_locked_dir ? : d_inode_rcu(parent);
+		dir = req->r_parent ? : d_inode_rcu(parent);
 
 		if (!dir || dir->i_sb != mdsc->fsc->sb) {
 			/*  not this fs or parent went negative */
@@ -1894,7 +1894,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
 	int ret;
 
 	ret = set_request_path_attr(req->r_inode, req->r_dentry,
-			      req->r_locked_dir, req->r_path1, req->r_ino1.ino,
+			      req->r_parent, req->r_path1, req->r_ino1.ino,
 			      &path1, &pathlen1, &ino1, &freepath1);
 	if (ret < 0) {
 		msg = ERR_PTR(ret);
@@ -1956,7 +1956,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
 		      mds, req->r_inode_drop, req->r_inode_unless, 0);
 	if (req->r_dentry_drop)
 		releases += ceph_encode_dentry_release(&p, req->r_dentry,
-				req->r_locked_dir, mds, req->r_dentry_drop,
+				req->r_parent, mds, req->r_dentry_drop,
 				req->r_dentry_unless);
 	if (req->r_old_dentry_drop)
 		releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
@@ -2095,14 +2095,14 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
 	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 		flags |= CEPH_MDS_FLAG_REPLAY;
-	if (req->r_locked_dir)
+	if (req->r_parent)
 		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
 	rhead->flags = cpu_to_le32(flags);
 	rhead->num_fwd = req->r_num_fwd;
 	rhead->num_retry = req->r_attempts - 1;
 	rhead->ino = 0;
 
-	dout(" r_locked_dir = %p\n", req->r_locked_dir);
+	dout(" r_parent = %p\n", req->r_parent);
 	return 0;
 }
 
@@ -2282,11 +2282,11 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 
 	dout("do_request on %p\n", req);
 
-	/* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
+	/* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
 	if (req->r_inode)
 		ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
-	if (req->r_locked_dir)
-		ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
+	if (req->r_parent)
+		ceph_get_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
 	if (req->r_old_dentry_dir)
 		ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
 				  CEPH_CAP_PIN);
@@ -2336,7 +2336,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
 		mutex_unlock(&req->r_fill_mutex);
 
-		if (req->r_locked_dir &&
+		if (req->r_parent &&
 		    (req->r_op & CEPH_MDS_OP_WRITE))
 			ceph_invalidate_dir_request(req);
 	} else {
@@ -2355,7 +2355,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
  */
 void ceph_invalidate_dir_request(struct ceph_mds_request *req)
 {
-	struct inode *inode = req->r_locked_dir;
+	struct inode *inode = req->r_parent;
 
 	dout("invalidate_dir_request %p (complete, lease(s))\n", inode);
 
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 409b0e3c3b7a..ac0475a2daa7 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -202,7 +202,7 @@ struct ceph_mds_request {
 	char *r_path1, *r_path2;
 	struct ceph_vino r_ino1, r_ino2;
 
-	struct inode *r_locked_dir; /* dir (if any) i_mutex locked by vfs */
+	struct inode *r_parent;		    /* parent dir inode */
 	struct inode *r_target_inode;       /* resulting inode */
 
 #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
@@ -211,6 +211,7 @@ struct ceph_mds_request {
 #define CEPH_MDS_R_GOT_SAFE		(4) /* got a safe reply */
 #define CEPH_MDS_R_GOT_RESULT		(5) /* got a result */
 #define CEPH_MDS_R_DID_PREPOPULATE	(6) /* prepopulated readdir */
+#define CEPH_MDS_R_PARENT_LOCKED	(7) /* is r_parent->i_rwsem wlocked? */
 	unsigned long	r_req_flags;
 
 	struct mutex r_fill_mutex;
-- 
2.9.3

[PATCH v2 10/13] ceph: don't update_dentry_lease unless we actually got one

[Jeff Layton]

This if block updates the dentry lease even in the case where
the MDS didn't grant one.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index cb94ec7961d2..da6e222dff84 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1288,8 +1288,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 				ceph_dir_clear_ordered(dir);
 				dout("d_delete %p\n", dn);
 				d_delete(dn);
-			} else {
-				if (have_lease && d_unhashed(dn))
+			} else if (have_lease) {
+				if (d_unhashed(dn))
 					d_add(dn, NULL);
 				update_dentry_lease(dn, rinfo->dlease,
 						    session,
-- 
2.9.3

[PATCH v2 11/13] ceph: vet the parent inode before updating dentry lease

[Jeff Layton]

In a later patch, we're going to need to allow ceph_fill_trace to
update the dentry's lease when the parent is not locked. This is
potentially racy though -- by the time we get around to processing the
trace, the parent may have already changed.

Change update_dentry_lease to take a ceph_vino pointer and use that to
ensure that the dentry's parent still matches it before updating the
lease.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 24 ++++++++++++++++--------
 1 file changed, 16 insertions(+), 8 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index da6e222dff84..15e042a8d71f 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1016,7 +1016,8 @@ static int fill_inode(struct inode *inode, struct page *locked_page,
 static void update_dentry_lease(struct dentry *dentry,
 				struct ceph_mds_reply_lease *lease,
 				struct ceph_mds_session *session,
-				unsigned long from_time)
+				unsigned long from_time,
+				struct ceph_vino *dir_vino)
 {
 	struct ceph_dentry_info *di = ceph_dentry(dentry);
 	long unsigned duration = le32_to_cpu(lease->duration_ms);
@@ -1031,6 +1032,9 @@ static void update_dentry_lease(struct dentry *dentry,
 	/* make lease_rdcache_gen match directory */
 	dir = d_inode(dentry->d_parent);
 
+	if (!ceph_ino_compare(dir, dir_vino))
+		goto out_unlock;
+
 	/* only track leases on regular dentries */
 	if (ceph_snap(dir) != CEPH_NOSNAP)
 		goto out_unlock;
@@ -1233,10 +1237,12 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 		BUG_ON(!dn);
 		BUG_ON(!dir);
 		BUG_ON(d_inode(dn->d_parent) != dir);
-		BUG_ON(ceph_ino(dir) !=
-		       le64_to_cpu(rinfo->diri.in->ino));
-		BUG_ON(ceph_snap(dir) !=
-		       le64_to_cpu(rinfo->diri.in->snapid));
+
+		vino.ino = le64_to_cpu(rinfo->diri.in->ino);
+		vino.snap = le64_to_cpu(rinfo->diri.in->snapid);
+
+		BUG_ON(ceph_ino(dir) != vino.ino);
+		BUG_ON(ceph_snap(dir) != vino.snap);
 
 		/* do we have a lease on the whole dir? */
 		have_dir_cap =
@@ -1293,7 +1299,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 					d_add(dn, NULL);
 				update_dentry_lease(dn, rinfo->dlease,
 						    session,
-						    req->r_request_started);
+						    req->r_request_started,
+						    &vino);
 			}
 			goto done;
 		}
@@ -1318,7 +1325,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 
 		if (have_lease)
 			update_dentry_lease(dn, rinfo->dlease, session,
-					    req->r_request_started);
+					    req->r_request_started, &vino);
 		dout(" final dn %p\n", dn);
 	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
 		   req->r_op == CEPH_MDS_OP_MKSNAP) {
@@ -1585,8 +1592,9 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 
 		ceph_dentry(dn)->offset = rde->offset;
 
+		vino = ceph_vino(d_inode(parent));
 		update_dentry_lease(dn, rde->lease, req->r_session,
-				    req->r_request_started);
+				    req->r_request_started, &vino);
 
 		if (err == 0 && skipped == 0 && cache_ctl.index >= 0) {
 			ret = fill_readdir_cache(d_inode(parent), dn,
-- 
2.9.3

[PATCH v2 12/13] ceph: call update_dentry_lease even when r_locked dir is not set

[Jeff Layton]

We don't really require that the parent be locked in order to update the
lease on a dentry. Lease info is protected by the d_lock. In the event
that the parent is not locked in ceph_fill_trace, and we have both
parent and target info, go ahead and update the dentry lease.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 15e042a8d71f..87863f962d50 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1345,6 +1345,16 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 			goto done;
 		}
 		req->r_dentry = dn;  /* may have spliced */
+	} else if (rinfo->head->is_dentry && rinfo->head->is_target) {
+		if ((le32_to_cpu(rinfo->diri.in->cap.caps) & CEPH_CAP_FILE_SHARED) ||
+		    le32_to_cpu(rinfo->dlease->duration_ms)) {
+			vino.ino = le64_to_cpu(rinfo->diri.in->ino);
+			vino.snap = le64_to_cpu(rinfo->diri.in->snapid);
+			update_dentry_lease(req->r_dentry, rinfo->dlease,
+				session, req->r_request_started, &vino);
+		} else {
+			dout("%s: no dentry lease or dir cap\n", __func__);
+		}
 	}
 done:
 	dout("fill_trace done err=%d\n", err);
-- 
2.9.3

[PATCH v2 13/13] ceph: do a LOOKUP in d_revalidate instead of GETATTR

[Jeff Layton]

In commit c3f4688a08f (ceph: don't set req->r_locked_dir in
ceph_d_revalidate), we changed the code to do a GETATTR instead of a
LOOKUP as the parent info isn't strictly necessary to revalidate the
dentry. What we missed there though is that in order to update the lease
on the dentry after revalidating it, we _do_ need parent info.

Change ceph_d_revalidate back to doing a LOOKUP instead of a GETATTR so
that we can get the parent info in order to update the lease from
ceph_fill_trace. Note that we set req->r_parent here, but we cannot set
the CEPH_MDS_R_PARENT_LOCKED flag as we can't guarantee that it is.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/dir.c | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index b889a3b555cf..32248017e928 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -1244,11 +1244,12 @@ static int ceph_d_revalidate(struct dentry *dentry, unsigned int flags)
 			return -ECHILD;
 
 		op = ceph_snap(dir) == CEPH_SNAPDIR ?
-			CEPH_MDS_OP_LOOKUPSNAP : CEPH_MDS_OP_GETATTR;
+			CEPH_MDS_OP_LOOKUPSNAP : CEPH_MDS_OP_LOOKUP;
 		req = ceph_mdsc_create_request(mdsc, op, USE_ANY_MDS);
 		if (!IS_ERR(req)) {
 			req->r_dentry = dget(dentry);
-			req->r_num_caps = op == CEPH_MDS_OP_GETATTR ? 1 : 2;
+			req->r_num_caps = 2;
+			req->r_parent = dir;
 
 			mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED;
 			if (ceph_security_xattr_wanted(dir))
-- 
2.9.3

Re: [PATCH v2 01/13] ceph: add new r_req_flags field to ceph_mds_request

[Ilya Dryomov]

On Wed, Feb 1, 2017 at 12:49 PM, Jeff Layton <jlayton@redhat.com> wrote:
> ...and start moving bool flags into it.
>
> Signed-off-by: Jeff Layton <jlayton@redhat.com>
> ---
>  fs/ceph/dir.c        | 2 +-
>  fs/ceph/mds_client.c | 2 +-
>  fs/ceph/mds_client.h | 4 +++-
>  3 files changed, 5 insertions(+), 3 deletions(-)
>
> diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
> index d4385563b70a..04fa4ae3deca 100644
> --- a/fs/ceph/dir.c
> +++ b/fs/ceph/dir.c
> @@ -371,7 +371,7 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
>                 /* hints to request -> mds selection code */
>                 req->r_direct_mode = USE_AUTH_MDS;
>                 req->r_direct_hash = ceph_frag_value(frag);
> -               req->r_direct_is_hash = true;
> +               set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);

Hi Jeff,

Just a couple of nits:

These are atomic -- should probably mention in the commit message why
is atomicity needed.

>                 if (fi->last_name) {
>                         req->r_path2 = kstrdup(fi->last_name, GFP_KERNEL);
>                         if (!req->r_path2) {
> diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
> index 176512960b14..1f2ef02832d9 100644
> --- a/fs/ceph/mds_client.c
> +++ b/fs/ceph/mds_client.c
> @@ -705,7 +705,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
>         int mode = req->r_direct_mode;
>         int mds = -1;
>         u32 hash = req->r_direct_hash;
> -       bool is_hash = req->r_direct_is_hash;
> +       bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
>
>         /*
>          * is there a specific mds we should try?  ignore hint if we have
> diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
> index 3c6f77b7bb02..a58cacccc986 100644
> --- a/fs/ceph/mds_client.h
> +++ b/fs/ceph/mds_client.h
> @@ -205,6 +205,9 @@ struct ceph_mds_request {
>         struct inode *r_locked_dir; /* dir (if any) i_mutex locked by vfs */
>         struct inode *r_target_inode;       /* resulting inode */
>
> +#define CEPH_MDS_R_DIRECT_IS_HASH      (1) /* r_direct_hash is valid */

Why parens?

> +       unsigned long   r_req_flags;
> +
>         struct mutex r_fill_mutex;
>
>         union ceph_mds_request_args r_args;
> @@ -216,7 +219,6 @@ struct ceph_mds_request {
>         /* for choosing which mds to send this request to */
>         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 */
>
>         /* data payload is used for xattr ops */
>         struct ceph_pagelist *r_pagelist;

3, 4, 5 and 6 can be merged into this patch, IMO.  They are trivial and
some change the same if statement over and over again.

Thanks,

                Ilya

Re: [PATCH v2 01/13] ceph: add new r_req_flags field to ceph_mds_request

[Jeff Layton]

On Wed, 2017-02-01 at 15:08 +0100, Ilya Dryomov wrote:
> On Wed, Feb 1, 2017 at 12:49 PM, Jeff Layton <jlayton@redhat.com> wrote:
> > ...and start moving bool flags into it.
> > 
> > Signed-off-by: Jeff Layton <jlayton@redhat.com>
> > ---
> >  fs/ceph/dir.c        | 2 +-
> >  fs/ceph/mds_client.c | 2 +-
> >  fs/ceph/mds_client.h | 4 +++-
> >  3 files changed, 5 insertions(+), 3 deletions(-)
> > 
> > diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
> > index d4385563b70a..04fa4ae3deca 100644
> > --- a/fs/ceph/dir.c
> > +++ b/fs/ceph/dir.c
> > @@ -371,7 +371,7 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
> >                 /* hints to request -> mds selection code */
> >                 req->r_direct_mode = USE_AUTH_MDS;
> >                 req->r_direct_hash = ceph_frag_value(frag);
> > -               req->r_direct_is_hash = true;
> > +               set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
> 
> Hi Jeff,
> 
> Just a couple of nits:
> 
> These are atomic -- should probably mention in the commit message why
> is atomicity needed.
> 

It's not strictly needed for most of this stuff. DIRECT_IS_HASH in
particular could just use __set_bit.

The rest of the flags, I think are already protected from concurrent
writes by mutexes in the code. What I'm not sure of is whether they are
protected by the _same_ lock. That's a requirement if we mix all of the
flags together into the same word and don't want to use the atomic
*_bit macros.

I can look and see if that's possible. Even if it's not though, using
the atomic *_bit macros is generally not that expensive (particularly
in a situation where we're already using sleeping locks anyway).

> >                 if (fi->last_name) {
> >                         req->r_path2 = kstrdup(fi->last_name, GFP_KERNEL);
> >                         if (!req->r_path2) {
> > diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
> > index 176512960b14..1f2ef02832d9 100644
> > --- a/fs/ceph/mds_client.c
> > +++ b/fs/ceph/mds_client.c
> > @@ -705,7 +705,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
> >         int mode = req->r_direct_mode;
> >         int mds = -1;
> >         u32 hash = req->r_direct_hash;
> > -       bool is_hash = req->r_direct_is_hash;
> > +       bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
> > 
> >         /*
> >          * is there a specific mds we should try?  ignore hint if we have
> > diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
> > index 3c6f77b7bb02..a58cacccc986 100644
> > --- a/fs/ceph/mds_client.h
> > +++ b/fs/ceph/mds_client.h
> > @@ -205,6 +205,9 @@ struct ceph_mds_request {
> >         struct inode *r_locked_dir; /* dir (if any) i_mutex locked by vfs */
> >         struct inode *r_target_inode;       /* resulting inode */
> > 
> > +#define CEPH_MDS_R_DIRECT_IS_HASH      (1) /* r_direct_hash is valid */
> 
> Why parens?
> 

Habit. I think we can safely remove them here.

> > +       unsigned long   r_req_flags;
> > +
> >         struct mutex r_fill_mutex;
> > 
> >         union ceph_mds_request_args r_args;
> > @@ -216,7 +219,6 @@ struct ceph_mds_request {
> >         /* for choosing which mds to send this request to */
> >         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 */
> > 
> >         /* data payload is used for xattr ops */
> >         struct ceph_pagelist *r_pagelist;
> 
> 3, 4, 5 and 6 can be merged into this patch, IMO.  They are trivial and
> some change the same if statement over and over again.
> 

Sure. I did it this way as that's how I put it together, but they can
definitely be squashed together. I'll plan to do that before the next
posting or before merging into testing branch.

-- 
Jeff Layton <jlayton@redhat.com>

Re: [PATCH v2 01/13] ceph: add new r_req_flags field to ceph_mds_request

[Ilya Dryomov]

On Wed, Feb 1, 2017 at 6:47 PM, Jeff Layton <jlayton@redhat.com> wrote:
> On Wed, 2017-02-01 at 15:08 +0100, Ilya Dryomov wrote:
>> On Wed, Feb 1, 2017 at 12:49 PM, Jeff Layton <jlayton@redhat.com> wrote:
>> > ...and start moving bool flags into it.
>> >
>> > Signed-off-by: Jeff Layton <jlayton@redhat.com>
>> > ---
>> >  fs/ceph/dir.c        | 2 +-
>> >  fs/ceph/mds_client.c | 2 +-
>> >  fs/ceph/mds_client.h | 4 +++-
>> >  3 files changed, 5 insertions(+), 3 deletions(-)
>> >
>> > diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
>> > index d4385563b70a..04fa4ae3deca 100644
>> > --- a/fs/ceph/dir.c
>> > +++ b/fs/ceph/dir.c
>> > @@ -371,7 +371,7 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
>> >                 /* hints to request -> mds selection code */
>> >                 req->r_direct_mode = USE_AUTH_MDS;
>> >                 req->r_direct_hash = ceph_frag_value(frag);
>> > -               req->r_direct_is_hash = true;
>> > +               set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
>>
>> Hi Jeff,
>>
>> Just a couple of nits:
>>
>> These are atomic -- should probably mention in the commit message why
>> is atomicity needed.
>>
>
> It's not strictly needed for most of this stuff. DIRECT_IS_HASH in
> particular could just use __set_bit.
>
> The rest of the flags, I think are already protected from concurrent
> writes by mutexes in the code. What I'm not sure of is whether they are
> protected by the _same_ lock. That's a requirement if we mix all of the
> flags together into the same word and don't want to use the atomic
> *_bit macros.
>
> I can look and see if that's possible. Even if it's not though, using
> the atomic *_bit macros is generally not that expensive (particularly
> in a situation where we're already using sleeping locks anyway).

Auditing for whether __set_bit, etc can be used instead is probably not
worth it.  I'm not saying we shouldn't use atomic bitops here -- just
wanted to get this explanation in the commit message.

Thanks,

                Ilya

Re: [PATCH v2 01/13] ceph: add new r_req_flags field to ceph_mds_request

[Jeff Layton]

On Wed, 2017-02-01 at 20:10 +0100, Ilya Dryomov wrote:
> On Wed, Feb 1, 2017 at 6:47 PM, Jeff Layton <jlayton@redhat.com> wrote:
> > On Wed, 2017-02-01 at 15:08 +0100, Ilya Dryomov wrote:
> > > On Wed, Feb 1, 2017 at 12:49 PM, Jeff Layton <jlayton@redhat.com> wrote:
> > > > ...and start moving bool flags into it.
> > > > 
> > > > Signed-off-by: Jeff Layton <jlayton@redhat.com>
> > > > ---
> > > >  fs/ceph/dir.c        | 2 +-
> > > >  fs/ceph/mds_client.c | 2 +-
> > > >  fs/ceph/mds_client.h | 4 +++-
> > > >  3 files changed, 5 insertions(+), 3 deletions(-)
> > > > 
> > > > diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
> > > > index d4385563b70a..04fa4ae3deca 100644
> > > > --- a/fs/ceph/dir.c
> > > > +++ b/fs/ceph/dir.c
> > > > @@ -371,7 +371,7 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
> > > >                 /* hints to request -> mds selection code */
> > > >                 req->r_direct_mode = USE_AUTH_MDS;
> > > >                 req->r_direct_hash = ceph_frag_value(frag);
> > > > -               req->r_direct_is_hash = true;
> > > > +               set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
> > > 
> > > Hi Jeff,
> > > 
> > > Just a couple of nits:
> > > 
> > > These are atomic -- should probably mention in the commit message why
> > > is atomicity needed.
> > > 
> > 
> > It's not strictly needed for most of this stuff. DIRECT_IS_HASH in
> > particular could just use __set_bit.
> > 
> > The rest of the flags, I think are already protected from concurrent
> > writes by mutexes in the code. What I'm not sure of is whether they are
> > protected by the _same_ lock. That's a requirement if we mix all of the
> > flags together into the same word and don't want to use the atomic
> > *_bit macros.
> > 
> > I can look and see if that's possible. Even if it's not though, using
> > the atomic *_bit macros is generally not that expensive (particularly
> > in a situation where we're already using sleeping locks anyway).
> 
> Auditing for whether __set_bit, etc can be used instead is probably not
> worth it.  I'm not saying we shouldn't use atomic bitops here -- just
> wanted to get this explanation in the commit message.
> 
> 

Thanks, done. I went ahead and changed DIRECT_IS_HASH to use __set_bit
(it's pretty clear that that's safe there), but the others do need the
atomic versions as they can have other tasks manipulating them.

I went ahead and merged the squashed set into ceph-client/testing with
an updated commit message. Let me know if you see any other issues with
the set.

-- 
Jeff Layton <jlayton@redhat.com>

Re: [PATCH v2 02/13] ceph: move r_aborted flag into r_req_flags

[Yan, Zheng]

> On 1 Feb 2017, at 19:49, Jeff Layton <jlayton@redhat.com> wrote:
> 
> ...and simplify the handling of it in ceph_fill_trace.
> 
> Signed-off-by: Jeff Layton <jlayton@redhat.com>
> ---
> fs/ceph/inode.c      | 17 +++++++++--------
> fs/ceph/mds_client.c |  8 ++++----
> fs/ceph/mds_client.h |  2 +-
> 3 files changed, 14 insertions(+), 13 deletions(-)
> 
> diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> index 4926265f4223..bd2e94a78057 100644
> --- a/fs/ceph/inode.c
> +++ b/fs/ceph/inode.c
> @@ -1233,8 +1233,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> 
> 		err = fill_inode(in, req->r_locked_page, &rinfo->targeti, NULL,
> 				session, req->r_request_started,
> -				(!req->r_aborted && rinfo->head->result == 0) ?
> -				req->r_fmode : -1,
> +				(!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
> +				rinfo->head->result == 0) ?  req->r_fmode : -1,
> 				&req->r_caps_reservation);
> 		if (err < 0) {
> 			pr_err("fill_inode badness %p %llx.%llx\n",
> @@ -1243,12 +1243,14 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> 		}
> 	}
> 
> +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> +		goto done;
> +

This seems wrong. If we got the reply, we should update inode’s caps even request is aborted.
Otherwise, MDS can hang at revoking the dropped caps.

> 	/*
> 	 * ignore null lease/binding on snapdir ENOENT, or else we
> 	 * will have trouble splicing in the virtual snapdir later
> 	 */
> -	if (rinfo->head->is_dentry && !req->r_aborted &&
> -	    req->r_locked_dir &&
> +	if (rinfo->head->is_dentry && req->r_locked_dir &&
> 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
> 					       fsc->mount_options->snapdir_name,
> 					       req->r_dentry->d_name.len))) {
> @@ -1351,9 +1353,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> 			update_dentry_lease(dn, rinfo->dlease, session,
> 					    req->r_request_started);
> 		dout(" final dn %p\n", dn);
> -	} else if (!req->r_aborted &&
> -		   (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> -		    req->r_op == CEPH_MDS_OP_MKSNAP)) {
> +	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> +		   req->r_op == CEPH_MDS_OP_MKSNAP) {
> 		struct dentry *dn = req->r_dentry;
> 		struct inode *dir = req->r_locked_dir;
> 
> @@ -1478,7 +1479,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> 	u32 fpos_offset;
> 	struct ceph_readdir_cache_control cache_ctl = {};
> 
> -	if (req->r_aborted)
> +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> 		return readdir_prepopulate_inodes_only(req, session);
> 
> 	if (rinfo->hash_order && req->r_path2) {
> diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
> index 1f2ef02832d9..a5156b6a0aed 100644
> --- a/fs/ceph/mds_client.c
> +++ b/fs/ceph/mds_client.c
> @@ -2115,7 +2115,7 @@ static int __do_request(struct ceph_mds_client *mdsc,
> 	int err = 0;
> 
> 	if (req->r_err || req->r_got_result) {
> -		if (req->r_aborted)
> +		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> 			__unregister_request(mdsc, req);
> 		goto out;
> 	}
> @@ -2331,7 +2331,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
> 		 */
> 		mutex_lock(&req->r_fill_mutex);
> 		req->r_err = err;
> -		req->r_aborted = true;
> +		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
> 		mutex_unlock(&req->r_fill_mutex);
> 
> 		if (req->r_locked_dir &&
> @@ -2538,7 +2538,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
> 	}
> out_err:
> 	mutex_lock(&mdsc->mutex);
> -	if (!req->r_aborted) {
> +	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
> 		if (err) {
> 			req->r_err = err;
> 		} else {
> @@ -2587,7 +2587,7 @@ static void handle_forward(struct ceph_mds_client *mdsc,
> 		goto out;  /* dup reply? */
> 	}
> 
> -	if (req->r_aborted) {
> +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
> 		dout("forward tid %llu aborted, unregistering\n", tid);
> 		__unregister_request(mdsc, req);
> 	} else if (fwd_seq <= req->r_num_fwd) {
> diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
> index a58cacccc986..3da20955d5e6 100644
> --- a/fs/ceph/mds_client.h
> +++ b/fs/ceph/mds_client.h
> @@ -206,6 +206,7 @@ struct ceph_mds_request {
> 	struct inode *r_target_inode;       /* resulting inode */
> 
> #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
> +#define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
> 	unsigned long	r_req_flags;
> 
> 	struct mutex r_fill_mutex;
> @@ -236,7 +237,6 @@ struct ceph_mds_request {
> 	struct ceph_mds_reply_info_parsed r_reply_info;
> 	struct page *r_locked_page;
> 	int r_err;
> -	bool r_aborted;
> 
> 	unsigned long r_timeout;  /* optional.  jiffies, 0 is "wait forever" */
> 	unsigned long r_started;  /* start time to measure timeout against */
> -- 
> 2.9.3
> 

Re: [PATCH v2 12/13] ceph: call update_dentry_lease even when r_locked dir is not set

[Yan, Zheng]

> On 1 Feb 2017, at 19:49, Jeff Layton <jlayton@redhat.com> wrote:
> 
> We don't really require that the parent be locked in order to update the
> lease on a dentry. Lease info is protected by the d_lock. In the event
> that the parent is not locked in ceph_fill_trace, and we have both
> parent and target info, go ahead and update the dentry lease.
> 
> Signed-off-by: Jeff Layton <jlayton@redhat.com>
> ---
> fs/ceph/inode.c | 10 ++++++++++
> 1 file changed, 10 insertions(+)
> 
> diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> index 15e042a8d71f..87863f962d50 100644
> --- a/fs/ceph/inode.c
> +++ b/fs/ceph/inode.c
> @@ -1345,6 +1345,16 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 			goto done;
> 		}
> 		req->r_dentry = dn;  /* may have spliced */
> +	} else if (rinfo->head->is_dentry && rinfo->head->is_target) {
> +		if ((le32_to_cpu(rinfo->diri.in->cap.caps) & CEPH_CAP_FILE_SHARED) ||
> +		    le32_to_cpu(rinfo->dlease->duration_ms)) {
> +			vino.ino = le64_to_cpu(rinfo->diri.in->ino);
> +			vino.snap = le64_to_cpu(rinfo->diri.in->snapid);
> +			update_dentry_lease(req->r_dentry, rinfo->dlease,
> +				session, req->r_request_started, &vino);
> +		} else {
> +			dout("%s: no dentry lease or dir cap\n", __func__);
> +		}
> 	}

I think checking rinfo->head->is_target is not needed here, because null dentry can also have lease.
Besides, I think we need to check if rinfo->head->is_target matches d_really_is_negative(dn) and if
the target inode matches d_inode(dn).

> 	dout("fill_trace done err=%d\n", err);
> -- 
> 2.9.3
> 

Re: [PATCH v2 02/13] ceph: move r_aborted flag into r_req_flags

[Jeff Layton]

On Thu, 2017-02-02 at 16:06 +0800, Yan, Zheng wrote:
> > On 1 Feb 2017, at 19:49, Jeff Layton <jlayton@redhat.com> wrote:
> > 
> > ...and simplify the handling of it in ceph_fill_trace.
> > 
> > Signed-off-by: Jeff Layton <jlayton@redhat.com>
> > ---
> > fs/ceph/inode.c      | 17 +++++++++--------
> > fs/ceph/mds_client.c |  8 ++++----
> > fs/ceph/mds_client.h |  2 +-
> > 3 files changed, 14 insertions(+), 13 deletions(-)
> > 
> > diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> > index 4926265f4223..bd2e94a78057 100644
> > --- a/fs/ceph/inode.c
> > +++ b/fs/ceph/inode.c
> > @@ -1233,8 +1233,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> > 
> > 		err = fill_inode(in, req->r_locked_page, &rinfo->targeti, NULL,
> > 				session, req->r_request_started,
> > -				(!req->r_aborted && rinfo->head->result == 0) ?
> > -				req->r_fmode : -1,
> > +				(!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
> > +				rinfo->head->result == 0) ?  req->r_fmode : -1,
> > 				&req->r_caps_reservation);
> > 		if (err < 0) {
> > 			pr_err("fill_inode badness %p %llx.%llx\n",
> > @@ -1243,12 +1243,14 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> > 		}
> > 	}
> > 
> > +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> > +		goto done;
> > +
> 
> This seems wrong. If we got the reply, we should update inode’s caps even request is aborted.
> Otherwise, MDS can hang at revoking the dropped caps.
> 

FWIW, I hit that exact problem in an earlier iteration of this patchset
until I figured out what was going on. The caps do get updated with this
patch. That gets done in fill_inode, and that's called even when the
call was aborted.

None of the "move the flag into r_req_flags" patches materially change
the behavior of the code. In this patch, we're just moving both of the
subsequent !req->r_aborted flag checks into this one spot.

> > 	/*
> > 	 * ignore null lease/binding on snapdir ENOENT, or else we
> > 	 * will have trouble splicing in the virtual snapdir later
> > 	 */
> > -	if (rinfo->head->is_dentry && !req->r_aborted &&
> > -	    req->r_locked_dir &&
> > +	if (rinfo->head->is_dentry && req->r_locked_dir &&
> > 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
> > 					       fsc->mount_options->snapdir_name,
> > 					       req->r_dentry->d_name.len))) {
> > @@ -1351,9 +1353,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> > 			update_dentry_lease(dn, rinfo->dlease, session,
> > 					    req->r_request_started);
> > 		dout(" final dn %p\n", dn);
> > -	} else if (!req->r_aborted &&
> > -		   (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> > -		    req->r_op == CEPH_MDS_OP_MKSNAP)) {
> > +	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> > +		   req->r_op == CEPH_MDS_OP_MKSNAP) {
> > 		struct dentry *dn = req->r_dentry;
> > 		struct inode *dir = req->r_locked_dir;
> > 
> > @@ -1478,7 +1479,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> > 	u32 fpos_offset;
> > 	struct ceph_readdir_cache_control cache_ctl = {};
> > 
> > -	if (req->r_aborted)
> > +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> > 		return readdir_prepopulate_inodes_only(req, session);
> > 
> > 	if (rinfo->hash_order && req->r_path2) {
> > diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
> > index 1f2ef02832d9..a5156b6a0aed 100644
> > --- a/fs/ceph/mds_client.c
> > +++ b/fs/ceph/mds_client.c
> > @@ -2115,7 +2115,7 @@ static int __do_request(struct ceph_mds_client *mdsc,
> > 	int err = 0;
> > 
> > 	if (req->r_err || req->r_got_result) {
> > -		if (req->r_aborted)
> > +		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> > 			__unregister_request(mdsc, req);
> > 		goto out;
> > 	}
> > @@ -2331,7 +2331,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
> > 		 */
> > 		mutex_lock(&req->r_fill_mutex);
> > 		req->r_err = err;
> > -		req->r_aborted = true;
> > +		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
> > 		mutex_unlock(&req->r_fill_mutex);
> > 
> > 		if (req->r_locked_dir &&
> > @@ -2538,7 +2538,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
> > 	}
> > out_err:
> > 	mutex_lock(&mdsc->mutex);
> > -	if (!req->r_aborted) {
> > +	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
> > 		if (err) {
> > 			req->r_err = err;
> > 		} else {
> > @@ -2587,7 +2587,7 @@ static void handle_forward(struct ceph_mds_client *mdsc,
> > 		goto out;  /* dup reply? */
> > 	}
> > 
> > -	if (req->r_aborted) {
> > +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
> > 		dout("forward tid %llu aborted, unregistering\n", tid);
> > 		__unregister_request(mdsc, req);
> > 	} else if (fwd_seq <= req->r_num_fwd) {
> > diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
> > index a58cacccc986..3da20955d5e6 100644
> > --- a/fs/ceph/mds_client.h
> > +++ b/fs/ceph/mds_client.h
> > @@ -206,6 +206,7 @@ struct ceph_mds_request {
> > 	struct inode *r_target_inode;       /* resulting inode */
> > 
> > #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
> > +#define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
> > 	unsigned long	r_req_flags;
> > 
> > 	struct mutex r_fill_mutex;
> > @@ -236,7 +237,6 @@ struct ceph_mds_request {
> > 	struct ceph_mds_reply_info_parsed r_reply_info;
> > 	struct page *r_locked_page;
> > 	int r_err;
> > -	bool r_aborted;
> > 
> > 	unsigned long r_timeout;  /* optional.  jiffies, 0 is "wait forever" */
> > 	unsigned long r_started;  /* start time to measure timeout against */
> > -- 
> > 2.9.3
> > 
> 
> 

-- 
Jeff Layton <jlayton@redhat.com>

Re: [PATCH v2 12/13] ceph: call update_dentry_lease even when r_locked dir is not set

[Jeff Layton]

On Thu, 2017-02-02 at 16:34 +0800, Yan, Zheng wrote:
> > On 1 Feb 2017, at 19:49, Jeff Layton <jlayton@redhat.com> wrote:
> > 
> > We don't really require that the parent be locked in order to update the
> > lease on a dentry. Lease info is protected by the d_lock. In the event
> > that the parent is not locked in ceph_fill_trace, and we have both
> > parent and target info, go ahead and update the dentry lease.
> > 
> > Signed-off-by: Jeff Layton <jlayton@redhat.com>
> > ---
> > fs/ceph/inode.c | 10 ++++++++++
> > 1 file changed, 10 insertions(+)
> > 
> > diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> > index 15e042a8d71f..87863f962d50 100644
> > --- a/fs/ceph/inode.c
> > +++ b/fs/ceph/inode.c
> > @@ -1345,6 +1345,16 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 			goto done;
> > 		}
> > 		req->r_dentry = dn;  /* may have spliced */
> > +	} else if (rinfo->head->is_dentry && rinfo->head->is_target) {
> > +		if ((le32_to_cpu(rinfo->diri.in->cap.caps) & CEPH_CAP_FILE_SHARED) ||
> > +		    le32_to_cpu(rinfo->dlease->duration_ms)) {
> > +			vino.ino = le64_to_cpu(rinfo->diri.in->ino);
> > +			vino.snap = le64_to_cpu(rinfo->diri.in->snapid);
> > +			update_dentry_lease(req->r_dentry, rinfo->dlease,
> > +				session, req->r_request_started, &vino);
> > +		} else {
> > +			dout("%s: no dentry lease or dir cap\n", __func__);
> > +		}
> > 	}
> 
> I think checking rinfo->head->is_target is not needed here, because null dentry can also have lease.
> Besides, I think we need to check if rinfo->head->is_target matches d_really_is_negative(dn) and if
> the target inode matches d_inode(dn).
> 

Yeah, I think you're right here. I'll respin this one.
-- 
Jeff Layton <jlayton@redhat.com>

Re: [PATCH v2 02/13] ceph: move r_aborted flag into r_req_flags

[Yan, Zheng]

> On 2 Feb 2017, at 19:26, Jeff Layton <jlayton@redhat.com> wrote:
> 
> On Thu, 2017-02-02 at 16:06 +0800, Yan, Zheng wrote:
>>> On 1 Feb 2017, at 19:49, Jeff Layton <jlayton@redhat.com> wrote:
>>> 
>>> ...and simplify the handling of it in ceph_fill_trace.
>>> 
>>> Signed-off-by: Jeff Layton <jlayton@redhat.com>
>>> ---
>>> fs/ceph/inode.c      | 17 +++++++++--------
>>> fs/ceph/mds_client.c |  8 ++++----
>>> fs/ceph/mds_client.h |  2 +-
>>> 3 files changed, 14 insertions(+), 13 deletions(-)
>>> 
>>> diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
>>> index 4926265f4223..bd2e94a78057 100644
>>> --- a/fs/ceph/inode.c
>>> +++ b/fs/ceph/inode.c
>>> @@ -1233,8 +1233,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
>>> 
>>> 		err = fill_inode(in, req->r_locked_page, &rinfo->targeti, NULL,
>>> 				session, req->r_request_started,
>>> -				(!req->r_aborted && rinfo->head->result == 0) ?
>>> -				req->r_fmode : -1,
>>> +				(!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
>>> +				rinfo->head->result == 0) ?  req->r_fmode : -1,
>>> 				&req->r_caps_reservation);
>>> 		if (err < 0) {
>>> 			pr_err("fill_inode badness %p %llx.%llx\n",
>>> @@ -1243,12 +1243,14 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
>>> 		}
>>> 	}
>>> 
>>> +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
>>> +		goto done;
>>> +
>> 
>> This seems wrong. If we got the reply, we should update inode’s caps even request is aborted.
>> Otherwise, MDS can hang at revoking the dropped caps.
>> 
> 
> FWIW, I hit that exact problem in an earlier iteration of this patchset
> until I figured out what was going on. The caps do get updated with this
> patch. That gets done in fill_inode, and that's called even when the
> call was aborted.
> 
> None of the "move the flag into r_req_flags" patches materially change
> the behavior of the code. In this patch, we're just moving both of the
> subsequent !req->r_aborted flag checks into this one spot.

I think you are right. Thank you for explanation

Yan, Zheng
> 
>>> 	/*
>>> 	 * ignore null lease/binding on snapdir ENOENT, or else we
>>> 	 * will have trouble splicing in the virtual snapdir later
>>> 	 */
>>> -	if (rinfo->head->is_dentry && !req->r_aborted &&
>>> -	    req->r_locked_dir &&
>>> +	if (rinfo->head->is_dentry && req->r_locked_dir &&
>>> 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
>>> 					       fsc->mount_options->snapdir_name,
>>> 					       req->r_dentry->d_name.len))) {
>>> @@ -1351,9 +1353,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
>>> 			update_dentry_lease(dn, rinfo->dlease, session,
>>> 					    req->r_request_started);
>>> 		dout(" final dn %p\n", dn);
>>> -	} else if (!req->r_aborted &&
>>> -		   (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
>>> -		    req->r_op == CEPH_MDS_OP_MKSNAP)) {
>>> +	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
>>> +		   req->r_op == CEPH_MDS_OP_MKSNAP) {
>>> 		struct dentry *dn = req->r_dentry;
>>> 		struct inode *dir = req->r_locked_dir;
>>> 
>>> @@ -1478,7 +1479,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
>>> 	u32 fpos_offset;
>>> 	struct ceph_readdir_cache_control cache_ctl = {};
>>> 
>>> -	if (req->r_aborted)
>>> +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
>>> 		return readdir_prepopulate_inodes_only(req, session);
>>> 
>>> 	if (rinfo->hash_order && req->r_path2) {
>>> diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
>>> index 1f2ef02832d9..a5156b6a0aed 100644
>>> --- a/fs/ceph/mds_client.c
>>> +++ b/fs/ceph/mds_client.c
>>> @@ -2115,7 +2115,7 @@ static int __do_request(struct ceph_mds_client *mdsc,
>>> 	int err = 0;
>>> 
>>> 	if (req->r_err || req->r_got_result) {
>>> -		if (req->r_aborted)
>>> +		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
>>> 			__unregister_request(mdsc, req);
>>> 		goto out;
>>> 	}
>>> @@ -2331,7 +2331,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
>>> 		 */
>>> 		mutex_lock(&req->r_fill_mutex);
>>> 		req->r_err = err;
>>> -		req->r_aborted = true;
>>> +		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
>>> 		mutex_unlock(&req->r_fill_mutex);
>>> 
>>> 		if (req->r_locked_dir &&
>>> @@ -2538,7 +2538,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
>>> 	}
>>> out_err:
>>> 	mutex_lock(&mdsc->mutex);
>>> -	if (!req->r_aborted) {
>>> +	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
>>> 		if (err) {
>>> 			req->r_err = err;
>>> 		} else {
>>> @@ -2587,7 +2587,7 @@ static void handle_forward(struct ceph_mds_client *mdsc,
>>> 		goto out;  /* dup reply? */
>>> 	}
>>> 
>>> -	if (req->r_aborted) {
>>> +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
>>> 		dout("forward tid %llu aborted, unregistering\n", tid);
>>> 		__unregister_request(mdsc, req);
>>> 	} else if (fwd_seq <= req->r_num_fwd) {
>>> diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
>>> index a58cacccc986..3da20955d5e6 100644
>>> --- a/fs/ceph/mds_client.h
>>> +++ b/fs/ceph/mds_client.h
>>> @@ -206,6 +206,7 @@ struct ceph_mds_request {
>>> 	struct inode *r_target_inode;       /* resulting inode */
>>> 
>>> #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
>>> +#define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
>>> 	unsigned long	r_req_flags;
>>> 
>>> 	struct mutex r_fill_mutex;
>>> @@ -236,7 +237,6 @@ struct ceph_mds_request {
>>> 	struct ceph_mds_reply_info_parsed r_reply_info;
>>> 	struct page *r_locked_page;
>>> 	int r_err;
>>> -	bool r_aborted;
>>> 
>>> 	unsigned long r_timeout;  /* optional.  jiffies, 0 is "wait forever" */
>>> 	unsigned long r_started;  /* start time to measure timeout against */
>>> -- 
>>> 2.9.3
>>> 
>> 
>> 
> 
> -- 
> Jeff Layton <jlayton@redhat.com>

Re: [PATCH v2 02/13] ceph: move r_aborted flag into r_req_flags

[Jeff Layton]

On Thu, 2017-02-02 at 23:16 +0800, Yan, Zheng wrote:
> > On 2 Feb 2017, at 19:26, Jeff Layton <jlayton@redhat.com> wrote:
> > 
> > On Thu, 2017-02-02 at 16:06 +0800, Yan, Zheng wrote:
> > > > On 1 Feb 2017, at 19:49, Jeff Layton <jlayton@redhat.com> wrote:
> > > > 
> > > > ...and simplify the handling of it in ceph_fill_trace.
> > > > 
> > > > Signed-off-by: Jeff Layton <jlayton@redhat.com>
> > > > ---
> > > > fs/ceph/inode.c      | 17 +++++++++--------
> > > > fs/ceph/mds_client.c |  8 ++++----
> > > > fs/ceph/mds_client.h |  2 +-
> > > > 3 files changed, 14 insertions(+), 13 deletions(-)
> > > > 
> > > > diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> > > > index 4926265f4223..bd2e94a78057 100644
> > > > --- a/fs/ceph/inode.c
> > > > +++ b/fs/ceph/inode.c
> > > > @@ -1233,8 +1233,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> > > > 
> > > > 		err = fill_inode(in, req->r_locked_page, &rinfo->targeti, NULL,
> > > > 				session, req->r_request_started,
> > > > -				(!req->r_aborted && rinfo->head->result == 0) ?
> > > > -				req->r_fmode : -1,
> > > > +				(!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
> > > > +				rinfo->head->result == 0) ?  req->r_fmode : -1,
> > > > 				&req->r_caps_reservation);
> > > > 		if (err < 0) {
> > > > 			pr_err("fill_inode badness %p %llx.%llx\n",
> > > > @@ -1243,12 +1243,14 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> > > > 		}
> > > > 	}
> > > > 
> > > > +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> > > > +		goto done;
> > > > +
> > > 
> > > This seems wrong. If we got the reply, we should update inode’s caps even request is aborted.
> > > Otherwise, MDS can hang at revoking the dropped caps.
> > > 
> > 
> > FWIW, I hit that exact problem in an earlier iteration of this patchset
> > until I figured out what was going on. The caps do get updated with this
> > patch. That gets done in fill_inode, and that's called even when the
> > call was aborted.
> > 
> > None of the "move the flag into r_req_flags" patches materially change
> > the behavior of the code. In this patch, we're just moving both of the
> > subsequent !req->r_aborted flag checks into this one spot.
> 
> I think you are right. Thank you for explanation
> 
> Yan, Zheng

Well...this patch doesn't break the existing code, but I think we do
still want to update the dentry lease, even when the call was cancelled.
 With this change that won't happen. That'll be fixed in the next
posting.
 
> > 
> > > > 	/*
> > > > 	 * ignore null lease/binding on snapdir ENOENT, or else we
> > > > 	 * will have trouble splicing in the virtual snapdir later
> > > > 	 */
> > > > -	if (rinfo->head->is_dentry && !req->r_aborted &&
> > > > -	    req->r_locked_dir &&
> > > > +	if (rinfo->head->is_dentry && req->r_locked_dir &&
> > > > 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
> > > > 					       fsc->mount_options->snapdir_name,
> > > > 					       req->r_dentry->d_name.len))) {
> > > > @@ -1351,9 +1353,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
> > > > 			update_dentry_lease(dn, rinfo->dlease, session,
> > > > 					    req->r_request_started);
> > > > 		dout(" final dn %p\n", dn);
> > > > -	} else if (!req->r_aborted &&
> > > > -		   (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> > > > -		    req->r_op == CEPH_MDS_OP_MKSNAP)) {
> > > > +	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> > > > +		   req->r_op == CEPH_MDS_OP_MKSNAP) {
> > > > 		struct dentry *dn = req->r_dentry;
> > > > 		struct inode *dir = req->r_locked_dir;
> > > > 
> > > > @@ -1478,7 +1479,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> > > > 	u32 fpos_offset;
> > > > 	struct ceph_readdir_cache_control cache_ctl = {};
> > > > 
> > > > -	if (req->r_aborted)
> > > > +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> > > > 		return readdir_prepopulate_inodes_only(req, session);
> > > > 
> > > > 	if (rinfo->hash_order && req->r_path2) {
> > > > diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
> > > > index 1f2ef02832d9..a5156b6a0aed 100644
> > > > --- a/fs/ceph/mds_client.c
> > > > +++ b/fs/ceph/mds_client.c
> > > > @@ -2115,7 +2115,7 @@ static int __do_request(struct ceph_mds_client *mdsc,
> > > > 	int err = 0;
> > > > 
> > > > 	if (req->r_err || req->r_got_result) {
> > > > -		if (req->r_aborted)
> > > > +		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
> > > > 			__unregister_request(mdsc, req);
> > > > 		goto out;
> > > > 	}
> > > > @@ -2331,7 +2331,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
> > > > 		 */
> > > > 		mutex_lock(&req->r_fill_mutex);
> > > > 		req->r_err = err;
> > > > -		req->r_aborted = true;
> > > > +		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
> > > > 		mutex_unlock(&req->r_fill_mutex);
> > > > 
> > > > 		if (req->r_locked_dir &&
> > > > @@ -2538,7 +2538,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
> > > > 	}
> > > > out_err:
> > > > 	mutex_lock(&mdsc->mutex);
> > > > -	if (!req->r_aborted) {
> > > > +	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
> > > > 		if (err) {
> > > > 			req->r_err = err;
> > > > 		} else {
> > > > @@ -2587,7 +2587,7 @@ static void handle_forward(struct ceph_mds_client *mdsc,
> > > > 		goto out;  /* dup reply? */
> > > > 	}
> > > > 
> > > > -	if (req->r_aborted) {
> > > > +	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
> > > > 		dout("forward tid %llu aborted, unregistering\n", tid);
> > > > 		__unregister_request(mdsc, req);
> > > > 	} else if (fwd_seq <= req->r_num_fwd) {
> > > > diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
> > > > index a58cacccc986..3da20955d5e6 100644
> > > > --- a/fs/ceph/mds_client.h
> > > > +++ b/fs/ceph/mds_client.h
> > > > @@ -206,6 +206,7 @@ struct ceph_mds_request {
> > > > 	struct inode *r_target_inode;       /* resulting inode */
> > > > 
> > > > #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
> > > > +#define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
> > > > 	unsigned long	r_req_flags;
> > > > 
> > > > 	struct mutex r_fill_mutex;
> > > > @@ -236,7 +237,6 @@ struct ceph_mds_request {
> > > > 	struct ceph_mds_reply_info_parsed r_reply_info;
> > > > 	struct page *r_locked_page;
> > > > 	int r_err;
> > > > -	bool r_aborted;
> > > > 
> > > > 	unsigned long r_timeout;  /* optional.  jiffies, 0 is "wait forever" */
> > > > 	unsigned long r_started;  /* start time to measure timeout against */
> > > > -- 
> > > > 2.9.3
> > > > 
> > > 
> > > 
> > 
> > -- 
> > Jeff Layton <jlayton@redhat.com>
> 
> 

-- 
Jeff Layton <jlayton@redhat.com>

[PATCH v3 0/8] ceph: fix performance regression due to ceph_d_revalidate fixes

[Jeff Layton]

v3: Squash bool to flag change patches, and better explain why we need
to use atomic *_bit calls. Vet target and parent inodes when updating
lease. Update lease properly on negative dentries (when is_target is 0).
Also, update dentry leases even when call that triggered the request
is cancelled.

v2: Fix a hang that would occasionally occur under fsstress

Zheng mentioned that he had gotten a report that testing with vdbench on
a v4.8 kernel showed a large increase in the number of GETATTR calls
being made. Since we had changed the d_revalidate code to use a GETATTR
call and not set r_locked_dir in the request, the dentry lease wasn't
getting updated from the traces, even when the MDS had granted one.

We need to use a LOOKUP when revalidating dentries so we can update
the dentry lease if things look ok. That currently depends however on
having r_locked_dir set to the parent directory.

In d_revalidate though, we may very well not hold the parent's i_rwsem.
Manipulating the dcache is not allowed without it, but we can update the
inode and the dentry leases.

The basic approach here is to separate r_locked_dir into an inode
pointer representing the parent, and a flag indicating whether we
know that it's locked (note that it _may_ still be locked even if
that flag isn't set).

In ceph_fill_trace, we can then do all of the parts that don't require a
locked parent whenever the parent pointer is set, but avoid the
dcache manipulation when it isn't.

With this, dentry leases are again being updated as a result of
d_revalidate lookups. The switch to using a flags fields instead
of bools also shrinks the request size slightly. 

Jeff Layton (8):
  ceph: remove "Debugging hook" from ceph_fill_trace
  ceph: drop session argument to ceph_fill_trace
  ceph: convert bools in ceph_mds_request to a new r_req_flags field
  ceph: add a new flag to indicate whether parent is locked
  ceph: don't update_dentry_lease unless we actually got one
  ceph: vet the target and parent inodes before updating dentry lease
  ceph: call update_dentry_lease even when r_locked dir is not set
  ceph: do a LOOKUP in d_revalidate instead of GETATTR

 fs/ceph/debugfs.c    |   2 +-
 fs/ceph/dir.c        |  30 ++++++----
 fs/ceph/export.c     |   3 +-
 fs/ceph/file.c       |   3 +-
 fs/ceph/inode.c      | 164 ++++++++++++++++++++++++++++-----------------------
 fs/ceph/mds_client.c |  73 ++++++++++++-----------
 fs/ceph/mds_client.h |  15 +++--
 fs/ceph/super.h      |   3 +-
 8 files changed, 162 insertions(+), 131 deletions(-)

-- 
2.9.3

[PATCH v3 1/8] ceph: remove "Debugging hook" from ceph_fill_trace

[Jeff Layton]

Keeping around commented out code is just asking for it to bitrot and
makes viewing the code under cscope more confusing.  If
we really need this, then we can revert this patch and put it under a
Kconfig option.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 34 ----------------------------------
 1 file changed, 34 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 4926265f4223..38a54ebf647d 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1120,40 +1120,6 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 	dout("fill_trace %p is_dentry %d is_target %d\n", req,
 	     rinfo->head->is_dentry, rinfo->head->is_target);
 
-#if 0
-	/*
-	 * Debugging hook:
-	 *
-	 * If we resend completed ops to a recovering mds, we get no
-	 * trace.  Since that is very rare, pretend this is the case
-	 * to ensure the 'no trace' handlers in the callers behave.
-	 *
-	 * Fill in inodes unconditionally to avoid breaking cap
-	 * invariants.
-	 */
-	if (rinfo->head->op & CEPH_MDS_OP_WRITE) {
-		pr_info("fill_trace faking empty trace on %lld %s\n",
-			req->r_tid, ceph_mds_op_name(rinfo->head->op));
-		if (rinfo->head->is_dentry) {
-			rinfo->head->is_dentry = 0;
-			err = fill_inode(req->r_locked_dir,
-					 &rinfo->diri, rinfo->dirfrag,
-					 session, req->r_request_started, -1);
-		}
-		if (rinfo->head->is_target) {
-			rinfo->head->is_target = 0;
-			ininfo = rinfo->targeti.in;
-			vino.ino = le64_to_cpu(ininfo->ino);
-			vino.snap = le64_to_cpu(ininfo->snapid);
-			in = ceph_get_inode(sb, vino);
-			err = fill_inode(in, &rinfo->targeti, NULL,
-					 session, req->r_request_started,
-					 req->r_fmode);
-			iput(in);
-		}
-	}
-#endif
-
 	if (!rinfo->head->is_target && !rinfo->head->is_dentry) {
 		dout("fill_trace reply is empty!\n");
 		if (rinfo->head->result == 0 && req->r_locked_dir)
-- 
2.9.3

[PATCH v3 2/8] ceph: drop session argument to ceph_fill_trace

[Jeff Layton]

Just get it from r_session since that's what's always passed in.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c      | 4 ++--
 fs/ceph/mds_client.c | 2 +-
 fs/ceph/super.h      | 3 +--
 3 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 38a54ebf647d..b18462c64cdd 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1108,9 +1108,9 @@ static struct dentry *splice_dentry(struct dentry *dn, struct inode *in)
  *
  * Called with snap_rwsem (read).
  */
-int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
-		    struct ceph_mds_session *session)
+int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 {
+	struct ceph_mds_session *session = req->r_session;
 	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
 	struct inode *in = NULL;
 	struct ceph_vino vino;
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 176512960b14..4a68067a76b5 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -2516,7 +2516,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	/* insert trace into our cache */
 	mutex_lock(&req->r_fill_mutex);
 	current->journal_info = req;
-	err = ceph_fill_trace(mdsc->fsc->sb, req, req->r_session);
+	err = ceph_fill_trace(mdsc->fsc->sb, req);
 	if (err == 0) {
 		if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
 				    req->r_op == CEPH_MDS_OP_LSSNAP))
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 6477264bfc7e..950170136be9 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -764,8 +764,7 @@ extern void ceph_fill_file_time(struct inode *inode, int issued,
 				u64 time_warp_seq, struct timespec *ctime,
 				struct timespec *mtime, struct timespec *atime);
 extern int ceph_fill_trace(struct super_block *sb,
-			   struct ceph_mds_request *req,
-			   struct ceph_mds_session *session);
+			   struct ceph_mds_request *req);
 extern int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 				    struct ceph_mds_session *session);
 
-- 
2.9.3

[PATCH v3 3/8] ceph: convert bools in ceph_mds_request to a new r_req_flags field

[Jeff Layton]

Currently, we have a bunch of bool flags in struct ceph_mds_request. We
need more flags though, but each bool takes (at least) a byte. Those
add up over time.

Merge all of the existing bools in this struct into a single unsigned
long, and use the set/test/clear_bit macros to manipulate them. These
are atomic operations, but that is required here to prevent
load/modify/store races. The existing flags are protected by different
locks, so we can't rely on them for that purpose.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/debugfs.c    |  2 +-
 fs/ceph/dir.c        |  4 ++--
 fs/ceph/inode.c      | 19 ++++++++++---------
 fs/ceph/mds_client.c | 47 +++++++++++++++++++++++++----------------------
 fs/ceph/mds_client.h | 12 ++++++++----
 5 files changed, 46 insertions(+), 38 deletions(-)

diff --git a/fs/ceph/debugfs.c b/fs/ceph/debugfs.c
index 39ff678e567f..f2ae393e2c31 100644
--- a/fs/ceph/debugfs.c
+++ b/fs/ceph/debugfs.c
@@ -70,7 +70,7 @@ static int mdsc_show(struct seq_file *s, void *p)
 
 		seq_printf(s, "%s", ceph_mds_op_name(req->r_op));
 
-		if (req->r_got_unsafe)
+		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 			seq_puts(s, "\t(unsafe)");
 		else
 			seq_puts(s, "\t");
diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index d4385563b70a..5fb86d71220e 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -371,7 +371,7 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
 		/* hints to request -> mds selection code */
 		req->r_direct_mode = USE_AUTH_MDS;
 		req->r_direct_hash = ceph_frag_value(frag);
-		req->r_direct_is_hash = true;
+		__set_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
 		if (fi->last_name) {
 			req->r_path2 = kstrdup(fi->last_name, GFP_KERNEL);
 			if (!req->r_path2) {
@@ -417,7 +417,7 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
 		fi->frag = frag;
 		fi->last_readdir = req;
 
-		if (req->r_did_prepopulate) {
+		if (test_bit(CEPH_MDS_R_DID_PREPOPULATE, &req->r_req_flags)) {
 			fi->readdir_cache_idx = req->r_readdir_cache_idx;
 			if (fi->readdir_cache_idx < 0) {
 				/* preclude from marking dir ordered */
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index b18462c64cdd..e54ba03d3df7 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1199,8 +1199,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 
 		err = fill_inode(in, req->r_locked_page, &rinfo->targeti, NULL,
 				session, req->r_request_started,
-				(!req->r_aborted && rinfo->head->result == 0) ?
-				req->r_fmode : -1,
+				(!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
+				rinfo->head->result == 0) ?  req->r_fmode : -1,
 				&req->r_caps_reservation);
 		if (err < 0) {
 			pr_err("fill_inode badness %p %llx.%llx\n",
@@ -1209,12 +1209,14 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 		}
 	}
 
+	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
+		goto done;
+
 	/*
 	 * ignore null lease/binding on snapdir ENOENT, or else we
 	 * will have trouble splicing in the virtual snapdir later
 	 */
-	if (rinfo->head->is_dentry && !req->r_aborted &&
-	    req->r_locked_dir &&
+	if (rinfo->head->is_dentry && req->r_locked_dir &&
 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
 					       fsc->mount_options->snapdir_name,
 					       req->r_dentry->d_name.len))) {
@@ -1317,9 +1319,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 			update_dentry_lease(dn, rinfo->dlease, session,
 					    req->r_request_started);
 		dout(" final dn %p\n", dn);
-	} else if (!req->r_aborted &&
-		   (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
-		    req->r_op == CEPH_MDS_OP_MKSNAP)) {
+	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
+		   req->r_op == CEPH_MDS_OP_MKSNAP) {
 		struct dentry *dn = req->r_dentry;
 		struct inode *dir = req->r_locked_dir;
 
@@ -1444,7 +1445,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 	u32 fpos_offset;
 	struct ceph_readdir_cache_control cache_ctl = {};
 
-	if (req->r_aborted)
+	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
 		return readdir_prepopulate_inodes_only(req, session);
 
 	if (rinfo->hash_order && req->r_path2) {
@@ -1598,7 +1599,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 	}
 out:
 	if (err == 0 && skipped == 0) {
-		req->r_did_prepopulate = true;
+		set_bit(CEPH_MDS_R_DID_PREPOPULATE, &req->r_req_flags);
 		req->r_readdir_cache_idx = cache_ctl.index;
 	}
 	ceph_readdir_cache_release(&cache_ctl);
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 4a68067a76b5..ccf75a3260e8 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -644,13 +644,15 @@ static void __unregister_request(struct ceph_mds_client *mdsc,
 
 	erase_request(&mdsc->request_tree, req);
 
-	if (req->r_unsafe_dir && req->r_got_unsafe) {
+	if (req->r_unsafe_dir  &&
+	    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 		struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
 		spin_lock(&ci->i_unsafe_lock);
 		list_del_init(&req->r_unsafe_dir_item);
 		spin_unlock(&ci->i_unsafe_lock);
 	}
-	if (req->r_target_inode && req->r_got_unsafe) {
+	if (req->r_target_inode &&
+	    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
 		spin_lock(&ci->i_unsafe_lock);
 		list_del_init(&req->r_unsafe_target_item);
@@ -705,7 +707,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
 	int mode = req->r_direct_mode;
 	int mds = -1;
 	u32 hash = req->r_direct_hash;
-	bool is_hash = req->r_direct_is_hash;
+	bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
 
 	/*
 	 * is there a specific mds we should try?  ignore hint if we have
@@ -2042,7 +2044,7 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
 	dout("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_got_unsafe) {
+	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 		void *p;
 		/*
 		 * Replay.  Do not regenerate message (and rebuild
@@ -2091,7 +2093,7 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
 
 	rhead = msg->front.iov_base;
 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
-	if (req->r_got_unsafe)
+	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 		flags |= CEPH_MDS_FLAG_REPLAY;
 	if (req->r_locked_dir)
 		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
@@ -2114,8 +2116,8 @@ static int __do_request(struct ceph_mds_client *mdsc,
 	int mds = -1;
 	int err = 0;
 
-	if (req->r_err || req->r_got_result) {
-		if (req->r_aborted)
+	if (req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
+		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
 			__unregister_request(mdsc, req);
 		goto out;
 	}
@@ -2245,7 +2247,7 @@ static void kick_requests(struct ceph_mds_client *mdsc, int mds)
 	while (p) {
 		req = rb_entry(p, struct ceph_mds_request, r_node);
 		p = rb_next(p);
-		if (req->r_got_unsafe)
+		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 			continue;
 		if (req->r_attempts > 0)
 			continue; /* only new requests */
@@ -2319,7 +2321,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 	mutex_lock(&mdsc->mutex);
 
 	/* only abort if we didn't race with a real reply */
-	if (req->r_got_result) {
+	if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
 		err = le32_to_cpu(req->r_reply_info.head->result);
 	} else if (err < 0) {
 		dout("aborted request %lld with %d\n", req->r_tid, err);
@@ -2331,7 +2333,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 		 */
 		mutex_lock(&req->r_fill_mutex);
 		req->r_err = err;
-		req->r_aborted = true;
+		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
 		mutex_unlock(&req->r_fill_mutex);
 
 		if (req->r_locked_dir &&
@@ -2409,14 +2411,14 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	}
 
 	/* dup? */
-	if ((req->r_got_unsafe && !head->safe) ||
-	    (req->r_got_safe && head->safe)) {
+	if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) ||
+	    (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags) && head->safe)) {
 		pr_warn("got a dup %s reply on %llu from mds%d\n",
 			   head->safe ? "safe" : "unsafe", tid, mds);
 		mutex_unlock(&mdsc->mutex);
 		goto out;
 	}
-	if (req->r_got_safe) {
+	if (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags)) {
 		pr_warn("got unsafe after safe on %llu from mds%d\n",
 			   tid, mds);
 		mutex_unlock(&mdsc->mutex);
@@ -2455,10 +2457,10 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 
 
 	if (head->safe) {
-		req->r_got_safe = true;
+		set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags);
 		__unregister_request(mdsc, req);
 
-		if (req->r_got_unsafe) {
+		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 			/*
 			 * We already handled the unsafe response, now do the
 			 * cleanup.  No need to examine the response; the MDS
@@ -2476,7 +2478,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 			goto out;
 		}
 	} else {
-		req->r_got_unsafe = true;
+		set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags);
 		list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
 		if (req->r_unsafe_dir) {
 			struct ceph_inode_info *ci =
@@ -2530,7 +2532,8 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	if (realm)
 		ceph_put_snap_realm(mdsc, realm);
 
-	if (err == 0 && req->r_got_unsafe && req->r_target_inode) {
+	if (err == 0 && req->r_target_inode &&
+	    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
 		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
 		spin_lock(&ci->i_unsafe_lock);
 		list_add_tail(&req->r_unsafe_target_item, &ci->i_unsafe_iops);
@@ -2538,12 +2541,12 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	}
 out_err:
 	mutex_lock(&mdsc->mutex);
-	if (!req->r_aborted) {
+	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
 		if (err) {
 			req->r_err = err;
 		} else {
 			req->r_reply =  ceph_msg_get(msg);
-			req->r_got_result = true;
+			set_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
 		}
 	} else {
 		dout("reply arrived after request %lld was aborted\n", tid);
@@ -2587,7 +2590,7 @@ static void handle_forward(struct ceph_mds_client *mdsc,
 		goto out;  /* dup reply? */
 	}
 
-	if (req->r_aborted) {
+	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
 		dout("forward tid %llu aborted, unregistering\n", tid);
 		__unregister_request(mdsc, req);
 	} else if (fwd_seq <= req->r_num_fwd) {
@@ -2597,7 +2600,7 @@ static void handle_forward(struct ceph_mds_client *mdsc,
 		/* resend. forward race not possible; mds would drop */
 		dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
 		BUG_ON(req->r_err);
-		BUG_ON(req->r_got_result);
+		BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags));
 		req->r_attempts = 0;
 		req->r_num_fwd = fwd_seq;
 		req->r_resend_mds = next_mds;
@@ -2762,7 +2765,7 @@ static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
 	while (p) {
 		req = rb_entry(p, struct ceph_mds_request, r_node);
 		p = rb_next(p);
-		if (req->r_got_unsafe)
+		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 			continue;
 		if (req->r_attempts == 0)
 			continue; /* only old requests */
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 3c6f77b7bb02..409b0e3c3b7a 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -205,6 +205,14 @@ struct ceph_mds_request {
 	struct inode *r_locked_dir; /* dir (if any) i_mutex locked by vfs */
 	struct inode *r_target_inode;       /* resulting inode */
 
+#define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
+#define CEPH_MDS_R_ABORTED		(2) /* call was aborted */
+#define CEPH_MDS_R_GOT_UNSAFE		(3) /* got an unsafe reply */
+#define CEPH_MDS_R_GOT_SAFE		(4) /* got a safe reply */
+#define CEPH_MDS_R_GOT_RESULT		(5) /* got a result */
+#define CEPH_MDS_R_DID_PREPOPULATE	(6) /* prepopulated readdir */
+	unsigned long	r_req_flags;
+
 	struct mutex r_fill_mutex;
 
 	union ceph_mds_request_args r_args;
@@ -216,7 +224,6 @@ struct ceph_mds_request {
 	/* for choosing which mds to send this request to */
 	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 */
 
 	/* data payload is used for xattr ops */
 	struct ceph_pagelist *r_pagelist;
@@ -234,7 +241,6 @@ struct ceph_mds_request {
 	struct ceph_mds_reply_info_parsed r_reply_info;
 	struct page *r_locked_page;
 	int r_err;
-	bool r_aborted;
 
 	unsigned long r_timeout;  /* optional.  jiffies, 0 is "wait forever" */
 	unsigned long r_started;  /* start time to measure timeout against */
@@ -262,9 +268,7 @@ struct ceph_mds_request {
 	ceph_mds_request_callback_t r_callback;
 	ceph_mds_request_wait_callback_t r_wait_for_completion;
 	struct list_head  r_unsafe_item;  /* per-session unsafe list item */
-	bool		  r_got_unsafe, r_got_safe, r_got_result;
 
-	bool              r_did_prepopulate;
 	long long	  r_dir_release_cnt;
 	long long	  r_dir_ordered_cnt;
 	int		  r_readdir_cache_idx;
-- 
2.9.3

[PATCH v3 4/8] ceph: add a new flag to indicate whether parent is locked

[Jeff Layton]

struct ceph_mds_request has an r_locked_dir pointer, which is set to
indicate the parent inode and that its i_rwsem is locked.  In some
critical places, we need to be able to indicate the parent inode to the
request handling code, even when its i_rwsem may not be locked.

Most of the code that operates on r_locked_dir doesn't require that the
i_rwsem be locked. We only really need it to handle manipulation of the
dcache. The rest (filling of the inode, updating dentry leases, etc.)
already has its own locking.

Add a new r_req_flags bit that indicates whether the parent is locked
when doing the request, and rename the pointer to "r_parent". For now,
all the places that set r_parent also set this flag, but that will
change in a later patch.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/dir.c        | 21 ++++++++++++++-------
 fs/ceph/export.c     |  3 ++-
 fs/ceph/file.c       |  3 ++-
 fs/ceph/inode.c      | 17 ++++++++++-------
 fs/ceph/mds_client.c | 24 ++++++++++++------------
 fs/ceph/mds_client.h |  3 ++-
 6 files changed, 42 insertions(+), 29 deletions(-)

diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index 5fb86d71220e..c8562b8a8c7e 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -752,7 +752,8 @@ static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
 		mask |= CEPH_CAP_XATTR_SHARED;
 	req->r_args.getattr.mask = cpu_to_le32(mask);
 
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	err = ceph_mdsc_do_request(mdsc, NULL, req);
 	err = ceph_handle_snapdir(req, dentry, err);
 	dentry = ceph_finish_lookup(req, dentry, err);
@@ -813,7 +814,8 @@ static int ceph_mknod(struct inode *dir, struct dentry *dentry,
 	}
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_args.mknod.mode = cpu_to_le32(mode);
 	req->r_args.mknod.rdev = cpu_to_le32(rdev);
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
@@ -864,7 +866,8 @@ static int ceph_symlink(struct inode *dir, struct dentry *dentry,
 		ceph_mdsc_put_request(req);
 		goto out;
 	}
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
@@ -913,7 +916,8 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, umode_t mode)
 
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_args.mkdir.mode = cpu_to_le32(mode);
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
@@ -957,7 +961,8 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
 	req->r_old_dentry = dget(old_dentry);
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
 	/* release LINK_SHARED on source inode (mds will lock it) */
@@ -1023,7 +1028,8 @@ static int ceph_unlink(struct inode *dir, struct dentry *dentry)
 	}
 	req->r_dentry = dget(dentry);
 	req->r_num_caps = 2;
-	req->r_locked_dir = dir;
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
 	req->r_inode_drop = drop_caps_for_unlink(inode);
@@ -1066,7 +1072,8 @@ static int ceph_rename(struct inode *old_dir, struct dentry *old_dentry,
 	req->r_num_caps = 2;
 	req->r_old_dentry = dget(old_dentry);
 	req->r_old_dentry_dir = old_dir;
-	req->r_locked_dir = new_dir;
+	req->r_parent = new_dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_old_dentry_drop = CEPH_CAP_FILE_SHARED;
 	req->r_old_dentry_unless = CEPH_CAP_FILE_EXCL;
 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
diff --git a/fs/ceph/export.c b/fs/ceph/export.c
index 180bbef760f2..e8f11fa565c5 100644
--- a/fs/ceph/export.c
+++ b/fs/ceph/export.c
@@ -207,7 +207,8 @@ static int ceph_get_name(struct dentry *parent, char *name,
 	req->r_inode = d_inode(child);
 	ihold(d_inode(child));
 	req->r_ino2 = ceph_vino(d_inode(parent));
-	req->r_locked_dir = d_inode(parent);
+	req->r_parent = d_inode(parent);
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	req->r_num_caps = 2;
 	err = ceph_mdsc_do_request(mdsc, NULL, req);
 
diff --git a/fs/ceph/file.c b/fs/ceph/file.c
index 19419eb87f38..a91239c4b31c 100644
--- a/fs/ceph/file.c
+++ b/fs/ceph/file.c
@@ -310,7 +310,8 @@ int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
                mask |= CEPH_CAP_XATTR_SHARED;
        req->r_args.open.mask = cpu_to_le32(mask);
 
-	req->r_locked_dir = dir;           /* caller holds dir->i_mutex */
+	req->r_parent = dir;
+	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
 	err = ceph_mdsc_do_request(mdsc,
 				   (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
 				   req);
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index e54ba03d3df7..b4b61da03dbf 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1122,13 +1122,13 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 
 	if (!rinfo->head->is_target && !rinfo->head->is_dentry) {
 		dout("fill_trace reply is empty!\n");
-		if (rinfo->head->result == 0 && req->r_locked_dir)
+		if (rinfo->head->result == 0 && req->r_parent)
 			ceph_invalidate_dir_request(req);
 		return 0;
 	}
 
 	if (rinfo->head->is_dentry) {
-		struct inode *dir = req->r_locked_dir;
+		struct inode *dir = req->r_parent;
 
 		if (dir) {
 			err = fill_inode(dir, NULL,
@@ -1216,7 +1216,9 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 	 * ignore null lease/binding on snapdir ENOENT, or else we
 	 * will have trouble splicing in the virtual snapdir later
 	 */
-	if (rinfo->head->is_dentry && req->r_locked_dir &&
+	if (rinfo->head->is_dentry &&
+            !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
+	    test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags) &&
 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
 					       fsc->mount_options->snapdir_name,
 					       req->r_dentry->d_name.len))) {
@@ -1225,7 +1227,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 		 * mknod symlink mkdir  : null -> new inode
 		 * unlink               : linked -> null
 		 */
-		struct inode *dir = req->r_locked_dir;
+		struct inode *dir = req->r_parent;
 		struct dentry *dn = req->r_dentry;
 		bool have_dir_cap, have_lease;
 
@@ -1319,10 +1321,11 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 			update_dentry_lease(dn, rinfo->dlease, session,
 					    req->r_request_started);
 		dout(" final dn %p\n", dn);
-	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
-		   req->r_op == CEPH_MDS_OP_MKSNAP) {
+	} else if ((req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
+		    req->r_op == CEPH_MDS_OP_MKSNAP) &&
+		   !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
 		struct dentry *dn = req->r_dentry;
-		struct inode *dir = req->r_locked_dir;
+		struct inode *dir = req->r_parent;
 
 		/* fill out a snapdir LOOKUPSNAP dentry */
 		BUG_ON(!dn);
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index ccf75a3260e8..52521f339745 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -547,8 +547,8 @@ void ceph_mdsc_release_request(struct kref *kref)
 		ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
 		iput(req->r_inode);
 	}
-	if (req->r_locked_dir)
-		ceph_put_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
+	if (req->r_parent)
+		ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
 	iput(req->r_target_inode);
 	if (req->r_dentry)
 		dput(req->r_dentry);
@@ -735,7 +735,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
 
 		rcu_read_lock();
 		parent = req->r_dentry->d_parent;
-		dir = req->r_locked_dir ? : d_inode_rcu(parent);
+		dir = req->r_parent ? : d_inode_rcu(parent);
 
 		if (!dir || dir->i_sb != mdsc->fsc->sb) {
 			/*  not this fs or parent went negative */
@@ -1894,7 +1894,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
 	int ret;
 
 	ret = set_request_path_attr(req->r_inode, req->r_dentry,
-			      req->r_locked_dir, req->r_path1, req->r_ino1.ino,
+			      req->r_parent, req->r_path1, req->r_ino1.ino,
 			      &path1, &pathlen1, &ino1, &freepath1);
 	if (ret < 0) {
 		msg = ERR_PTR(ret);
@@ -1956,7 +1956,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
 		      mds, req->r_inode_drop, req->r_inode_unless, 0);
 	if (req->r_dentry_drop)
 		releases += ceph_encode_dentry_release(&p, req->r_dentry,
-				req->r_locked_dir, mds, req->r_dentry_drop,
+				req->r_parent, mds, req->r_dentry_drop,
 				req->r_dentry_unless);
 	if (req->r_old_dentry_drop)
 		releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
@@ -2095,14 +2095,14 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
 	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
 		flags |= CEPH_MDS_FLAG_REPLAY;
-	if (req->r_locked_dir)
+	if (req->r_parent)
 		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
 	rhead->flags = cpu_to_le32(flags);
 	rhead->num_fwd = req->r_num_fwd;
 	rhead->num_retry = req->r_attempts - 1;
 	rhead->ino = 0;
 
-	dout(" r_locked_dir = %p\n", req->r_locked_dir);
+	dout(" r_parent = %p\n", req->r_parent);
 	return 0;
 }
 
@@ -2282,11 +2282,11 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 
 	dout("do_request on %p\n", req);
 
-	/* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
+	/* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
 	if (req->r_inode)
 		ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
-	if (req->r_locked_dir)
-		ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
+	if (req->r_parent)
+		ceph_get_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
 	if (req->r_old_dentry_dir)
 		ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
 				  CEPH_CAP_PIN);
@@ -2336,7 +2336,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
 		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
 		mutex_unlock(&req->r_fill_mutex);
 
-		if (req->r_locked_dir &&
+		if (req->r_parent &&
 		    (req->r_op & CEPH_MDS_OP_WRITE))
 			ceph_invalidate_dir_request(req);
 	} else {
@@ -2355,7 +2355,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
  */
 void ceph_invalidate_dir_request(struct ceph_mds_request *req)
 {
-	struct inode *inode = req->r_locked_dir;
+	struct inode *inode = req->r_parent;
 
 	dout("invalidate_dir_request %p (complete, lease(s))\n", inode);
 
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 409b0e3c3b7a..ac0475a2daa7 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -202,7 +202,7 @@ struct ceph_mds_request {
 	char *r_path1, *r_path2;
 	struct ceph_vino r_ino1, r_ino2;
 
-	struct inode *r_locked_dir; /* dir (if any) i_mutex locked by vfs */
+	struct inode *r_parent;		    /* parent dir inode */
 	struct inode *r_target_inode;       /* resulting inode */
 
 #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
@@ -211,6 +211,7 @@ struct ceph_mds_request {
 #define CEPH_MDS_R_GOT_SAFE		(4) /* got a safe reply */
 #define CEPH_MDS_R_GOT_RESULT		(5) /* got a result */
 #define CEPH_MDS_R_DID_PREPOPULATE	(6) /* prepopulated readdir */
+#define CEPH_MDS_R_PARENT_LOCKED	(7) /* is r_parent->i_rwsem wlocked? */
 	unsigned long	r_req_flags;
 
 	struct mutex r_fill_mutex;
-- 
2.9.3

[PATCH v3 5/8] ceph: don't update_dentry_lease unless we actually got one

[Jeff Layton]

This if block updates the dentry lease even in the case where
the MDS didn't grant one.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index b4b61da03dbf..e5c1ca02dbe3 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1289,8 +1289,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 				ceph_dir_clear_ordered(dir);
 				dout("d_delete %p\n", dn);
 				d_delete(dn);
-			} else {
-				if (have_lease && d_unhashed(dn))
+			} else if (have_lease) {
+				if (d_unhashed(dn))
 					d_add(dn, NULL);
 				update_dentry_lease(dn, rinfo->dlease,
 						    session,
-- 
2.9.3

[PATCH v3 6/8] ceph: vet the target and parent inodes before updating dentry lease

[Jeff Layton]

In a later patch, we're going to need to allow ceph_fill_trace to
update the dentry's lease when the parent is not locked. This is
potentially racy though -- by the time we get around to processing the
trace, the parent may have already changed.

Change update_dentry_lease to take a ceph_vino pointer and use that to
ensure that the dentry's parent still matches it before updating the
lease.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 72 ++++++++++++++++++++++++++++++++++++++-------------------
 1 file changed, 48 insertions(+), 24 deletions(-)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index e5c1ca02dbe3..41cbddd22091 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1016,7 +1016,9 @@ static int fill_inode(struct inode *inode, struct page *locked_page,
 static void update_dentry_lease(struct dentry *dentry,
 				struct ceph_mds_reply_lease *lease,
 				struct ceph_mds_session *session,
-				unsigned long from_time)
+				unsigned long from_time,
+				struct ceph_vino *tgt_vino,
+				struct ceph_vino *dir_vino)
 {
 	struct ceph_dentry_info *di = ceph_dentry(dentry);
 	long unsigned duration = le32_to_cpu(lease->duration_ms);
@@ -1024,13 +1026,27 @@ static void update_dentry_lease(struct dentry *dentry,
 	long unsigned half_ttl = from_time + (duration * HZ / 2) / 1000;
 	struct inode *dir;
 
+	/*
+	 * Make sure dentry's inode matches tgt_vino. NULL tgt_vino means that
+	 * we expect a negative dentry.
+	 */
+	if (!tgt_vino && d_really_is_positive(dentry))
+		return;
+
+	if (tgt_vino && (d_really_is_negative(dentry) ||
+			!ceph_ino_compare(d_inode(dentry), tgt_vino)))
+		return;
+
 	spin_lock(&dentry->d_lock);
 	dout("update_dentry_lease %p duration %lu ms ttl %lu\n",
 	     dentry, duration, ttl);
 
-	/* make lease_rdcache_gen match directory */
 	dir = d_inode(dentry->d_parent);
 
+	/* make sure parent matches dir_vino */
+	if (!ceph_ino_compare(dir, dir_vino))
+		goto out_unlock;
+
 	/* only track leases on regular dentries */
 	if (ceph_snap(dir) != CEPH_NOSNAP)
 		goto out_unlock;
@@ -1113,7 +1129,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 	struct ceph_mds_session *session = req->r_session;
 	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
 	struct inode *in = NULL;
-	struct ceph_vino vino;
+	struct ceph_vino tvino, dvino;
 	struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
 	int err = 0;
 
@@ -1154,8 +1170,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 			dname.name = rinfo->dname;
 			dname.len = rinfo->dname_len;
 			dname.hash = full_name_hash(parent, dname.name, dname.len);
-			vino.ino = le64_to_cpu(rinfo->targeti.in->ino);
-			vino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
+			tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
+			tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
 retry_lookup:
 			dn = d_lookup(parent, &dname);
 			dout("d_lookup on parent=%p name=%.*s got %p\n",
@@ -1172,8 +1188,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 				}
 				err = 0;
 			} else if (d_really_is_positive(dn) &&
-				   (ceph_ino(d_inode(dn)) != vino.ino ||
-				    ceph_snap(d_inode(dn)) != vino.snap)) {
+				   (ceph_ino(d_inode(dn)) != tvino.ino ||
+				    ceph_snap(d_inode(dn)) != tvino.snap)) {
 				dout(" dn %p points to wrong inode %p\n",
 				     dn, d_inode(dn));
 				d_delete(dn);
@@ -1187,10 +1203,10 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 	}
 
 	if (rinfo->head->is_target) {
-		vino.ino = le64_to_cpu(rinfo->targeti.in->ino);
-		vino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
+		tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
+		tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
 
-		in = ceph_get_inode(sb, vino);
+		in = ceph_get_inode(sb, tvino);
 		if (IS_ERR(in)) {
 			err = PTR_ERR(in);
 			goto done;
@@ -1234,10 +1250,12 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 		BUG_ON(!dn);
 		BUG_ON(!dir);
 		BUG_ON(d_inode(dn->d_parent) != dir);
-		BUG_ON(ceph_ino(dir) !=
-		       le64_to_cpu(rinfo->diri.in->ino));
-		BUG_ON(ceph_snap(dir) !=
-		       le64_to_cpu(rinfo->diri.in->snapid));
+
+		dvino.ino = le64_to_cpu(rinfo->diri.in->ino);
+		dvino.snap = le64_to_cpu(rinfo->diri.in->snapid);
+
+		BUG_ON(ceph_ino(dir) != dvino.ino);
+		BUG_ON(ceph_snap(dir) != dvino.snap);
 
 		/* do we have a lease on the whole dir? */
 		have_dir_cap =
@@ -1294,7 +1312,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 					d_add(dn, NULL);
 				update_dentry_lease(dn, rinfo->dlease,
 						    session,
-						    req->r_request_started);
+						    req->r_request_started,
+						    NULL, &dvino);
 			}
 			goto done;
 		}
@@ -1317,9 +1336,13 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 			have_lease = false;
 		}
 
-		if (have_lease)
+		if (have_lease) {
+			tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
+			tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
 			update_dentry_lease(dn, rinfo->dlease, session,
-					    req->r_request_started);
+					    req->r_request_started,
+					    &tvino, &dvino);
+		}
 		dout(" final dn %p\n", dn);
 	} else if ((req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
 		    req->r_op == CEPH_MDS_OP_MKSNAP) &&
@@ -1493,14 +1516,14 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 	/* FIXME: release caps/leases if error occurs */
 	for (i = 0; i < rinfo->dir_nr; i++) {
 		struct ceph_mds_reply_dir_entry *rde = rinfo->dir_entries + i;
-		struct ceph_vino vino;
+		struct ceph_vino tvino, dvino;
 
 		dname.name = rde->name;
 		dname.len = rde->name_len;
 		dname.hash = full_name_hash(parent, dname.name, dname.len);
 
-		vino.ino = le64_to_cpu(rde->inode.in->ino);
-		vino.snap = le64_to_cpu(rde->inode.in->snapid);
+		tvino.ino = le64_to_cpu(rde->inode.in->ino);
+		tvino.snap = le64_to_cpu(rde->inode.in->snapid);
 
 		if (rinfo->hash_order) {
 			u32 hash = ceph_str_hash(ci->i_dir_layout.dl_dir_hash,
@@ -1529,8 +1552,8 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 				goto out;
 			}
 		} else if (d_really_is_positive(dn) &&
-			   (ceph_ino(d_inode(dn)) != vino.ino ||
-			    ceph_snap(d_inode(dn)) != vino.snap)) {
+			   (ceph_ino(d_inode(dn)) != tvino.ino ||
+			    ceph_snap(d_inode(dn)) != tvino.snap)) {
 			dout(" dn %p points to wrong inode %p\n",
 			     dn, d_inode(dn));
 			d_delete(dn);
@@ -1542,7 +1565,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 		if (d_really_is_positive(dn)) {
 			in = d_inode(dn);
 		} else {
-			in = ceph_get_inode(parent->d_sb, vino);
+			in = ceph_get_inode(parent->d_sb, tvino);
 			if (IS_ERR(in)) {
 				dout("new_inode badness\n");
 				d_drop(dn);
@@ -1587,8 +1610,9 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
 
 		ceph_dentry(dn)->offset = rde->offset;
 
+		dvino = ceph_vino(d_inode(parent));
 		update_dentry_lease(dn, rde->lease, req->r_session,
-				    req->r_request_started);
+				    req->r_request_started, &tvino, &dvino);
 
 		if (err == 0 && skipped == 0 && cache_ctl.index >= 0) {
 			ret = fill_readdir_cache(d_inode(parent), dn,
-- 
2.9.3

[PATCH v3 7/8] ceph: call update_dentry_lease even when r_locked dir is not set

[Jeff Layton]

We don't really require that the parent be locked in order to update the
lease on a dentry. Lease info is protected by the d_lock. In the event
that the parent is not locked in ceph_fill_trace, and we have both
parent and target info, go ahead and update the dentry lease.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/inode.c | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 41cbddd22091..b5de158abaaa 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1363,6 +1363,26 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 			goto done;
 		}
 		req->r_dentry = dn;  /* may have spliced */
+	} else if (rinfo->head->is_dentry) {
+		struct ceph_vino *ptvino = NULL;
+
+		if ((le32_to_cpu(rinfo->diri.in->cap.caps) & CEPH_CAP_FILE_SHARED) ||
+		    le32_to_cpu(rinfo->dlease->duration_ms)) {
+			dvino.ino = le64_to_cpu(rinfo->diri.in->ino);
+			dvino.snap = le64_to_cpu(rinfo->diri.in->snapid);
+
+			if (rinfo->head->is_target) {
+				tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
+				tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
+				ptvino = &tvino;
+			}
+
+			update_dentry_lease(req->r_dentry, rinfo->dlease,
+				session, req->r_request_started, ptvino,
+				&dvino);
+		} else {
+			dout("%s: no dentry lease or dir cap\n", __func__);
+		}
 	}
 done:
 	dout("fill_trace done err=%d\n", err);
-- 
2.9.3

[PATCH v3 8/8] ceph: do a LOOKUP in d_revalidate instead of GETATTR

[Jeff Layton]

In commit c3f4688a08f (ceph: don't set req->r_locked_dir in
ceph_d_revalidate), we changed the code to do a GETATTR instead of a
LOOKUP as the parent info isn't strictly necessary to revalidate the
dentry. What we missed there though is that in order to update the lease
on the dentry after revalidating it, we _do_ need parent info.

Change ceph_d_revalidate back to doing a LOOKUP instead of a GETATTR so
that we can get the parent info in order to update the lease from
ceph_fill_trace. Note that we set req->r_parent here, but we cannot set
the CEPH_MDS_R_PARENT_LOCKED flag as we can't guarantee that it is.

Signed-off-by: Jeff Layton <jlayton@redhat.com>
---
 fs/ceph/dir.c | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index c8562b8a8c7e..3e9ad501addf 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -1244,11 +1244,12 @@ static int ceph_d_revalidate(struct dentry *dentry, unsigned int flags)
 			return -ECHILD;
 
 		op = ceph_snap(dir) == CEPH_SNAPDIR ?
-			CEPH_MDS_OP_LOOKUPSNAP : CEPH_MDS_OP_GETATTR;
+			CEPH_MDS_OP_LOOKUPSNAP : CEPH_MDS_OP_LOOKUP;
 		req = ceph_mdsc_create_request(mdsc, op, USE_ANY_MDS);
 		if (!IS_ERR(req)) {
 			req->r_dentry = dget(dentry);
-			req->r_num_caps = op == CEPH_MDS_OP_GETATTR ? 1 : 2;
+			req->r_num_caps = 2;
+			req->r_parent = dir;
 
 			mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED;
 			if (ceph_security_xattr_wanted(dir))
-- 
2.9.3

Re: [PATCH v3 4/8] ceph: add a new flag to indicate whether parent is locked

[Yan, Zheng]

> On 4 Feb 2017, at 02:04, Jeff Layton <jlayton@redhat.com> wrote:
> 
> struct ceph_mds_request has an r_locked_dir pointer, which is set to
> indicate the parent inode and that its i_rwsem is locked.  In some
> critical places, we need to be able to indicate the parent inode to the
> request handling code, even when its i_rwsem may not be locked.
> 
> Most of the code that operates on r_locked_dir doesn't require that the
> i_rwsem be locked. We only really need it to handle manipulation of the
> dcache. The rest (filling of the inode, updating dentry leases, etc.)
> already has its own locking.
> 
> Add a new r_req_flags bit that indicates whether the parent is locked
> when doing the request, and rename the pointer to "r_parent". For now,
> all the places that set r_parent also set this flag, but that will
> change in a later patch.
> 
> Signed-off-by: Jeff Layton <jlayton@redhat.com>
> ---
> fs/ceph/dir.c        | 21 ++++++++++++++-------
> fs/ceph/export.c     |  3 ++-
> fs/ceph/file.c       |  3 ++-
> fs/ceph/inode.c      | 17 ++++++++++-------
> fs/ceph/mds_client.c | 24 ++++++++++++------------
> fs/ceph/mds_client.h |  3 ++-
> 6 files changed, 42 insertions(+), 29 deletions(-)
> 
> diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
> index 5fb86d71220e..c8562b8a8c7e 100644
> --- a/fs/ceph/dir.c
> +++ b/fs/ceph/dir.c
> @@ -752,7 +752,8 @@ static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
> 		mask |= CEPH_CAP_XATTR_SHARED;
> 	req->r_args.getattr.mask = cpu_to_le32(mask);
> 
> -	req->r_locked_dir = dir;
> +	req->r_parent = dir;
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	err = ceph_mdsc_do_request(mdsc, NULL, req);
> 	err = ceph_handle_snapdir(req, dentry, err);
> 	dentry = ceph_finish_lookup(req, dentry, err);
> @@ -813,7 +814,8 @@ static int ceph_mknod(struct inode *dir, struct dentry *dentry,
> 	}
> 	req->r_dentry = dget(dentry);
> 	req->r_num_caps = 2;
> -	req->r_locked_dir = dir;
> +	req->r_parent = dir;
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	req->r_args.mknod.mode = cpu_to_le32(mode);
> 	req->r_args.mknod.rdev = cpu_to_le32(rdev);
> 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> @@ -864,7 +866,8 @@ static int ceph_symlink(struct inode *dir, struct dentry *dentry,
> 		ceph_mdsc_put_request(req);
> 		goto out;
> 	}
> -	req->r_locked_dir = dir;
> +	req->r_parent = dir;
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	req->r_dentry = dget(dentry);
> 	req->r_num_caps = 2;
> 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> @@ -913,7 +916,8 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, umode_t mode)
> 
> 	req->r_dentry = dget(dentry);
> 	req->r_num_caps = 2;
> -	req->r_locked_dir = dir;
> +	req->r_parent = dir;
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	req->r_args.mkdir.mode = cpu_to_le32(mode);
> 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
> @@ -957,7 +961,8 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
> 	req->r_dentry = dget(dentry);
> 	req->r_num_caps = 2;
> 	req->r_old_dentry = dget(old_dentry);
> -	req->r_locked_dir = dir;
> +	req->r_parent = dir;
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
> 	/* release LINK_SHARED on source inode (mds will lock it) */
> @@ -1023,7 +1028,8 @@ static int ceph_unlink(struct inode *dir, struct dentry *dentry)
> 	}
> 	req->r_dentry = dget(dentry);
> 	req->r_num_caps = 2;
> -	req->r_locked_dir = dir;
> +	req->r_parent = dir;
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
> 	req->r_inode_drop = drop_caps_for_unlink(inode);
> @@ -1066,7 +1072,8 @@ static int ceph_rename(struct inode *old_dir, struct dentry *old_dentry,
> 	req->r_num_caps = 2;
> 	req->r_old_dentry = dget(old_dentry);
> 	req->r_old_dentry_dir = old_dir;
> -	req->r_locked_dir = new_dir;
> +	req->r_parent = new_dir;
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	req->r_old_dentry_drop = CEPH_CAP_FILE_SHARED;
> 	req->r_old_dentry_unless = CEPH_CAP_FILE_EXCL;
> 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> diff --git a/fs/ceph/export.c b/fs/ceph/export.c
> index 180bbef760f2..e8f11fa565c5 100644
> --- a/fs/ceph/export.c
> +++ b/fs/ceph/export.c
> @@ -207,7 +207,8 @@ static int ceph_get_name(struct dentry *parent, char *name,
> 	req->r_inode = d_inode(child);
> 	ihold(d_inode(child));
> 	req->r_ino2 = ceph_vino(d_inode(parent));
> -	req->r_locked_dir = d_inode(parent);
> +	req->r_parent = d_inode(parent);
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	req->r_num_caps = 2;
> 	err = ceph_mdsc_do_request(mdsc, NULL, req);
> 
> diff --git a/fs/ceph/file.c b/fs/ceph/file.c
> index 19419eb87f38..a91239c4b31c 100644
> --- a/fs/ceph/file.c
> +++ b/fs/ceph/file.c
> @@ -310,7 +310,8 @@ int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
>                mask |= CEPH_CAP_XATTR_SHARED;
>        req->r_args.open.mask = cpu_to_le32(mask);
> 
> -	req->r_locked_dir = dir;           /* caller holds dir->i_mutex */
> +	req->r_parent = dir;
> +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> 	err = ceph_mdsc_do_request(mdsc,
> 				   (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
> 				   req);
> diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> index e54ba03d3df7..b4b61da03dbf 100644
> --- a/fs/ceph/inode.c
> +++ b/fs/ceph/inode.c
> @@ -1122,13 +1122,13 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 
> 	if (!rinfo->head->is_target && !rinfo->head->is_dentry) {
> 		dout("fill_trace reply is empty!\n");
> -		if (rinfo->head->result == 0 && req->r_locked_dir)
> +		if (rinfo->head->result == 0 && req->r_parent)
> 			ceph_invalidate_dir_request(req);
> 		return 0;
> 	}
> 
> 	if (rinfo->head->is_dentry) {
> -		struct inode *dir = req->r_locked_dir;
> +		struct inode *dir = req->r_parent;
> 
> 		if (dir) {
> 			err = fill_inode(dir, NULL,
> @@ -1216,7 +1216,9 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 	 * ignore null lease/binding on snapdir ENOENT, or else we
> 	 * will have trouble splicing in the virtual snapdir later
> 	 */
> -	if (rinfo->head->is_dentry && req->r_locked_dir &&
> +	if (rinfo->head->is_dentry &&
> +            !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&

why adding the !ABORTED check back. I think previous patch add:

if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
  goto done;

> +	    test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags) &&
> 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
> 					       fsc->mount_options->snapdir_name,
> 					       req->r_dentry->d_name.len))) {
> @@ -1225,7 +1227,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 		 * mknod symlink mkdir  : null -> new inode
> 		 * unlink               : linked -> null
> 		 */
> -		struct inode *dir = req->r_locked_dir;
> +		struct inode *dir = req->r_parent;
> 		struct dentry *dn = req->r_dentry;
> 		bool have_dir_cap, have_lease;
> 
> @@ -1319,10 +1321,11 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 			update_dentry_lease(dn, rinfo->dlease, session,
> 					    req->r_request_started);
> 		dout(" final dn %p\n", dn);
> -	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> -		   req->r_op == CEPH_MDS_OP_MKSNAP) {
> +	} else if ((req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> +		    req->r_op == CEPH_MDS_OP_MKSNAP) &&
> +		   !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {

same here

> 		struct dentry *dn = req->r_dentry;
> -		struct inode *dir = req->r_locked_dir;
> +		struct inode *dir = req->r_parent;
> 
> 		/* fill out a snapdir LOOKUPSNAP dentry */
> 		BUG_ON(!dn);
> diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
> index ccf75a3260e8..52521f339745 100644
> --- a/fs/ceph/mds_client.c
> +++ b/fs/ceph/mds_client.c
> @@ -547,8 +547,8 @@ void ceph_mdsc_release_request(struct kref *kref)
> 		ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
> 		iput(req->r_inode);
> 	}
> -	if (req->r_locked_dir)
> -		ceph_put_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
> +	if (req->r_parent)
> +		ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
> 	iput(req->r_target_inode);
> 	if (req->r_dentry)
> 		dput(req->r_dentry);
> @@ -735,7 +735,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
> 
> 		rcu_read_lock();
> 		parent = req->r_dentry->d_parent;
> -		dir = req->r_locked_dir ? : d_inode_rcu(parent);
> +		dir = req->r_parent ? : d_inode_rcu(parent);
> 
> 		if (!dir || dir->i_sb != mdsc->fsc->sb) {
> 			/*  not this fs or parent went negative */
> @@ -1894,7 +1894,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
> 	int ret;
> 
> 	ret = set_request_path_attr(req->r_inode, req->r_dentry,
> -			      req->r_locked_dir, req->r_path1, req->r_ino1.ino,
> +			      req->r_parent, req->r_path1, req->r_ino1.ino,
> 			      &path1, &pathlen1, &ino1, &freepath1);
> 	if (ret < 0) {
> 		msg = ERR_PTR(ret);
> @@ -1956,7 +1956,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
> 		      mds, req->r_inode_drop, req->r_inode_unless, 0);
> 	if (req->r_dentry_drop)
> 		releases += ceph_encode_dentry_release(&p, req->r_dentry,
> -				req->r_locked_dir, mds, req->r_dentry_drop,
> +				req->r_parent, mds, req->r_dentry_drop,
> 				req->r_dentry_unless);
> 	if (req->r_old_dentry_drop)
> 		releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
> @@ -2095,14 +2095,14 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
> 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
> 	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
> 		flags |= CEPH_MDS_FLAG_REPLAY;
> -	if (req->r_locked_dir)
> +	if (req->r_parent)
> 		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
> 	rhead->flags = cpu_to_le32(flags);
> 	rhead->num_fwd = req->r_num_fwd;
> 	rhead->num_retry = req->r_attempts - 1;
> 	rhead->ino = 0;
> 
> -	dout(" r_locked_dir = %p\n", req->r_locked_dir);
> +	dout(" r_parent = %p\n", req->r_parent);
> 	return 0;
> }
> 
> @@ -2282,11 +2282,11 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
> 
> 	dout("do_request on %p\n", req);
> 
> -	/* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
> +	/* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
> 	if (req->r_inode)
> 		ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
> -	if (req->r_locked_dir)
> -		ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
> +	if (req->r_parent)
> +		ceph_get_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
> 	if (req->r_old_dentry_dir)
> 		ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
> 				  CEPH_CAP_PIN);
> @@ -2336,7 +2336,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
> 		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
> 		mutex_unlock(&req->r_fill_mutex);
> 
> -		if (req->r_locked_dir &&
> +		if (req->r_parent &&
> 		    (req->r_op & CEPH_MDS_OP_WRITE))
> 			ceph_invalidate_dir_request(req);
> 	} else {
> @@ -2355,7 +2355,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
>  */
> void ceph_invalidate_dir_request(struct ceph_mds_request *req)
> {
> -	struct inode *inode = req->r_locked_dir;
> +	struct inode *inode = req->r_parent;
> 
> 	dout("invalidate_dir_request %p (complete, lease(s))\n", inode);
> 
> diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
> index 409b0e3c3b7a..ac0475a2daa7 100644
> --- a/fs/ceph/mds_client.h
> +++ b/fs/ceph/mds_client.h
> @@ -202,7 +202,7 @@ struct ceph_mds_request {
> 	char *r_path1, *r_path2;
> 	struct ceph_vino r_ino1, r_ino2;
> 
> -	struct inode *r_locked_dir; /* dir (if any) i_mutex locked by vfs */
> +	struct inode *r_parent;		    /* parent dir inode */
> 	struct inode *r_target_inode;       /* resulting inode */
> 
> #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
> @@ -211,6 +211,7 @@ struct ceph_mds_request {
> #define CEPH_MDS_R_GOT_SAFE		(4) /* got a safe reply */
> #define CEPH_MDS_R_GOT_RESULT		(5) /* got a result */
> #define CEPH_MDS_R_DID_PREPOPULATE	(6) /* prepopulated readdir */
> +#define CEPH_MDS_R_PARENT_LOCKED	(7) /* is r_parent->i_rwsem wlocked? */
> 	unsigned long	r_req_flags;
> 
> 	struct mutex r_fill_mutex;
> -- 
> 2.9.3
> 

Re: [PATCH v3 6/8] ceph: vet the target and parent inodes before updating dentry lease

[Yan, Zheng]

> On 4 Feb 2017, at 02:04, Jeff Layton <jlayton@redhat.com> wrote:
> 
> In a later patch, we're going to need to allow ceph_fill_trace to
> update the dentry's lease when the parent is not locked. This is
> potentially racy though -- by the time we get around to processing the
> trace, the parent may have already changed.
> 
> Change update_dentry_lease to take a ceph_vino pointer and use that to
> ensure that the dentry's parent still matches it before updating the
> lease.
> 
> Signed-off-by: Jeff Layton <jlayton@redhat.com>
> ---
> fs/ceph/inode.c | 72 ++++++++++++++++++++++++++++++++++++++-------------------
> 1 file changed, 48 insertions(+), 24 deletions(-)
> 
> diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> index e5c1ca02dbe3..41cbddd22091 100644
> --- a/fs/ceph/inode.c
> +++ b/fs/ceph/inode.c
> @@ -1016,7 +1016,9 @@ static int fill_inode(struct inode *inode, struct page *locked_page,
> static void update_dentry_lease(struct dentry *dentry,
> 				struct ceph_mds_reply_lease *lease,
> 				struct ceph_mds_session *session,
> -				unsigned long from_time)
> +				unsigned long from_time,
> +				struct ceph_vino *tgt_vino,
> +				struct ceph_vino *dir_vino)
> {
> 	struct ceph_dentry_info *di = ceph_dentry(dentry);
> 	long unsigned duration = le32_to_cpu(lease->duration_ms);
> @@ -1024,13 +1026,27 @@ static void update_dentry_lease(struct dentry *dentry,
> 	long unsigned half_ttl = from_time + (duration * HZ / 2) / 1000;
> 	struct inode *dir;
> 
> +	/*
> +	 * Make sure dentry's inode matches tgt_vino. NULL tgt_vino means that
> +	 * we expect a negative dentry.
> +	 */
> +	if (!tgt_vino && d_really_is_positive(dentry))
> +		return;
> +
> +	if (tgt_vino && (d_really_is_negative(dentry) ||
> +			!ceph_ino_compare(d_inode(dentry), tgt_vino)))
> +		return;
> +

we may call this function without locked parent. maybe it’s better to move these checks into critical section of dentry->d_lock


> 	spin_lock(&dentry->d_lock);
> 	dout("update_dentry_lease %p duration %lu ms ttl %lu\n",
> 	     dentry, duration, ttl);
> 
> -	/* make lease_rdcache_gen match directory */
> 	dir = d_inode(dentry->d_parent);
> 
> +	/* make sure parent matches dir_vino */
> +	if (!ceph_ino_compare(dir, dir_vino))
> +		goto out_unlock;
> +
> 	/* only track leases on regular dentries */
> 	if (ceph_snap(dir) != CEPH_NOSNAP)
> 		goto out_unlock;
> @@ -1113,7 +1129,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 	struct ceph_mds_session *session = req->r_session;
> 	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
> 	struct inode *in = NULL;
> -	struct ceph_vino vino;
> +	struct ceph_vino tvino, dvino;
> 	struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
> 	int err = 0;
> 
> @@ -1154,8 +1170,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 			dname.name = rinfo->dname;
> 			dname.len = rinfo->dname_len;
> 			dname.hash = full_name_hash(parent, dname.name, dname.len);
> -			vino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> -			vino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> +			tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> +			tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> retry_lookup:
> 			dn = d_lookup(parent, &dname);
> 			dout("d_lookup on parent=%p name=%.*s got %p\n",
> @@ -1172,8 +1188,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 				}
> 				err = 0;
> 			} else if (d_really_is_positive(dn) &&
> -				   (ceph_ino(d_inode(dn)) != vino.ino ||
> -				    ceph_snap(d_inode(dn)) != vino.snap)) {
> +				   (ceph_ino(d_inode(dn)) != tvino.ino ||
> +				    ceph_snap(d_inode(dn)) != tvino.snap)) {
> 				dout(" dn %p points to wrong inode %p\n",
> 				     dn, d_inode(dn));
> 				d_delete(dn);
> @@ -1187,10 +1203,10 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 	}
> 
> 	if (rinfo->head->is_target) {
> -		vino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> -		vino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> +		tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> +		tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> 
> -		in = ceph_get_inode(sb, vino);
> +		in = ceph_get_inode(sb, tvino);
> 		if (IS_ERR(in)) {
> 			err = PTR_ERR(in);
> 			goto done;
> @@ -1234,10 +1250,12 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 		BUG_ON(!dn);
> 		BUG_ON(!dir);
> 		BUG_ON(d_inode(dn->d_parent) != dir);
> -		BUG_ON(ceph_ino(dir) !=
> -		       le64_to_cpu(rinfo->diri.in->ino));
> -		BUG_ON(ceph_snap(dir) !=
> -		       le64_to_cpu(rinfo->diri.in->snapid));
> +
> +		dvino.ino = le64_to_cpu(rinfo->diri.in->ino);
> +		dvino.snap = le64_to_cpu(rinfo->diri.in->snapid);
> +
> +		BUG_ON(ceph_ino(dir) != dvino.ino);
> +		BUG_ON(ceph_snap(dir) != dvino.snap);
> 
> 		/* do we have a lease on the whole dir? */
> 		have_dir_cap =
> @@ -1294,7 +1312,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 					d_add(dn, NULL);
> 				update_dentry_lease(dn, rinfo->dlease,
> 						    session,
> -						    req->r_request_started);
> +						    req->r_request_started,
> +						    NULL, &dvino);
> 			}
> 			goto done;
> 		}
> @@ -1317,9 +1336,13 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> 			have_lease = false;
> 		}
> 
> -		if (have_lease)
> +		if (have_lease) {
> +			tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> +			tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> 			update_dentry_lease(dn, rinfo->dlease, session,
> -					    req->r_request_started);
> +					    req->r_request_started,
> +					    &tvino, &dvino);
> +		}
> 		dout(" final dn %p\n", dn);
> 	} else if ((req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> 		    req->r_op == CEPH_MDS_OP_MKSNAP) &&
> @@ -1493,14 +1516,14 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> 	/* FIXME: release caps/leases if error occurs */
> 	for (i = 0; i < rinfo->dir_nr; i++) {
> 		struct ceph_mds_reply_dir_entry *rde = rinfo->dir_entries + i;
> -		struct ceph_vino vino;
> +		struct ceph_vino tvino, dvino;
> 
> 		dname.name = rde->name;
> 		dname.len = rde->name_len;
> 		dname.hash = full_name_hash(parent, dname.name, dname.len);
> 
> -		vino.ino = le64_to_cpu(rde->inode.in->ino);
> -		vino.snap = le64_to_cpu(rde->inode.in->snapid);
> +		tvino.ino = le64_to_cpu(rde->inode.in->ino);
> +		tvino.snap = le64_to_cpu(rde->inode.in->snapid);
> 
> 		if (rinfo->hash_order) {
> 			u32 hash = ceph_str_hash(ci->i_dir_layout.dl_dir_hash,
> @@ -1529,8 +1552,8 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> 				goto out;
> 			}
> 		} else if (d_really_is_positive(dn) &&
> -			   (ceph_ino(d_inode(dn)) != vino.ino ||
> -			    ceph_snap(d_inode(dn)) != vino.snap)) {
> +			   (ceph_ino(d_inode(dn)) != tvino.ino ||
> +			    ceph_snap(d_inode(dn)) != tvino.snap)) {
> 			dout(" dn %p points to wrong inode %p\n",
> 			     dn, d_inode(dn));
> 			d_delete(dn);
> @@ -1542,7 +1565,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> 		if (d_really_is_positive(dn)) {
> 			in = d_inode(dn);
> 		} else {
> -			in = ceph_get_inode(parent->d_sb, vino);
> +			in = ceph_get_inode(parent->d_sb, tvino);
> 			if (IS_ERR(in)) {
> 				dout("new_inode badness\n");
> 				d_drop(dn);
> @@ -1587,8 +1610,9 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> 
> 		ceph_dentry(dn)->offset = rde->offset;
> 
> +		dvino = ceph_vino(d_inode(parent));
> 		update_dentry_lease(dn, rde->lease, req->r_session,
> -				    req->r_request_started);
> +				    req->r_request_started, &tvino, &dvino);
> 
> 		if (err == 0 && skipped == 0 && cache_ctl.index >= 0) {
> 			ret = fill_readdir_cache(d_inode(parent), dn,
> -- 
> 2.9.3
> 

Re: [PATCH v3 0/8] ceph: fix performance regression due to ceph_d_revalidate fixes

[Yan, Zheng]

> On 4 Feb 2017, at 02:03, Jeff Layton <jlayton@redhat.com> wrote:
> 
> v3: Squash bool to flag change patches, and better explain why we need
> to use atomic *_bit calls. Vet target and parent inodes when updating
> lease. Update lease properly on negative dentries (when is_target is 0).
> Also, update dentry leases even when call that triggered the request
> is cancelled.
> 
> v2: Fix a hang that would occasionally occur under fsstress
> 
> Zheng mentioned that he had gotten a report that testing with vdbench on
> a v4.8 kernel showed a large increase in the number of GETATTR calls
> being made. Since we had changed the d_revalidate code to use a GETATTR
> call and not set r_locked_dir in the request, the dentry lease wasn't
> getting updated from the traces, even when the MDS had granted one.
> 
> We need to use a LOOKUP when revalidating dentries so we can update
> the dentry lease if things look ok. That currently depends however on
> having r_locked_dir set to the parent directory.
> 
> In d_revalidate though, we may very well not hold the parent's i_rwsem.
> Manipulating the dcache is not allowed without it, but we can update the
> inode and the dentry leases.
> 
> The basic approach here is to separate r_locked_dir into an inode
> pointer representing the parent, and a flag indicating whether we
> know that it's locked (note that it _may_ still be locked even if
> that flag isn't set).
> 
> In ceph_fill_trace, we can then do all of the parts that don't require a
> locked parent whenever the parent pointer is set, but avoid the
> dcache manipulation when it isn't.
> 
> With this, dentry leases are again being updated as a result of
> d_revalidate lookups. The switch to using a flags fields instead
> of bools also shrinks the request size slightly. 
> 
> Jeff Layton (8):
>  ceph: remove "Debugging hook" from ceph_fill_trace
>  ceph: drop session argument to ceph_fill_trace
>  ceph: convert bools in ceph_mds_request to a new r_req_flags field
>  ceph: add a new flag to indicate whether parent is locked
>  ceph: don't update_dentry_lease unless we actually got one
>  ceph: vet the target and parent inodes before updating dentry lease
>  ceph: call update_dentry_lease even when r_locked dir is not set
>  ceph: do a LOOKUP in d_revalidate instead of GETATTR
> 
> fs/ceph/debugfs.c    |   2 +-
> fs/ceph/dir.c        |  30 ++++++----
> fs/ceph/export.c     |   3 +-
> fs/ceph/file.c       |   3 +-
> fs/ceph/inode.c      | 164 ++++++++++++++++++++++++++++-----------------------
> fs/ceph/mds_client.c |  73 ++++++++++++-----------
> fs/ceph/mds_client.h |  15 +++--
> fs/ceph/super.h      |   3 +-
> 8 files changed, 162 insertions(+), 131 deletions(-)

Besides few minor comments. This series looks good in general.

Regards
Yan, Zheng
> 
> -- 
> 2.9.3
> 

Re: [PATCH v3 6/8] ceph: vet the target and parent inodes before updating dentry lease

[Jeff Layton]

On Sat, 2017-02-04 at 11:20 +0800, Yan, Zheng wrote:
> > On 4 Feb 2017, at 02:04, Jeff Layton <jlayton@redhat.com> wrote:
> > 
> > In a later patch, we're going to need to allow ceph_fill_trace to
> > update the dentry's lease when the parent is not locked. This is
> > potentially racy though -- by the time we get around to processing the
> > trace, the parent may have already changed.
> > 
> > Change update_dentry_lease to take a ceph_vino pointer and use that to
> > ensure that the dentry's parent still matches it before updating the
> > lease.
> > 
> > Signed-off-by: Jeff Layton <jlayton@redhat.com>
> > ---
> > fs/ceph/inode.c | 72 ++++++++++++++++++++++++++++++++++++++-------------------
> > 1 file changed, 48 insertions(+), 24 deletions(-)
> > 
> > diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> > index e5c1ca02dbe3..41cbddd22091 100644
> > --- a/fs/ceph/inode.c
> > +++ b/fs/ceph/inode.c
> > @@ -1016,7 +1016,9 @@ static int fill_inode(struct inode *inode, struct page *locked_page,
> > static void update_dentry_lease(struct dentry *dentry,
> > 				struct ceph_mds_reply_lease *lease,
> > 				struct ceph_mds_session *session,
> > -				unsigned long from_time)
> > +				unsigned long from_time,
> > +				struct ceph_vino *tgt_vino,
> > +				struct ceph_vino *dir_vino)
> > {
> > 	struct ceph_dentry_info *di = ceph_dentry(dentry);
> > 	long unsigned duration = le32_to_cpu(lease->duration_ms);
> > @@ -1024,13 +1026,27 @@ static void update_dentry_lease(struct dentry *dentry,
> > 	long unsigned half_ttl = from_time + (duration * HZ / 2) / 1000;
> > 	struct inode *dir;
> > 
> > +	/*
> > +	 * Make sure dentry's inode matches tgt_vino. NULL tgt_vino means that
> > +	 * we expect a negative dentry.
> > +	 */
> > +	if (!tgt_vino && d_really_is_positive(dentry))
> > +		return;
> > +
> > +	if (tgt_vino && (d_really_is_negative(dentry) ||
> > +			!ceph_ino_compare(d_inode(dentry), tgt_vino)))
> > +		return;
> > +
> 
> we may call this function without locked parent. maybe it’s better to move these checks into critical section of dentry->d_lock
> 
> 

d_inode(dentry) is stable even without the parent being locked. We only
need the d_lock to ensure stability of the parent here. Certainly we
could take the lock before checking it, but it's not required here.

> > 	spin_lock(&dentry->d_lock);
> > 	dout("update_dentry_lease %p duration %lu ms ttl %lu\n",
> > 	     dentry, duration, ttl);
> > 
> > -	/* make lease_rdcache_gen match directory */
> > 	dir = d_inode(dentry->d_parent);
> > 
> > +	/* make sure parent matches dir_vino */
> > +	if (!ceph_ino_compare(dir, dir_vino))
> > +		goto out_unlock;
> > +
> > 	/* only track leases on regular dentries */
> > 	if (ceph_snap(dir) != CEPH_NOSNAP)
> > 		goto out_unlock;
> > @@ -1113,7 +1129,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 	struct ceph_mds_session *session = req->r_session;
> > 	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
> > 	struct inode *in = NULL;
> > -	struct ceph_vino vino;
> > +	struct ceph_vino tvino, dvino;
> > 	struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
> > 	int err = 0;
> > 
> > @@ -1154,8 +1170,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 			dname.name = rinfo->dname;
> > 			dname.len = rinfo->dname_len;
> > 			dname.hash = full_name_hash(parent, dname.name, dname.len);
> > -			vino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> > -			vino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> > +			tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> > +			tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> > retry_lookup:
> > 			dn = d_lookup(parent, &dname);
> > 			dout("d_lookup on parent=%p name=%.*s got %p\n",
> > @@ -1172,8 +1188,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 				}
> > 				err = 0;
> > 			} else if (d_really_is_positive(dn) &&
> > -				   (ceph_ino(d_inode(dn)) != vino.ino ||
> > -				    ceph_snap(d_inode(dn)) != vino.snap)) {
> > +				   (ceph_ino(d_inode(dn)) != tvino.ino ||
> > +				    ceph_snap(d_inode(dn)) != tvino.snap)) {
> > 				dout(" dn %p points to wrong inode %p\n",
> > 				     dn, d_inode(dn));
> > 				d_delete(dn);
> > @@ -1187,10 +1203,10 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 	}
> > 
> > 	if (rinfo->head->is_target) {
> > -		vino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> > -		vino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> > +		tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> > +		tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> > 
> > -		in = ceph_get_inode(sb, vino);
> > +		in = ceph_get_inode(sb, tvino);
> > 		if (IS_ERR(in)) {
> > 			err = PTR_ERR(in);
> > 			goto done;
> > @@ -1234,10 +1250,12 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 		BUG_ON(!dn);
> > 		BUG_ON(!dir);
> > 		BUG_ON(d_inode(dn->d_parent) != dir);
> > -		BUG_ON(ceph_ino(dir) !=
> > -		       le64_to_cpu(rinfo->diri.in->ino));
> > -		BUG_ON(ceph_snap(dir) !=
> > -		       le64_to_cpu(rinfo->diri.in->snapid));
> > +
> > +		dvino.ino = le64_to_cpu(rinfo->diri.in->ino);
> > +		dvino.snap = le64_to_cpu(rinfo->diri.in->snapid);
> > +
> > +		BUG_ON(ceph_ino(dir) != dvino.ino);
> > +		BUG_ON(ceph_snap(dir) != dvino.snap);
> > 
> > 		/* do we have a lease on the whole dir? */
> > 		have_dir_cap =
> > @@ -1294,7 +1312,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 					d_add(dn, NULL);
> > 				update_dentry_lease(dn, rinfo->dlease,
> > 						    session,
> > -						    req->r_request_started);
> > +						    req->r_request_started,
> > +						    NULL, &dvino);
> > 			}
> > 			goto done;
> > 		}
> > @@ -1317,9 +1336,13 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 			have_lease = false;
> > 		}
> > 
> > -		if (have_lease)
> > +		if (have_lease) {
> > +			tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
> > +			tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
> > 			update_dentry_lease(dn, rinfo->dlease, session,
> > -					    req->r_request_started);
> > +					    req->r_request_started,
> > +					    &tvino, &dvino);
> > +		}
> > 		dout(" final dn %p\n", dn);
> > 	} else if ((req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> > 		    req->r_op == CEPH_MDS_OP_MKSNAP) &&
> > @@ -1493,14 +1516,14 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> > 	/* FIXME: release caps/leases if error occurs */
> > 	for (i = 0; i < rinfo->dir_nr; i++) {
> > 		struct ceph_mds_reply_dir_entry *rde = rinfo->dir_entries + i;
> > -		struct ceph_vino vino;
> > +		struct ceph_vino tvino, dvino;
> > 
> > 		dname.name = rde->name;
> > 		dname.len = rde->name_len;
> > 		dname.hash = full_name_hash(parent, dname.name, dname.len);
> > 
> > -		vino.ino = le64_to_cpu(rde->inode.in->ino);
> > -		vino.snap = le64_to_cpu(rde->inode.in->snapid);
> > +		tvino.ino = le64_to_cpu(rde->inode.in->ino);
> > +		tvino.snap = le64_to_cpu(rde->inode.in->snapid);
> > 
> > 		if (rinfo->hash_order) {
> > 			u32 hash = ceph_str_hash(ci->i_dir_layout.dl_dir_hash,
> > @@ -1529,8 +1552,8 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> > 				goto out;
> > 			}
> > 		} else if (d_really_is_positive(dn) &&
> > -			   (ceph_ino(d_inode(dn)) != vino.ino ||
> > -			    ceph_snap(d_inode(dn)) != vino.snap)) {
> > +			   (ceph_ino(d_inode(dn)) != tvino.ino ||
> > +			    ceph_snap(d_inode(dn)) != tvino.snap)) {
> > 			dout(" dn %p points to wrong inode %p\n",
> > 			     dn, d_inode(dn));
> > 			d_delete(dn);
> > @@ -1542,7 +1565,7 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> > 		if (d_really_is_positive(dn)) {
> > 			in = d_inode(dn);
> > 		} else {
> > -			in = ceph_get_inode(parent->d_sb, vino);
> > +			in = ceph_get_inode(parent->d_sb, tvino);
> > 			if (IS_ERR(in)) {
> > 				dout("new_inode badness\n");
> > 				d_drop(dn);
> > @@ -1587,8 +1610,9 @@ int ceph_readdir_prepopulate(struct ceph_mds_request *req,
> > 
> > 		ceph_dentry(dn)->offset = rde->offset;
> > 
> > +		dvino = ceph_vino(d_inode(parent));
> > 		update_dentry_lease(dn, rde->lease, req->r_session,
> > -				    req->r_request_started);
> > +				    req->r_request_started, &tvino, &dvino);
> > 
> > 		if (err == 0 && skipped == 0 && cache_ctl.index >= 0) {
> > 			ret = fill_readdir_cache(d_inode(parent), dn,
> > -- 
> > 2.9.3
> > 
> 
> 

-- 
Jeff Layton <jlayton@redhat.com>

Re: [PATCH v3 4/8] ceph: add a new flag to indicate whether parent is locked

[Jeff Layton]

On Sat, 2017-02-04 at 11:16 +0800, Yan, Zheng wrote:
> > On 4 Feb 2017, at 02:04, Jeff Layton <jlayton@redhat.com> wrote:
> > 
> > struct ceph_mds_request has an r_locked_dir pointer, which is set to
> > indicate the parent inode and that its i_rwsem is locked.  In some
> > critical places, we need to be able to indicate the parent inode to the
> > request handling code, even when its i_rwsem may not be locked.
> > 
> > Most of the code that operates on r_locked_dir doesn't require that the
> > i_rwsem be locked. We only really need it to handle manipulation of the
> > dcache. The rest (filling of the inode, updating dentry leases, etc.)
> > already has its own locking.
> > 
> > Add a new r_req_flags bit that indicates whether the parent is locked
> > when doing the request, and rename the pointer to "r_parent". For now,
> > all the places that set r_parent also set this flag, but that will
> > change in a later patch.
> > 
> > Signed-off-by: Jeff Layton <jlayton@redhat.com>
> > ---
> > fs/ceph/dir.c        | 21 ++++++++++++++-------
> > fs/ceph/export.c     |  3 ++-
> > fs/ceph/file.c       |  3 ++-
> > fs/ceph/inode.c      | 17 ++++++++++-------
> > fs/ceph/mds_client.c | 24 ++++++++++++------------
> > fs/ceph/mds_client.h |  3 ++-
> > 6 files changed, 42 insertions(+), 29 deletions(-)
> > 
> > diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
> > index 5fb86d71220e..c8562b8a8c7e 100644
> > --- a/fs/ceph/dir.c
> > +++ b/fs/ceph/dir.c
> > @@ -752,7 +752,8 @@ static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
> > 		mask |= CEPH_CAP_XATTR_SHARED;
> > 	req->r_args.getattr.mask = cpu_to_le32(mask);
> > 
> > -	req->r_locked_dir = dir;
> > +	req->r_parent = dir;
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	err = ceph_mdsc_do_request(mdsc, NULL, req);
> > 	err = ceph_handle_snapdir(req, dentry, err);
> > 	dentry = ceph_finish_lookup(req, dentry, err);
> > @@ -813,7 +814,8 @@ static int ceph_mknod(struct inode *dir, struct dentry *dentry,
> > 	}
> > 	req->r_dentry = dget(dentry);
> > 	req->r_num_caps = 2;
> > -	req->r_locked_dir = dir;
> > +	req->r_parent = dir;
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	req->r_args.mknod.mode = cpu_to_le32(mode);
> > 	req->r_args.mknod.rdev = cpu_to_le32(rdev);
> > 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> > @@ -864,7 +866,8 @@ static int ceph_symlink(struct inode *dir, struct dentry *dentry,
> > 		ceph_mdsc_put_request(req);
> > 		goto out;
> > 	}
> > -	req->r_locked_dir = dir;
> > +	req->r_parent = dir;
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	req->r_dentry = dget(dentry);
> > 	req->r_num_caps = 2;
> > 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> > @@ -913,7 +916,8 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, umode_t mode)
> > 
> > 	req->r_dentry = dget(dentry);
> > 	req->r_num_caps = 2;
> > -	req->r_locked_dir = dir;
> > +	req->r_parent = dir;
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	req->r_args.mkdir.mode = cpu_to_le32(mode);
> > 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> > 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
> > @@ -957,7 +961,8 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
> > 	req->r_dentry = dget(dentry);
> > 	req->r_num_caps = 2;
> > 	req->r_old_dentry = dget(old_dentry);
> > -	req->r_locked_dir = dir;
> > +	req->r_parent = dir;
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> > 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
> > 	/* release LINK_SHARED on source inode (mds will lock it) */
> > @@ -1023,7 +1028,8 @@ static int ceph_unlink(struct inode *dir, struct dentry *dentry)
> > 	}
> > 	req->r_dentry = dget(dentry);
> > 	req->r_num_caps = 2;
> > -	req->r_locked_dir = dir;
> > +	req->r_parent = dir;
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> > 	req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
> > 	req->r_inode_drop = drop_caps_for_unlink(inode);
> > @@ -1066,7 +1072,8 @@ static int ceph_rename(struct inode *old_dir, struct dentry *old_dentry,
> > 	req->r_num_caps = 2;
> > 	req->r_old_dentry = dget(old_dentry);
> > 	req->r_old_dentry_dir = old_dir;
> > -	req->r_locked_dir = new_dir;
> > +	req->r_parent = new_dir;
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	req->r_old_dentry_drop = CEPH_CAP_FILE_SHARED;
> > 	req->r_old_dentry_unless = CEPH_CAP_FILE_EXCL;
> > 	req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
> > diff --git a/fs/ceph/export.c b/fs/ceph/export.c
> > index 180bbef760f2..e8f11fa565c5 100644
> > --- a/fs/ceph/export.c
> > +++ b/fs/ceph/export.c
> > @@ -207,7 +207,8 @@ static int ceph_get_name(struct dentry *parent, char *name,
> > 	req->r_inode = d_inode(child);
> > 	ihold(d_inode(child));
> > 	req->r_ino2 = ceph_vino(d_inode(parent));
> > -	req->r_locked_dir = d_inode(parent);
> > +	req->r_parent = d_inode(parent);
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	req->r_num_caps = 2;
> > 	err = ceph_mdsc_do_request(mdsc, NULL, req);
> > 
> > diff --git a/fs/ceph/file.c b/fs/ceph/file.c
> > index 19419eb87f38..a91239c4b31c 100644
> > --- a/fs/ceph/file.c
> > +++ b/fs/ceph/file.c
> > @@ -310,7 +310,8 @@ int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
> >                mask |= CEPH_CAP_XATTR_SHARED;
> >        req->r_args.open.mask = cpu_to_le32(mask);
> > 
> > -	req->r_locked_dir = dir;           /* caller holds dir->i_mutex */
> > +	req->r_parent = dir;
> > +	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
> > 	err = ceph_mdsc_do_request(mdsc,
> > 				   (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
> > 				   req);
> > diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> > index e54ba03d3df7..b4b61da03dbf 100644
> > --- a/fs/ceph/inode.c
> > +++ b/fs/ceph/inode.c
> > @@ -1122,13 +1122,13 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 
> > 	if (!rinfo->head->is_target && !rinfo->head->is_dentry) {
> > 		dout("fill_trace reply is empty!\n");
> > -		if (rinfo->head->result == 0 && req->r_locked_dir)
> > +		if (rinfo->head->result == 0 && req->r_parent)
> > 			ceph_invalidate_dir_request(req);
> > 		return 0;
> > 	}
> > 
> > 	if (rinfo->head->is_dentry) {
> > -		struct inode *dir = req->r_locked_dir;
> > +		struct inode *dir = req->r_parent;
> > 
> > 		if (dir) {
> > 			err = fill_inode(dir, NULL,
> > @@ -1216,7 +1216,9 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 	 * ignore null lease/binding on snapdir ENOENT, or else we
> > 	 * will have trouble splicing in the virtual snapdir later
> > 	 */
> > -	if (rinfo->head->is_dentry && req->r_locked_dir &&
> > +	if (rinfo->head->is_dentry &&
> > +            !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
> 
> why adding the !ABORTED check back. I think previous patch add:
> 
> if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
>   goto done;
> 

Oof, good catch -- I screwed that up.

We want to keep the R_ABORTED checks where r_aborted is checked now, as
that will allow us to update the dentry leases even when the task that
spawned the call was aborted.

I'll fix that in the ceph-client/testing branch, but I'll not bother
reposting unless there are other problems that need fixing in here.

> > +	    test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags) &&
> > 	    (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
> > 					       fsc->mount_options->snapdir_name,
> > 					       req->r_dentry->d_name.len))) {
> > @@ -1225,7 +1227,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 		 * mknod symlink mkdir  : null -> new inode
> > 		 * unlink               : linked -> null
> > 		 */
> > -		struct inode *dir = req->r_locked_dir;
> > +		struct inode *dir = req->r_parent;
> > 		struct dentry *dn = req->r_dentry;
> > 		bool have_dir_cap, have_lease;
> > 
> > @@ -1319,10 +1321,11 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
> > 			update_dentry_lease(dn, rinfo->dlease, session,
> > 					    req->r_request_started);
> > 		dout(" final dn %p\n", dn);
> > -	} else if (req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> > -		   req->r_op == CEPH_MDS_OP_MKSNAP) {
> > +	} else if ((req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
> > +		    req->r_op == CEPH_MDS_OP_MKSNAP) &&
> > +		   !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
> 
> same here
> 
> > 		struct dentry *dn = req->r_dentry;
> > -		struct inode *dir = req->r_locked_dir;
> > +		struct inode *dir = req->r_parent;
> > 
> > 		/* fill out a snapdir LOOKUPSNAP dentry */
> > 		BUG_ON(!dn);
> > diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
> > index ccf75a3260e8..52521f339745 100644
> > --- a/fs/ceph/mds_client.c
> > +++ b/fs/ceph/mds_client.c
> > @@ -547,8 +547,8 @@ void ceph_mdsc_release_request(struct kref *kref)
> > 		ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
> > 		iput(req->r_inode);
> > 	}
> > -	if (req->r_locked_dir)
> > -		ceph_put_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
> > +	if (req->r_parent)
> > +		ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
> > 	iput(req->r_target_inode);
> > 	if (req->r_dentry)
> > 		dput(req->r_dentry);
> > @@ -735,7 +735,7 @@ static int __choose_mds(struct ceph_mds_client *mdsc,
> > 
> > 		rcu_read_lock();
> > 		parent = req->r_dentry->d_parent;
> > -		dir = req->r_locked_dir ? : d_inode_rcu(parent);
> > +		dir = req->r_parent ? : d_inode_rcu(parent);
> > 
> > 		if (!dir || dir->i_sb != mdsc->fsc->sb) {
> > 			/*  not this fs or parent went negative */
> > @@ -1894,7 +1894,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
> > 	int ret;
> > 
> > 	ret = set_request_path_attr(req->r_inode, req->r_dentry,
> > -			      req->r_locked_dir, req->r_path1, req->r_ino1.ino,
> > +			      req->r_parent, req->r_path1, req->r_ino1.ino,
> > 			      &path1, &pathlen1, &ino1, &freepath1);
> > 	if (ret < 0) {
> > 		msg = ERR_PTR(ret);
> > @@ -1956,7 +1956,7 @@ static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
> > 		      mds, req->r_inode_drop, req->r_inode_unless, 0);
> > 	if (req->r_dentry_drop)
> > 		releases += ceph_encode_dentry_release(&p, req->r_dentry,
> > -				req->r_locked_dir, mds, req->r_dentry_drop,
> > +				req->r_parent, mds, req->r_dentry_drop,
> > 				req->r_dentry_unless);
> > 	if (req->r_old_dentry_drop)
> > 		releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
> > @@ -2095,14 +2095,14 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc,
> > 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
> > 	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
> > 		flags |= CEPH_MDS_FLAG_REPLAY;
> > -	if (req->r_locked_dir)
> > +	if (req->r_parent)
> > 		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
> > 	rhead->flags = cpu_to_le32(flags);
> > 	rhead->num_fwd = req->r_num_fwd;
> > 	rhead->num_retry = req->r_attempts - 1;
> > 	rhead->ino = 0;
> > 
> > -	dout(" r_locked_dir = %p\n", req->r_locked_dir);
> > +	dout(" r_parent = %p\n", req->r_parent);
> > 	return 0;
> > }
> > 
> > @@ -2282,11 +2282,11 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
> > 
> > 	dout("do_request on %p\n", req);
> > 
> > -	/* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
> > +	/* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
> > 	if (req->r_inode)
> > 		ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
> > -	if (req->r_locked_dir)
> > -		ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
> > +	if (req->r_parent)
> > +		ceph_get_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
> > 	if (req->r_old_dentry_dir)
> > 		ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
> > 				  CEPH_CAP_PIN);
> > @@ -2336,7 +2336,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
> > 		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
> > 		mutex_unlock(&req->r_fill_mutex);
> > 
> > -		if (req->r_locked_dir &&
> > +		if (req->r_parent &&
> > 		    (req->r_op & CEPH_MDS_OP_WRITE))
> > 			ceph_invalidate_dir_request(req);
> > 	} else {
> > @@ -2355,7 +2355,7 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
> >  */
> > void ceph_invalidate_dir_request(struct ceph_mds_request *req)
> > {
> > -	struct inode *inode = req->r_locked_dir;
> > +	struct inode *inode = req->r_parent;
> > 
> > 	dout("invalidate_dir_request %p (complete, lease(s))\n", inode);
> > 
> > diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
> > index 409b0e3c3b7a..ac0475a2daa7 100644
> > --- a/fs/ceph/mds_client.h
> > +++ b/fs/ceph/mds_client.h
> > @@ -202,7 +202,7 @@ struct ceph_mds_request {
> > 	char *r_path1, *r_path2;
> > 	struct ceph_vino r_ino1, r_ino2;
> > 
> > -	struct inode *r_locked_dir; /* dir (if any) i_mutex locked by vfs */
> > +	struct inode *r_parent;		    /* parent dir inode */
> > 	struct inode *r_target_inode;       /* resulting inode */
> > 
> > #define CEPH_MDS_R_DIRECT_IS_HASH	(1) /* r_direct_hash is valid */
> > @@ -211,6 +211,7 @@ struct ceph_mds_request {
> > #define CEPH_MDS_R_GOT_SAFE		(4) /* got a safe reply */
> > #define CEPH_MDS_R_GOT_RESULT		(5) /* got a result */
> > #define CEPH_MDS_R_DID_PREPOPULATE	(6) /* prepopulated readdir */
> > +#define CEPH_MDS_R_PARENT_LOCKED	(7) /* is r_parent->i_rwsem wlocked? */
> > 	unsigned long	r_req_flags;
> > 
> > 	struct mutex r_fill_mutex;
> > -- 
> > 2.9.3
> > 
> 
> 

-- 
Jeff Layton <jlayton@redhat.com>