[PATCH] fd/ceph/mds_client: add per-request lock for completion

[Max Kellermann <max.kellermann@ionos.com>]

ceph_mdsc_wait_request() took `ceph_mds_client.mutex` only to resolve
a request's terminal state after the wait returned: either a reply had
arrived and handle_reply() had set `r_reply`/`CEPH_MDS_R_GOT_RESULT`,
or there was an error and the waiter had to abort the request by
setting `r_err`/`CEPH_MDS_R_ABORTED` so a late reply is discarded.  It
never touched any `ceph_mds_client` fields; it used the big mutex
purely as the serialization point against the matching section at the
tail of handle_reply(), and (indirectly) against __do_request() and
handle_forward().

`ceph_mds_client.mutex` is a coarse lock, also covering the request
tree, tid counters, the session table and the mdsmap.  Making the
waiter take it means request completion contends with request
submission (ceph_mdsc_submit_request() holds it across
__register_request() and __do_request()) even though the two only
share a handful of per-request fields.

This patch adds a per-request spinlock that guards a request's
terminal state: `r_err`, `r_reply`, `CEPH_MDS_R_GOT_RESULT` and
`CEPH_MDS_R_ABORTED`.

Lock ordering is `ceph_mds_client.mutex`, `r_fill_mutex`,
`r_completion_lock`; the spinlock is always innermost and no sleeping
lock is taken while it is held.

The two BUG_ON()s in handle_forward()'s resend branch are removed.
They asserted that a request being resent had neither errored nor
completed, an invariant that held only because the client mutex
serialized the resend against the waiter's abort.  Now that the waiter
aborts under `r_completion_lock` without the client mutex, a forward
can race an abort; this is harmless because __do_request() re-checks
the terminal state under `r_completion_lock` and unregisters the
request instead of resending it.

This weakens one invariant: previously, the abort path held
`ceph_mds_client.mutex`, so it was mutually exclusive with
__do_request() and a request could never be sent once
`CEPH_MDS_R_ABORTED` was set.  With the abort now protected only by
`r_completion_lock`, the bit can be set concurrently with a
resend-path __do_request() (kick_requests(), __wake_requests(),
handle_forward()) after that function's one-time entry check, so an
aborted request may still be sent or queued.  This is harmless:
`ABORTED` is re-checked at every consumer: handle_reply() discards the
reply, ceph_fill_trace() skips the dcache work that depends on locks
held by the (now returned) caller, and the next __do_request()
unregisters it.  This matches the existing property that an
interrupted request may still execute on the MDS.

Signed-off-by: Max Kellermann <max.kellermann@ionos.com>
---
 fs/ceph/locks.c      | 21 ++++++----
 fs/ceph/mds_client.c | 99 ++++++++++++++++++++++++++++++++++----------
 fs/ceph/mds_client.h | 10 +++++
 3 files changed, 102 insertions(+), 28 deletions(-)

diff --git a/fs/ceph/locks.c b/fs/ceph/locks.c
index 677221bd64e0..35cadcd7cb2a 100644
--- a/fs/ceph/locks.c
+++ b/fs/ceph/locks.c
@@ -177,25 +177,32 @@ static int ceph_lock_wait_for_completion(struct ceph_mds_client *mdsc,
 
 	doutc(cl, "request %llu was interrupted\n", req->r_tid);
 
+	/*
+	 * mdsc->mutex guards the r_session check below.  The terminal-state
+	 * arbitration against a concurrent reply/forward handler is done under
+	 * r_completion_lock (the same lock ceph_mdsc_wait_request() uses),
+	 * since those handlers no longer serialize on mdsc->mutex.
+	 * r_fill_mutex additionally fences ceph_fill_trace() /
+	 * ceph_readdir_prepopulate(), which consult CEPH_MDS_R_ABORTED while
+	 * relying on locks (dir mutex) held by our caller.
+	 * Lock ordering: mdsc->mutex -> r_fill_mutex -> r_completion_lock.
+	 */
 	mutex_lock(&mdsc->mutex);
+	mutex_lock(&req->r_fill_mutex);
+	spin_lock(&req->r_completion_lock);
 	if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
 		err = 0;
 	} else {
-		/*
-		 * ensure we aren't running concurrently with
-		 * ceph_fill_trace or ceph_readdir_prepopulate, which
-		 * rely on locks (dir mutex) held by our caller.
-		 */
-		mutex_lock(&req->r_fill_mutex);
 		req->r_err = err;
 		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
-		mutex_unlock(&req->r_fill_mutex);
 
 		if (!req->r_session) {
 			// haven't sent the request
 			err = 0;
 		}
 	}
