LU-5435 lnet: lustre network latency simulation

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

diff --git a/lnet/lnet/lib-move.c b/lnet/lnet/lib-move.c
index 00f6614..4211e4f 100644 (file)
--- a/lnet/lnet/lib-move.c
+++ b/lnet/lnet/lib-move.c
@@ -42,11 +42,6 @@
 
 #include <lnet/lib-lnet.h>
 
-/** lnet message has credit and can be submitted to lnd for send/receive */
-#define LNET_CREDIT_OK         0
-/** lnet message is waiting for credit */
-#define LNET_CREDIT_WAIT       1
-
 static int local_nid_dist_zero = 1;
 CFS_MODULE_PARM(local_nid_dist_zero, "i", int, 0444,
                 "Reserved");
@@ -1472,7 +1467,7 @@ lnet_send(lnet_nid_t src_nid, lnet_msg_t *msg, lnet_nid_t rtr_nid)
        return 0; /* rc == LNET_CREDIT_OK or LNET_CREDIT_WAIT */
 }
 
-static void
+void
 lnet_drop_message(lnet_ni_t *ni, int cpt, void *private, unsigned int nob)
 {
        lnet_net_lock(cpt);
@@ -1740,7 +1735,7 @@ lnet_parse_ack(lnet_ni_t *ni, lnet_msg_t *msg)
  * \retval LNET_CREDIT_WAIT    If \a msg is blocked because w/o buffer
  * \retval -ve                 error code
  */
-static int
+int
 lnet_parse_forward_locked(lnet_ni_t *ni, lnet_msg_t *msg)
 {
        int     rc = 0;
@@ -1768,6 +1763,33 @@ lnet_parse_forward_locked(lnet_ni_t *ni, lnet_msg_t *msg)
        return rc;
 }
 
+int
+lnet_parse_local(lnet_ni_t *ni, lnet_msg_t *msg)
+{
+       int     rc;
+
+       switch (msg->msg_type) {
+       case LNET_MSG_ACK:
+               rc = lnet_parse_ack(ni, msg);
+               break;
+       case LNET_MSG_PUT:
+               rc = lnet_parse_put(ni, msg);
+               break;
+       case LNET_MSG_GET:
+               rc = lnet_parse_get(ni, msg, msg->msg_rdma_get);
+               break;
+       case LNET_MSG_REPLY:
+               rc = lnet_parse_reply(ni, msg);
+               break;
+       default: /* prevent an unused label if !kernel */
+               LASSERT(0);
+               return -EPROTO;
+       }
+
+       LASSERT(rc == 0 || rc == ENOENT);
+       return rc;
+}
+
 char *
 lnet_msgtyp2str (int type)
 {
@@ -2000,6 +2022,7 @@ lnet_parse(lnet_ni_t *ni, lnet_hdr_t *hdr, lnet_nid_t from_nid,
        msg->msg_type = type;
        msg->msg_private = private;
        msg->msg_receiving = 1;
+       msg->msg_rdma_get = rdma_req;
        msg->msg_len = msg->msg_wanted = payload_length;
        msg->msg_offset = 0;
        msg->msg_hdr = *hdr;
@@ -2034,6 +2057,13 @@ lnet_parse(lnet_ni_t *ni, lnet_hdr_t *hdr, lnet_nid_t from_nid,
 
        lnet_msg_commit(msg, cpt);
 
+       /* message delay simulation */
+       if (unlikely(!list_empty(&the_lnet.ln_delay_rules) &&
+                    lnet_delay_rule_match_locked(hdr, msg))) {
+               lnet_net_unlock(cpt);
+               return 0;
+       }
+
        if (!for_me) {
                rc = lnet_parse_forward_locked(ni, msg);
                lnet_net_unlock(cpt);
@@ -2050,29 +2080,10 @@ lnet_parse(lnet_ni_t *ni, lnet_hdr_t *hdr, lnet_nid_t from_nid,
 
        lnet_net_unlock(cpt);
 
-       switch (type) {
-       case LNET_MSG_ACK:
-               rc = lnet_parse_ack(ni, msg);
-               break;
-       case LNET_MSG_PUT:
-               rc = lnet_parse_put(ni, msg);
-               break;
-       case LNET_MSG_GET:
-               rc = lnet_parse_get(ni, msg, rdma_req);
-               break;
-       case LNET_MSG_REPLY:
-               rc = lnet_parse_reply(ni, msg);
-               break;
-       default:
-               LASSERT(0);
-               rc = -EPROTO;
-               goto free_drop;  /* prevent an unused label if !kernel */
-       }
-
-       if (rc == 0)
-               return 0;
-
-       LASSERT(rc == ENOENT);
+       rc = lnet_parse_local(ni, msg);
+       if (rc != 0)
+               goto free_drop;
+       return 0;
 
  free_drop:
        LASSERT(msg->msg_md == NULL);