Branch b1_x

lnet_parse_put() should re-match after lnet_eager_recv_locked()

b=17537
i=isaac
i=maxim

diff --git a/lnet/include/lnet/lib-types.h b/lnet/include/lnet/lib-types.h
index e456514..ca60fb6 100644 (file)
--- a/lnet/include/lnet/lib-types.h
+++ b/lnet/include/lnet/lib-types.h
@@ -475,6 +475,7 @@ typedef struct {
 typedef struct {
         struct list_head ptl_ml;  /* match list */
         struct list_head ptl_msgq; /* messages blocking for MD */
+        __u64            ptl_ml_version;    /* validity stamp, only changed for new attached MD */
         __u64            ptl_msgq_version;  /* validity stamp */
         unsigned int     ptl_options;
 } lnet_portal_t;

diff --git a/lnet/lnet/lib-md.c b/lnet/lnet/lib-md.c
index 39fa978..be16ba6 100644 (file)
--- a/lnet/lnet/lib-md.c
+++ b/lnet/lnet/lib-md.c
@@ -244,6 +244,8 @@ LNetMDAttach(lnet_handle_me_t meh, lnet_md_t umd,
         } else {
                 rc = lib_md_build(md, &umd, unlink);
                 if (rc == 0) {
+                        the_lnet.ln_portals[me->me_portal].ptl_ml_version++;
+
                         me->me_md = md;
                         md->md_me = me;
 

diff --git a/lnet/lnet/lib-move.c b/lnet/lnet/lib-move.c
index 26fccb4..972676c 100644 (file)
--- a/lnet/lnet/lib-move.c
+++ b/lnet/lnet/lib-move.c
@@ -1604,6 +1604,7 @@ lnet_parse_put(lnet_ni_t *ni, lnet_msg_t *msg)
 {
         int               rc;
         int               index;
+        __u64             version;
         lnet_hdr_t       *hdr = &msg->msg_hdr;
         unsigned int      rlength = hdr->payload_length;
         unsigned int      mlength = 0;
@@ -1611,6 +1612,7 @@ lnet_parse_put(lnet_ni_t *ni, lnet_msg_t *msg)
         lnet_process_id_t src = {/* .nid = */ hdr->src_nid,
                                  /* .pid = */ hdr->src_pid};
         lnet_libmd_t     *md;
+        lnet_portal_t    *ptl;
 
         /* Convert put fields to host byte order */
         hdr->msg.put.match_bits = le64_to_cpu(hdr->msg.put.match_bits);
@@ -1618,9 +1620,11 @@ lnet_parse_put(lnet_ni_t *ni, lnet_msg_t *msg)
         hdr->msg.put.offset = le32_to_cpu(hdr->msg.put.offset);
 
         index = hdr->msg.put.ptl_index;
+        ptl = &the_lnet.ln_portals[index];
 
         LNET_LOCK();
 
+ again:
         rc = lnet_match_md(index, LNET_MD_OP_PUT, src,
                            rlength, hdr->msg.put.offset,
                            hdr->msg.put.match_bits, msg,
@@ -1631,24 +1635,31 @@ lnet_parse_put(lnet_ni_t *ni, lnet_msg_t *msg)
 
         case LNET_MATCHMD_OK:
                 LNET_UNLOCK();
-                lnet_recv_put(md, msg, 0, offset, mlength);
+                lnet_recv_put(md, msg, msg->msg_delayed, offset, mlength);
                 return 0;
 
         case LNET_MATCHMD_NONE:
-                rc = lnet_eager_recv_locked(msg);
-                if (rc == 0 && !the_lnet.ln_shutdown) {
-                        list_add_tail(&msg->msg_list,
-                                      &the_lnet.ln_portals[index].ptl_msgq);
+                version = ptl->ptl_ml_version;
+
+                rc = 0;
+                if (!msg->msg_delayed)
+                        rc = lnet_eager_recv_locked(msg);
+
+                if (rc == 0 &&
+                    !the_lnet.ln_shutdown &&
+                    ((ptl->ptl_options & LNET_PTL_LAZY) != 0)) {
+                        if (version != ptl->ptl_ml_version)
+                                goto again;
 
-                        the_lnet.ln_portals[index].ptl_msgq_version++;
+                        list_add_tail(&msg->msg_list, &ptl->ptl_msgq);
+                        ptl->ptl_msgq_version++;
+                        LNET_UNLOCK();
 
                         CDEBUG(D_NET, "Delaying PUT from %s portal %d match "
                                LPU64" offset %d length %d: no match \n",
                                libcfs_id2str(src), index,
                                hdr->msg.put.match_bits,
                                hdr->msg.put.offset, rlength);
-
-                        LNET_UNLOCK();
                         return 0;
                 }
                 /* fall through */