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