Whamcloud - gitweb
LU-7734 lnet: proper cpt locking
[fs/lustre-release.git] / 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_pid       = cpu_to_le32(target.pid);
633         /* src_nid will be set later */
634         msg->msg_hdr.src_pid        = cpu_to_le32(the_lnet.ln_pid);
635         msg->msg_hdr.payload_length = cpu_to_le32(len);
636 }
637
638 static void
639 lnet_ni_send(lnet_ni_t *ni, lnet_msg_t *msg)
640 {
641         void   *priv = msg->msg_private;
642         int     rc;
643
644         LASSERT (!in_interrupt ());
645         LASSERT (LNET_NETTYP(LNET_NIDNET(ni->ni_nid)) == LOLND ||
646                  (msg->msg_txcredit && msg->msg_peertxcredit));
647
648         rc = (ni->ni_net->net_lnd->lnd_send)(ni, priv, msg);
649         if (rc < 0)
650                 lnet_finalize(ni, msg, rc);
651 }
652
653 static int
654 lnet_ni_eager_recv(lnet_ni_t *ni, lnet_msg_t *msg)
655 {
656         int     rc;
657
658         LASSERT(!msg->msg_sending);
659         LASSERT(msg->msg_receiving);
660         LASSERT(!msg->msg_rx_ready_delay);
661         LASSERT(ni->ni_net->net_lnd->lnd_eager_recv != NULL);
662
663         msg->msg_rx_ready_delay = 1;
664         rc = (ni->ni_net->net_lnd->lnd_eager_recv)(ni, msg->msg_private, msg,
665                                                   &msg->msg_private);
666         if (rc != 0) {
667                 CERROR("recv from %s / send to %s aborted: "
668                        "eager_recv failed %d\n",
669                        libcfs_nid2str(msg->msg_rxpeer->lpni_nid),
670                        libcfs_id2str(msg->msg_target), rc);
671                 LASSERT(rc < 0); /* required by my callers */
672         }
673
674         return rc;
675 }
676
677 /*
678  * This function can be called from two paths:
679  *      1. when sending a message
680  *      2. when decommiting a message (lnet_msg_decommit_tx())
681  * In both these cases the peer_ni should have it's reference count
682  * acquired by the caller and therefore it is safe to drop the spin
683  * lock before calling lnd_query()
684  */
685 static void
686 lnet_ni_query_locked(lnet_ni_t *ni, struct lnet_peer_ni *lp)
687 {
688         cfs_time_t last_alive = 0;
689         int cpt = lnet_cpt_of_nid_locked(lp->lpni_nid, ni);
690
691         LASSERT(lnet_peer_aliveness_enabled(lp));
692         LASSERT(ni->ni_net->net_lnd->lnd_query != NULL);
693
694         lnet_net_unlock(cpt);
695         (ni->ni_net->net_lnd->lnd_query)(ni, lp->lpni_nid, &last_alive);
696         lnet_net_lock(cpt);
697
698         lp->lpni_last_query = cfs_time_current();
699
700         if (last_alive != 0) /* NI has updated timestamp */
701                 lp->lpni_last_alive = last_alive;
702 }
703
704 /* NB: always called with lnet_net_lock held */
705 static inline int
706 lnet_peer_is_alive (struct lnet_peer_ni *lp, cfs_time_t now)
707 {
708         int        alive;
709         cfs_time_t deadline;
710
711         LASSERT (lnet_peer_aliveness_enabled(lp));
712
713         /*
714          * Trust lnet_notify() if it has more recent aliveness news, but
715          * ignore the initial assumed death (see lnet_peers_start_down()).
716          */
717         spin_lock(&lp->lpni_lock);
718         if (!lp->lpni_alive && lp->lpni_alive_count > 0 &&
719             cfs_time_aftereq(lp->lpni_timestamp, lp->lpni_last_alive)) {
720                 spin_unlock(&lp->lpni_lock);
721                 return 0;
722         }
723
724         deadline =
725           cfs_time_add(lp->lpni_last_alive,
726                        cfs_time_seconds(lp->lpni_net->net_tunables.
727                                         lct_peer_timeout));
728         alive = cfs_time_after(deadline, now);
729
730         /*
731          * Update obsolete lp_alive except for routers assumed to be dead
732          * initially, because router checker would update aliveness in this
733          * case, and moreover lpni_last_alive at peer creation is assumed.
734          */
735         if (alive && !lp->lpni_alive &&
736             !(lnet_isrouter(lp) && lp->lpni_alive_count == 0)) {
737                 spin_unlock(&lp->lpni_lock);
738                 lnet_notify_locked(lp, 0, 1, lp->lpni_last_alive);
739         } else {
740                 spin_unlock(&lp->lpni_lock);
741         }
742
743         return alive;
744 }
745
746
747 /* NB: returns 1 when alive, 0 when dead, negative when error;
748  *     may drop the lnet_net_lock */
749 static int
750 lnet_peer_alive_locked (struct lnet_ni *ni, struct lnet_peer_ni *lp)
751 {
752         cfs_time_t now = cfs_time_current();
753
754         if (!lnet_peer_aliveness_enabled(lp))
755                 return -ENODEV;
756
757         if (lnet_peer_is_alive(lp, now))
758                 return 1;
759
760         /*
761          * Peer appears dead, but we should avoid frequent NI queries (at
762          * most once per lnet_queryinterval seconds).
763          */
764         if (lp->lpni_last_query != 0) {
765                 static const int lnet_queryinterval = 1;
766
767                 cfs_time_t next_query =
768                            cfs_time_add(lp->lpni_last_query,
769                                         cfs_time_seconds(lnet_queryinterval));
770
771                 if (cfs_time_before(now, next_query)) {
772                         if (lp->lpni_alive)
773                                 CWARN("Unexpected aliveness of peer %s: "
774                                       "%d < %d (%d/%d)\n",
775                                       libcfs_nid2str(lp->lpni_nid),
776                                       (int)now, (int)next_query,
777                                       lnet_queryinterval,
778                                       lp->lpni_net->net_tunables.lct_peer_timeout);
779                         return 0;
780                 }
781         }
782
783         /* query NI for latest aliveness news */
784         lnet_ni_query_locked(ni, lp);
785
786         if (lnet_peer_is_alive(lp, now))
787                 return 1;
788
789         lnet_notify_locked(lp, 0, 0, lp->lpni_last_alive);
790         return 0;
791 }
792
793 /**
794  * \param msg The message to be sent.
795  * \param do_send True if lnet_ni_send() should be called in this function.
796  *        lnet_send() is going to lnet_net_unlock immediately after this, so
797  *        it sets do_send FALSE and I don't do the unlock/send/lock bit.
798  *
799  * \retval LNET_CREDIT_OK If \a msg sent or OK to send.
800  * \retval LNET_CREDIT_WAIT If \a msg blocked for credit.
801  * \retval -EHOSTUNREACH If the next hop of the message appears dead.
802  * \retval -ECANCELED If the MD of the message has been unlinked.
803  */
804 static int
805 lnet_post_send_locked(lnet_msg_t *msg, int do_send)
806 {
807         struct lnet_peer_ni     *lp = msg->msg_txpeer;
808         struct lnet_ni          *ni = msg->msg_txni;
809         int                     cpt = msg->msg_tx_cpt;
810         struct lnet_tx_queue    *tq = ni->ni_tx_queues[cpt];
811
812         /* non-lnet_send() callers have checked before */
813         LASSERT(!do_send || msg->msg_tx_delayed);
814         LASSERT(!msg->msg_receiving);
815         LASSERT(msg->msg_tx_committed);
816
817         /* NB 'lp' is always the next hop */
818         if ((msg->msg_target.pid & LNET_PID_USERFLAG) == 0 &&
819             lnet_peer_alive_locked(ni, lp) == 0) {
820                 the_lnet.ln_counters[cpt]->drop_count++;
821                 the_lnet.ln_counters[cpt]->drop_length += msg->msg_len;
822                 lnet_net_unlock(cpt);
823                 if (msg->msg_txpeer)
824                         atomic_inc(&msg->msg_txpeer->lpni_stats.drop_count);
825                 if (msg->msg_txni)
826                         atomic_inc(&msg->msg_txni->ni_stats.drop_count);
827
828                 CNETERR("Dropping message for %s: peer not alive\n",
829                         libcfs_id2str(msg->msg_target));
830                 if (do_send)
831                         lnet_finalize(ni, msg, -EHOSTUNREACH);
832
833                 lnet_net_lock(cpt);
834                 return -EHOSTUNREACH;
835         }
836
837         if (msg->msg_md != NULL &&
838             (msg->msg_md->md_flags & LNET_MD_FLAG_ABORTED) != 0) {
839                 lnet_net_unlock(cpt);
840
841                 CNETERR("Aborting message for %s: LNetM[DE]Unlink() already "
842                         "called on the MD/ME.\n",
843                         libcfs_id2str(msg->msg_target));
844                 if (do_send)
845                         lnet_finalize(ni, msg, -ECANCELED);
846
847                 lnet_net_lock(cpt);
848                 return -ECANCELED;
849         }
850
851         if (!msg->msg_peertxcredit) {
852                 LASSERT((lp->lpni_txcredits < 0) ==
853                         !list_empty(&lp->lpni_txq));
854
855                 msg->msg_peertxcredit = 1;
856                 lp->lpni_txqnob += msg->msg_len + sizeof(lnet_hdr_t);
857                 lp->lpni_txcredits--;
858
859                 if (lp->lpni_txcredits < lp->lpni_mintxcredits)
860                         lp->lpni_mintxcredits = lp->lpni_txcredits;
861
862                 if (lp->lpni_txcredits < 0) {
863                         msg->msg_tx_delayed = 1;
864                         list_add_tail(&msg->msg_list, &lp->lpni_txq);
865                         return LNET_CREDIT_WAIT;
866                 }
867         }
868
869         if (!msg->msg_txcredit) {
870                 LASSERT((tq->tq_credits < 0) ==
871                         !list_empty(&tq->tq_delayed));
872
873                 msg->msg_txcredit = 1;
874                 tq->tq_credits--;
875                 atomic_dec(&ni->ni_tx_credits);
876
877                 if (tq->tq_credits < tq->tq_credits_min)
878                         tq->tq_credits_min = tq->tq_credits;
879
880                 if (tq->tq_credits < 0) {
881                         msg->msg_tx_delayed = 1;
882                         list_add_tail(&msg->msg_list, &tq->tq_delayed);
883                         return LNET_CREDIT_WAIT;
884                 }
885         }
886
887         if (do_send) {
888                 lnet_net_unlock(cpt);
889                 lnet_ni_send(ni, msg);
890                 lnet_net_lock(cpt);
891         }
892         return LNET_CREDIT_OK;
893 }
894
895
896 static lnet_rtrbufpool_t *
897 lnet_msg2bufpool(lnet_msg_t *msg)
898 {
899         lnet_rtrbufpool_t       *rbp;
900         int                     cpt;
901
902         LASSERT(msg->msg_rx_committed);
903
904         cpt = msg->msg_rx_cpt;
905         rbp = &the_lnet.ln_rtrpools[cpt][0];
906
907         LASSERT(msg->msg_len <= LNET_MTU);
908         while (msg->msg_len > (unsigned int)rbp->rbp_npages * PAGE_SIZE) {
909                 rbp++;
910                 LASSERT(rbp < &the_lnet.ln_rtrpools[cpt][LNET_NRBPOOLS]);
911         }
912
913         return rbp;
914 }
915
916 static int
917 lnet_post_routed_recv_locked (lnet_msg_t *msg, int do_recv)
918 {
919         /* lnet_parse is going to lnet_net_unlock immediately after this, so it
920          * sets do_recv FALSE and I don't do the unlock/send/lock bit.
921          * I return LNET_CREDIT_WAIT if msg blocked and LNET_CREDIT_OK if
922          * received or OK to receive */
923         struct lnet_peer_ni *lp = msg->msg_rxpeer;
924         lnet_rtrbufpool_t   *rbp;
925         lnet_rtrbuf_t       *rb;
926
927         LASSERT (msg->msg_iov == NULL);
928         LASSERT (msg->msg_kiov == NULL);
929         LASSERT (msg->msg_niov == 0);
930         LASSERT (msg->msg_routing);
931         LASSERT (msg->msg_receiving);
932         LASSERT (!msg->msg_sending);
933
934         /* non-lnet_parse callers only receive delayed messages */
935         LASSERT(!do_recv || msg->msg_rx_delayed);
936
937         if (!msg->msg_peerrtrcredit) {
938                 LASSERT((lp->lpni_rtrcredits < 0) ==
939                         !list_empty(&lp->lpni_rtrq));
940
941                 msg->msg_peerrtrcredit = 1;
942                 lp->lpni_rtrcredits--;
943                 if (lp->lpni_rtrcredits < lp->lpni_minrtrcredits)
944                         lp->lpni_minrtrcredits = lp->lpni_rtrcredits;
945
946                 if (lp->lpni_rtrcredits < 0) {
947                         /* must have checked eager_recv before here */
948                         LASSERT(msg->msg_rx_ready_delay);
949                         msg->msg_rx_delayed = 1;
950                         list_add_tail(&msg->msg_list, &lp->lpni_rtrq);
951                         return LNET_CREDIT_WAIT;
952                 }
953         }
954
955         rbp = lnet_msg2bufpool(msg);
956
957         if (!msg->msg_rtrcredit) {
958                 msg->msg_rtrcredit = 1;
959                 rbp->rbp_credits--;
960                 if (rbp->rbp_credits < rbp->rbp_mincredits)
961                         rbp->rbp_mincredits = rbp->rbp_credits;
962
963                 if (rbp->rbp_credits < 0) {
964                         /* must have checked eager_recv before here */
965                         LASSERT(msg->msg_rx_ready_delay);
966                         msg->msg_rx_delayed = 1;
967                         list_add_tail(&msg->msg_list, &rbp->rbp_msgs);
968                         return LNET_CREDIT_WAIT;
969                 }
970         }
971
972         LASSERT(!list_empty(&rbp->rbp_bufs));
973         rb = list_entry(rbp->rbp_bufs.next, lnet_rtrbuf_t, rb_list);
974         list_del(&rb->rb_list);
975
976         msg->msg_niov = rbp->rbp_npages;
977         msg->msg_kiov = &rb->rb_kiov[0];
978
979         if (do_recv) {
980                 int cpt = msg->msg_rx_cpt;
981
982                 lnet_net_unlock(cpt);
983                 lnet_ni_recv(msg->msg_rxni, msg->msg_private, msg, 1,
984                              0, msg->msg_len, msg->msg_len);
985                 lnet_net_lock(cpt);
986         }
987         return LNET_CREDIT_OK;
988 }
989
990 void
991 lnet_return_tx_credits_locked(lnet_msg_t *msg)
992 {
993         struct lnet_peer_ni     *txpeer = msg->msg_txpeer;
994         struct lnet_ni          *txni = msg->msg_txni;
995         lnet_msg_t              *msg2;
996
997         if (msg->msg_txcredit) {
998                 struct lnet_ni       *ni = msg->msg_txni;
999                 struct lnet_tx_queue *tq = ni->ni_tx_queues[msg->msg_tx_cpt];
1000
1001                 /* give back NI txcredits */
1002                 msg->msg_txcredit = 0;
1003
1004                 LASSERT((tq->tq_credits < 0) ==
1005                         !list_empty(&tq->tq_delayed));
1006
1007                 tq->tq_credits++;
1008                 atomic_inc(&ni->ni_tx_credits);
1009                 if (tq->tq_credits <= 0) {
1010                         msg2 = list_entry(tq->tq_delayed.next,
1011                                           lnet_msg_t, msg_list);
1012                         list_del(&msg2->msg_list);
1013
1014                         LASSERT(msg2->msg_txni == ni);
1015                         LASSERT(msg2->msg_tx_delayed);
1016
1017                         (void) lnet_post_send_locked(msg2, 1);
1018                 }
1019         }
1020
1021         if (msg->msg_peertxcredit) {
1022                 /* give back peer txcredits */
1023                 msg->msg_peertxcredit = 0;
1024
1025                 LASSERT((txpeer->lpni_txcredits < 0) ==
1026                         !list_empty(&txpeer->lpni_txq));
1027
1028                 txpeer->lpni_txqnob -= msg->msg_len + sizeof(lnet_hdr_t);
1029                 LASSERT (txpeer->lpni_txqnob >= 0);
1030
1031                 txpeer->lpni_txcredits++;
1032                 if (txpeer->lpni_txcredits <= 0) {
1033                         msg2 = list_entry(txpeer->lpni_txq.next,
1034                                               lnet_msg_t, msg_list);
1035                         list_del(&msg2->msg_list);
1036
1037                         LASSERT(msg2->msg_txpeer == txpeer);
1038                         LASSERT(msg2->msg_tx_delayed);
1039
1040                         (void) lnet_post_send_locked(msg2, 1);
1041                 }
1042         }
1043
1044         if (txni != NULL) {
1045                 msg->msg_txni = NULL;
1046                 lnet_ni_decref_locked(txni, msg->msg_tx_cpt);
1047         }
1048
1049         if (txpeer != NULL) {
1050                 /*
1051                  * TODO:
1052                  * Once the patch for the health comes in we need to set
1053                  * the health of the peer ni to bad when we fail to send
1054                  * a message.
1055                  * int status = msg->msg_ev.status;
1056                  * if (status != 0)
1057                  *      lnet_set_peer_ni_health_locked(txpeer, false)
1058                  */
1059                 msg->msg_txpeer = NULL;
1060                 lnet_peer_ni_decref_locked(txpeer);
1061         }
1062 }
1063
1064 void
1065 lnet_schedule_blocked_locked(lnet_rtrbufpool_t *rbp)
1066 {
1067         lnet_msg_t      *msg;
1068
1069         if (list_empty(&rbp->rbp_msgs))
1070                 return;
1071         msg = list_entry(rbp->rbp_msgs.next,
1072                          lnet_msg_t, msg_list);
1073         list_del(&msg->msg_list);
1074
1075         (void)lnet_post_routed_recv_locked(msg, 1);
1076 }
1077
1078 void
1079 lnet_drop_routed_msgs_locked(struct list_head *list, int cpt)
1080 {
1081         lnet_msg_t       *msg;
1082         lnet_msg_t       *tmp;
1083         struct list_head drop;
1084
1085         INIT_LIST_HEAD(&drop);
1086
1087         list_splice_init(list, &drop);
1088
1089         lnet_net_unlock(cpt);
1090
1091         list_for_each_entry_safe(msg, tmp, &drop, msg_list) {
1092                 lnet_ni_recv(msg->msg_rxni, msg->msg_private, NULL,
1093                              0, 0, 0, msg->msg_hdr.payload_length);
1094                 list_del_init(&msg->msg_list);
1095                 lnet_finalize(NULL, msg, -ECANCELED);
1096         }
1097
1098         lnet_net_lock(cpt);
1099 }
1100
1101 void
1102 lnet_return_rx_credits_locked(lnet_msg_t *msg)
1103 {
1104         struct lnet_peer_ni     *rxpeer = msg->msg_rxpeer;
1105         struct lnet_ni          *rxni = msg->msg_rxni;
1106         lnet_msg_t              *msg2;
1107
1108         if (msg->msg_rtrcredit) {
1109                 /* give back global router credits */
1110                 lnet_rtrbuf_t     *rb;
1111                 lnet_rtrbufpool_t *rbp;
1112
1113                 /* NB If a msg ever blocks for a buffer in rbp_msgs, it stays
1114                  * there until it gets one allocated, or aborts the wait
1115                  * itself */
1116                 LASSERT(msg->msg_kiov != NULL);
1117
1118                 rb = list_entry(msg->msg_kiov, lnet_rtrbuf_t, rb_kiov[0]);
1119                 rbp = rb->rb_pool;
1120
1121                 msg->msg_kiov = NULL;
1122                 msg->msg_rtrcredit = 0;
1123
1124                 LASSERT(rbp == lnet_msg2bufpool(msg));
1125
1126                 LASSERT((rbp->rbp_credits > 0) ==
1127                         !list_empty(&rbp->rbp_bufs));
1128
1129                 /* If routing is now turned off, we just drop this buffer and
1130                  * don't bother trying to return credits.  */
1131                 if (!the_lnet.ln_routing) {
1132                         lnet_destroy_rtrbuf(rb, rbp->rbp_npages);
1133                         goto routing_off;
1134                 }
1135
1136                 /* It is possible that a user has lowered the desired number of
1137                  * buffers in this pool.  Make sure we never put back
1138                  * more buffers than the stated number. */
1139                 if (unlikely(rbp->rbp_credits >= rbp->rbp_req_nbuffers)) {
1140                         /* Discard this buffer so we don't have too
1141                          * many. */
1142                         lnet_destroy_rtrbuf(rb, rbp->rbp_npages);
1143                         rbp->rbp_nbuffers--;
1144                 } else {
1145                         list_add(&rb->rb_list, &rbp->rbp_bufs);
1146                         rbp->rbp_credits++;
1147                         if (rbp->rbp_credits <= 0)
1148                                 lnet_schedule_blocked_locked(rbp);
1149                 }
1150         }
1151
1152 routing_off:
1153         if (msg->msg_peerrtrcredit) {
1154                 /* give back peer router credits */
1155                 msg->msg_peerrtrcredit = 0;
1156
1157                 LASSERT((rxpeer->lpni_rtrcredits < 0) ==
1158                         !list_empty(&rxpeer->lpni_rtrq));
1159
1160                 rxpeer->lpni_rtrcredits++;
1161
1162                 /* drop all messages which are queued to be routed on that
1163                  * peer. */
1164                 if (!the_lnet.ln_routing) {
1165                         lnet_drop_routed_msgs_locked(&rxpeer->lpni_rtrq,
1166                                                      msg->msg_rx_cpt);
1167                 } else if (rxpeer->lpni_rtrcredits <= 0) {
1168                         msg2 = list_entry(rxpeer->lpni_rtrq.next,
1169                                           lnet_msg_t, msg_list);
1170                         list_del(&msg2->msg_list);
1171
1172                         (void) lnet_post_routed_recv_locked(msg2, 1);
1173                 }
1174         }
1175         if (rxni != NULL) {
1176                 msg->msg_rxni = NULL;
1177                 lnet_ni_decref_locked(rxni, msg->msg_rx_cpt);
1178         }
1179         if (rxpeer != NULL) {
1180                 msg->msg_rxpeer = NULL;
1181                 lnet_peer_ni_decref_locked(rxpeer);
1182         }
1183 }
1184
1185 static int
1186 lnet_compare_peers(struct lnet_peer_ni *p1, struct lnet_peer_ni *p2)
1187 {
1188         if (p1->lpni_txqnob < p2->lpni_txqnob)
1189                 return 1;
1190
1191         if (p1->lpni_txqnob > p2->lpni_txqnob)
1192                 return -1;
1193
1194         if (p1->lpni_txcredits > p2->lpni_txcredits)
1195                 return 1;
1196
1197         if (p1->lpni_txcredits < p2->lpni_txcredits)
1198                 return -1;
1199
1200         return 0;
1201 }
1202
1203 static int
1204 lnet_compare_routes(lnet_route_t *r1, lnet_route_t *r2)
1205 {
1206         struct lnet_peer_ni *p1 = r1->lr_gateway;
1207         struct lnet_peer_ni *p2 = r2->lr_gateway;
1208         int r1_hops = (r1->lr_hops == LNET_UNDEFINED_HOPS) ? 1 : r1->lr_hops;
1209         int r2_hops = (r2->lr_hops == LNET_UNDEFINED_HOPS) ? 1 : r2->lr_hops;
1210         int rc;
1211
1212         if (r1->lr_priority < r2->lr_priority)
1213                 return 1;
1214
1215         if (r1->lr_priority > r2->lr_priority)
1216                 return -1;
1217
1218         if (r1_hops < r2_hops)
1219                 return 1;
1220
1221         if (r1_hops > r2_hops)
1222                 return -1;
1223
1224         rc = lnet_compare_peers(p1, p2);
1225         if (rc)
1226                 return rc;
1227
1228         if (r1->lr_seq - r2->lr_seq <= 0)
1229                 return 1;
1230
1231         return -1;
1232 }
1233
1234 static struct lnet_peer_ni *
1235 lnet_find_route_locked(struct lnet_net *net, lnet_nid_t target,
1236                        lnet_nid_t rtr_nid)
1237 {
1238         lnet_remotenet_t        *rnet;
1239         lnet_route_t            *route;
1240         lnet_route_t            *best_route;
1241         lnet_route_t            *last_route;
1242         struct lnet_peer_ni     *lpni_best;
1243         struct lnet_peer_ni     *lp;
1244         int                     rc;
1245
1246         /* If @rtr_nid is not LNET_NID_ANY, return the gateway with
1247          * rtr_nid nid, otherwise find the best gateway I can use */
1248
1249         rnet = lnet_find_rnet_locked(LNET_NIDNET(target));
1250         if (rnet == NULL)
1251                 return NULL;
1252
1253         lpni_best = NULL;
1254         best_route = last_route = NULL;
1255         list_for_each_entry(route, &rnet->lrn_routes, lr_list) {
1256                 lp = route->lr_gateway;
1257
1258                 if (!lnet_is_route_alive(route))
1259                         continue;
1260
1261                 if (net != NULL && lp->lpni_net != net)
1262                         continue;
1263
1264                 if (lp->lpni_nid == rtr_nid) /* it's pre-determined router */
1265                         return lp;
1266
1267                 if (lpni_best == NULL) {
1268                         best_route = last_route = route;
1269                         lpni_best = lp;
1270                         continue;
1271                 }
1272
1273                 /* no protection on below fields, but it's harmless */
1274                 if (last_route->lr_seq - route->lr_seq < 0)
1275                         last_route = route;
1276
1277                 rc = lnet_compare_routes(route, best_route);
1278                 if (rc < 0)
1279                         continue;
1280
1281                 best_route = route;
1282                 lpni_best = lp;
1283         }
1284
1285         /* set sequence number on the best router to the latest sequence + 1
1286          * so we can round-robin all routers, it's race and inaccurate but
1287          * harmless and functional  */
1288         if (best_route != NULL)
1289                 best_route->lr_seq = last_route->lr_seq + 1;
1290         return lpni_best;
1291 }
1292
1293 static int
1294 lnet_select_pathway(lnet_nid_t src_nid, lnet_nid_t dst_nid,
1295                     struct lnet_msg *msg, lnet_nid_t rtr_nid, bool *lo_sent)
1296 {
1297         struct lnet_ni          *best_ni = NULL;
1298         struct lnet_peer_ni     *best_lpni = NULL;
1299         struct lnet_peer_ni     *net_gw = NULL;
1300         struct lnet_peer_ni     *best_gw = NULL;
1301         struct lnet_peer_ni     *lpni;
1302         struct lnet_peer        *peer = NULL;
1303         struct lnet_peer_net    *peer_net;
1304         struct lnet_net         *local_net;
1305         struct lnet_ni          *ni = NULL;
1306         int                     cpt, cpt2, rc;
1307         bool                    routing = false;
1308         bool                    ni_is_pref = false;
1309         bool                    preferred = false;
1310         int                     best_credits = 0;
1311         __u32                   seq, seq2;
1312         int                     best_lpni_credits = INT_MIN;
1313         int                     md_cpt = 0;
1314         int                     shortest_distance = INT_MAX;
1315         int                     distance = 0;
1316         bool                    found_ir = false;
1317
1318 again:
1319         /*
1320          * get an initial CPT to use for locking. The idea here is not to
1321          * serialize the calls to select_pathway, so that as many
1322          * operations can run concurrently as possible. To do that we use
1323          * the CPT where this call is being executed. Later on when we
1324          * determine the CPT to use in lnet_message_commit, we switch the
1325          * lock and check if there was any configuration changes, if none,
1326          * then we proceed, if there is, then we'll need to update the cpt
1327          * and redo the operation.
1328          */
1329         cpt = lnet_net_lock_current();
1330
1331         best_gw = NULL;
1332         routing = false;
1333         local_net = NULL;
1334         best_ni = NULL;
1335         shortest_distance = INT_MAX;
1336         found_ir = false;
1337
1338         if (the_lnet.ln_shutdown) {
1339                 lnet_net_unlock(cpt);
1340                 return -ESHUTDOWN;
1341         }
1342
1343         if (msg->msg_md != NULL)
1344                 /* get the cpt of the MD, used during NUMA based selection */
1345                 md_cpt = lnet_cpt_of_cookie(msg->msg_md->md_lh.lh_cookie);
1346         else
1347                 md_cpt = CFS_CPT_ANY;
1348
1349         /*
1350          * initialize the variables which could be reused if we go to
1351          * again
1352          */
1353         lpni = NULL;
1354         seq = lnet_get_dlc_seq_locked();
1355
1356         peer = lnet_find_or_create_peer_locked(dst_nid, cpt);
1357         if (IS_ERR(peer)) {
1358                 lnet_net_unlock(cpt);
1359                 return PTR_ERR(peer);
1360         }
1361
1362         /* If peer is not healthy then can not send anything to it */
1363         if (!lnet_is_peer_healthy_locked(peer)) {
1364                 lnet_net_unlock(cpt);
1365                 return -EHOSTUNREACH;
1366         }
1367
1368         if (!peer->lp_multi_rail && lnet_get_num_peer_nis(peer) > 1) {
1369                 CERROR("peer %s is declared to be non MR capable, "
1370                        "yet configured with more than one NID\n",
1371                        libcfs_nid2str(dst_nid));
1372                 return -EINVAL;
1373         }
1374
1375         /*
1376          * STEP 1: first jab at determineing best_ni
1377          * if src_nid is explicitly specified, then best_ni is already
1378          * pre-determiend for us. Otherwise we need to select the best
1379          * one to use later on
1380          */
1381         if (src_nid != LNET_NID_ANY) {
1382                 best_ni = lnet_nid2ni_locked(src_nid, cpt);
1383                 if (!best_ni) {
1384                         lnet_net_unlock(cpt);
1385                         LCONSOLE_WARN("Can't send to %s: src %s is not a "
1386                                       "local nid\n", libcfs_nid2str(dst_nid),
1387                                       libcfs_nid2str(src_nid));
1388                         return -EINVAL;
1389                 }
1390
1391                 if (best_ni->ni_net->net_id != LNET_NIDNET(dst_nid)) {
1392                         lnet_net_unlock(cpt);
1393                         LCONSOLE_WARN("No route to %s via from %s\n",
1394                                       libcfs_nid2str(dst_nid),
1395                                       libcfs_nid2str(src_nid));
1396                         return -EINVAL;
1397                 }
1398         }
1399
1400         if (best_ni)
1401                 goto pick_peer;
1402
1403         /*
1404          * Decide whether we need to route to peer_ni.
1405          * Get the local net that I need to be on to be able to directly
1406          * send to that peer.
1407          *
1408          * a. Find the peer which the dst_nid belongs to.
1409          * b. Iterate through each of the peer_nets/nis to decide
1410          * the best peer/local_ni pair to use
1411          */
1412         list_for_each_entry(peer_net, &peer->lp_peer_nets, lpn_on_peer_list) {
1413                 if (!lnet_is_peer_net_healthy_locked(peer_net))
1414                         continue;
1415
1416                 local_net = lnet_get_net_locked(peer_net->lpn_net_id);
1417                 if (!local_net) {
1418                         /*
1419                          * go through each peer_ni on that peer_net and
1420                          * determine the best possible gw to go through
1421                          */
1422                         list_for_each_entry(lpni, &peer_net->lpn_peer_nis,
1423                                             lpni_on_peer_net_list) {
1424                                 net_gw = lnet_find_route_locked(NULL,
1425                                                                 lpni->lpni_nid,
1426                                                                 rtr_nid);
1427
1428                                 /*
1429                                  * if no route is found for that network then
1430                                  * move onto the next peer_ni in the peer
1431                                  */
1432                                 if (!net_gw)
1433                                         continue;
1434
1435                                 if (!best_gw) {
1436                                         best_gw = net_gw;
1437                                         best_lpni = lpni;
1438                                 } else  {
1439                                         rc = lnet_compare_peers(net_gw,
1440                                                                 best_gw);
1441                                         if (rc > 0) {
1442                                                 best_gw = net_gw;
1443                                                 best_lpni = lpni;
1444                                         }
1445                                 }
1446                         }
1447
1448                         if (!best_gw)
1449                                 continue;
1450
1451                         local_net = lnet_get_net_locked
1452                                         (LNET_NIDNET(best_gw->lpni_nid));
1453                         routing = true;
1454                 } else {
1455                         routing = false;
1456                         best_gw = NULL;
1457                 }
1458
1459                 /* no routable net found go on to a different net */
1460                 if (!local_net)
1461                         continue;
1462
1463                 /*
1464                  * Iterate through the NIs in this local Net and select
1465                  * the NI to send from. The selection is determined by
1466                  * these 3 criterion in the following priority:
1467                  *      1. NUMA
1468                  *      2. NI available credits
1469                  *      3. Round Robin
1470                  */
1471                 while ((ni = lnet_get_next_ni_locked(local_net, ni))) {
1472                         int ni_credits;
1473
1474                         if (!lnet_is_ni_healthy_locked(ni))
1475                                 continue;
1476
1477                         ni_credits = atomic_read(&ni->ni_tx_credits);
1478
1479                         /*
1480                          * calculate the distance from the cpt on which
1481                          * the message memory is allocated to the CPT of
1482                          * the NI's physical device
1483                          */
1484                         distance = cfs_cpt_distance(lnet_cpt_table(),
1485                                                     md_cpt,
1486                                                     ni->dev_cpt);
1487
1488                         /*
1489                          * If we already have a closer NI within the NUMA
1490                          * range provided, then there is no need to
1491                          * consider the current NI. Move on to the next
1492                          * one.
1493                          */
1494                         if (distance > shortest_distance &&
1495                             distance > lnet_get_numa_range())
1496                                 continue;
1497
1498                         if (distance < shortest_distance &&
1499                             distance > lnet_get_numa_range()) {
1500                                 /*
1501                                  * The current NI is the closest one that we
1502                                  * have found, even though it's not in the
1503                                  * NUMA range specified. This occurs if
1504                                  * the NUMA range is less than the least
1505                                  * of the distances in the system.
1506                                  * In effect NUMA range consideration is
1507                                  * turned off.
1508                                  */
1509                                 shortest_distance = distance;
1510                         } else if ((distance <= shortest_distance &&
1511                                     distance < lnet_get_numa_range()) ||
1512                                    distance == shortest_distance) {
1513                                 /*
1514                                  * This NI is either within range or it's
1515                                  * equidistant. In both of these cases we
1516                                  * would want to select the NI based on
1517                                  * its available credits first, and then
1518                                  * via Round Robin.
1519                                  */
1520                                 if (distance <= shortest_distance &&
1521                                     distance < lnet_get_numa_range()) {
1522                                         /*
1523                                          * If this is the first NI that's
1524                                          * within range, then set the
1525                                          * shortest distance to the range
1526                                          * specified by the user. In
1527                                          * effect we're saying that all
1528                                          * NIs that fall within this NUMA
1529                                          * range shall be dealt with as
1530                                          * having equal NUMA weight. Which
1531                                          * will mean that we should select
1532                                          * through that set by their
1533                                          * available credits first
1534                                          * followed by Round Robin.
1535                                          *
1536                                          * And since this is the first NI
1537                                          * in the range, let's just set it
1538                                          * as our best_ni for now. The
1539                                          * following NIs found in the
1540                                          * range will be dealt with as
1541                                          * mentioned previously.
1542                                          */
1543                                         shortest_distance = lnet_get_numa_range();
1544                                         if (!found_ir) {
1545                                                 found_ir = true;
1546                                                 goto set_ni;
1547                                         }
1548                                 }
1549                                 /*
1550                                  * This NI is NUMA equidistant let's
1551                                  * select using credits followed by Round
1552                                  * Robin.
1553                                  */
1554                                 if (ni_credits < best_credits) {
1555                                         continue;
1556                                 } else if (ni_credits == best_credits) {
1557                                         if (best_ni) {
1558                                                 if (best_ni->ni_seq <= ni->ni_seq)
1559                                                         continue;
1560                                         }
1561                                 }
1562                         }
1563 set_ni:
1564                         best_ni = ni;
1565                         best_credits = ni_credits;
1566                 }
1567         }
1568         /*
1569          * if the peer is not MR capable, then we should always send to it
1570          * using the first NI in the NET we determined.
1571          */
1572         if (!peer->lp_multi_rail && local_net != NULL)
1573                 best_ni = lnet_net2ni_locked(local_net->net_id, cpt);
1574
1575         if (!best_ni) {
1576                 lnet_net_unlock(cpt);
1577                 LCONSOLE_WARN("No local ni found to send from to %s\n",
1578                         libcfs_nid2str(dst_nid));
1579                 return -EINVAL;
1580         }
1581
1582         /*
1583          * Now that we selected the NI to use increment its sequence
1584          * number so the Round Robin algorithm will detect that it has
1585          * been used and pick the next NI.
1586          */
1587         best_ni->ni_seq++;
1588
1589         if (routing)
1590                 goto send;
1591
1592 pick_peer:
1593         if (best_ni == the_lnet.ln_loni) {
1594                 /* No send credit hassles with LOLND */
1595                 lnet_ni_addref_locked(best_ni, cpt);
1596                 msg->msg_hdr.dest_nid = cpu_to_le64(best_ni->ni_nid);
1597                 if (!msg->msg_routing)
1598                         msg->msg_hdr.src_nid = cpu_to_le64(best_ni->ni_nid);
1599                 msg->msg_target.nid = best_ni->ni_nid;
1600                 lnet_msg_commit(msg, cpt);
1601
1602                 lnet_net_unlock(cpt);
1603                 msg->msg_txni = best_ni;
1604                 lnet_ni_send(best_ni, msg);
1605
1606                 *lo_sent = true;
1607                 return 0;
1608         }
1609
1610         lpni = NULL;
1611
1612         if (msg->msg_type == LNET_MSG_REPLY ||
1613             msg->msg_type == LNET_MSG_ACK) {
1614                 /*
1615                  * for replies we want to respond on the same peer_ni we
1616                  * received the message on if possible. If not, then pick
1617                  * a peer_ni to send to
1618                  */
1619                 best_lpni = lnet_find_peer_ni_locked(dst_nid);
1620                 if (best_lpni) {
1621                         lnet_peer_ni_decref_locked(best_lpni);
1622                         goto send;
1623                 } else {
1624                         CDEBUG(D_NET, "unable to send msg_type %d to "
1625                               "originating %s\n", msg->msg_type,
1626                               libcfs_nid2str(dst_nid));
1627                 }
1628         }
1629
1630         peer_net = lnet_peer_get_net_locked(peer,
1631                                             best_ni->ni_net->net_id);
1632         /*
1633          * peer_net is not available or the src_nid is explicitly defined
1634          * and the peer_net for that src_nid is unhealthy. find a route to
1635          * the destination nid.
1636          */
1637         if (!peer_net ||
1638             (src_nid != LNET_NID_ANY &&
1639              !lnet_is_peer_net_healthy_locked(peer_net))) {
1640                 best_gw = lnet_find_route_locked(best_ni->ni_net,
1641                                                  dst_nid,
1642                                                  rtr_nid);
1643                 /*
1644                  * if no route is found for that network then
1645                  * move onto the next peer_ni in the peer
1646                  */
1647                 if (!best_gw) {
1648                         lnet_net_unlock(cpt);
1649                         LCONSOLE_WARN("No route to peer from %s\n",
1650                                 libcfs_nid2str(best_ni->ni_nid));
1651                         return -EHOSTUNREACH;
1652                 }
1653
1654                 CDEBUG(D_NET, "Best route to %s via %s for %s %d\n",
1655                         libcfs_nid2str(dst_nid),
1656                         libcfs_nid2str(best_gw->lpni_nid),
1657                         lnet_msgtyp2str(msg->msg_type), msg->msg_len);
1658
1659                 best_lpni = lnet_find_peer_ni_locked(dst_nid);
1660                 LASSERT(best_lpni != NULL);
1661                 lnet_peer_ni_decref_locked(best_lpni);
1662
1663                 routing = true;
1664
1665                 goto send;
1666         } else if (!lnet_is_peer_net_healthy_locked(peer_net)) {
1667                 /*
1668                  * this peer_net is unhealthy but we still have an opportunity
1669                  * to find another peer_net that we can use
1670                  */
1671                 __u32 net_id = peer_net->lpn_net_id;
1672                 lnet_net_unlock(cpt);
1673                 if (!best_lpni)
1674                         LCONSOLE_WARN("peer net %s unhealthy\n",
1675                                       libcfs_net2str(net_id));
1676                 goto again;
1677         }
1678
1679         best_lpni = NULL;
1680         while ((lpni = lnet_get_next_peer_ni_locked(peer, peer_net, lpni))) {
1681                 /*
1682                  * if this peer ni is not healthy just skip it, no point in
1683                  * examining it further
1684                  */
1685                 if (!lnet_is_peer_ni_healthy_locked(lpni))
1686                         continue;
1687                 ni_is_pref = lnet_peer_is_ni_pref_locked(lpni, best_ni);
1688
1689                 /* if this is a preferred peer use it */
1690                 if (!preferred && ni_is_pref) {
1691                         preferred = true;
1692                 } else if (preferred && !ni_is_pref) {
1693                         /*
1694                          * this is not the preferred peer so let's ignore
1695                          * it.
1696                          */
1697                         continue;
1698                 } if (lpni->lpni_txcredits < best_lpni_credits)
1699                         /*
1700                          * We already have a peer that has more credits
1701                          * available than this one. No need to consider
1702                          * this peer further.
1703                          */
1704                         continue;
1705                 else if (lpni->lpni_txcredits == best_lpni_credits) {
1706                         /*
1707                          * The best peer found so far and the current peer
1708                          * have the same number of available credits let's
1709                          * make sure to select between them using Round
1710                          * Robin
1711                          */
1712                         if (best_lpni) {
1713                                 if (best_lpni->lpni_seq <= lpni->lpni_seq)
1714                                         continue;
1715                         }
1716                 }
1717
1718                 best_lpni = lpni;
1719                 best_lpni_credits = lpni->lpni_txcredits;
1720         }
1721
1722         /*
1723          * Increment sequence number of the peer selected so that we can
1724          * pick the next one in Round Robin.
1725          */
1726         best_lpni->lpni_seq++;
1727
1728         /* if we still can't find a peer ni then we can't reach it */
1729         if (!best_lpni) {
1730                 __u32 net_id = peer_net->lpn_net_id;
1731                 lnet_net_unlock(cpt);
1732                 LCONSOLE_WARN("no peer_ni found on peer net %s\n",
1733                                 libcfs_net2str(net_id));
1734                 return -EHOSTUNREACH;
1735         }
1736
1737 send:
1738         /*
1739          * Use lnet_cpt_of_nid() to determine the CPT used to commit the
1740          * message. This ensures that we get a CPT that is correct for
1741          * the NI when the NI has been restricted to a subset of all CPTs.
1742          * If the selected CPT differs from the one currently locked, we
1743          * must unlock and relock the lnet_net_lock(), and then check whether
1744          * the configuration has changed. We don't have a hold on the best_ni
1745          * or best_peer_ni yet, and they may have vanished.
1746          */
1747         cpt2 = lnet_cpt_of_nid_locked(best_lpni->lpni_nid, best_ni);
1748         if (cpt != cpt2) {
1749                 lnet_net_unlock(cpt);
1750                 cpt = cpt2;
1751                 lnet_net_lock(cpt);
1752                 seq2 = lnet_get_dlc_seq_locked();
1753                 if (seq2 != seq) {
1754                         lnet_net_unlock(cpt);
1755                         goto again;
1756                 }
1757         }
1758
1759         /*
1760          * store the best_lpni in the message right away to avoid having
1761          * to do the same operation under different conditions
1762          */
1763         msg->msg_txpeer = (routing) ? best_gw : best_lpni;
1764         msg->msg_txni = best_ni;
1765         /*
1766          * grab a reference for the best_ni since now it's in use in this
1767          * send. the reference will need to be dropped when the message is
1768          * finished in lnet_finalize()
1769          */
1770         lnet_ni_addref_locked(msg->msg_txni, cpt);
1771         lnet_peer_ni_addref_locked(msg->msg_txpeer);
1772
1773         /*
1774          * set the destination nid in the message here because it's
1775          * possible that we'd be sending to a different nid than the one
1776          * originaly given.
1777          */
1778         msg->msg_hdr.dest_nid = cpu_to_le64(msg->msg_txpeer->lpni_nid);
1779
1780         /*
1781          * Always set the target.nid to the best peer picked. Either the
1782          * nid will be one of the preconfigured NIDs, or the same NID as
1783          * what was originaly set in the target or it will be the NID of
1784          * a router if this message should be routed
1785          */
1786         msg->msg_target.nid = msg->msg_txpeer->lpni_nid;
1787
1788         /*
1789          * lnet_msg_commit assigns the correct cpt to the message, which
1790          * is used to decrement the correct refcount on the ni when it's
1791          * time to return the credits
1792          */
1793         lnet_msg_commit(msg, cpt);
1794
1795         /*
1796          * If we are routing the message then we don't need to overwrite
1797          * the src_nid since it would've been set at the origin. Otherwise
1798          * we are the originator so we need to set it.
1799          */
1800         if (!msg->msg_routing)
1801                 msg->msg_hdr.src_nid = cpu_to_le64(msg->msg_txni->ni_nid);
1802
1803         if (routing) {
1804                 msg->msg_target_is_router = 1;
1805                 msg->msg_target.pid = LNET_PID_LUSTRE;
1806         }
1807
1808         rc = lnet_post_send_locked(msg, 0);
1809
1810         lnet_net_unlock(cpt);
1811
1812         return rc;
1813 }
1814
1815 int
1816 lnet_send(lnet_nid_t src_nid, lnet_msg_t *msg, lnet_nid_t rtr_nid)
1817 {
1818         lnet_nid_t              dst_nid = msg->msg_target.nid;
1819         int                     rc;
1820         bool                    lo_sent = false;
1821
1822         /*
1823          * NB: rtr_nid is set to LNET_NID_ANY for all current use-cases,
1824          * but we might want to use pre-determined router for ACK/REPLY
1825          * in the future
1826          */
1827         /* NB: ni != NULL == interface pre-determined (ACK/REPLY) */
1828         LASSERT (msg->msg_txpeer == NULL);
1829         LASSERT (!msg->msg_sending);
1830         LASSERT (!msg->msg_target_is_router);
1831         LASSERT (!msg->msg_receiving);
1832
1833         msg->msg_sending = 1;
1834
1835         LASSERT(!msg->msg_tx_committed);
1836
1837         rc = lnet_select_pathway(src_nid, dst_nid, msg, rtr_nid, &lo_sent);
1838         if (rc < 0 || lo_sent)
1839                 return rc;
1840
1841         if (rc == LNET_CREDIT_OK)
1842                 lnet_ni_send(msg->msg_txni, msg);
1843
1844         /* rc == LNET_CREDIT_OK or LNET_CREDIT_WAIT */
1845         return 0;
1846 }
1847
1848 void
1849 lnet_drop_message(lnet_ni_t *ni, int cpt, void *private, unsigned int nob)
1850 {
1851         lnet_net_lock(cpt);
1852         the_lnet.ln_counters[cpt]->drop_count++;
1853         the_lnet.ln_counters[cpt]->drop_length += nob;
1854         lnet_net_unlock(cpt);
1855
1856         lnet_ni_recv(ni, private, NULL, 0, 0, 0, nob);
1857 }
1858
1859 static void
1860 lnet_recv_put(lnet_ni_t *ni, lnet_msg_t *msg)
1861 {
1862         lnet_hdr_t      *hdr = &msg->msg_hdr;
1863
1864         if (msg->msg_wanted != 0)
1865                 lnet_setpayloadbuffer(msg);
1866
1867         lnet_build_msg_event(msg, LNET_EVENT_PUT);
1868
1869         /* Must I ACK?  If so I'll grab the ack_wmd out of the header and put
1870          * it back into the ACK during lnet_finalize() */
1871         msg->msg_ack = (!lnet_is_wire_handle_none(&hdr->msg.put.ack_wmd) &&
1872                         (msg->msg_md->md_options & LNET_MD_ACK_DISABLE) == 0);
1873
1874         lnet_ni_recv(ni, msg->msg_private, msg, msg->msg_rx_delayed,
1875                      msg->msg_offset, msg->msg_wanted, hdr->payload_length);
1876 }
1877
1878 static int
1879 lnet_parse_put(lnet_ni_t *ni, lnet_msg_t *msg)
1880 {
1881         lnet_hdr_t              *hdr = &msg->msg_hdr;
1882         struct lnet_match_info  info;
1883         int                     rc;
1884         bool                    ready_delay;
1885
1886         /* Convert put fields to host byte order */
1887         hdr->msg.put.match_bits = le64_to_cpu(hdr->msg.put.match_bits);
1888         hdr->msg.put.ptl_index  = le32_to_cpu(hdr->msg.put.ptl_index);
1889         hdr->msg.put.offset     = le32_to_cpu(hdr->msg.put.offset);
1890
1891         /* Primary peer NID. */
1892         info.mi_id.nid  = msg->msg_initiator;
1893         info.mi_id.pid  = hdr->src_pid;
1894         info.mi_opc     = LNET_MD_OP_PUT;
1895         info.mi_portal  = hdr->msg.put.ptl_index;
1896         info.mi_rlength = hdr->payload_length;
1897         info.mi_roffset = hdr->msg.put.offset;
1898         info.mi_mbits   = hdr->msg.put.match_bits;
1899         info.mi_cpt     = lnet_cpt_of_nid(msg->msg_rxpeer->lpni_nid, ni);
1900
1901         msg->msg_rx_ready_delay = ni->ni_net->net_lnd->lnd_eager_recv == NULL;
1902         ready_delay = msg->msg_rx_ready_delay;
1903
1904  again:
1905         rc = lnet_ptl_match_md(&info, msg);
1906         switch (rc) {
1907         default:
1908                 LBUG();
1909
1910         case LNET_MATCHMD_OK:
1911                 lnet_recv_put(ni, msg);
1912                 return 0;
1913
1914         case LNET_MATCHMD_NONE:
1915                 if (ready_delay)
1916                         /* no eager_recv or has already called it, should
1917                          * have been attached on delayed list */
1918                         return 0;
1919
1920                 rc = lnet_ni_eager_recv(ni, msg);
1921                 if (rc == 0) {
1922                         ready_delay = true;
1923                         goto again;
1924                 }
1925                 /* fall through */
1926
1927         case LNET_MATCHMD_DROP:
1928                 CNETERR("Dropping PUT from %s portal %d match %llu"
1929                         " offset %d length %d: %d\n",
1930                         libcfs_id2str(info.mi_id), info.mi_portal,
1931                         info.mi_mbits, info.mi_roffset, info.mi_rlength, rc);
1932
1933                 return -ENOENT; /* -ve: OK but no match */
1934         }
1935 }
1936
1937 static int
1938 lnet_parse_get(lnet_ni_t *ni, lnet_msg_t *msg, int rdma_get)
1939 {
1940         struct lnet_match_info  info;
1941         lnet_hdr_t              *hdr = &msg->msg_hdr;
1942         lnet_process_id_t       source_id;
1943         struct lnet_handle_wire reply_wmd;
1944         int                     rc;
1945
1946         /* Convert get fields to host byte order */
1947         hdr->msg.get.match_bits   = le64_to_cpu(hdr->msg.get.match_bits);
1948         hdr->msg.get.ptl_index    = le32_to_cpu(hdr->msg.get.ptl_index);
1949         hdr->msg.get.sink_length  = le32_to_cpu(hdr->msg.get.sink_length);
1950         hdr->msg.get.src_offset   = le32_to_cpu(hdr->msg.get.src_offset);
1951
1952         source_id.nid = hdr->src_nid;
1953         source_id.pid = hdr->src_pid;
1954         /* Primary peer NID */
1955         info.mi_id.nid  = msg->msg_initiator;
1956         info.mi_id.pid  = hdr->src_pid;
1957         info.mi_opc     = LNET_MD_OP_GET;
1958         info.mi_portal  = hdr->msg.get.ptl_index;
1959         info.mi_rlength = hdr->msg.get.sink_length;
1960         info.mi_roffset = hdr->msg.get.src_offset;
1961         info.mi_mbits   = hdr->msg.get.match_bits;
1962
1963         rc = lnet_ptl_match_md(&info, msg);
1964         if (rc == LNET_MATCHMD_DROP) {
1965                 CNETERR("Dropping GET from %s portal %d match %llu"
1966                         " offset %d length %d\n",
1967                         libcfs_id2str(info.mi_id), info.mi_portal,
1968                         info.mi_mbits, info.mi_roffset, info.mi_rlength);
1969                 return -ENOENT; /* -ve: OK but no match */
1970         }
1971
1972         LASSERT(rc == LNET_MATCHMD_OK);
1973
1974         lnet_build_msg_event(msg, LNET_EVENT_GET);
1975
1976         reply_wmd = hdr->msg.get.return_wmd;
1977
1978         lnet_prep_send(msg, LNET_MSG_REPLY, source_id,
1979                        msg->msg_offset, msg->msg_wanted);
1980
1981         msg->msg_hdr.msg.reply.dst_wmd = reply_wmd;
1982
1983         if (rdma_get) {
1984                 /* The LND completes the REPLY from her recv procedure */
1985                 lnet_ni_recv(ni, msg->msg_private, msg, 0,
1986                              msg->msg_offset, msg->msg_len, msg->msg_len);
1987                 return 0;
1988         }
1989
1990         lnet_ni_recv(ni, msg->msg_private, NULL, 0, 0, 0, 0);
1991         msg->msg_receiving = 0;
1992
1993         rc = lnet_send(ni->ni_nid, msg, LNET_NID_ANY);
1994         if (rc < 0) {
1995                 /* didn't get as far as lnet_ni_send() */
1996                 CERROR("%s: Unable to send REPLY for GET from %s: %d\n",
1997                        libcfs_nid2str(ni->ni_nid),
1998                        libcfs_id2str(info.mi_id), rc);
1999
2000                 lnet_finalize(ni, msg, rc);
2001         }
2002
2003         return 0;
2004 }
2005
2006 static int
2007 lnet_parse_reply(lnet_ni_t *ni, lnet_msg_t *msg)
2008 {
2009         void             *private = msg->msg_private;
2010         lnet_hdr_t       *hdr = &msg->msg_hdr;
2011         lnet_process_id_t src = {0};
2012         lnet_libmd_t     *md;
2013         int               rlength;
2014         int               mlength;
2015         int                     cpt;
2016
2017         cpt = lnet_cpt_of_cookie(hdr->msg.reply.dst_wmd.wh_object_cookie);
2018         lnet_res_lock(cpt);
2019
2020         src.nid = hdr->src_nid;
2021         src.pid = hdr->src_pid;
2022
2023         /* NB handles only looked up by creator (no flips) */
2024         md = lnet_wire_handle2md(&hdr->msg.reply.dst_wmd);
2025         if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
2026                 CNETERR("%s: Dropping REPLY from %s for %s "
2027                         "MD %#llx.%#llx\n",
2028                         libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
2029                         (md == NULL) ? "invalid" : "inactive",
2030                         hdr->msg.reply.dst_wmd.wh_interface_cookie,
2031                         hdr->msg.reply.dst_wmd.wh_object_cookie);
2032                 if (md != NULL && md->md_me != NULL)
2033                         CERROR("REPLY MD also attached to portal %d\n",
2034                                md->md_me->me_portal);
2035
2036                 lnet_res_unlock(cpt);
2037                 return -ENOENT; /* -ve: OK but no match */
2038         }
2039
2040         LASSERT(md->md_offset == 0);
2041
2042         rlength = hdr->payload_length;
2043         mlength = MIN(rlength, (int)md->md_length);
2044
2045         if (mlength < rlength &&
2046             (md->md_options & LNET_MD_TRUNCATE) == 0) {
2047                 CNETERR("%s: Dropping REPLY from %s length %d "
2048                         "for MD %#llx would overflow (%d)\n",
2049                         libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
2050                         rlength, hdr->msg.reply.dst_wmd.wh_object_cookie,
2051                         mlength);
2052                 lnet_res_unlock(cpt);
2053                 return -ENOENT; /* -ve: OK but no match */
2054         }
2055
2056         CDEBUG(D_NET, "%s: Reply from %s of length %d/%d into md %#llx\n",
2057                libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
2058                mlength, rlength, hdr->msg.reply.dst_wmd.wh_object_cookie);
2059
2060         lnet_msg_attach_md(msg, md, 0, mlength);
2061
2062         if (mlength != 0)
2063                 lnet_setpayloadbuffer(msg);
2064
2065         lnet_res_unlock(cpt);
2066
2067         lnet_build_msg_event(msg, LNET_EVENT_REPLY);
2068
2069         lnet_ni_recv(ni, private, msg, 0, 0, mlength, rlength);
2070         return 0;
2071 }
2072
2073 static int
2074 lnet_parse_ack(lnet_ni_t *ni, lnet_msg_t *msg)
2075 {
2076         lnet_hdr_t       *hdr = &msg->msg_hdr;
2077         lnet_process_id_t src = {0};
2078         lnet_libmd_t     *md;
2079         int                     cpt;
2080
2081         src.nid = hdr->src_nid;
2082         src.pid = hdr->src_pid;
2083
2084         /* Convert ack fields to host byte order */
2085         hdr->msg.ack.match_bits = le64_to_cpu(hdr->msg.ack.match_bits);
2086         hdr->msg.ack.mlength = le32_to_cpu(hdr->msg.ack.mlength);
2087
2088         cpt = lnet_cpt_of_cookie(hdr->msg.ack.dst_wmd.wh_object_cookie);
2089         lnet_res_lock(cpt);
2090
2091         /* NB handles only looked up by creator (no flips) */
2092         md = lnet_wire_handle2md(&hdr->msg.ack.dst_wmd);
2093         if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
2094                 /* Don't moan; this is expected */
2095                 CDEBUG(D_NET,
2096                        "%s: Dropping ACK from %s to %s MD %#llx.%#llx\n",
2097                        libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
2098                        (md == NULL) ? "invalid" : "inactive",
2099                        hdr->msg.ack.dst_wmd.wh_interface_cookie,
2100                        hdr->msg.ack.dst_wmd.wh_object_cookie);
2101                 if (md != NULL && md->md_me != NULL)
2102                         CERROR("Source MD also attached to portal %d\n",
2103                                md->md_me->me_portal);
2104
2105                 lnet_res_unlock(cpt);
2106                 return -ENOENT;                  /* -ve! */
2107         }
2108
2109         CDEBUG(D_NET, "%s: ACK from %s into md %#llx\n",
2110                libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
2111                hdr->msg.ack.dst_wmd.wh_object_cookie);
2112
2113         lnet_msg_attach_md(msg, md, 0, 0);
2114
2115         lnet_res_unlock(cpt);
2116
2117         lnet_build_msg_event(msg, LNET_EVENT_ACK);
2118
2119         lnet_ni_recv(ni, msg->msg_private, msg, 0, 0, 0, msg->msg_len);
2120         return 0;
2121 }
2122
2123 /**
2124  * \retval LNET_CREDIT_OK       If \a msg is forwarded
2125  * \retval LNET_CREDIT_WAIT     If \a msg is blocked because w/o buffer
2126  * \retval -ve                  error code
2127  */
2128 int
2129 lnet_parse_forward_locked(lnet_ni_t *ni, lnet_msg_t *msg)
2130 {
2131         int     rc = 0;
2132
2133         if (!the_lnet.ln_routing)
2134                 return -ECANCELED;
2135
2136         if (msg->msg_rxpeer->lpni_rtrcredits <= 0 ||
2137             lnet_msg2bufpool(msg)->rbp_credits <= 0) {
2138                 if (ni->ni_net->net_lnd->lnd_eager_recv == NULL) {
2139                         msg->msg_rx_ready_delay = 1;
2140                 } else {
2141                         lnet_net_unlock(msg->msg_rx_cpt);
2142                         rc = lnet_ni_eager_recv(ni, msg);
2143                         lnet_net_lock(msg->msg_rx_cpt);
2144                 }
2145         }
2146
2147         if (rc == 0)
2148                 rc = lnet_post_routed_recv_locked(msg, 0);
2149         return rc;
2150 }
2151
2152 int
2153 lnet_parse_local(lnet_ni_t *ni, lnet_msg_t *msg)
2154 {
2155         int     rc;
2156
2157         switch (msg->msg_type) {
2158         case LNET_MSG_ACK:
2159                 rc = lnet_parse_ack(ni, msg);
2160                 break;
2161         case LNET_MSG_PUT:
2162                 rc = lnet_parse_put(ni, msg);
2163                 break;
2164         case LNET_MSG_GET:
2165                 rc = lnet_parse_get(ni, msg, msg->msg_rdma_get);
2166                 break;
2167         case LNET_MSG_REPLY:
2168                 rc = lnet_parse_reply(ni, msg);
2169                 break;
2170         default: /* prevent an unused label if !kernel */
2171                 LASSERT(0);
2172                 return -EPROTO;
2173         }
2174
2175         LASSERT(rc == 0 || rc == -ENOENT);
2176         return rc;
2177 }
2178
2179 char *
2180 lnet_msgtyp2str (int type)
2181 {
2182         switch (type) {
2183         case LNET_MSG_ACK:
2184                 return ("ACK");
2185         case LNET_MSG_PUT:
2186                 return ("PUT");
2187         case LNET_MSG_GET:
2188                 return ("GET");
2189         case LNET_MSG_REPLY:
2190                 return ("REPLY");
2191         case LNET_MSG_HELLO:
2192                 return ("HELLO");
2193         default:
2194                 return ("<UNKNOWN>");
2195         }
2196 }
2197
2198 void
2199 lnet_print_hdr(lnet_hdr_t * hdr)
2200 {
2201         lnet_process_id_t src = {0};
2202         lnet_process_id_t dst = {0};
2203         char *type_str = lnet_msgtyp2str(hdr->type);
2204
2205         src.nid = hdr->src_nid;
2206         src.pid = hdr->src_pid;
2207
2208         dst.nid = hdr->dest_nid;
2209         dst.pid = hdr->dest_pid;
2210
2211         CWARN("P3 Header at %p of type %s\n", hdr, type_str);
2212         CWARN("    From %s\n", libcfs_id2str(src));
2213         CWARN("    To   %s\n", libcfs_id2str(dst));
2214
2215         switch (hdr->type) {
2216         default:
2217                 break;
2218
2219         case LNET_MSG_PUT:
2220                 CWARN("    Ptl index %d, ack md %#llx.%#llx, "
2221                       "match bits %llu\n",
2222                       hdr->msg.put.ptl_index,
2223                       hdr->msg.put.ack_wmd.wh_interface_cookie,
2224                       hdr->msg.put.ack_wmd.wh_object_cookie,
2225                       hdr->msg.put.match_bits);
2226                 CWARN("    Length %d, offset %d, hdr data %#llx\n",
2227                       hdr->payload_length, hdr->msg.put.offset,
2228                       hdr->msg.put.hdr_data);
2229                 break;
2230
2231         case LNET_MSG_GET:
2232                 CWARN("    Ptl index %d, return md %#llx.%#llx, "
2233                       "match bits %llu\n", hdr->msg.get.ptl_index,
2234                       hdr->msg.get.return_wmd.wh_interface_cookie,
2235                       hdr->msg.get.return_wmd.wh_object_cookie,
2236                       hdr->msg.get.match_bits);
2237                 CWARN("    Length %d, src offset %d\n",
2238                       hdr->msg.get.sink_length,
2239                       hdr->msg.get.src_offset);
2240                 break;
2241
2242         case LNET_MSG_ACK:
2243                 CWARN("    dst md %#llx.%#llx, "
2244                       "manipulated length %d\n",
2245                       hdr->msg.ack.dst_wmd.wh_interface_cookie,
2246                       hdr->msg.ack.dst_wmd.wh_object_cookie,
2247                       hdr->msg.ack.mlength);
2248                 break;
2249
2250         case LNET_MSG_REPLY:
2251                 CWARN("    dst md %#llx.%#llx, "
2252                       "length %d\n",
2253                       hdr->msg.reply.dst_wmd.wh_interface_cookie,
2254                       hdr->msg.reply.dst_wmd.wh_object_cookie,
2255                       hdr->payload_length);
2256         }
2257
2258 }
2259
2260 int
2261 lnet_parse(lnet_ni_t *ni, lnet_hdr_t *hdr, lnet_nid_t from_nid,
2262            void *private, int rdma_req)
2263 {
2264         int             rc = 0;
2265         int             cpt;
2266         int             for_me;
2267         struct lnet_msg *msg;
2268         lnet_pid_t     dest_pid;
2269         lnet_nid_t     dest_nid;
2270         lnet_nid_t     src_nid;
2271         struct lnet_peer_ni *lpni;
2272         __u32          payload_length;
2273         __u32          type;
2274
2275         LASSERT (!in_interrupt ());
2276
2277         type = le32_to_cpu(hdr->type);
2278         src_nid = le64_to_cpu(hdr->src_nid);
2279         dest_nid = le64_to_cpu(hdr->dest_nid);
2280         dest_pid = le32_to_cpu(hdr->dest_pid);
2281         payload_length = le32_to_cpu(hdr->payload_length);
2282
2283         for_me = (ni->ni_nid == dest_nid);
2284         cpt = lnet_cpt_of_nid(from_nid, ni);
2285
2286         switch (type) {
2287         case LNET_MSG_ACK:
2288         case LNET_MSG_GET:
2289                 if (payload_length > 0) {
2290                         CERROR("%s, src %s: bad %s payload %d (0 expected)\n",
2291                                libcfs_nid2str(from_nid),
2292                                libcfs_nid2str(src_nid),
2293                                lnet_msgtyp2str(type), payload_length);
2294                         return -EPROTO;
2295                 }
2296                 break;
2297
2298         case LNET_MSG_PUT:
2299         case LNET_MSG_REPLY:
2300                 if (payload_length >
2301                     (__u32)(for_me ? LNET_MAX_PAYLOAD : LNET_MTU)) {
2302                         CERROR("%s, src %s: bad %s payload %d "
2303                                "(%d max expected)\n",
2304                                libcfs_nid2str(from_nid),
2305                                libcfs_nid2str(src_nid),
2306                                lnet_msgtyp2str(type),
2307                                payload_length,
2308                                for_me ? LNET_MAX_PAYLOAD : LNET_MTU);
2309                         return -EPROTO;
2310                 }
2311                 break;
2312
2313         default:
2314                 CERROR("%s, src %s: Bad message type 0x%x\n",
2315                        libcfs_nid2str(from_nid),
2316                        libcfs_nid2str(src_nid), type);
2317                 return -EPROTO;
2318         }
2319
2320         if (the_lnet.ln_routing &&
2321             ni->ni_last_alive != ktime_get_real_seconds()) {
2322                 /* NB: so far here is the only place to set NI status to "up */
2323                 lnet_ni_lock(ni);
2324                 ni->ni_last_alive = ktime_get_real_seconds();
2325                 if (ni->ni_status != NULL &&
2326                     ni->ni_status->ns_status == LNET_NI_STATUS_DOWN)
2327                         ni->ni_status->ns_status = LNET_NI_STATUS_UP;
2328                 lnet_ni_unlock(ni);
2329         }
2330
2331         /* Regard a bad destination NID as a protocol error.  Senders should
2332          * know what they're doing; if they don't they're misconfigured, buggy
2333          * or malicious so we chop them off at the knees :) */
2334
2335         if (!for_me) {
2336                 if (LNET_NIDNET(dest_nid) == LNET_NIDNET(ni->ni_nid)) {
2337                         /* should have gone direct */
2338                         CERROR("%s, src %s: Bad dest nid %s "
2339                                "(should have been sent direct)\n",
2340                                 libcfs_nid2str(from_nid),
2341                                 libcfs_nid2str(src_nid),
2342                                 libcfs_nid2str(dest_nid));
2343                         return -EPROTO;
2344                 }
2345
2346                 if (lnet_islocalnid(dest_nid)) {
2347                         /* dest is another local NI; sender should have used
2348                          * this node's NID on its own network */
2349                         CERROR("%s, src %s: Bad dest nid %s "
2350                                "(it's my nid but on a different network)\n",
2351                                 libcfs_nid2str(from_nid),
2352                                 libcfs_nid2str(src_nid),
2353                                 libcfs_nid2str(dest_nid));
2354                         return -EPROTO;
2355                 }
2356
2357                 if (rdma_req && type == LNET_MSG_GET) {
2358                         CERROR("%s, src %s: Bad optimized GET for %s "
2359                                "(final destination must be me)\n",
2360                                 libcfs_nid2str(from_nid),
2361                                 libcfs_nid2str(src_nid),
2362                                 libcfs_nid2str(dest_nid));
2363                         return -EPROTO;
2364                 }
2365
2366                 if (!the_lnet.ln_routing) {
2367                         CERROR("%s, src %s: Dropping message for %s "
2368                                "(routing not enabled)\n",
2369                                 libcfs_nid2str(from_nid),
2370                                 libcfs_nid2str(src_nid),
2371                                 libcfs_nid2str(dest_nid));
2372                         goto drop;
2373                 }
2374         }
2375
2376         /* Message looks OK; we're not going to return an error, so we MUST
2377          * call back lnd_recv() come what may... */
2378
2379         if (!list_empty(&the_lnet.ln_test_peers) &&     /* normally we don't */
2380             fail_peer(src_nid, 0)) {                    /* shall we now? */
2381                 CERROR("%s, src %s: Dropping %s to simulate failure\n",
2382                        libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
2383                        lnet_msgtyp2str(type));
2384                 goto drop;
2385         }
2386
2387         if (!list_empty(&the_lnet.ln_drop_rules) &&
2388             lnet_drop_rule_match(hdr)) {
2389                 CDEBUG(D_NET, "%s, src %s, dst %s: Dropping %s to simulate"
2390                               "silent message loss\n",
2391                        libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
2392                        libcfs_nid2str(dest_nid), lnet_msgtyp2str(type));
2393                 goto drop;
2394         }
2395
2396
2397         msg = lnet_msg_alloc();
2398         if (msg == NULL) {
2399                 CERROR("%s, src %s: Dropping %s (out of memory)\n",
2400                        libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
2401                        lnet_msgtyp2str(type));
2402                 goto drop;
2403         }
2404
2405         /* msg zeroed in lnet_msg_alloc; i.e. flags all clear,
2406          * pointers NULL etc */
2407
2408         msg->msg_type = type;
2409         msg->msg_private = private;
2410         msg->msg_receiving = 1;
2411         msg->msg_rdma_get = rdma_req;
2412         msg->msg_len = msg->msg_wanted = payload_length;
2413         msg->msg_offset = 0;
2414         msg->msg_hdr = *hdr;
2415         /* for building message event */
2416         msg->msg_from = from_nid;
2417         if (!for_me) {
2418                 msg->msg_target.pid     = dest_pid;
2419                 msg->msg_target.nid     = dest_nid;
2420                 msg->msg_routing        = 1;
2421
2422         } else {
2423                 /* convert common msg->hdr fields to host byteorder */
2424                 msg->msg_hdr.type       = type;
2425                 msg->msg_hdr.src_nid    = src_nid;
2426                 msg->msg_hdr.src_pid    = le32_to_cpu(msg->msg_hdr.src_pid);
2427                 msg->msg_hdr.dest_nid   = dest_nid;
2428                 msg->msg_hdr.dest_pid   = dest_pid;
2429                 msg->msg_hdr.payload_length = payload_length;
2430         }
2431         /* Multi-Rail: Primary NID of source. */
2432         msg->msg_initiator = lnet_peer_primary_nid(src_nid);
2433
2434         lnet_net_lock(cpt);
2435         lpni = lnet_nid2peerni_locked(from_nid, cpt);
2436         if (IS_ERR(lpni)) {
2437                 lnet_net_unlock(cpt);
2438                 CERROR("%s, src %s: Dropping %s "
2439                        "(error %ld looking up sender)\n",
2440                        libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
2441                        lnet_msgtyp2str(type), PTR_ERR(lpni));
2442                 lnet_msg_free(msg);
2443                 if (rc == -ESHUTDOWN)
2444                         /* We are shutting down.  Don't do anything more */
2445                         return 0;
2446                 goto drop;
2447         }
2448         msg->msg_rxpeer = lpni;
2449         msg->msg_rxni = ni;
2450         lnet_ni_addref_locked(ni, cpt);
2451
2452         if (lnet_isrouter(msg->msg_rxpeer)) {
2453                 lnet_peer_set_alive(msg->msg_rxpeer);
2454                 if (avoid_asym_router_failure &&
2455                     LNET_NIDNET(src_nid) != LNET_NIDNET(from_nid)) {
2456                         /* received a remote message from router, update
2457                          * remote NI status on this router.
2458                          * NB: multi-hop routed message will be ignored.
2459                          */
2460                         lnet_router_ni_update_locked(msg->msg_rxpeer,
2461                                                      LNET_NIDNET(src_nid));
2462                 }
2463         }
2464
2465         lnet_msg_commit(msg, cpt);
2466
2467         /* message delay simulation */
2468         if (unlikely(!list_empty(&the_lnet.ln_delay_rules) &&
2469                      lnet_delay_rule_match_locked(hdr, msg))) {
2470                 lnet_net_unlock(cpt);
2471                 return 0;
2472         }
2473
2474         if (!for_me) {
2475                 rc = lnet_parse_forward_locked(ni, msg);
2476                 lnet_net_unlock(cpt);
2477
2478                 if (rc < 0)
2479                         goto free_drop;
2480
2481                 if (rc == LNET_CREDIT_OK) {
2482                         lnet_ni_recv(ni, msg->msg_private, msg, 0,
2483                                      0, payload_length, payload_length);
2484                 }
2485                 return 0;
2486         }
2487
2488         lnet_net_unlock(cpt);
2489
2490         rc = lnet_parse_local(ni, msg);
2491         if (rc != 0)
2492                 goto free_drop;
2493         return 0;
2494
2495  free_drop:
2496         LASSERT(msg->msg_md == NULL);
2497         lnet_finalize(ni, msg, rc);
2498
2499  drop:
2500         lnet_drop_message(ni, cpt, private, payload_length);
2501         return 0;
2502 }
2503 EXPORT_SYMBOL(lnet_parse);
2504
2505 void
2506 lnet_drop_delayed_msg_list(struct list_head *head, char *reason)
2507 {
2508         while (!list_empty(head)) {
2509                 lnet_process_id_t       id = {0};
2510                 lnet_msg_t              *msg;
2511
2512                 msg = list_entry(head->next, lnet_msg_t, msg_list);
2513                 list_del(&msg->msg_list);
2514
2515                 id.nid = msg->msg_hdr.src_nid;
2516                 id.pid = msg->msg_hdr.src_pid;
2517
2518                 LASSERT(msg->msg_md == NULL);
2519                 LASSERT(msg->msg_rx_delayed);
2520                 LASSERT(msg->msg_rxpeer != NULL);
2521                 LASSERT(msg->msg_hdr.type == LNET_MSG_PUT);
2522
2523                 CWARN("Dropping delayed PUT from %s portal %d match %llu"
2524                       " offset %d length %d: %s\n",
2525                       libcfs_id2str(id),
2526                       msg->msg_hdr.msg.put.ptl_index,
2527                       msg->msg_hdr.msg.put.match_bits,
2528                       msg->msg_hdr.msg.put.offset,
2529                       msg->msg_hdr.payload_length, reason);
2530
2531                 /* NB I can't drop msg's ref on msg_rxpeer until after I've
2532                  * called lnet_drop_message(), so I just hang onto msg as well
2533                  * until that's done */
2534
2535                 lnet_drop_message(msg->msg_rxni, msg->msg_rx_cpt,
2536                                   msg->msg_private, msg->msg_len);
2537                 /*
2538                  * NB: message will not generate event because w/o attached MD,
2539                  * but we still should give error code so lnet_msg_decommit()
2540                  * can skip counters operations and other checks.
2541                  */
2542                 lnet_finalize(msg->msg_rxni, msg, -ENOENT);
2543         }
2544 }
2545
2546 void
2547 lnet_recv_delayed_msg_list(struct list_head *head)
2548 {
2549         while (!list_empty(head)) {
2550                 lnet_msg_t        *msg;
2551                 lnet_process_id_t  id;
2552
2553                 msg = list_entry(head->next, lnet_msg_t, msg_list);
2554                 list_del(&msg->msg_list);
2555
2556                 /* md won't disappear under me, since each msg
2557                  * holds a ref on it */
2558
2559                 id.nid = msg->msg_hdr.src_nid;
2560                 id.pid = msg->msg_hdr.src_pid;
2561
2562                 LASSERT(msg->msg_rx_delayed);
2563                 LASSERT(msg->msg_md != NULL);
2564                 LASSERT(msg->msg_rxpeer != NULL);
2565                 LASSERT(msg->msg_rxni != NULL);
2566                 LASSERT(msg->msg_hdr.type == LNET_MSG_PUT);
2567
2568                 CDEBUG(D_NET, "Resuming delayed PUT from %s portal %d "
2569                        "match %llu offset %d length %d.\n",
2570                         libcfs_id2str(id), msg->msg_hdr.msg.put.ptl_index,
2571                         msg->msg_hdr.msg.put.match_bits,
2572                         msg->msg_hdr.msg.put.offset,
2573                         msg->msg_hdr.payload_length);
2574
2575                 lnet_recv_put(msg->msg_rxni, msg);
2576         }
2577 }
2578
2579 /**
2580  * Initiate an asynchronous PUT operation.
2581  *
2582  * There are several events associated with a PUT: completion of the send on
2583  * the initiator node (LNET_EVENT_SEND), and when the send completes
2584  * successfully, the receipt of an acknowledgment (LNET_EVENT_ACK) indicating
2585  * that the operation was accepted by the target. The event LNET_EVENT_PUT is
2586  * used at the target node to indicate the completion of incoming data
2587  * delivery.
2588  *
2589  * The local events will be logged in the EQ associated with the MD pointed to
2590  * by \a mdh handle. Using a MD without an associated EQ results in these
2591  * events being discarded. In this case, the caller must have another
2592  * mechanism (e.g., a higher level protocol) for determining when it is safe
2593  * to modify the memory region associated with the MD.
2594  *
2595  * Note that LNet does not guarantee the order of LNET_EVENT_SEND and
2596  * LNET_EVENT_ACK, though intuitively ACK should happen after SEND.
2597  *
2598  * \param self Indicates the NID of a local interface through which to send
2599  * the PUT request. Use LNET_NID_ANY to let LNet choose one by itself.
2600  * \param mdh A handle for the MD that describes the memory to be sent. The MD
2601  * must be "free floating" (See LNetMDBind()).
2602  * \param ack Controls whether an acknowledgment is requested.
2603  * Acknowledgments are only sent when they are requested by the initiating
2604  * process and the target MD enables them.
2605  * \param target A process identifier for the target process.
2606  * \param portal The index in the \a target's portal table.
2607  * \param match_bits The match bits to use for MD selection at the target
2608  * process.
2609  * \param offset The offset into the target MD (only used when the target
2610  * MD has the LNET_MD_MANAGE_REMOTE option set).
2611  * \param hdr_data 64 bits of user data that can be included in the message
2612  * header. This data is written to an event queue entry at the target if an
2613  * EQ is present on the matching MD.
2614  *
2615  * \retval  0      Success, and only in this case events will be generated
2616  * and logged to EQ (if it exists).
2617  * \retval -EIO    Simulated failure.
2618  * \retval -ENOMEM Memory allocation failure.
2619  * \retval -ENOENT Invalid MD object.
2620  *
2621  * \see lnet_event_t::hdr_data and lnet_event_kind_t.
2622  */
2623 int
2624 LNetPut(lnet_nid_t self, lnet_handle_md_t mdh, lnet_ack_req_t ack,
2625         lnet_process_id_t target, unsigned int portal,
2626         __u64 match_bits, unsigned int offset,
2627         __u64 hdr_data)
2628 {
2629         struct lnet_msg         *msg;
2630         struct lnet_libmd       *md;
2631         int                     cpt;
2632         int                     rc;
2633
2634         LASSERT(the_lnet.ln_refcount > 0);
2635
2636         if (!list_empty(&the_lnet.ln_test_peers) &&     /* normally we don't */
2637             fail_peer(target.nid, 1)) {                 /* shall we now? */
2638                 CERROR("Dropping PUT to %s: simulated failure\n",
2639                        libcfs_id2str(target));
2640                 return -EIO;
2641         }
2642
2643         msg = lnet_msg_alloc();
2644         if (msg == NULL) {
2645                 CERROR("Dropping PUT to %s: ENOMEM on lnet_msg_t\n",
2646                        libcfs_id2str(target));
2647                 return -ENOMEM;
2648         }
2649         msg->msg_vmflush = !!memory_pressure_get();
2650
2651         cpt = lnet_cpt_of_cookie(mdh.cookie);
2652         lnet_res_lock(cpt);
2653
2654         md = lnet_handle2md(&mdh);
2655         if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
2656                 CERROR("Dropping PUT (%llu:%d:%s): MD (%d) invalid\n",
2657                        match_bits, portal, libcfs_id2str(target),
2658                        md == NULL ? -1 : md->md_threshold);
2659                 if (md != NULL && md->md_me != NULL)
2660                         CERROR("Source MD also attached to portal %d\n",
2661                                md->md_me->me_portal);
2662                 lnet_res_unlock(cpt);
2663
2664                 lnet_msg_free(msg);
2665                 return -ENOENT;
2666         }
2667
2668         CDEBUG(D_NET, "LNetPut -> %s\n", libcfs_id2str(target));
2669
2670         lnet_msg_attach_md(msg, md, 0, 0);
2671
2672         lnet_prep_send(msg, LNET_MSG_PUT, target, 0, md->md_length);
2673
2674         msg->msg_hdr.msg.put.match_bits = cpu_to_le64(match_bits);
2675         msg->msg_hdr.msg.put.ptl_index = cpu_to_le32(portal);
2676         msg->msg_hdr.msg.put.offset = cpu_to_le32(offset);
2677         msg->msg_hdr.msg.put.hdr_data = hdr_data;
2678
2679         /* NB handles only looked up by creator (no flips) */
2680         if (ack == LNET_ACK_REQ) {
2681                 msg->msg_hdr.msg.put.ack_wmd.wh_interface_cookie =
2682                         the_lnet.ln_interface_cookie;
2683                 msg->msg_hdr.msg.put.ack_wmd.wh_object_cookie =
2684                         md->md_lh.lh_cookie;
2685         } else {
2686                 msg->msg_hdr.msg.put.ack_wmd.wh_interface_cookie =
2687                         LNET_WIRE_HANDLE_COOKIE_NONE;
2688                 msg->msg_hdr.msg.put.ack_wmd.wh_object_cookie =
2689                         LNET_WIRE_HANDLE_COOKIE_NONE;
2690         }
2691
2692         lnet_res_unlock(cpt);
2693
2694         lnet_build_msg_event(msg, LNET_EVENT_SEND);
2695
2696         rc = lnet_send(self, msg, LNET_NID_ANY);
2697         if (rc != 0) {
2698                 CNETERR("Error sending PUT to %s: %d\n",
2699                         libcfs_id2str(target), rc);
2700                 lnet_finalize(NULL, msg, rc);
2701         }
2702
2703         /* completion will be signalled by an event */
2704         return 0;
2705 }
2706 EXPORT_SYMBOL(LNetPut);
2707
2708 lnet_msg_t *
2709 lnet_create_reply_msg (lnet_ni_t *ni, lnet_msg_t *getmsg)
2710 {
2711         /* The LND can DMA direct to the GET md (i.e. no REPLY msg).  This
2712          * returns a msg for the LND to pass to lnet_finalize() when the sink
2713          * data has been received.
2714          *
2715          * CAVEAT EMPTOR: 'getmsg' is the original GET, which is freed when
2716          * lnet_finalize() is called on it, so the LND must call this first */
2717
2718         struct lnet_msg         *msg = lnet_msg_alloc();
2719         struct lnet_libmd       *getmd = getmsg->msg_md;
2720         lnet_process_id_t       peer_id = getmsg->msg_target;
2721         int                     cpt;
2722
2723         LASSERT(!getmsg->msg_target_is_router);
2724         LASSERT(!getmsg->msg_routing);
2725
2726         if (msg == NULL) {
2727                 CERROR("%s: Dropping REPLY from %s: can't allocate msg\n",
2728                        libcfs_nid2str(ni->ni_nid), libcfs_id2str(peer_id));
2729                 goto drop;
2730         }
2731
2732         cpt = lnet_cpt_of_cookie(getmd->md_lh.lh_cookie);
2733         lnet_res_lock(cpt);
2734
2735         LASSERT(getmd->md_refcount > 0);
2736
2737         if (getmd->md_threshold == 0) {
2738                 CERROR("%s: Dropping REPLY from %s for inactive MD %p\n",
2739                         libcfs_nid2str(ni->ni_nid), libcfs_id2str(peer_id),
2740                         getmd);
2741                 lnet_res_unlock(cpt);
2742                 goto drop;
2743         }
2744
2745         LASSERT(getmd->md_offset == 0);
2746
2747         CDEBUG(D_NET, "%s: Reply from %s md %p\n",
2748                libcfs_nid2str(ni->ni_nid), libcfs_id2str(peer_id), getmd);
2749
2750         /* setup information for lnet_build_msg_event */
2751         msg->msg_initiator = lnet_peer_primary_nid(peer_id.nid);
2752         /* Cheaper: msg->msg_initiator = getmsg->msg_txpeer->lp_nid; */
2753         msg->msg_from = peer_id.nid;
2754         msg->msg_type = LNET_MSG_GET; /* flag this msg as an "optimized" GET */
2755         msg->msg_hdr.src_nid = peer_id.nid;
2756         msg->msg_hdr.payload_length = getmd->md_length;
2757         msg->msg_receiving = 1; /* required by lnet_msg_attach_md */
2758
2759         lnet_msg_attach_md(msg, getmd, getmd->md_offset, getmd->md_length);
2760         lnet_res_unlock(cpt);
2761
2762         cpt = lnet_cpt_of_nid(peer_id.nid, ni);
2763
2764         lnet_net_lock(cpt);
2765         lnet_msg_commit(msg, cpt);
2766         lnet_net_unlock(cpt);
2767
2768         lnet_build_msg_event(msg, LNET_EVENT_REPLY);
2769
2770         return msg;
2771
2772  drop:
2773         cpt = lnet_cpt_of_nid(peer_id.nid, ni);
2774
2775         lnet_net_lock(cpt);
2776         the_lnet.ln_counters[cpt]->drop_count++;
2777         the_lnet.ln_counters[cpt]->drop_length += getmd->md_length;
2778         lnet_net_unlock(cpt);
2779
2780         if (msg != NULL)
2781                 lnet_msg_free(msg);
2782
2783         return NULL;
2784 }
2785 EXPORT_SYMBOL(lnet_create_reply_msg);
2786
2787 void
2788 lnet_set_reply_msg_len(lnet_ni_t *ni, lnet_msg_t *reply, unsigned int len)
2789 {
2790         /* Set the REPLY length, now the RDMA that elides the REPLY message has
2791          * completed and I know it. */
2792         LASSERT(reply != NULL);
2793         LASSERT(reply->msg_type == LNET_MSG_GET);
2794         LASSERT(reply->msg_ev.type == LNET_EVENT_REPLY);
2795
2796         /* NB I trusted my peer to RDMA.  If she tells me she's written beyond
2797          * the end of my buffer, I might as well be dead. */
2798         LASSERT(len <= reply->msg_ev.mlength);
2799
2800         reply->msg_ev.mlength = len;
2801 }
2802 EXPORT_SYMBOL(lnet_set_reply_msg_len);
2803
2804 /**
2805  * Initiate an asynchronous GET operation.
2806  *
2807  * On the initiator node, an LNET_EVENT_SEND is logged when the GET request
2808  * is sent, and an LNET_EVENT_REPLY is logged when the data returned from
2809  * the target node in the REPLY has been written to local MD.
2810  *
2811  * On the target node, an LNET_EVENT_GET is logged when the GET request
2812  * arrives and is accepted into a MD.
2813  *
2814  * \param self,target,portal,match_bits,offset See the discussion in LNetPut().
2815  * \param mdh A handle for the MD that describes the memory into which the
2816  * requested data will be received. The MD must be "free floating" (See LNetMDBind()).
2817  *
2818  * \retval  0      Success, and only in this case events will be generated
2819  * and logged to EQ (if it exists) of the MD.
2820  * \retval -EIO    Simulated failure.
2821  * \retval -ENOMEM Memory allocation failure.
2822  * \retval -ENOENT Invalid MD object.
2823  */
2824 int
2825 LNetGet(lnet_nid_t self, lnet_handle_md_t mdh,
2826         lnet_process_id_t target, unsigned int portal,
2827         __u64 match_bits, unsigned int offset)
2828 {
2829         struct lnet_msg         *msg;
2830         struct lnet_libmd       *md;
2831         int                     cpt;
2832         int                     rc;
2833
2834         LASSERT(the_lnet.ln_refcount > 0);
2835
2836         if (!list_empty(&the_lnet.ln_test_peers) &&     /* normally we don't */
2837             fail_peer(target.nid, 1))                   /* shall we now? */
2838         {
2839                 CERROR("Dropping GET to %s: simulated failure\n",
2840                        libcfs_id2str(target));
2841                 return -EIO;
2842         }
2843
2844         msg = lnet_msg_alloc();
2845         if (msg == NULL) {
2846                 CERROR("Dropping GET to %s: ENOMEM on lnet_msg_t\n",
2847                        libcfs_id2str(target));
2848                 return -ENOMEM;
2849         }
2850
2851         cpt = lnet_cpt_of_cookie(mdh.cookie);
2852         lnet_res_lock(cpt);
2853
2854         md = lnet_handle2md(&mdh);
2855         if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
2856                 CERROR("Dropping GET (%llu:%d:%s): MD (%d) invalid\n",
2857                        match_bits, portal, libcfs_id2str(target),
2858                        md == NULL ? -1 : md->md_threshold);
2859                 if (md != NULL && md->md_me != NULL)
2860                         CERROR("REPLY MD also attached to portal %d\n",
2861                                md->md_me->me_portal);
2862
2863                 lnet_res_unlock(cpt);
2864
2865                 lnet_msg_free(msg);
2866                 return -ENOENT;
2867         }
2868
2869         CDEBUG(D_NET, "LNetGet -> %s\n", libcfs_id2str(target));
2870
2871         lnet_msg_attach_md(msg, md, 0, 0);
2872
2873         lnet_prep_send(msg, LNET_MSG_GET, target, 0, 0);
2874
2875         msg->msg_hdr.msg.get.match_bits = cpu_to_le64(match_bits);
2876         msg->msg_hdr.msg.get.ptl_index = cpu_to_le32(portal);
2877         msg->msg_hdr.msg.get.src_offset = cpu_to_le32(offset);
2878         msg->msg_hdr.msg.get.sink_length = cpu_to_le32(md->md_length);
2879
2880         /* NB handles only looked up by creator (no flips) */
2881         msg->msg_hdr.msg.get.return_wmd.wh_interface_cookie =
2882                 the_lnet.ln_interface_cookie;
2883         msg->msg_hdr.msg.get.return_wmd.wh_object_cookie =
2884                 md->md_lh.lh_cookie;
2885
2886         lnet_res_unlock(cpt);
2887
2888         lnet_build_msg_event(msg, LNET_EVENT_SEND);
2889
2890         rc = lnet_send(self, msg, LNET_NID_ANY);
2891         if (rc < 0) {
2892                 CNETERR("Error sending GET to %s: %d\n",
2893                         libcfs_id2str(target), rc);
2894                 lnet_finalize(NULL, msg, rc);
2895         }
2896
2897         /* completion will be signalled by an event */
2898         return 0;
2899 }
2900 EXPORT_SYMBOL(LNetGet);
2901
2902 /**
2903  * Calculate distance to node at \a dstnid.
2904  *
2905  * \param dstnid Target NID.
2906  * \param srcnidp If not NULL, NID of the local interface to reach \a dstnid
2907  * is saved here.
2908  * \param orderp If not NULL, order of the route to reach \a dstnid is saved
2909  * here.
2910  *
2911  * \retval 0 If \a dstnid belongs to a local interface, and reserved option
2912  * local_nid_dist_zero is set, which is the default.
2913  * \retval positives Distance to target NID, i.e. number of hops plus one.
2914  * \retval -EHOSTUNREACH If \a dstnid is not reachable.
2915  */
2916 int
2917 LNetDist(lnet_nid_t dstnid, lnet_nid_t *srcnidp, __u32 *orderp)
2918 {
2919         struct list_head        *e;
2920         struct lnet_ni          *ni = NULL;
2921         lnet_remotenet_t        *rnet;
2922         __u32                   dstnet = LNET_NIDNET(dstnid);
2923         int                     hops;
2924         int                     cpt;
2925         __u32                   order = 2;
2926         struct list_head        *rn_list;
2927
2928         /* if !local_nid_dist_zero, I don't return a distance of 0 ever
2929          * (when lustre sees a distance of 0, it substitutes 0@lo), so I
2930          * keep order 0 free for 0@lo and order 1 free for a local NID
2931          * match */
2932
2933         LASSERT(the_lnet.ln_refcount > 0);
2934
2935         cpt = lnet_net_lock_current();
2936
2937         while ((ni = lnet_get_next_ni_locked(NULL, ni))) {
2938                 if (ni->ni_nid == dstnid) {
2939                         if (srcnidp != NULL)
2940                                 *srcnidp = dstnid;
2941                         if (orderp != NULL) {
2942                                 if (LNET_NETTYP(LNET_NIDNET(dstnid)) == LOLND)
2943                                         *orderp = 0;
2944                                 else
2945                                         *orderp = 1;
2946                         }
2947                         lnet_net_unlock(cpt);
2948
2949                         return local_nid_dist_zero ? 0 : 1;
2950                 }
2951
2952                 if (LNET_NIDNET(ni->ni_nid) == dstnet) {
2953                         /* Check if ni was originally created in
2954                          * current net namespace.
2955                          * If not, assign order above 0xffff0000,
2956                          * to make this ni not a priority. */
2957                         if (!net_eq(ni->ni_net_ns, current->nsproxy->net_ns))
2958                                 order += 0xffff0000;
2959
2960                         if (srcnidp != NULL)
2961                                 *srcnidp = ni->ni_nid;
2962                         if (orderp != NULL)
2963                                 *orderp = order;
2964                         lnet_net_unlock(cpt);
2965                         return 1;
2966                 }
2967
2968                 order++;
2969         }
2970
2971         rn_list = lnet_net2rnethash(dstnet);
2972         list_for_each(e, rn_list) {
2973                 rnet = list_entry(e, lnet_remotenet_t, lrn_list);
2974
2975                 if (rnet->lrn_net == dstnet) {
2976                         lnet_route_t *route;
2977                         lnet_route_t *shortest = NULL;
2978                         __u32 shortest_hops = LNET_UNDEFINED_HOPS;
2979                         __u32 route_hops;
2980
2981                         LASSERT(!list_empty(&rnet->lrn_routes));
2982
2983                         list_for_each_entry(route, &rnet->lrn_routes,
2984                                             lr_list) {
2985                                 route_hops = route->lr_hops;
2986                                 if (route_hops == LNET_UNDEFINED_HOPS)
2987                                         route_hops = 1;
2988                                 if (shortest == NULL ||
2989                                     route_hops < shortest_hops) {
2990                                         shortest = route;
2991                                         shortest_hops = route_hops;
2992                                 }
2993                         }
2994
2995                         LASSERT(shortest != NULL);
2996                         hops = shortest_hops;
2997                         if (srcnidp != NULL) {
2998                                 ni = lnet_get_next_ni_locked(
2999                                         shortest->lr_gateway->lpni_net,
3000                                         NULL);
3001                                 *srcnidp = ni->ni_nid;
3002                         }
3003                         if (orderp != NULL)
3004                                 *orderp = order;
3005                         lnet_net_unlock(cpt);
3006                         return hops + 1;
3007                 }
3008                 order++;
3009         }
3010
3011         lnet_net_unlock(cpt);
3012         return -EHOSTUNREACH;
3013 }
3014 EXPORT_SYMBOL(LNetDist);