drivers/staging/lustre/lnet/lnet/lib-ptl.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 021110-1307, USA
  20  *
  21  * GPL HEADER END
  22  */
  23 /*
  24  * Copyright (c) 2012, 2015, Intel Corporation.
  25  */
  26 /*
  27  * This file is part of Lustre, http://www.lustre.org/
  28  * Lustre is a trademark of Sun Microsystems, Inc.
  29  *
  30  * lnet/lnet/lib-ptl.c
  31  *
  32  * portal & match routines
  33  *
  34  * Author: liang@whamcloud.com
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_LNET
  38
  39 #include "../../include/linux/lnet/lib-lnet.h"
  40
  41 /* NB: add /proc interfaces in upcoming patches */
  42 int portal_rotor = LNET_PTL_ROTOR_HASH_RT;
  43 module_param(portal_rotor, int, 0644);
  44 MODULE_PARM_DESC(portal_rotor, "redirect PUTs to different cpu-partitions");
  45
  46 static int
  47 lnet_ptl_match_type(unsigned int index, lnet_process_id_t match_id,
  48                     __u64 mbits, __u64 ignore_bits)
  49 {
  50         struct lnet_portal *ptl = the_lnet.ln_portals[index];
  51         int unique;
  52
  53         unique = !ignore_bits &&
  54                  match_id.nid != LNET_NID_ANY &&
  55                  match_id.pid != LNET_PID_ANY;
  56
  57         LASSERT(!lnet_ptl_is_unique(ptl) || !lnet_ptl_is_wildcard(ptl));
  58
  59         /* prefer to check w/o any lock */
  60         if (likely(lnet_ptl_is_unique(ptl) || lnet_ptl_is_wildcard(ptl)))
  61                 goto match;
  62
  63         /* unset, new portal */
  64         lnet_ptl_lock(ptl);
  65         /* check again with lock */
  66         if (unlikely(lnet_ptl_is_unique(ptl) || lnet_ptl_is_wildcard(ptl))) {
  67                 lnet_ptl_unlock(ptl);
  68                 goto match;
  69         }
  70
  71         /* still not set */
  72         if (unique)
  73                 lnet_ptl_setopt(ptl, LNET_PTL_MATCH_UNIQUE);
  74         else
  75                 lnet_ptl_setopt(ptl, LNET_PTL_MATCH_WILDCARD);
  76
  77         lnet_ptl_unlock(ptl);
  78
  79         return 1;
  80
  81  match:
  82         if ((lnet_ptl_is_unique(ptl) && !unique) ||
  83             (lnet_ptl_is_wildcard(ptl) && unique))
  84                 return 0;
  85         return 1;
  86 }
  87
  88 static void
  89 lnet_ptl_enable_mt(struct lnet_portal *ptl, int cpt)
  90 {
  91         struct lnet_match_table *mtable = ptl->ptl_mtables[cpt];
  92         int i;
  93
  94         /* with hold of both lnet_res_lock(cpt) and lnet_ptl_lock */
  95         LASSERT(lnet_ptl_is_wildcard(ptl));
  96
  97         mtable->mt_enabled = 1;
  98
  99         ptl->ptl_mt_maps[ptl->ptl_mt_nmaps] = cpt;
 100         for (i = ptl->ptl_mt_nmaps - 1; i >= 0; i--) {
 101                 LASSERT(ptl->ptl_mt_maps[i] != cpt);
 102                 if (ptl->ptl_mt_maps[i] < cpt)
 103                         break;
 104
 105                 /* swap to order */
 106                 ptl->ptl_mt_maps[i + 1] = ptl->ptl_mt_maps[i];
 107                 ptl->ptl_mt_maps[i] = cpt;
 108         }
 109
 110         ptl->ptl_mt_nmaps++;
 111 }
 112
 113 static void
 114 lnet_ptl_disable_mt(struct lnet_portal *ptl, int cpt)
 115 {
 116         struct lnet_match_table *mtable = ptl->ptl_mtables[cpt];
 117         int i;
 118
 119         /* with hold of both lnet_res_lock(cpt) and lnet_ptl_lock */
 120         LASSERT(lnet_ptl_is_wildcard(ptl));
 121
 122         if (LNET_CPT_NUMBER == 1)
 123                 return; /* never disable the only match-table */
 124
 125         mtable->mt_enabled = 0;
 126
 127         LASSERT(ptl->ptl_mt_nmaps > 0 &&
 128                 ptl->ptl_mt_nmaps <= LNET_CPT_NUMBER);
 129
 130         /* remove it from mt_maps */
 131         ptl->ptl_mt_nmaps--;
 132         for (i = 0; i < ptl->ptl_mt_nmaps; i++) {
 133                 if (ptl->ptl_mt_maps[i] >= cpt) /* overwrite it */
 134                         ptl->ptl_mt_maps[i] = ptl->ptl_mt_maps[i + 1];
 135         }
 136 }
 137
 138 static int
 139 lnet_try_match_md(lnet_libmd_t *md,
 140                   struct lnet_match_info *info, struct lnet_msg *msg)
 141 {
 142         /*
 143          * ALWAYS called holding the lnet_res_lock, and can't lnet_res_unlock;
 144          * lnet_match_blocked_msg() relies on this to avoid races
 145          */
 146         unsigned int offset;
 147         unsigned int mlength;
 148         lnet_me_t *me = md->md_me;
 149
 150         /* MD exhausted */
 151         if (lnet_md_exhausted(md))
 152                 return LNET_MATCHMD_NONE | LNET_MATCHMD_EXHAUSTED;
 153
 154         /* mismatched MD op */
 155         if (!(md->md_options & info->mi_opc))
 156                 return LNET_MATCHMD_NONE;
 157
 158         /* mismatched ME nid/pid? */
 159         if (me->me_match_id.nid != LNET_NID_ANY &&
 160             me->me_match_id.nid != info->mi_id.nid)
 161                 return LNET_MATCHMD_NONE;
 162
 163         if (me->me_match_id.pid != LNET_PID_ANY &&
 164             me->me_match_id.pid != info->mi_id.pid)
 165                 return LNET_MATCHMD_NONE;
 166
 167         /* mismatched ME matchbits? */
 168         if ((me->me_match_bits ^ info->mi_mbits) & ~me->me_ignore_bits)
 169                 return LNET_MATCHMD_NONE;
 170
 171         /* Hurrah! This _is_ a match; check it out... */
 172
 173         if (!(md->md_options & LNET_MD_MANAGE_REMOTE))
 174                 offset = md->md_offset;
 175         else
 176                 offset = info->mi_roffset;
 177
 178         if (md->md_options & LNET_MD_MAX_SIZE) {
 179                 mlength = md->md_max_size;
 180                 LASSERT(md->md_offset + mlength <= md->md_length);
 181         } else {
 182                 mlength = md->md_length - offset;
 183         }
 184
 185         if (info->mi_rlength <= mlength) {      /* fits in allowed space */
 186                 mlength = info->mi_rlength;
 187         } else if (!(md->md_options & LNET_MD_TRUNCATE)) {
 188                 /* this packet _really_ is too big */
 189                 CERROR("Matching packet from %s, match %llu length %d too big: %d left, %d allowed\n",
 190                        libcfs_id2str(info->mi_id), info->mi_mbits,
 191                        info->mi_rlength, md->md_length - offset, mlength);
 192
 193                 return LNET_MATCHMD_DROP;
 194         }
 195
 196         /* Commit to this ME/MD */
 197         CDEBUG(D_NET, "Incoming %s index %x from %s of length %d/%d into md %#llx [%d] + %d\n",
 198                (info->mi_opc == LNET_MD_OP_PUT) ? "put" : "get",
 199                info->mi_portal, libcfs_id2str(info->mi_id), mlength,
 200                info->mi_rlength, md->md_lh.lh_cookie, md->md_niov, offset);
 201
 202         lnet_msg_attach_md(msg, md, offset, mlength);
 203         md->md_offset = offset + mlength;
 204
 205         if (!lnet_md_exhausted(md))
 206                 return LNET_MATCHMD_OK;
 207
 208         /*
 209          * Auto-unlink NOW, so the ME gets unlinked if required.
 210          * We bumped md->md_refcount above so the MD just gets flagged
 211          * for unlink when it is finalized.
 212          */
 213         if (md->md_flags & LNET_MD_FLAG_AUTO_UNLINK)
 214                 lnet_md_unlink(md);
 215
 216         return LNET_MATCHMD_OK | LNET_MATCHMD_EXHAUSTED;
 217 }
 218
 219 static struct lnet_match_table *
 220 lnet_match2mt(struct lnet_portal *ptl, lnet_process_id_t id, __u64 mbits)
 221 {
 222         if (LNET_CPT_NUMBER == 1)
 223                 return ptl->ptl_mtables[0]; /* the only one */
 224
 225         /* if it's a unique portal, return match-table hashed by NID */
 226         return lnet_ptl_is_unique(ptl) ?
 227                ptl->ptl_mtables[lnet_cpt_of_nid(id.nid)] : NULL;
 228 }
 229
 230 struct lnet_match_table *
 231 lnet_mt_of_attach(unsigned int index, lnet_process_id_t id,
 232                   __u64 mbits, __u64 ignore_bits, lnet_ins_pos_t pos)
 233 {
 234         struct lnet_portal *ptl;
 235         struct lnet_match_table *mtable;
 236
 237         /* NB: called w/o lock */
 238         LASSERT(index < the_lnet.ln_nportals);
 239
 240         if (!lnet_ptl_match_type(index, id, mbits, ignore_bits))
 241                 return NULL;
 242
 243         ptl = the_lnet.ln_portals[index];
 244
 245         mtable = lnet_match2mt(ptl, id, mbits);
 246         if (mtable) /* unique portal or only one match-table */
 247                 return mtable;
 248
 249         /* it's a wildcard portal */
 250         switch (pos) {
 251         default:
 252                 return NULL;
 253         case LNET_INS_BEFORE:
 254         case LNET_INS_AFTER:
 255                 /*
 256                  * posted by no affinity thread, always hash to specific
 257                  * match-table to avoid buffer stealing which is heavy
 258                  */
 259                 return ptl->ptl_mtables[ptl->ptl_index % LNET_CPT_NUMBER];
 260         case LNET_INS_LOCAL:
 261                 /* posted by cpu-affinity thread */
 262                 return ptl->ptl_mtables[lnet_cpt_current()];
 263         }
 264 }
 265
 266 static struct lnet_match_table *
 267 lnet_mt_of_match(struct lnet_match_info *info, struct lnet_msg *msg)
 268 {
 269         struct lnet_match_table *mtable;
 270         struct lnet_portal *ptl;
 271         unsigned int nmaps;
 272         unsigned int rotor;
 273         unsigned int cpt;
 274         bool routed;
 275
 276         /* NB: called w/o lock */
 277         LASSERT(info->mi_portal < the_lnet.ln_nportals);
 278         ptl = the_lnet.ln_portals[info->mi_portal];
 279
 280         LASSERT(lnet_ptl_is_wildcard(ptl) || lnet_ptl_is_unique(ptl));
 281
 282         mtable = lnet_match2mt(ptl, info->mi_id, info->mi_mbits);
 283         if (mtable)
 284                 return mtable;
 285
 286         /* it's a wildcard portal */
 287         routed = LNET_NIDNET(msg->msg_hdr.src_nid) !=
 288                  LNET_NIDNET(msg->msg_hdr.dest_nid);
 289
 290         if (portal_rotor == LNET_PTL_ROTOR_OFF ||
 291             (portal_rotor != LNET_PTL_ROTOR_ON && !routed)) {
 292                 cpt = lnet_cpt_current();
 293                 if (ptl->ptl_mtables[cpt]->mt_enabled)
 294                         return ptl->ptl_mtables[cpt];
 295         }
 296
 297         rotor = ptl->ptl_rotor++; /* get round-robin factor */
 298         if (portal_rotor == LNET_PTL_ROTOR_HASH_RT && routed)
 299                 cpt = lnet_cpt_of_nid(msg->msg_hdr.src_nid);
 300         else
 301                 cpt = rotor % LNET_CPT_NUMBER;
 302
 303         if (!ptl->ptl_mtables[cpt]->mt_enabled) {
 304                 /* is there any active entry for this portal? */
 305                 nmaps = ptl->ptl_mt_nmaps;
 306                 /* map to an active mtable to avoid heavy "stealing" */
 307                 if (nmaps) {
 308                         /*
 309                          * NB: there is possibility that ptl_mt_maps is being
 310                          * changed because we are not under protection of
 311                          * lnet_ptl_lock, but it shouldn't hurt anything
 312                          */
 313                         cpt = ptl->ptl_mt_maps[rotor % nmaps];
 314                 }
 315         }
 316
 317         return ptl->ptl_mtables[cpt];
 318 }
 319
 320 static int
 321 lnet_mt_test_exhausted(struct lnet_match_table *mtable, int pos)
 322 {
 323         __u64 *bmap;
 324         int i;
 325
 326         if (!lnet_ptl_is_wildcard(the_lnet.ln_portals[mtable->mt_portal]))
 327                 return 0;
 328
 329         if (pos < 0) { /* check all bits */
 330                 for (i = 0; i < LNET_MT_EXHAUSTED_BMAP; i++) {
 331                         if (mtable->mt_exhausted[i] != (__u64)(-1))
 332                                 return 0;
 333                 }
 334                 return 1;
 335         }
 336
 337         LASSERT(pos <= LNET_MT_HASH_IGNORE);
 338         /* mtable::mt_mhash[pos] is marked as exhausted or not */
 339         bmap = &mtable->mt_exhausted[pos >> LNET_MT_BITS_U64];
 340         pos &= (1 << LNET_MT_BITS_U64) - 1;
 341
 342         return (*bmap & (1ULL << pos));
 343 }
 344
 345 static void
 346 lnet_mt_set_exhausted(struct lnet_match_table *mtable, int pos, int exhausted)
 347 {
 348         __u64 *bmap;
 349
 350         LASSERT(lnet_ptl_is_wildcard(the_lnet.ln_portals[mtable->mt_portal]));
 351         LASSERT(pos <= LNET_MT_HASH_IGNORE);
 352
 353         /* set mtable::mt_mhash[pos] as exhausted/non-exhausted */
 354         bmap = &mtable->mt_exhausted[pos >> LNET_MT_BITS_U64];
 355         pos &= (1 << LNET_MT_BITS_U64) - 1;
 356
 357         if (!exhausted)
 358                 *bmap &= ~(1ULL << pos);
 359         else
 360                 *bmap |= 1ULL << pos;
 361 }
 362
 363 struct list_head *
 364 lnet_mt_match_head(struct lnet_match_table *mtable,
 365                    lnet_process_id_t id, __u64 mbits)
 366 {
 367         struct lnet_portal *ptl = the_lnet.ln_portals[mtable->mt_portal];
 368
 369         if (lnet_ptl_is_wildcard(ptl)) {
 370                 return &mtable->mt_mhash[mbits & LNET_MT_HASH_MASK];
 371         } else {
 372                 unsigned long hash = mbits + id.nid + id.pid;
 373
 374                 LASSERT(lnet_ptl_is_unique(ptl));
 375                 hash = hash_long(hash, LNET_MT_HASH_BITS);
 376                 return &mtable->mt_mhash[hash];
 377         }
 378 }
 379
 380 int
 381 lnet_mt_match_md(struct lnet_match_table *mtable,
 382                  struct lnet_match_info *info, struct lnet_msg *msg)
 383 {
 384         struct list_head *head;
 385         lnet_me_t *me;
 386         lnet_me_t *tmp;
 387         int exhausted = 0;
 388         int rc;
 389
 390         /* any ME with ignore bits? */
 391         if (!list_empty(&mtable->mt_mhash[LNET_MT_HASH_IGNORE]))
 392                 head = &mtable->mt_mhash[LNET_MT_HASH_IGNORE];
 393         else
 394                 head = lnet_mt_match_head(mtable, info->mi_id, info->mi_mbits);
 395  again:
 396         /* NB: only wildcard portal needs to return LNET_MATCHMD_EXHAUSTED */
 397         if (lnet_ptl_is_wildcard(the_lnet.ln_portals[mtable->mt_portal]))
 398                 exhausted = LNET_MATCHMD_EXHAUSTED;
 399
 400         list_for_each_entry_safe(me, tmp, head, me_list) {
 401                 /* ME attached but MD not attached yet */
 402                 if (!me->me_md)
 403                         continue;
 404
 405                 LASSERT(me == me->me_md->md_me);
 406
 407                 rc = lnet_try_match_md(me->me_md, info, msg);
 408                 if (!(rc & LNET_MATCHMD_EXHAUSTED))
 409                         exhausted = 0; /* mlist is not empty */
 410
 411                 if (rc & LNET_MATCHMD_FINISH) {
 412                         /*
 413                          * don't return EXHAUSTED bit because we don't know
 414                          * whether the mlist is empty or not
 415                          */
 416                         return rc & ~LNET_MATCHMD_EXHAUSTED;
 417                 }
 418         }
 419
 420         if (exhausted == LNET_MATCHMD_EXHAUSTED) { /* @head is exhausted */
 421                 lnet_mt_set_exhausted(mtable, head - mtable->mt_mhash, 1);
 422                 if (!lnet_mt_test_exhausted(mtable, -1))
 423                         exhausted = 0;
 424         }
 425
 426         if (!exhausted && head == &mtable->mt_mhash[LNET_MT_HASH_IGNORE]) {
 427                 head = lnet_mt_match_head(mtable, info->mi_id, info->mi_mbits);
 428                 goto again; /* re-check MEs w/o ignore-bits */
 429         }
 430
 431         if (info->mi_opc == LNET_MD_OP_GET ||
 432             !lnet_ptl_is_lazy(the_lnet.ln_portals[info->mi_portal]))
 433                 return exhausted | LNET_MATCHMD_DROP;
 434
 435         return exhausted | LNET_MATCHMD_NONE;
 436 }
 437
 438 static int
 439 lnet_ptl_match_early(struct lnet_portal *ptl, struct lnet_msg *msg)
 440 {
 441         int rc;
 442
 443         /*
 444          * message arrived before any buffer posting on this portal,
 445          * simply delay or drop this message
 446          */
 447         if (likely(lnet_ptl_is_wildcard(ptl) || lnet_ptl_is_unique(ptl)))
 448                 return 0;
 449
 450         lnet_ptl_lock(ptl);
 451         /* check it again with hold of lock */
 452         if (lnet_ptl_is_wildcard(ptl) || lnet_ptl_is_unique(ptl)) {
 453                 lnet_ptl_unlock(ptl);
 454                 return 0;
 455         }
 456
 457         if (lnet_ptl_is_lazy(ptl)) {
 458                 if (msg->msg_rx_ready_delay) {
 459                         msg->msg_rx_delayed = 1;
 460                         list_add_tail(&msg->msg_list,
 461                                       &ptl->ptl_msg_delayed);
 462                 }
 463                 rc = LNET_MATCHMD_NONE;
 464         } else {
 465                 rc = LNET_MATCHMD_DROP;
 466         }
 467
 468         lnet_ptl_unlock(ptl);
 469         return rc;
 470 }
 471
 472 static int
 473 lnet_ptl_match_delay(struct lnet_portal *ptl,
 474                      struct lnet_match_info *info, struct lnet_msg *msg)
 475 {
 476         int first = ptl->ptl_mt_maps[0]; /* read w/o lock */
 477         int rc = 0;
 478         int i;
 479
 480         /*
 481          * steal buffer from other CPTs, and delay it if nothing to steal,
 482          * this function is more expensive than a regular match, but we
 483          * don't expect it can happen a lot
 484          */
 485         LASSERT(lnet_ptl_is_wildcard(ptl));
 486
 487         for (i = 0; i < LNET_CPT_NUMBER; i++) {
 488                 struct lnet_match_table *mtable;
 489                 int cpt;
 490
 491                 cpt = (first + i) % LNET_CPT_NUMBER;
 492                 mtable = ptl->ptl_mtables[cpt];
 493                 if (i && i != LNET_CPT_NUMBER - 1 && !mtable->mt_enabled)
 494                         continue;
 495
 496                 lnet_res_lock(cpt);
 497                 lnet_ptl_lock(ptl);
 498
 499                 if (!i) { /* the first try, attach on stealing list */
 500                         list_add_tail(&msg->msg_list,
 501                                       &ptl->ptl_msg_stealing);
 502                 }
 503
 504                 if (!list_empty(&msg->msg_list)) { /* on stealing list */
 505                         rc = lnet_mt_match_md(mtable, info, msg);
 506
 507                         if ((rc & LNET_MATCHMD_EXHAUSTED) &&
 508                             mtable->mt_enabled)
 509                                 lnet_ptl_disable_mt(ptl, cpt);
 510
 511                         if (rc & LNET_MATCHMD_FINISH)
 512                                 list_del_init(&msg->msg_list);
 513
 514                 } else {
 515                         /*
 516                          * could be matched by lnet_ptl_attach_md()
 517                          * which is called by another thread
 518                          */
 519                         rc = !msg->msg_md ?
 520                              LNET_MATCHMD_DROP : LNET_MATCHMD_OK;
 521                 }
 522
 523                 if (!list_empty(&msg->msg_list) && /* not matched yet */
 524                     (i == LNET_CPT_NUMBER - 1 || /* the last CPT */
 525                      !ptl->ptl_mt_nmaps ||   /* no active CPT */
 526                      (ptl->ptl_mt_nmaps == 1 &&  /* the only active CPT */
 527                       ptl->ptl_mt_maps[0] == cpt))) {
 528                         /* nothing to steal, delay or drop */
 529                         list_del_init(&msg->msg_list);
 530
 531                         if (lnet_ptl_is_lazy(ptl)) {
 532                                 msg->msg_rx_delayed = 1;
 533                                 list_add_tail(&msg->msg_list,
 534                                               &ptl->ptl_msg_delayed);
 535                                 rc = LNET_MATCHMD_NONE;
 536                         } else {
 537                                 rc = LNET_MATCHMD_DROP;
 538                         }
 539                 }
 540
 541                 lnet_ptl_unlock(ptl);
 542                 lnet_res_unlock(cpt);
 543
 544                 if ((rc & LNET_MATCHMD_FINISH) || msg->msg_rx_delayed)
 545                         break;
 546         }
 547
 548         return rc;
 549 }
 550
 551 int
 552 lnet_ptl_match_md(struct lnet_match_info *info, struct lnet_msg *msg)
 553 {
 554         struct lnet_match_table *mtable;
 555         struct lnet_portal *ptl;
 556         int rc;
 557
 558         CDEBUG(D_NET, "Request from %s of length %d into portal %d MB=%#llx\n",
 559                libcfs_id2str(info->mi_id), info->mi_rlength, info->mi_portal,
 560                info->mi_mbits);
 561
 562         if (info->mi_portal >= the_lnet.ln_nportals) {
 563                 CERROR("Invalid portal %d not in [0-%d]\n",
 564                        info->mi_portal, the_lnet.ln_nportals);
 565                 return LNET_MATCHMD_DROP;
 566         }
 567
 568         ptl = the_lnet.ln_portals[info->mi_portal];
 569         rc = lnet_ptl_match_early(ptl, msg);
 570         if (rc) /* matched or delayed early message */
 571                 return rc;
 572
 573         mtable = lnet_mt_of_match(info, msg);
 574         lnet_res_lock(mtable->mt_cpt);
 575
 576         if (the_lnet.ln_shutdown) {
 577                 rc = LNET_MATCHMD_DROP;
 578                 goto out1;
 579         }
 580
 581         rc = lnet_mt_match_md(mtable, info, msg);
 582         if ((rc & LNET_MATCHMD_EXHAUSTED) && mtable->mt_enabled) {
 583                 lnet_ptl_lock(ptl);
 584                 lnet_ptl_disable_mt(ptl, mtable->mt_cpt);
 585                 lnet_ptl_unlock(ptl);
 586         }
 587
 588         if (rc & LNET_MATCHMD_FINISH)   /* matched or dropping */
 589                 goto out1;
 590
 591         if (!msg->msg_rx_ready_delay)
 592                 goto out1;
 593
 594         LASSERT(lnet_ptl_is_lazy(ptl));
 595         LASSERT(!msg->msg_rx_delayed);
 596
 597         /* NB: we don't expect "delay" can happen a lot */
 598         if (lnet_ptl_is_unique(ptl) || LNET_CPT_NUMBER == 1) {
 599                 lnet_ptl_lock(ptl);
 600
 601                 msg->msg_rx_delayed = 1;
 602                 list_add_tail(&msg->msg_list, &ptl->ptl_msg_delayed);
 603
 604                 lnet_ptl_unlock(ptl);
 605                 lnet_res_unlock(mtable->mt_cpt);
 606
 607         } else  {
 608                 lnet_res_unlock(mtable->mt_cpt);
 609                 rc = lnet_ptl_match_delay(ptl, info, msg);
 610         }
 611
 612         if (msg->msg_rx_delayed) {
 613                 CDEBUG(D_NET,
 614                        "Delaying %s from %s ptl %d MB %#llx off %d len %d\n",
 615                        info->mi_opc == LNET_MD_OP_PUT ? "PUT" : "GET",
 616                        libcfs_id2str(info->mi_id), info->mi_portal,
 617                        info->mi_mbits, info->mi_roffset, info->mi_rlength);
 618         }
 619         goto out0;
 620  out1:
 621         lnet_res_unlock(mtable->mt_cpt);
 622  out0:
 623         /* EXHAUSTED bit is only meaningful for internal functions */
 624         return rc & ~LNET_MATCHMD_EXHAUSTED;
 625 }
 626
 627 void
 628 lnet_ptl_detach_md(lnet_me_t *me, lnet_libmd_t *md)
 629 {
 630         LASSERT(me->me_md == md && md->md_me == me);
 631
 632         me->me_md = NULL;
 633         md->md_me = NULL;
 634 }
 635
 636 /* called with lnet_res_lock held */
 637 void
 638 lnet_ptl_attach_md(lnet_me_t *me, lnet_libmd_t *md,
 639                    struct list_head *matches, struct list_head *drops)
 640 {
 641         struct lnet_portal *ptl = the_lnet.ln_portals[me->me_portal];
 642         struct lnet_match_table *mtable;
 643         struct list_head *head;
 644         lnet_msg_t *tmp;
 645         lnet_msg_t *msg;
 646         int exhausted = 0;
 647         int cpt;
 648
 649         LASSERT(!md->md_refcount); /* a brand new MD */
 650
 651         me->me_md = md;
 652         md->md_me = me;
 653
 654         cpt = lnet_cpt_of_cookie(md->md_lh.lh_cookie);
 655         mtable = ptl->ptl_mtables[cpt];
 656
 657         if (list_empty(&ptl->ptl_msg_stealing) &&
 658             list_empty(&ptl->ptl_msg_delayed) &&
 659             !lnet_mt_test_exhausted(mtable, me->me_pos))
 660                 return;
 661
 662         lnet_ptl_lock(ptl);
 663         head = &ptl->ptl_msg_stealing;
 664  again:
 665         list_for_each_entry_safe(msg, tmp, head, msg_list) {
 666                 struct lnet_match_info info;
 667                 lnet_hdr_t *hdr;
 668                 int rc;
 669
 670                 LASSERT(msg->msg_rx_delayed || head == &ptl->ptl_msg_stealing);
 671
 672                 hdr = &msg->msg_hdr;
 673                 info.mi_id.nid  = hdr->src_nid;
 674                 info.mi_id.pid  = hdr->src_pid;
 675                 info.mi_opc     = LNET_MD_OP_PUT;
 676                 info.mi_portal  = hdr->msg.put.ptl_index;
 677                 info.mi_rlength = hdr->payload_length;
 678                 info.mi_roffset = hdr->msg.put.offset;
 679                 info.mi_mbits   = hdr->msg.put.match_bits;
 680
 681                 rc = lnet_try_match_md(md, &info, msg);
 682
 683                 exhausted = (rc & LNET_MATCHMD_EXHAUSTED);
 684                 if (rc & LNET_MATCHMD_NONE) {
 685                         if (exhausted)
 686                                 break;
 687                         continue;
 688                 }
 689
 690                 /* Hurrah! This _is_ a match */
 691                 LASSERT(rc & LNET_MATCHMD_FINISH);
 692                 list_del_init(&msg->msg_list);
 693
 694                 if (head == &ptl->ptl_msg_stealing) {
 695                         if (exhausted)
 696                                 break;
 697                         /* stealing thread will handle the message */
 698                         continue;
 699                 }
 700
 701                 if (rc & LNET_MATCHMD_OK) {
 702                         list_add_tail(&msg->msg_list, matches);
 703
 704                         CDEBUG(D_NET, "Resuming delayed PUT from %s portal %d match %llu offset %d length %d.\n",
 705                                libcfs_id2str(info.mi_id),
 706                                info.mi_portal, info.mi_mbits,
 707                                info.mi_roffset, info.mi_rlength);
 708                 } else {
 709                         list_add_tail(&msg->msg_list, drops);
 710                 }
 711
 712                 if (exhausted)
 713                         break;
 714         }
 715
 716         if (!exhausted && head == &ptl->ptl_msg_stealing) {
 717                 head = &ptl->ptl_msg_delayed;
 718                 goto again;
 719         }
 720
 721         if (lnet_ptl_is_wildcard(ptl) && !exhausted) {
 722                 lnet_mt_set_exhausted(mtable, me->me_pos, 0);
 723                 if (!mtable->mt_enabled)
 724                         lnet_ptl_enable_mt(ptl, cpt);
 725         }
 726
 727         lnet_ptl_unlock(ptl);
 728 }
 729
 730 static void
 731 lnet_ptl_cleanup(struct lnet_portal *ptl)
 732 {
 733         struct lnet_match_table *mtable;
 734         int i;
 735
 736         if (!ptl->ptl_mtables) /* uninitialized portal */
 737                 return;
 738
 739         LASSERT(list_empty(&ptl->ptl_msg_delayed));
 740         LASSERT(list_empty(&ptl->ptl_msg_stealing));
 741         cfs_percpt_for_each(mtable, i, ptl->ptl_mtables) {
 742                 struct list_head *mhash;
 743                 lnet_me_t *me;
 744                 int j;
 745
 746                 if (!mtable->mt_mhash) /* uninitialized match-table */
 747                         continue;
 748
 749                 mhash = mtable->mt_mhash;
 750                 /* cleanup ME */
 751                 for (j = 0; j < LNET_MT_HASH_SIZE + 1; j++) {
 752                         while (!list_empty(&mhash[j])) {
 753                                 me = list_entry(mhash[j].next,
 754                                                 lnet_me_t, me_list);
 755                                 CERROR("Active ME %p on exit\n", me);
 756                                 list_del(&me->me_list);
 757                                 lnet_me_free(me);
 758                         }
 759                 }
 760                 /* the extra entry is for MEs with ignore bits */
 761                 LIBCFS_FREE(mhash, sizeof(*mhash) * (LNET_MT_HASH_SIZE + 1));
 762         }
 763
 764         cfs_percpt_free(ptl->ptl_mtables);
 765         ptl->ptl_mtables = NULL;
 766 }
 767
 768 static int
 769 lnet_ptl_setup(struct lnet_portal *ptl, int index)
 770 {
 771         struct lnet_match_table *mtable;
 772         struct list_head *mhash;
 773         int i;
 774         int j;
 775
 776         ptl->ptl_mtables = cfs_percpt_alloc(lnet_cpt_table(),
 777                                             sizeof(struct lnet_match_table));
 778         if (!ptl->ptl_mtables) {
 779                 CERROR("Failed to create match table for portal %d\n", index);
 780                 return -ENOMEM;
 781         }
 782
 783         ptl->ptl_index = index;
 784         INIT_LIST_HEAD(&ptl->ptl_msg_delayed);
 785         INIT_LIST_HEAD(&ptl->ptl_msg_stealing);
 786         spin_lock_init(&ptl->ptl_lock);
 787         cfs_percpt_for_each(mtable, i, ptl->ptl_mtables) {
 788                 /* the extra entry is for MEs with ignore bits */
 789                 LIBCFS_CPT_ALLOC(mhash, lnet_cpt_table(), i,
 790                                  sizeof(*mhash) * (LNET_MT_HASH_SIZE + 1));
 791                 if (!mhash) {
 792                         CERROR("Failed to create match hash for portal %d\n",
 793                                index);
 794                         goto failed;
 795                 }
 796
 797                 memset(&mtable->mt_exhausted[0], -1,
 798                        sizeof(mtable->mt_exhausted[0]) *
 799                        LNET_MT_EXHAUSTED_BMAP);
 800                 mtable->mt_mhash = mhash;
 801                 for (j = 0; j < LNET_MT_HASH_SIZE + 1; j++)
 802                         INIT_LIST_HEAD(&mhash[j]);
 803
 804                 mtable->mt_portal = index;
 805                 mtable->mt_cpt = i;
 806         }
 807
 808         return 0;
 809  failed:
 810         lnet_ptl_cleanup(ptl);
 811         return -ENOMEM;
 812 }
 813
 814 void
 815 lnet_portals_destroy(void)
 816 {
 817         int i;
 818
 819         if (!the_lnet.ln_portals)
 820                 return;
 821
 822         for (i = 0; i < the_lnet.ln_nportals; i++)
 823                 lnet_ptl_cleanup(the_lnet.ln_portals[i]);
 824
 825         cfs_array_free(the_lnet.ln_portals);
 826         the_lnet.ln_portals = NULL;
 827 }
 828
 829 int
 830 lnet_portals_create(void)
 831 {
 832         int size;
 833         int i;
 834
 835         size = offsetof(struct lnet_portal, ptl_mt_maps[LNET_CPT_NUMBER]);
 836
 837         the_lnet.ln_nportals = MAX_PORTALS;
 838         the_lnet.ln_portals = cfs_array_alloc(the_lnet.ln_nportals, size);
 839         if (!the_lnet.ln_portals) {
 840                 CERROR("Failed to allocate portals table\n");
 841                 return -ENOMEM;
 842         }
 843
 844         for (i = 0; i < the_lnet.ln_nportals; i++) {
 845                 if (lnet_ptl_setup(the_lnet.ln_portals[i], i)) {
 846                         lnet_portals_destroy();
 847                         return -ENOMEM;
 848                 }
 849         }
 850
 851         return 0;
 852 }
 853
 854 /**
 855  * Turn on the lazy portal attribute. Use with caution!
 856  *
 857  * This portal attribute only affects incoming PUT requests to the portal,
 858  * and is off by default. By default, if there's no matching MD for an
 859  * incoming PUT request, it is simply dropped. With the lazy attribute on,
 860  * such requests are queued indefinitely until either a matching MD is
 861  * posted to the portal or the lazy attribute is turned off.
 862  *
 863  * It would prevent dropped requests, however it should be regarded as the
 864  * last line of defense - i.e. users must keep a close watch on active
 865  * buffers on a lazy portal and once it becomes too low post more buffers as
 866  * soon as possible. This is because delayed requests usually have detrimental
 867  * effects on underlying network connections. A few delayed requests often
 868  * suffice to bring an underlying connection to a complete halt, due to flow
 869  * control mechanisms.
 870  *
 871  * There's also a DOS attack risk. If users don't post match-all MDs on a
 872  * lazy portal, a malicious peer can easily stop a service by sending some
 873  * PUT requests with match bits that won't match any MD. A routed server is
 874  * especially vulnerable since the connections to its neighbor routers are
 875  * shared among all clients.
 876  *
 877  * \param portal Index of the portal to enable the lazy attribute on.
 878  *
 879  * \retval 0       On success.
 880  * \retval -EINVAL If \a portal is not a valid index.
 881  */
 882 int
 883 LNetSetLazyPortal(int portal)
 884 {
 885         struct lnet_portal *ptl;
 886
 887         if (portal < 0 || portal >= the_lnet.ln_nportals)
 888                 return -EINVAL;
 889
 890         CDEBUG(D_NET, "Setting portal %d lazy\n", portal);
 891         ptl = the_lnet.ln_portals[portal];
 892
 893         lnet_res_lock(LNET_LOCK_EX);
 894         lnet_ptl_lock(ptl);
 895
 896         lnet_ptl_setopt(ptl, LNET_PTL_LAZY);
 897
 898         lnet_ptl_unlock(ptl);
 899         lnet_res_unlock(LNET_LOCK_EX);
 900
 901         return 0;
 902 }
 903 EXPORT_SYMBOL(LNetSetLazyPortal);
 904
 905 /**
 906  * Turn off the lazy portal attribute. Delayed requests on the portal,
 907  * if any, will be all dropped when this function returns.
 908  *
 909  * \param portal Index of the portal to disable the lazy attribute on.
 910  *
 911  * \retval 0       On success.
 912  * \retval -EINVAL If \a portal is not a valid index.
 913  */
 914 int
 915 LNetClearLazyPortal(int portal)
 916 {
 917         struct lnet_portal *ptl;
 918         LIST_HEAD(zombies);
 919
 920         if (portal < 0 || portal >= the_lnet.ln_nportals)
 921                 return -EINVAL;
 922
 923         ptl = the_lnet.ln_portals[portal];
 924
 925         lnet_res_lock(LNET_LOCK_EX);
 926         lnet_ptl_lock(ptl);
 927
 928         if (!lnet_ptl_is_lazy(ptl)) {
 929                 lnet_ptl_unlock(ptl);
 930                 lnet_res_unlock(LNET_LOCK_EX);
 931                 return 0;
 932         }
 933
 934         if (the_lnet.ln_shutdown)
 935                 CWARN("Active lazy portal %d on exit\n", portal);
 936         else
 937                 CDEBUG(D_NET, "clearing portal %d lazy\n", portal);
 938
 939         /* grab all the blocked messages atomically */
 940         list_splice_init(&ptl->ptl_msg_delayed, &zombies);
 941
 942         lnet_ptl_unsetopt(ptl, LNET_PTL_LAZY);
 943
 944         lnet_ptl_unlock(ptl);
 945         lnet_res_unlock(LNET_LOCK_EX);
 946
 947         lnet_drop_delayed_msg_list(&zombies, "Clearing lazy portal attr");
 948
 949         return 0;
 950 }
 951 EXPORT_SYMBOL(LNetClearLazyPortal);