#endif
#include <linux/uaccess.h>
+#include <lnet/udsp.h>
#include <lnet/lib-lnet.h>
#include <uapi/linux/lnet/lnet-dlc.h>
INIT_LIST_HEAD(&lpni->lpni_on_remote_peer_ni_list);
INIT_LIST_HEAD(&lpni->lpni_rtr_pref_nids);
LNetInvalidateMDHandle(&lpni->lpni_recovery_ping_mdh);
- atomic_set(&lpni->lpni_refcount, 1);
+ kref_init(&lpni->lpni_kref);
lpni->lpni_sel_priority = LNET_MAX_SELECTION_PRIORITY;
spin_lock_init(&lpni->lpni_lock);
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);
return rc;
}
+void
+lnet_peer_ni_set_selection_priority(struct lnet_peer_ni *lpni, __u32 priority)
+{
+ lpni->lpni_sel_priority = priority;
+}
+
/*
* Clear the preferred NIDs from a non-multi-rail peer.
*/
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 */
unsigned flags)
{
struct lnet_peer_table *ptable;
+ bool new_lpn = false;
+ int rc;
/* Install the new peer_ni */
lnet_net_lock(LNET_LOCK_EX);
/* Add peer_net to peer */
if (!lpn->lpn_peer) {
+ new_lpn = true;
lpn->lpn_peer = lp;
list_add_tail(&lpn->lpn_peer_nets, &lp->lp_peer_nets);
lnet_peer_addref_locked(lp);
lp->lp_nnis++;
+ /* apply UDSPs */
+ if (new_lpn) {
+ rc = lnet_udsp_apply_policies_on_lpn(lpn);
+ if (rc)
+ CERROR("Failed to apply UDSPs on lpn %s\n",
+ libcfs_net2str(lpn->lpn_net_id));
+ }
+ rc = lnet_udsp_apply_policies_on_lpni(lpni);
+ if (rc)
+ CERROR("Failed to apply UDSPs on lpni %s\n",
+ libcfs_nid2str(lpni->lpni_nid));
+
CDEBUG(D_NET, "peer %s NID %s flags %#x\n",
libcfs_nid2str(lp->lp_primary_nid),
libcfs_nid2str(lpni->lpni_nid), flags);
}
void
-lnet_destroy_peer_ni_locked(struct lnet_peer_ni *lpni)
+lnet_destroy_peer_ni_locked(struct kref *ref)
{
+ struct lnet_peer_ni *lpni = container_of(ref, struct lnet_peer_ni,
+ lpni_kref);
struct lnet_peer_table *ptable;
struct lnet_peer_net *lpn;
CDEBUG(D_NET, "%p nid %s\n", lpni, libcfs_nid2str(lpni->lpni_nid));
- LASSERT(atomic_read(&lpni->lpni_refcount) == 0);
+ LASSERT(kref_read(&lpni->lpni_kref) == 0);
LASSERT(list_empty(&lpni->lpni_txq));
LASSERT(lpni->lpni_txqnob == 0);
LASSERT(list_empty(&lpni->lpni_peer_nis));
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
return false;
}
+/* Delete a peer that has been marked for deletion. NB: when this peer was added
+ * to the discovery queue a reference was taken that will prevent the peer from
+ * actually being freed by this function. After this function exits the
+ * discovery thread should call lnet_peer_discovery_complete() which will
+ * drop that reference as well as wake any waiters that may also be holding a
+ * ref on the peer
+ */
+static int lnet_peer_deletion(struct lnet_peer *lp)
+__must_hold(&lp->lp_lock)
+{
+ struct list_head rlist;
+ struct lnet_route *route, *tmp;
+ int sensitivity = lp->lp_health_sensitivity;
+
+ INIT_LIST_HEAD(&rlist);
+
+ lp->lp_state &= ~(LNET_PEER_DISCOVERING | LNET_PEER_FORCE_PING |
+ LNET_PEER_FORCE_PUSH);
+ 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;
+
+ spin_unlock(&lp->lp_lock);
+
+ mutex_lock(&the_lnet.ln_api_mutex);
+
+ lnet_net_lock(LNET_LOCK_EX);
+ /* remove the peer from the discovery work
+ * queue if it's on there in preparation
+ * of deleting it.
+ */
+ if (!list_empty(&lp->lp_dc_list))
+ list_del(&lp->lp_dc_list);
+ list_for_each_entry_safe(route, tmp,
+ &lp->lp_routes,
+ lr_gwlist)
+ lnet_move_route(route, NULL, &rlist);
+ lnet_net_unlock(LNET_LOCK_EX);
+
+ /* lnet_peer_del() deletes all the peer NIs owned by this peer */
+ lnet_peer_del(lp);
+
+ list_for_each_entry_safe(route, tmp,
+ &rlist, lr_list) {
+ /* re-add these routes */
+ lnet_add_route(route->lr_net,
+ route->lr_hops,
+ route->lr_nid,
+ route->lr_priority,
+ sensitivity);
+ LIBCFS_FREE(route, sizeof(*route));
+ }
+
+ mutex_unlock(&the_lnet.ln_api_mutex);
+
+ spin_lock(&lp->lp_lock);
+
+ return 0;
+}
+
/*
* Update a peer using the data received.
*/
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_DATA_PRESENT)
+ 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);
else if (lp->lp_state & LNET_PEER_PING_FAILED)
rc = lnet_peer_ping_failed(lp);
if (the_lnet.ln_dc_state == LNET_DC_STATE_STOPPING)
break;
- if (lp->lp_state & LNET_PEER_MARK_DELETION) {
- struct list_head rlist;
- struct lnet_route *route, *tmp;
- int sensitivity = lp->lp_health_sensitivity;
-
- INIT_LIST_HEAD(&rlist);
-
- /*
- * remove the peer from the discovery work
- * queue if it's on there in preparation
- * of deleting it.
- */
- if (!list_empty(&lp->lp_dc_list))
- list_del(&lp->lp_dc_list);
-
- lnet_net_unlock(LNET_LOCK_EX);
-
- mutex_lock(&the_lnet.ln_api_mutex);
-
- lnet_net_lock(LNET_LOCK_EX);
- list_for_each_entry_safe(route, tmp,
- &lp->lp_routes,
- lr_gwlist)
- lnet_move_route(route, NULL, &rlist);
- lnet_net_unlock(LNET_LOCK_EX);
-
- /* delete the peer */
- lnet_peer_del(lp);
-
- list_for_each_entry_safe(route, tmp,
- &rlist, lr_list) {
- /* re-add these routes */
- lnet_add_route(route->lr_net,
- route->lr_hops,
- route->lr_nid,
- route->lr_priority,
- sensitivity);
- LIBCFS_FREE(route, sizeof(*route));
- }
- mutex_unlock(&the_lnet.ln_api_mutex);
-
- lnet_net_lock(LNET_LOCK_EX);
- }
}
lnet_net_unlock(LNET_LOCK_EX);
aliveness = (lnet_is_peer_ni_alive(lp)) ? "up" : "down";
CDEBUG(D_WARNING, "%-24s %4d %5s %5d %5d %5d %5d %5d %ld\n",
- libcfs_nid2str(lp->lpni_nid), atomic_read(&lp->lpni_refcount),
+ libcfs_nid2str(lp->lpni_nid), kref_read(&lp->lpni_kref),
aliveness, lp->lpni_net->net_tunables.lct_peer_tx_credits,
lp->lpni_rtrcredits, lp->lpni_minrtrcredits,
lp->lpni_txcredits, lp->lpni_mintxcredits, lp->lpni_txqnob);
lnet_is_peer_ni_alive(lp) ? "up" : "down");
*nid = lp->lpni_nid;
- *refcount = atomic_read(&lp->lpni_refcount);
+ *refcount = kref_read(&lp->lpni_kref);
*ni_peer_tx_credits =
lp->lpni_net->net_tunables.lct_peer_tx_credits;
*peer_tx_credits = lp->lpni_txcredits;
snprintf(lpni_info->cr_aliveness, LNET_MAX_STR_LEN,
lnet_is_peer_ni_alive(lpni) ? "up" : "down");
- lpni_info->cr_refcount = atomic_read(&lpni->lpni_refcount);
+ lpni_info->cr_refcount = kref_read(&lpni->lpni_kref);
lpni_info->cr_ni_peer_tx_credits = (lpni->lpni_net != NULL) ?
lpni->lpni_net->net_tunables.lct_peer_tx_credits : 0;
lpni_info->cr_peer_tx_credits = lpni->lpni_txcredits;
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 (!lpni->lpni_last_alive) {
+ CDEBUG(D_NET,
+ "lpni %s(%p) not eligible for recovery last alive %lld\n",
+ libcfs_nid2str(lpni->lpni_nid), lpni,
+ lpni->lpni_last_alive);
+ 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);
+ /* Reset the ping count so that if this peer NI is added back to
+ * the recovery queue we will send the first ping right away.
+ */
+ lpni->lpni_ping_count = 0;
+ return;
+ }
+
+ /* This peer NI is going on the recovery queue, so take a ref on it */
+ lnet_peer_ni_addref_locked(lpni);
+
+ lnet_peer_ni_set_next_ping(lpni, now);
+
+ CDEBUG(D_NET, "%s added to recovery queue. ping count: %u next ping: %lld last alive: %lld health: %d\n",
+ libcfs_nid2str(lpni->lpni_nid),
+ lpni->lpni_ping_count,
+ lpni->lpni_next_ping,
+ 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