LU-12222 lnet: Check if we're sending to ourselves

[fs/lustre-release.git] / lnet / lnet / lib-move.c

diff --git a/lnet/lnet/lib-move.c b/lnet/lnet/lib-move.c
index 1706ef7..262e432 100644 (file)
--- a/lnet/lnet/lib-move.c
+++ b/lnet/lnet/lib-move.c
@@ -42,8 +42,6 @@
 #include <linux/nsproxy.h>
 #include <net/net_namespace.h>
 
-extern unsigned int lnet_current_net_count;
-
 static int local_nid_dist_zero = 1;
 module_param(local_nid_dist_zero, int, 0444);
 MODULE_PARM_DESC(local_nid_dist_zero, "Reserved");
@@ -1910,8 +1908,11 @@ lnet_handle_spec_local_nmr_dst(struct lnet_send_data *sd)
  * Local Destination
  * MR Peer
  *
- * Run the selection algorithm on the peer NIs unless we're sending
- * a response, in this case just send to the destination
+ * Don't run the selection algorithm on the peer NIs. By specifying the
+ * local NID, we're also saying that we should always use the destination NID
+ * provided. This handles the case where we should be using the same
+ * destination NID for the all the messages which belong to the same RPC
+ * request.
  */
 static int
 lnet_handle_spec_local_mr_dst(struct lnet_send_data *sd)
@@ -1924,17 +1925,6 @@ lnet_handle_spec_local_mr_dst(struct lnet_send_data *sd)
                return -EINVAL;
        }
 
-       /*
-        * only run the selection algorithm to pick the peer_ni if we're
-        * sending a GET or a PUT. Responses are sent to the same
-        * destination NID provided.
-        */
-       if (!(sd->sd_send_case & SND_RESP)) {
-               sd->sd_best_lpni =
-                 lnet_find_best_lpni_on_net(sd, sd->sd_peer,
-                                            sd->sd_best_ni->ni_net->net_id);
-       }
-
        if (sd->sd_best_lpni &&
            sd->sd_best_lpni->lpni_nid == the_lnet.ln_loni->ni_nid)
                return lnet_handle_lo_send(sd);
@@ -2009,15 +1999,21 @@ lnet_initiate_peer_discovery(struct lnet_peer_ni *lpni,
        }
        /* The peer may have changed. */
        peer = lpni->lpni_peer_net->lpn_peer;
+       spin_lock(&peer->lp_lock);
+       if (lnet_peer_is_uptodate_locked(peer)) {
+               spin_unlock(&peer->lp_lock);
+               lnet_peer_ni_decref_locked(lpni);
+               return 0;
+       }
        /* queue message and return */
        msg->msg_rtr_nid_param = rtr_nid;
        msg->msg_sending = 0;
        msg->msg_txpeer = NULL;
-       spin_lock(&peer->lp_lock);
        list_add_tail(&msg->msg_list, &peer->lp_dc_pendq);
+       primary_nid = peer->lp_primary_nid;
        spin_unlock(&peer->lp_lock);
+
        lnet_peer_ni_decref_locked(lpni);
-       primary_nid = peer->lp_primary_nid;
 
        CDEBUG(D_NET, "msg %p delayed. %s pending discovery\n",
                msg, libcfs_nid2str(primary_nid));
@@ -2638,13 +2634,20 @@ lnet_select_pathway(lnet_nid_t src_nid, lnet_nid_t dst_nid,
 again:
 
        /*
-        * If we're being asked to send to the loopback interface, there
-        * is no need to go through any selection. We can just shortcut
-        * the entire process and send over lolnd
+        * If we're sending to ourselves then there is no need to go through
+        * any selection. We can shortcut the entire process and send over
+        * lolnd.
+        *
+        * However, we make two exceptions to this rule:
+        * 1. If the src_nid is specified then our API defines that we must send
+        *    via that interface.
+        * 2. Recovery messages must be sent to the lnet_ni that is being
+        *    recovered.
         */
        send_data.sd_msg = msg;
        send_data.sd_cpt = cpt;
-       if (LNET_NETTYP(LNET_NIDNET(dst_nid)) == LOLND) {
+       if (src_nid == LNET_NID_ANY && !msg->msg_recovery &&
+           lnet_nid2ni_locked(dst_nid, cpt)) {
                rc = lnet_handle_lo_send(&send_data);
                lnet_net_unlock(cpt);
                return rc;
@@ -2671,11 +2674,10 @@ again:
        msg->msg_src_nid_param = src_nid;
 
        /*
-        * Now that we have a peer_ni, check if we want to discover
-        * the peer. Traffic to the LNET_RESERVED_PORTAL should not
-        * trigger discovery.
+        * If necessary, perform discovery on the peer that owns this peer_ni.
+        * Note, this can result in the ownership of this peer_ni changing
+        * to another peer object.
         */
-       peer = lpni->lpni_peer_net->lpn_peer;
        rc = lnet_initiate_peer_discovery(lpni, msg, rtr_nid, cpt);
        if (rc) {
                lnet_peer_ni_decref_locked(lpni);
@@ -2684,6 +2686,8 @@ again:
        }
        lnet_peer_ni_decref_locked(lpni);
 
+       peer = lpni->lpni_peer_net->lpn_peer;
+
        /*
         * Identify the different send cases
         */
@@ -3824,8 +3828,6 @@ void lnet_monitor_thr_stop(void)
        lnet_clean_local_ni_recoveryq();
        lnet_clean_peer_ni_recoveryq();
        lnet_clean_resendqs();
-
-       return;
 }
 
 void
@@ -5098,9 +5100,9 @@ LNetDist(lnet_nid_t dstnid, lnet_nid_t *srcnidp, __u32 *orderp)
                         * current net namespace.
                         * If not, assign order above 0xffff0000,
                         * to make this ni not a priority. */
-                       if (!net_eq(ni->ni_net_ns, current->nsproxy->net_ns))
-                               order += 0xffff0000;
-
+                       if (current->nsproxy &&
+                           !net_eq(ni->ni_net_ns, current->nsproxy->net_ns))
+                                       order += 0xffff0000;
                        if (srcnidp != NULL)
                                *srcnidp = ni->ni_nid;
                        if (orderp != NULL)