CDEBUG(D_NET, "peer %s\n", libcfs_nid2str(peer->lp_primary_nid));
+ spin_lock(&peer->lp_lock);
+ peer->lp_state |= LNET_PEER_MARK_DELETED;
+ spin_unlock(&peer->lp_lock);
+
lpni = lnet_get_next_peer_ni_locked(peer, NULL, lpni);
while (lpni != NULL) {
lpni2 = lnet_get_next_peer_ni_locked(peer, NULL, lpni);
return rc2;
}
+/*
+ * Discovering this peer is taking too long. Cancel any Ping or Push
+ * that discovery is waiting on by unlinking the relevant MDs. The
+ * lnet_discovery_event_handler() will proceed from here and complete
+ * the cleanup.
+ */
+static void lnet_peer_cancel_discovery(struct lnet_peer *lp)
+{
+ struct lnet_handle_md ping_mdh;
+ struct lnet_handle_md push_mdh;
+
+ LNetInvalidateMDHandle(&ping_mdh);
+ LNetInvalidateMDHandle(&push_mdh);
+
+ spin_lock(&lp->lp_lock);
+ if (lp->lp_state & LNET_PEER_PING_SENT) {
+ ping_mdh = lp->lp_ping_mdh;
+ LNetInvalidateMDHandle(&lp->lp_ping_mdh);
+ }
+ if (lp->lp_state & LNET_PEER_PUSH_SENT) {
+ push_mdh = lp->lp_push_mdh;
+ LNetInvalidateMDHandle(&lp->lp_push_mdh);
+ }
+ spin_unlock(&lp->lp_lock);
+
+ if (!LNetMDHandleIsInvalid(ping_mdh))
+ LNetMDUnlink(ping_mdh);
+ if (!LNetMDHandleIsInvalid(push_mdh))
+ LNetMDUnlink(push_mdh);
+}
+
static int
lnet_peer_del(struct lnet_peer *peer)
{
+ lnet_peer_cancel_discovery(peer);
lnet_net_lock(LNET_LOCK_EX);
lnet_peer_del_locked(peer);
lnet_net_unlock(LNET_LOCK_EX);
rc = lnet_discover_peer_locked(lpni, cpt, true);
if (rc)
goto out_decref;
+ /* The lpni (or lp) for this NID may have changed and our ref is
+ * the only thing keeping the old one around. Release the ref
+ * and lookup the lpni again
+ */
+ lnet_peer_ni_decref_locked(lpni);
+ lpni = lnet_find_peer_ni_locked(nid);
+ if (!lpni) {
+ rc = -ENOENT;
+ goto out_unlock;
+ }
lp = lpni->lpni_peer_net->lpn_peer;
/* Only try once if discovery is disabled */
return rc;
}
+/* Add the message to the peer's lp_dc_pendq and queue the peer for discovery */
+void
+lnet_peer_queue_message(struct lnet_peer *lp, struct lnet_msg *msg)
+{
+ /* The discovery thread holds net_lock/EX and lp_lock when it splices
+ * the lp_dc_pendq onto a local list for resending. Thus, we do the same
+ * when adding to the list and queuing the peer to ensure that we do not
+ * strand any messages on the lp_dc_pendq. This scheme ensures the
+ * message will be resent even if the peer is already being discovered.
+ * Therefore we needn't check the return value of
+ * lnet_peer_queue_for_discovery(lp).
+ */
+ lnet_net_lock(LNET_LOCK_EX);
+ spin_lock(&lp->lp_lock);
+ list_add_tail(&msg->msg_list, &lp->lp_dc_pendq);
+ spin_unlock(&lp->lp_lock);
+ lnet_peer_queue_for_discovery(lp);
+ lnet_net_unlock(LNET_LOCK_EX);
+}
+
/*
* Queue a peer for the attention of the discovery thread. Call with
* lnet_net_lock/EX held. Returns 0 if the peer was queued, and
CDEBUG(D_NET, "peer %s(%p) state %#x\n",
libcfs_nid2str(lp->lp_primary_nid), lp, lp->lp_state);
+ /* no-op if lnet_peer_del() has already been called on this peer */
+ if (lp->lp_state & LNET_PEER_MARK_DELETED)
+ return 0;
+
if (the_lnet.ln_dc_state != LNET_DC_STATE_RUNNING)
return -ESHUTDOWN;
rc = lnet_peer_merge_data(lp, pbuf);
} else {
lpni = lnet_find_peer_ni_locked(nid);
- if (!lpni) {
+ if (!lpni || lp == lpni->lpni_peer_net->lpn_peer) {
rc = lnet_peer_set_primary_nid(lp, nid, flags);
if (rc) {
CERROR("Primary NID error %s versus %s: %d\n",
} else {
rc = lnet_peer_merge_data(lp, pbuf);
}
+ if (lpni)
+ lnet_peer_ni_decref_locked(lpni);
} else {
struct lnet_peer *new_lp;
new_lp = lpni->lpni_peer_net->lpn_peer;
* should have discovery/MR enabled as well, since
* it's the same peer, which we're about to merge
*/
+ spin_lock(&lp->lp_lock);
+ spin_lock(&new_lp->lp_lock);
if (!(lp->lp_state & LNET_PEER_NO_DISCOVERY))
new_lp->lp_state &= ~LNET_PEER_NO_DISCOVERY;
if (lp->lp_state & LNET_PEER_MULTI_RAIL)
new_lp->lp_state |= LNET_PEER_MULTI_RAIL;
+ /* If we're processing a ping reply then we may be
+ * about to send a push to the peer that we ping'd.
+ * Since the ping reply that we're processing was
+ * received by lp, we need to set the discovery source
+ * NID for new_lp to the NID stored in lp.
+ */
+ if (lp->lp_disc_src_nid != LNET_NID_ANY)
+ new_lp->lp_disc_src_nid = lp->lp_disc_src_nid;
+ spin_unlock(&new_lp->lp_lock);
+ spin_unlock(&lp->lp_lock);
rc = lnet_peer_set_primary_data(new_lp, pbuf);
lnet_consolidate_routes_locked(lp, new_lp);
}
/*
- * Discovering this peer is taking too long. Cancel any Ping or Push
- * that discovery is waiting on by unlinking the relevant MDs. The
- * lnet_discovery_event_handler() will proceed from here and complete
- * the cleanup.
- */
-static void lnet_peer_cancel_discovery(struct lnet_peer *lp)
-{
- struct lnet_handle_md ping_mdh;
- struct lnet_handle_md push_mdh;
-
- LNetInvalidateMDHandle(&ping_mdh);
- LNetInvalidateMDHandle(&push_mdh);
-
- spin_lock(&lp->lp_lock);
- if (lp->lp_state & LNET_PEER_PING_SENT) {
- ping_mdh = lp->lp_ping_mdh;
- LNetInvalidateMDHandle(&lp->lp_ping_mdh);
- }
- if (lp->lp_state & LNET_PEER_PUSH_SENT) {
- push_mdh = lp->lp_push_mdh;
- LNetInvalidateMDHandle(&lp->lp_push_mdh);
- }
- spin_unlock(&lp->lp_lock);
-
- if (!LNetMDHandleIsInvalid(ping_mdh))
- LNetMDUnlink(ping_mdh);
- if (!LNetMDHandleIsInvalid(push_mdh))
- LNetMDUnlink(push_mdh);
-}
-
-/*
* Wait for work to be queued or some other change that must be
* attended to. Returns non-zero if the discovery thread should shut
* down.
CDEBUG(D_NET, "peer %s(%p) state %#x\n",
libcfs_nid2str(lp->lp_primary_nid), lp,
lp->lp_state);
- if (lp->lp_state & LNET_PEER_MARK_DELETION)
+ if (lp->lp_state & (LNET_PEER_MARK_DELETION |
+ LNET_PEER_MARK_DELETED))
rc = lnet_peer_deletion(lp);
else if (lp->lp_state & LNET_PEER_DATA_PRESENT)
rc = lnet_peer_data_present(lp);
return rc;
}
+/* must hold net_lock/0 */
void
-lnet_peer_ni_add_to_recoveryq_locked(struct lnet_peer_ni *lpni)
+lnet_peer_ni_add_to_recoveryq_locked(struct lnet_peer_ni *lpni,
+ struct list_head *recovery_queue,
+ time64_t now)
{
/* the mt could've shutdown and cleaned up the queues */
if (the_lnet.ln_mt_state != LNET_MT_STATE_RUNNING)
return;
- if (list_empty(&lpni->lpni_recovery) &&
- atomic_read(&lpni->lpni_healthv) < LNET_MAX_HEALTH_VALUE) {
- CDEBUG(D_NET, "lpni %s added to recovery queue. Health = %d\n",
- libcfs_nid2str(lpni->lpni_nid),
- atomic_read(&lpni->lpni_healthv));
- list_add_tail(&lpni->lpni_recovery, &the_lnet.ln_mt_peerNIRecovq);
- lnet_peer_ni_addref_locked(lpni);
+ if (!list_empty(&lpni->lpni_recovery))
+ return;
+
+ if (atomic_read(&lpni->lpni_healthv) == LNET_MAX_HEALTH_VALUE)
+ return;
+
+ if (now > lpni->lpni_last_alive + lnet_recovery_limit) {
+ CDEBUG(D_NET, "lpni %s aged out last alive %lld\n",
+ libcfs_nid2str(lpni->lpni_nid),
+ lpni->lpni_last_alive);
+ return;
}
+
+ /* This peer NI is going on the recovery queue, so take a ref on it */
+ lnet_peer_ni_addref_locked(lpni);
+
+ CDEBUG(D_NET, "%s added to recovery queue. last alive: %lld health: %d\n",
+ libcfs_nid2str(lpni->lpni_nid),
+ lpni->lpni_last_alive,
+ atomic_read(&lpni->lpni_healthv));
+
+ list_add_tail(&lpni->lpni_recovery, recovery_queue);
}
/* Call with the ln_api_mutex held */
struct lnet_peer_ni *lpni;
int lncpt;
int cpt;
+ time64_t now;
if (the_lnet.ln_state != LNET_STATE_RUNNING)
return;
+ now = ktime_get_seconds();
+
if (!all) {
lnet_net_lock(LNET_LOCK_EX);
lpni = lnet_find_peer_ni_locked(nid);
return;
}
atomic_set(&lpni->lpni_healthv, value);
- lnet_peer_ni_add_to_recoveryq_locked(lpni);
+ lnet_peer_ni_add_to_recoveryq_locked(lpni,
+ &the_lnet.ln_mt_peerNIRecovq, now);
lnet_peer_ni_decref_locked(lpni);
lnet_net_unlock(LNET_LOCK_EX);
return;
lncpt = cfs_percpt_number(the_lnet.ln_peer_tables);
/*
- * Walk all the peers and reset the healhv for each one to the
- * maximum value.
+ * Walk all the peers and reset the health value for each one to the
+ * specified value.
*/
lnet_net_lock(LNET_LOCK_EX);
for (cpt = 0; cpt < lncpt; cpt++) {
list_for_each_entry(lpni, &lpn->lpn_peer_nis,
lpni_peer_nis) {
atomic_set(&lpni->lpni_healthv, value);
- lnet_peer_ni_add_to_recoveryq_locked(lpni);
+ lnet_peer_ni_add_to_recoveryq_locked(lpni,
+ &the_lnet.ln_mt_peerNIRecovq, now);
}
}
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff