[PATCH v4 3/3] ceph: don't unregister an MDS session before removing its caps

@@ -4261,6 +4261,10 @@ static void handle_cap_export(struct inode *inode, struct ceph_mds_caps *ex,
 		}
 		new_cap = ceph_get_cap(mdsc, NULL);
 	} else {
+		if (tsession == ERR_PTR(-EAGAIN))
+			/* locks already dropped */
+			return;
+
 		WARN_ON(1);
 		tsession = NULL;
 		target = -1;

@@ -1783,6 +1783,30 @@ static int __open_session(struct ceph_mds_client *mdsc,
 	return 0;
 }
 
+/* Is this rank still occupied by a session being torn down? */
+static bool __mds_rank_closing(struct ceph_mds_client *mdsc, int mds)
+{
+	return mds < mdsc->max_sessions && mdsc->sessions[mds] &&
+		mdsc->sessions[mds]->s_state == CEPH_MDS_SESSION_CLOSED;
+}
+
+/*
+ * Wait until the rank is no longer occupied by a CLOSED session.
+ *
+ * The caller must hold mdsc->mutex.  The mutex is dropped while sleeping and
+ * held again on return.  Another session may occupy the rank by then.  The
+ * wait also ends when shutdown starts, so the caller must check
+ * mdsc->stopping before proceeding.
+ */
+static void wait_for_mds_rank_not_closing(struct ceph_mds_client *mdsc,
+					  int mds)
+{
+	wait_event_cmd(mdsc->session_close_wq,
+		       !__mds_rank_closing(mdsc, mds) || mdsc->stopping,
+		       mutex_unlock(&mdsc->mutex),
+		       mutex_lock(&mdsc->mutex));
+}
+
 /*
  * open sessions for any export targets for the given mds
  *

@@ -1800,6 +1824,16 @@ __open_export_target_session(struct ceph_mds_client *mdsc, int target)
 		if (IS_ERR(session))
 			return session;
 	}
+	if (session->s_state == CEPH_MDS_SESSION_CLOSED) {
+		/*
+		 * handle_session() is currently closing this session;
+		 * it stays registered until its caps are gone.  Do
+		 * not return it to our caller because we don't want
+		 * it to attach new caps to it.
+		 */
+		ceph_put_mds_session(session);
+		return ERR_PTR(-EAGAIN);
+	}
 	if (session->s_state == CEPH_MDS_SESSION_NEW ||
 	    session->s_state == CEPH_MDS_SESSION_CLOSING) {
 		ret = __open_session(mdsc, session);

@@ -1819,7 +1853,19 @@ ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target)
 	doutc(cl, "to mds%d\n", target);
 
 	mutex_lock(&mdsc->mutex);
-	session = __open_export_target_session(mdsc, target);
+	for (;;) {
+		session = __open_export_target_session(mdsc, target);
+		if (session != ERR_PTR(-EAGAIN) || mdsc->stopping)
+			break;
+
+		/*
+		 * Keep the exported cap on its old session until the
+		 * target rank is vacant.  Dropping it here can discard a
+		 * dirty auth cap; handle_session() wakes us after removing
+		 * the old target session's caps and unregistering it.
+		 */
+		wait_for_mds_rank_not_closing(mdsc, target);
+	}
 	mutex_unlock(&mdsc->mutex);
 
 	return session;

@@ -4553,8 +4599,20 @@ static void handle_session(struct ceph_mds_session *session,
 			ceph_metric_bind_session(mdsc, session);
 	}
 	if (op == CEPH_SESSION_CLOSE) {
+		/*
+		 * Pin the session for the rest of this function.  The
+		 * __unregister_session() call is deferred until after
+		 * remove_session_caps() below, or else other
+		 * processes may find caps still assigned to this
+		 * session while working with a new session object.
+		 */
 		ceph_get_mds_session(session);
-		__unregister_session(mdsc, session);
+
+		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
+			pr_info_client(cl, "mds%d reconnect denied\n",
+				       session->s_mds);
+
+		session->s_state = CEPH_MDS_SESSION_CLOSED;
 	}
 	/* FIXME: this ttl calculation is generous */
 	session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;