+	spin_unlock(&req->r_completion_lock);
+	mutex_unlock(&req->r_fill_mutex);
 	mutex_unlock(&mdsc->mutex);
 	if (!err)
 		return 0;
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 853bf698b356..824d271e55c9 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -2722,6 +2722,7 @@ ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
 		return ERR_PTR(-ENOMEM);
 
 	mutex_init(&req->r_fill_mutex);
+	spin_lock_init(&req->r_completion_lock);
 	req->r_mdsc = mdsc;
 	req->r_started = jiffies;
 	req->r_start_latency = ktime_get();
@@ -3528,7 +3529,9 @@ static int __prepare_send_request(struct ceph_mds_session *session,
 	}
 	msg = create_request_message(session, req, drop_cap_releases);
 	if (IS_ERR(msg)) {
+		spin_lock(&req->r_completion_lock);
 		req->r_err = PTR_ERR(msg);
+		spin_unlock(&req->r_completion_lock);
 		return PTR_ERR(msg);
 	}
 	req->r_request = msg;
@@ -3586,9 +3589,34 @@ static void __do_request(struct ceph_mds_client *mdsc,
 	int mds = -1;
 	int err = 0;
 	bool random;
+	bool done, 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))
+	/*
+	 * The request may already have reached its terminal state,
+	 * either because a reply/forward arrived or because
+	 * ceph_mdsc_wait_request() aborted it.  Sample that state
+	 * under r_completion_lock so we agree with those paths, which
+	 * no longer all serialize on mdsc->mutex.
+	 *
+	 * This is only a point-in-time check.  Since the abort path
+	 * does not own mdsc->mutex, CEPH_MDS_R_ABORTED may be set
+	 * concurrently after this snapshot while we go on to choose a
+	 * MDS and send the request, so a request can be sent (or
+	 * queued) after it has been aborted.  That is harmless:
+	 * ABORTED is re-checked at every consumer: handle_reply()
+	 * discards the reply, ceph_fill_trace() skips the work that
+	 * relies on locks held by the (now returned) caller, and the
+	 * next __do_request() sees the flag and unregisters the
+	 * request.  ABORTED is an advisory "stop waiting" flag, not a
+	 * send barrier.
+	 */
+	spin_lock(&req->r_completion_lock);
+	done = req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
+	aborted = test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
+	spin_unlock(&req->r_completion_lock);
+
+	if (done) {
+		if (aborted)
 			__unregister_request(mdsc, req);
 		return;
 	}
@@ -3742,7 +3770,9 @@ static void __do_request(struct ceph_mds_client *mdsc,
 					  TASK_KILLABLE);
 			if (err) {
 				mutex_lock(&req->r_fill_mutex);
+				spin_lock(&req->r_completion_lock);
 				set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
+				spin_unlock(&req->r_completion_lock);
 				mutex_unlock(&req->r_fill_mutex);
 				goto out_session;
 			}
@@ -3783,7 +3813,9 @@ static void __do_request(struct ceph_mds_client *mdsc,
 finish:
 	if (err) {
 		doutc(cl, "early error %d\n", err);
+		spin_lock(&req->r_completion_lock);
 		req->r_err = err;
+		spin_unlock(&req->r_completion_lock);
 		complete_request(mdsc, req);
 		__unregister_request(mdsc, req);
 	}
@@ -3910,6 +3942,8 @@ int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc,
 			   ceph_mds_request_wait_callback_t wait_func)
 {
 	struct ceph_client *cl = mdsc->fsc->client;
+	bool aborted = false;
+	bool got_result;
 	int err;
 
 	/* wait */
@@ -3928,32 +3962,47 @@ int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc,
 			err = timeleft;  /* killed */
 	}
 	doutc(cl, "do_request waited, got %d\n", err);
