[PATCH] ceph: acquire write lock only if snap trace has really changed

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

MDS replies and cap imports acquire `snap_rwsem` for writing whenever
they contain a snap trace, even when the cached realm information is
already current.  This causes a lot of lock contention with all
processes writing to Ceph, because check_quota_exceeded() needs a read
lock on `snap_rwsem`.  This not only delays all writing processes, but
also the messenger thread, which adds a lot of latency to all MDS
requests.

This patchs adds a wrapper function for ceph_update_snap_trace() which
acquires a read lock instead of a write lock, parses the new snap
trace and calls ceph_update_snap_trace() with a write-upgraded lock
only if the snap trace has really changed.  This avoids lock
contention almost all of the time.

Signed-off-by: Max Kellermann <max.kellermann@ionos.com>
---
 fs/ceph/caps.c       |  5 +--
 fs/ceph/mds_client.c | 11 ++----
 fs/ceph/snap.c       | 92 ++++++++++++++++++++++++++++++++++++++++++++
 fs/ceph/super.h      |  3 ++
 4 files changed, 100 insertions(+), 11 deletions(-)

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index bcb04c6cb92c..42c59e4c3716 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -4578,15 +4578,12 @@ void ceph_handle_caps(struct ceph_mds_session *session,
 	case CEPH_CAP_OP_IMPORT:
 		realm = NULL;
 		if (snaptrace_len) {
-			down_write(&mdsc->snap_rwsem);
-			if (ceph_update_snap_trace(mdsc, snaptrace,
+			if (ceph_handle_snap_trace(mdsc, snaptrace,
 						   snaptrace + snaptrace_len,
 						   false, &realm)) {
-				up_write(&mdsc->snap_rwsem);
 				close_sessions = true;
 				goto done;
 			}
-			downgrade_write(&mdsc->snap_rwsem);
 		} else {
 			down_read(&mdsc->snap_rwsem);
 		}
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 085ae0cfb5f7..f94926c520bf 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -4198,19 +4198,16 @@ static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
 	/* snap trace */
 	realm = NULL;
 	if (rinfo->snapblob_len) {
-		down_write(&mdsc->snap_rwsem);
-		err = ceph_update_snap_trace(mdsc, rinfo->snapblob,
-				rinfo->snapblob + rinfo->snapblob_len,
-				le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
-				&realm);
+		err = ceph_handle_snap_trace(mdsc, rinfo->snapblob,
+					     rinfo->snapblob + rinfo->snapblob_len,
+					     le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
+					     &realm);
 		if (err) {
-			up_write(&mdsc->snap_rwsem);
 			close_sessions = true;
 			if (err == -EIO)
 				ceph_msg_dump(msg);
 			goto out_err;
 		}
-		downgrade_write(&mdsc->snap_rwsem);
 	} else {
 		down_read(&mdsc->snap_rwsem);
 	}
diff --git a/fs/ceph/snap.c b/fs/ceph/snap.c
index 9b79a5eaca93..89f4be790583 100644
--- a/fs/ceph/snap.c
+++ b/fs/ceph/snap.c
@@ -932,6 +932,98 @@ int ceph_update_snap_trace(struct ceph_mds_client *mdsc,
 	return err;
 }
 
+/*
+ * Return a referenced first realm only if the entire snap trace can
+ * be consumed without changing the cached topology or snapshot
+ * contexts.
+ *
+ * Caller must lock snap_rwsem for reading.
+ */
+static struct ceph_snap_realm *get_snap_trace_if_unmodified(struct ceph_mds_client *mdsc,
+							    void *p, void *e)
+{
+	struct ceph_snap_realm *first = NULL, *realm;
+	struct ceph_mds_snap_realm *ri;
+	bool empty;
+	u64 num;
+
+	lockdep_assert_held_read(&mdsc->snap_rwsem);
+
+	do {
+		ceph_decode_need(&p, e, sizeof(*ri), call_update);
+		ri = p;
+		p += sizeof(*ri);
+		num = (u64)le32_to_cpu(ri->num_snaps) +
+		      le32_to_cpu(ri->num_prior_parent_snaps);
+		if (num > (e - p) / sizeof(u64))
+			goto call_update;
+		p += num * sizeof(u64);
+
+		realm = __lookup_snap_realm(mdsc, le64_to_cpu(ri->ino));
+		if (!realm || !realm->parent ||
+		    realm->parent->ino != le64_to_cpu(ri->parent) ||
+		    le64_to_cpu(ri->seq) > realm->seq ||
+		    !realm->cached_context)
+			goto call_update;
+		if (!first)
+			first = realm;
+	} while (p < e);
+
+	/* if there are empty realms, ceph_update_snap_trace() should
+	 * be called for its deferred realm cleanup
+	 */
+	spin_lock(&mdsc->snap_empty_lock);
+	empty = list_empty(&mdsc->snap_empty);
+	spin_unlock(&mdsc->snap_empty_lock);
+	if (!empty)
+		goto call_update;
+
+	/* the ceph_update_snap_trace() call can be omitted (and the
+	 * write lock on snap_rwsem is not necessary); acquire a
+	 * reference to the return value
+	 */
+	ceph_get_snap_realm(mdsc, first);
+	return first;
+
+call_update:
+	/* ceph_update_snap_trace() must be called */
+	return NULL;
+}
+
+/*
+ * Wrapper for ceph_update_snap_trace() which acquires snap_rwsem for
+ * writing only if the new snap trace has really changed.
+ *
+ * Caller must not lock snap_rwsem.  Upon successful return,
+ * snap_rwsem is left locked for reading, but is unlocked on error.
+ */
+int ceph_handle_snap_trace(struct ceph_mds_client *mdsc,
+			   void *p, void *e, bool deletion,
+			   struct ceph_snap_realm **realm_ret)
+{
+	int err;
+
+	lockdep_assert_not_held(&mdsc->snap_rwsem);
+
+	*realm_ret = NULL;
+	down_read(&mdsc->snap_rwsem);
+	if (!deletion) {
+		*realm_ret = get_snap_trace_if_unmodified(mdsc, p, e);
+		if (*realm_ret)
+			return 0;
+	}
+	up_read(&mdsc->snap_rwsem);
+
+	/* reparse from the beginning: the topology may change while unlocked */
+	down_write(&mdsc->snap_rwsem);
+	err = ceph_update_snap_trace(mdsc, p, e, deletion, realm_ret);
+	if (err)
+		up_write(&mdsc->snap_rwsem);
+	else
+		downgrade_write(&mdsc->snap_rwsem);
+	return err;
+}
+
 
 /*
  * Send any cap_snaps that are queued for flush.  Try to carry
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 72d4e30304dc..e5782fbfeeb5 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -1064,6 +1064,9 @@ extern void ceph_put_snap_realm(struct ceph_mds_client *mdsc,
 extern int ceph_update_snap_trace(struct ceph_mds_client *m,
 				  void *p, void *e, bool deletion,
 				  struct ceph_snap_realm **realm_ret);
+int ceph_handle_snap_trace(struct ceph_mds_client *mdsc,
+			   void *p, void *e, bool deletion,
+			   struct ceph_snap_realm **realm_ret);
 void ceph_change_snap_realm(struct inode *inode, struct ceph_snap_realm *realm);
 extern void ceph_handle_snap(struct ceph_mds_client *mdsc,
 			     struct ceph_mds_session *session,
-- 
2.47.3