@@ -4612,12 +4670,24 @@ static void handle_session(struct ceph_mds_session *session,
 		break;
 
 	case CEPH_SESSION_CLOSE:
-		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
-			pr_info_client(cl, "mds%d reconnect denied\n",
-				       session->s_mds);
-		session->s_state = CEPH_MDS_SESSION_CLOSED;
 		cleanup_session_requests(mdsc, session);
 		remove_session_caps(session);
+
+		/*
+		 * Now that all caps are removed, it is safe to release
+		 * the MDS rank and allow other processes to create a
+		 * new session object.
+		 *
+		 * A concurrent ceph_mdsc_close_sessions() or
+		 * check_new_map() may have unregistered the session
+		 * already, so check __verify_registered_session()
+		 * first.
+		 */
+		mutex_lock(&mdsc->mutex);
+		if (!__verify_registered_session(mdsc, session))
+			__unregister_session(mdsc, session);
+		mutex_unlock(&mdsc->mutex);
+
 		wake = 2; /* for good measure */
 		wake_up_all(&mdsc->session_close_wq);
 		break;

@@ -5168,16 +5238,6 @@ static int send_mds_reconnect(struct ceph_mds_client *mdsc,
 	/* Serialized by s_mutex against concurrent ceph_get_deleg_ino(). */
 	xa_destroy(&session->s_delegated_inos);
 	atomic_set(&session->s_num_deleg_inos, 0);
-	if (session->s_state == CEPH_MDS_SESSION_CLOSED ||
-	    session->s_state == CEPH_MDS_SESSION_REJECTED) {
-		pr_info_client(cl, "mds%d skipping reconnect, session %s\n",
-			       mds,
-			       ceph_session_state_name(session->s_state));
-		mutex_unlock(&session->s_mutex);
-		ceph_msg_put(reply);
-		err = -ESTALE;
-		goto fail_return;
-	}
 
 	/* s_mutex -> mdsc->mutex matches cleanup_session_requests() order. */
 	mutex_lock(&mdsc->mutex);

@@ -5191,12 +5251,23 @@ static int send_mds_reconnect(struct ceph_mds_client *mdsc,
 		err = -ENOENT;
 		goto fail_return;
 	}
-	mutex_unlock(&mdsc->mutex);
+	if (session->s_state == CEPH_MDS_SESSION_CLOSED ||
+	    session->s_state == CEPH_MDS_SESSION_REJECTED) {
+		mutex_unlock(&mdsc->mutex);
+		pr_info_client(cl, "mds%d skipping reconnect, session %s\n",
+			       mds,
+			       ceph_session_state_name(session->s_state));
+		mutex_unlock(&session->s_mutex);
+		ceph_msg_put(reply);
+		err = -ESTALE;
+		goto fail_return;
+	}
 
 	pr_info_client(cl, "mds%d reconnect start\n", mds);
 	old_state = session->s_state;
 	session->s_state = CEPH_MDS_SESSION_RECONNECTING;
 	session->s_seq = 0;
+	mutex_unlock(&mdsc->mutex);
 
 	doutc(cl, "session %p state %s\n", session,
 	      ceph_session_state_name(session->s_state));

@@ -5338,7 +5409,10 @@ static int send_mds_reconnect(struct ceph_mds_client *mdsc,
 	 * (check_new_map) can retry.  Without this, a transient build
 	 * failure strands the session in RECONNECTING indefinitely.
 	 */
-	session->s_state = old_state;
+	mutex_lock(&mdsc->mutex);
+	if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
+		session->s_state = old_state;
+	mutex_unlock(&mdsc->mutex);
 	mutex_unlock(&session->s_mutex);
 fail_nomsg:
 	ceph_pagelist_release(recon_state.pagelist);

@@ -5746,23 +5820,29 @@ static void ceph_mdsc_reset_workfn(struct work_struct *work)
 			continue;
 		}
 		sessions[i]->s_state = CEPH_MDS_SESSION_CLOSED;
-		__unregister_session(mdsc, sessions[i]);
-		__wake_requests(mdsc, &sessions[i]->s_waiting);
 		mutex_unlock(&mdsc->mutex);
 
 		mutex_lock(&sessions[i]->s_mutex);
 		cleanup_session_requests(mdsc, sessions[i]);
 		remove_session_caps(sessions[i]);
+
+		/* Keep the rank occupied until all old-session caps are gone. */
+		mutex_lock(&mdsc->mutex);
+		if (!__verify_registered_session(mdsc, sessions[i]))
+			__unregister_session(mdsc, sessions[i]);
+		mutex_unlock(&mdsc->mutex);
+
 		mutex_unlock(&sessions[i]->s_mutex);
 
 		wake_up_all(&mdsc->session_close_wq);
 
-		ceph_put_mds_session(sessions[i]);
-
 		mutex_lock(&mdsc->mutex);
+		__wake_requests(mdsc, &sessions[i]->s_waiting);
 		kick_requests(mdsc, mds);
 		mutex_unlock(&mdsc->mutex);
 
+		ceph_put_mds_session(sessions[i]);
+
 		torn_down++;
 		pr_info_client(cl, "mds%d session reset complete\n", mds);
 	}

@@ -5864,6 +5944,12 @@ static void check_new_map(struct ceph_mds_client *mdsc,
 		s = __ceph_lookup_mds_session(mdsc, i);
 		if (!s)
 			continue;
+
+		if (s->s_state == CEPH_MDS_SESSION_CLOSED) {
+			ceph_put_mds_session(s);
+			continue;
+		}
+
 		oldstate = ceph_mdsmap_get_state(oldmap, i);
 		newstate = ceph_mdsmap_get_state(newmap, i);
 

@@ -5876,18 +5962,24 @@ static void check_new_map(struct ceph_mds_client *mdsc,
 
 		if (i >= newmap->possible_max_rank) {
 			/* force close session for stopped mds */
-			__unregister_session(mdsc, s);
-			__wake_requests(mdsc, &s->s_waiting);
+			s->s_state = CEPH_MDS_SESSION_CLOSED;
 			mutex_unlock(&mdsc->mutex);
 
 			mutex_lock(&s->s_mutex);
 			cleanup_session_requests(mdsc, s);
 			remove_session_caps(s);
+
+			mutex_lock(&mdsc->mutex);
+			if (!__verify_registered_session(mdsc, s))
+				__unregister_session(mdsc, s);
+			mutex_unlock(&mdsc->mutex);
 			mutex_unlock(&s->s_mutex);
 
-			ceph_put_mds_session(s);
+			wake_up_all(&mdsc->session_close_wq);
 
 			mutex_lock(&mdsc->mutex);
+			__wake_requests(mdsc, &s->s_waiting);
+			ceph_put_mds_session(s);
 			if (mdsc->mdsmap->m_epoch != map_epoch)
 				return;
 			kick_requests(mdsc, i);

@@ -5906,6 +5998,16 @@ static void check_new_map(struct ceph_mds_client *mdsc,
 				ceph_put_mds_session(s);
 				return;
 			}
+			if (s->s_state == CEPH_MDS_SESSION_CLOSED) {
+				/*
+				 * handle_session() set state=CLOSED
+				 * in the mutex gap above and is tearing it
+				 * down
+				 */
+				mutex_unlock(&s->s_mutex);
+				ceph_put_mds_session(s);
+				continue;
+			}
 			ceph_con_close(&s->s_con);
 			mutex_unlock(&s->s_mutex);
 			s->s_state = CEPH_MDS_SESSION_RESTARTING;

@@ -5964,6 +6066,9 @@ static void check_new_map(struct ceph_mds_client *mdsc,
 	 * Only open and reconnect sessions that don't exist yet.
 	 */
 	for (i = 0; i < newmap->possible_max_rank; i++) {
+		if (mdsc->stopping)
+			return;
+
 		/*
 		 * In case the import MDS is crashed just after
 		 * the EImportStart journal is flushed, so when

@@ -5989,6 +6094,19 @@ static void check_new_map(struct ceph_mds_client *mdsc,
 		 * reconnection request in up:reconnect state.
 		 */
 		s = __ceph_lookup_mds_session(mdsc, i);
+		if (s && s->s_state == CEPH_MDS_SESSION_CLOSED) {
+			/*
+			 * Wait for handle_session() to remove the caps and
+			 * unregister this session, so the reconnect below
+			 * uses a fresh session on the now vacant rank
+			 */
+			ceph_put_mds_session(s);
+			wait_for_mds_rank_not_closing(mdsc, i);
+			if (mdsc->stopping ||
+			    mdsc->mdsmap->m_epoch != map_epoch)
+				return;
+			s = NULL;
+		}
 		if (likely(!s)) {
 			s = __open_export_target_session(mdsc, i);
 			if (IS_ERR(s)) {

@@ -7149,9 +7267,7 @@ static void mds_peer_reset(struct ceph_connection *con)
 	 * Snapshot session state with READ_ONCE, then revalidate under
 	 * mdsc->mutex before acting.  The subsequent mdsc->mutex
 	 * section rechecks s_state to catch concurrent transitions, so
-	 * the lockless snapshot here is safe.  s->s_mutex is taken
-	 * separately for cleanup after unregistration, which avoids
-	 * introducing a new s->s_mutex + mdsc->mutex nesting.
+	 * the lockless snapshot here is safe.
 	 */
 	session_state = READ_ONCE(s->s_state);
 

@@ -7176,18 +7292,24 @@ static void mds_peer_reset(struct ceph_connection *con)
 
 		ceph_get_mds_session(s);
 		s->s_state = CEPH_MDS_SESSION_CLOSED;
-		__unregister_session(mdsc, s);
-		__wake_requests(mdsc, &s->s_waiting);
 		mutex_unlock(&mdsc->mutex);
 
 		mutex_lock(&s->s_mutex);
 		cleanup_session_requests(mdsc, s);
 		remove_session_caps(s);
+
+		/* Keep the rank occupied until all old-session caps are gone. */
+		mutex_lock(&mdsc->mutex);
+		if (!__verify_registered_session(mdsc, s))
+			__unregister_session(mdsc, s);
+		mutex_unlock(&mdsc->mutex);
+
 		mutex_unlock(&s->s_mutex);
 
 		wake_up_all(&mdsc->session_close_wq);
 
 		mutex_lock(&mdsc->mutex);
+		__wake_requests(mdsc, &s->s_waiting);
 		kick_requests(mdsc, s->s_mds);
 		mutex_unlock(&mdsc->mutex);