-	mutex_lock(&mdsc->mutex);
-
-	/* only abort if we didn't race with a real reply */
-	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) {
-		doutc(cl, "aborted request %lld with %d\n", req->r_tid, err);
 
+	if (err < 0) {
 		/*
-		 * ensure we aren't running concurrently with
-		 * ceph_fill_trace or ceph_readdir_prepopulate, which
-		 * rely on locks (dir mutex) held by our caller.
+		 * Timed out or killed.  A reply may be getting published
+		 * concurrently, so serialize against handle_reply() on
+		 * r_completion_lock before deciding to abort.  r_fill_mutex
+		 * additionally fences ceph_fill_trace()/ceph_readdir_prepopulate(),
+		 * which consult CEPH_MDS_R_ABORTED while relying on locks (dir
+		 * mutex) held by our caller.
 		 */
 		mutex_lock(&req->r_fill_mutex);
-		req->r_err = err;
-		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
+		spin_lock(&req->r_completion_lock);
+		got_result = test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
+		if (!got_result) {
+			req->r_err = err;
+			set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
+			aborted = true;
+		}
+		spin_unlock(&req->r_completion_lock);
 		mutex_unlock(&req->r_fill_mutex);
+	} else {
+		/*
+		 * The wait completed, so complete_all() has run and the terminal
+		 * state stored by handle_reply()/__do_request() before it is
+		 * visible without locking.
+		 */
+		got_result = test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
+	}
 
+	if (got_result)
+		err = le32_to_cpu(req->r_reply_info.head->result);
+	else if (!aborted)
+		err = req->r_err;
+
+	if (aborted) {
+		doutc(cl, "aborted request %lld with %d\n", req->r_tid, err);
 		if (req->r_parent &&
 		    (req->r_op & CEPH_MDS_OP_WRITE))
 			ceph_invalidate_dir_request(req);
-	} else {
-		err = req->r_err;
 	}
 
-	mutex_unlock(&mdsc->mutex);
 	return err;
 }
 
@@ -4186,7 +4235,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 		ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
 	}
 out_err:
-	mutex_lock(&mdsc->mutex);
+	spin_lock(&req->r_completion_lock);
 	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
 		if (err) {
 			req->r_err = err;
@@ -4197,7 +4246,7 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	} else {
 		doutc(cl, "reply arrived after request %lld was aborted\n", tid);
 	}
-	mutex_unlock(&mdsc->mutex);
+	spin_unlock(&req->r_completion_lock);
 
 	mutex_unlock(&session->s_mutex);
 
@@ -4259,8 +4308,10 @@ static void handle_forward(struct ceph_mds_client *mdsc,
 		 * 8 bits.
 		 */
 		mutex_lock(&req->r_fill_mutex);
+		spin_lock(&req->r_completion_lock);
 		req->r_err = -EMULTIHOP;
 		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
+		spin_unlock(&req->r_completion_lock);
 		mutex_unlock(&req->r_fill_mutex);
 		aborted = true;
 		pr_warn_ratelimited_client(cl, "forward tid %llu seq overflow\n",
@@ -4268,8 +4319,14 @@ static void handle_forward(struct ceph_mds_client *mdsc,
 	} else {
 		/* resend. forward race not possible; mds would drop */
 		doutc(cl, "forward tid %llu to mds%d (we resend)\n", tid, next_mds);
-		BUG_ON(req->r_err);
-		BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags));
+		/*
+		 * The request had not reached its terminal state when sampled
+		 * above.  It may now race an abort from
+		 * ceph_mdsc_wait_request(), which no longer serializes on
+		 * mdsc->mutex; __do_request() re-checks the terminal state
+		 * under r_completion_lock and unregisters instead of resending
+		 * if that happened, so this is safe.
+		 */
 		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 731d6ad04956..449b5e28b2de 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -394,6 +394,16 @@ struct ceph_mds_request {
 	struct ceph_msg  *r_reply;
 	struct ceph_mds_reply_info_parsed r_reply_info;
 	int r_err;
+
+	/*
+	 * Serializes the request's terminal state (r_err, r_reply,
+	 * CEPH_MDS_R_GOT_RESULT and CEPH_MDS_R_ABORTED) between the reply and
+	 * forward handlers, __do_request() and ceph_mdsc_wait_request().
+	 *
+	 * Lock ordering: mdsc->mutex -> r_fill_mutex -> r_completion_lock.
+	 */
+	spinlock_t r_completion_lock;
+
 	u32               r_readdir_offset;
 
 	struct page *r_locked_page;
-- 
2.47.3