lnet/lnet/lib-move.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, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, 2015, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lnet/lnet/lib-move.c
  37  *
  38  * Data movement routines
  39  */
  40
  41 #define DEBUG_SUBSYSTEM S_LNET
  42
  43 #include <lnet/lib-lnet.h>
  44 #include <linux/nsproxy.h>
  45 #include <net/net_namespace.h>
  46
  47 static int local_nid_dist_zero = 1;
  48 module_param(local_nid_dist_zero, int, 0444);
  49 MODULE_PARM_DESC(local_nid_dist_zero, "Reserved");
  50
  51 int
  52 lnet_fail_nid(lnet_nid_t nid, unsigned int threshold)
  53 {
  54         lnet_test_peer_t *tp;
  55         struct list_head *el;
  56         struct list_head *next;
  57         struct list_head  cull;
  58
  59         /* NB: use lnet_net_lock(0) to serialize operations on test peers */
  60         if (threshold != 0) {
  61                 /* Adding a new entry */
  62                 LIBCFS_ALLOC(tp, sizeof(*tp));
  63                 if (tp == NULL)
  64                         return -ENOMEM;
  65
  66                 tp->tp_nid = nid;
  67                 tp->tp_threshold = threshold;
  68
  69                 lnet_net_lock(0);
  70                 list_add_tail(&tp->tp_list, &the_lnet.ln_test_peers);
  71                 lnet_net_unlock(0);
  72                 return 0;
  73         }
  74
  75         /* removing entries */
  76         INIT_LIST_HEAD(&cull);
  77
  78         lnet_net_lock(0);
  79
  80         list_for_each_safe(el, next, &the_lnet.ln_test_peers) {
  81                 tp = list_entry(el, lnet_test_peer_t, tp_list);
  82
  83                 if (tp->tp_threshold == 0 ||    /* needs culling anyway */
  84                     nid == LNET_NID_ANY ||      /* removing all entries */
  85                     tp->tp_nid == nid) {        /* matched this one */
  86                         list_del(&tp->tp_list);
  87                         list_add(&tp->tp_list, &cull);
  88                 }
  89         }
  90
  91         lnet_net_unlock(0);
  92
  93         while (!list_empty(&cull)) {
  94                 tp = list_entry(cull.next, lnet_test_peer_t, tp_list);
  95
  96                 list_del(&tp->tp_list);
  97                 LIBCFS_FREE(tp, sizeof(*tp));
  98         }
  99         return 0;
 100 }
 101
 102 static int
 103 fail_peer (lnet_nid_t nid, int outgoing)
 104 {
 105         lnet_test_peer_t *tp;
 106         struct list_head *el;
 107         struct list_head *next;
 108         struct list_head  cull;
 109         int               fail = 0;
 110
 111         INIT_LIST_HEAD(&cull);
 112
 113         /* NB: use lnet_net_lock(0) to serialize operations on test peers */
 114         lnet_net_lock(0);
 115
 116         list_for_each_safe(el, next, &the_lnet.ln_test_peers) {
 117                 tp = list_entry(el, lnet_test_peer_t, tp_list);
 118
 119                 if (tp->tp_threshold == 0) {
 120                         /* zombie entry */
 121                         if (outgoing) {
 122                                 /* only cull zombies on outgoing tests,
 123                                  * since we may be at interrupt priority on
 124                                  * incoming messages. */
 125                                 list_del(&tp->tp_list);
 126                                 list_add(&tp->tp_list, &cull);
 127                         }
 128                         continue;
 129                 }
 130
 131                 if (tp->tp_nid == LNET_NID_ANY ||       /* fail every peer */
 132                     nid == tp->tp_nid) {                /* fail this peer */
 133                         fail = 1;
 134
 135                         if (tp->tp_threshold != LNET_MD_THRESH_INF) {
 136                                 tp->tp_threshold--;
 137                                 if (outgoing &&
 138                                     tp->tp_threshold == 0) {
 139                                         /* see above */
 140                                         list_del(&tp->tp_list);
 141                                         list_add(&tp->tp_list, &cull);
 142                                 }
 143                         }
 144                         break;
 145                 }
 146         }
 147
 148         lnet_net_unlock(0);
 149
 150         while (!list_empty(&cull)) {
 151                 tp = list_entry(cull.next, lnet_test_peer_t, tp_list);
 152                 list_del(&tp->tp_list);
 153
 154                 LIBCFS_FREE(tp, sizeof(*tp));
 155         }
 156
 157         return fail;
 158 }
 159
 160 unsigned int
 161 lnet_iov_nob(unsigned int niov, struct kvec *iov)
 162 {
 163         unsigned int nob = 0;
 164
 165         LASSERT(niov == 0 || iov != NULL);
 166         while (niov-- > 0)
 167                 nob += (iov++)->iov_len;
 168
 169         return (nob);
 170 }
 171 EXPORT_SYMBOL(lnet_iov_nob);
 172
 173 void
 174 lnet_copy_iov2iov(unsigned int ndiov, struct kvec *diov, unsigned int doffset,
 175                   unsigned int nsiov, struct kvec *siov, unsigned int soffset,
 176                   unsigned int nob)
 177 {
 178         /* NB diov, siov are READ-ONLY */
 179         unsigned int  this_nob;
 180
 181         if (nob == 0)
 182                 return;
 183
 184         /* skip complete frags before 'doffset' */
 185         LASSERT (ndiov > 0);
 186         while (doffset >= diov->iov_len) {
 187                 doffset -= diov->iov_len;
 188                 diov++;
 189                 ndiov--;
 190                 LASSERT (ndiov > 0);
 191         }
 192
 193         /* skip complete frags before 'soffset' */
 194         LASSERT (nsiov > 0);
 195         while (soffset >= siov->iov_len) {
 196                 soffset -= siov->iov_len;
 197                 siov++;
 198                 nsiov--;
 199                 LASSERT (nsiov > 0);
 200         }
 201
 202         do {
 203                 LASSERT (ndiov > 0);
 204                 LASSERT (nsiov > 0);
 205                 this_nob = MIN(diov->iov_len - doffset,
 206                                siov->iov_len - soffset);
 207                 this_nob = MIN(this_nob, nob);
 208
 209                 memcpy ((char *)diov->iov_base + doffset,
 210                         (char *)siov->iov_base + soffset, this_nob);
 211                 nob -= this_nob;
 212
 213                 if (diov->iov_len > doffset + this_nob) {
 214                         doffset += this_nob;
 215                 } else {
 216                         diov++;
 217                         ndiov--;
 218                         doffset = 0;
 219                 }
 220
 221                 if (siov->iov_len > soffset + this_nob) {
 222                         soffset += this_nob;
 223                 } else {
 224                         siov++;
 225                         nsiov--;
 226                         soffset = 0;
 227                 }
 228         } while (nob > 0);
 229 }
 230 EXPORT_SYMBOL(lnet_copy_iov2iov);
 231
 232 int
 233 lnet_extract_iov(int dst_niov, struct kvec *dst,
 234                  int src_niov, struct kvec *src,
 235                  unsigned int offset, unsigned int len)
 236 {
 237         /* Initialise 'dst' to the subset of 'src' starting at 'offset',
 238          * for exactly 'len' bytes, and return the number of entries.
 239          * NB not destructive to 'src' */
 240         unsigned int    frag_len;
 241         unsigned int    niov;
 242
 243         if (len == 0)                           /* no data => */
 244                 return (0);                     /* no frags */
 245
 246         LASSERT (src_niov > 0);
 247         while (offset >= src->iov_len) {      /* skip initial frags */
 248                 offset -= src->iov_len;
 249                 src_niov--;
 250                 src++;
 251                 LASSERT (src_niov > 0);
 252         }
 253
 254         niov = 1;
 255         for (;;) {
 256                 LASSERT (src_niov > 0);
 257                 LASSERT ((int)niov <= dst_niov);
 258
 259                 frag_len = src->iov_len - offset;
 260                 dst->iov_base = ((char *)src->iov_base) + offset;
 261
 262                 if (len <= frag_len) {
 263                         dst->iov_len = len;
 264                         return (niov);
 265                 }
 266
 267                 dst->iov_len = frag_len;
 268
 269                 len -= frag_len;
 270                 dst++;
 271                 src++;
 272                 niov++;
 273                 src_niov--;
 274                 offset = 0;
 275         }
 276 }
 277 EXPORT_SYMBOL(lnet_extract_iov);
 278
 279
 280 unsigned int
 281 lnet_kiov_nob(unsigned int niov, lnet_kiov_t *kiov)
 282 {
 283         unsigned int  nob = 0;
 284
 285         LASSERT(niov == 0 || kiov != NULL);
 286         while (niov-- > 0)
 287                 nob += (kiov++)->kiov_len;
 288
 289         return (nob);
 290 }
 291 EXPORT_SYMBOL(lnet_kiov_nob);
 292
 293 void
 294 lnet_copy_kiov2kiov(unsigned int ndiov, lnet_kiov_t *diov, unsigned int doffset,
 295                     unsigned int nsiov, lnet_kiov_t *siov, unsigned int soffset,
 296                     unsigned int nob)
 297 {
 298         /* NB diov, siov are READ-ONLY */
 299         unsigned int    this_nob;
 300         char           *daddr = NULL;
 301         char           *saddr = NULL;
 302
 303         if (nob == 0)
 304                 return;
 305
 306         LASSERT (!in_interrupt ());
 307
 308         LASSERT (ndiov > 0);
 309         while (doffset >= diov->kiov_len) {
 310                 doffset -= diov->kiov_len;
 311                 diov++;
 312                 ndiov--;
 313                 LASSERT (ndiov > 0);
 314         }
 315
 316         LASSERT (nsiov > 0);
 317         while (soffset >= siov->kiov_len) {
 318                 soffset -= siov->kiov_len;
 319                 siov++;
 320                 nsiov--;
 321                 LASSERT (nsiov > 0);
 322         }
 323
 324         do {
 325                 LASSERT (ndiov > 0);
 326                 LASSERT (nsiov > 0);
 327                 this_nob = MIN(diov->kiov_len - doffset,
 328                                siov->kiov_len - soffset);
 329                 this_nob = MIN(this_nob, nob);
 330
 331                 if (daddr == NULL)
 332                         daddr = ((char *)kmap(diov->kiov_page)) +
 333                                 diov->kiov_offset + doffset;
 334                 if (saddr == NULL)
 335                         saddr = ((char *)kmap(siov->kiov_page)) +
 336                                 siov->kiov_offset + soffset;
 337
 338                 /* Vanishing risk of kmap deadlock when mapping 2 pages.
 339                  * However in practice at least one of the kiovs will be mapped
 340                  * kernel pages and the map/unmap will be NOOPs */
 341
 342                 memcpy (daddr, saddr, this_nob);
 343                 nob -= this_nob;
 344
 345                 if (diov->kiov_len > doffset + this_nob) {
 346                         daddr += this_nob;
 347                         doffset += this_nob;
 348                 } else {
 349                         kunmap(diov->kiov_page);
 350                         daddr = NULL;
 351                         diov++;
 352                         ndiov--;
 353                         doffset = 0;
 354                 }
 355
 356                 if (siov->kiov_len > soffset + this_nob) {
 357                         saddr += this_nob;
 358                         soffset += this_nob;
 359                 } else {
 360                         kunmap(siov->kiov_page);
 361                         saddr = NULL;
 362                         siov++;
 363                         nsiov--;
 364                         soffset = 0;
 365                 }
 366         } while (nob > 0);
 367
 368         if (daddr != NULL)
 369                 kunmap(diov->kiov_page);
 370         if (saddr != NULL)
 371                 kunmap(siov->kiov_page);
 372 }
 373 EXPORT_SYMBOL(lnet_copy_kiov2kiov);
 374
 375 void
 376 lnet_copy_kiov2iov (unsigned int niov, struct kvec *iov, unsigned int iovoffset,
 377                     unsigned int nkiov, lnet_kiov_t *kiov, unsigned int kiovoffset,
 378                     unsigned int nob)
 379 {
 380         /* NB iov, kiov are READ-ONLY */
 381         unsigned int    this_nob;
 382         char           *addr = NULL;
 383
 384         if (nob == 0)
 385                 return;
 386
 387         LASSERT (!in_interrupt ());
 388
 389         LASSERT (niov > 0);
 390         while (iovoffset >= iov->iov_len) {
 391                 iovoffset -= iov->iov_len;
 392                 iov++;
 393                 niov--;
 394                 LASSERT (niov > 0);
 395         }
 396
 397         LASSERT (nkiov > 0);
 398         while (kiovoffset >= kiov->kiov_len) {
 399                 kiovoffset -= kiov->kiov_len;
 400                 kiov++;
 401                 nkiov--;
 402                 LASSERT (nkiov > 0);
 403         }
 404
 405         do {
 406                 LASSERT (niov > 0);
 407                 LASSERT (nkiov > 0);
 408                 this_nob = MIN(iov->iov_len - iovoffset,
 409                                kiov->kiov_len - kiovoffset);
 410                 this_nob = MIN(this_nob, nob);
 411
 412                 if (addr == NULL)
 413                         addr = ((char *)kmap(kiov->kiov_page)) +
 414                                 kiov->kiov_offset + kiovoffset;
 415
 416                 memcpy ((char *)iov->iov_base + iovoffset, addr, this_nob);
 417                 nob -= this_nob;
 418
 419                 if (iov->iov_len > iovoffset + this_nob) {
 420                         iovoffset += this_nob;
 421                 } else {
 422                         iov++;
 423                         niov--;
 424                         iovoffset = 0;
 425                 }
 426
 427                 if (kiov->kiov_len > kiovoffset + this_nob) {
 428                         addr += this_nob;
 429                         kiovoffset += this_nob;
 430                 } else {
 431                         kunmap(kiov->kiov_page);
 432                         addr = NULL;
 433                         kiov++;
 434                         nkiov--;
 435                         kiovoffset = 0;
 436                 }
 437
 438         } while (nob > 0);
 439
 440         if (addr != NULL)
 441                 kunmap(kiov->kiov_page);
 442 }
 443 EXPORT_SYMBOL(lnet_copy_kiov2iov);
 444
 445 void
 446 lnet_copy_iov2kiov(unsigned int nkiov, lnet_kiov_t *kiov, unsigned int kiovoffset,
 447                    unsigned int niov, struct kvec *iov, unsigned int iovoffset,
 448                    unsigned int nob)
 449 {
 450         /* NB kiov, iov are READ-ONLY */
 451         unsigned int    this_nob;
 452         char           *addr = NULL;
 453
 454         if (nob == 0)
 455                 return;
 456
 457         LASSERT (!in_interrupt ());
 458
 459         LASSERT (nkiov > 0);
 460         while (kiovoffset >= kiov->kiov_len) {
 461                 kiovoffset -= kiov->kiov_len;
 462                 kiov++;
 463                 nkiov--;
 464                 LASSERT (nkiov > 0);
 465         }
 466
 467         LASSERT (niov > 0);
 468         while (iovoffset >= iov->iov_len) {
 469                 iovoffset -= iov->iov_len;
 470                 iov++;
 471                 niov--;
 472                 LASSERT (niov > 0);
 473         }
 474
 475         do {
 476                 LASSERT (nkiov > 0);
 477                 LASSERT (niov > 0);
 478                 this_nob = MIN(kiov->kiov_len - kiovoffset,
 479                                iov->iov_len - iovoffset);
 480                 this_nob = MIN(this_nob, nob);
 481
 482                 if (addr == NULL)
 483                         addr = ((char *)kmap(kiov->kiov_page)) +
 484                                 kiov->kiov_offset + kiovoffset;
 485
 486                 memcpy (addr, (char *)iov->iov_base + iovoffset, this_nob);
 487                 nob -= this_nob;
 488
 489                 if (kiov->kiov_len > kiovoffset + this_nob) {
 490                         addr += this_nob;
 491                         kiovoffset += this_nob;
 492                 } else {
 493                         kunmap(kiov->kiov_page);
 494                         addr = NULL;
 495                         kiov++;
 496                         nkiov--;
 497                         kiovoffset = 0;
 498                 }
 499
 500                 if (iov->iov_len > iovoffset + this_nob) {
 501                         iovoffset += this_nob;
 502                 } else {
 503                         iov++;
 504                         niov--;
 505                         iovoffset = 0;
 506                 }
 507         } while (nob > 0);
 508
 509         if (addr != NULL)
 510                 kunmap(kiov->kiov_page);
 511 }
 512 EXPORT_SYMBOL(lnet_copy_iov2kiov);
 513
 514 int
 515 lnet_extract_kiov(int dst_niov, lnet_kiov_t *dst,
 516                   int src_niov, lnet_kiov_t *src,
 517                   unsigned int offset, unsigned int len)
 518 {
 519         /* Initialise 'dst' to the subset of 'src' starting at 'offset',
 520          * for exactly 'len' bytes, and return the number of entries.
 521          * NB not destructive to 'src' */
 522         unsigned int    frag_len;
 523         unsigned int    niov;
 524
 525         if (len == 0)                           /* no data => */
 526                 return (0);                     /* no frags */
 527
 528         LASSERT (src_niov > 0);
 529         while (offset >= src->kiov_len) {      /* skip initial frags */
 530                 offset -= src->kiov_len;
 531                 src_niov--;
 532                 src++;
 533                 LASSERT (src_niov > 0);
 534         }
 535
 536         niov = 1;
 537         for (;;) {
 538                 LASSERT (src_niov > 0);
 539                 LASSERT ((int)niov <= dst_niov);
 540
 541                 frag_len = src->kiov_len - offset;
 542                 dst->kiov_page = src->kiov_page;
 543                 dst->kiov_offset = src->kiov_offset + offset;
 544
 545                 if (len <= frag_len) {
 546                         dst->kiov_len = len;
 547                         LASSERT (dst->kiov_offset + dst->kiov_len <= PAGE_CACHE_SIZE);
 548                         return niov;
 549                 }
 550
 551                 dst->kiov_len = frag_len;
 552                 LASSERT (dst->kiov_offset + dst->kiov_len <= PAGE_CACHE_SIZE);
 553
 554                 len -= frag_len;
 555                 dst++;
 556                 src++;
 557                 niov++;
 558                 src_niov--;
 559                 offset = 0;
 560         }
 561 }
 562 EXPORT_SYMBOL(lnet_extract_kiov);
 563
 564 void
 565 lnet_ni_recv(lnet_ni_t *ni, void *private, lnet_msg_t *msg, int delayed,
 566              unsigned int offset, unsigned int mlen, unsigned int rlen)
 567 {
 568         unsigned int  niov = 0;
 569         struct kvec *iov = NULL;
 570         lnet_kiov_t  *kiov = NULL;
 571         int           rc;
 572
 573         LASSERT (!in_interrupt ());
 574         LASSERT (mlen == 0 || msg != NULL);
 575
 576         if (msg != NULL) {
 577                 LASSERT(msg->msg_receiving);
 578                 LASSERT(!msg->msg_sending);
 579                 LASSERT(rlen == msg->msg_len);
 580                 LASSERT(mlen <= msg->msg_len);
 581                 LASSERT(msg->msg_offset == offset);
 582                 LASSERT(msg->msg_wanted == mlen);
 583
 584                 msg->msg_receiving = 0;
 585
 586                 if (mlen != 0) {
 587                         niov = msg->msg_niov;
 588                         iov  = msg->msg_iov;
 589                         kiov = msg->msg_kiov;
 590
 591                         LASSERT (niov > 0);
 592                         LASSERT ((iov == NULL) != (kiov == NULL));
 593                 }
 594         }
 595
 596         rc = (ni->ni_lnd->lnd_recv)(ni, private, msg, delayed,
 597                                     niov, iov, kiov, offset, mlen, rlen);
 598         if (rc < 0)
 599                 lnet_finalize(ni, msg, rc);
 600 }
 601
 602 static void
 603 lnet_setpayloadbuffer(lnet_msg_t *msg)
 604 {
 605         lnet_libmd_t *md = msg->msg_md;
 606
 607         LASSERT (msg->msg_len > 0);
 608         LASSERT (!msg->msg_routing);
 609         LASSERT (md != NULL);
 610         LASSERT (msg->msg_niov == 0);
 611         LASSERT (msg->msg_iov == NULL);
 612         LASSERT (msg->msg_kiov == NULL);
 613
 614         msg->msg_niov = md->md_niov;
 615         if ((md->md_options & LNET_MD_KIOV) != 0)
 616                 msg->msg_kiov = md->md_iov.kiov;
 617         else
 618                 msg->msg_iov = md->md_iov.iov;
 619 }
 620
 621 void
 622 lnet_prep_send(lnet_msg_t *msg, int type, lnet_process_id_t target,
 623                unsigned int offset, unsigned int len)
 624 {
 625         msg->msg_type = type;
 626         msg->msg_target = target;
 627         msg->msg_len = len;
 628         msg->msg_offset = offset;
 629
 630         if (len != 0)
 631                 lnet_setpayloadbuffer(msg);
 632
 633         memset (&msg->msg_hdr, 0, sizeof (msg->msg_hdr));
 634         msg->msg_hdr.type           = cpu_to_le32(type);
 635         msg->msg_hdr.dest_nid       = cpu_to_le64(target.nid);
 636         msg->msg_hdr.dest_pid       = cpu_to_le32(target.pid);
 637         /* src_nid will be set later */
 638         msg->msg_hdr.src_pid        = cpu_to_le32(the_lnet.ln_pid);
 639         msg->msg_hdr.payload_length = cpu_to_le32(len);
 640 }
 641
 642 static void
 643 lnet_ni_send(lnet_ni_t *ni, lnet_msg_t *msg)
 644 {
 645         void   *priv = msg->msg_private;
 646         int     rc;
 647
 648         LASSERT (!in_interrupt ());
 649         LASSERT (LNET_NETTYP(LNET_NIDNET(ni->ni_nid)) == LOLND ||
 650                  (msg->msg_txcredit && msg->msg_peertxcredit));
 651
 652         rc = (ni->ni_lnd->lnd_send)(ni, priv, msg);
 653         if (rc < 0)
 654                 lnet_finalize(ni, msg, rc);
 655 }
 656
 657 static int
 658 lnet_ni_eager_recv(lnet_ni_t *ni, lnet_msg_t *msg)
 659 {
 660         int     rc;
 661
 662         LASSERT(!msg->msg_sending);
 663         LASSERT(msg->msg_receiving);
 664         LASSERT(!msg->msg_rx_ready_delay);
 665         LASSERT(ni->ni_lnd->lnd_eager_recv != NULL);
 666
 667         msg->msg_rx_ready_delay = 1;
 668         rc = (ni->ni_lnd->lnd_eager_recv)(ni, msg->msg_private, msg,
 669                                           &msg->msg_private);
 670         if (rc != 0) {
 671                 CERROR("recv from %s / send to %s aborted: "
 672                        "eager_recv failed %d\n",
 673                        libcfs_nid2str(msg->msg_rxpeer->lp_nid),
 674                        libcfs_id2str(msg->msg_target), rc);
 675                 LASSERT(rc < 0); /* required by my callers */
 676         }
 677
 678         return rc;
 679 }
 680
 681 /* NB: caller shall hold a ref on 'lp' as I'd drop lnet_net_lock */
 682 static void
 683 lnet_ni_query_locked(lnet_ni_t *ni, lnet_peer_t *lp)
 684 {
 685         cfs_time_t last_alive = 0;
 686
 687         LASSERT(lnet_peer_aliveness_enabled(lp));
 688         LASSERT(ni->ni_lnd->lnd_query != NULL);
 689
 690         lnet_net_unlock(lp->lp_cpt);
 691         (ni->ni_lnd->lnd_query)(ni, lp->lp_nid, &last_alive);
 692         lnet_net_lock(lp->lp_cpt);
 693
 694         lp->lp_last_query = cfs_time_current();
 695
 696         if (last_alive != 0) /* NI has updated timestamp */
 697                 lp->lp_last_alive = last_alive;
 698 }
 699
 700 /* NB: always called with lnet_net_lock held */
 701 static inline int
 702 lnet_peer_is_alive (lnet_peer_t *lp, cfs_time_t now)
 703 {
 704         int        alive;
 705         cfs_time_t deadline;
 706
 707         LASSERT (lnet_peer_aliveness_enabled(lp));
 708
 709         /* Trust lnet_notify() if it has more recent aliveness news, but
 710          * ignore the initial assumed death (see lnet_peers_start_down()).
 711          */
 712         if (!lp->lp_alive && lp->lp_alive_count > 0 &&
 713             cfs_time_aftereq(lp->lp_timestamp, lp->lp_last_alive))
 714                 return 0;
 715
 716         deadline = cfs_time_add(lp->lp_last_alive,
 717                                 cfs_time_seconds(lp->lp_ni->ni_peertimeout));
 718         alive = cfs_time_after(deadline, now);
 719
 720         /* Update obsolete lp_alive except for routers assumed to be dead
 721          * initially, because router checker would update aliveness in this
 722          * case, and moreover lp_last_alive at peer creation is assumed.
 723          */
 724         if (alive && !lp->lp_alive &&
 725             !(lnet_isrouter(lp) && lp->lp_alive_count == 0))
 726                 lnet_notify_locked(lp, 0, 1, lp->lp_last_alive);
 727
 728         return alive;
 729 }
 730
 731
 732 /* NB: returns 1 when alive, 0 when dead, negative when error;
 733  *     may drop the lnet_net_lock */
 734 static int
 735 lnet_peer_alive_locked (lnet_peer_t *lp)
 736 {
 737         cfs_time_t now = cfs_time_current();
 738
 739         if (!lnet_peer_aliveness_enabled(lp))
 740                 return -ENODEV;
 741
 742         if (lnet_peer_is_alive(lp, now))
 743                 return 1;
 744
 745         /* Peer appears dead, but we should avoid frequent NI queries (at
 746          * most once per lnet_queryinterval seconds). */
 747         if (lp->lp_last_query != 0) {
 748                 static const int lnet_queryinterval = 1;
 749
 750                 cfs_time_t next_query =
 751                            cfs_time_add(lp->lp_last_query,
 752                                         cfs_time_seconds(lnet_queryinterval));
 753
 754                 if (cfs_time_before(now, next_query)) {
 755                         if (lp->lp_alive)
 756                                 CWARN("Unexpected aliveness of peer %s: "
 757                                       "%d < %d (%d/%d)\n",
 758                                       libcfs_nid2str(lp->lp_nid),
 759                                       (int)now, (int)next_query,
 760                                       lnet_queryinterval,
 761                                       lp->lp_ni->ni_peertimeout);
 762                         return 0;
 763                 }
 764         }
 765
 766         /* query NI for latest aliveness news */
 767         lnet_ni_query_locked(lp->lp_ni, lp);
 768
 769         if (lnet_peer_is_alive(lp, now))
 770                 return 1;
 771
 772         lnet_notify_locked(lp, 0, 0, lp->lp_last_alive);
 773         return 0;
 774 }
 775
 776 /**
 777  * \param msg The message to be sent.
 778  * \param do_send True if lnet_ni_send() should be called in this function.
 779  *        lnet_send() is going to lnet_net_unlock immediately after this, so
 780  *        it sets do_send FALSE and I don't do the unlock/send/lock bit.
 781  *
 782  * \retval LNET_CREDIT_OK If \a msg sent or OK to send.
 783  * \retval LNET_CREDIT_WAIT If \a msg blocked for credit.
 784  * \retval -EHOSTUNREACH If the next hop of the message appears dead.
 785  * \retval -ECANCELED If the MD of the message has been unlinked.
 786  */
 787 static int
 788 lnet_post_send_locked(lnet_msg_t *msg, int do_send)
 789 {
 790         lnet_peer_t             *lp = msg->msg_txpeer;
 791         lnet_ni_t               *ni = lp->lp_ni;
 792         int                     cpt = msg->msg_tx_cpt;
 793         struct lnet_tx_queue    *tq = ni->ni_tx_queues[cpt];
 794
 795         /* non-lnet_send() callers have checked before */
 796         LASSERT(!do_send || msg->msg_tx_delayed);
 797         LASSERT(!msg->msg_receiving);
 798         LASSERT(msg->msg_tx_committed);
 799
 800         /* NB 'lp' is always the next hop */
 801         if ((msg->msg_target.pid & LNET_PID_USERFLAG) == 0 &&
 802             lnet_peer_alive_locked(lp) == 0) {
 803                 the_lnet.ln_counters[cpt]->drop_count++;
 804                 the_lnet.ln_counters[cpt]->drop_length += msg->msg_len;
 805                 lnet_net_unlock(cpt);
 806
 807                 CNETERR("Dropping message for %s: peer not alive\n",
 808                         libcfs_id2str(msg->msg_target));
 809                 if (do_send)
 810                         lnet_finalize(ni, msg, -EHOSTUNREACH);
 811
 812                 lnet_net_lock(cpt);
 813                 return -EHOSTUNREACH;
 814         }
 815
 816         if (msg->msg_md != NULL &&
 817             (msg->msg_md->md_flags & LNET_MD_FLAG_ABORTED) != 0) {
 818                 lnet_net_unlock(cpt);
 819
 820                 CNETERR("Aborting message for %s: LNetM[DE]Unlink() already "
 821                         "called on the MD/ME.\n",
 822                         libcfs_id2str(msg->msg_target));
 823                 if (do_send)
 824                         lnet_finalize(ni, msg, -ECANCELED);
 825
 826                 lnet_net_lock(cpt);
 827                 return -ECANCELED;
 828         }
 829
 830         if (!msg->msg_peertxcredit) {
 831                 LASSERT((lp->lp_txcredits < 0) ==
 832                         !list_empty(&lp->lp_txq));
 833
 834                 msg->msg_peertxcredit = 1;
 835                 lp->lp_txqnob += msg->msg_len + sizeof(lnet_hdr_t);
 836                 lp->lp_txcredits--;
 837
 838                 if (lp->lp_txcredits < lp->lp_mintxcredits)
 839                         lp->lp_mintxcredits = lp->lp_txcredits;
 840
 841                 if (lp->lp_txcredits < 0) {
 842                         msg->msg_tx_delayed = 1;
 843                         list_add_tail(&msg->msg_list, &lp->lp_txq);
 844                         return LNET_CREDIT_WAIT;
 845                 }
 846         }
 847
 848         if (!msg->msg_txcredit) {
 849                 LASSERT((tq->tq_credits < 0) ==
 850                         !list_empty(&tq->tq_delayed));
 851
 852                 msg->msg_txcredit = 1;
 853                 tq->tq_credits--;
 854
 855                 if (tq->tq_credits < tq->tq_credits_min)
 856                         tq->tq_credits_min = tq->tq_credits;
 857
 858                 if (tq->tq_credits < 0) {
 859                         msg->msg_tx_delayed = 1;
 860                         list_add_tail(&msg->msg_list, &tq->tq_delayed);
 861                         return LNET_CREDIT_WAIT;
 862                 }
 863         }
 864
 865         if (do_send) {
 866                 lnet_net_unlock(cpt);
 867                 lnet_ni_send(ni, msg);
 868                 lnet_net_lock(cpt);
 869         }
 870         return LNET_CREDIT_OK;
 871 }
 872
 873
 874 static lnet_rtrbufpool_t *
 875 lnet_msg2bufpool(lnet_msg_t *msg)
 876 {
 877         lnet_rtrbufpool_t       *rbp;
 878         int                     cpt;
 879
 880         LASSERT(msg->msg_rx_committed);
 881
 882         cpt = msg->msg_rx_cpt;
 883         rbp = &the_lnet.ln_rtrpools[cpt][0];
 884
 885         LASSERT(msg->msg_len <= LNET_MTU);
 886         while (msg->msg_len > (unsigned int)rbp->rbp_npages * PAGE_CACHE_SIZE) {
 887                 rbp++;
 888                 LASSERT(rbp < &the_lnet.ln_rtrpools[cpt][LNET_NRBPOOLS]);
 889         }
 890
 891         return rbp;
 892 }
 893
 894 static int
 895 lnet_post_routed_recv_locked (lnet_msg_t *msg, int do_recv)
 896 {
 897         /* lnet_parse is going to lnet_net_unlock immediately after this, so it
 898          * sets do_recv FALSE and I don't do the unlock/send/lock bit.
 899          * I return LNET_CREDIT_WAIT if msg blocked and LNET_CREDIT_OK if
 900          * received or OK to receive */
 901         lnet_peer_t         *lp = msg->msg_rxpeer;
 902         lnet_rtrbufpool_t   *rbp;
 903         lnet_rtrbuf_t       *rb;
 904
 905         LASSERT (msg->msg_iov == NULL);
 906         LASSERT (msg->msg_kiov == NULL);
 907         LASSERT (msg->msg_niov == 0);
 908         LASSERT (msg->msg_routing);
 909         LASSERT (msg->msg_receiving);
 910         LASSERT (!msg->msg_sending);
 911
 912         /* non-lnet_parse callers only receive delayed messages */
 913         LASSERT(!do_recv || msg->msg_rx_delayed);
 914
 915         if (!msg->msg_peerrtrcredit) {
 916                 LASSERT((lp->lp_rtrcredits < 0) ==
 917                         !list_empty(&lp->lp_rtrq));
 918
 919                 msg->msg_peerrtrcredit = 1;
 920                 lp->lp_rtrcredits--;
 921                 if (lp->lp_rtrcredits < lp->lp_minrtrcredits)
 922                         lp->lp_minrtrcredits = lp->lp_rtrcredits;
 923
 924                 if (lp->lp_rtrcredits < 0) {
 925                         /* must have checked eager_recv before here */
 926                         LASSERT(msg->msg_rx_ready_delay);
 927                         msg->msg_rx_delayed = 1;
 928                         list_add_tail(&msg->msg_list, &lp->lp_rtrq);
 929                         return LNET_CREDIT_WAIT;
 930                 }
 931         }
 932
 933         rbp = lnet_msg2bufpool(msg);
 934
 935         if (!msg->msg_rtrcredit) {
 936                 msg->msg_rtrcredit = 1;
 937                 rbp->rbp_credits--;
 938                 if (rbp->rbp_credits < rbp->rbp_mincredits)
 939                         rbp->rbp_mincredits = rbp->rbp_credits;
 940
 941                 if (rbp->rbp_credits < 0) {
 942                         /* must have checked eager_recv before here */
 943                         LASSERT(msg->msg_rx_ready_delay);
 944                         msg->msg_rx_delayed = 1;
 945                         list_add_tail(&msg->msg_list, &rbp->rbp_msgs);
 946                         return LNET_CREDIT_WAIT;
 947                 }
 948         }
 949
 950         LASSERT(!list_empty(&rbp->rbp_bufs));
 951         rb = list_entry(rbp->rbp_bufs.next, lnet_rtrbuf_t, rb_list);
 952         list_del(&rb->rb_list);
 953
 954         msg->msg_niov = rbp->rbp_npages;
 955         msg->msg_kiov = &rb->rb_kiov[0];
 956
 957         if (do_recv) {
 958                 int cpt = msg->msg_rx_cpt;
 959
 960                 lnet_net_unlock(cpt);
 961                 lnet_ni_recv(lp->lp_ni, msg->msg_private, msg, 1,
 962                              0, msg->msg_len, msg->msg_len);
 963                 lnet_net_lock(cpt);
 964         }
 965         return LNET_CREDIT_OK;
 966 }
 967
 968 void
 969 lnet_return_tx_credits_locked(lnet_msg_t *msg)
 970 {
 971         lnet_peer_t     *txpeer = msg->msg_txpeer;
 972         lnet_msg_t      *msg2;
 973
 974         if (msg->msg_txcredit) {
 975                 struct lnet_ni       *ni = txpeer->lp_ni;
 976                 struct lnet_tx_queue *tq = ni->ni_tx_queues[msg->msg_tx_cpt];
 977
 978                 /* give back NI txcredits */
 979                 msg->msg_txcredit = 0;
 980
 981                 LASSERT((tq->tq_credits < 0) ==
 982                         !list_empty(&tq->tq_delayed));
 983
 984                 tq->tq_credits++;
 985                 if (tq->tq_credits <= 0) {
 986                         msg2 = list_entry(tq->tq_delayed.next,
 987                                           lnet_msg_t, msg_list);
 988                         list_del(&msg2->msg_list);
 989
 990                         LASSERT(msg2->msg_txpeer->lp_ni == ni);
 991                         LASSERT(msg2->msg_tx_delayed);
 992
 993                         (void) lnet_post_send_locked(msg2, 1);
 994                 }
 995         }
 996
 997         if (msg->msg_peertxcredit) {
 998                 /* give back peer txcredits */
 999                 msg->msg_peertxcredit = 0;
1000
1001                 LASSERT((txpeer->lp_txcredits < 0) ==
1002                         !list_empty(&txpeer->lp_txq));
1003
1004                 txpeer->lp_txqnob -= msg->msg_len + sizeof(lnet_hdr_t);
1005                 LASSERT (txpeer->lp_txqnob >= 0);
1006
1007                 txpeer->lp_txcredits++;
1008                 if (txpeer->lp_txcredits <= 0) {
1009                         msg2 = list_entry(txpeer->lp_txq.next,
1010                                               lnet_msg_t, msg_list);
1011                         list_del(&msg2->msg_list);
1012
1013                         LASSERT(msg2->msg_txpeer == txpeer);
1014                         LASSERT(msg2->msg_tx_delayed);
1015
1016                         (void) lnet_post_send_locked(msg2, 1);
1017                 }
1018         }
1019
1020         if (txpeer != NULL) {
1021                 msg->msg_txpeer = NULL;
1022                 lnet_peer_decref_locked(txpeer);
1023         }
1024 }
1025
1026 void
1027 lnet_schedule_blocked_locked(lnet_rtrbufpool_t *rbp)
1028 {
1029         lnet_msg_t      *msg;
1030
1031         if (list_empty(&rbp->rbp_msgs))
1032                 return;
1033         msg = list_entry(rbp->rbp_msgs.next,
1034                          lnet_msg_t, msg_list);
1035         list_del(&msg->msg_list);
1036
1037         (void)lnet_post_routed_recv_locked(msg, 1);
1038 }
1039
1040 void
1041 lnet_drop_routed_msgs_locked(struct list_head *list, int cpt)
1042 {
1043         lnet_msg_t       *msg;
1044         lnet_msg_t       *tmp;
1045         struct list_head drop;
1046
1047         INIT_LIST_HEAD(&drop);
1048
1049         list_splice_init(list, &drop);
1050
1051         lnet_net_unlock(cpt);
1052
1053         list_for_each_entry_safe(msg, tmp, &drop, msg_list) {
1054                 lnet_ni_recv(msg->msg_rxpeer->lp_ni, msg->msg_private, NULL,
1055                              0, 0, 0, msg->msg_hdr.payload_length);
1056                 list_del_init(&msg->msg_list);
1057                 lnet_finalize(NULL, msg, -ECANCELED);
1058         }
1059
1060         lnet_net_lock(cpt);
1061 }
1062
1063 void
1064 lnet_return_rx_credits_locked(lnet_msg_t *msg)
1065 {
1066         lnet_peer_t     *rxpeer = msg->msg_rxpeer;
1067         lnet_msg_t      *msg2;
1068
1069         if (msg->msg_rtrcredit) {
1070                 /* give back global router credits */
1071                 lnet_rtrbuf_t     *rb;
1072                 lnet_rtrbufpool_t *rbp;
1073
1074                 /* NB If a msg ever blocks for a buffer in rbp_msgs, it stays
1075                  * there until it gets one allocated, or aborts the wait
1076                  * itself */
1077                 LASSERT(msg->msg_kiov != NULL);
1078
1079                 rb = list_entry(msg->msg_kiov, lnet_rtrbuf_t, rb_kiov[0]);
1080                 rbp = rb->rb_pool;
1081
1082                 msg->msg_kiov = NULL;
1083                 msg->msg_rtrcredit = 0;
1084
1085                 LASSERT(rbp == lnet_msg2bufpool(msg));
1086
1087                 LASSERT((rbp->rbp_credits > 0) ==
1088                         !list_empty(&rbp->rbp_bufs));
1089
1090                 /* If routing is now turned off, we just drop this buffer and
1091                  * don't bother trying to return credits.  */
1092                 if (!the_lnet.ln_routing) {
1093                         lnet_destroy_rtrbuf(rb, rbp->rbp_npages);
1094                         goto routing_off;
1095                 }
1096
1097                 /* It is possible that a user has lowered the desired number of
1098                  * buffers in this pool.  Make sure we never put back
1099                  * more buffers than the stated number. */
1100                 if (unlikely(rbp->rbp_credits >= rbp->rbp_req_nbuffers)) {
1101                         /* Discard this buffer so we don't have too
1102                          * many. */
1103                         lnet_destroy_rtrbuf(rb, rbp->rbp_npages);
1104                         rbp->rbp_nbuffers--;
1105                 } else {
1106                         list_add(&rb->rb_list, &rbp->rbp_bufs);
1107                         rbp->rbp_credits++;
1108                         if (rbp->rbp_credits <= 0)
1109                                 lnet_schedule_blocked_locked(rbp);
1110                 }
1111         }
1112
1113 routing_off:
1114         if (msg->msg_peerrtrcredit) {
1115                 /* give back peer router credits */
1116                 msg->msg_peerrtrcredit = 0;
1117
1118                 LASSERT((rxpeer->lp_rtrcredits < 0) ==
1119                         !list_empty(&rxpeer->lp_rtrq));
1120
1121                 rxpeer->lp_rtrcredits++;
1122
1123                 /* drop all messages which are queued to be routed on that
1124                  * peer. */
1125                 if (!the_lnet.ln_routing) {
1126                         lnet_drop_routed_msgs_locked(&rxpeer->lp_rtrq,
1127                                                      msg->msg_rx_cpt);
1128                 } else if (rxpeer->lp_rtrcredits <= 0) {
1129                         msg2 = list_entry(rxpeer->lp_rtrq.next,
1130                                           lnet_msg_t, msg_list);
1131                         list_del(&msg2->msg_list);
1132
1133                         (void) lnet_post_routed_recv_locked(msg2, 1);
1134                 }
1135         }
1136         if (rxpeer != NULL) {
1137                 msg->msg_rxpeer = NULL;
1138                 lnet_peer_decref_locked(rxpeer);
1139         }
1140 }
1141
1142 static int
1143 lnet_compare_routes(lnet_route_t *r1, lnet_route_t *r2)
1144 {
1145         lnet_peer_t *p1 = r1->lr_gateway;
1146         lnet_peer_t *p2 = r2->lr_gateway;
1147         int r1_hops = (r1->lr_hops == LNET_UNDEFINED_HOPS) ? 1 : r1->lr_hops;
1148         int r2_hops = (r2->lr_hops == LNET_UNDEFINED_HOPS) ? 1 : r2->lr_hops;
1149
1150         if (r1->lr_priority < r2->lr_priority)
1151                 return 1;
1152
1153         if (r1->lr_priority > r2->lr_priority)
1154                 return -ERANGE;
1155
1156         if (r1_hops < r2_hops)
1157                 return 1;
1158
1159         if (r1_hops > r2_hops)
1160                 return -ERANGE;
1161
1162         if (p1->lp_txqnob < p2->lp_txqnob)
1163                 return 1;
1164
1165         if (p1->lp_txqnob > p2->lp_txqnob)
1166                 return -ERANGE;
1167
1168         if (p1->lp_txcredits > p2->lp_txcredits)
1169                 return 1;
1170
1171         if (p1->lp_txcredits < p2->lp_txcredits)
1172                 return -ERANGE;
1173
1174         if (r1->lr_seq - r2->lr_seq <= 0)
1175                 return 1;
1176
1177         return -ERANGE;
1178 }
1179
1180 static lnet_peer_t *
1181 lnet_find_route_locked(lnet_ni_t *ni, lnet_nid_t target, lnet_nid_t rtr_nid)
1182 {
1183         lnet_remotenet_t        *rnet;
1184         lnet_route_t            *route;
1185         lnet_route_t            *best_route;
1186         lnet_route_t            *last_route;
1187         struct lnet_peer        *lp_best;
1188         struct lnet_peer        *lp;
1189         int                     rc;
1190
1191         /* If @rtr_nid is not LNET_NID_ANY, return the gateway with
1192          * rtr_nid nid, otherwise find the best gateway I can use */
1193
1194         rnet = lnet_find_net_locked(LNET_NIDNET(target));
1195         if (rnet == NULL)
1196                 return NULL;
1197
1198         lp_best = NULL;
1199         best_route = last_route = NULL;
1200         list_for_each_entry(route, &rnet->lrn_routes, lr_list) {
1201                 lp = route->lr_gateway;
1202
1203                 if (!lnet_is_route_alive(route))
1204                         continue;
1205
1206                 if (ni != NULL && lp->lp_ni != ni)
1207                         continue;
1208
1209                 if (lp->lp_nid == rtr_nid) /* it's pre-determined router */
1210                         return lp;
1211
1212                 if (lp_best == NULL) {
1213                         best_route = last_route = route;
1214                         lp_best = lp;
1215                         continue;
1216                 }
1217
1218                 /* no protection on below fields, but it's harmless */
1219                 if (last_route->lr_seq - route->lr_seq < 0)
1220                         last_route = route;
1221
1222                 rc = lnet_compare_routes(route, best_route);
1223                 if (rc < 0)
1224                         continue;
1225
1226                 best_route = route;
1227                 lp_best = lp;
1228         }
1229
1230         /* set sequence number on the best router to the latest sequence + 1
1231          * so we can round-robin all routers, it's race and inaccurate but
1232          * harmless and functional  */
1233         if (best_route != NULL)
1234                 best_route->lr_seq = last_route->lr_seq + 1;
1235         return lp_best;
1236 }
1237
1238 int
1239 lnet_send(lnet_nid_t src_nid, lnet_msg_t *msg, lnet_nid_t rtr_nid)
1240 {
1241         lnet_nid_t              dst_nid = msg->msg_target.nid;
1242         struct lnet_ni          *src_ni;
1243         struct lnet_ni          *local_ni;
1244         struct lnet_peer        *lp;
1245         int                     cpt;
1246         int                     cpt2;
1247         int                     rc;
1248
1249         /* NB: rtr_nid is set to LNET_NID_ANY for all current use-cases,
1250          * but we might want to use pre-determined router for ACK/REPLY
1251          * in the future */
1252         /* NB: ni != NULL == interface pre-determined (ACK/REPLY) */
1253         LASSERT (msg->msg_txpeer == NULL);
1254         LASSERT (!msg->msg_sending);
1255         LASSERT (!msg->msg_target_is_router);
1256         LASSERT (!msg->msg_receiving);
1257
1258         msg->msg_sending = 1;
1259
1260         LASSERT(!msg->msg_tx_committed);
1261         cpt = lnet_cpt_of_nid(rtr_nid == LNET_NID_ANY ? dst_nid : rtr_nid);
1262  again:
1263         lnet_net_lock(cpt);
1264
1265         if (the_lnet.ln_shutdown) {
1266                 lnet_net_unlock(cpt);
1267                 return -ESHUTDOWN;
1268         }
1269
1270         if (src_nid == LNET_NID_ANY) {
1271                 src_ni = NULL;
1272         } else {
1273                 src_ni = lnet_nid2ni_locked(src_nid, cpt);
1274                 if (src_ni == NULL) {
1275                         lnet_net_unlock(cpt);
1276                         LCONSOLE_WARN("Can't send to %s: src %s is not a "
1277                                       "local nid\n", libcfs_nid2str(dst_nid),
1278                                       libcfs_nid2str(src_nid));
1279                         return -EINVAL;
1280                 }
1281                 LASSERT (!msg->msg_routing);
1282         }
1283
1284         /* Is this for someone on a local network? */
1285         local_ni = lnet_net2ni_locked(LNET_NIDNET(dst_nid), cpt);
1286
1287         if (local_ni != NULL) {
1288                 if (src_ni == NULL) {
1289                         src_ni = local_ni;
1290                         src_nid = src_ni->ni_nid;
1291                 } else if (src_ni == local_ni) {
1292                         lnet_ni_decref_locked(local_ni, cpt);
1293                 } else {
1294                         lnet_ni_decref_locked(local_ni, cpt);
1295                         lnet_ni_decref_locked(src_ni, cpt);
1296                         lnet_net_unlock(cpt);
1297                         LCONSOLE_WARN("No route to %s via from %s\n",
1298                                       libcfs_nid2str(dst_nid),
1299                                       libcfs_nid2str(src_nid));
1300                         return -EINVAL;
1301                 }
1302
1303                 LASSERT(src_nid != LNET_NID_ANY);
1304                 lnet_msg_commit(msg, cpt);
1305
1306                 if (!msg->msg_routing)
1307                         msg->msg_hdr.src_nid = cpu_to_le64(src_nid);
1308
1309                 if (src_ni == the_lnet.ln_loni) {
1310                         /* No send credit hassles with LOLND */
1311                         lnet_net_unlock(cpt);
1312                         lnet_ni_send(src_ni, msg);
1313
1314                         lnet_net_lock(cpt);
1315                         lnet_ni_decref_locked(src_ni, cpt);
1316                         lnet_net_unlock(cpt);
1317                         return 0;
1318                 }
1319
1320                 rc = lnet_nid2peer_locked(&lp, dst_nid, cpt);
1321                 /* lp has ref on src_ni; lose mine */
1322                 lnet_ni_decref_locked(src_ni, cpt);
1323                 if (rc != 0) {
1324                         lnet_net_unlock(cpt);
1325                         LCONSOLE_WARN("Error %d finding peer %s\n", rc,
1326                                       libcfs_nid2str(dst_nid));
1327                         /* ENOMEM or shutting down */
1328                         return rc;
1329                 }
1330                 LASSERT (lp->lp_ni == src_ni);
1331         } else {
1332                 /* sending to a remote network */
1333                 lp = lnet_find_route_locked(src_ni, dst_nid, rtr_nid);
1334                 if (lp == NULL) {
1335                         if (src_ni != NULL)
1336                                 lnet_ni_decref_locked(src_ni, cpt);
1337                         lnet_net_unlock(cpt);
1338
1339                         LCONSOLE_WARN("No route to %s via %s "
1340                                       "(all routers down)\n",
1341                                       libcfs_id2str(msg->msg_target),
1342                                       libcfs_nid2str(src_nid));
1343                         return -EHOSTUNREACH;
1344                 }
1345
1346                 /* rtr_nid is LNET_NID_ANY or NID of pre-determined router,
1347                  * it's possible that rtr_nid isn't LNET_NID_ANY and lp isn't
1348                  * pre-determined router, this can happen if router table
1349                  * was changed when we release the lock */
1350                 if (rtr_nid != lp->lp_nid) {
1351                         cpt2 = lnet_cpt_of_nid_locked(lp->lp_nid);
1352                         if (cpt2 != cpt) {
1353                                 if (src_ni != NULL)
1354                                         lnet_ni_decref_locked(src_ni, cpt);
1355                                 lnet_net_unlock(cpt);
1356
1357                                 rtr_nid = lp->lp_nid;
1358                                 cpt = cpt2;
1359                                 goto again;
1360                         }
1361                 }
1362
1363                 CDEBUG(D_NET, "Best route to %s via %s for %s %d\n",
1364                        libcfs_nid2str(dst_nid), libcfs_nid2str(lp->lp_nid),
1365                        lnet_msgtyp2str(msg->msg_type), msg->msg_len);
1366
1367                 if (src_ni == NULL) {
1368                         src_ni = lp->lp_ni;
1369                         src_nid = src_ni->ni_nid;
1370                 } else {
1371                         LASSERT (src_ni == lp->lp_ni);
1372                         lnet_ni_decref_locked(src_ni, cpt);
1373                 }
1374
1375                 lnet_peer_addref_locked(lp);
1376
1377                 LASSERT(src_nid != LNET_NID_ANY);
1378                 lnet_msg_commit(msg, cpt);
1379
1380                 if (!msg->msg_routing) {
1381                         /* I'm the source and now I know which NI to send on */
1382                         msg->msg_hdr.src_nid = cpu_to_le64(src_nid);
1383                 }
1384
1385                 msg->msg_target_is_router = 1;
1386                 msg->msg_target.nid = lp->lp_nid;
1387                 msg->msg_target.pid = LNET_PID_LUSTRE;
1388         }
1389
1390         /* 'lp' is our best choice of peer */
1391
1392         LASSERT (!msg->msg_peertxcredit);
1393         LASSERT (!msg->msg_txcredit);
1394         LASSERT (msg->msg_txpeer == NULL);
1395
1396         msg->msg_txpeer = lp;                   /* msg takes my ref on lp */
1397
1398         rc = lnet_post_send_locked(msg, 0);
1399         lnet_net_unlock(cpt);
1400
1401         if (rc < 0)
1402                 return rc;
1403
1404         if (rc == LNET_CREDIT_OK)
1405                 lnet_ni_send(src_ni, msg);
1406
1407         return 0; /* rc == LNET_CREDIT_OK or LNET_CREDIT_WAIT */
1408 }
1409
1410 void
1411 lnet_drop_message(lnet_ni_t *ni, int cpt, void *private, unsigned int nob)
1412 {
1413         lnet_net_lock(cpt);
1414         the_lnet.ln_counters[cpt]->drop_count++;
1415         the_lnet.ln_counters[cpt]->drop_length += nob;
1416         lnet_net_unlock(cpt);
1417
1418         lnet_ni_recv(ni, private, NULL, 0, 0, 0, nob);
1419 }
1420
1421 static void
1422 lnet_recv_put(lnet_ni_t *ni, lnet_msg_t *msg)
1423 {
1424         lnet_hdr_t      *hdr = &msg->msg_hdr;
1425
1426         if (msg->msg_wanted != 0)
1427                 lnet_setpayloadbuffer(msg);
1428
1429         lnet_build_msg_event(msg, LNET_EVENT_PUT);
1430
1431         /* Must I ACK?  If so I'll grab the ack_wmd out of the header and put
1432          * it back into the ACK during lnet_finalize() */
1433         msg->msg_ack = (!lnet_is_wire_handle_none(&hdr->msg.put.ack_wmd) &&
1434                         (msg->msg_md->md_options & LNET_MD_ACK_DISABLE) == 0);
1435
1436         lnet_ni_recv(ni, msg->msg_private, msg, msg->msg_rx_delayed,
1437                      msg->msg_offset, msg->msg_wanted, hdr->payload_length);
1438 }
1439
1440 static int
1441 lnet_parse_put(lnet_ni_t *ni, lnet_msg_t *msg)
1442 {
1443         lnet_hdr_t              *hdr = &msg->msg_hdr;
1444         struct lnet_match_info  info;
1445         int                     rc;
1446         bool                    ready_delay;
1447
1448         /* Convert put fields to host byte order */
1449         hdr->msg.put.match_bits = le64_to_cpu(hdr->msg.put.match_bits);
1450         hdr->msg.put.ptl_index  = le32_to_cpu(hdr->msg.put.ptl_index);
1451         hdr->msg.put.offset     = le32_to_cpu(hdr->msg.put.offset);
1452
1453         info.mi_id.nid  = hdr->src_nid;
1454         info.mi_id.pid  = hdr->src_pid;
1455         info.mi_opc     = LNET_MD_OP_PUT;
1456         info.mi_portal  = hdr->msg.put.ptl_index;
1457         info.mi_rlength = hdr->payload_length;
1458         info.mi_roffset = hdr->msg.put.offset;
1459         info.mi_mbits   = hdr->msg.put.match_bits;
1460
1461         msg->msg_rx_ready_delay = ni->ni_lnd->lnd_eager_recv == NULL;
1462         ready_delay = msg->msg_rx_ready_delay;
1463
1464  again:
1465         rc = lnet_ptl_match_md(&info, msg);
1466         switch (rc) {
1467         default:
1468                 LBUG();
1469
1470         case LNET_MATCHMD_OK:
1471                 lnet_recv_put(ni, msg);
1472                 return 0;
1473
1474         case LNET_MATCHMD_NONE:
1475                 if (ready_delay)
1476                         /* no eager_recv or has already called it, should
1477                          * have been attached on delayed list */
1478                         return 0;
1479
1480                 rc = lnet_ni_eager_recv(ni, msg);
1481                 if (rc == 0) {
1482                         ready_delay = true;
1483                         goto again;
1484                 }
1485                 /* fall through */
1486
1487         case LNET_MATCHMD_DROP:
1488                 CNETERR("Dropping PUT from %s portal %d match %llu"
1489                         " offset %d length %d: %d\n",
1490                         libcfs_id2str(info.mi_id), info.mi_portal,
1491                         info.mi_mbits, info.mi_roffset, info.mi_rlength, rc);
1492
1493                 return -ENOENT; /* -ve: OK but no match */
1494         }
1495 }
1496
1497 static int
1498 lnet_parse_get(lnet_ni_t *ni, lnet_msg_t *msg, int rdma_get)
1499 {
1500         struct lnet_match_info  info;
1501         lnet_hdr_t              *hdr = &msg->msg_hdr;
1502         lnet_handle_wire_t      reply_wmd;
1503         int                     rc;
1504
1505         /* Convert get fields to host byte order */
1506         hdr->msg.get.match_bits   = le64_to_cpu(hdr->msg.get.match_bits);
1507         hdr->msg.get.ptl_index    = le32_to_cpu(hdr->msg.get.ptl_index);
1508         hdr->msg.get.sink_length  = le32_to_cpu(hdr->msg.get.sink_length);
1509         hdr->msg.get.src_offset   = le32_to_cpu(hdr->msg.get.src_offset);
1510
1511         info.mi_id.nid  = hdr->src_nid;
1512         info.mi_id.pid  = hdr->src_pid;
1513         info.mi_opc     = LNET_MD_OP_GET;
1514         info.mi_portal  = hdr->msg.get.ptl_index;
1515         info.mi_rlength = hdr->msg.get.sink_length;
1516         info.mi_roffset = hdr->msg.get.src_offset;
1517         info.mi_mbits   = hdr->msg.get.match_bits;
1518
1519         rc = lnet_ptl_match_md(&info, msg);
1520         if (rc == LNET_MATCHMD_DROP) {
1521                 CNETERR("Dropping GET from %s portal %d match %llu"
1522                         " offset %d length %d\n",
1523                         libcfs_id2str(info.mi_id), info.mi_portal,
1524                         info.mi_mbits, info.mi_roffset, info.mi_rlength);
1525                 return -ENOENT; /* -ve: OK but no match */
1526         }
1527
1528         LASSERT(rc == LNET_MATCHMD_OK);
1529
1530         lnet_build_msg_event(msg, LNET_EVENT_GET);
1531
1532         reply_wmd = hdr->msg.get.return_wmd;
1533
1534         lnet_prep_send(msg, LNET_MSG_REPLY, info.mi_id,
1535                        msg->msg_offset, msg->msg_wanted);
1536
1537         msg->msg_hdr.msg.reply.dst_wmd = reply_wmd;
1538
1539         if (rdma_get) {
1540                 /* The LND completes the REPLY from her recv procedure */
1541                 lnet_ni_recv(ni, msg->msg_private, msg, 0,
1542                              msg->msg_offset, msg->msg_len, msg->msg_len);
1543                 return 0;
1544         }
1545
1546         lnet_ni_recv(ni, msg->msg_private, NULL, 0, 0, 0, 0);
1547         msg->msg_receiving = 0;
1548
1549         rc = lnet_send(ni->ni_nid, msg, LNET_NID_ANY);
1550         if (rc < 0) {
1551                 /* didn't get as far as lnet_ni_send() */
1552                 CERROR("%s: Unable to send REPLY for GET from %s: %d\n",
1553                        libcfs_nid2str(ni->ni_nid),
1554                        libcfs_id2str(info.mi_id), rc);
1555
1556                 lnet_finalize(ni, msg, rc);
1557         }
1558
1559         return 0;
1560 }
1561
1562 static int
1563 lnet_parse_reply(lnet_ni_t *ni, lnet_msg_t *msg)
1564 {
1565         void             *private = msg->msg_private;
1566         lnet_hdr_t       *hdr = &msg->msg_hdr;
1567         lnet_process_id_t src = {0};
1568         lnet_libmd_t     *md;
1569         int               rlength;
1570         int               mlength;
1571         int                     cpt;
1572
1573         cpt = lnet_cpt_of_cookie(hdr->msg.reply.dst_wmd.wh_object_cookie);
1574         lnet_res_lock(cpt);
1575
1576         src.nid = hdr->src_nid;
1577         src.pid = hdr->src_pid;
1578
1579         /* NB handles only looked up by creator (no flips) */
1580         md = lnet_wire_handle2md(&hdr->msg.reply.dst_wmd);
1581         if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
1582                 CNETERR("%s: Dropping REPLY from %s for %s "
1583                         "MD %#llx.%#llx\n",
1584                         libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
1585                         (md == NULL) ? "invalid" : "inactive",
1586                         hdr->msg.reply.dst_wmd.wh_interface_cookie,
1587                         hdr->msg.reply.dst_wmd.wh_object_cookie);
1588                 if (md != NULL && md->md_me != NULL)
1589                         CERROR("REPLY MD also attached to portal %d\n",
1590                                md->md_me->me_portal);
1591
1592                 lnet_res_unlock(cpt);
1593                 return -ENOENT; /* -ve: OK but no match */
1594         }
1595
1596         LASSERT (md->md_offset == 0);
1597
1598         rlength = hdr->payload_length;
1599         mlength = MIN(rlength, (int)md->md_length);
1600
1601         if (mlength < rlength &&
1602             (md->md_options & LNET_MD_TRUNCATE) == 0) {
1603                 CNETERR("%s: Dropping REPLY from %s length %d "
1604                         "for MD %#llx would overflow (%d)\n",
1605                         libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
1606                         rlength, hdr->msg.reply.dst_wmd.wh_object_cookie,
1607                         mlength);
1608                 lnet_res_unlock(cpt);
1609                 return -ENOENT; /* -ve: OK but no match */
1610         }
1611
1612         CDEBUG(D_NET, "%s: Reply from %s of length %d/%d into md %#llx\n",
1613                libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
1614                mlength, rlength, hdr->msg.reply.dst_wmd.wh_object_cookie);
1615
1616         lnet_msg_attach_md(msg, md, 0, mlength);
1617
1618         if (mlength != 0)
1619                 lnet_setpayloadbuffer(msg);
1620
1621         lnet_res_unlock(cpt);
1622
1623         lnet_build_msg_event(msg, LNET_EVENT_REPLY);
1624
1625         lnet_ni_recv(ni, private, msg, 0, 0, mlength, rlength);
1626         return 0;
1627 }
1628
1629 static int
1630 lnet_parse_ack(lnet_ni_t *ni, lnet_msg_t *msg)
1631 {
1632         lnet_hdr_t       *hdr = &msg->msg_hdr;
1633         lnet_process_id_t src = {0};
1634         lnet_libmd_t     *md;
1635         int                     cpt;
1636
1637         src.nid = hdr->src_nid;
1638         src.pid = hdr->src_pid;
1639
1640         /* Convert ack fields to host byte order */
1641         hdr->msg.ack.match_bits = le64_to_cpu(hdr->msg.ack.match_bits);
1642         hdr->msg.ack.mlength = le32_to_cpu(hdr->msg.ack.mlength);
1643
1644         cpt = lnet_cpt_of_cookie(hdr->msg.ack.dst_wmd.wh_object_cookie);
1645         lnet_res_lock(cpt);
1646
1647         /* NB handles only looked up by creator (no flips) */
1648         md = lnet_wire_handle2md(&hdr->msg.ack.dst_wmd);
1649         if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
1650                 /* Don't moan; this is expected */
1651                 CDEBUG(D_NET,
1652                        "%s: Dropping ACK from %s to %s MD %#llx.%#llx\n",
1653                        libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
1654                        (md == NULL) ? "invalid" : "inactive",
1655                        hdr->msg.ack.dst_wmd.wh_interface_cookie,
1656                        hdr->msg.ack.dst_wmd.wh_object_cookie);
1657                 if (md != NULL && md->md_me != NULL)
1658                         CERROR("Source MD also attached to portal %d\n",
1659                                md->md_me->me_portal);
1660
1661                 lnet_res_unlock(cpt);
1662                 return -ENOENT;                  /* -ve! */
1663         }
1664
1665         CDEBUG(D_NET, "%s: ACK from %s into md %#llx\n",
1666                libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
1667                hdr->msg.ack.dst_wmd.wh_object_cookie);
1668
1669         lnet_msg_attach_md(msg, md, 0, 0);
1670
1671         lnet_res_unlock(cpt);
1672
1673         lnet_build_msg_event(msg, LNET_EVENT_ACK);
1674
1675         lnet_ni_recv(ni, msg->msg_private, msg, 0, 0, 0, msg->msg_len);
1676         return 0;
1677 }
1678
1679 /**
1680  * \retval LNET_CREDIT_OK       If \a msg is forwarded
1681  * \retval LNET_CREDIT_WAIT     If \a msg is blocked because w/o buffer
1682  * \retval -ve                  error code
1683  */
1684 int
1685 lnet_parse_forward_locked(lnet_ni_t *ni, lnet_msg_t *msg)
1686 {
1687         int     rc = 0;
1688
1689         if (!the_lnet.ln_routing)
1690                 return -ECANCELED;
1691
1692         if (msg->msg_rxpeer->lp_rtrcredits <= 0 ||
1693             lnet_msg2bufpool(msg)->rbp_credits <= 0) {
1694                 if (ni->ni_lnd->lnd_eager_recv == NULL) {
1695                         msg->msg_rx_ready_delay = 1;
1696                 } else {
1697                         lnet_net_unlock(msg->msg_rx_cpt);
1698                         rc = lnet_ni_eager_recv(ni, msg);
1699                         lnet_net_lock(msg->msg_rx_cpt);
1700                 }
1701         }
1702
1703         if (rc == 0)
1704                 rc = lnet_post_routed_recv_locked(msg, 0);
1705         return rc;
1706 }
1707
1708 int
1709 lnet_parse_local(lnet_ni_t *ni, lnet_msg_t *msg)
1710 {
1711         int     rc;
1712
1713         switch (msg->msg_type) {
1714         case LNET_MSG_ACK:
1715                 rc = lnet_parse_ack(ni, msg);
1716                 break;
1717         case LNET_MSG_PUT:
1718                 rc = lnet_parse_put(ni, msg);
1719                 break;
1720         case LNET_MSG_GET:
1721                 rc = lnet_parse_get(ni, msg, msg->msg_rdma_get);
1722                 break;
1723         case LNET_MSG_REPLY:
1724                 rc = lnet_parse_reply(ni, msg);
1725                 break;
1726         default: /* prevent an unused label if !kernel */
1727                 LASSERT(0);
1728                 return -EPROTO;
1729         }
1730
1731         LASSERT(rc == 0 || rc == -ENOENT);
1732         return rc;
1733 }
1734
1735 char *
1736 lnet_msgtyp2str (int type)
1737 {
1738         switch (type) {
1739         case LNET_MSG_ACK:
1740                 return ("ACK");
1741         case LNET_MSG_PUT:
1742                 return ("PUT");
1743         case LNET_MSG_GET:
1744                 return ("GET");
1745         case LNET_MSG_REPLY:
1746                 return ("REPLY");
1747         case LNET_MSG_HELLO:
1748                 return ("HELLO");
1749         default:
1750                 return ("<UNKNOWN>");
1751         }
1752 }
1753
1754 void
1755 lnet_print_hdr(lnet_hdr_t * hdr)
1756 {
1757         lnet_process_id_t src = {0};
1758         lnet_process_id_t dst = {0};
1759         char *type_str = lnet_msgtyp2str (hdr->type);
1760
1761         src.nid = hdr->src_nid;
1762         src.pid = hdr->src_pid;
1763
1764         dst.nid = hdr->dest_nid;
1765         dst.pid = hdr->dest_pid;
1766
1767         CWARN("P3 Header at %p of type %s\n", hdr, type_str);
1768         CWARN("    From %s\n", libcfs_id2str(src));
1769         CWARN("    To   %s\n", libcfs_id2str(dst));
1770
1771         switch (hdr->type) {
1772         default:
1773                 break;
1774
1775         case LNET_MSG_PUT:
1776                 CWARN("    Ptl index %d, ack md %#llx.%#llx, "
1777                       "match bits %llu\n",
1778                       hdr->msg.put.ptl_index,
1779                       hdr->msg.put.ack_wmd.wh_interface_cookie,
1780                       hdr->msg.put.ack_wmd.wh_object_cookie,
1781                       hdr->msg.put.match_bits);
1782                 CWARN("    Length %d, offset %d, hdr data %#llx\n",
1783                       hdr->payload_length, hdr->msg.put.offset,
1784                       hdr->msg.put.hdr_data);
1785                 break;
1786
1787         case LNET_MSG_GET:
1788                 CWARN("    Ptl index %d, return md %#llx.%#llx, "
1789                       "match bits %llu\n", hdr->msg.get.ptl_index,
1790                       hdr->msg.get.return_wmd.wh_interface_cookie,
1791                       hdr->msg.get.return_wmd.wh_object_cookie,
1792                       hdr->msg.get.match_bits);
1793                 CWARN("    Length %d, src offset %d\n",
1794                       hdr->msg.get.sink_length,
1795                       hdr->msg.get.src_offset);
1796                 break;
1797
1798         case LNET_MSG_ACK:
1799                 CWARN("    dst md %#llx.%#llx, "
1800                       "manipulated length %d\n",
1801                       hdr->msg.ack.dst_wmd.wh_interface_cookie,
1802                       hdr->msg.ack.dst_wmd.wh_object_cookie,
1803                       hdr->msg.ack.mlength);
1804                 break;
1805
1806         case LNET_MSG_REPLY:
1807                 CWARN("    dst md %#llx.%#llx, "
1808                       "length %d\n",
1809                       hdr->msg.reply.dst_wmd.wh_interface_cookie,
1810                       hdr->msg.reply.dst_wmd.wh_object_cookie,
1811                       hdr->payload_length);
1812         }
1813
1814 }
1815
1816 int
1817 lnet_parse(lnet_ni_t *ni, lnet_hdr_t *hdr, lnet_nid_t from_nid,
1818            void *private, int rdma_req)
1819 {
1820         int             rc = 0;
1821         int             cpt;
1822         int             for_me;
1823         struct lnet_msg *msg;
1824         lnet_pid_t     dest_pid;
1825         lnet_nid_t     dest_nid;
1826         lnet_nid_t     src_nid;
1827         __u32          payload_length;
1828         __u32          type;
1829
1830         LASSERT (!in_interrupt ());
1831
1832         type = le32_to_cpu(hdr->type);
1833         src_nid = le64_to_cpu(hdr->src_nid);
1834         dest_nid = le64_to_cpu(hdr->dest_nid);
1835         dest_pid = le32_to_cpu(hdr->dest_pid);
1836         payload_length = le32_to_cpu(hdr->payload_length);
1837
1838         for_me = (ni->ni_nid == dest_nid);
1839         cpt = lnet_cpt_of_nid(from_nid);
1840
1841         switch (type) {
1842         case LNET_MSG_ACK:
1843         case LNET_MSG_GET:
1844                 if (payload_length > 0) {
1845                         CERROR("%s, src %s: bad %s payload %d (0 expected)\n",
1846                                libcfs_nid2str(from_nid),
1847                                libcfs_nid2str(src_nid),
1848                                lnet_msgtyp2str(type), payload_length);
1849                         return -EPROTO;
1850                 }
1851                 break;
1852
1853         case LNET_MSG_PUT:
1854         case LNET_MSG_REPLY:
1855                 if (payload_length >
1856                     (__u32)(for_me ? LNET_MAX_PAYLOAD : LNET_MTU)) {
1857                         CERROR("%s, src %s: bad %s payload %d "
1858                                "(%d max expected)\n",
1859                                libcfs_nid2str(from_nid),
1860                                libcfs_nid2str(src_nid),
1861                                lnet_msgtyp2str(type),
1862                                payload_length,
1863                                for_me ? LNET_MAX_PAYLOAD : LNET_MTU);
1864                         return -EPROTO;
1865                 }
1866                 break;
1867
1868         default:
1869                 CERROR("%s, src %s: Bad message type 0x%x\n",
1870                        libcfs_nid2str(from_nid),
1871                        libcfs_nid2str(src_nid), type);
1872                 return -EPROTO;
1873         }
1874
1875         if (the_lnet.ln_routing &&
1876             ni->ni_last_alive != cfs_time_current_sec()) {
1877                 /* NB: so far here is the only place to set NI status to "up */
1878                 lnet_ni_lock(ni);
1879                 ni->ni_last_alive = cfs_time_current_sec();
1880                 if (ni->ni_status != NULL &&
1881                     ni->ni_status->ns_status == LNET_NI_STATUS_DOWN)
1882                         ni->ni_status->ns_status = LNET_NI_STATUS_UP;
1883                 lnet_ni_unlock(ni);
1884         }
1885
1886         /* Regard a bad destination NID as a protocol error.  Senders should
1887          * know what they're doing; if they don't they're misconfigured, buggy
1888          * or malicious so we chop them off at the knees :) */
1889
1890         if (!for_me) {
1891                 if (LNET_NIDNET(dest_nid) == LNET_NIDNET(ni->ni_nid)) {
1892                         /* should have gone direct */
1893                         CERROR("%s, src %s: Bad dest nid %s "
1894                                "(should have been sent direct)\n",
1895                                 libcfs_nid2str(from_nid),
1896                                 libcfs_nid2str(src_nid),
1897                                 libcfs_nid2str(dest_nid));
1898                         return -EPROTO;
1899                 }
1900
1901                 if (lnet_islocalnid(dest_nid)) {
1902                         /* dest is another local NI; sender should have used
1903                          * this node's NID on its own network */
1904                         CERROR("%s, src %s: Bad dest nid %s "
1905                                "(it's my nid but on a different network)\n",
1906                                 libcfs_nid2str(from_nid),
1907                                 libcfs_nid2str(src_nid),
1908                                 libcfs_nid2str(dest_nid));
1909                         return -EPROTO;
1910                 }
1911
1912                 if (rdma_req && type == LNET_MSG_GET) {
1913                         CERROR("%s, src %s: Bad optimized GET for %s "
1914                                "(final destination must be me)\n",
1915                                 libcfs_nid2str(from_nid),
1916                                 libcfs_nid2str(src_nid),
1917                                 libcfs_nid2str(dest_nid));
1918                         return -EPROTO;
1919                 }
1920
1921                 if (!the_lnet.ln_routing) {
1922                         CERROR("%s, src %s: Dropping message for %s "
1923                                "(routing not enabled)\n",
1924                                 libcfs_nid2str(from_nid),
1925                                 libcfs_nid2str(src_nid),
1926                                 libcfs_nid2str(dest_nid));
1927                         goto drop;
1928                 }
1929         }
1930
1931         /* Message looks OK; we're not going to return an error, so we MUST
1932          * call back lnd_recv() come what may... */
1933
1934         if (!list_empty(&the_lnet.ln_test_peers) &&     /* normally we don't */
1935             fail_peer(src_nid, 0)) {                    /* shall we now? */
1936                 CERROR("%s, src %s: Dropping %s to simulate failure\n",
1937                        libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
1938                        lnet_msgtyp2str(type));
1939                 goto drop;
1940         }
1941
1942         if (!list_empty(&the_lnet.ln_drop_rules) &&
1943             lnet_drop_rule_match(hdr)) {
1944                 CDEBUG(D_NET, "%s, src %s, dst %s: Dropping %s to simulate"
1945                               "silent message loss\n",
1946                        libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
1947                        libcfs_nid2str(dest_nid), lnet_msgtyp2str(type));
1948                 goto drop;
1949         }
1950
1951
1952         msg = lnet_msg_alloc();
1953         if (msg == NULL) {
1954                 CERROR("%s, src %s: Dropping %s (out of memory)\n",
1955                        libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
1956                        lnet_msgtyp2str(type));
1957                 goto drop;
1958         }
1959
1960         /* msg zeroed in lnet_msg_alloc; i.e. flags all clear,
1961          * pointers NULL etc */
1962
1963         msg->msg_type = type;
1964         msg->msg_private = private;
1965         msg->msg_receiving = 1;
1966         msg->msg_rdma_get = rdma_req;
1967         msg->msg_len = msg->msg_wanted = payload_length;
1968         msg->msg_offset = 0;
1969         msg->msg_hdr = *hdr;
1970         /* for building message event */
1971         msg->msg_from = from_nid;
1972         if (!for_me) {
1973                 msg->msg_target.pid     = dest_pid;
1974                 msg->msg_target.nid     = dest_nid;
1975                 msg->msg_routing        = 1;
1976
1977         } else {
1978                 /* convert common msg->hdr fields to host byteorder */
1979                 msg->msg_hdr.type       = type;
1980                 msg->msg_hdr.src_nid    = src_nid;
1981                 msg->msg_hdr.src_pid    = le32_to_cpu(msg->msg_hdr.src_pid);
1982                 msg->msg_hdr.dest_nid   = dest_nid;
1983                 msg->msg_hdr.dest_pid   = dest_pid;
1984                 msg->msg_hdr.payload_length = payload_length;
1985         }
1986
1987         lnet_net_lock(cpt);
1988         rc = lnet_nid2peer_locked(&msg->msg_rxpeer, from_nid, cpt);
1989         if (rc != 0) {
1990                 lnet_net_unlock(cpt);
1991                 CERROR("%s, src %s: Dropping %s "
1992                        "(error %d looking up sender)\n",
1993                        libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
1994                        lnet_msgtyp2str(type), rc);
1995                 lnet_msg_free(msg);
1996                 if (rc == -ESHUTDOWN)
1997                         /* We are shutting down.  Don't do anything more */
1998                         return 0;
1999                 goto drop;
2000         }
2001
2002         if (lnet_isrouter(msg->msg_rxpeer)) {
2003                 lnet_peer_set_alive(msg->msg_rxpeer);
2004                 if (avoid_asym_router_failure &&
2005                     LNET_NIDNET(src_nid) != LNET_NIDNET(from_nid)) {
2006                         /* received a remote message from router, update
2007                          * remote NI status on this router.
2008                          * NB: multi-hop routed message will be ignored.
2009                          */
2010                         lnet_router_ni_update_locked(msg->msg_rxpeer,
2011                                                      LNET_NIDNET(src_nid));
2012                 }
2013         }
2014
2015         lnet_msg_commit(msg, cpt);
2016
2017         /* message delay simulation */
2018         if (unlikely(!list_empty(&the_lnet.ln_delay_rules) &&
2019                      lnet_delay_rule_match_locked(hdr, msg))) {
2020                 lnet_net_unlock(cpt);
2021                 return 0;
2022         }
2023
2024         if (!for_me) {
2025                 rc = lnet_parse_forward_locked(ni, msg);
2026                 lnet_net_unlock(cpt);
2027
2028                 if (rc < 0)
2029                         goto free_drop;
2030
2031                 if (rc == LNET_CREDIT_OK) {
2032                         lnet_ni_recv(ni, msg->msg_private, msg, 0,
2033                                      0, payload_length, payload_length);
2034                 }
2035                 return 0;
2036         }
2037
2038         lnet_net_unlock(cpt);
2039
2040         rc = lnet_parse_local(ni, msg);
2041         if (rc != 0)
2042                 goto free_drop;
2043         return 0;
2044
2045  free_drop:
2046         LASSERT(msg->msg_md == NULL);
2047         lnet_finalize(ni, msg, rc);
2048
2049  drop:
2050         lnet_drop_message(ni, cpt, private, payload_length);
2051         return 0;
2052 }
2053 EXPORT_SYMBOL(lnet_parse);
2054
2055 void
2056 lnet_drop_delayed_msg_list(struct list_head *head, char *reason)
2057 {
2058         while (!list_empty(head)) {
2059                 lnet_process_id_t       id = {0};
2060                 lnet_msg_t              *msg;
2061
2062                 msg = list_entry(head->next, lnet_msg_t, msg_list);
2063                 list_del(&msg->msg_list);
2064
2065                 id.nid = msg->msg_hdr.src_nid;
2066                 id.pid = msg->msg_hdr.src_pid;
2067
2068                 LASSERT(msg->msg_md == NULL);
2069                 LASSERT(msg->msg_rx_delayed);
2070                 LASSERT(msg->msg_rxpeer != NULL);
2071                 LASSERT(msg->msg_hdr.type == LNET_MSG_PUT);
2072
2073                 CWARN("Dropping delayed PUT from %s portal %d match %llu"
2074                       " offset %d length %d: %s\n",
2075                       libcfs_id2str(id),
2076                       msg->msg_hdr.msg.put.ptl_index,
2077                       msg->msg_hdr.msg.put.match_bits,
2078                       msg->msg_hdr.msg.put.offset,
2079                       msg->msg_hdr.payload_length, reason);
2080
2081                 /* NB I can't drop msg's ref on msg_rxpeer until after I've
2082                  * called lnet_drop_message(), so I just hang onto msg as well
2083                  * until that's done */
2084
2085                 lnet_drop_message(msg->msg_rxpeer->lp_ni,
2086                                   msg->msg_rxpeer->lp_cpt,
2087                                   msg->msg_private, msg->msg_len);
2088                 /*
2089                  * NB: message will not generate event because w/o attached MD,
2090                  * but we still should give error code so lnet_msg_decommit()
2091                  * can skip counters operations and other checks.
2092                  */
2093                 lnet_finalize(msg->msg_rxpeer->lp_ni, msg, -ENOENT);
2094         }
2095 }
2096
2097 void
2098 lnet_recv_delayed_msg_list(struct list_head *head)
2099 {
2100         while (!list_empty(head)) {
2101                 lnet_msg_t        *msg;
2102                 lnet_process_id_t  id;
2103
2104                 msg = list_entry(head->next, lnet_msg_t, msg_list);
2105                 list_del(&msg->msg_list);
2106
2107                 /* md won't disappear under me, since each msg
2108                  * holds a ref on it */
2109
2110                 id.nid = msg->msg_hdr.src_nid;
2111                 id.pid = msg->msg_hdr.src_pid;
2112
2113                 LASSERT(msg->msg_rx_delayed);
2114                 LASSERT(msg->msg_md != NULL);
2115                 LASSERT(msg->msg_rxpeer != NULL);
2116                 LASSERT(msg->msg_hdr.type == LNET_MSG_PUT);
2117
2118                 CDEBUG(D_NET, "Resuming delayed PUT from %s portal %d "
2119                        "match %llu offset %d length %d.\n",
2120                         libcfs_id2str(id), msg->msg_hdr.msg.put.ptl_index,
2121                         msg->msg_hdr.msg.put.match_bits,
2122                         msg->msg_hdr.msg.put.offset,
2123                         msg->msg_hdr.payload_length);
2124
2125                 lnet_recv_put(msg->msg_rxpeer->lp_ni, msg);
2126         }
2127 }
2128
2129 /**
2130  * Initiate an asynchronous PUT operation.
2131  *
2132  * There are several events associated with a PUT: completion of the send on
2133  * the initiator node (LNET_EVENT_SEND), and when the send completes
2134  * successfully, the receipt of an acknowledgment (LNET_EVENT_ACK) indicating
2135  * that the operation was accepted by the target. The event LNET_EVENT_PUT is
2136  * used at the target node to indicate the completion of incoming data
2137  * delivery.
2138  *
2139  * The local events will be logged in the EQ associated with the MD pointed to
2140  * by \a mdh handle. Using a MD without an associated EQ results in these
2141  * events being discarded. In this case, the caller must have another
2142  * mechanism (e.g., a higher level protocol) for determining when it is safe
2143  * to modify the memory region associated with the MD.
2144  *
2145  * Note that LNet does not guarantee the order of LNET_EVENT_SEND and
2146  * LNET_EVENT_ACK, though intuitively ACK should happen after SEND.
2147  *
2148  * \param self Indicates the NID of a local interface through which to send
2149  * the PUT request. Use LNET_NID_ANY to let LNet choose one by itself.
2150  * \param mdh A handle for the MD that describes the memory to be sent. The MD
2151  * must be "free floating" (See LNetMDBind()).
2152  * \param ack Controls whether an acknowledgment is requested.
2153  * Acknowledgments are only sent when they are requested by the initiating
2154  * process and the target MD enables them.
2155  * \param target A process identifier for the target process.
2156  * \param portal The index in the \a target's portal table.
2157  * \param match_bits The match bits to use for MD selection at the target
2158  * process.
2159  * \param offset The offset into the target MD (only used when the target
2160  * MD has the LNET_MD_MANAGE_REMOTE option set).
2161  * \param hdr_data 64 bits of user data that can be included in the message
2162  * header. This data is written to an event queue entry at the target if an
2163  * EQ is present on the matching MD.
2164  *
2165  * \retval  0      Success, and only in this case events will be generated
2166  * and logged to EQ (if it exists).
2167  * \retval -EIO    Simulated failure.
2168  * \retval -ENOMEM Memory allocation failure.
2169  * \retval -ENOENT Invalid MD object.
2170  *
2171  * \see lnet_event_t::hdr_data and lnet_event_kind_t.
2172  */
2173 int
2174 LNetPut(lnet_nid_t self, lnet_handle_md_t mdh, lnet_ack_req_t ack,
2175         lnet_process_id_t target, unsigned int portal,
2176         __u64 match_bits, unsigned int offset,
2177         __u64 hdr_data)
2178 {
2179         struct lnet_msg         *msg;
2180         struct lnet_libmd       *md;
2181         int                     cpt;
2182         int                     rc;
2183
2184         LASSERT(the_lnet.ln_refcount > 0);
2185
2186         if (!list_empty(&the_lnet.ln_test_peers) &&     /* normally we don't */
2187             fail_peer(target.nid, 1)) {                 /* shall we now? */
2188                 CERROR("Dropping PUT to %s: simulated failure\n",
2189                        libcfs_id2str(target));
2190                 return -EIO;
2191         }
2192
2193         msg = lnet_msg_alloc();
2194         if (msg == NULL) {
2195                 CERROR("Dropping PUT to %s: ENOMEM on lnet_msg_t\n",
2196                        libcfs_id2str(target));
2197                 return -ENOMEM;
2198         }
2199         msg->msg_vmflush = !!memory_pressure_get();
2200
2201         cpt = lnet_cpt_of_cookie(mdh.cookie);
2202         lnet_res_lock(cpt);
2203
2204         md = lnet_handle2md(&mdh);
2205         if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
2206                 CERROR("Dropping PUT (%llu:%d:%s): MD (%d) invalid\n",
2207                        match_bits, portal, libcfs_id2str(target),
2208                        md == NULL ? -1 : md->md_threshold);
2209                 if (md != NULL && md->md_me != NULL)
2210                         CERROR("Source MD also attached to portal %d\n",
2211                                md->md_me->me_portal);
2212                 lnet_res_unlock(cpt);
2213
2214                 lnet_msg_free(msg);
2215                 return -ENOENT;
2216         }
2217
2218         CDEBUG(D_NET, "LNetPut -> %s\n", libcfs_id2str(target));
2219
2220         lnet_msg_attach_md(msg, md, 0, 0);
2221
2222         lnet_prep_send(msg, LNET_MSG_PUT, target, 0, md->md_length);
2223
2224         msg->msg_hdr.msg.put.match_bits = cpu_to_le64(match_bits);
2225         msg->msg_hdr.msg.put.ptl_index = cpu_to_le32(portal);
2226         msg->msg_hdr.msg.put.offset = cpu_to_le32(offset);
2227         msg->msg_hdr.msg.put.hdr_data = hdr_data;
2228
2229         /* NB handles only looked up by creator (no flips) */
2230         if (ack == LNET_ACK_REQ) {
2231                 msg->msg_hdr.msg.put.ack_wmd.wh_interface_cookie =
2232                         the_lnet.ln_interface_cookie;
2233                 msg->msg_hdr.msg.put.ack_wmd.wh_object_cookie =
2234                         md->md_lh.lh_cookie;
2235         } else {
2236                 msg->msg_hdr.msg.put.ack_wmd.wh_interface_cookie =
2237                         LNET_WIRE_HANDLE_COOKIE_NONE;
2238                 msg->msg_hdr.msg.put.ack_wmd.wh_object_cookie =
2239                         LNET_WIRE_HANDLE_COOKIE_NONE;
2240         }
2241
2242         lnet_res_unlock(cpt);
2243
2244         lnet_build_msg_event(msg, LNET_EVENT_SEND);
2245
2246         rc = lnet_send(self, msg, LNET_NID_ANY);
2247         if (rc != 0) {
2248                 CNETERR( "Error sending PUT to %s: %d\n",
2249                        libcfs_id2str(target), rc);
2250                 lnet_finalize (NULL, msg, rc);
2251         }
2252
2253         /* completion will be signalled by an event */
2254         return 0;
2255 }
2256 EXPORT_SYMBOL(LNetPut);
2257
2258 lnet_msg_t *
2259 lnet_create_reply_msg (lnet_ni_t *ni, lnet_msg_t *getmsg)
2260 {
2261         /* The LND can DMA direct to the GET md (i.e. no REPLY msg).  This
2262          * returns a msg for the LND to pass to lnet_finalize() when the sink
2263          * data has been received.
2264          *
2265          * CAVEAT EMPTOR: 'getmsg' is the original GET, which is freed when
2266          * lnet_finalize() is called on it, so the LND must call this first */
2267
2268         struct lnet_msg         *msg = lnet_msg_alloc();
2269         struct lnet_libmd       *getmd = getmsg->msg_md;
2270         lnet_process_id_t       peer_id = getmsg->msg_target;
2271         int                     cpt;
2272
2273         LASSERT(!getmsg->msg_target_is_router);
2274         LASSERT(!getmsg->msg_routing);
2275
2276         if (msg == NULL) {
2277                 CERROR("%s: Dropping REPLY from %s: can't allocate msg\n",
2278                        libcfs_nid2str(ni->ni_nid), libcfs_id2str(peer_id));
2279                 goto drop;
2280         }
2281
2282         cpt = lnet_cpt_of_cookie(getmd->md_lh.lh_cookie);
2283         lnet_res_lock(cpt);
2284
2285         LASSERT(getmd->md_refcount > 0);
2286
2287         if (getmd->md_threshold == 0) {
2288                 CERROR ("%s: Dropping REPLY from %s for inactive MD %p\n",
2289                         libcfs_nid2str(ni->ni_nid), libcfs_id2str(peer_id),
2290                         getmd);
2291                 lnet_res_unlock(cpt);
2292                 goto drop;
2293         }
2294
2295         LASSERT(getmd->md_offset == 0);
2296
2297         CDEBUG(D_NET, "%s: Reply from %s md %p\n",
2298                libcfs_nid2str(ni->ni_nid), libcfs_id2str(peer_id), getmd);
2299
2300         /* setup information for lnet_build_msg_event */
2301         msg->msg_from = peer_id.nid;
2302         msg->msg_type = LNET_MSG_GET; /* flag this msg as an "optimized" GET */
2303         msg->msg_hdr.src_nid = peer_id.nid;
2304         msg->msg_hdr.payload_length = getmd->md_length;
2305         msg->msg_receiving = 1; /* required by lnet_msg_attach_md */
2306
2307         lnet_msg_attach_md(msg, getmd, getmd->md_offset, getmd->md_length);
2308         lnet_res_unlock(cpt);
2309
2310         cpt = lnet_cpt_of_nid(peer_id.nid);
2311
2312         lnet_net_lock(cpt);
2313         lnet_msg_commit(msg, cpt);
2314         lnet_net_unlock(cpt);
2315
2316         lnet_build_msg_event(msg, LNET_EVENT_REPLY);
2317
2318         return msg;
2319
2320  drop:
2321         cpt = lnet_cpt_of_nid(peer_id.nid);
2322
2323         lnet_net_lock(cpt);
2324         the_lnet.ln_counters[cpt]->drop_count++;
2325         the_lnet.ln_counters[cpt]->drop_length += getmd->md_length;
2326         lnet_net_unlock(cpt);
2327
2328         if (msg != NULL)
2329                 lnet_msg_free(msg);
2330
2331         return NULL;
2332 }
2333 EXPORT_SYMBOL(lnet_create_reply_msg);
2334
2335 void
2336 lnet_set_reply_msg_len(lnet_ni_t *ni, lnet_msg_t *reply, unsigned int len)
2337 {
2338         /* Set the REPLY length, now the RDMA that elides the REPLY message has
2339          * completed and I know it. */
2340         LASSERT (reply != NULL);
2341         LASSERT (reply->msg_type == LNET_MSG_GET);
2342         LASSERT (reply->msg_ev.type == LNET_EVENT_REPLY);
2343
2344         /* NB I trusted my peer to RDMA.  If she tells me she's written beyond
2345          * the end of my buffer, I might as well be dead. */
2346         LASSERT (len <= reply->msg_ev.mlength);
2347
2348         reply->msg_ev.mlength = len;
2349 }
2350 EXPORT_SYMBOL(lnet_set_reply_msg_len);
2351
2352 /**
2353  * Initiate an asynchronous GET operation.
2354  *
2355  * On the initiator node, an LNET_EVENT_SEND is logged when the GET request
2356  * is sent, and an LNET_EVENT_REPLY is logged when the data returned from
2357  * the target node in the REPLY has been written to local MD.
2358  *
2359  * On the target node, an LNET_EVENT_GET is logged when the GET request
2360  * arrives and is accepted into a MD.
2361  *
2362  * \param self,target,portal,match_bits,offset See the discussion in LNetPut().
2363  * \param mdh A handle for the MD that describes the memory into which the
2364  * requested data will be received. The MD must be "free floating" (See LNetMDBind()).
2365  *
2366  * \retval  0      Success, and only in this case events will be generated
2367  * and logged to EQ (if it exists) of the MD.
2368  * \retval -EIO    Simulated failure.
2369  * \retval -ENOMEM Memory allocation failure.
2370  * \retval -ENOENT Invalid MD object.
2371  */
2372 int
2373 LNetGet(lnet_nid_t self, lnet_handle_md_t mdh,
2374         lnet_process_id_t target, unsigned int portal,
2375         __u64 match_bits, unsigned int offset)
2376 {
2377         struct lnet_msg         *msg;
2378         struct lnet_libmd       *md;
2379         int                     cpt;
2380         int                     rc;
2381
2382         LASSERT (the_lnet.ln_refcount > 0);
2383
2384         if (!list_empty(&the_lnet.ln_test_peers) &&     /* normally we don't */
2385             fail_peer(target.nid, 1))                   /* shall we now? */
2386         {
2387                 CERROR("Dropping GET to %s: simulated failure\n",
2388                        libcfs_id2str(target));
2389                 return -EIO;
2390         }
2391
2392         msg = lnet_msg_alloc();
2393         if (msg == NULL) {
2394                 CERROR("Dropping GET to %s: ENOMEM on lnet_msg_t\n",
2395                        libcfs_id2str(target));
2396                 return -ENOMEM;
2397         }
2398
2399         cpt = lnet_cpt_of_cookie(mdh.cookie);
2400         lnet_res_lock(cpt);
2401
2402         md = lnet_handle2md(&mdh);
2403         if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
2404                 CERROR("Dropping GET (%llu:%d:%s): MD (%d) invalid\n",
2405                        match_bits, portal, libcfs_id2str(target),
2406                        md == NULL ? -1 : md->md_threshold);
2407                 if (md != NULL && md->md_me != NULL)
2408                         CERROR("REPLY MD also attached to portal %d\n",
2409                                md->md_me->me_portal);
2410
2411                 lnet_res_unlock(cpt);
2412
2413                 lnet_msg_free(msg);
2414                 return -ENOENT;
2415         }
2416
2417         CDEBUG(D_NET, "LNetGet -> %s\n", libcfs_id2str(target));
2418
2419         lnet_msg_attach_md(msg, md, 0, 0);
2420
2421         lnet_prep_send(msg, LNET_MSG_GET, target, 0, 0);
2422
2423         msg->msg_hdr.msg.get.match_bits = cpu_to_le64(match_bits);
2424         msg->msg_hdr.msg.get.ptl_index = cpu_to_le32(portal);
2425         msg->msg_hdr.msg.get.src_offset = cpu_to_le32(offset);
2426         msg->msg_hdr.msg.get.sink_length = cpu_to_le32(md->md_length);
2427
2428         /* NB handles only looked up by creator (no flips) */
2429         msg->msg_hdr.msg.get.return_wmd.wh_interface_cookie =
2430                 the_lnet.ln_interface_cookie;
2431         msg->msg_hdr.msg.get.return_wmd.wh_object_cookie =
2432                 md->md_lh.lh_cookie;
2433
2434         lnet_res_unlock(cpt);
2435
2436         lnet_build_msg_event(msg, LNET_EVENT_SEND);
2437
2438         rc = lnet_send(self, msg, LNET_NID_ANY);
2439         if (rc < 0) {
2440                 CNETERR("Error sending GET to %s: %d\n",
2441                         libcfs_id2str(target), rc);
2442                 lnet_finalize(NULL, msg, rc);
2443         }
2444
2445         /* completion will be signalled by an event */
2446         return 0;
2447 }
2448 EXPORT_SYMBOL(LNetGet);
2449
2450 /**
2451  * Calculate distance to node at \a dstnid.
2452  *
2453  * \param dstnid Target NID.
2454  * \param srcnidp If not NULL, NID of the local interface to reach \a dstnid
2455  * is saved here.
2456  * \param orderp If not NULL, order of the route to reach \a dstnid is saved
2457  * here.
2458  *
2459  * \retval 0 If \a dstnid belongs to a local interface, and reserved option
2460  * local_nid_dist_zero is set, which is the default.
2461  * \retval positives Distance to target NID, i.e. number of hops plus one.
2462  * \retval -EHOSTUNREACH If \a dstnid is not reachable.
2463  */
2464 int
2465 LNetDist(lnet_nid_t dstnid, lnet_nid_t *srcnidp, __u32 *orderp)
2466 {
2467         struct list_head        *e;
2468         struct lnet_ni          *ni;
2469         lnet_remotenet_t        *rnet;
2470         __u32                   dstnet = LNET_NIDNET(dstnid);
2471         int                     hops;
2472         int                     cpt;
2473         __u32                   order = 2;
2474         struct list_head        *rn_list;
2475
2476         /* if !local_nid_dist_zero, I don't return a distance of 0 ever
2477          * (when lustre sees a distance of 0, it substitutes 0@lo), so I
2478          * keep order 0 free for 0@lo and order 1 free for a local NID
2479          * match */
2480
2481         LASSERT (the_lnet.ln_refcount > 0);
2482
2483         cpt = lnet_net_lock_current();
2484
2485         list_for_each(e, &the_lnet.ln_nis) {
2486                 ni = list_entry(e, lnet_ni_t, ni_list);
2487
2488                 if (ni->ni_nid == dstnid) {
2489                         if (srcnidp != NULL)
2490                                 *srcnidp = dstnid;
2491                         if (orderp != NULL) {
2492                                 if (LNET_NETTYP(LNET_NIDNET(dstnid)) == LOLND)
2493                                         *orderp = 0;
2494                                 else
2495                                         *orderp = 1;
2496                         }
2497                         lnet_net_unlock(cpt);
2498
2499                         return local_nid_dist_zero ? 0 : 1;
2500                 }
2501
2502                 if (LNET_NIDNET(ni->ni_nid) == dstnet) {
2503                         /* Check if ni was originally created in
2504                          * current net namespace.
2505                          * If not, assign order above 0xffff0000,
2506                          * to make this ni not a priority. */
2507                         if (!net_eq(ni->ni_net_ns, current->nsproxy->net_ns))
2508                                 order += 0xffff0000;
2509
2510                         if (srcnidp != NULL)
2511                                 *srcnidp = ni->ni_nid;
2512                         if (orderp != NULL)
2513                                 *orderp = order;
2514                         lnet_net_unlock(cpt);
2515                         return 1;
2516                 }
2517
2518                 order++;
2519         }
2520
2521         rn_list = lnet_net2rnethash(dstnet);
2522         list_for_each(e, rn_list) {
2523                 rnet = list_entry(e, lnet_remotenet_t, lrn_list);
2524
2525                 if (rnet->lrn_net == dstnet) {
2526                         lnet_route_t *route;
2527                         lnet_route_t *shortest = NULL;
2528                         __u32 shortest_hops = LNET_UNDEFINED_HOPS;
2529                         __u32 route_hops;
2530
2531                         LASSERT(!list_empty(&rnet->lrn_routes));
2532
2533                         list_for_each_entry(route, &rnet->lrn_routes,
2534                                             lr_list) {
2535                                 route_hops = route->lr_hops;
2536                                 if (route_hops == LNET_UNDEFINED_HOPS)
2537                                         route_hops = 1;
2538                                 if (shortest == NULL ||
2539                                     route_hops < shortest_hops) {
2540                                         shortest = route;
2541                                         shortest_hops = route_hops;
2542                                 }
2543                         }
2544
2545                         LASSERT(shortest != NULL);
2546                         hops = shortest_hops;
2547                         if (srcnidp != NULL)
2548                                 *srcnidp = shortest->lr_gateway->lp_ni->ni_nid;
2549                         if (orderp != NULL)
2550                                 *orderp = order;
2551                         lnet_net_unlock(cpt);
2552                         return hops + 1;
2553                 }
2554                 order++;
2555         }
2556
2557         lnet_net_unlock(cpt);
2558         return -EHOSTUNREACH;
2559 }
2560 EXPORT_SYMBOL(LNetDist);