LU-1622 lnet: hash MEs on wildcard portal

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

diff --git a/lnet/lnet/lib-me.c b/lnet/lnet/lib-me.c
index d4f325d..8df9917 100644 (file)
--- a/lnet/lnet/lib-me.c
+++ b/lnet/lnet/lib-me.c
@@ -78,25 +78,25 @@ LNetMEAttach(unsigned int portal,
              lnet_handle_me_t *handle)
 {
        struct lnet_match_table *mtable;
              lnet_handle_me_t *handle)
 {
        struct lnet_match_table *mtable;

-        lnet_me_t        *me;
-        cfs_list_t       *head;
+       struct lnet_me          *me;
+       cfs_list_t              *head;

 
 

-        LASSERT (the_lnet.ln_init);
-        LASSERT (the_lnet.ln_refcount > 0);
+       LASSERT(the_lnet.ln_init);
+       LASSERT(the_lnet.ln_refcount > 0);

 
 

-        if ((int)portal >= the_lnet.ln_nportals)
-                return -EINVAL;
+       if ((int)portal >= the_lnet.ln_nportals)
+               return -EINVAL;

 
        mtable = lnet_mt_of_attach(portal, match_id,
                                   match_bits, ignore_bits, pos);
        if (mtable == NULL) /* can't match portal type */
                return -EPERM;
 
 
        mtable = lnet_mt_of_attach(portal, match_id,
                                   match_bits, ignore_bits, pos);
        if (mtable == NULL) /* can't match portal type */
                return -EPERM;
 

-        me = lnet_me_alloc();
-        if (me == NULL)
-                return -ENOMEM;
+       me = lnet_me_alloc();
+       if (me == NULL)
+               return -ENOMEM;

 
 

-       lnet_res_lock();
+       lnet_res_lock(mtable->mt_cpt);

 
         me->me_portal = portal;
         me->me_match_id = match_id;
 
         me->me_portal = portal;
         me->me_match_id = match_id;


@@ -105,21 +105,25 @@ LNetMEAttach(unsigned int portal,
         me->me_unlink = unlink;
         me->me_md = NULL;
 
         me->me_unlink = unlink;
         me->me_md = NULL;
 

-       lnet_res_lh_initialize(&the_lnet.ln_me_container, &me->me_lh);
-       head = lnet_mt_match_head(mtable, match_id, match_bits);
-        LASSERT (head != NULL);
+       lnet_res_lh_initialize(the_lnet.ln_me_containers[mtable->mt_cpt],
+                              &me->me_lh);
+       if (ignore_bits != 0)
+               head = &mtable->mt_mhash[LNET_MT_HASH_IGNORE];
+       else
+               head = lnet_mt_match_head(mtable, match_id, match_bits);

 
 

-        if (pos == LNET_INS_AFTER)
-                cfs_list_add_tail(&me->me_list, head);
-        else
-                cfs_list_add(&me->me_list, head);
-
-        lnet_me2handle(handle, me);
+       me->me_pos = head - &mtable->mt_mhash[0];
+       if (pos == LNET_INS_AFTER || pos == LNET_INS_LOCAL)
+               cfs_list_add_tail(&me->me_list, head);
+       else
+               cfs_list_add(&me->me_list, head);

 
 

-       lnet_res_unlock();
+       lnet_me2handle(handle, me);

 
 

-        return 0;
+       lnet_res_unlock(mtable->mt_cpt);
+       return 0;

 }
 }

+EXPORT_SYMBOL(LNetMEAttach);

 
 /**
  * Create and a match entry and insert it before or after the ME pointed to by
 
 /**
  * Create and a match entry and insert it before or after the ME pointed to by


@@ -145,24 +149,30 @@ LNetMEInsert(lnet_handle_me_t current_meh,
              lnet_unlink_t unlink, lnet_ins_pos_t pos,
              lnet_handle_me_t *handle)
 {
              lnet_unlink_t unlink, lnet_ins_pos_t pos,
              lnet_handle_me_t *handle)
 {

-        lnet_me_t     *current_me;
-        lnet_me_t     *new_me;
-        lnet_portal_t *ptl;
+       struct lnet_me          *current_me;
+       struct lnet_me          *new_me;
+       struct lnet_portal      *ptl;
+       int                     cpt;
+
+       LASSERT(the_lnet.ln_init);
+       LASSERT(the_lnet.ln_refcount > 0);

 
 

-        LASSERT (the_lnet.ln_init);
-        LASSERT (the_lnet.ln_refcount > 0);
+       if (pos == LNET_INS_LOCAL)
+               return -EPERM;

 
 

-        new_me = lnet_me_alloc();
-        if (new_me == NULL)
-                return -ENOMEM;
+       new_me = lnet_me_alloc();
+       if (new_me == NULL)
+               return -ENOMEM;

 
 

-       lnet_res_lock();
+       cpt = lnet_cpt_of_cookie(current_meh.cookie);
+
+       lnet_res_lock(cpt);

 
        current_me = lnet_handle2me(&current_meh);
        if (current_me == NULL) {
                lnet_me_free_locked(new_me);
 
 
        current_me = lnet_handle2me(&current_meh);
        if (current_me == NULL) {
                lnet_me_free_locked(new_me);
 

-               lnet_res_unlock();
+               lnet_res_unlock(cpt);

                return -ENOENT;
        }
 
                return -ENOENT;
        }
 


@@ -170,12 +180,13 @@ LNetMEInsert(lnet_handle_me_t current_meh,
 
        ptl = the_lnet.ln_portals[current_me->me_portal];
        if (lnet_ptl_is_unique(ptl)) {
 
        ptl = the_lnet.ln_portals[current_me->me_portal];
        if (lnet_ptl_is_unique(ptl)) {

-                /* nosense to insertion on unique portal */
+               /* nosense to insertion on unique portal */

                lnet_me_free_locked(new_me);
                lnet_me_free_locked(new_me);

