b21619 hash ME on RDMA portal i=isaac i=maxim

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

diff --git a/lnet/lnet/lib-me.c b/lnet/lnet/lib-me.c
index 65b28dd..381e512 100644 (file)
--- a/lnet/lnet/lib-me.c
+++ b/lnet/lnet/lib-me.c
@@ -42,6 +42,60 @@
 
 #include <lnet/lib-lnet.h>
 
+static int
+lnet_me_match_portal(lnet_portal_t *ptl, lnet_process_id_t id,
+                     __u64 match_bits, __u64 ignore_bits)
+{
+        cfs_list_t       *mhash = NULL;
+        int               unique;
+
+        LASSERT (!(lnet_portal_is_unique(ptl) &&
+                   lnet_portal_is_wildcard(ptl)));
+
+        /* prefer to check w/o any lock */
+        unique = lnet_match_is_unique(id, match_bits, ignore_bits);
+        if (likely(lnet_portal_is_unique(ptl) ||
+                   lnet_portal_is_wildcard(ptl)))
+                goto match;
+
+        /* unset, new portal */
+        if (unique) {
+                mhash = lnet_portal_mhash_alloc();
+                if (mhash == NULL)
+                        return -ENOMEM;
+        }
+
+        LNET_LOCK();
+        if (lnet_portal_is_unique(ptl) ||
+            lnet_portal_is_wildcard(ptl)) {
+                /* someone set it before me */
+                if (mhash != NULL)
+                        lnet_portal_mhash_free(mhash);
+                LNET_UNLOCK();
+                goto match;
+        }
+
+        /* still not set */
+        LASSERT (ptl->ptl_mhash == NULL);
+        if (unique) {
+                ptl->ptl_mhash = mhash;
+                lnet_portal_setopt(ptl, LNET_PTL_MATCH_UNIQUE);
+        } else {
+                lnet_portal_setopt(ptl, LNET_PTL_MATCH_WILDCARD);
+        }
+        LNET_UNLOCK();
+        return 0;
+
+ match:
+        if (lnet_portal_is_unique(ptl) && !unique)
+                return -EPERM;
+
+        if (lnet_portal_is_wildcard(ptl) && unique)
+                return -EPERM;
+
+        return 0;
+}
+
 int
 LNetMEAttach(unsigned int portal,
              lnet_process_id_t match_id,
@@ -49,7 +103,10 @@ LNetMEAttach(unsigned int portal,
              lnet_unlink_t unlink, lnet_ins_pos_t pos,
              lnet_handle_me_t *handle)
 {
-        lnet_me_t     *me;
+        lnet_me_t        *me;
+        lnet_portal_t    *ptl;
+        cfs_list_t       *head;
+        int               rc;
 
         LASSERT (the_lnet.ln_init);
         LASSERT (the_lnet.ln_refcount > 0);
@@ -57,6 +114,11 @@ LNetMEAttach(unsigned int portal,
         if ((int)portal >= the_lnet.ln_nportals)
                 return -EINVAL;
 
+        ptl = &the_lnet.ln_portals[portal];
+        rc = lnet_me_match_portal(ptl, match_id, match_bits, ignore_bits);
+        if (rc != 0)
+                return rc;
+
         me = lnet_me_alloc();
         if (me == NULL)
                 return -ENOMEM;
@@ -71,13 +133,13 @@ LNetMEAttach(unsigned int portal,
         me->me_md = NULL;
 
         lnet_initialise_handle (&me->me_lh, LNET_COOKIE_TYPE_ME);
+        head = lnet_portal_me_head(portal, match_id, match_bits);
+        LASSERT (head != NULL);
 
         if (pos == LNET_INS_AFTER)
-                cfs_list_add_tail(&me->me_list,
-                                  &(the_lnet.ln_portals[portal].ptl_ml));
+                cfs_list_add_tail(&me->me_list, head);
         else
-                cfs_list_add(&me->me_list,
-                             &(the_lnet.ln_portals[portal].ptl_ml));
+                cfs_list_add(&me->me_list, head);
 
         lnet_me2handle(handle, me);
 
@@ -95,6 +157,7 @@ LNetMEInsert(lnet_handle_me_t current_meh,
 {
         lnet_me_t     *current_me;
         lnet_me_t     *new_me;
+        lnet_portal_t *ptl;
 
         LASSERT (the_lnet.ln_init);
         LASSERT (the_lnet.ln_refcount > 0);
@@ -113,6 +176,16 @@ LNetMEInsert(lnet_handle_me_t current_meh,
                 return -ENOENT;
         }
 
+        LASSERT (current_me->me_portal < the_lnet.ln_nportals);
+
+        ptl = &the_lnet.ln_portals[current_me->me_portal];
+        if (lnet_portal_is_unique(ptl)) {
+                /* nosense to insertion on unique portal */
+                lnet_me_free (new_me);
+                LNET_UNLOCK();
+                return -EPERM;
+        }
+
         new_me->me_portal = current_me->me_portal;
         new_me->me_match_id = match_id;
         new_me->me_match_bits = match_bits;