X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lnet%2Flnet%2Flib-ptl.c;h=e8f7ef7308cf415f53fbc8010fedc6264008220a;hp=ca24586f8d9a53c0861a9b713841559d4741fccc;hb=546993d587c5fc380e9745eae98f863e02e68575;hpb=38fcdd3966da09517ca176b962230b7dae43514c

diff --git a/lnet/lnet/lib-ptl.c b/lnet/lnet/lib-ptl.c
index ca24586..e8f7ef7 100644
--- a/lnet/lnet/lib-ptl.c
+++ b/lnet/lnet/lib-ptl.c
@@ -21,7 +21,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2011, 2012, Whamcloud, Inc.
+ * Copyright (c) 2012, 2017, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -38,8 +38,13 @@
 
 #include <lnet/lib-lnet.h>
 
+/* NB: add /proc interfaces in upcoming patches */
+int portal_rotor = LNET_PTL_ROTOR_HASH_RT;
+module_param(portal_rotor, int, 0644);
+MODULE_PARM_DESC(portal_rotor, "redirect PUTs to different cpu-partitions");
+
 static int
-lnet_ptl_match_type(unsigned int index, lnet_process_id_t match_id,
+lnet_ptl_match_type(unsigned int index, struct lnet_process_id match_id,
 		    __u64 mbits, __u64 ignore_bits)
 {
 	struct lnet_portal	*ptl = the_lnet.ln_portals[index];
@@ -56,10 +61,10 @@ lnet_ptl_match_type(unsigned int index, lnet_process_id_t match_id,
 		goto match;
 
 	/* unset, new portal */
-	LNET_LOCK();
+	lnet_ptl_lock(ptl);
 	/* check again with lock */
 	if (unlikely(lnet_ptl_is_unique(ptl) || lnet_ptl_is_wildcard(ptl))) {
-		LNET_UNLOCK();
+		lnet_ptl_unlock(ptl);
 		goto match;
 	}
 
@@ -69,7 +74,7 @@ lnet_ptl_match_type(unsigned int index, lnet_process_id_t match_id,
 	else
 		lnet_ptl_setopt(ptl, LNET_PTL_MATCH_WILDCARD);
 
-	LNET_UNLOCK();
+	lnet_ptl_unlock(ptl);
 
 	return 1;
 
@@ -80,36 +85,85 @@ lnet_ptl_match_type(unsigned int index, lnet_process_id_t match_id,
 	return 1;
 }
 
+static void
+lnet_ptl_enable_mt(struct lnet_portal *ptl, int cpt)
+{
+	struct lnet_match_table	*mtable = ptl->ptl_mtables[cpt];
+	int			i;
+
+	/* with hold of both lnet_res_lock(cpt) and lnet_ptl_lock */
+	LASSERT(lnet_ptl_is_wildcard(ptl));
+
+	mtable->mt_enabled = 1;
+
+	ptl->ptl_mt_maps[ptl->ptl_mt_nmaps] = cpt;
+	for (i = ptl->ptl_mt_nmaps - 1; i >= 0; i--) {
+		LASSERT(ptl->ptl_mt_maps[i] != cpt);
+		if (ptl->ptl_mt_maps[i] < cpt)
+			break;
+
+		/* swap to order */
+		ptl->ptl_mt_maps[i + 1] = ptl->ptl_mt_maps[i];
+		ptl->ptl_mt_maps[i] = cpt;
+	}
+
+	ptl->ptl_mt_nmaps++;
+}
+
+static void
+lnet_ptl_disable_mt(struct lnet_portal *ptl, int cpt)
+{
+	struct lnet_match_table	*mtable = ptl->ptl_mtables[cpt];
+	int			i;
+
+	/* with hold of both lnet_res_lock(cpt) and lnet_ptl_lock */
+	LASSERT(lnet_ptl_is_wildcard(ptl));
+
+	if (LNET_CPT_NUMBER == 1)
+		return; /* never disable the only match-table */
+
+	mtable->mt_enabled = 0;
+
+	LASSERT(ptl->ptl_mt_nmaps > 0 &&
+		ptl->ptl_mt_nmaps <= LNET_CPT_NUMBER);
+
+	/* remove it from mt_maps */
+	ptl->ptl_mt_nmaps--;
+	for (i = 0; i < ptl->ptl_mt_nmaps; i++) {
+		if (ptl->ptl_mt_maps[i] >= cpt) /* overwrite it */
+			ptl->ptl_mt_maps[i] = ptl->ptl_mt_maps[i + 1];
+	}
+}
+
 static int
-lnet_try_match_md(int index, int op_mask, lnet_process_id_t src,
-		  unsigned int rlength, unsigned int roffset,
-		  __u64 match_bits, lnet_libmd_t *md, lnet_msg_t *msg)
+lnet_try_match_md(struct lnet_libmd *md,
+		  struct lnet_match_info *info, struct lnet_msg *msg)
 {
-	/* ALWAYS called holding the LNET_LOCK, and can't LNET_UNLOCK;
+	/* ALWAYS called holding the lnet_res_lock, and can't lnet_res_unlock;
 	 * lnet_match_blocked_msg() relies on this to avoid races */
 	unsigned int	offset;
 	unsigned int	mlength;
-	lnet_me_t	*me = md->md_me;
-
-	/* mismatched MD op */
-	if ((md->md_options & op_mask) == 0)
-		return LNET_MATCHMD_NONE;
+	struct lnet_me	*me = md->md_me;
 
 	/* MD exhausted */
 	if (lnet_md_exhausted(md))
+		return LNET_MATCHMD_NONE | LNET_MATCHMD_EXHAUSTED;
+
+	/* mismatched MD op */
+	if ((md->md_options & info->mi_opc) == 0)
 		return LNET_MATCHMD_NONE;
 
 	/* mismatched ME nid/pid? */
 	if (me->me_match_id.nid != LNET_NID_ANY &&
-	    me->me_match_id.nid != src.nid)
+	    me->me_match_id.nid != info->mi_id.nid)
 		return LNET_MATCHMD_NONE;
 
 	if (me->me_match_id.pid != LNET_PID_ANY &&
-	    me->me_match_id.pid != src.pid)
+	    me->me_match_id.pid != info->mi_id.pid)
 		return LNET_MATCHMD_NONE;
 
 	/* mismatched ME matchbits? */
-	if (((me->me_match_bits ^ match_bits) & ~me->me_ignore_bits) != 0)
+	if (((me->me_match_bits ^ info->mi_mbits) & ~me->me_ignore_bits) != 0)
 		return LNET_MATCHMD_NONE;
 
 	/* Hurrah! This _is_ a match; check it out... */
@@ -117,7 +171,7 @@ lnet_try_match_md(int index, int op_mask, lnet_process_id_t src,
 	if ((md->md_options & LNET_MD_MANAGE_REMOTE) == 0)
 		offset = md->md_offset;
 	else
-		offset = roffset;
+		offset = info->mi_roffset;
 
 	if ((md->md_options & LNET_MD_MAX_SIZE) != 0) {
 		mlength = md->md_max_size;
@@ -126,202 +180,464 @@ lnet_try_match_md(int index, int op_mask, lnet_process_id_t src,
 		mlength = md->md_length - offset;
 	}
 
-	if (rlength <= mlength) {        /* fits in allowed space */
-		mlength = rlength;
+	if (info->mi_rlength <= mlength) {	/* fits in allowed space */
+		mlength = info->mi_rlength;
 	} else if ((md->md_options & LNET_MD_TRUNCATE) == 0) {
 		/* this packet _really_ is too big */
-		CERROR("Matching packet from %s, match "LPU64
+		CERROR("Matching packet from %s, match %llu"
 		       " length %d too big: %d left, %d allowed\n",
-		       libcfs_id2str(src), match_bits, rlength,
-		       md->md_length - offset, mlength);
+		       libcfs_id2str(info->mi_id), info->mi_mbits,
+		       info->mi_rlength, md->md_length - offset, mlength);
 
 		return LNET_MATCHMD_DROP;
 	}
 
 	/* Commit to this ME/MD */
 	CDEBUG(D_NET, "Incoming %s index %x from %s of "
-	       "length %d/%d into md "LPX64" [%d] + %d\n",
-	       (op_mask == LNET_MD_OP_PUT) ? "put" : "get",
-	       index, libcfs_id2str(src), mlength, rlength,
-	       md->md_lh.lh_cookie, md->md_niov, offset);
+	       "length %d/%d into md %#llx [%d] + %d\n",
+	       (info->mi_opc == LNET_MD_OP_PUT) ? "put" : "get",
+	       info->mi_portal, libcfs_id2str(info->mi_id), mlength,
+	       info->mi_rlength, md->md_lh.lh_cookie, md->md_niov, offset);
 
 	lnet_msg_attach_md(msg, md, offset, mlength);
 	md->md_offset = offset + mlength;
 
+	if (!lnet_md_exhausted(md))
+		return LNET_MATCHMD_OK;
+
 	/* Auto-unlink NOW, so the ME gets unlinked if required.
 	 * We bumped md->md_refcount above so the MD just gets flagged
 	 * for unlink when it is finalized. */
-	if ((md->md_flags & LNET_MD_FLAG_AUTO_UNLINK) != 0 &&
-	    lnet_md_exhausted(md)) {
+	if ((md->md_flags & LNET_MD_FLAG_AUTO_UNLINK) != 0)
 		lnet_md_unlink(md);
-	}
 
-	return LNET_MATCHMD_OK;
+	return LNET_MATCHMD_OK | LNET_MATCHMD_EXHAUSTED;
+}
+
+static struct lnet_match_table *
+lnet_match2mt(struct lnet_portal *ptl, struct lnet_process_id id, __u64 mbits)
+{
+	if (LNET_CPT_NUMBER == 1)
+		return ptl->ptl_mtables[0]; /* the only one */
+
+	/* if it's a unique portal, return match-table hashed by NID */
+	return lnet_ptl_is_unique(ptl) ?
+	       ptl->ptl_mtables[lnet_cpt_of_nid(id.nid, NULL)] : NULL;
 }
 
 struct lnet_match_table *
-lnet_mt_of_attach(unsigned int index, lnet_process_id_t id,
-		  __u64 mbits, __u64 ignore_bits, lnet_ins_pos_t pos)
+lnet_mt_of_attach(unsigned int index, struct lnet_process_id id,
+		  __u64 mbits, __u64 ignore_bits, enum lnet_ins_pos pos)
 {
-	struct lnet_portal *ptl;
+	struct lnet_portal	*ptl;
+	struct lnet_match_table	*mtable;
 
+	/* NB: called w/o lock */
 	LASSERT(index < the_lnet.ln_nportals);
 
 	if (!lnet_ptl_match_type(index, id, mbits, ignore_bits))
 		return NULL;
 
 	ptl = the_lnet.ln_portals[index];
-	/* NB: Now we only have one match-table for each portal,
-	 * and will have match-table per CPT in upcoming changes,
-	 * ME will be scattered to different match-tables based
-	 * on attaching information */
-	return ptl->ptl_mtable;
+
+	mtable = lnet_match2mt(ptl, id, mbits);
+	if (mtable != NULL) /* unique portal or only one match-table */
+		return mtable;
+
+	/* it's a wildcard portal */
+	switch (pos) {
+	default:
+		return NULL;
+	case LNET_INS_BEFORE:
+	case LNET_INS_AFTER:
+		/* posted by no affinity thread, always hash to specific
+		 * match-table to avoid buffer stealing which is heavy */
+		return ptl->ptl_mtables[ptl->ptl_index % LNET_CPT_NUMBER];
+	case LNET_INS_LOCAL:
+		/* posted by cpu-affinity thread */
+		return ptl->ptl_mtables[lnet_cpt_current()];
+	}
 }
 
-struct lnet_match_table *
-lnet_mt_of_match(unsigned int index, lnet_process_id_t id, __u64 mbits)
+static struct lnet_match_table *
+lnet_mt_of_match(struct lnet_match_info *info, struct lnet_msg *msg)
 {
-	struct lnet_portal *ptl;
+	struct lnet_match_table	*mtable;
+	struct lnet_portal	*ptl;
+	unsigned int		nmaps;
+	unsigned int		rotor;
+	unsigned int		cpt;
+	bool			routed;
+
+	/* NB: called w/o lock */
+	LASSERT(info->mi_portal < the_lnet.ln_nportals);
+	ptl = the_lnet.ln_portals[info->mi_portal];
+
+	LASSERT(lnet_ptl_is_wildcard(ptl) || lnet_ptl_is_unique(ptl));
+
+	mtable = lnet_match2mt(ptl, info->mi_id, info->mi_mbits);
+	if (mtable != NULL)
+		return mtable;
+
+	/* it's a wildcard portal */
+	routed = LNET_NIDNET(msg->msg_hdr.src_nid) !=
+		 LNET_NIDNET(msg->msg_hdr.dest_nid);
+
+	if (portal_rotor == LNET_PTL_ROTOR_OFF ||
+	    (portal_rotor != LNET_PTL_ROTOR_ON && !routed)) {
+		cpt = lnet_cpt_current();
+		if (ptl->ptl_mtables[cpt]->mt_enabled)
+			return ptl->ptl_mtables[cpt];
+	}
 
-	LASSERT(index < the_lnet.ln_nportals);
+	rotor = ptl->ptl_rotor++; /* get round-robin factor */
+	if (portal_rotor == LNET_PTL_ROTOR_HASH_RT && routed)
+		cpt = info->mi_cpt;
+	else
+		cpt = rotor % LNET_CPT_NUMBER;
+
+	if (!ptl->ptl_mtables[cpt]->mt_enabled) {
+		/* is there any active entry for this portal? */
+		nmaps = ptl->ptl_mt_nmaps;
+		/* map to an active mtable to avoid heavy "stealing" */
+		if (nmaps != 0) {
+			/* NB: there is possibility that ptl_mt_maps is being
+			 * changed because we are not under protection of
+			 * lnet_ptl_lock, but it shouldn't hurt anything */
+			cpt = ptl->ptl_mt_maps[rotor % nmaps];
+		}
+	}
 
-	ptl = the_lnet.ln_portals[index];
-	if (!lnet_ptl_is_unique(ptl) &&
-	    !lnet_ptl_is_wildcard(ptl) && !lnet_ptl_is_lazy(ptl))
-		return NULL;
+	return ptl->ptl_mtables[cpt];
+}
+
+static int
+lnet_mt_test_exhausted(struct lnet_match_table *mtable, int pos)
+{
+	__u64	*bmap;
+	int	i;
+
+	if (!lnet_ptl_is_wildcard(the_lnet.ln_portals[mtable->mt_portal]))
+		return 0;
 
-	/* NB: Now we only have one match-table for each portal,
-	 * and will have match-table per CPT in upcoming changes,
-	 * request will be scattered to different match-tables based
-	 * on matching information */
-	return ptl->ptl_mtable;
+	if (pos < 0) { /* check all bits */
+		for (i = 0; i < LNET_MT_EXHAUSTED_BMAP; i++) {
+			if (mtable->mt_exhausted[i] != (__u64)(-1))
+				return 0;
+		}
+		return 1;
+	}
+
+	LASSERT(pos <= LNET_MT_HASH_IGNORE);
+	/* mtable::mt_mhash[pos] is marked as exhausted or not */
+	bmap = &mtable->mt_exhausted[pos >> LNET_MT_BITS_U64];
+	pos &= (1 << LNET_MT_BITS_U64) - 1;
+
+	return ((*bmap) & (1ULL << pos)) != 0;
+}
+
+static void
+lnet_mt_set_exhausted(struct lnet_match_table *mtable, int pos, int exhausted)
+{
+	__u64	*bmap;
+
+	LASSERT(lnet_ptl_is_wildcard(the_lnet.ln_portals[mtable->mt_portal]));
+	LASSERT(pos <= LNET_MT_HASH_IGNORE);
+
+	/* set mtable::mt_mhash[pos] as exhausted/non-exhausted */
+	bmap = &mtable->mt_exhausted[pos >> LNET_MT_BITS_U64];
+	pos &= (1 << LNET_MT_BITS_U64) - 1;
+
+	if (!exhausted)
+		*bmap &= ~(1ULL << pos);
+	else
+		*bmap |= 1ULL << pos;
 }
 
-cfs_list_t *
+struct list_head *
 lnet_mt_match_head(struct lnet_match_table *mtable,
-		   lnet_process_id_t id, __u64 mbits)
+		   struct lnet_process_id id, __u64 mbits)
 {
 	struct lnet_portal *ptl = the_lnet.ln_portals[mtable->mt_portal];
 
 	if (lnet_ptl_is_wildcard(ptl)) {
-		return &mtable->mt_mlist;
-
-	} else if (lnet_ptl_is_unique(ptl)) {
+		return &mtable->mt_mhash[mbits & LNET_MT_HASH_MASK];
+	} else {
 		unsigned long hash = mbits + id.nid + id.pid;
 
-		hash = cfs_hash_long(hash, LNET_MT_HASH_BITS);
-		return &mtable->mt_mhash[hash];
+		LASSERT(lnet_ptl_is_unique(ptl));
+		hash = hash_long(hash, LNET_MT_HASH_BITS);
+		return &mtable->mt_mhash[hash & LNET_MT_HASH_MASK];
 	}
-
-	return NULL;
 }
 
 int
 lnet_mt_match_md(struct lnet_match_table *mtable,
-		 int op_mask, lnet_process_id_t src,
-		 unsigned int rlength, unsigned int roffset,
-		 __u64 match_bits, lnet_msg_t *msg)
+		 struct lnet_match_info *info, struct lnet_msg *msg)
 {
-	cfs_list_t		*head;
-	lnet_me_t		*me;
-	lnet_me_t		*tmp;
+	struct list_head	*head;
+	struct lnet_me		*me;
+	struct lnet_me		*tmp;
+	int			exhausted = 0;
 	int			rc;
 
-	head = lnet_mt_match_head(mtable, src, match_bits);
-	if (head == NULL) /* nobody posted anything on this portal */
-		goto out;
+	/* any ME with ignore bits? */
+	if (!list_empty(&mtable->mt_mhash[LNET_MT_HASH_IGNORE]))
+		head = &mtable->mt_mhash[LNET_MT_HASH_IGNORE];
+	else
+		head = lnet_mt_match_head(mtable, info->mi_id, info->mi_mbits);
+ again:
+	/* NB: only wildcard portal needs to return LNET_MATCHMD_EXHAUSTED */
+	if (lnet_ptl_is_wildcard(the_lnet.ln_portals[mtable->mt_portal]))
+		exhausted = LNET_MATCHMD_EXHAUSTED;
 
-	cfs_list_for_each_entry_safe(me, tmp, head, me_list) {
+	list_for_each_entry_safe(me, tmp, head, me_list) {
 		/* ME attached but MD not attached yet */
 		if (me->me_md == NULL)
 			continue;
 
 		LASSERT(me == me->me_md->md_me);
 
-		rc = lnet_try_match_md(mtable->mt_portal,
-				       op_mask, src, rlength, roffset,
-				       match_bits, me->me_md, msg);
-		switch (rc) {
-		default:
-			LBUG();
+		rc = lnet_try_match_md(me->me_md, info, msg);
+		if ((rc & LNET_MATCHMD_EXHAUSTED) == 0)
+			exhausted = 0; /* mlist is not empty */
 
-		case LNET_MATCHMD_NONE:
-			continue;
+		if ((rc & LNET_MATCHMD_FINISH) != 0) {
+			/* don't return EXHAUSTED bit because we don't know
+			 * whether the mlist is empty or not */
+			return rc & ~LNET_MATCHMD_EXHAUSTED;
+		}
+	}
+
+	if (exhausted == LNET_MATCHMD_EXHAUSTED) { /* @head is exhausted */
+		lnet_mt_set_exhausted(mtable, head - mtable->mt_mhash, 1);
+		if (!lnet_mt_test_exhausted(mtable, -1))
+			exhausted = 0;
+	}
+
+	if (exhausted == 0 && head == &mtable->mt_mhash[LNET_MT_HASH_IGNORE]) {
+		head = lnet_mt_match_head(mtable, info->mi_id, info->mi_mbits);
+		goto again; /* re-check MEs w/o ignore-bits */
+	}
 
-		case LNET_MATCHMD_OK:
-			return LNET_MATCHMD_OK;
+	if (info->mi_opc == LNET_MD_OP_GET ||
+	    !lnet_ptl_is_lazy(the_lnet.ln_portals[info->mi_portal]))
+		return LNET_MATCHMD_DROP | exhausted;
 
-		case LNET_MATCHMD_DROP:
-			return LNET_MATCHMD_DROP;
+	return LNET_MATCHMD_NONE | exhausted;
+}
+
+static int
+lnet_ptl_match_early(struct lnet_portal *ptl, struct lnet_msg *msg)
+{
+	int	rc;
+
+	/* message arrived before any buffer posting on this portal,
+	 * simply delay or drop this message */
+	if (likely(lnet_ptl_is_wildcard(ptl) || lnet_ptl_is_unique(ptl)))
+		return 0;
+
+	lnet_ptl_lock(ptl);
+	/* check it again with hold of lock */
+	if (lnet_ptl_is_wildcard(ptl) || lnet_ptl_is_unique(ptl)) {
+		lnet_ptl_unlock(ptl);
+		return 0;
+	}
+
+	if (lnet_ptl_is_lazy(ptl)) {
+		if (msg->msg_rx_ready_delay) {
+			msg->msg_rx_delayed = 1;
+			list_add_tail(&msg->msg_list,
+				      &ptl->ptl_msg_delayed);
 		}
-		/* not reached */
+		rc = LNET_MATCHMD_NONE;
+	} else {
+		rc = LNET_MATCHMD_DROP;
 	}
 
- out:
-	if (op_mask == LNET_MD_OP_GET ||
-	    !lnet_ptl_is_lazy(the_lnet.ln_portals[mtable->mt_portal]))
-		return LNET_MATCHMD_DROP;
+	lnet_ptl_unlock(ptl);
+	return rc;
+}
+
+static int
+lnet_ptl_match_delay(struct lnet_portal *ptl,
+		     struct lnet_match_info *info, struct lnet_msg *msg)
+{
+	int	first = ptl->ptl_mt_maps[0]; /* read w/o lock */
+	int	rc = 0;
+	int	i;
+
+	/*
+	 * Steal buffer from other CPTs, and delay msg if nothing to
+	 * steal.  This function is more expensive than a regular
+	 * match, but we don't expect it can happen a lot. The return
+	 * code contains one of LNET_MATCHMD_OK, LNET_MATCHMD_DROP, or
+	 * LNET_MATCHMD_NONE.
+	 */
+	LASSERT(lnet_ptl_is_wildcard(ptl));
+
+	for (i = 0; i < LNET_CPT_NUMBER; i++) {
+		struct lnet_match_table *mtable;
+		int			cpt;
+
+		cpt = (first + i) % LNET_CPT_NUMBER;
+		mtable = ptl->ptl_mtables[cpt];
+		if (i != 0 && i != LNET_CPT_NUMBER - 1 && !mtable->mt_enabled)
+			continue;
+
+		lnet_res_lock(cpt);
+		lnet_ptl_lock(ptl);
+
+		if (i == 0) {
+			/* The first try, add to stealing list. */
+			list_add_tail(&msg->msg_list,
+				      &ptl->ptl_msg_stealing);
+		}
 
-	return LNET_MATCHMD_NONE;
+		if (!list_empty(&msg->msg_list)) {
+			/* On stealing list. */
+			rc = lnet_mt_match_md(mtable, info, msg);
+
+			if ((rc & LNET_MATCHMD_EXHAUSTED) != 0 &&
+			    mtable->mt_enabled)
+				lnet_ptl_disable_mt(ptl, cpt);
+
+			if ((rc & LNET_MATCHMD_FINISH) != 0) {
+				/* Match found, remove from stealing list. */
+				list_del_init(&msg->msg_list);
+			} else if (i == LNET_CPT_NUMBER - 1 || /* (1) */
+				   ptl->ptl_mt_nmaps == 0 ||   /* (2) */
+				   (ptl->ptl_mt_nmaps == 1 &&  /* (3) */
+				    ptl->ptl_mt_maps[0] == cpt)) {
+				/*
+				 * No match found, and this is either
+				 * (1) the last cpt to check, or
+				 * (2) there is no active cpt, or
+				 * (3) this is the only active cpt.
+				 * There is nothing to steal: delay or
+				 * drop the message.
+				 */
+				list_del_init(&msg->msg_list);
+
+				if (lnet_ptl_is_lazy(ptl)) {
+					msg->msg_rx_delayed = 1;
+					list_add_tail(&msg->msg_list,
+						      &ptl->ptl_msg_delayed);
+					rc = LNET_MATCHMD_NONE;
+				} else {
+					rc = LNET_MATCHMD_DROP;
+				}
+			} else {
+				/* Do another iteration. */
+				rc = 0;
+			}
+		} else {
+			/*
+			 * No longer on stealing list: another thread
+			 * matched the message in lnet_ptl_attach_md().
+			 * We are now expected to handle the message.
+			 */
+			rc = msg->msg_md == NULL ?
+				LNET_MATCHMD_DROP : LNET_MATCHMD_OK;
+		}
+
+		lnet_ptl_unlock(ptl);
+		lnet_res_unlock(cpt);
+
+		/*
+		 * Note that test (1) above ensures that we always
+		 * exit the loop through this break statement.
+		 *
+		 * LNET_MATCHMD_NONE means msg was added to the
+		 * delayed queue, and we may no longer reference it
+		 * after lnet_ptl_unlock() and lnet_res_unlock().
+		 */
+		if (rc & (LNET_MATCHMD_FINISH | LNET_MATCHMD_NONE))
+			break;
+	}
+
+	return rc;
 }
 
 int
-lnet_ptl_match_md(unsigned int index, int op_mask, lnet_process_id_t src,
-		  unsigned int rlength, unsigned int roffset,
-		  __u64 match_bits, lnet_msg_t *msg)
+lnet_ptl_match_md(struct lnet_match_info *info, struct lnet_msg *msg)
 {
 	struct lnet_match_table	*mtable;
 	struct lnet_portal	*ptl;
 	int			rc;
 
 	CDEBUG(D_NET, "Request from %s of length %d into portal %d "
-	       "MB="LPX64"\n", libcfs_id2str(src), rlength, index, match_bits);
+	       "MB=%#llx\n", libcfs_id2str(info->mi_id),
+	       info->mi_rlength, info->mi_portal, info->mi_mbits);
 
-	if (index >= the_lnet.ln_nportals) {
+	if (info->mi_portal >= the_lnet.ln_nportals) {
 		CERROR("Invalid portal %d not in [0-%d]\n",
-		       index, the_lnet.ln_nportals);
+		       info->mi_portal, the_lnet.ln_nportals);
 		return LNET_MATCHMD_DROP;
 	}
 
-	mtable = lnet_mt_of_match(index, src, match_bits);
-	if (mtable == NULL) {
-		CDEBUG(D_NET, "Drop early message from %s of length %d into "
-			      "portal %d MB="LPX64"\n",
-			      libcfs_id2str(src), rlength, index, match_bits);
-		return LNET_MATCHMD_DROP;
-	}
+	ptl = the_lnet.ln_portals[info->mi_portal];
+	rc = lnet_ptl_match_early(ptl, msg);
+	if (rc != 0) /* matched or delayed early message */
+		return rc;
 
-	ptl = the_lnet.ln_portals[index];
-	LNET_LOCK();
+	mtable = lnet_mt_of_match(info, msg);
+	lnet_res_lock(mtable->mt_cpt);
 
-	if (the_lnet.ln_shutdown) {
-		rc =  LNET_MATCHMD_DROP;
-		goto out;
+	if (the_lnet.ln_state != LNET_STATE_RUNNING) {
+		rc = LNET_MATCHMD_DROP;
+		goto out1;
 	}
 
-	rc = lnet_mt_match_md(mtable, op_mask, src, rlength,
-			      roffset, match_bits, msg);
-	if (rc != LNET_MATCHMD_NONE) /* matched or dropping */
-		goto out;
+	rc = lnet_mt_match_md(mtable, info, msg);
+	if ((rc & LNET_MATCHMD_EXHAUSTED) != 0 && mtable->mt_enabled) {
+		lnet_ptl_lock(ptl);
+		lnet_ptl_disable_mt(ptl, mtable->mt_cpt);
+		lnet_ptl_unlock(ptl);
+	}
+
+	if ((rc & LNET_MATCHMD_FINISH) != 0)	/* matched or dropping */
+		goto out1;
 
 	if (!msg->msg_rx_ready_delay)
-		goto out;
+		goto out1;
 
+	LASSERT(lnet_ptl_is_lazy(ptl));
 	LASSERT(!msg->msg_rx_delayed);
-	msg->msg_rx_delayed = 1;
-	cfs_list_add_tail(&msg->msg_list, &ptl->ptl_msgq);
-
-	CDEBUG(D_NET,
-	       "Delaying %s from %s portal %d MB "LPX64" offset %d len %d\n",
-	       op_mask == LNET_MD_OP_PUT ? "PUT" : "GET",
-	       libcfs_id2str(src), index, match_bits, roffset, rlength);
- out:
-	LNET_UNLOCK();
-	return rc;
+
+	/* NB: we don't expect "delay" can happen a lot */
+	if (lnet_ptl_is_unique(ptl) || LNET_CPT_NUMBER == 1) {
+		lnet_ptl_lock(ptl);
+
+		msg->msg_rx_delayed = 1;
+		list_add_tail(&msg->msg_list, &ptl->ptl_msg_delayed);
+
+		lnet_ptl_unlock(ptl);
+		lnet_res_unlock(mtable->mt_cpt);
+		rc = LNET_MATCHMD_NONE;
+	} else	{
+		lnet_res_unlock(mtable->mt_cpt);
+		rc = lnet_ptl_match_delay(ptl, info, msg);
+	}
+
+	/* LNET_MATCHMD_NONE means msg was added to the delay queue */
+	if (rc & LNET_MATCHMD_NONE) {
+		CDEBUG(D_NET,
+		       "Delaying %s from %s ptl %d MB %#llx off %d len %d\n",
+		       info->mi_opc == LNET_MD_OP_PUT ? "PUT" : "GET",
+		       libcfs_id2str(info->mi_id), info->mi_portal,
+		       info->mi_mbits, info->mi_roffset, info->mi_rlength);
+	}
+	goto out0;
+ out1:
+	lnet_res_unlock(mtable->mt_cpt);
+ out0:
+	/* EXHAUSTED bit is only meaningful for internal functions */
+	return rc & ~LNET_MATCHMD_EXHAUSTED;
 }
 
 void
-lnet_ptl_detach_md(lnet_me_t *me, lnet_libmd_t *md)
+lnet_ptl_detach_md(struct lnet_me *me, struct lnet_libmd *md)
 {
 	LASSERT(me->me_md == md && md->md_me == me);
 
@@ -329,145 +645,179 @@ lnet_ptl_detach_md(lnet_me_t *me, lnet_libmd_t *md)
 	md->md_me = NULL;
 }
 
-/* called with LNET_LOCK held */
+/* called with lnet_res_lock held */
 void
-lnet_ptl_attach_md(lnet_me_t *me, lnet_libmd_t *md,
-		   cfs_list_t *matches, cfs_list_t *drops)
+lnet_ptl_attach_md(struct lnet_me *me, struct lnet_libmd *md,
+		   struct list_head *matches, struct list_head *drops)
 {
-	struct lnet_portal	*ptl = the_lnet.ln_portals[me->me_portal];
-	lnet_msg_t		*tmp;
-	lnet_msg_t		*msg;
+	struct lnet_portal *ptl = the_lnet.ln_portals[me->me_portal];
+	struct lnet_match_table	*mtable;
+	struct list_head *head;
+	struct lnet_msg	*tmp;
+	struct lnet_msg	*msg;
+	int exhausted = 0;
+	int cpt;
 
 	LASSERT(md->md_refcount == 0); /* a brand new MD */
 
 	me->me_md = md;
 	md->md_me = me;
 
-	cfs_list_for_each_entry_safe(msg, tmp, &ptl->ptl_msgq, msg_list) {
-		int               rc;
-		int               index;
-		lnet_hdr_t       *hdr;
-		lnet_process_id_t src;
-
-		LASSERT(msg->msg_rx_delayed);
+	cpt = lnet_cpt_of_cookie(md->md_lh.lh_cookie);
+	mtable = ptl->ptl_mtables[cpt];
 
-		hdr   = &msg->msg_hdr;
-		index = hdr->msg.put.ptl_index;
+	if (list_empty(&ptl->ptl_msg_stealing) &&
+	    list_empty(&ptl->ptl_msg_delayed) &&
+	    !lnet_mt_test_exhausted(mtable, me->me_pos))
+		return;
 
-		src.nid = hdr->src_nid;
-		src.pid = hdr->src_pid;
+	lnet_ptl_lock(ptl);
+	head = &ptl->ptl_msg_stealing;
+ again:
+	list_for_each_entry_safe(msg, tmp, head, msg_list) {
+		struct lnet_match_info	info;
+		struct lnet_hdr		*hdr;
+		int			rc;
 
-		rc = lnet_try_match_md(index, LNET_MD_OP_PUT, src,
-				       hdr->payload_length,
-				       hdr->msg.put.offset,
-				       hdr->msg.put.match_bits, md, msg);
+		LASSERT(msg->msg_rx_delayed || head == &ptl->ptl_msg_stealing);
 
-		if (rc == LNET_MATCHMD_NONE)
+		hdr   = &msg->msg_hdr;
+		/* Multi-Rail: Primary peer NID */
+		info.mi_id.nid	= msg->msg_initiator;
+		info.mi_id.pid	= hdr->src_pid;
+		info.mi_opc	= LNET_MD_OP_PUT;
+		info.mi_portal	= hdr->msg.put.ptl_index;
+		info.mi_rlength	= hdr->payload_length;
+		info.mi_roffset	= hdr->msg.put.offset;
+		info.mi_mbits	= hdr->msg.put.match_bits;
+
+		rc = lnet_try_match_md(md, &info, msg);
+
+		exhausted = (rc & LNET_MATCHMD_EXHAUSTED) != 0;
+		if ((rc & LNET_MATCHMD_NONE) != 0) {
+			if (exhausted)
+				break;
 			continue;
+		}
 
 		/* Hurrah! This _is_ a match */
-		cfs_list_del(&msg->msg_list);
+		LASSERT((rc & LNET_MATCHMD_FINISH) != 0);
+		list_del_init(&msg->msg_list);
 
-		if (rc == LNET_MATCHMD_OK) {
-			cfs_list_add_tail(&msg->msg_list, matches);
+		if (head == &ptl->ptl_msg_stealing) {
+			if (exhausted)
+				break;
+			/* stealing thread will handle the message */
+			continue;
+		}
+
+		if ((rc & LNET_MATCHMD_OK) != 0) {
+			list_add_tail(&msg->msg_list, matches);
 
 			CDEBUG(D_NET, "Resuming delayed PUT from %s portal %d "
-			       "match "LPU64" offset %d length %d.\n",
-			       libcfs_id2str(src),
-			       hdr->msg.put.ptl_index,
-			       hdr->msg.put.match_bits,
-			       hdr->msg.put.offset,
-			       hdr->payload_length);
+			       "match %llu offset %d length %d.\n",
+			       libcfs_id2str(info.mi_id),
+			       info.mi_portal, info.mi_mbits,
+			       info.mi_roffset, info.mi_rlength);
 		} else {
-			LASSERT(rc == LNET_MATCHMD_DROP);
-
-			cfs_list_add_tail(&msg->msg_list, drops);
+			list_add_tail(&msg->msg_list, drops);
 		}
 
-		if (lnet_md_exhausted(md))
+		if (exhausted)
 			break;
 	}
+
+	if (!exhausted && head == &ptl->ptl_msg_stealing) {
+		head = &ptl->ptl_msg_delayed;
+		goto again;
+	}
+
+	if (lnet_ptl_is_wildcard(ptl) && !exhausted) {
+		lnet_mt_set_exhausted(mtable, me->me_pos, 0);
+		if (!mtable->mt_enabled)
+			lnet_ptl_enable_mt(ptl, cpt);
+	}
+
+	lnet_ptl_unlock(ptl);
 }
 
-void
+static void
 lnet_ptl_cleanup(struct lnet_portal *ptl)
 {
 	struct lnet_match_table	*mtable;
+	int			i;
 
-	LASSERT(cfs_list_empty(&ptl->ptl_msgq));
-
-	if (ptl->ptl_mtable == NULL) /* uninitialized portal */
+	if (ptl->ptl_mtables == NULL) /* uninitialized portal */
 		return;
 
-	do { /* iterate over match-tables when we have percpt match-table */
-		cfs_list_t	*mhash;
-		lnet_me_t	*me;
-		int		j;
-
-		mtable = ptl->ptl_mtable;
+	LASSERT(list_empty(&ptl->ptl_msg_delayed));
+	LASSERT(list_empty(&ptl->ptl_msg_stealing));
+	cfs_percpt_for_each(mtable, i, ptl->ptl_mtables) {
+		struct list_head *mhash;
+		struct lnet_me	 *me;
+		int		  j;
 
 		if (mtable->mt_mhash == NULL) /* uninitialized match-table */
 			continue;
 
 		mhash = mtable->mt_mhash;
 		/* cleanup ME */
-		while (!cfs_list_empty(&mtable->mt_mlist)) {
-			me = cfs_list_entry(mtable->mt_mlist.next,
-					    lnet_me_t, me_list);
-			CERROR("Active wildcard ME %p on exit\n", me);
-			cfs_list_del(&me->me_list);
-			lnet_me_free(me);
-		}
-
-		for (j = 0; j < LNET_MT_HASH_SIZE; j++) {
-			while (!cfs_list_empty(&mhash[j])) {
-				me = cfs_list_entry(mhash[j].next,
-						    lnet_me_t, me_list);
-				CERROR("Active unique ME %p on exit\n", me);
-				cfs_list_del(&me->me_list);
+		for (j = 0; j < LNET_MT_HASH_SIZE + 1; j++) {
+			while (!list_empty(&mhash[j])) {
+				me = list_entry(mhash[j].next,
+						struct lnet_me, me_list);
+				CERROR("Active ME %p on exit\n", me);
+				list_del(&me->me_list);
 				lnet_me_free(me);
 			}
 		}
+		/* the extra entry is for MEs with ignore bits */
+		LIBCFS_FREE(mhash, sizeof(*mhash) * (LNET_MT_HASH_SIZE + 1));
+	}
 
-		LIBCFS_FREE(mhash, sizeof(*mhash) * LNET_MT_HASH_SIZE);
-	} while (0);
-
-	LIBCFS_FREE(ptl->ptl_mtable, sizeof(*mtable));
-	ptl->ptl_mtable = NULL;
+	cfs_percpt_free(ptl->ptl_mtables);
+	ptl->ptl_mtables = NULL;
 }
 
-int
+static int
 lnet_ptl_setup(struct lnet_portal *ptl, int index)
 {
 	struct lnet_match_table	*mtable;
-	cfs_list_t		*mhash;
+	struct list_head	*mhash;
+	int			i;
 	int			j;
 
-	ptl->ptl_index = index;
-	CFS_INIT_LIST_HEAD(&ptl->ptl_msgq);
-
-	LIBCFS_ALLOC(mtable, sizeof(*mtable));
-	if (mtable == NULL) {
+	ptl->ptl_mtables = cfs_percpt_alloc(lnet_cpt_table(),
+					    sizeof(struct lnet_match_table));
+	if (ptl->ptl_mtables == NULL) {
 		CERROR("Failed to create match table for portal %d\n", index);
 		return -ENOMEM;
 	}
 
-	ptl->ptl_mtable = mtable;
-	do { /* iterate over match-tables when we have percpt match-table */
-		LIBCFS_ALLOC(mhash, sizeof(*mhash) * LNET_MT_HASH_SIZE);
+	ptl->ptl_index = index;
+	INIT_LIST_HEAD(&ptl->ptl_msg_delayed);
+	INIT_LIST_HEAD(&ptl->ptl_msg_stealing);
+	spin_lock_init(&ptl->ptl_lock);
+	cfs_percpt_for_each(mtable, i, ptl->ptl_mtables) {
+		/* the extra entry is for MEs with ignore bits */
+		LIBCFS_CPT_ALLOC(mhash, lnet_cpt_table(), i,
+				 sizeof(*mhash) * (LNET_MT_HASH_SIZE + 1));
 		if (mhash == NULL) {
 			CERROR("Failed to create match hash for portal %d\n",
 			       index);
 			goto failed;
 		}
 
+		memset(&mtable->mt_exhausted[0], -1,
+		       sizeof(mtable->mt_exhausted[0]) *
+		       LNET_MT_EXHAUSTED_BMAP);
 		mtable->mt_mhash = mhash;
-		for (j = 0; j < LNET_MT_HASH_SIZE; j++)
-			CFS_INIT_LIST_HEAD(&mhash[j]);
+		for (j = 0; j < LNET_MT_HASH_SIZE + 1; j++)
+			INIT_LIST_HEAD(&mhash[j]);
 
-		CFS_INIT_LIST_HEAD(&mtable->mt_mlist);
 		mtable->mt_portal = index;
-	} while (0);
+		mtable->mt_cpt = i;
+	}
 
 	return 0;
  failed:
@@ -496,7 +846,7 @@ lnet_portals_create(void)
 	int	size;
 	int	i;
 
-	size = sizeof(struct lnet_portal);
+	size = offsetof(struct lnet_portal, ptl_mt_maps[LNET_CPT_NUMBER]);
 
 	the_lnet.ln_nportals = MAX_PORTALS;
 	the_lnet.ln_portals = cfs_array_alloc(the_lnet.ln_nportals, size);
@@ -540,7 +890,7 @@ lnet_portals_create(void)
  *
  * \param portal Index of the portal to enable the lazy attribute on.
  *
- * \retval 0       On success.
+ * \retval 0	   On success.
  * \retval -EINVAL If \a portal is not a valid index.
  */
 int
@@ -554,53 +904,80 @@ LNetSetLazyPortal(int portal)
 	CDEBUG(D_NET, "Setting portal %d lazy\n", portal);
 	ptl = the_lnet.ln_portals[portal];
 
-	LNET_LOCK();
+	lnet_res_lock(LNET_LOCK_EX);
+	lnet_ptl_lock(ptl);
+
 	lnet_ptl_setopt(ptl, LNET_PTL_LAZY);
-	LNET_UNLOCK();
+
+	lnet_ptl_unlock(ptl);
+	lnet_res_unlock(LNET_LOCK_EX);
 
 	return 0;
 }
+EXPORT_SYMBOL(LNetSetLazyPortal);
 
-/**
- * Turn off the lazy portal attribute. Delayed requests on the portal,
- * if any, will be all dropped when this function returns.
- *
- * \param portal Index of the portal to disable the lazy attribute on.
- *
- * \retval 0       On success.
- * \retval -EINVAL If \a portal is not a valid index.
- */
 int
-LNetClearLazyPortal(int portal)
+lnet_clear_lazy_portal(struct lnet_ni *ni, int portal, char *reason)
 {
 	struct lnet_portal	*ptl;
-	CFS_LIST_HEAD		(zombies);
+	LIST_HEAD(zombies);
 
 	if (portal < 0 || portal >= the_lnet.ln_nportals)
 		return -EINVAL;
 
 	ptl = the_lnet.ln_portals[portal];
 
-	LNET_LOCK();
+	lnet_res_lock(LNET_LOCK_EX);
+	lnet_ptl_lock(ptl);
 
 	if (!lnet_ptl_is_lazy(ptl)) {
-		LNET_UNLOCK();
+		lnet_ptl_unlock(ptl);
+		lnet_res_unlock(LNET_LOCK_EX);
 		return 0;
 	}
 
-	if (the_lnet.ln_shutdown)
-		CWARN("Active lazy portal %d on exit\n", portal);
-	else
-		CDEBUG(D_NET, "clearing portal %d lazy\n", portal);
+	if (ni != NULL) {
+		struct lnet_msg *msg, *tmp;
 
-	/* grab all the blocked messages atomically */
-	cfs_list_splice_init(&ptl->ptl_msgq, &zombies);
+		/* grab all messages which are on the NI passed in */
+		list_for_each_entry_safe(msg, tmp, &ptl->ptl_msg_delayed,
+					 msg_list) {
+			if (msg->msg_txni == ni || msg->msg_rxni == ni)
+				list_move(&msg->msg_list, &zombies);
+		}
+	} else {
+		if (the_lnet.ln_state != LNET_STATE_RUNNING)
+			CWARN("Active lazy portal %d on exit\n", portal);
+		else
+			CDEBUG(D_NET, "clearing portal %d lazy\n", portal);
+
+		/* grab all the blocked messages atomically */
+		list_splice_init(&ptl->ptl_msg_delayed, &zombies);
 
-	lnet_ptl_unsetopt(ptl, LNET_PTL_LAZY);
+		lnet_ptl_unsetopt(ptl, LNET_PTL_LAZY);
+	}
 
-	LNET_UNLOCK();
+	lnet_ptl_unlock(ptl);
+	lnet_res_unlock(LNET_LOCK_EX);
 
-	lnet_drop_delayed_msg_list(&zombies, "Clearing lazy portal attr");
+	lnet_drop_delayed_msg_list(&zombies, reason);
 
 	return 0;
 }
+
+/**
+ * Turn off the lazy portal attribute. Delayed requests on the portal,
+ * if any, will be all dropped when this function returns.
+ *
+ * \param portal Index of the portal to disable the lazy attribute on.
+ *
+ * \retval 0	   On success.
+ * \retval -EINVAL If \a portal is not a valid index.
+ */
+int
+LNetClearLazyPortal(int portal)
+{
+	return lnet_clear_lazy_portal(NULL, portal,
+				      "Clearing lazy portal attr");
+}
+EXPORT_SYMBOL(LNetClearLazyPortal);