-               lnet_res_unlock();
-                return -EPERM;
+               lnet_res_unlock(cpt);
+               return -EPERM;

         }
 
         }
 

+       new_me->me_pos = current_me->me_pos;

         new_me->me_portal = current_me->me_portal;
         new_me->me_match_id = match_id;
         new_me->me_match_bits = match_bits;
         new_me->me_portal = current_me->me_portal;
         new_me->me_match_id = match_id;
         new_me->me_match_bits = match_bits;


@@ -183,7 +194,7 @@ LNetMEInsert(lnet_handle_me_t current_meh,
         new_me->me_unlink = unlink;
         new_me->me_md = NULL;
 
         new_me->me_unlink = unlink;
         new_me->me_md = NULL;
 

-       lnet_res_lh_initialize(&the_lnet.ln_me_container, &new_me->me_lh);
+       lnet_res_lh_initialize(the_lnet.ln_me_containers[cpt], &new_me->me_lh);

 
         if (pos == LNET_INS_AFTER)
                 cfs_list_add(&new_me->me_list, &current_me->me_list);
 
         if (pos == LNET_INS_AFTER)
                 cfs_list_add(&new_me->me_list, &current_me->me_list);


@@ -192,10 +203,11 @@ LNetMEInsert(lnet_handle_me_t current_meh,
 
         lnet_me2handle(handle, new_me);
 
 
         lnet_me2handle(handle, new_me);
 

-       lnet_res_unlock();
+       lnet_res_unlock(cpt);

 
        return 0;
 }
 
        return 0;
 }

+EXPORT_SYMBOL(LNetMEInsert);

 
 /**
  * Unlink a match entry from its match list.
 
 /**
  * Unlink a match entry from its match list.


@@ -214,18 +226,20 @@ LNetMEInsert(lnet_handle_me_t current_meh,
 int
 LNetMEUnlink(lnet_handle_me_t meh)
 {
 int
 LNetMEUnlink(lnet_handle_me_t meh)
 {

-        lnet_me_t    *me;
-        lnet_libmd_t *md;
-        lnet_event_t  ev;
+       lnet_me_t       *me;
+       lnet_libmd_t    *md;
+       lnet_event_t    ev;
+       int             cpt;

 
 

-        LASSERT (the_lnet.ln_init);
-        LASSERT (the_lnet.ln_refcount > 0);
+       LASSERT(the_lnet.ln_init);
+       LASSERT(the_lnet.ln_refcount > 0);

 
 

-       lnet_res_lock();
+       cpt = lnet_cpt_of_cookie(meh.cookie);
+       lnet_res_lock(cpt);

 
 

-        me = lnet_handle2me(&meh);
-        if (me == NULL) {
-               lnet_res_unlock();
+       me = lnet_handle2me(&meh);
+       if (me == NULL) {
+               lnet_res_unlock(cpt);

                 return -ENOENT;
         }
 
                 return -ENOENT;
         }
 


@@ -235,13 +249,14 @@ LNetMEUnlink(lnet_handle_me_t meh)
             md->md_refcount == 0) {
                 lnet_build_unlink_event(md, &ev);
                lnet_eq_enqueue_event(md->md_eq, &ev);
             md->md_refcount == 0) {
                 lnet_build_unlink_event(md, &ev);
                lnet_eq_enqueue_event(md->md_eq, &ev);

-        }
+       }

 
 

-        lnet_me_unlink(me);
+       lnet_me_unlink(me);

 
 

-       lnet_res_unlock();
-        return 0;
+       lnet_res_unlock(cpt);
+       return 0;

 }
 }

+EXPORT_SYMBOL(LNetMEUnlink);

 
 /* call with lnet_res_lock please */
 void
 
 /* call with lnet_res_lock please */
 void