Whamcloud - gitweb
LU-1617 build: skip generated files in .gitignore
[fs/lustre-release.git] / lnet / klnds / mxlnd / mxlnd_cb.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (C) 2006 Myricom, Inc.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lnet/klnds/mxlnd/mxlnd.c
37  *
38  * Author: Eric Barton <eric@bartonsoftware.com>
39  * Author: Scott Atchley <atchley at myri.com>
40  */
41
42 #include "mxlnd.h"
43
44 mx_endpoint_addr_t MX_EPA_NULL; /* use to determine if an endpoint is NULL */
45
46 inline int
47 mxlnd_endpoint_addr_null(mx_endpoint_addr_t epa)
48 {
49         /* if memcmp() == 0, it is NULL */
50         return !(memcmp(&epa, &MX_EPA_NULL, sizeof(epa)));
51 }
52
53 char *
54 mxlnd_ctxstate_to_str(int mxc_state)
55 {
56         switch (mxc_state) {
57         case MXLND_CTX_INIT:
58                 return "MXLND_CTX_INIT";
59         case MXLND_CTX_IDLE:
60                 return "MXLND_CTX_IDLE";
61         case MXLND_CTX_PREP:
62                 return "MXLND_CTX_PREP";
63         case MXLND_CTX_PENDING:
64                 return "MXLND_CTX_PENDING";
65         case MXLND_CTX_COMPLETED:
66                 return "MXLND_CTX_COMPLETED";
67         case MXLND_CTX_CANCELED:
68                 return "MXLND_CTX_CANCELED";
69         default:
70                 return "*unknown*";
71         }
72 }
73
74 char *
75 mxlnd_connstatus_to_str(int mxk_status)
76 {
77         switch (mxk_status) {
78         case MXLND_CONN_READY:
79                 return "MXLND_CONN_READY";
80         case MXLND_CONN_INIT:
81                 return "MXLND_CONN_INIT";
82         case MXLND_CONN_WAIT:
83                 return "MXLND_CONN_WAIT";
84         case MXLND_CONN_DISCONNECT:
85                 return "MXLND_CONN_DISCONNECT";
86         case MXLND_CONN_FAIL:
87                 return "MXLND_CONN_FAIL";
88         default:
89                 return "unknown";
90         }
91 }
92
93 char *
94 mxlnd_msgtype_to_str(int type) {
95         switch (type) {
96         case MXLND_MSG_EAGER:
97                 return "MXLND_MSG_EAGER";
98         case MXLND_MSG_CONN_REQ:
99                 return "MXLND_MSG_CONN_REQ";
100         case MXLND_MSG_CONN_ACK:
101                 return "MXLND_MSG_CONN_ACK";
102         case MXLND_MSG_BYE:
103                 return "MXLND_MSG_BYE";
104         case MXLND_MSG_NOOP:
105                 return "MXLND_MSG_NOOP";
106         case MXLND_MSG_PUT_REQ:
107                 return "MXLND_MSG_PUT_REQ";
108         case MXLND_MSG_PUT_ACK:
109                 return "MXLND_MSG_PUT_ACK";
110         case MXLND_MSG_PUT_DATA:
111                 return "MXLND_MSG_PUT_DATA";
112         case MXLND_MSG_GET_REQ:
113                 return "MXLND_MSG_GET_REQ";
114         case MXLND_MSG_GET_DATA:
115                 return "MXLND_MSG_GET_DATA";
116         default:
117                 return "unknown";
118         }
119 }
120
121 char *
122 mxlnd_lnetmsg_to_str(int type)
123 {
124         switch (type) {
125         case LNET_MSG_ACK:
126                 return "LNET_MSG_ACK";
127         case LNET_MSG_PUT:
128                 return "LNET_MSG_PUT";
129         case LNET_MSG_GET:
130                 return "LNET_MSG_GET";
131         case LNET_MSG_REPLY:
132                 return "LNET_MSG_REPLY";
133         case LNET_MSG_HELLO:
134                 return "LNET_MSG_HELLO";
135         default:
136                 LBUG();
137                 return "*unknown*";
138         }
139 }
140
141 static inline u64
142 mxlnd_create_match(kmx_ctx_t *ctx, u8 error)
143 {
144         u64 type        = (u64) ctx->mxc_msg_type;
145         u64 err         = (u64) error;
146         u64 match       = 0ULL;
147
148         mxlnd_valid_msg_type(ctx->mxc_msg_type);
149         LASSERT(ctx->mxc_cookie >> MXLND_ERROR_OFFSET == 0);
150         match = (type << MXLND_MSG_OFFSET) | (err << MXLND_ERROR_OFFSET) | ctx->mxc_cookie;
151         return match;
152 }
153
154 static inline void
155 mxlnd_parse_match(u64 match, u8 *msg_type, u8 *error, u64 *cookie)
156 {
157         *msg_type = (u8) MXLND_MSG_TYPE(match);
158         *error    = (u8) MXLND_ERROR_VAL(match);
159         *cookie   = match & MXLND_MAX_COOKIE;
160         mxlnd_valid_msg_type(*msg_type);
161         return;
162 }
163
164 kmx_ctx_t *
165 mxlnd_get_idle_rx(kmx_conn_t *conn)
166 {
167         cfs_list_t              *rxs    = NULL;
168         kmx_ctx_t               *rx     = NULL;
169
170         LASSERT(conn != NULL);
171
172         rxs = &conn->mxk_rx_idle;
173
174         cfs_spin_lock(&conn->mxk_lock);
175
176         if (cfs_list_empty (rxs)) {
177                 cfs_spin_unlock(&conn->mxk_lock);
178                 return NULL;
179         }
180
181         rx = cfs_list_entry (rxs->next, kmx_ctx_t, mxc_list);
182         cfs_list_del_init(&rx->mxc_list);
183         cfs_spin_unlock(&conn->mxk_lock);
184
185 #if MXLND_DEBUG
186         if (rx->mxc_get != rx->mxc_put) {
187                 CNETERR("*** RX get (%llu) != put (%llu) ***\n", rx->mxc_get, rx->mxc_put);
188                 CNETERR("*** incarnation= %lld ***\n", rx->mxc_incarnation);
189                 CNETERR("*** deadline= %ld ***\n", rx->mxc_deadline);
190                 CNETERR("*** state= %s ***\n", mxlnd_ctxstate_to_str(rx->mxc_state));
191                 CNETERR("*** listed?= %d ***\n", !cfs_list_empty(&rx->mxc_list));
192                 CNETERR("*** nid= 0x%llx ***\n", rx->mxc_nid);
193                 CNETERR("*** peer= 0x%p ***\n", rx->mxc_peer);
194                 CNETERR("*** msg_type= %s ***\n", mxlnd_msgtype_to_str(rx->mxc_msg_type));
195                 CNETERR("*** cookie= 0x%llx ***\n", rx->mxc_cookie);
196                 CNETERR("*** nob= %d ***\n", rx->mxc_nob);
197         }
198 #endif
199         LASSERT (rx->mxc_get == rx->mxc_put);
200
201         rx->mxc_get++;
202
203         LASSERT (rx->mxc_state == MXLND_CTX_IDLE);
204         rx->mxc_state = MXLND_CTX_PREP;
205         rx->mxc_deadline = jiffies + MXLND_COMM_TIMEOUT;
206
207         return rx;
208 }
209
210 int
211 mxlnd_put_idle_rx(kmx_ctx_t *rx)
212 {
213         kmx_conn_t              *conn   = rx->mxc_conn;
214         cfs_list_t              *rxs    = &conn->mxk_rx_idle;
215
216         LASSERT(rx->mxc_type == MXLND_REQ_RX);
217
218         mxlnd_ctx_init(rx);
219
220         rx->mxc_put++;
221         LASSERT(rx->mxc_get == rx->mxc_put);
222
223         cfs_spin_lock(&conn->mxk_lock);
224         cfs_list_add(&rx->mxc_list, rxs);
225         cfs_spin_unlock(&conn->mxk_lock);
226         return 0;
227 }
228
229 kmx_ctx_t *
230 mxlnd_get_idle_tx(void)
231 {
232         cfs_list_t              *tmp    = &kmxlnd_data.kmx_tx_idle;
233         kmx_ctx_t               *tx     = NULL;
234
235         cfs_spin_lock(&kmxlnd_data.kmx_tx_idle_lock);
236
237         if (cfs_list_empty (&kmxlnd_data.kmx_tx_idle)) {
238                 CNETERR("%d txs in use\n", kmxlnd_data.kmx_tx_used);
239                 cfs_spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);
240                 return NULL;
241         }
242
243         tmp = &kmxlnd_data.kmx_tx_idle;
244         tx = cfs_list_entry (tmp->next, kmx_ctx_t, mxc_list);
245         cfs_list_del_init(&tx->mxc_list);
246
247         /* Allocate a new completion cookie.  It might not be needed,
248          * but we've got a lock right now and we're unlikely to
249          * wrap... */
250         tx->mxc_cookie = kmxlnd_data.kmx_tx_next_cookie++;
251         if (kmxlnd_data.kmx_tx_next_cookie > MXLND_MAX_COOKIE) {
252                 kmxlnd_data.kmx_tx_next_cookie = 1;
253         }
254         kmxlnd_data.kmx_tx_used++;
255         cfs_spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);
256
257         LASSERT (tx->mxc_get == tx->mxc_put);
258
259         tx->mxc_get++;
260
261         LASSERT (tx->mxc_state == MXLND_CTX_IDLE);
262         LASSERT (tx->mxc_lntmsg[0] == NULL);
263         LASSERT (tx->mxc_lntmsg[1] == NULL);
264
265         tx->mxc_state = MXLND_CTX_PREP;
266         tx->mxc_deadline = jiffies + MXLND_COMM_TIMEOUT;
267
268         return tx;
269 }
270
271 void
272 mxlnd_conn_disconnect(kmx_conn_t *conn, int mx_dis, int send_bye);
273
274 int
275 mxlnd_put_idle_tx(kmx_ctx_t *tx)
276 {
277         int             result  = 0;
278         lnet_msg_t      *lntmsg[2];
279
280         LASSERT(tx->mxc_type == MXLND_REQ_TX);
281
282         if (tx->mxc_status.code != MX_STATUS_SUCCESS || tx->mxc_errno != 0) {
283                 kmx_conn_t      *conn   = tx->mxc_conn;
284
285                 result = -EIO;
286                 if (tx->mxc_errno != 0) result = tx->mxc_errno;
287                 /* FIXME should we set mx_dis? */
288                 mxlnd_conn_disconnect(conn, 0, 1);
289         }
290
291         lntmsg[0] = tx->mxc_lntmsg[0];
292         lntmsg[1] = tx->mxc_lntmsg[1];
293
294         mxlnd_ctx_init(tx);
295
296         tx->mxc_put++;
297         LASSERT(tx->mxc_get == tx->mxc_put);
298
299         cfs_spin_lock(&kmxlnd_data.kmx_tx_idle_lock);
300         cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_tx_idle);
301         kmxlnd_data.kmx_tx_used--;
302         cfs_spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);
303
304         if (lntmsg[0] != NULL) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[0], result);
305         if (lntmsg[1] != NULL) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[1], result);
306         return 0;
307 }
308
309
310 void
311 mxlnd_connparams_free(kmx_connparams_t *cp)
312 {
313         LASSERT(cfs_list_empty(&cp->mxr_list));
314         MXLND_FREE(cp, sizeof(*cp));
315         return;
316 }
317
318 int
319 mxlnd_connparams_alloc(kmx_connparams_t **cp, void *context,
320                             mx_endpoint_addr_t epa, u64 match, u32 length,
321                             kmx_conn_t *conn, kmx_peer_t *peer, void *data)
322 {
323         kmx_connparams_t *c = NULL;
324
325         MXLND_ALLOC(c, sizeof(*c));
326         if (!c) return -ENOMEM;
327
328         CFS_INIT_LIST_HEAD(&c->mxr_list);
329         c->mxr_context = context;
330         c->mxr_epa = epa;
331         c->mxr_match = match;
332         c->mxr_nob = length;
333         c->mxr_conn = conn;
334         c->mxr_peer = peer;
335         c->mxr_msg = *((kmx_msg_t *) data);
336
337         *cp = c;
338         return 0;
339 }
340
341 static inline void
342 mxlnd_set_conn_status(kmx_conn_t *conn, int status)
343 {
344         conn->mxk_status = status;
345         cfs_mb();
346 }
347
348 /**
349  * mxlnd_conn_free_locked - free the conn
350  * @conn - a kmx_conn pointer
351  *
352  * The calling function should remove the conn from the conns list first
353  * then destroy it. Caller should have write-locked kmx_global_lock.
354  */
355 void
356 mxlnd_conn_free_locked(kmx_conn_t *conn)
357 {
358         int             valid   = !mxlnd_endpoint_addr_null(conn->mxk_epa);
359         kmx_peer_t      *peer   = conn->mxk_peer;
360
361         CDEBUG(D_NET, "freeing conn 0x%p *****\n", conn);
362         LASSERT (cfs_list_empty (&conn->mxk_tx_credit_queue) &&
363                  cfs_list_empty (&conn->mxk_tx_free_queue) &&
364                  cfs_list_empty (&conn->mxk_pending));
365         if (!cfs_list_empty(&conn->mxk_list)) {
366                 cfs_list_del_init(&conn->mxk_list);
367                 if (peer->mxp_conn == conn) {
368                         peer->mxp_conn = NULL;
369                         if (valid) {
370                                 kmx_conn_t      *temp   = NULL;
371
372                                 mx_get_endpoint_addr_context(conn->mxk_epa,
373                                                              (void **) &temp);
374                                 if (conn == temp) {
375                                         mx_set_endpoint_addr_context(conn->mxk_epa,
376                                                                      (void *) NULL);
377                                 }
378                         }
379                         /* unlink from global list and drop its ref */
380                         cfs_list_del_init(&peer->mxp_list);
381                         mxlnd_peer_decref(peer);
382                 }
383         }
384         mxlnd_peer_decref(peer); /* drop conn's ref to peer */
385         if (conn->mxk_rx_pages) {
386                 LASSERT (conn->mxk_rxs != NULL);
387                 mxlnd_free_pages(conn->mxk_rx_pages);
388         }
389         if (conn->mxk_rxs) {
390                 int             i       = 0;
391                 kmx_ctx_t       *rx     = NULL;
392
393                 for (i = 0; i < MXLND_RX_MSGS(); i++) {
394                         rx = &conn->mxk_rxs[i];
395                         if (rx->mxc_seg_list != NULL) {
396                                 LASSERT(rx->mxc_nseg > 0);
397                                 MXLND_FREE(rx->mxc_seg_list,
398                                            rx->mxc_nseg *
399                                            sizeof(*rx->mxc_seg_list));
400                         }
401                 }
402                 MXLND_FREE(conn->mxk_rxs, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));
403         }
404
405         MXLND_FREE(conn, sizeof (*conn));
406         return;
407 }
408
409
410 int
411 mxlnd_conn_cancel_pending_rxs(kmx_conn_t *conn)
412 {
413         int             found   = 0;
414         int             count   = 0;
415         kmx_ctx_t       *ctx    = NULL;
416         kmx_ctx_t       *next   = NULL;
417         mx_return_t     mxret   = MX_SUCCESS;
418         u32             result  = 0;
419
420         do {
421                 found = 0;
422                 cfs_spin_lock(&conn->mxk_lock);
423                 cfs_list_for_each_entry_safe(ctx, next, &conn->mxk_pending,
424                                              mxc_list) {
425                         cfs_list_del_init(&ctx->mxc_list);
426                         if (ctx->mxc_type == MXLND_REQ_RX) {
427                                 found = 1;
428                                 mxret = mx_cancel(kmxlnd_data.kmx_endpt,
429                                                   &ctx->mxc_mxreq,
430                                                   &result);
431                                 if (mxret != MX_SUCCESS) {
432                                         CNETERR("mx_cancel() returned %s (%d)\n", mx_strerror(mxret), mxret);
433                                 }
434                                 if (result == 1) {
435                                         ctx->mxc_errno = -ECONNABORTED;
436                                         ctx->mxc_state = MXLND_CTX_CANCELED;
437                                         cfs_spin_unlock(&conn->mxk_lock);
438                                         cfs_spin_lock(&kmxlnd_data.kmx_conn_lock);
439                                         /* we may be holding the global lock,
440                                          * move to orphan list so that it can free it */
441                                         cfs_list_add_tail(&ctx->mxc_list,
442                                                           &kmxlnd_data.kmx_orphan_msgs);
443                                         count++;
444                                         cfs_spin_unlock(&kmxlnd_data.kmx_conn_lock);
445                                         cfs_spin_lock(&conn->mxk_lock);
446                                 }
447                                 break;
448                         }
449                 }
450                 cfs_spin_unlock(&conn->mxk_lock);
451         }
452         while (found);
453
454         return count;
455 }
456
457 int
458 mxlnd_cancel_queued_txs(kmx_conn_t *conn)
459 {
460         int                     count   = 0;
461         cfs_list_t             *tmp    = NULL;
462
463         cfs_spin_lock(&conn->mxk_lock);
464         while (!cfs_list_empty(&conn->mxk_tx_free_queue) ||
465                !cfs_list_empty(&conn->mxk_tx_credit_queue)) {
466
467                 kmx_ctx_t       *tx     = NULL;
468
469                 if (!cfs_list_empty(&conn->mxk_tx_free_queue)) {
470                         tmp = &conn->mxk_tx_free_queue;
471                 } else {
472                         tmp = &conn->mxk_tx_credit_queue;
473                 }
474
475                 tx = cfs_list_entry(tmp->next, kmx_ctx_t, mxc_list);
476                 cfs_list_del_init(&tx->mxc_list);
477                 cfs_spin_unlock(&conn->mxk_lock);
478                 tx->mxc_errno = -ECONNABORTED;
479                 tx->mxc_state = MXLND_CTX_CANCELED;
480                 /* move to orphan list and then abort */
481                 cfs_spin_lock(&kmxlnd_data.kmx_conn_lock);
482                 cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_orphan_msgs);
483                 cfs_spin_unlock(&kmxlnd_data.kmx_conn_lock);
484                 count++;
485                 cfs_spin_lock(&conn->mxk_lock);
486         }
487         cfs_spin_unlock(&conn->mxk_lock);
488
489         return count;
490 }
491
492 void
493 mxlnd_send_message(mx_endpoint_addr_t epa, u8 msg_type, int error, u64 cookie)
494 {
495         u64 match = (((u64) msg_type) << MXLND_MSG_OFFSET) |
496                     (((u64) error) << MXLND_ERROR_OFFSET) | cookie;
497
498         mx_kisend(kmxlnd_data.kmx_endpt, NULL, 0, MX_PIN_PHYSICAL,
499                   epa, match, NULL, NULL);
500         return;
501 }
502
503 /**
504  * mxlnd_conn_disconnect - shutdown a connection
505  * @conn - a kmx_conn pointer
506  * @mx_dis - call mx_disconnect()
507  * @send_bye - send peer a BYE msg
508  *
509  * This function sets the status to DISCONNECT, completes queued
510  * txs with failure, calls mx_disconnect, which will complete
511  * pending txs and matched rxs with failure.
512  */
513 void
514 mxlnd_conn_disconnect(kmx_conn_t *conn, int mx_dis, int send_bye)
515 {
516         mx_endpoint_addr_t      epa     = conn->mxk_epa;
517         int                     valid   = !mxlnd_endpoint_addr_null(epa);
518         int                     count   = 0;
519
520         cfs_spin_lock(&conn->mxk_lock);
521         if (conn->mxk_status == MXLND_CONN_DISCONNECT) {
522                 cfs_spin_unlock(&conn->mxk_lock);
523                 return;
524         }
525         mxlnd_set_conn_status(conn, MXLND_CONN_DISCONNECT);
526         conn->mxk_timeout = 0;
527         cfs_spin_unlock(&conn->mxk_lock);
528
529         count = mxlnd_cancel_queued_txs(conn);
530         count += mxlnd_conn_cancel_pending_rxs(conn);
531
532         if (count)
533                 cfs_up(&kmxlnd_data.kmx_conn_sem); /* let connd call kmxlnd_abort_msgs() */
534
535         if (send_bye && valid &&
536             conn->mxk_peer->mxp_nid != kmxlnd_data.kmx_ni->ni_nid) {
537                 /* send a BYE to the peer */
538                 CDEBUG(D_NET, "%s: sending a BYE msg to %s\n", __func__,
539                                 libcfs_nid2str(conn->mxk_peer->mxp_nid));
540                 mxlnd_send_message(epa, MXLND_MSG_BYE, 0, 0);
541                 /* wait to allow the peer to ack our message */
542                 mxlnd_sleep(msecs_to_jiffies(20));
543         }
544
545         if (cfs_atomic_read(&kmxlnd_data.kmx_shutdown) != 1) {
546                 unsigned long   last_msg        = 0;
547
548                 /* notify LNET that we are giving up on this peer */
549                 if (cfs_time_after(conn->mxk_last_rx, conn->mxk_last_tx))
550                         last_msg = conn->mxk_last_rx;
551                 else
552                         last_msg = conn->mxk_last_tx;
553
554                 lnet_notify(kmxlnd_data.kmx_ni, conn->mxk_peer->mxp_nid, 0, last_msg);
555
556                 if (mx_dis && valid &&
557                     (memcmp(&epa, &kmxlnd_data.kmx_epa, sizeof(epa) != 0)))
558                         mx_disconnect(kmxlnd_data.kmx_endpt, epa);
559         }
560         mxlnd_conn_decref(conn); /* drop the owning peer's reference */
561
562         return;
563 }
564
565 /**
566  * mxlnd_conn_alloc - allocate and initialize a new conn struct
567  * @connp - address of a kmx_conn pointer
568  * @peer - owning kmx_peer
569  *
570  * Returns 0 on success and -ENOMEM on failure
571  */
572 int
573 mxlnd_conn_alloc_locked(kmx_conn_t **connp, kmx_peer_t *peer)
574 {
575         int             i       = 0;
576         int             ret     = 0;
577         int             ipage   = 0;
578         int             offset  = 0;
579         void           *addr    = NULL;
580         kmx_conn_t     *conn    = NULL;
581         kmx_pages_t    *pages   = NULL;
582         struct page    *page    = NULL;
583         kmx_ctx_t      *rx      = NULL;
584
585         LASSERT(peer != NULL);
586
587         MXLND_ALLOC(conn, sizeof (*conn));
588         if (conn == NULL) {
589                 CNETERR("Cannot allocate conn\n");
590                 return -ENOMEM;
591         }
592         CDEBUG(D_NET, "allocated conn 0x%p for peer 0x%p\n", conn, peer);
593
594         memset(conn, 0, sizeof(*conn));
595
596         ret = mxlnd_alloc_pages(&pages, MXLND_RX_MSG_PAGES());
597         if (ret != 0) {
598                 CERROR("Can't allocate rx pages\n");
599                 MXLND_FREE(conn, sizeof(*conn));
600                 return -ENOMEM;
601         }
602         conn->mxk_rx_pages = pages;
603
604         MXLND_ALLOC(conn->mxk_rxs, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));
605         if (conn->mxk_rxs == NULL) {
606                 CERROR("Can't allocate %d rx descriptors\n", MXLND_RX_MSGS());
607                 mxlnd_free_pages(pages);
608                 MXLND_FREE(conn, sizeof(*conn));
609                 return -ENOMEM;
610         }
611
612         memset(conn->mxk_rxs, 0, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));
613
614         conn->mxk_peer = peer;
615         CFS_INIT_LIST_HEAD(&conn->mxk_list);
616         CFS_INIT_LIST_HEAD(&conn->mxk_zombie);
617         cfs_atomic_set(&conn->mxk_refcount, 2); /* ref for owning peer
618                                                    and one for the caller */
619         if (peer->mxp_nid == kmxlnd_data.kmx_ni->ni_nid) {
620                 u64     nic_id  = 0ULL;
621                 u32     ep_id   = 0;
622
623                 /* this is localhost, set the epa and status as up */
624                 mxlnd_set_conn_status(conn, MXLND_CONN_READY);
625                 conn->mxk_epa = kmxlnd_data.kmx_epa;
626                 mx_set_endpoint_addr_context(conn->mxk_epa, (void *) conn);
627                 peer->mxp_reconnect_time = 0;
628                 mx_decompose_endpoint_addr(kmxlnd_data.kmx_epa, &nic_id, &ep_id);
629                 peer->mxp_nic_id = nic_id;
630                 peer->mxp_ep_id = ep_id;
631                 conn->mxk_incarnation = kmxlnd_data.kmx_incarnation;
632                 conn->mxk_timeout = 0;
633         } else {
634                 /* conn->mxk_incarnation = 0 - will be set by peer */
635                 /* conn->mxk_sid = 0 - will be set by peer */
636                 mxlnd_set_conn_status(conn, MXLND_CONN_INIT);
637                 /* mxk_epa - to be set after mx_iconnect() */
638         }
639         cfs_spin_lock_init(&conn->mxk_lock);
640         /* conn->mxk_timeout = 0 */
641         /* conn->mxk_last_tx = 0 */
642         /* conn->mxk_last_rx = 0 */
643         CFS_INIT_LIST_HEAD(&conn->mxk_rx_idle);
644
645         conn->mxk_credits = *kmxlnd_tunables.kmx_peercredits;
646         /* mxk_outstanding = 0 */
647
648         CFS_INIT_LIST_HEAD(&conn->mxk_tx_credit_queue);
649         CFS_INIT_LIST_HEAD(&conn->mxk_tx_free_queue);
650         /* conn->mxk_ntx_msgs = 0 */
651         /* conn->mxk_ntx_data = 0 */
652         /* conn->mxk_ntx_posted = 0 */
653         /* conn->mxk_data_posted = 0 */
654         CFS_INIT_LIST_HEAD(&conn->mxk_pending);
655
656         for (i = 0; i < MXLND_RX_MSGS(); i++) {
657
658                 rx = &conn->mxk_rxs[i];
659                 rx->mxc_type = MXLND_REQ_RX;
660                 CFS_INIT_LIST_HEAD(&rx->mxc_list);
661
662                 /* map mxc_msg to page */
663                 page = pages->mxg_pages[ipage];
664                 addr = page_address(page);
665                 LASSERT(addr != NULL);
666                 rx->mxc_msg = (kmx_msg_t *)(addr + offset);
667                 rx->mxc_seg.segment_ptr = MX_PA_TO_U64(virt_to_phys(rx->mxc_msg));
668
669                 rx->mxc_conn = conn;
670                 rx->mxc_peer = peer;
671                 rx->mxc_nid = peer->mxp_nid;
672
673                 mxlnd_ctx_init(rx);
674
675                 offset += MXLND_MSG_SIZE;
676                 LASSERT (offset <= PAGE_SIZE);
677
678                 if (offset == PAGE_SIZE) {
679                         offset = 0;
680                         ipage++;
681                         LASSERT (ipage <= MXLND_TX_MSG_PAGES());
682                 }
683
684                 cfs_list_add_tail(&rx->mxc_list, &conn->mxk_rx_idle);
685         }
686
687         *connp = conn;
688
689         mxlnd_peer_addref(peer);        /* add a ref for this conn */
690
691         /* add to front of peer's conns list */
692         cfs_list_add(&conn->mxk_list, &peer->mxp_conns);
693         peer->mxp_conn = conn;
694         return 0;
695 }
696
697 int
698 mxlnd_conn_alloc(kmx_conn_t **connp, kmx_peer_t *peer)
699 {
700         int             ret     = 0;
701         cfs_rwlock_t   *g_lock  = &kmxlnd_data.kmx_global_lock;
702
703         cfs_write_lock(g_lock);
704         ret = mxlnd_conn_alloc_locked(connp, peer);
705         cfs_write_unlock(g_lock);
706         return ret;
707 }
708
709 int
710 mxlnd_q_pending_ctx(kmx_ctx_t *ctx)
711 {
712         int             ret     = 0;
713         kmx_conn_t      *conn   = ctx->mxc_conn;
714
715         ctx->mxc_state = MXLND_CTX_PENDING;
716         if (conn != NULL) {
717                 cfs_spin_lock(&conn->mxk_lock);
718                 if (conn->mxk_status >= MXLND_CONN_INIT) {
719                         cfs_list_add_tail(&ctx->mxc_list, &conn->mxk_pending);
720                         if (conn->mxk_timeout == 0 || ctx->mxc_deadline < conn->mxk_timeout) {
721                                 conn->mxk_timeout = ctx->mxc_deadline;
722                         }
723                 } else {
724                         ctx->mxc_state = MXLND_CTX_COMPLETED;
725                         ret = -1;
726                 }
727                 cfs_spin_unlock(&conn->mxk_lock);
728         }
729         return ret;
730 }
731
732 int
733 mxlnd_deq_pending_ctx(kmx_ctx_t *ctx)
734 {
735         LASSERT(ctx->mxc_state == MXLND_CTX_PENDING ||
736                 ctx->mxc_state == MXLND_CTX_COMPLETED);
737         if (ctx->mxc_state != MXLND_CTX_PENDING &&
738             ctx->mxc_state != MXLND_CTX_COMPLETED) {
739                 CNETERR("deq ctx->mxc_state = %s\n",
740                         mxlnd_ctxstate_to_str(ctx->mxc_state));
741         }
742         ctx->mxc_state = MXLND_CTX_COMPLETED;
743         if (!cfs_list_empty(&ctx->mxc_list)) {
744                 kmx_conn_t      *conn = ctx->mxc_conn;
745                 kmx_ctx_t       *next = NULL;
746
747                 LASSERT(conn != NULL);
748                 cfs_spin_lock(&conn->mxk_lock);
749                 cfs_list_del_init(&ctx->mxc_list);
750                 conn->mxk_timeout = 0;
751                 if (!cfs_list_empty(&conn->mxk_pending)) {
752                         next = cfs_list_entry(conn->mxk_pending.next,
753                                               kmx_ctx_t, mxc_list);
754                         conn->mxk_timeout = next->mxc_deadline;
755                 }
756                 cfs_spin_unlock(&conn->mxk_lock);
757         }
758         return 0;
759 }
760
761 /**
762  * mxlnd_peer_free - free the peer
763  * @peer - a kmx_peer pointer
764  *
765  * The calling function should decrement the rxs, drain the tx queues and
766  * remove the peer from the peers list first then destroy it.
767  */
768 void
769 mxlnd_peer_free(kmx_peer_t *peer)
770 {
771         CDEBUG(D_NET, "freeing peer 0x%p %s\n", peer, libcfs_nid2str(peer->mxp_nid));
772
773         LASSERT (cfs_atomic_read(&peer->mxp_refcount) == 0);
774
775         if (!cfs_list_empty(&peer->mxp_list)) {
776                 /* assume we are locked */
777                 cfs_list_del_init(&peer->mxp_list);
778         }
779
780         MXLND_FREE(peer, sizeof (*peer));
781         cfs_atomic_dec(&kmxlnd_data.kmx_npeers);
782         return;
783 }
784
785 static int
786 mxlnd_lookup_mac(u32 ip, u64 *tmp_id)
787 {
788         int                     ret     = -EHOSTUNREACH;
789         unsigned char           *haddr  = NULL;
790         struct net_device       *dev    = NULL;
791         struct neighbour        *n      = NULL;
792         __be32                  dst_ip  = htonl(ip);
793
794         dev = dev_get_by_name(*kmxlnd_tunables.kmx_default_ipif);
795         if (dev == NULL)
796                 return -ENODEV;
797
798         haddr = (unsigned char *) tmp_id + 2; /* MAC is only 6 bytes */
799
800         n = neigh_lookup(&arp_tbl, &dst_ip, dev);
801         if (n) {
802                 n->used = jiffies;
803                 if (n->nud_state & NUD_VALID) {
804                         memcpy(haddr, n->ha, dev->addr_len);
805                         neigh_release(n);
806                         ret = 0;
807                 }
808         }
809
810         dev_put(dev);
811
812         return ret;
813 }
814
815
816 /* We only want the MAC address of the peer's Myricom NIC. We
817  * require that each node has the IPoMX interface (myriN) up.
818  * We will not pass any traffic over IPoMX, but it allows us
819  * to get the MAC address. */
820 static int
821 mxlnd_ip2nic_id(u32 ip, u64 *nic_id, int tries)
822 {
823         int                     ret     = 0;
824         int                     try     = 1;
825         int                     fatal   = 0;
826         u64                     tmp_id  = 0ULL;
827         cfs_socket_t            *sock   = NULL;
828
829         do {
830                 CDEBUG(D_NET, "try %d of %d tries\n", try, tries);
831                 ret = mxlnd_lookup_mac(ip, &tmp_id);
832                 if (ret == 0) {
833                         break;
834                 } else {
835                         /* not found, try to connect (force an arp) */
836                         ret = libcfs_sock_connect(&sock, &fatal, 0, 0, ip, 987);
837                         if (ret == -ECONNREFUSED) {
838                                 /* peer is there, get the MAC address */
839                                 mxlnd_lookup_mac(ip, &tmp_id);
840                                 if (tmp_id != 0ULL)
841                                         ret = 0;
842                                 break;
843                         } else if (ret == -EHOSTUNREACH && try < tries) {
844                                 /* add a little backoff */
845                                 CDEBUG(D_NET, "sleeping for %d jiffies\n",
846                                        CFS_HZ/4);
847                                 mxlnd_sleep(CFS_HZ/4);
848                         }
849                 }
850         } while (try++ < tries);
851         CDEBUG(D_NET, "done trying. ret = %d\n", ret);
852
853         if (tmp_id == 0ULL)
854                 ret = -EHOSTUNREACH;
855 #ifdef __LITTLE_ENDIAN
856         *nic_id = ___arch__swab64(tmp_id);
857 #else
858         *nic_id = tmp_id;
859 #endif
860         return ret;
861 }
862
863 /**
864  * mxlnd_peer_alloc - allocate and initialize a new peer struct
865  * @peerp - address of a kmx_peer pointer
866  * @nid - LNET node id
867  *
868  * Returns 0 on success and -ENOMEM on failure
869  */
870 int
871 mxlnd_peer_alloc(kmx_peer_t **peerp, lnet_nid_t nid, u32 board, u32 ep_id, u64 nic_id)
872 {
873         int             ret     = 0;
874         u32             ip      = LNET_NIDADDR(nid);
875         kmx_peer_t     *peer    = NULL;
876
877         LASSERT (nid != LNET_NID_ANY && nid != 0LL);
878
879         MXLND_ALLOC(peer, sizeof (*peer));
880         if (peer == NULL) {
881                 CNETERR("Cannot allocate peer for NID 0x%llx\n",
882                        nid);
883                 return -ENOMEM;
884         }
885         CDEBUG(D_NET, "allocated peer 0x%p for NID 0x%llx\n", peer, nid);
886
887         memset(peer, 0, sizeof(*peer));
888
889         CFS_INIT_LIST_HEAD(&peer->mxp_list);
890         peer->mxp_nid = nid;
891         /* peer->mxp_ni unused - may be used for multi-rail */
892         cfs_atomic_set(&peer->mxp_refcount, 1);     /* ref for kmx_peers list */
893
894         peer->mxp_board = board;
895         peer->mxp_ep_id = ep_id;
896         peer->mxp_nic_id = nic_id;
897
898         CFS_INIT_LIST_HEAD(&peer->mxp_conns);
899         ret = mxlnd_conn_alloc(&peer->mxp_conn, peer); /* adds 2nd conn ref here... */
900         if (ret != 0) {
901                 mxlnd_peer_decref(peer);
902                 return ret;
903         }
904         CFS_INIT_LIST_HEAD(&peer->mxp_tx_queue);
905
906         if (peer->mxp_nic_id != 0ULL)
907                 nic_id = peer->mxp_nic_id;
908
909         if (nic_id == 0ULL) {
910                 ret = mxlnd_ip2nic_id(ip, &nic_id, 1);
911                 if (ret == 0) {
912                         peer->mxp_nic_id = nic_id;
913                         mx_nic_id_to_board_number(nic_id, &peer->mxp_board);
914                 }
915         }
916
917         peer->mxp_nic_id = nic_id; /* may be 0ULL if ip2nic_id() failed */
918
919         /* peer->mxp_reconnect_time = 0 */
920         /* peer->mxp_incompatible = 0 */
921
922         *peerp = peer;
923         return 0;
924 }
925
926 static inline kmx_peer_t *
927 mxlnd_find_peer_by_nid_locked(lnet_nid_t nid)
928 {
929         int             found   = 0;
930         int             hash    = 0;
931         kmx_peer_t      *peer   = NULL;
932
933         hash = mxlnd_nid_to_hash(nid);
934
935         cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[hash], mxp_list) {
936                 if (peer->mxp_nid == nid) {
937                         found = 1;
938                         mxlnd_peer_addref(peer);
939                         break;
940                 }
941         }
942         return (found ? peer : NULL);
943 }
944
945 static kmx_peer_t *
946 mxlnd_find_peer_by_nid(lnet_nid_t nid, int create)
947 {
948         int             ret     = 0;
949         int             hash    = 0;
950         kmx_peer_t      *peer   = NULL;
951         kmx_peer_t      *old    = NULL;
952         cfs_rwlock_t    *g_lock = &kmxlnd_data.kmx_global_lock;
953
954         cfs_read_lock(g_lock);
955         peer = mxlnd_find_peer_by_nid_locked(nid); /* adds peer ref */
956
957         if ((peer && peer->mxp_conn) || /* found peer with conn or */
958             (!peer && !create)) {       /* did not find peer and do not create one */
959                 cfs_read_unlock(g_lock);
960                 return peer;
961         }
962
963         cfs_read_unlock(g_lock);
964
965         /* if peer but _not_ conn */
966         if (peer && !peer->mxp_conn) {
967                 if (create) {
968                         cfs_write_lock(g_lock);
969                         if (!peer->mxp_conn) { /* check again */
970                                 /* create the conn */
971                                 ret = mxlnd_conn_alloc_locked(&peer->mxp_conn, peer);
972                                 if (ret != 0) {
973                                         /* we tried, return the peer only.
974                                          * the caller needs to see if the conn exists */
975                                         CNETERR("%s: %s could not alloc conn\n",
976                                         __func__, libcfs_nid2str(peer->mxp_nid));
977                                 } else {
978                                         /* drop extra conn ref */
979                                         mxlnd_conn_decref(peer->mxp_conn);
980                                 }
981                         }
982                         cfs_write_unlock(g_lock);
983                 }
984                 return peer;
985         }
986
987         /* peer not found and we need to create one */
988         hash = mxlnd_nid_to_hash(nid);
989
990         /* create peer (and conn) */
991         /* adds conn ref for peer and one for this function */
992         ret = mxlnd_peer_alloc(&peer, nid, *kmxlnd_tunables.kmx_board,
993                                *kmxlnd_tunables.kmx_ep_id, 0ULL);
994         if (ret != 0) /* no memory, peer is NULL */
995                 return NULL;
996
997         cfs_write_lock(g_lock);
998
999         /* look again */
1000         old = mxlnd_find_peer_by_nid_locked(nid);
1001         if (old) {
1002                 /* someone already created one */
1003                 mxlnd_conn_decref(peer->mxp_conn); /* drop ref taken above.. */
1004                 mxlnd_conn_decref(peer->mxp_conn); /* drop peer's ref */
1005                 mxlnd_peer_decref(peer);
1006                 peer = old;
1007         } else {
1008                 /* no other peer, use this one */
1009                 cfs_list_add_tail(&peer->mxp_list,
1010                                   &kmxlnd_data.kmx_peers[hash]);
1011                 cfs_atomic_inc(&kmxlnd_data.kmx_npeers);
1012                 mxlnd_peer_addref(peer);
1013                 mxlnd_conn_decref(peer->mxp_conn); /* drop ref from peer_alloc */
1014         }
1015
1016         cfs_write_unlock(g_lock);
1017
1018         return peer;
1019 }
1020
1021 static inline int
1022 mxlnd_tx_requires_credit(kmx_ctx_t *tx)
1023 {
1024         return (tx->mxc_msg_type == MXLND_MSG_EAGER ||
1025                 tx->mxc_msg_type == MXLND_MSG_GET_REQ ||
1026                 tx->mxc_msg_type == MXLND_MSG_PUT_REQ ||
1027                 tx->mxc_msg_type == MXLND_MSG_NOOP);
1028 }
1029
1030 /**
1031  * mxlnd_init_msg - set type and number of bytes
1032  * @msg - msg pointer
1033  * @type - of message
1034  * @body_nob - bytes in msg body
1035  */
1036 static inline void
1037 mxlnd_init_msg(kmx_msg_t *msg, u8 type, int body_nob)
1038 {
1039         msg->mxm_type = type;
1040         msg->mxm_nob  = offsetof(kmx_msg_t, mxm_u) + body_nob;
1041 }
1042
1043 static inline void
1044 mxlnd_init_tx_msg (kmx_ctx_t *tx, u8 type, int body_nob, lnet_nid_t nid)
1045 {
1046         int             nob     = offsetof (kmx_msg_t, mxm_u) + body_nob;
1047         kmx_msg_t       *msg    = NULL;
1048
1049         LASSERT (tx != NULL);
1050         LASSERT (nob <= MXLND_MSG_SIZE);
1051
1052         tx->mxc_nid = nid;
1053         /* tx->mxc_peer should have already been set if we know it */
1054         tx->mxc_msg_type = type;
1055         tx->mxc_nseg = 1;
1056         /* tx->mxc_seg.segment_ptr is already pointing to mxc_page */
1057         tx->mxc_seg.segment_length = nob;
1058         tx->mxc_pin_type = MX_PIN_PHYSICAL;
1059
1060         msg = tx->mxc_msg;
1061         msg->mxm_type = type;
1062         msg->mxm_nob  = nob;
1063
1064         return;
1065 }
1066
1067 static inline __u32
1068 mxlnd_cksum (void *ptr, int nob)
1069 {
1070         char  *c  = ptr;
1071         __u32  sum = 0;
1072
1073         while (nob-- > 0)
1074                 sum = ((sum << 1) | (sum >> 31)) + *c++;
1075
1076         /* ensure I don't return 0 (== no checksum) */
1077         return (sum == 0) ? 1 : sum;
1078 }
1079
1080 /**
1081  * mxlnd_pack_msg_locked - complete msg info
1082  * @tx - msg to send
1083  */
1084 static inline void
1085 mxlnd_pack_msg_locked(kmx_ctx_t *tx)
1086 {
1087         kmx_msg_t       *msg    = tx->mxc_msg;
1088
1089         /* type and nob should already be set in init_msg() */
1090         msg->mxm_magic    = MXLND_MSG_MAGIC;
1091         msg->mxm_version  = MXLND_MSG_VERSION;
1092         /*   mxm_type */
1093         /* don't use mxlnd_tx_requires_credit() since we want PUT_ACK to
1094          * return credits as well */
1095         if (tx->mxc_msg_type != MXLND_MSG_CONN_REQ &&
1096             tx->mxc_msg_type != MXLND_MSG_CONN_ACK) {
1097                 msg->mxm_credits  = tx->mxc_conn->mxk_outstanding;
1098                 tx->mxc_conn->mxk_outstanding = 0;
1099         } else {
1100                 msg->mxm_credits  = 0;
1101         }
1102         /*   mxm_nob */
1103         msg->mxm_cksum    = 0;
1104         msg->mxm_srcnid   = kmxlnd_data.kmx_ni->ni_nid;
1105         msg->mxm_srcstamp = kmxlnd_data.kmx_incarnation;
1106         msg->mxm_dstnid   = tx->mxc_nid;
1107         /* if it is a new peer, the dststamp will be 0 */
1108         msg->mxm_dststamp = tx->mxc_conn->mxk_incarnation;
1109
1110         if (*kmxlnd_tunables.kmx_cksum) {
1111                 msg->mxm_cksum = mxlnd_cksum(msg, msg->mxm_nob);
1112         }
1113 }
1114
1115 int
1116 mxlnd_unpack_msg(kmx_msg_t *msg, int nob)
1117 {
1118         const int hdr_size      = offsetof(kmx_msg_t, mxm_u);
1119         __u32     msg_cksum     = 0;
1120         int       flip          = 0;
1121         int       msg_nob       = 0;
1122
1123         /* 6 bytes are enough to have received magic + version */
1124         if (nob < 6) {
1125                 CNETERR("not enough bytes for magic + hdr: %d\n", nob);
1126                 return -EPROTO;
1127         }
1128
1129         if (msg->mxm_magic == MXLND_MSG_MAGIC) {
1130                 flip = 0;
1131         } else if (msg->mxm_magic == __swab32(MXLND_MSG_MAGIC)) {
1132                 flip = 1;
1133         } else {
1134                 CNETERR("Bad magic: %08x\n", msg->mxm_magic);
1135                 return -EPROTO;
1136         }
1137
1138         if (msg->mxm_version !=
1139             (flip ? __swab16(MXLND_MSG_VERSION) : MXLND_MSG_VERSION)) {
1140                 CNETERR("Bad version: %d\n", msg->mxm_version);
1141                 return -EPROTO;
1142         }
1143
1144         if (nob < hdr_size) {
1145                 CNETERR("not enough for a header: %d\n", nob);
1146                 return -EPROTO;
1147         }
1148
1149         msg_nob = flip ? __swab32(msg->mxm_nob) : msg->mxm_nob;
1150         if (msg_nob > nob) {
1151                 CNETERR("Short message: got %d, wanted %d\n", nob, msg_nob);
1152                 return -EPROTO;
1153         }
1154
1155         /* checksum must be computed with mxm_cksum zero and BEFORE anything
1156          * gets flipped */
1157         msg_cksum = flip ? __swab32(msg->mxm_cksum) : msg->mxm_cksum;
1158         msg->mxm_cksum = 0;
1159         if (msg_cksum != 0 && msg_cksum != mxlnd_cksum(msg, msg_nob)) {
1160                 CNETERR("Bad checksum\n");
1161                 return -EPROTO;
1162         }
1163         msg->mxm_cksum = msg_cksum;
1164
1165         if (flip) {
1166                 /* leave magic unflipped as a clue to peer endianness */
1167                 __swab16s(&msg->mxm_version);
1168                 CLASSERT (sizeof(msg->mxm_type) == 1);
1169                 CLASSERT (sizeof(msg->mxm_credits) == 1);
1170                 msg->mxm_nob = msg_nob;
1171                 __swab64s(&msg->mxm_srcnid);
1172                 __swab64s(&msg->mxm_srcstamp);
1173                 __swab64s(&msg->mxm_dstnid);
1174                 __swab64s(&msg->mxm_dststamp);
1175         }
1176
1177         if (msg->mxm_srcnid == LNET_NID_ANY) {
1178                 CNETERR("Bad src nid: %s\n", libcfs_nid2str(msg->mxm_srcnid));
1179                 return -EPROTO;
1180         }
1181
1182         switch (msg->mxm_type) {
1183         default:
1184                 CNETERR("Unknown message type %x\n", msg->mxm_type);
1185                 return -EPROTO;
1186
1187         case MXLND_MSG_NOOP:
1188                 break;
1189
1190         case MXLND_MSG_EAGER:
1191                 if (msg_nob < offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[0])) {
1192                         CNETERR("Short EAGER: %d(%d)\n", msg_nob,
1193                                (int)offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[0]));
1194                         return -EPROTO;
1195                 }
1196                 break;
1197
1198         case MXLND_MSG_PUT_REQ:
1199                 if (msg_nob < hdr_size + sizeof(msg->mxm_u.put_req)) {
1200                         CNETERR("Short PUT_REQ: %d(%d)\n", msg_nob,
1201                                (int)(hdr_size + sizeof(msg->mxm_u.put_req)));
1202                         return -EPROTO;
1203                 }
1204                 if (flip)
1205                         __swab64s(&msg->mxm_u.put_req.mxprm_cookie);
1206                 break;
1207
1208         case MXLND_MSG_PUT_ACK:
1209                 if (msg_nob < hdr_size + sizeof(msg->mxm_u.put_ack)) {
1210                         CNETERR("Short PUT_ACK: %d(%d)\n", msg_nob,
1211                                (int)(hdr_size + sizeof(msg->mxm_u.put_ack)));
1212                         return -EPROTO;
1213                 }
1214                 if (flip) {
1215                         __swab64s(&msg->mxm_u.put_ack.mxpam_src_cookie);
1216                         __swab64s(&msg->mxm_u.put_ack.mxpam_dst_cookie);
1217                 }
1218                 break;
1219
1220         case MXLND_MSG_GET_REQ:
1221                 if (msg_nob < hdr_size + sizeof(msg->mxm_u.get_req)) {
1222                         CNETERR("Short GET_REQ: %d(%d)\n", msg_nob,
1223                                (int)(hdr_size + sizeof(msg->mxm_u.get_req)));
1224                         return -EPROTO;
1225                 }
1226                 if (flip) {
1227                         __swab64s(&msg->mxm_u.get_req.mxgrm_cookie);
1228                 }
1229                 break;
1230
1231         case MXLND_MSG_CONN_REQ:
1232         case MXLND_MSG_CONN_ACK:
1233                 if (msg_nob < hdr_size + sizeof(msg->mxm_u.conn_req)) {
1234                         CNETERR("Short connreq/ack: %d(%d)\n", msg_nob,
1235                                (int)(hdr_size + sizeof(msg->mxm_u.conn_req)));
1236                         return -EPROTO;
1237                 }
1238                 if (flip) {
1239                         __swab32s(&msg->mxm_u.conn_req.mxcrm_queue_depth);
1240                         __swab32s(&msg->mxm_u.conn_req.mxcrm_eager_size);
1241                 }
1242                 break;
1243         }
1244         return 0;
1245 }
1246
1247
1248 /**
1249  * mxlnd_recv_msg
1250  * @lntmsg - the LNET msg that this is continuing. If EAGER, then NULL.
1251  * @rx
1252  * @msg_type
1253  * @cookie
1254  * @length - length of incoming message
1255  * @pending - add to kmx_pending (0 is NO and 1 is YES)
1256  *
1257  * The caller gets the rx and sets nid, peer and conn if known.
1258  *
1259  * Returns 0 on success and -1 on failure
1260  */
1261 int
1262 mxlnd_recv_msg(lnet_msg_t *lntmsg, kmx_ctx_t *rx, u8 msg_type, u64 cookie, u32 length)
1263 {
1264         int             ret     = 0;
1265         mx_return_t     mxret   = MX_SUCCESS;
1266         uint64_t        mask    = ~(MXLND_ERROR_MASK);
1267
1268         rx->mxc_msg_type = msg_type;
1269         rx->mxc_lntmsg[0] = lntmsg; /* may be NULL if EAGER */
1270         rx->mxc_cookie = cookie;
1271         /* rx->mxc_match may already be set */
1272         /* rx->mxc_seg.segment_ptr is already set */
1273         rx->mxc_seg.segment_length = length;
1274         ret = mxlnd_q_pending_ctx(rx);
1275         if (ret == -1) {
1276                 /* the caller is responsible for calling conn_decref() if needed */
1277                 return -1;
1278         }
1279         mxret = mx_kirecv(kmxlnd_data.kmx_endpt, &rx->mxc_seg, 1, MX_PIN_PHYSICAL,
1280                           cookie, mask, (void *) rx, &rx->mxc_mxreq);
1281         if (mxret != MX_SUCCESS) {
1282                 mxlnd_deq_pending_ctx(rx);
1283                 CNETERR("mx_kirecv() failed with %s (%d)\n",
1284                         mx_strerror(mxret), (int) mxret);
1285                 return -1;
1286         }
1287         return 0;
1288 }
1289
1290
1291 /**
1292  * mxlnd_unexpected_recv - this is the callback function that will handle
1293  *                         unexpected receives
1294  * @context - NULL, ignore
1295  * @source - the peer's mx_endpoint_addr_t
1296  * @match_value - the msg's bits, should be MXLND_MSG_EAGER
1297  * @length - length of incoming message
1298  * @data_if_available - used for CONN_[REQ|ACK]
1299  *
1300  * If it is an eager-sized msg, we will call recv_msg() with the actual
1301  * length. If it is a large message, we will call recv_msg() with a
1302  * length of 0 bytes to drop it because we should never have a large,
1303  * unexpected message.
1304  *
1305  * NOTE - The MX library blocks until this function completes. Make it as fast as
1306  * possible. DO NOT allocate memory which can block!
1307  *
1308  * If we cannot get a rx or the conn is closed, drop the message on the floor
1309  * (i.e. recv 0 bytes and ignore).
1310  */
1311 mx_unexp_handler_action_t
1312 mxlnd_unexpected_recv(void *context, mx_endpoint_addr_t source,
1313                  uint64_t match_value, uint32_t length, void *data_if_available)
1314 {
1315         int             ret             = 0;
1316         kmx_ctx_t       *rx             = NULL;
1317         mx_ksegment_t   seg;
1318         u8              msg_type        = 0;
1319         u8              error           = 0;
1320         u64             cookie          = 0ULL;
1321         kmx_conn_t      *conn           = NULL;
1322         kmx_peer_t      *peer           = NULL;
1323         u64             nic_id          = 0ULL;
1324         u32             ep_id           = 0;
1325         u32             sid             = 0;
1326
1327         /* TODO this will change to the net struct */
1328         if (context != NULL) {
1329                 CNETERR("non-NULL context\n");
1330         }
1331
1332 #if MXLND_DEBUG
1333         CDEBUG(D_NET, "bits=0x%llx length=%d\n", match_value, length);
1334 #endif
1335
1336         mx_decompose_endpoint_addr2(source, &nic_id, &ep_id, &sid);
1337         mxlnd_parse_match(match_value, &msg_type, &error, &cookie);
1338         cfs_read_lock(&kmxlnd_data.kmx_global_lock);
1339         mx_get_endpoint_addr_context(source, (void **) &conn);
1340         if (conn) {
1341                 mxlnd_conn_addref(conn); /* add ref for this function */
1342                 peer = conn->mxk_peer;
1343         }
1344         cfs_read_unlock(&kmxlnd_data.kmx_global_lock);
1345
1346         if (msg_type == MXLND_MSG_BYE) {
1347                 if (conn) {
1348                         CDEBUG(D_NET, "peer %s sent BYE msg\n",
1349                                         libcfs_nid2str(peer->mxp_nid));
1350                         mxlnd_conn_disconnect(conn, 1, 0);
1351                         mxlnd_conn_decref(conn); /* drop ref taken above */
1352                 }
1353                 return MX_RECV_FINISHED;
1354         }
1355
1356         if (msg_type == MXLND_MSG_CONN_REQ) {
1357                 kmx_connparams_t       *cp      = NULL;
1358                 const int       expected        = offsetof(kmx_msg_t, mxm_u) +
1359                                                   sizeof(kmx_connreq_msg_t);
1360
1361                 if (conn) mxlnd_conn_decref(conn); /* drop ref taken above */
1362                 if (unlikely(length != expected || !data_if_available)) {
1363                         CNETERR("received invalid CONN_REQ from %llx "
1364                                 "length=%d (expected %d)\n", nic_id, length, expected);
1365                         mxlnd_send_message(source, MXLND_MSG_CONN_ACK, EPROTO, 0);
1366                         return MX_RECV_FINISHED;
1367                 }
1368
1369                 ret = mxlnd_connparams_alloc(&cp, context, source, match_value, length,
1370                                          conn, peer, data_if_available);
1371                 if (unlikely(ret != 0)) {
1372                         CNETERR("unable to alloc CONN_REQ from %llx:%d\n",
1373                                 nic_id, ep_id);
1374                         mxlnd_send_message(source, MXLND_MSG_CONN_ACK, ENOMEM, 0);
1375                         return MX_RECV_FINISHED;
1376                 }
1377                 cfs_spin_lock(&kmxlnd_data.kmx_conn_lock);
1378                 cfs_list_add_tail(&cp->mxr_list, &kmxlnd_data.kmx_conn_reqs);
1379                 cfs_spin_unlock(&kmxlnd_data.kmx_conn_lock);
1380                 cfs_up(&kmxlnd_data.kmx_conn_sem);
1381                 return MX_RECV_FINISHED;
1382         }
1383         if (msg_type == MXLND_MSG_CONN_ACK) {
1384                 kmx_connparams_t  *cp           = NULL;
1385                 const int       expected        = offsetof(kmx_msg_t, mxm_u) +
1386                                                   sizeof(kmx_connreq_msg_t);
1387
1388                 LASSERT(conn);
1389                 if (unlikely(error != 0)) {
1390                         CNETERR("received CONN_ACK from %s with error -%d\n",
1391                                libcfs_nid2str(peer->mxp_nid), (int) error);
1392                         mxlnd_conn_disconnect(conn, 1, 0);
1393                 } else if (unlikely(length != expected || !data_if_available)) {
1394                         CNETERR("received %s CONN_ACK from %s "
1395                                "length=%d (expected %d)\n",
1396                                data_if_available ? "short" : "missing",
1397                                libcfs_nid2str(peer->mxp_nid), length, expected);
1398                         mxlnd_conn_disconnect(conn, 1, 1);
1399                 } else {
1400                         /* peer is ready for messages */
1401                         ret = mxlnd_connparams_alloc(&cp, context, source, match_value, length,
1402                                          conn, peer, data_if_available);
1403                         if (unlikely(ret != 0)) {
1404                                 CNETERR("unable to alloc kmx_connparams_t"
1405                                                " from %llx:%d\n", nic_id, ep_id);
1406                                 mxlnd_conn_disconnect(conn, 1, 1);
1407                         } else {
1408                                 cfs_spin_lock(&kmxlnd_data.kmx_conn_lock);
1409                                 cfs_list_add_tail(&cp->mxr_list,
1410                                                   &kmxlnd_data.kmx_conn_reqs);
1411                                 cfs_spin_unlock(&kmxlnd_data.kmx_conn_lock);
1412                                 cfs_up(&kmxlnd_data.kmx_conn_sem);
1413                         }
1414                 }
1415                 mxlnd_conn_decref(conn); /* drop ref taken above */
1416
1417                 return MX_RECV_FINISHED;
1418         }
1419
1420         /* Handle unexpected messages (PUT_REQ and GET_REQ) */
1421
1422         LASSERT(peer != NULL && conn != NULL);
1423
1424         rx = mxlnd_get_idle_rx(conn);
1425         if (rx != NULL) {
1426                 if (length <= MXLND_MSG_SIZE) {
1427                         ret = mxlnd_recv_msg(NULL, rx, msg_type, match_value, length);
1428                 } else {
1429                         CNETERR("unexpected large receive with "
1430                                 "match_value=0x%llx length=%d\n",
1431                                 match_value, length);
1432                         ret = mxlnd_recv_msg(NULL, rx, msg_type, match_value, 0);
1433                 }
1434
1435                 if (ret == 0) {
1436                         /* hold conn ref until rx completes */
1437                         rx->mxc_conn = conn;
1438                         rx->mxc_peer = peer;
1439                         rx->mxc_nid = peer->mxp_nid;
1440                 } else {
1441                         CNETERR("could not post receive\n");
1442                         mxlnd_put_idle_rx(rx);
1443                 }
1444         }
1445
1446         /* Encountered error, drop incoming message on the floor */
1447         /* We could use MX_RECV_FINISHED but posting the receive of 0 bytes
1448          * uses the standard code path and acks the sender normally */
1449
1450         if (rx == NULL || ret != 0) {
1451                 mxlnd_conn_decref(conn); /* drop ref taken above */
1452                 if (rx == NULL) {
1453                         CNETERR("no idle rxs available - dropping rx"
1454                                 " 0x%llx from %s\n", match_value,
1455                                 libcfs_nid2str(peer->mxp_nid));
1456                 } else {
1457                         /* ret != 0 */
1458                         CNETERR("disconnected peer - dropping rx\n");
1459                 }
1460                 seg.segment_ptr = 0ULL;
1461                 seg.segment_length = 0;
1462                 mx_kirecv(kmxlnd_data.kmx_endpt, &seg, 1, MX_PIN_PHYSICAL,
1463                           match_value, ~0ULL, NULL, NULL);
1464         }
1465
1466         return MX_RECV_CONTINUE;
1467 }
1468
1469
1470 int
1471 mxlnd_get_peer_info(int index, lnet_nid_t *nidp, int *count)
1472 {
1473         int              i      = 0;
1474         int              ret    = -ENOENT;
1475         kmx_peer_t      *peer   = NULL;
1476
1477         cfs_read_lock(&kmxlnd_data.kmx_global_lock);
1478         for (i = 0; i < MXLND_HASH_SIZE; i++) {
1479                 cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],
1480                                         mxp_list) {
1481                         if (index-- == 0) {
1482                                 *nidp = peer->mxp_nid;
1483                                 *count = cfs_atomic_read(&peer->mxp_refcount);
1484                                 ret = 0;
1485                                 break;
1486                         }
1487                 }
1488         }
1489         cfs_read_unlock(&kmxlnd_data.kmx_global_lock);
1490
1491         return ret;
1492 }
1493
1494 void
1495 mxlnd_del_peer_locked(kmx_peer_t *peer)
1496 {
1497         if (peer->mxp_conn) {
1498                 mxlnd_conn_disconnect(peer->mxp_conn, 1, 1);
1499         } else {
1500                 cfs_list_del_init(&peer->mxp_list); /* remove from the global list */
1501                 mxlnd_peer_decref(peer); /* drop global list ref */
1502         }
1503         return;
1504 }
1505
1506 int
1507 mxlnd_del_peer(lnet_nid_t nid)
1508 {
1509         int             i       = 0;
1510         int             ret     = 0;
1511         kmx_peer_t      *peer   = NULL;
1512         kmx_peer_t      *next   = NULL;
1513
1514         if (nid != LNET_NID_ANY) {
1515                 peer = mxlnd_find_peer_by_nid(nid, 0); /* adds peer ref */
1516         }
1517         cfs_write_lock(&kmxlnd_data.kmx_global_lock);
1518         if (nid != LNET_NID_ANY) {
1519                 if (peer == NULL) {
1520                         ret = -ENOENT;
1521                 } else {
1522                         mxlnd_peer_decref(peer); /* and drops it */
1523                         mxlnd_del_peer_locked(peer);
1524                 }
1525         } else { /* LNET_NID_ANY */
1526                 for (i = 0; i < MXLND_HASH_SIZE; i++) {
1527                         cfs_list_for_each_entry_safe(peer, next,
1528                                                      &kmxlnd_data.kmx_peers[i],
1529                                                      mxp_list) {
1530                                 mxlnd_del_peer_locked(peer);
1531                         }
1532                 }
1533         }
1534         cfs_write_unlock(&kmxlnd_data.kmx_global_lock);
1535
1536         return ret;
1537 }
1538
1539 kmx_conn_t *
1540 mxlnd_get_conn_by_idx(int index)
1541 {
1542         int              i      = 0;
1543         kmx_peer_t      *peer   = NULL;
1544         kmx_conn_t      *conn   = NULL;
1545
1546         cfs_read_lock(&kmxlnd_data.kmx_global_lock);
1547         for (i = 0; i < MXLND_HASH_SIZE; i++) {
1548                 cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],
1549                                         mxp_list) {
1550                         cfs_list_for_each_entry(conn, &peer->mxp_conns,
1551                                                 mxk_list) {
1552                                 if (index-- > 0) {
1553                                         continue;
1554                                 }
1555
1556                                 mxlnd_conn_addref(conn); /* add ref here, dec in ctl() */
1557                                 cfs_read_unlock(&kmxlnd_data.kmx_global_lock);
1558                                 return conn;
1559                         }
1560                 }
1561         }
1562         cfs_read_unlock(&kmxlnd_data.kmx_global_lock);
1563
1564         return NULL;
1565 }
1566
1567 void
1568 mxlnd_close_matching_conns_locked(kmx_peer_t *peer)
1569 {
1570         kmx_conn_t      *conn   = NULL;
1571         kmx_conn_t      *next   = NULL;
1572
1573         cfs_list_for_each_entry_safe(conn, next, &peer->mxp_conns, mxk_list)
1574                 mxlnd_conn_disconnect(conn, 0, 1);
1575
1576         return;
1577 }
1578
1579 int
1580 mxlnd_close_matching_conns(lnet_nid_t nid)
1581 {
1582         int             i       = 0;
1583         int             ret     = 0;
1584         kmx_peer_t      *peer   = NULL;
1585
1586         cfs_write_lock(&kmxlnd_data.kmx_global_lock);
1587         if (nid != LNET_NID_ANY) {
1588                 peer = mxlnd_find_peer_by_nid_locked(nid); /* adds peer ref */
1589                 if (peer == NULL) {
1590                         ret = -ENOENT;
1591                 } else {
1592                         mxlnd_close_matching_conns_locked(peer);
1593                         mxlnd_peer_decref(peer); /* and drops it here */
1594                 }
1595         } else { /* LNET_NID_ANY */
1596                 for (i = 0; i < MXLND_HASH_SIZE; i++) {
1597                         cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],                                                mxp_list)
1598                                 mxlnd_close_matching_conns_locked(peer);
1599                 }
1600         }
1601         cfs_write_unlock(&kmxlnd_data.kmx_global_lock);
1602
1603         return ret;
1604 }
1605
1606 /**
1607  * mxlnd_ctl - modify MXLND parameters
1608  * @ni - LNET interface handle
1609  * @cmd - command to change
1610  * @arg - the ioctl data
1611  */
1612 int
1613 mxlnd_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
1614 {
1615         struct libcfs_ioctl_data *data  = arg;
1616         int                       ret   = -EINVAL;
1617
1618         LASSERT (ni == kmxlnd_data.kmx_ni);
1619
1620         switch (cmd) {
1621         case IOC_LIBCFS_GET_PEER: {
1622                 lnet_nid_t      nid     = 0;
1623                 int             count   = 0;
1624
1625                 ret = mxlnd_get_peer_info(data->ioc_count, &nid, &count);
1626                 data->ioc_nid    = nid;
1627                 data->ioc_count  = count;
1628                 break;
1629         }
1630         case IOC_LIBCFS_DEL_PEER: {
1631                 ret = mxlnd_del_peer(data->ioc_nid);
1632                 break;
1633         }
1634         case IOC_LIBCFS_GET_CONN: {
1635                 kmx_conn_t      *conn = NULL;
1636
1637                 conn = mxlnd_get_conn_by_idx(data->ioc_count);
1638                 if (conn == NULL) {
1639                         ret = -ENOENT;
1640                 } else {
1641                         ret = 0;
1642                         data->ioc_nid = conn->mxk_peer->mxp_nid;
1643                         mxlnd_conn_decref(conn); /* dec ref taken in get_conn_by_idx() */
1644                 }
1645                 break;
1646         }
1647         case IOC_LIBCFS_CLOSE_CONNECTION: {
1648                 ret = mxlnd_close_matching_conns(data->ioc_nid);
1649                 break;
1650         }
1651         default:
1652                 CNETERR("unknown ctl(%d)\n", cmd);
1653                 break;
1654         }
1655
1656         return ret;
1657 }
1658
1659 /**
1660  * mxlnd_peer_queue_tx_locked - add the tx to the peer's tx queue
1661  * @tx
1662  *
1663  * Add the tx to the peer's msg or data queue. The caller has locked the peer.
1664  */
1665 void
1666 mxlnd_peer_queue_tx_locked(kmx_ctx_t *tx)
1667 {
1668         u8              msg_type        = tx->mxc_msg_type;
1669         kmx_conn_t      *conn           = tx->mxc_conn;
1670
1671         LASSERT (msg_type != 0);
1672         LASSERT (tx->mxc_nid != 0);
1673         LASSERT (tx->mxc_peer != NULL);
1674         LASSERT (tx->mxc_conn != NULL);
1675
1676         tx->mxc_incarnation = conn->mxk_incarnation;
1677
1678         if (msg_type != MXLND_MSG_PUT_DATA &&
1679             msg_type != MXLND_MSG_GET_DATA) {
1680                 /* msg style tx */
1681                 if (mxlnd_tx_requires_credit(tx)) {
1682                         cfs_list_add_tail(&tx->mxc_list,
1683                                           &conn->mxk_tx_credit_queue);
1684                         conn->mxk_ntx_msgs++;
1685                 } else if (msg_type == MXLND_MSG_CONN_REQ ||
1686                            msg_type == MXLND_MSG_CONN_ACK) {
1687                         /* put conn msgs at the front of the queue */
1688                         cfs_list_add(&tx->mxc_list, &conn->mxk_tx_free_queue);
1689                 } else {
1690                         /* PUT_ACK, PUT_NAK */
1691                         cfs_list_add_tail(&tx->mxc_list,
1692                                           &conn->mxk_tx_free_queue);
1693                         conn->mxk_ntx_msgs++;
1694                 }
1695         } else {
1696                 /* data style tx */
1697                 cfs_list_add_tail(&tx->mxc_list, &conn->mxk_tx_free_queue);
1698                 conn->mxk_ntx_data++;
1699         }
1700
1701         return;
1702 }
1703
1704 /**
1705  * mxlnd_peer_queue_tx - add the tx to the global tx queue
1706  * @tx
1707  *
1708  * Add the tx to the peer's msg or data queue
1709  */
1710 static inline void
1711 mxlnd_peer_queue_tx(kmx_ctx_t *tx)
1712 {
1713         LASSERT(tx->mxc_peer != NULL);
1714         LASSERT(tx->mxc_conn != NULL);
1715         cfs_spin_lock(&tx->mxc_conn->mxk_lock);
1716         mxlnd_peer_queue_tx_locked(tx);
1717         cfs_spin_unlock(&tx->mxc_conn->mxk_lock);
1718
1719         return;
1720 }
1721
1722 /**
1723  * mxlnd_queue_tx - add the tx to the global tx queue
1724  * @tx
1725  *
1726  * Add the tx to the global queue and up the tx_queue_sem
1727  */
1728 void
1729 mxlnd_queue_tx(kmx_ctx_t *tx)
1730 {
1731         kmx_peer_t *peer   = tx->mxc_peer;
1732         LASSERT (tx->mxc_nid != 0);
1733
1734         if (peer != NULL) {
1735                 if (peer->mxp_incompatible &&
1736                     tx->mxc_msg_type != MXLND_MSG_CONN_ACK) {
1737                         /* let this fail now */
1738                         tx->mxc_errno = -ECONNABORTED;
1739                         mxlnd_conn_decref(peer->mxp_conn);
1740                         mxlnd_put_idle_tx(tx);
1741                         return;
1742                 }
1743                 if (tx->mxc_conn == NULL) {
1744                         int             ret     = 0;
1745                         kmx_conn_t      *conn   = NULL;
1746
1747                         ret = mxlnd_conn_alloc(&conn, peer); /* adds 2nd ref for tx... */
1748                         if (ret != 0) {
1749                                 tx->mxc_errno = ret;
1750                                 mxlnd_put_idle_tx(tx);
1751                                 goto done;
1752                         }
1753                         tx->mxc_conn = conn;
1754                         mxlnd_peer_decref(peer); /* and takes it from peer */
1755                 }
1756                 LASSERT(tx->mxc_conn != NULL);
1757                 mxlnd_peer_queue_tx(tx);
1758                 mxlnd_check_sends(peer);
1759         } else {
1760                 cfs_spin_lock(&kmxlnd_data.kmx_tx_queue_lock);
1761                 cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_tx_queue);
1762                 cfs_spin_unlock(&kmxlnd_data.kmx_tx_queue_lock);
1763                 cfs_up(&kmxlnd_data.kmx_tx_queue_sem);
1764         }
1765 done:
1766         return;
1767 }
1768
1769 int
1770 mxlnd_setup_iov(kmx_ctx_t *ctx, u32 niov, struct iovec *iov, u32 offset, u32 nob)
1771 {
1772         int             i                       = 0;
1773         int             sum                     = 0;
1774         int             old_sum                 = 0;
1775         int             nseg                    = 0;
1776         int             first_iov               = -1;
1777         int             first_iov_offset        = 0;
1778         int             first_found             = 0;
1779         int             last_iov                = -1;
1780         int             last_iov_length         = 0;
1781         mx_ksegment_t  *seg                     = NULL;
1782
1783         if (niov == 0) return 0;
1784         LASSERT(iov != NULL);
1785
1786         for (i = 0; i < niov; i++) {
1787                 sum = old_sum + (u32) iov[i].iov_len;
1788                 if (!first_found && (sum > offset)) {
1789                         first_iov = i;
1790                         first_iov_offset = offset - old_sum;
1791                         first_found = 1;
1792                         sum = (u32) iov[i].iov_len - first_iov_offset;
1793                         old_sum = 0;
1794                 }
1795                 if (sum >= nob) {
1796                         last_iov = i;
1797                         last_iov_length = (u32) iov[i].iov_len - (sum - nob);
1798                         if (first_iov == last_iov) last_iov_length -= first_iov_offset;
1799                         break;
1800                 }
1801                 old_sum = sum;
1802         }
1803         LASSERT(first_iov >= 0 && last_iov >= first_iov);
1804         nseg = last_iov - first_iov + 1;
1805         LASSERT(nseg > 0);
1806
1807         MXLND_ALLOC(seg, nseg * sizeof(*seg));
1808         if (seg == NULL) {
1809                 CNETERR("MXLND_ALLOC() failed\n");
1810                 return -1;
1811         }
1812         memset(seg, 0, nseg * sizeof(*seg));
1813         ctx->mxc_nseg = nseg;
1814         sum = 0;
1815         for (i = 0; i < nseg; i++) {
1816                 seg[i].segment_ptr = MX_PA_TO_U64(virt_to_phys(iov[first_iov + i].iov_base));
1817                 seg[i].segment_length = (u32) iov[first_iov + i].iov_len;
1818                 if (i == 0) {
1819                         seg[i].segment_ptr += (u64) first_iov_offset;
1820                         seg[i].segment_length -= (u32) first_iov_offset;
1821                 }
1822                 if (i == (nseg - 1)) {
1823                         seg[i].segment_length = (u32) last_iov_length;
1824                 }
1825                 sum += seg[i].segment_length;
1826         }
1827         ctx->mxc_seg_list = seg;
1828         ctx->mxc_pin_type = MX_PIN_PHYSICAL;
1829 #ifdef MX_PIN_FULLPAGES
1830         ctx->mxc_pin_type |= MX_PIN_FULLPAGES;
1831 #endif
1832         LASSERT(nob == sum);
1833         return 0;
1834 }
1835
1836 int
1837 mxlnd_setup_kiov(kmx_ctx_t *ctx, u32 niov, lnet_kiov_t *kiov, u32 offset, u32 nob)
1838 {
1839         int             i                       = 0;
1840         int             sum                     = 0;
1841         int             old_sum                 = 0;
1842         int             nseg                    = 0;
1843         int             first_kiov              = -1;
1844         int             first_kiov_offset       = 0;
1845         int             first_found             = 0;
1846         int             last_kiov               = -1;
1847         int             last_kiov_length        = 0;
1848         mx_ksegment_t  *seg                     = NULL;
1849
1850         if (niov == 0) return 0;
1851         LASSERT(kiov != NULL);
1852
1853         for (i = 0; i < niov; i++) {
1854                 sum = old_sum + kiov[i].kiov_len;
1855                 if (i == 0) sum -= kiov[i].kiov_offset;
1856                 if (!first_found && (sum > offset)) {
1857                         first_kiov = i;
1858                         first_kiov_offset = offset - old_sum;
1859                         if (i == 0) first_kiov_offset = kiov[i].kiov_offset;
1860                         first_found = 1;
1861                         sum = kiov[i].kiov_len - first_kiov_offset;
1862                         old_sum = 0;
1863                 }
1864                 if (sum >= nob) {
1865                         last_kiov = i;
1866                         last_kiov_length = kiov[i].kiov_len - (sum - nob);
1867                         if (first_kiov == last_kiov) last_kiov_length -= first_kiov_offset;
1868                         break;
1869                 }
1870                 old_sum = sum;
1871         }
1872         LASSERT(first_kiov >= 0 && last_kiov >= first_kiov);
1873         nseg = last_kiov - first_kiov + 1;
1874         LASSERT(nseg > 0);
1875
1876         MXLND_ALLOC(seg, nseg * sizeof(*seg));
1877         if (seg == NULL) {
1878                 CNETERR("MXLND_ALLOC() failed\n");
1879                 return -1;
1880         }
1881         memset(seg, 0, niov * sizeof(*seg));
1882         ctx->mxc_nseg = niov;
1883         sum = 0;
1884         for (i = 0; i < niov; i++) {
1885                 seg[i].segment_ptr = lnet_page2phys(kiov[first_kiov + i].kiov_page);
1886                 seg[i].segment_length = kiov[first_kiov + i].kiov_len;
1887                 if (i == 0) {
1888                         seg[i].segment_ptr += (u64) first_kiov_offset;
1889                         /* we have to add back the original kiov_offset */
1890                         seg[i].segment_length -= first_kiov_offset +
1891                                                  kiov[first_kiov].kiov_offset;
1892                 }
1893                 if (i == (nseg - 1)) {
1894                         seg[i].segment_length = last_kiov_length;
1895                 }
1896                 sum += seg[i].segment_length;
1897         }
1898         ctx->mxc_seg_list = seg;
1899         ctx->mxc_pin_type = MX_PIN_PHYSICAL;
1900 #ifdef MX_PIN_FULLPAGES
1901         ctx->mxc_pin_type |= MX_PIN_FULLPAGES;
1902 #endif
1903         LASSERT(nob == sum);
1904         return 0;
1905 }
1906
1907 void
1908 mxlnd_send_nak(kmx_ctx_t *tx, lnet_nid_t nid, int type, int status, __u64 cookie)
1909 {
1910         LASSERT(type == MXLND_MSG_PUT_ACK);
1911         mxlnd_init_tx_msg(tx, type, sizeof(kmx_putack_msg_t), tx->mxc_nid);
1912         tx->mxc_cookie = cookie;
1913         tx->mxc_msg->mxm_u.put_ack.mxpam_src_cookie = cookie;
1914         tx->mxc_msg->mxm_u.put_ack.mxpam_dst_cookie = ((u64) status << MXLND_ERROR_OFFSET); /* error code */
1915         tx->mxc_match = mxlnd_create_match(tx, status);
1916
1917         mxlnd_queue_tx(tx);
1918 }
1919
1920
1921 /**
1922  * mxlnd_send_data - get tx, map [k]iov, queue tx
1923  * @ni
1924  * @lntmsg
1925  * @peer
1926  * @msg_type
1927  * @cookie
1928  *
1929  * This setups the DATA send for PUT or GET.
1930  *
1931  * On success, it queues the tx, on failure it calls lnet_finalize()
1932  */
1933 void
1934 mxlnd_send_data(lnet_ni_t *ni, lnet_msg_t *lntmsg, kmx_peer_t *peer, u8 msg_type, u64 cookie)
1935 {
1936         int                     ret     = 0;
1937         lnet_process_id_t       target  = lntmsg->msg_target;
1938         unsigned int            niov    = lntmsg->msg_niov;
1939         struct iovec           *iov     = lntmsg->msg_iov;
1940         lnet_kiov_t            *kiov    = lntmsg->msg_kiov;
1941         unsigned int            offset  = lntmsg->msg_offset;
1942         unsigned int            nob     = lntmsg->msg_len;
1943         kmx_ctx_t              *tx      = NULL;
1944
1945         LASSERT(lntmsg != NULL);
1946         LASSERT(peer != NULL);
1947         LASSERT(msg_type == MXLND_MSG_PUT_DATA || msg_type == MXLND_MSG_GET_DATA);
1948         LASSERT((cookie>>MXLND_ERROR_OFFSET) == 0);
1949
1950         tx = mxlnd_get_idle_tx();
1951         if (tx == NULL) {
1952                 CNETERR("Can't allocate %s tx for %s\n",
1953                         msg_type == MXLND_MSG_PUT_DATA ? "PUT_DATA" : "GET_DATA",
1954                         libcfs_nid2str(target.nid));
1955                 goto failed_0;
1956         }
1957         tx->mxc_nid = target.nid;
1958         /* NOTE called when we have a ref on the conn, get one for this tx */
1959         mxlnd_conn_addref(peer->mxp_conn);
1960         tx->mxc_peer = peer;
1961         tx->mxc_conn = peer->mxp_conn;
1962         tx->mxc_msg_type = msg_type;
1963         tx->mxc_lntmsg[0] = lntmsg;
1964         tx->mxc_cookie = cookie;
1965         tx->mxc_match = mxlnd_create_match(tx, 0);
1966
1967         /* This setups up the mx_ksegment_t to send the DATA payload  */
1968         if (nob == 0) {
1969                 /* do not setup the segments */
1970                 CNETERR("nob = 0; why didn't we use an EAGER reply "
1971                         "to %s?\n", libcfs_nid2str(target.nid));
1972                 ret = 0;
1973         } else if (kiov == NULL) {
1974                 ret = mxlnd_setup_iov(tx, niov, iov, offset, nob);
1975         } else {
1976                 ret = mxlnd_setup_kiov(tx, niov, kiov, offset, nob);
1977         }
1978         if (ret != 0) {
1979                 CNETERR("Can't setup send DATA for %s\n",
1980                         libcfs_nid2str(target.nid));
1981                 tx->mxc_errno = -EIO;
1982                 goto failed_1;
1983         }
1984         mxlnd_queue_tx(tx);
1985         return;
1986
1987 failed_1:
1988         mxlnd_conn_decref(peer->mxp_conn);
1989         mxlnd_put_idle_tx(tx);
1990         return;
1991
1992 failed_0:
1993         CNETERR("no tx avail\n");
1994         lnet_finalize(ni, lntmsg, -EIO);
1995         return;
1996 }
1997
1998 /**
1999  * mxlnd_recv_data - map [k]iov, post rx
2000  * @ni
2001  * @lntmsg
2002  * @rx
2003  * @msg_type
2004  * @cookie
2005  *
2006  * This setups the DATA receive for PUT or GET.
2007  *
2008  * On success, it returns 0, on failure it returns -1
2009  */
2010 int
2011 mxlnd_recv_data(lnet_ni_t *ni, lnet_msg_t *lntmsg, kmx_ctx_t *rx, u8 msg_type, u64 cookie)
2012 {
2013         int                     ret     = 0;
2014         lnet_process_id_t       target  = lntmsg->msg_target;
2015         unsigned int            niov    = lntmsg->msg_niov;
2016         struct iovec           *iov     = lntmsg->msg_iov;
2017         lnet_kiov_t            *kiov    = lntmsg->msg_kiov;
2018         unsigned int            offset  = lntmsg->msg_offset;
2019         unsigned int            nob     = lntmsg->msg_len;
2020         mx_return_t             mxret   = MX_SUCCESS;
2021         u64                     mask    = ~(MXLND_ERROR_MASK);
2022
2023         /* above assumes MXLND_MSG_PUT_DATA */
2024         if (msg_type == MXLND_MSG_GET_DATA) {
2025                 niov = lntmsg->msg_md->md_niov;
2026                 iov = lntmsg->msg_md->md_iov.iov;
2027                 kiov = lntmsg->msg_md->md_iov.kiov;
2028                 offset = 0;
2029                 nob = lntmsg->msg_md->md_length;
2030         }
2031
2032         LASSERT(lntmsg != NULL);
2033         LASSERT(rx != NULL);
2034         LASSERT(msg_type == MXLND_MSG_PUT_DATA || msg_type == MXLND_MSG_GET_DATA);
2035         LASSERT((cookie>>MXLND_ERROR_OFFSET) == 0); /* ensure top 12 bits are 0 */
2036
2037         rx->mxc_msg_type = msg_type;
2038         rx->mxc_state = MXLND_CTX_PENDING;
2039         rx->mxc_nid = target.nid;
2040         /* if posting a GET_DATA, we may not yet know the peer */
2041         if (rx->mxc_peer != NULL) {
2042                 rx->mxc_conn = rx->mxc_peer->mxp_conn;
2043         }
2044         rx->mxc_lntmsg[0] = lntmsg;
2045         rx->mxc_cookie = cookie;
2046         rx->mxc_match = mxlnd_create_match(rx, 0);
2047         /* This setups up the mx_ksegment_t to receive the DATA payload  */
2048         if (kiov == NULL) {
2049                 ret = mxlnd_setup_iov(rx, niov, iov, offset, nob);
2050         } else {
2051                 ret = mxlnd_setup_kiov(rx, niov, kiov, offset, nob);
2052         }
2053         if (msg_type == MXLND_MSG_GET_DATA) {
2054                 rx->mxc_lntmsg[1] = lnet_create_reply_msg(kmxlnd_data.kmx_ni, lntmsg);
2055                 if (rx->mxc_lntmsg[1] == NULL) {
2056                         CNETERR("Can't create reply for GET -> %s\n",
2057                                 libcfs_nid2str(target.nid));
2058                         ret = -1;
2059                 }
2060         }
2061         if (ret != 0) {
2062                 CNETERR("Can't setup %s rx for %s\n",
2063                        msg_type == MXLND_MSG_PUT_DATA ? "PUT_DATA" : "GET_DATA",
2064                        libcfs_nid2str(target.nid));
2065                 return -1;
2066         }
2067         ret = mxlnd_q_pending_ctx(rx);
2068         if (ret == -1) {
2069                 return -1;
2070         }
2071         CDEBUG(D_NET, "receiving %s 0x%llx\n", mxlnd_msgtype_to_str(msg_type), rx->mxc_cookie);
2072         mxret = mx_kirecv(kmxlnd_data.kmx_endpt,
2073                           rx->mxc_seg_list, rx->mxc_nseg,
2074                           rx->mxc_pin_type, rx->mxc_match,
2075                           mask, (void *) rx,
2076                           &rx->mxc_mxreq);
2077         if (mxret != MX_SUCCESS) {
2078                 if (rx->mxc_conn != NULL) {
2079                         mxlnd_deq_pending_ctx(rx);
2080                 }
2081                 CNETERR("mx_kirecv() failed with %d for %s\n",
2082                         (int) mxret, libcfs_nid2str(target.nid));
2083                 return -1;
2084         }
2085
2086         return 0;
2087 }
2088
2089 /**
2090  * mxlnd_send - the LND required send function
2091  * @ni
2092  * @private
2093  * @lntmsg
2094  *
2095  * This must not block. Since we may not have a peer struct for the receiver,
2096  * it will append send messages on a global tx list. We will then up the
2097  * tx_queued's semaphore to notify it of the new send.
2098  */
2099 int
2100 mxlnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
2101 {
2102         int                     ret             = 0;
2103         int                     type            = lntmsg->msg_type;
2104         lnet_hdr_t             *hdr             = &lntmsg->msg_hdr;
2105         lnet_process_id_t       target          = lntmsg->msg_target;
2106         lnet_nid_t              nid             = target.nid;
2107         int                     target_is_router = lntmsg->msg_target_is_router;
2108         int                     routing         = lntmsg->msg_routing;
2109         unsigned int            payload_niov    = lntmsg->msg_niov;
2110         struct iovec           *payload_iov     = lntmsg->msg_iov;
2111         lnet_kiov_t            *payload_kiov    = lntmsg->msg_kiov;
2112         unsigned int            payload_offset  = lntmsg->msg_offset;
2113         unsigned int            payload_nob     = lntmsg->msg_len;
2114         kmx_ctx_t              *tx              = NULL;
2115         kmx_msg_t              *txmsg           = NULL;
2116         kmx_ctx_t              *rx              = (kmx_ctx_t *) private; /* for REPLY */
2117         kmx_ctx_t              *rx_data         = NULL;
2118         kmx_conn_t             *conn            = NULL;
2119         int                     nob             = 0;
2120         uint32_t                length          = 0;
2121         kmx_peer_t             *peer            = NULL;
2122         cfs_rwlock_t           *g_lock          = &kmxlnd_data.kmx_global_lock;
2123
2124         CDEBUG(D_NET, "sending %d bytes in %d frags to %s\n",
2125                        payload_nob, payload_niov, libcfs_id2str(target));
2126
2127         LASSERT (payload_nob == 0 || payload_niov > 0);
2128         LASSERT (payload_niov <= LNET_MAX_IOV);
2129         /* payload is either all vaddrs or all pages */
2130         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
2131
2132         /* private is used on LNET_GET_REPLY only, NULL for all other cases */
2133
2134         /* NOTE we may not know the peer if it is the very first PUT_REQ or GET_REQ
2135          * to a new peer, so create one if not found */
2136         peer = mxlnd_find_peer_by_nid(nid, 1); /* adds peer ref */
2137         if (peer == NULL || peer->mxp_conn == NULL) {
2138                 /* we could not find it nor could we create one or
2139                  * one exists but we cannot create a conn,
2140                  * fail this message */
2141                 if (peer) {
2142                         /* found peer without conn, drop ref taken above */
2143                         LASSERT(peer->mxp_conn == NULL);
2144                         mxlnd_peer_decref(peer);
2145                 }
2146                 return -ENOMEM;
2147         }
2148
2149         /* we have a peer with a conn */
2150
2151         if (unlikely(peer->mxp_incompatible)) {
2152                 mxlnd_peer_decref(peer); /* drop ref taken above */
2153         } else {
2154                 cfs_read_lock(g_lock);
2155                 conn = peer->mxp_conn;
2156                 if (conn && conn->mxk_status != MXLND_CONN_DISCONNECT) {
2157                         mxlnd_conn_addref(conn);
2158                 } else {
2159                         conn = NULL;
2160                 }
2161                 cfs_read_unlock(g_lock);
2162                 mxlnd_peer_decref(peer); /* drop peer ref taken above */
2163                 if (!conn)
2164                         return -ENOTCONN;
2165         }
2166
2167         LASSERT(peer && conn);
2168
2169         CDEBUG(D_NET, "%s: peer 0x%llx is 0x%p\n", __func__, nid, peer);
2170
2171         switch (type) {
2172         case LNET_MSG_ACK:
2173                 LASSERT (payload_nob == 0);
2174                 break;
2175
2176         case LNET_MSG_REPLY:
2177         case LNET_MSG_PUT:
2178                 /* Is the payload small enough not to need DATA? */
2179                 nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[payload_nob]);
2180                 if (nob <= MXLND_MSG_SIZE)
2181                         break;                  /* send EAGER */
2182
2183                 tx = mxlnd_get_idle_tx();
2184                 if (unlikely(tx == NULL)) {
2185                         CNETERR("Can't allocate %s tx for %s\n",
2186                                 type == LNET_MSG_PUT ? "PUT" : "REPLY",
2187                                 libcfs_nid2str(nid));
2188                         if (conn) mxlnd_conn_decref(conn);
2189                         return -ENOMEM;
2190                 }
2191
2192                 tx->mxc_peer = peer;
2193                 tx->mxc_conn = conn;
2194                 /* we added a conn ref above */
2195                 mxlnd_init_tx_msg (tx, MXLND_MSG_PUT_REQ, sizeof(kmx_putreq_msg_t), nid);
2196                 txmsg = tx->mxc_msg;
2197                 txmsg->mxm_u.put_req.mxprm_hdr = *hdr;
2198                 txmsg->mxm_u.put_req.mxprm_cookie = tx->mxc_cookie;
2199                 tx->mxc_match = mxlnd_create_match(tx, 0);
2200
2201                 /* we must post a receive _before_ sending the request.
2202                  * we need to determine how much to receive, it will be either
2203                  * a put_ack or a put_nak. The put_ack is larger, so use it. */
2204
2205                 rx = mxlnd_get_idle_rx(conn);
2206                 if (unlikely(rx == NULL)) {
2207                         CNETERR("Can't allocate rx for PUT_ACK for %s\n",
2208                                 libcfs_nid2str(nid));
2209                         mxlnd_put_idle_tx(tx);
2210                         if (conn) mxlnd_conn_decref(conn); /* for the ref taken above */
2211                         return -ENOMEM;
2212                 }
2213                 rx->mxc_nid = nid;
2214                 rx->mxc_peer = peer;
2215                 mxlnd_conn_addref(conn); /* for this rx */
2216                 rx->mxc_conn = conn;
2217                 rx->mxc_msg_type = MXLND_MSG_PUT_ACK;
2218                 rx->mxc_cookie = tx->mxc_cookie;
2219                 rx->mxc_match = mxlnd_create_match(rx, 0);
2220
2221                 length = offsetof(kmx_msg_t, mxm_u) + sizeof(kmx_putack_msg_t);
2222                 ret = mxlnd_recv_msg(lntmsg, rx, MXLND_MSG_PUT_ACK, rx->mxc_match, length);
2223                 if (unlikely(ret != 0)) {
2224                         CNETERR("recv_msg() failed for PUT_ACK for %s\n",
2225                                            libcfs_nid2str(nid));
2226                         rx->mxc_lntmsg[0] = NULL;
2227                         mxlnd_put_idle_rx(rx);
2228                         mxlnd_put_idle_tx(tx);
2229                         mxlnd_conn_decref(conn); /* for the rx... */
2230                         mxlnd_conn_decref(conn); /* and for the tx */
2231                         return -EHOSTUNREACH;
2232                 }
2233
2234                 mxlnd_queue_tx(tx);
2235                 return 0;
2236
2237         case LNET_MSG_GET:
2238                 if (routing || target_is_router)
2239                         break;                  /* send EAGER */
2240
2241                 /* is the REPLY message too small for DATA? */
2242                 nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[lntmsg->msg_md->md_length]);
2243                 if (nob <= MXLND_MSG_SIZE)
2244                         break;                  /* send EAGER */
2245
2246                 /* get tx (we need the cookie) , post rx for incoming DATA,
2247                  * then post GET_REQ tx */
2248                 tx = mxlnd_get_idle_tx();
2249                 if (unlikely(tx == NULL)) {
2250                         CNETERR("Can't allocate GET tx for %s\n",
2251                                 libcfs_nid2str(nid));
2252                         mxlnd_conn_decref(conn); /* for the ref taken above */
2253                         return -ENOMEM;
2254                 }
2255                 rx_data = mxlnd_get_idle_rx(conn);
2256                 if (unlikely(rx_data == NULL)) {
2257                         CNETERR("Can't allocate DATA rx for %s\n",
2258                                 libcfs_nid2str(nid));
2259                         mxlnd_put_idle_tx(tx);
2260                         mxlnd_conn_decref(conn); /* for the ref taken above */
2261                         return -ENOMEM;
2262                 }
2263                 rx_data->mxc_peer = peer;
2264                 /* NOTE no need to lock peer before adding conn ref since we took
2265                  * a conn ref for the tx (it cannot be freed between there and here ) */
2266                 mxlnd_conn_addref(conn); /* for the rx_data */
2267                 rx_data->mxc_conn = conn;
2268
2269                 ret = mxlnd_recv_data(ni, lntmsg, rx_data, MXLND_MSG_GET_DATA, tx->mxc_cookie);
2270                 if (unlikely(ret != 0)) {
2271                         CNETERR("Can't setup GET sink for %s\n",
2272                                 libcfs_nid2str(nid));
2273                         mxlnd_put_idle_rx(rx_data);
2274                         mxlnd_put_idle_tx(tx);
2275                         mxlnd_conn_decref(conn); /* for the rx_data... */
2276                         mxlnd_conn_decref(conn); /* and for the tx */
2277                         return -EIO;
2278                 }
2279
2280                 tx->mxc_peer = peer;
2281                 tx->mxc_conn = conn;
2282                 /* conn ref taken above */
2283                 mxlnd_init_tx_msg(tx, MXLND_MSG_GET_REQ, sizeof(kmx_getreq_msg_t), nid);
2284                 txmsg = tx->mxc_msg;
2285                 txmsg->mxm_u.get_req.mxgrm_hdr = *hdr;
2286                 txmsg->mxm_u.get_req.mxgrm_cookie = tx->mxc_cookie;
2287                 tx->mxc_match = mxlnd_create_match(tx, 0);
2288
2289                 mxlnd_queue_tx(tx);
2290                 return 0;
2291
2292         default:
2293                 LBUG();
2294                 mxlnd_conn_decref(conn); /* drop ref taken above */
2295                 return -EIO;
2296         }
2297
2298         /* send EAGER */
2299
2300         LASSERT (offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[payload_nob])
2301                 <= MXLND_MSG_SIZE);
2302
2303         tx = mxlnd_get_idle_tx();
2304         if (unlikely(tx == NULL)) {
2305                 CNETERR("Can't send %s to %s: tx descs exhausted\n",
2306                         mxlnd_lnetmsg_to_str(type), libcfs_nid2str(nid));
2307                 mxlnd_conn_decref(conn); /* drop ref taken above */
2308                 return -ENOMEM;
2309         }
2310
2311         tx->mxc_peer = peer;
2312         tx->mxc_conn = conn;
2313         /* conn ref taken above */
2314         nob = offsetof(kmx_eager_msg_t, mxem_payload[payload_nob]);
2315         mxlnd_init_tx_msg (tx, MXLND_MSG_EAGER, nob, nid);
2316         tx->mxc_match = mxlnd_create_match(tx, 0);
2317
2318         txmsg = tx->mxc_msg;
2319         txmsg->mxm_u.eager.mxem_hdr = *hdr;
2320
2321         if (payload_kiov != NULL)
2322                 lnet_copy_kiov2flat(MXLND_MSG_SIZE, txmsg,
2323                             offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
2324                             payload_niov, payload_kiov, payload_offset, payload_nob);
2325         else
2326                 lnet_copy_iov2flat(MXLND_MSG_SIZE, txmsg,
2327                             offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
2328                             payload_niov, payload_iov, payload_offset, payload_nob);
2329
2330         tx->mxc_lntmsg[0] = lntmsg;              /* finalise lntmsg on completion */
2331         mxlnd_queue_tx(tx);
2332         return 0;
2333 }
2334
2335 /**
2336  * mxlnd_recv - the LND required recv function
2337  * @ni
2338  * @private
2339  * @lntmsg
2340  * @delayed
2341  * @niov
2342  * @kiov
2343  * @offset
2344  * @mlen
2345  * @rlen
2346  *
2347  * This must not block.
2348  */
2349 int
2350 mxlnd_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg, int delayed,
2351              unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
2352              unsigned int offset, unsigned int mlen, unsigned int rlen)
2353 {
2354         int             ret             = 0;
2355         int             nob             = 0;
2356         int             len             = 0;
2357         kmx_ctx_t       *rx             = private;
2358         kmx_msg_t       *rxmsg          = rx->mxc_msg;
2359         lnet_nid_t       nid            = rx->mxc_nid;
2360         kmx_ctx_t       *tx             = NULL;
2361         kmx_msg_t       *txmsg          = NULL;
2362         kmx_peer_t      *peer           = rx->mxc_peer;
2363         kmx_conn_t      *conn           = peer->mxp_conn;
2364         u64              cookie         = 0ULL;
2365         int              msg_type       = rxmsg->mxm_type;
2366         int              repost         = 1;
2367         int              credit         = 0;
2368         int              finalize       = 0;
2369
2370         LASSERT (mlen <= rlen);
2371         /* Either all pages or all vaddrs */
2372         LASSERT (!(kiov != NULL && iov != NULL));
2373         LASSERT (peer && conn);
2374
2375         /* conn_addref(conn) already taken for the primary rx */
2376
2377         switch (msg_type) {
2378         case MXLND_MSG_EAGER:
2379                 nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[rlen]);
2380                 len = rx->mxc_status.xfer_length;
2381                 if (unlikely(nob > len)) {
2382                         CNETERR("Eager message from %s too big: %d(%d)\n",
2383                                 libcfs_nid2str(nid), nob, len);
2384                         ret = -EPROTO;
2385                         break;
2386                 }
2387
2388                 if (kiov != NULL)
2389                         lnet_copy_flat2kiov(niov, kiov, offset,
2390                                 MXLND_MSG_SIZE, rxmsg,
2391                                 offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
2392                                 mlen);
2393                 else
2394                         lnet_copy_flat2iov(niov, iov, offset,
2395                                 MXLND_MSG_SIZE, rxmsg,
2396                                 offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
2397                                 mlen);
2398                 finalize = 1;
2399                 credit = 1;
2400                 break;
2401
2402         case MXLND_MSG_PUT_REQ:
2403                 /* we are going to reuse the rx, store the needed info */
2404                 cookie = rxmsg->mxm_u.put_req.mxprm_cookie;
2405
2406                 /* get tx, post rx, send PUT_ACK */
2407
2408                 tx = mxlnd_get_idle_tx();
2409                 if (unlikely(tx == NULL)) {
2410                         CNETERR("Can't allocate tx for %s\n", libcfs_nid2str(nid));
2411                         /* Not replying will break the connection */
2412                         ret = -ENOMEM;
2413                         break;
2414                 }
2415                 if (unlikely(mlen == 0)) {
2416                         finalize = 1;
2417                         tx->mxc_peer = peer;
2418                         tx->mxc_conn = conn;
2419                         mxlnd_send_nak(tx, nid, MXLND_MSG_PUT_ACK, 0, cookie);
2420                         /* repost = 1 */
2421                         break;
2422                 }
2423
2424                 mxlnd_init_tx_msg(tx, MXLND_MSG_PUT_ACK, sizeof(kmx_putack_msg_t), nid);
2425                 tx->mxc_peer = peer;
2426                 tx->mxc_conn = conn;
2427                 /* no need to lock peer first since we already have a ref */
2428                 mxlnd_conn_addref(conn); /* for the tx */
2429                 txmsg = tx->mxc_msg;
2430                 txmsg->mxm_u.put_ack.mxpam_src_cookie = cookie;
2431                 txmsg->mxm_u.put_ack.mxpam_dst_cookie = tx->mxc_cookie;
2432                 tx->mxc_cookie = cookie;
2433                 tx->mxc_match = mxlnd_create_match(tx, 0);
2434
2435                 /* we must post a receive _before_ sending the PUT_ACK */
2436                 mxlnd_ctx_init(rx);
2437                 rx->mxc_state = MXLND_CTX_PREP;
2438                 rx->mxc_peer = peer;
2439                 rx->mxc_conn = conn;
2440                 /* do not take another ref for this rx, it is already taken */
2441                 rx->mxc_nid = peer->mxp_nid;
2442                 ret = mxlnd_recv_data(ni, lntmsg, rx, MXLND_MSG_PUT_DATA,
2443                                       txmsg->mxm_u.put_ack.mxpam_dst_cookie);
2444
2445                 if (unlikely(ret != 0)) {
2446                         /* Notify peer that it's over */
2447                         CNETERR("Can't setup PUT_DATA rx for %s: %d\n",
2448                                 libcfs_nid2str(nid), ret);
2449                         mxlnd_ctx_init(tx);
2450                         tx->mxc_state = MXLND_CTX_PREP;
2451                         tx->mxc_peer = peer;
2452                         tx->mxc_conn = conn;
2453                         /* finalize = 0, let the PUT_ACK tx finalize this */
2454                         tx->mxc_lntmsg[0] = rx->mxc_lntmsg[0];
2455                         tx->mxc_lntmsg[1] = rx->mxc_lntmsg[1];
2456                         /* conn ref already taken above */
2457                         mxlnd_send_nak(tx, nid, MXLND_MSG_PUT_ACK, ret, cookie);
2458                         /* repost = 1 */
2459                         break;
2460                 }
2461
2462                 mxlnd_queue_tx(tx);
2463                 /* do not return a credit until after PUT_DATA returns */
2464                 repost = 0;
2465                 break;
2466
2467         case MXLND_MSG_GET_REQ:
2468                 cookie = rxmsg->mxm_u.get_req.mxgrm_cookie;
2469
2470                 if (likely(lntmsg != NULL)) {
2471                         mxlnd_send_data(ni, lntmsg, rx->mxc_peer, MXLND_MSG_GET_DATA,
2472                                         cookie);
2473                 } else {
2474                         /* GET didn't match anything */
2475                         /* The initiator has a rx mapped to [k]iov. We cannot send a nak.
2476                          * We have to embed the error code in the match bits.
2477                          * Send the error in bits 52-59 and the cookie in bits 0-51 */
2478                         tx = mxlnd_get_idle_tx();
2479                         if (unlikely(tx == NULL)) {
2480                                 CNETERR("Can't get tx for GET NAK for %s\n",
2481                                         libcfs_nid2str(nid));
2482                                 /* we can't get a tx, notify the peer that the GET failed */
2483                                 mxlnd_send_message(conn->mxk_epa, MXLND_MSG_GET_DATA,
2484                                                    ENODATA, cookie);
2485                                 ret = -ENOMEM;
2486                                 break;
2487                         }
2488                         tx->mxc_msg_type = MXLND_MSG_GET_DATA;
2489                         tx->mxc_state = MXLND_CTX_PENDING;
2490                         tx->mxc_nid = nid;
2491                         tx->mxc_peer = peer;
2492                         tx->mxc_conn = conn;
2493                         /* no need to lock peer first since we already have a ref */
2494                         mxlnd_conn_addref(conn); /* for this tx */
2495                         tx->mxc_cookie = cookie;
2496                         tx->mxc_match = mxlnd_create_match(tx, ENODATA);
2497                         tx->mxc_pin_type = MX_PIN_PHYSICAL;
2498                         mxlnd_queue_tx(tx);
2499                 }
2500                 /* finalize lntmsg after tx completes */
2501                 break;
2502
2503         default:
2504                 LBUG();
2505         }
2506
2507         if (repost) {
2508                 /* we received a message, increment peer's outstanding credits */
2509                 if (credit == 1) {
2510                         cfs_spin_lock(&conn->mxk_lock);
2511                         conn->mxk_outstanding++;
2512                         cfs_spin_unlock(&conn->mxk_lock);
2513                 }
2514                 /* we are done with the rx */
2515                 mxlnd_put_idle_rx(rx);
2516                 mxlnd_conn_decref(conn);
2517         }
2518
2519         if (finalize == 1) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg, 0);
2520
2521         /* we received a credit, see if we can use it to send a msg */
2522         if (credit) mxlnd_check_sends(peer);
2523
2524         return ret;
2525 }
2526
2527 void
2528 mxlnd_sleep(unsigned long timeout)
2529 {
2530         cfs_set_current_state(CFS_TASK_INTERRUPTIBLE);
2531         cfs_schedule_timeout(timeout);
2532         return;
2533 }
2534
2535 /**
2536  * mxlnd_tx_queued - the generic send queue thread
2537  * @arg - thread id (as a void *)
2538  *
2539  * This thread moves send messages from the global tx_queue to the owning
2540  * peer's tx_[msg|data]_queue. If the peer does not exist, it creates one and adds
2541  * it to the global peer list.
2542  */
2543 int
2544 mxlnd_tx_queued(void *arg)
2545 {
2546         long                    id      = (long) arg;
2547         int                     ret     = 0;
2548         int                     found   = 0;
2549         kmx_ctx_t              *tx      = NULL;
2550         kmx_peer_t             *peer    = NULL;
2551         cfs_list_t             *queue   = &kmxlnd_data.kmx_tx_queue;
2552         cfs_spinlock_t         *tx_q_lock = &kmxlnd_data.kmx_tx_queue_lock;
2553         cfs_rwlock_t           *g_lock  = &kmxlnd_data.kmx_global_lock;
2554
2555         cfs_daemonize("mxlnd_tx_queued");
2556
2557         while (!(cfs_atomic_read(&kmxlnd_data.kmx_shutdown))) {
2558                 ret = cfs_down_interruptible(&kmxlnd_data.kmx_tx_queue_sem);
2559                 if (cfs_atomic_read(&kmxlnd_data.kmx_shutdown))
2560                         break;
2561                 if (ret != 0) // Should we check for -EINTR?
2562                         continue;
2563                 cfs_spin_lock(tx_q_lock);
2564                 if (cfs_list_empty (&kmxlnd_data.kmx_tx_queue)) {
2565                         cfs_spin_unlock(tx_q_lock);
2566                         continue;
2567                 }
2568                 tx = cfs_list_entry (queue->next, kmx_ctx_t, mxc_list);
2569                 cfs_list_del_init(&tx->mxc_list);
2570                 cfs_spin_unlock(tx_q_lock);
2571
2572                 found = 0;
2573                 peer = mxlnd_find_peer_by_nid(tx->mxc_nid, 0); /* adds peer ref */
2574                 if (peer != NULL) {
2575                         tx->mxc_peer = peer;
2576                         cfs_write_lock(g_lock);
2577                         if (peer->mxp_conn == NULL) {
2578                                 ret = mxlnd_conn_alloc_locked(&peer->mxp_conn, peer);
2579                                 if (ret != 0) {
2580                                         /* out of memory, give up and fail tx */
2581                                         tx->mxc_errno = -ENOMEM;
2582                                         mxlnd_peer_decref(peer);
2583                                         cfs_write_unlock(g_lock);
2584                                         mxlnd_put_idle_tx(tx);
2585                                         continue;
2586                                 }
2587                         }
2588                         tx->mxc_conn = peer->mxp_conn;
2589                         mxlnd_conn_addref(tx->mxc_conn); /* for this tx */
2590                         mxlnd_peer_decref(peer); /* drop peer ref taken above */
2591                         cfs_write_unlock(g_lock);
2592                         mxlnd_queue_tx(tx);
2593                         found = 1;
2594                 }
2595                 if (found == 0) {
2596                         int             hash    = 0;
2597                         kmx_peer_t     *peer    = NULL;
2598                         kmx_peer_t     *old     = NULL;
2599
2600                         hash = mxlnd_nid_to_hash(tx->mxc_nid);
2601
2602                         LASSERT(tx->mxc_msg_type != MXLND_MSG_PUT_DATA &&
2603                                 tx->mxc_msg_type != MXLND_MSG_GET_DATA);
2604                         /* create peer */
2605                         /* adds conn ref for this function */
2606                         ret = mxlnd_peer_alloc(&peer, tx->mxc_nid,
2607                                         *kmxlnd_tunables.kmx_board,
2608                                         *kmxlnd_tunables.kmx_ep_id, 0ULL);
2609                         if (ret != 0) {
2610                                 /* finalize message */
2611                                 tx->mxc_errno = ret;
2612                                 mxlnd_put_idle_tx(tx);
2613                                 continue;
2614                         }
2615                         tx->mxc_peer = peer;
2616                         tx->mxc_conn = peer->mxp_conn;
2617                         /* this tx will keep the conn ref taken in peer_alloc() */
2618
2619                         /* add peer to global peer list, but look to see
2620                          * if someone already created it after we released
2621                          * the read lock */
2622                         cfs_write_lock(g_lock);
2623                         old = mxlnd_find_peer_by_nid_locked(peer->mxp_nid);
2624                         if (old) {
2625                                 /* we have a peer ref on old */
2626                                 if (old->mxp_conn) {
2627                                         found = 1;
2628                                 } else {
2629                                         /* no conn */
2630                                         /* drop our ref taken above... */
2631                                         mxlnd_peer_decref(old);
2632                                         /* and delete it */
2633                                         mxlnd_del_peer_locked(old);
2634                                 }
2635                         }
2636
2637                         if (found == 0) {
2638                                 cfs_list_add_tail(&peer->mxp_list,
2639                                                   &kmxlnd_data.kmx_peers[hash]);
2640                                 cfs_atomic_inc(&kmxlnd_data.kmx_npeers);
2641                         } else {
2642                                 tx->mxc_peer = old;
2643                                 tx->mxc_conn = old->mxp_conn;
2644                                 LASSERT(old->mxp_conn != NULL);
2645                                 mxlnd_conn_addref(old->mxp_conn);
2646                                 mxlnd_conn_decref(peer->mxp_conn); /* drop ref taken above.. */
2647                                 mxlnd_conn_decref(peer->mxp_conn); /* drop peer's ref */
2648                                 mxlnd_peer_decref(peer);
2649                         }
2650                         cfs_write_unlock(g_lock);
2651
2652                         mxlnd_queue_tx(tx);
2653                 }
2654         }
2655         mxlnd_thread_stop(id);
2656         return 0;
2657 }
2658
2659 /* When calling this, we must not have the peer lock. */
2660 void
2661 mxlnd_iconnect(kmx_peer_t *peer, u8 msg_type)
2662 {
2663         mx_return_t     mxret           = MX_SUCCESS;
2664         mx_request_t    request;
2665         kmx_conn_t      *conn           = peer->mxp_conn;
2666         u64             match           = ((u64) msg_type) << MXLND_MSG_OFFSET;
2667
2668         /* NOTE we are holding a conn ref every time we call this function,
2669          * we do not need to lock the peer before taking another ref */
2670         mxlnd_conn_addref(conn); /* hold until CONN_REQ or CONN_ACK completes */
2671
2672         LASSERT(msg_type == MXLND_MSG_ICON_REQ || msg_type == MXLND_MSG_ICON_ACK);
2673
2674         if (peer->mxp_reconnect_time == 0) {
2675                 peer->mxp_reconnect_time = jiffies;
2676         }
2677
2678         if (peer->mxp_nic_id == 0ULL) {
2679                 int     ret     = 0;
2680
2681                 ret = mxlnd_ip2nic_id(LNET_NIDADDR(peer->mxp_nid),
2682                                       &peer->mxp_nic_id, MXLND_LOOKUP_COUNT);
2683                 if (ret == 0) {
2684                         mx_nic_id_to_board_number(peer->mxp_nic_id, &peer->mxp_board);
2685                 }
2686                 if (peer->mxp_nic_id == 0ULL && conn->mxk_status == MXLND_CONN_WAIT) {
2687                         /* not mapped yet, return */
2688                         cfs_spin_lock(&conn->mxk_lock);
2689                         mxlnd_set_conn_status(conn, MXLND_CONN_INIT);
2690                         cfs_spin_unlock(&conn->mxk_lock);
2691                 }
2692         }
2693
2694         if (cfs_time_after(jiffies,
2695                            peer->mxp_reconnect_time + MXLND_CONNECT_TIMEOUT) &&
2696             conn->mxk_status != MXLND_CONN_DISCONNECT) {
2697                 /* give up and notify LNET */
2698                 CDEBUG(D_NET, "timeout trying to connect to %s\n",
2699                        libcfs_nid2str(peer->mxp_nid));
2700                 mxlnd_conn_disconnect(conn, 0, 0);
2701                 mxlnd_conn_decref(conn);
2702                 return;
2703         }
2704
2705         mxret = mx_iconnect(kmxlnd_data.kmx_endpt, peer->mxp_nic_id,
2706                             peer->mxp_ep_id, MXLND_MSG_MAGIC, match,
2707                             (void *) peer, &request);
2708         if (unlikely(mxret != MX_SUCCESS)) {
2709                 cfs_spin_lock(&conn->mxk_lock);
2710                 mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
2711                 cfs_spin_unlock(&conn->mxk_lock);
2712                 CNETERR("mx_iconnect() failed with %s (%d) to %s\n",
2713                        mx_strerror(mxret), mxret, libcfs_nid2str(peer->mxp_nid));
2714                 mxlnd_conn_decref(conn);
2715         }
2716         mx_set_request_timeout(kmxlnd_data.kmx_endpt, request,
2717                                MXLND_CONNECT_TIMEOUT/CFS_HZ*1000);
2718         return;
2719 }
2720
2721 #define MXLND_STATS 0
2722
2723 int
2724 mxlnd_check_sends(kmx_peer_t *peer)
2725 {
2726         int             ret             = 0;
2727         int             found           = 0;
2728         mx_return_t     mxret           = MX_SUCCESS;
2729         kmx_ctx_t       *tx             = NULL;
2730         kmx_conn_t      *conn           = NULL;
2731         u8              msg_type        = 0;
2732         int             credit          = 0;
2733         int             status          = 0;
2734         int             ntx_posted      = 0;
2735         int             credits         = 0;
2736 #if MXLND_STATS
2737         static unsigned long last       = 0;
2738 #endif
2739
2740         if (unlikely(peer == NULL)) {
2741                 LASSERT(peer != NULL);
2742                 return -1;
2743         }
2744         cfs_write_lock(&kmxlnd_data.kmx_global_lock);
2745         conn = peer->mxp_conn;
2746         /* NOTE take a ref for the duration of this function since it is called
2747          * when there might not be any queued txs for this peer */
2748         if (conn) {
2749                 if (conn->mxk_status == MXLND_CONN_DISCONNECT) {
2750                         cfs_write_unlock(&kmxlnd_data.kmx_global_lock);
2751                         return -1;
2752                 }
2753                 mxlnd_conn_addref(conn); /* for duration of this function */
2754         }
2755         cfs_write_unlock(&kmxlnd_data.kmx_global_lock);
2756
2757         /* do not add another ref for this tx */
2758
2759         if (conn == NULL) {
2760                 /* we do not have any conns */
2761                 CNETERR("peer %s has no conn\n", libcfs_nid2str(peer->mxp_nid));
2762                 return -1;
2763         }
2764
2765 #if MXLND_STATS
2766         if (cfs_time_after(jiffies, last)) {
2767                 last = jiffies + CFS_HZ;
2768                 CDEBUG(D_NET, "status= %s credits= %d outstanding= %d ntx_msgs= %d "
2769                               "ntx_posted= %d ntx_data= %d data_posted= %d\n",
2770                               mxlnd_connstatus_to_str(conn->mxk_status), conn->mxk_credits,
2771                               conn->mxk_outstanding, conn->mxk_ntx_msgs, conn->mxk_ntx_posted,
2772                               conn->mxk_ntx_data, conn->mxk_data_posted);
2773         }
2774 #endif
2775
2776         cfs_spin_lock(&conn->mxk_lock);
2777         ntx_posted = conn->mxk_ntx_posted;
2778         credits = conn->mxk_credits;
2779
2780         LASSERT(ntx_posted <= *kmxlnd_tunables.kmx_peercredits);
2781         LASSERT(ntx_posted >= 0);
2782
2783         LASSERT(credits <= *kmxlnd_tunables.kmx_peercredits);
2784         LASSERT(credits >= 0);
2785
2786         /* check number of queued msgs, ignore data */
2787         if (conn->mxk_outstanding >= MXLND_CREDIT_HIGHWATER()) {
2788                 /* check if any txs queued that could return credits... */
2789                 if (cfs_list_empty(&conn->mxk_tx_credit_queue) ||
2790                     conn->mxk_ntx_msgs == 0) {
2791                         /* if not, send a NOOP */
2792                         tx = mxlnd_get_idle_tx();
2793                         if (likely(tx != NULL)) {
2794                                 tx->mxc_peer = peer;
2795                                 tx->mxc_conn = peer->mxp_conn;
2796                                 mxlnd_conn_addref(conn); /* for this tx */
2797                                 mxlnd_init_tx_msg (tx, MXLND_MSG_NOOP, 0, peer->mxp_nid);
2798                                 tx->mxc_match = mxlnd_create_match(tx, 0);
2799                                 mxlnd_peer_queue_tx_locked(tx);
2800                                 found = 1;
2801                                 goto done_locked;
2802                         }
2803                 }
2804         }
2805
2806         /* if the peer is not ready, try to connect */
2807         if (unlikely(conn->mxk_status == MXLND_CONN_INIT ||
2808             conn->mxk_status == MXLND_CONN_FAIL)) {
2809                 CDEBUG(D_NET, "status=%s\n", mxlnd_connstatus_to_str(conn->mxk_status));
2810                 mxlnd_set_conn_status(conn, MXLND_CONN_WAIT);
2811                 cfs_spin_unlock(&conn->mxk_lock);
2812                 mxlnd_iconnect(peer, (u8) MXLND_MSG_ICON_REQ);
2813                 goto done;
2814         }
2815
2816         while (!cfs_list_empty(&conn->mxk_tx_free_queue) ||
2817                !cfs_list_empty(&conn->mxk_tx_credit_queue)) {
2818                 /* We have something to send. If we have a queued tx that does not
2819                  * require a credit (free), choose it since its completion will
2820                  * return a credit (here or at the peer), complete a DATA or
2821                  * CONN_REQ or CONN_ACK. */
2822                 cfs_list_t *tmp_tx = NULL;
2823                 if (!cfs_list_empty(&conn->mxk_tx_free_queue)) {
2824                         tmp_tx = &conn->mxk_tx_free_queue;
2825                 } else {
2826                         tmp_tx = &conn->mxk_tx_credit_queue;
2827                 }
2828                 tx = cfs_list_entry(tmp_tx->next, kmx_ctx_t, mxc_list);
2829
2830                 msg_type = tx->mxc_msg_type;
2831
2832                 /* don't try to send a rx */
2833                 LASSERT(tx->mxc_type == MXLND_REQ_TX);
2834
2835                 /* ensure that it is a valid msg type */
2836                 LASSERT(msg_type == MXLND_MSG_CONN_REQ ||
2837                         msg_type == MXLND_MSG_CONN_ACK ||
2838                         msg_type == MXLND_MSG_NOOP     ||
2839                         msg_type == MXLND_MSG_EAGER    ||
2840                         msg_type == MXLND_MSG_PUT_REQ  ||
2841                         msg_type == MXLND_MSG_PUT_ACK  ||
2842                         msg_type == MXLND_MSG_PUT_DATA ||
2843                         msg_type == MXLND_MSG_GET_REQ  ||
2844                         msg_type == MXLND_MSG_GET_DATA);
2845                 LASSERT(tx->mxc_peer == peer);
2846                 LASSERT(tx->mxc_nid == peer->mxp_nid);
2847
2848                 credit = mxlnd_tx_requires_credit(tx);
2849                 if (credit) {
2850
2851                         if (conn->mxk_ntx_posted == *kmxlnd_tunables.kmx_peercredits) {
2852                                 CDEBUG(D_NET, "%s: posted enough\n",
2853                                               libcfs_nid2str(peer->mxp_nid));
2854                                 goto done_locked;
2855                         }
2856
2857                         if (conn->mxk_credits == 0) {
2858                                 CDEBUG(D_NET, "%s: no credits\n",
2859                                               libcfs_nid2str(peer->mxp_nid));
2860                                 goto done_locked;
2861                         }
2862
2863                         if (conn->mxk_credits == 1 &&      /* last credit reserved for */
2864                             conn->mxk_outstanding == 0) {  /* giving back credits */
2865                                 CDEBUG(D_NET, "%s: not using last credit\n",
2866                                               libcfs_nid2str(peer->mxp_nid));
2867                                 goto done_locked;
2868                         }
2869                 }
2870
2871                 if (unlikely(conn->mxk_status != MXLND_CONN_READY)) {
2872                         if ( ! (msg_type == MXLND_MSG_CONN_REQ ||
2873                                 msg_type == MXLND_MSG_CONN_ACK)) {
2874                                 CDEBUG(D_NET, "peer status is %s for tx 0x%llx (%s)\n",
2875                                              mxlnd_connstatus_to_str(conn->mxk_status),
2876                                              tx->mxc_cookie,
2877                                              mxlnd_msgtype_to_str(tx->mxc_msg_type));
2878                                 if (conn->mxk_status == MXLND_CONN_DISCONNECT ||
2879                                     cfs_time_aftereq(jiffies, tx->mxc_deadline)) {
2880                                         cfs_list_del_init(&tx->mxc_list);
2881                                         tx->mxc_errno = -ECONNABORTED;
2882                                         cfs_spin_unlock(&conn->mxk_lock);
2883                                         mxlnd_put_idle_tx(tx);
2884                                         mxlnd_conn_decref(conn);
2885                                         goto done;
2886                                 }
2887                                 goto done_locked;
2888                         }
2889                 }
2890
2891                 cfs_list_del_init(&tx->mxc_list);
2892
2893                 /* handle credits, etc now while we have the lock to avoid races */
2894                 if (credit) {
2895                         conn->mxk_credits--;
2896                         conn->mxk_ntx_posted++;
2897                 }
2898                 if (msg_type != MXLND_MSG_PUT_DATA &&
2899                     msg_type != MXLND_MSG_GET_DATA) {
2900                         if (msg_type != MXLND_MSG_CONN_REQ &&
2901                             msg_type != MXLND_MSG_CONN_ACK) {
2902                                 conn->mxk_ntx_msgs--;
2903                         }
2904                 }
2905                 if (tx->mxc_incarnation == 0 &&
2906                     conn->mxk_incarnation != 0) {
2907                         tx->mxc_incarnation = conn->mxk_incarnation;
2908                 }
2909
2910                 /* if this is a NOOP and (1) mxp_conn->mxk_outstanding < CREDIT_HIGHWATER
2911                  * or (2) there is a non-DATA msg that can return credits in the
2912                  * queue, then drop this duplicate NOOP */
2913                 if (unlikely(msg_type == MXLND_MSG_NOOP)) {
2914                         if ((conn->mxk_outstanding < MXLND_CREDIT_HIGHWATER()) ||
2915                             (conn->mxk_ntx_msgs >= 1)) {
2916                                 conn->mxk_credits++;
2917                                 conn->mxk_ntx_posted--;
2918                                 cfs_spin_unlock(&conn->mxk_lock);
2919                                 /* redundant NOOP */
2920                                 mxlnd_put_idle_tx(tx);
2921                                 mxlnd_conn_decref(conn);
2922                                 CDEBUG(D_NET, "%s: redundant noop\n",
2923                                               libcfs_nid2str(peer->mxp_nid));
2924                                 found = 1;
2925                                 goto done;
2926                         }
2927                 }
2928
2929                 found = 1;
2930                 if (likely((msg_type != MXLND_MSG_PUT_DATA) &&
2931                     (msg_type != MXLND_MSG_GET_DATA))) {
2932                         mxlnd_pack_msg_locked(tx);
2933                 }
2934
2935                 mxret = MX_SUCCESS;
2936
2937                 status = conn->mxk_status;
2938                 cfs_spin_unlock(&conn->mxk_lock);
2939
2940                 if (likely((status == MXLND_CONN_READY) ||
2941                     (msg_type == MXLND_MSG_CONN_REQ) ||
2942                     (msg_type == MXLND_MSG_CONN_ACK))) {
2943                         ret = 0;
2944                         if (msg_type != MXLND_MSG_CONN_REQ &&
2945                             msg_type != MXLND_MSG_CONN_ACK) {
2946                                 /* add to the pending list */
2947                                 ret = mxlnd_q_pending_ctx(tx);
2948                         } else {
2949                                 /* CONN_REQ/ACK */
2950                                 tx->mxc_state = MXLND_CTX_PENDING;
2951                         }
2952
2953                         if (ret == 0) {
2954                                 if (likely(msg_type != MXLND_MSG_PUT_DATA &&
2955                                     msg_type != MXLND_MSG_GET_DATA)) {
2956                                         /* send a msg style tx */
2957                                         LASSERT(tx->mxc_nseg == 1);
2958                                         LASSERT(tx->mxc_pin_type == MX_PIN_PHYSICAL);
2959                                         CDEBUG(D_NET, "sending %s 0x%llx\n",
2960                                                mxlnd_msgtype_to_str(msg_type),
2961                                                tx->mxc_cookie);
2962                                         mxret = mx_kisend(kmxlnd_data.kmx_endpt,
2963                                                           &tx->mxc_seg,
2964                                                           tx->mxc_nseg,
2965                                                           tx->mxc_pin_type,
2966                                                           conn->mxk_epa,
2967                                                           tx->mxc_match,
2968                                                           (void *) tx,
2969                                                           &tx->mxc_mxreq);
2970                                 } else {
2971                                         /* send a DATA tx */
2972                                         cfs_spin_lock(&conn->mxk_lock);
2973                                         conn->mxk_ntx_data--;
2974                                         conn->mxk_data_posted++;
2975                                         cfs_spin_unlock(&conn->mxk_lock);
2976                                         CDEBUG(D_NET, "sending %s 0x%llx\n",
2977                                                mxlnd_msgtype_to_str(msg_type),
2978                                                tx->mxc_cookie);
2979                                         mxret = mx_kisend(kmxlnd_data.kmx_endpt,
2980                                                           tx->mxc_seg_list,
2981                                                           tx->mxc_nseg,
2982                                                           tx->mxc_pin_type,
2983                                                           conn->mxk_epa,
2984                                                           tx->mxc_match,
2985                                                           (void *) tx,
2986                                                           &tx->mxc_mxreq);
2987                                 }
2988                         } else {
2989                                 /* ret != 0 */
2990                                 mxret = MX_CONNECTION_FAILED;
2991                         }
2992                         if (likely(mxret == MX_SUCCESS)) {
2993                                 ret = 0;
2994                         } else {
2995                                 CNETERR("mx_kisend() failed with %s (%d) "
2996                                         "sending to %s\n", mx_strerror(mxret), (int) mxret,
2997                                        libcfs_nid2str(peer->mxp_nid));
2998                                 /* NOTE mx_kisend() only fails if there are not enough
2999                                 * resources. Do not change the connection status. */
3000                                 if (mxret == MX_NO_RESOURCES) {
3001                                         tx->mxc_errno = -ENOMEM;
3002                                 } else {
3003                                         tx->mxc_errno = -ECONNABORTED;
3004                                 }
3005                                 if (credit) {
3006                                         cfs_spin_lock(&conn->mxk_lock);
3007                                         conn->mxk_ntx_posted--;
3008                                         conn->mxk_credits++;
3009                                         cfs_spin_unlock(&conn->mxk_lock);
3010                                 } else if (msg_type == MXLND_MSG_PUT_DATA ||
3011                                         msg_type == MXLND_MSG_GET_DATA) {
3012                                         cfs_spin_lock(&conn->mxk_lock);
3013                                         conn->mxk_data_posted--;
3014                                         cfs_spin_unlock(&conn->mxk_lock);
3015                                 }
3016                                 if (msg_type != MXLND_MSG_PUT_DATA &&
3017                                     msg_type != MXLND_MSG_GET_DATA &&
3018                                     msg_type != MXLND_MSG_CONN_REQ &&
3019                                     msg_type != MXLND_MSG_CONN_ACK) {
3020                                         cfs_spin_lock(&conn->mxk_lock);
3021                                         conn->mxk_outstanding += tx->mxc_msg->mxm_credits;
3022                                         cfs_spin_unlock(&conn->mxk_lock);
3023                                 }
3024                                 if (msg_type != MXLND_MSG_CONN_REQ &&
3025                                     msg_type != MXLND_MSG_CONN_ACK) {
3026                                         /* remove from the pending list */
3027                                         mxlnd_deq_pending_ctx(tx);
3028                                 }
3029                                 mxlnd_put_idle_tx(tx);
3030                                 mxlnd_conn_decref(conn);
3031                         }
3032                 }
3033                 cfs_spin_lock(&conn->mxk_lock);
3034         }
3035 done_locked:
3036         cfs_spin_unlock(&conn->mxk_lock);
3037 done:
3038         mxlnd_conn_decref(conn); /* drop ref taken at start of function */