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) 2012, Intel Corporation.
  31  *
  32  * Copyright (C) 2006 Myricom, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lnet/klnds/mxlnd/mxlnd.c
  39  *
  40  * Author: Eric Barton <eric@bartonsoftware.com>
  41  * Author: Scott Atchley <atchley at myri.com>
  42  */
  43
  44 #include "mxlnd.h"
  45
  46 mx_endpoint_addr_t MX_EPA_NULL; /* use to determine if an endpoint is NULL */
  47
  48 inline int
  49 mxlnd_endpoint_addr_null(mx_endpoint_addr_t epa)
  50 {
  51         /* if memcmp() == 0, it is NULL */
  52         return !(memcmp(&epa, &MX_EPA_NULL, sizeof(epa)));
  53 }
  54
  55 char *
  56 mxlnd_ctxstate_to_str(int mxc_state)
  57 {
  58         switch (mxc_state) {
  59         case MXLND_CTX_INIT:
  60                 return "MXLND_CTX_INIT";
  61         case MXLND_CTX_IDLE:
  62                 return "MXLND_CTX_IDLE";
  63         case MXLND_CTX_PREP:
  64                 return "MXLND_CTX_PREP";
  65         case MXLND_CTX_PENDING:
  66                 return "MXLND_CTX_PENDING";
  67         case MXLND_CTX_COMPLETED:
  68                 return "MXLND_CTX_COMPLETED";
  69         case MXLND_CTX_CANCELED:
  70                 return "MXLND_CTX_CANCELED";
  71         default:
  72                 return "*unknown*";
  73         }
  74 }
  75
  76 char *
  77 mxlnd_connstatus_to_str(int mxk_status)
  78 {
  79         switch (mxk_status) {
  80         case MXLND_CONN_READY:
  81                 return "MXLND_CONN_READY";
  82         case MXLND_CONN_INIT:
  83                 return "MXLND_CONN_INIT";
  84         case MXLND_CONN_WAIT:
  85                 return "MXLND_CONN_WAIT";
  86         case MXLND_CONN_DISCONNECT:
  87                 return "MXLND_CONN_DISCONNECT";
  88         case MXLND_CONN_FAIL:
  89                 return "MXLND_CONN_FAIL";
  90         default:
  91                 return "unknown";
  92         }
  93 }
  94
  95 char *
  96 mxlnd_msgtype_to_str(int type) {
  97         switch (type) {
  98         case MXLND_MSG_EAGER:
  99                 return "MXLND_MSG_EAGER";
 100         case MXLND_MSG_CONN_REQ:
 101                 return "MXLND_MSG_CONN_REQ";
 102         case MXLND_MSG_CONN_ACK:
 103                 return "MXLND_MSG_CONN_ACK";
 104         case MXLND_MSG_BYE:
 105                 return "MXLND_MSG_BYE";
 106         case MXLND_MSG_NOOP:
 107                 return "MXLND_MSG_NOOP";
 108         case MXLND_MSG_PUT_REQ:
 109                 return "MXLND_MSG_PUT_REQ";
 110         case MXLND_MSG_PUT_ACK:
 111                 return "MXLND_MSG_PUT_ACK";
 112         case MXLND_MSG_PUT_DATA:
 113                 return "MXLND_MSG_PUT_DATA";
 114         case MXLND_MSG_GET_REQ:
 115                 return "MXLND_MSG_GET_REQ";
 116         case MXLND_MSG_GET_DATA:
 117                 return "MXLND_MSG_GET_DATA";
 118         default:
 119                 return "unknown";
 120         }
 121 }
 122
 123 char *
 124 mxlnd_lnetmsg_to_str(int type)
 125 {
 126         switch (type) {
 127         case LNET_MSG_ACK:
 128                 return "LNET_MSG_ACK";
 129         case LNET_MSG_PUT:
 130                 return "LNET_MSG_PUT";
 131         case LNET_MSG_GET:
 132                 return "LNET_MSG_GET";
 133         case LNET_MSG_REPLY:
 134                 return "LNET_MSG_REPLY";
 135         case LNET_MSG_HELLO:
 136                 return "LNET_MSG_HELLO";
 137         default:
 138                 LBUG();
 139                 return "*unknown*";
 140         }
 141 }
 142
 143 static inline u64
 144 mxlnd_create_match(kmx_ctx_t *ctx, u8 error)
 145 {
 146         u64 type        = (u64) ctx->mxc_msg_type;
 147         u64 err         = (u64) error;
 148         u64 match       = 0ULL;
 149
 150         mxlnd_valid_msg_type(ctx->mxc_msg_type);
 151         LASSERT(ctx->mxc_cookie >> MXLND_ERROR_OFFSET == 0);
 152         match = (type << MXLND_MSG_OFFSET) | (err << MXLND_ERROR_OFFSET) | ctx->mxc_cookie;
 153         return match;
 154 }
 155
 156 static inline void
 157 mxlnd_parse_match(u64 match, u8 *msg_type, u8 *error, u64 *cookie)
 158 {
 159         *msg_type = (u8) MXLND_MSG_TYPE(match);
 160         *error    = (u8) MXLND_ERROR_VAL(match);
 161         *cookie   = match & MXLND_MAX_COOKIE;
 162         mxlnd_valid_msg_type(*msg_type);
 163         return;
 164 }
 165
 166 kmx_ctx_t *
 167 mxlnd_get_idle_rx(kmx_conn_t *conn)
 168 {
 169         cfs_list_t              *rxs    = NULL;
 170         kmx_ctx_t               *rx     = NULL;
 171
 172         LASSERT(conn != NULL);
 173
 174         rxs = &conn->mxk_rx_idle;
 175
 176         spin_lock(&conn->mxk_lock);
 177
 178         if (cfs_list_empty (rxs)) {
 179                 spin_unlock(&conn->mxk_lock);
 180                 return NULL;
 181         }
 182
 183         rx = cfs_list_entry (rxs->next, kmx_ctx_t, mxc_list);
 184         cfs_list_del_init(&rx->mxc_list);
 185         spin_unlock(&conn->mxk_lock);
 186
 187 #if MXLND_DEBUG
 188         if (rx->mxc_get != rx->mxc_put) {
 189                 CNETERR("*** RX get (%llu) != put (%llu) ***\n", rx->mxc_get, rx->mxc_put);
 190                 CNETERR("*** incarnation= %lld ***\n", rx->mxc_incarnation);
 191                 CNETERR("*** deadline= %ld ***\n", rx->mxc_deadline);
 192                 CNETERR("*** state= %s ***\n", mxlnd_ctxstate_to_str(rx->mxc_state));
 193                 CNETERR("*** listed?= %d ***\n", !cfs_list_empty(&rx->mxc_list));
 194                 CNETERR("*** nid= 0x%llx ***\n", rx->mxc_nid);
 195                 CNETERR("*** peer= 0x%p ***\n", rx->mxc_peer);
 196                 CNETERR("*** msg_type= %s ***\n", mxlnd_msgtype_to_str(rx->mxc_msg_type));
 197                 CNETERR("*** cookie= 0x%llx ***\n", rx->mxc_cookie);
 198                 CNETERR("*** nob= %d ***\n", rx->mxc_nob);
 199         }
 200 #endif
 201         LASSERT (rx->mxc_get == rx->mxc_put);
 202
 203         rx->mxc_get++;
 204
 205         LASSERT (rx->mxc_state == MXLND_CTX_IDLE);
 206         rx->mxc_state = MXLND_CTX_PREP;
 207         rx->mxc_deadline = jiffies + MXLND_COMM_TIMEOUT;
 208
 209         return rx;
 210 }
 211
 212 int
 213 mxlnd_put_idle_rx(kmx_ctx_t *rx)
 214 {
 215         kmx_conn_t              *conn   = rx->mxc_conn;
 216         cfs_list_t              *rxs    = &conn->mxk_rx_idle;
 217
 218         LASSERT(rx->mxc_type == MXLND_REQ_RX);
 219
 220         mxlnd_ctx_init(rx);
 221
 222         rx->mxc_put++;
 223         LASSERT(rx->mxc_get == rx->mxc_put);
 224
 225         spin_lock(&conn->mxk_lock);
 226         cfs_list_add(&rx->mxc_list, rxs);
 227         spin_unlock(&conn->mxk_lock);
 228         return 0;
 229 }
 230
 231 kmx_ctx_t *
 232 mxlnd_get_idle_tx(void)
 233 {
 234         cfs_list_t              *tmp    = &kmxlnd_data.kmx_tx_idle;
 235         kmx_ctx_t               *tx     = NULL;
 236
 237         spin_lock(&kmxlnd_data.kmx_tx_idle_lock);
 238
 239         if (cfs_list_empty (&kmxlnd_data.kmx_tx_idle)) {
 240                 CNETERR("%d txs in use\n", kmxlnd_data.kmx_tx_used);
 241                 spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);
 242                 return NULL;
 243         }
 244
 245         tmp = &kmxlnd_data.kmx_tx_idle;
 246         tx = cfs_list_entry (tmp->next, kmx_ctx_t, mxc_list);
 247         cfs_list_del_init(&tx->mxc_list);
 248
 249         /* Allocate a new completion cookie.  It might not be needed,
 250          * but we've got a lock right now and we're unlikely to
 251          * wrap... */
 252         tx->mxc_cookie = kmxlnd_data.kmx_tx_next_cookie++;
 253         if (kmxlnd_data.kmx_tx_next_cookie > MXLND_MAX_COOKIE) {
 254                 kmxlnd_data.kmx_tx_next_cookie = 1;
 255         }
 256         kmxlnd_data.kmx_tx_used++;
 257         spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);
 258
 259         LASSERT (tx->mxc_get == tx->mxc_put);
 260
 261         tx->mxc_get++;
 262
 263         LASSERT (tx->mxc_state == MXLND_CTX_IDLE);
 264         LASSERT (tx->mxc_lntmsg[0] == NULL);
 265         LASSERT (tx->mxc_lntmsg[1] == NULL);
 266
 267         tx->mxc_state = MXLND_CTX_PREP;
 268         tx->mxc_deadline = jiffies + MXLND_COMM_TIMEOUT;
 269
 270         return tx;
 271 }
 272
 273 void
 274 mxlnd_conn_disconnect(kmx_conn_t *conn, int mx_dis, int send_bye);
 275
 276 int
 277 mxlnd_put_idle_tx(kmx_ctx_t *tx)
 278 {
 279         int             result  = 0;
 280         lnet_msg_t      *lntmsg[2];
 281
 282         LASSERT(tx->mxc_type == MXLND_REQ_TX);
 283
 284         if (tx->mxc_status.code != MX_STATUS_SUCCESS || tx->mxc_errno != 0) {
 285                 kmx_conn_t      *conn   = tx->mxc_conn;
 286
 287                 result = -EIO;
 288                 if (tx->mxc_errno != 0) result = tx->mxc_errno;
 289                 /* FIXME should we set mx_dis? */
 290                 mxlnd_conn_disconnect(conn, 0, 1);
 291         }
 292
 293         lntmsg[0] = tx->mxc_lntmsg[0];
 294         lntmsg[1] = tx->mxc_lntmsg[1];
 295
 296         mxlnd_ctx_init(tx);
 297
 298         tx->mxc_put++;
 299         LASSERT(tx->mxc_get == tx->mxc_put);
 300
 301         spin_lock(&kmxlnd_data.kmx_tx_idle_lock);
 302         cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_tx_idle);
 303         kmxlnd_data.kmx_tx_used--;
 304         spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);
 305
 306         if (lntmsg[0] != NULL)
 307                 lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[0], result);
 308         if (lntmsg[1] != NULL)
 309                 lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[1], result);
 310         return 0;
 311 }
 312
 313
 314 void
 315 mxlnd_connparams_free(kmx_connparams_t *cp)
 316 {
 317         LASSERT(cfs_list_empty(&cp->mxr_list));
 318         MXLND_FREE(cp, sizeof(*cp));
 319         return;
 320 }
 321
 322 int
 323 mxlnd_connparams_alloc(kmx_connparams_t **cp, void *context,
 324                             mx_endpoint_addr_t epa, u64 match, u32 length,
 325                             kmx_conn_t *conn, kmx_peer_t *peer, void *data)
 326 {
 327         kmx_connparams_t *c = NULL;
 328
 329         MXLND_ALLOC(c, sizeof(*c));
 330         if (!c) return -ENOMEM;
 331
 332         CFS_INIT_LIST_HEAD(&c->mxr_list);
 333         c->mxr_context = context;
 334         c->mxr_epa = epa;
 335         c->mxr_match = match;
 336         c->mxr_nob = length;
 337         c->mxr_conn = conn;
 338         c->mxr_peer = peer;
 339         c->mxr_msg = *((kmx_msg_t *) data);
 340
 341         *cp = c;
 342         return 0;
 343 }
 344
 345 static inline void
 346 mxlnd_set_conn_status(kmx_conn_t *conn, int status)
 347 {
 348         conn->mxk_status = status;
 349         smp_mb();
 350 }
 351
 352 /**
 353  * mxlnd_conn_free_locked - free the conn
 354  * @conn - a kmx_conn pointer
 355  *
 356  * The calling function should remove the conn from the conns list first
 357  * then destroy it. Caller should have write-locked kmx_global_lock.
 358  */
 359 void
 360 mxlnd_conn_free_locked(kmx_conn_t *conn)
 361 {
 362         int             valid   = !mxlnd_endpoint_addr_null(conn->mxk_epa);
 363         kmx_peer_t      *peer   = conn->mxk_peer;
 364
 365         CDEBUG(D_NET, "freeing conn 0x%p *****\n", conn);
 366         LASSERT (cfs_list_empty (&conn->mxk_tx_credit_queue) &&
 367                  cfs_list_empty (&conn->mxk_tx_free_queue) &&
 368                  cfs_list_empty (&conn->mxk_pending));
 369         if (!cfs_list_empty(&conn->mxk_list)) {
 370                 cfs_list_del_init(&conn->mxk_list);
 371                 if (peer->mxp_conn == conn) {
 372                         peer->mxp_conn = NULL;
 373                         if (valid) {
 374                                 kmx_conn_t      *temp   = NULL;
 375
 376                                 mx_get_endpoint_addr_context(conn->mxk_epa,
 377                                                              (void **) &temp);
 378                                 if (conn == temp) {
 379                                         mx_set_endpoint_addr_context(conn->mxk_epa,
 380                                                                      (void *) NULL);
 381                                 }
 382                         }
 383                         /* unlink from global list and drop its ref */
 384                         cfs_list_del_init(&peer->mxp_list);
 385                         mxlnd_peer_decref(peer);
 386                 }
 387         }
 388         mxlnd_peer_decref(peer); /* drop conn's ref to peer */
 389         if (conn->mxk_rx_pages) {
 390                 LASSERT (conn->mxk_rxs != NULL);
 391                 mxlnd_free_pages(conn->mxk_rx_pages);
 392         }
 393         if (conn->mxk_rxs) {
 394                 int             i       = 0;
 395                 kmx_ctx_t       *rx     = NULL;
 396
 397                 for (i = 0; i < MXLND_RX_MSGS(); i++) {
 398                         rx = &conn->mxk_rxs[i];
 399                         if (rx->mxc_seg_list != NULL) {
 400                                 LASSERT(rx->mxc_nseg > 0);
 401                                 MXLND_FREE(rx->mxc_seg_list,
 402                                            rx->mxc_nseg *
 403                                            sizeof(*rx->mxc_seg_list));
 404                         }
 405                 }
 406                 MXLND_FREE(conn->mxk_rxs, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));
 407         }
 408
 409         MXLND_FREE(conn, sizeof (*conn));
 410         return;
 411 }
 412
 413
 414 int
 415 mxlnd_conn_cancel_pending_rxs(kmx_conn_t *conn)
 416 {
 417         int             found   = 0;
 418         int             count   = 0;
 419         kmx_ctx_t       *ctx    = NULL;
 420         kmx_ctx_t       *next   = NULL;
 421         mx_return_t     mxret   = MX_SUCCESS;
 422         u32             result  = 0;
 423
 424         do {
 425                 found = 0;
 426                 spin_lock(&conn->mxk_lock);
 427                 cfs_list_for_each_entry_safe(ctx, next, &conn->mxk_pending,
 428                                              mxc_list) {
 429                         cfs_list_del_init(&ctx->mxc_list);
 430                         if (ctx->mxc_type == MXLND_REQ_RX) {
 431                                 found = 1;
 432                                 mxret = mx_cancel(kmxlnd_data.kmx_endpt,
 433                                                   &ctx->mxc_mxreq,
 434                                                   &result);
 435                                 if (mxret != MX_SUCCESS) {
 436                                         CNETERR("mx_cancel() returned %s (%d)\n", mx_strerror(mxret), mxret);
 437                                 }
 438                                 if (result == 1) {
 439                                         ctx->mxc_errno = -ECONNABORTED;
 440                                         ctx->mxc_state = MXLND_CTX_CANCELED;
 441                                         spin_unlock(&conn->mxk_lock);
 442                                         spin_lock(&kmxlnd_data.kmx_conn_lock);
 443                                         /* we may be holding the global lock,
 444                                          * move to orphan list so that it can free it */
 445                                         cfs_list_add_tail(&ctx->mxc_list,
 446                                                           &kmxlnd_data.kmx_orphan_msgs);
 447                                         count++;
 448                                         spin_unlock(&kmxlnd_data.kmx_conn_lock);
 449                                         spin_lock(&conn->mxk_lock);
 450                                 }
 451                                 break;
 452                         }
 453                 }
 454                 spin_unlock(&conn->mxk_lock);
 455         } while (found);
 456
 457         return count;
 458 }
 459
 460 int
 461 mxlnd_cancel_queued_txs(kmx_conn_t *conn)
 462 {
 463         int             count   = 0;
 464         cfs_list_t      *tmp    = NULL;
 465
 466         spin_lock(&conn->mxk_lock);
 467         while (!cfs_list_empty(&conn->mxk_tx_free_queue) ||
 468                !cfs_list_empty(&conn->mxk_tx_credit_queue)) {
 469
 470                 kmx_ctx_t       *tx     = NULL;
 471
 472                 if (!cfs_list_empty(&conn->mxk_tx_free_queue)) {
 473                         tmp = &conn->mxk_tx_free_queue;
 474                 } else {
 475                         tmp = &conn->mxk_tx_credit_queue;
 476                 }
 477
 478                 tx = cfs_list_entry(tmp->next, kmx_ctx_t, mxc_list);
 479                 cfs_list_del_init(&tx->mxc_list);
 480                 spin_unlock(&conn->mxk_lock);
 481                 tx->mxc_errno = -ECONNABORTED;
 482                 tx->mxc_state = MXLND_CTX_CANCELED;
 483                 /* move to orphan list and then abort */
 484                 spin_lock(&kmxlnd_data.kmx_conn_lock);
 485                 cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_orphan_msgs);
 486                 spin_unlock(&kmxlnd_data.kmx_conn_lock);
 487                 count++;
 488                 spin_lock(&conn->mxk_lock);
 489         }
 490         spin_unlock(&conn->mxk_lock);
 491
 492         return count;
 493 }
 494
 495 void
 496 mxlnd_send_message(mx_endpoint_addr_t epa, u8 msg_type, int error, u64 cookie)
 497 {
 498         u64 match = (((u64) msg_type) << MXLND_MSG_OFFSET) |
 499                     (((u64) error) << MXLND_ERROR_OFFSET) | cookie;
 500
 501         mx_kisend(kmxlnd_data.kmx_endpt, NULL, 0, MX_PIN_PHYSICAL,
 502                   epa, match, NULL, NULL);
 503         return;
 504 }
 505
 506 /**
 507  * mxlnd_conn_disconnect - shutdown a connection
 508  * @conn - a kmx_conn pointer
 509  * @mx_dis - call mx_disconnect()
 510  * @send_bye - send peer a BYE msg
 511  *
 512  * This function sets the status to DISCONNECT, completes queued
 513  * txs with failure, calls mx_disconnect, which will complete
 514  * pending txs and matched rxs with failure.
 515  */
 516 void
 517 mxlnd_conn_disconnect(kmx_conn_t *conn, int mx_dis, int send_bye)
 518 {
 519         mx_endpoint_addr_t      epa     = conn->mxk_epa;
 520         int                     valid   = !mxlnd_endpoint_addr_null(epa);
 521         int                     count   = 0;
 522
 523         spin_lock(&conn->mxk_lock);
 524         if (conn->mxk_status == MXLND_CONN_DISCONNECT) {
 525                 spin_unlock(&conn->mxk_lock);
 526                 return;
 527         }
 528         mxlnd_set_conn_status(conn, MXLND_CONN_DISCONNECT);
 529         conn->mxk_timeout = 0;
 530         spin_unlock(&conn->mxk_lock);
 531
 532         count = mxlnd_cancel_queued_txs(conn);
 533         count += mxlnd_conn_cancel_pending_rxs(conn);
 534
 535         if (count) /* let connd call kmxlnd_abort_msgs() */
 536                 up(&kmxlnd_data.kmx_conn_sem);
 537
 538         if (send_bye && valid &&
 539             conn->mxk_peer->mxp_nid != kmxlnd_data.kmx_ni->ni_nid) {
 540                 /* send a BYE to the peer */
 541                 CDEBUG(D_NET, "%s: sending a BYE msg to %s\n", __func__,
 542                                 libcfs_nid2str(conn->mxk_peer->mxp_nid));
 543                 mxlnd_send_message(epa, MXLND_MSG_BYE, 0, 0);
 544                 /* wait to allow the peer to ack our message */
 545                 mxlnd_sleep(msecs_to_jiffies(20));
 546         }
 547
 548         if (atomic_read(&kmxlnd_data.kmx_shutdown) != 1) {
 549                 unsigned long   last_msg        = 0;
 550
 551                 /* notify LNET that we are giving up on this peer */
 552                 if (cfs_time_after(conn->mxk_last_rx, conn->mxk_last_tx))
 553                         last_msg = conn->mxk_last_rx;
 554                 else
 555                         last_msg = conn->mxk_last_tx;
 556
 557                 lnet_notify(kmxlnd_data.kmx_ni, conn->mxk_peer->mxp_nid, 0, last_msg);
 558
 559                 if (mx_dis && valid &&
 560                     (memcmp(&epa, &kmxlnd_data.kmx_epa, sizeof(epa) != 0)))
 561                         mx_disconnect(kmxlnd_data.kmx_endpt, epa);
 562         }
 563         mxlnd_conn_decref(conn); /* drop the owning peer's reference */
 564
 565         return;
 566 }
 567
 568 /**
 569  * mxlnd_conn_alloc - allocate and initialize a new conn struct
 570  * @connp - address of a kmx_conn pointer
 571  * @peer - owning kmx_peer
 572  *
 573  * Returns 0 on success and -ENOMEM on failure
 574  */
 575 int
 576 mxlnd_conn_alloc_locked(kmx_conn_t **connp, kmx_peer_t *peer)
 577 {
 578         int             i       = 0;
 579         int             ret     = 0;
 580         int             ipage   = 0;
 581         int             offset  = 0;
 582         void           *addr    = NULL;
 583         kmx_conn_t     *conn    = NULL;
 584         kmx_pages_t    *pages   = NULL;
 585         struct page    *page    = NULL;
 586         kmx_ctx_t      *rx      = NULL;
 587
 588         LASSERT(peer != NULL);
 589
 590         MXLND_ALLOC(conn, sizeof (*conn));
 591         if (conn == NULL) {
 592                 CNETERR("Cannot allocate conn\n");
 593                 return -ENOMEM;
 594         }
 595         CDEBUG(D_NET, "allocated conn 0x%p for peer 0x%p\n", conn, peer);
 596
 597         memset(conn, 0, sizeof(*conn));
 598
 599         ret = mxlnd_alloc_pages(&pages, MXLND_RX_MSG_PAGES());
 600         if (ret != 0) {
 601                 CERROR("Can't allocate rx pages\n");
 602                 MXLND_FREE(conn, sizeof(*conn));
 603                 return -ENOMEM;
 604         }
 605         conn->mxk_rx_pages = pages;
 606
 607         MXLND_ALLOC(conn->mxk_rxs, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));
 608         if (conn->mxk_rxs == NULL) {
 609                 CERROR("Can't allocate %d rx descriptors\n", MXLND_RX_MSGS());
 610                 mxlnd_free_pages(pages);
 611                 MXLND_FREE(conn, sizeof(*conn));
 612                 return -ENOMEM;
 613         }
 614
 615         memset(conn->mxk_rxs, 0, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));
 616
 617         conn->mxk_peer = peer;
 618         CFS_INIT_LIST_HEAD(&conn->mxk_list);
 619         CFS_INIT_LIST_HEAD(&conn->mxk_zombie);
 620         atomic_set(&conn->mxk_refcount, 2); /* ref for owning peer
 621                                                    and one for the caller */
 622         if (peer->mxp_nid == kmxlnd_data.kmx_ni->ni_nid) {
 623                 u64     nic_id  = 0ULL;
 624                 u32     ep_id   = 0;
 625
 626                 /* this is localhost, set the epa and status as up */
 627                 mxlnd_set_conn_status(conn, MXLND_CONN_READY);
 628                 conn->mxk_epa = kmxlnd_data.kmx_epa;
 629                 mx_set_endpoint_addr_context(conn->mxk_epa, (void *) conn);
 630                 peer->mxp_reconnect_time = 0;
 631                 mx_decompose_endpoint_addr(kmxlnd_data.kmx_epa, &nic_id, &ep_id);
 632                 peer->mxp_nic_id = nic_id;
 633                 peer->mxp_ep_id = ep_id;
 634                 conn->mxk_incarnation = kmxlnd_data.kmx_incarnation;
 635                 conn->mxk_timeout = 0;
 636         } else {
 637                 /* conn->mxk_incarnation = 0 - will be set by peer */
 638                 /* conn->mxk_sid = 0 - will be set by peer */
 639                 mxlnd_set_conn_status(conn, MXLND_CONN_INIT);
 640                 /* mxk_epa - to be set after mx_iconnect() */
 641         }
 642         spin_lock_init(&conn->mxk_lock);
 643         /* conn->mxk_timeout = 0 */
 644         /* conn->mxk_last_tx = 0 */
 645         /* conn->mxk_last_rx = 0 */
 646         CFS_INIT_LIST_HEAD(&conn->mxk_rx_idle);
 647
 648         conn->mxk_credits = *kmxlnd_tunables.kmx_peercredits;
 649         /* mxk_outstanding = 0 */
 650
 651         CFS_INIT_LIST_HEAD(&conn->mxk_tx_credit_queue);
 652         CFS_INIT_LIST_HEAD(&conn->mxk_tx_free_queue);
 653         /* conn->mxk_ntx_msgs = 0 */
 654         /* conn->mxk_ntx_data = 0 */
 655         /* conn->mxk_ntx_posted = 0 */
 656         /* conn->mxk_data_posted = 0 */
 657         CFS_INIT_LIST_HEAD(&conn->mxk_pending);
 658
 659         for (i = 0; i < MXLND_RX_MSGS(); i++) {
 660
 661                 rx = &conn->mxk_rxs[i];
 662                 rx->mxc_type = MXLND_REQ_RX;
 663                 CFS_INIT_LIST_HEAD(&rx->mxc_list);
 664
 665                 /* map mxc_msg to page */
 666                 page = pages->mxg_pages[ipage];
 667                 addr = page_address(page);
 668                 LASSERT(addr != NULL);
 669                 rx->mxc_msg = (kmx_msg_t *)(addr + offset);
 670                 rx->mxc_seg.segment_ptr = MX_PA_TO_U64(virt_to_phys(rx->mxc_msg));
 671
 672                 rx->mxc_conn = conn;
 673                 rx->mxc_peer = peer;
 674                 rx->mxc_nid = peer->mxp_nid;
 675
 676                 mxlnd_ctx_init(rx);
 677
 678                 offset += MXLND_MSG_SIZE;
 679                 LASSERT (offset <= PAGE_SIZE);
 680
 681                 if (offset == PAGE_SIZE) {
 682                         offset = 0;
 683                         ipage++;
 684                         LASSERT (ipage <= MXLND_TX_MSG_PAGES());
 685                 }
 686
 687                 cfs_list_add_tail(&rx->mxc_list, &conn->mxk_rx_idle);
 688         }
 689
 690         *connp = conn;
 691
 692         mxlnd_peer_addref(peer);        /* add a ref for this conn */
 693
 694         /* add to front of peer's conns list */
 695         cfs_list_add(&conn->mxk_list, &peer->mxp_conns);
 696         peer->mxp_conn = conn;
 697         return 0;
 698 }
 699
 700 int
 701 mxlnd_conn_alloc(kmx_conn_t **connp, kmx_peer_t *peer)
 702 {
 703         int             ret     = 0;
 704         rwlock_t   *g_lock  = &kmxlnd_data.kmx_global_lock;
 705
 706         write_lock(g_lock);
 707         ret = mxlnd_conn_alloc_locked(connp, peer);
 708         write_unlock(g_lock);
 709         return ret;
 710 }
 711
 712 int
 713 mxlnd_q_pending_ctx(kmx_ctx_t *ctx)
 714 {
 715         int             ret     = 0;
 716         kmx_conn_t      *conn   = ctx->mxc_conn;
 717
 718         ctx->mxc_state = MXLND_CTX_PENDING;
 719         if (conn != NULL) {
 720                 spin_lock(&conn->mxk_lock);
 721                 if (conn->mxk_status >= MXLND_CONN_INIT) {
 722                         cfs_list_add_tail(&ctx->mxc_list, &conn->mxk_pending);
 723                         if (conn->mxk_timeout == 0 || ctx->mxc_deadline < conn->mxk_timeout) {
 724                                 conn->mxk_timeout = ctx->mxc_deadline;
 725                         }
 726                 } else {
 727                         ctx->mxc_state = MXLND_CTX_COMPLETED;
 728                         ret = -1;
 729                 }
 730                 spin_unlock(&conn->mxk_lock);
 731         }
 732         return ret;
 733 }
 734
 735 int
 736 mxlnd_deq_pending_ctx(kmx_ctx_t *ctx)
 737 {
 738         LASSERT(ctx->mxc_state == MXLND_CTX_PENDING ||
 739                 ctx->mxc_state == MXLND_CTX_COMPLETED);
 740         if (ctx->mxc_state != MXLND_CTX_PENDING &&
 741             ctx->mxc_state != MXLND_CTX_COMPLETED) {
 742                 CNETERR("deq ctx->mxc_state = %s\n",
 743                         mxlnd_ctxstate_to_str(ctx->mxc_state));
 744         }
 745         ctx->mxc_state = MXLND_CTX_COMPLETED;
 746         if (!cfs_list_empty(&ctx->mxc_list)) {
 747                 kmx_conn_t      *conn = ctx->mxc_conn;
 748                 kmx_ctx_t       *next = NULL;
 749
 750                 LASSERT(conn != NULL);
 751                 spin_lock(&conn->mxk_lock);
 752                 cfs_list_del_init(&ctx->mxc_list);
 753                 conn->mxk_timeout = 0;
 754                 if (!cfs_list_empty(&conn->mxk_pending)) {
 755                         next = cfs_list_entry(conn->mxk_pending.next,
 756                                               kmx_ctx_t, mxc_list);
 757                         conn->mxk_timeout = next->mxc_deadline;
 758                 }
 759                 spin_unlock(&conn->mxk_lock);
 760         }
 761         return 0;
 762 }
 763
 764 /**
 765  * mxlnd_peer_free - free the peer
 766  * @peer - a kmx_peer pointer
 767  *
 768  * The calling function should decrement the rxs, drain the tx queues and
 769  * remove the peer from the peers list first then destroy it.
 770  */
 771 void
 772 mxlnd_peer_free(kmx_peer_t *peer)
 773 {
 774         CDEBUG(D_NET, "freeing peer 0x%p %s\n", peer, libcfs_nid2str(peer->mxp_nid));
 775
 776         LASSERT (atomic_read(&peer->mxp_refcount) == 0);
 777
 778         if (!cfs_list_empty(&peer->mxp_list)) {
 779                 /* assume we are locked */
 780                 cfs_list_del_init(&peer->mxp_list);
 781         }
 782
 783         MXLND_FREE(peer, sizeof (*peer));
 784         atomic_dec(&kmxlnd_data.kmx_npeers);
 785         return;
 786 }
 787
 788 static int
 789 mxlnd_lookup_mac(u32 ip, u64 *tmp_id)
 790 {
 791         int                     ret     = -EHOSTUNREACH;
 792         unsigned char           *haddr  = NULL;
 793         struct net_device       *dev    = NULL;
 794         struct neighbour        *n      = NULL;
 795         __be32                  dst_ip  = htonl(ip);
 796
 797         dev = dev_get_by_name(*kmxlnd_tunables.kmx_default_ipif);
 798         if (dev == NULL)
 799                 return -ENODEV;
 800
 801         haddr = (unsigned char *) tmp_id + 2; /* MAC is only 6 bytes */
 802
 803         n = neigh_lookup(&arp_tbl, &dst_ip, dev);
 804         if (n) {
 805                 n->used = jiffies;
 806                 if (n->nud_state & NUD_VALID) {
 807                         memcpy(haddr, n->ha, dev->addr_len);
 808                         neigh_release(n);
 809                         ret = 0;
 810                 }
 811         }
 812
 813         dev_put(dev);
 814
 815         return ret;
 816 }
 817
 818
 819 /* We only want the MAC address of the peer's Myricom NIC. We
 820  * require that each node has the IPoMX interface (myriN) up.
 821  * We will not pass any traffic over IPoMX, but it allows us
 822  * to get the MAC address. */
 823 static int
 824 mxlnd_ip2nic_id(u32 ip, u64 *nic_id, int tries)
 825 {
 826         int                     ret     = 0;
 827         int                     try     = 1;
 828         int                     fatal   = 0;
 829         u64                     tmp_id  = 0ULL;
 830         cfs_socket_t            *sock   = NULL;
 831
 832         do {
 833                 CDEBUG(D_NET, "try %d of %d tries\n", try, tries);
 834                 ret = mxlnd_lookup_mac(ip, &tmp_id);
 835                 if (ret == 0) {
 836                         break;
 837                 } else {
 838                         /* not found, try to connect (force an arp) */
 839                         ret = libcfs_sock_connect(&sock, &fatal, 0, 0, ip, 987);
 840                         if (ret == -ECONNREFUSED) {
 841                                 /* peer is there, get the MAC address */
 842                                 mxlnd_lookup_mac(ip, &tmp_id);
 843                                 if (tmp_id != 0ULL)
 844                                         ret = 0;
 845                                 break;
 846                         } else if (ret == -EHOSTUNREACH && try < tries) {
 847                                 /* add a little backoff */
 848                                 CDEBUG(D_NET, "sleeping for %d jiffies\n",
 849                                        HZ/4);
 850                                 mxlnd_sleep(HZ/4);
 851                         }
 852                 }
 853         } while (try++ < tries);
 854         CDEBUG(D_NET, "done trying. ret = %d\n", ret);
 855
 856         if (tmp_id == 0ULL)
 857                 ret = -EHOSTUNREACH;
 858 #if __BYTE_ORDER == __LITTLE_ENDIAN
 859         *nic_id = ___arch__swab64(tmp_id);
 860 #else
 861         *nic_id = tmp_id;
 862 #endif
 863         return ret;
 864 }
 865
 866 /**
 867  * mxlnd_peer_alloc - allocate and initialize a new peer struct
 868  * @peerp - address of a kmx_peer pointer
 869  * @nid - LNET node id
 870  *
 871  * Returns 0 on success and -ENOMEM on failure
 872  */
 873 int
 874 mxlnd_peer_alloc(kmx_peer_t **peerp, lnet_nid_t nid, u32 board, u32 ep_id, u64 nic_id)
 875 {
 876         int             ret     = 0;
 877         u32             ip      = LNET_NIDADDR(nid);
 878         kmx_peer_t     *peer    = NULL;
 879
 880         LASSERT (nid != LNET_NID_ANY && nid != 0LL);
 881
 882         MXLND_ALLOC(peer, sizeof (*peer));
 883         if (peer == NULL) {
 884                 CNETERR("Cannot allocate peer for NID 0x%llx\n",
 885                        nid);
 886                 return -ENOMEM;
 887         }
 888         CDEBUG(D_NET, "allocated peer 0x%p for NID 0x%llx\n", peer, nid);
 889
 890         memset(peer, 0, sizeof(*peer));
 891
 892         CFS_INIT_LIST_HEAD(&peer->mxp_list);
 893         peer->mxp_nid = nid;
 894         /* peer->mxp_ni unused - may be used for multi-rail */
 895         atomic_set(&peer->mxp_refcount, 1);     /* ref for kmx_peers list */
 896
 897         peer->mxp_board = board;
 898         peer->mxp_ep_id = ep_id;
 899         peer->mxp_nic_id = nic_id;
 900
 901         CFS_INIT_LIST_HEAD(&peer->mxp_conns);
 902         ret = mxlnd_conn_alloc(&peer->mxp_conn, peer); /* adds 2nd conn ref here... */
 903         if (ret != 0) {
 904                 mxlnd_peer_decref(peer);
 905                 return ret;
 906         }
 907         CFS_INIT_LIST_HEAD(&peer->mxp_tx_queue);
 908
 909         if (peer->mxp_nic_id != 0ULL)
 910                 nic_id = peer->mxp_nic_id;
 911
 912         if (nic_id == 0ULL) {
 913                 ret = mxlnd_ip2nic_id(ip, &nic_id, 1);
 914                 if (ret == 0) {
 915                         peer->mxp_nic_id = nic_id;
 916                         mx_nic_id_to_board_number(nic_id, &peer->mxp_board);
 917                 }
 918         }
 919
 920         peer->mxp_nic_id = nic_id; /* may be 0ULL if ip2nic_id() failed */
 921
 922         /* peer->mxp_reconnect_time = 0 */
 923         /* peer->mxp_incompatible = 0 */
 924
 925         *peerp = peer;
 926         return 0;
 927 }
 928
 929 static inline kmx_peer_t *
 930 mxlnd_find_peer_by_nid_locked(lnet_nid_t nid)
 931 {
 932         int             found   = 0;
 933         int             hash    = 0;
 934         kmx_peer_t      *peer   = NULL;
 935
 936         hash = mxlnd_nid_to_hash(nid);
 937
 938         cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[hash], mxp_list) {
 939                 if (peer->mxp_nid == nid) {
 940                         found = 1;
 941                         mxlnd_peer_addref(peer);
 942                         break;
 943                 }
 944         }
 945         return (found ? peer : NULL);
 946 }
 947
 948 static kmx_peer_t *
 949 mxlnd_find_peer_by_nid(lnet_nid_t nid, int create)
 950 {
 951         int             ret     = 0;
 952         int             hash    = 0;
 953         kmx_peer_t      *peer   = NULL;
 954         kmx_peer_t      *old    = NULL;
 955         rwlock_t    *g_lock = &kmxlnd_data.kmx_global_lock;
 956
 957         read_lock(g_lock);
 958         peer = mxlnd_find_peer_by_nid_locked(nid); /* adds peer ref */
 959
 960         if ((peer && peer->mxp_conn) || /* found peer with conn or */
 961             (!peer && !create)) {       /* did not find peer and do not create one */
 962                 read_unlock(g_lock);
 963                 return peer;
 964         }
 965
 966         read_unlock(g_lock);
 967
 968         /* if peer but _not_ conn */
 969         if (peer && !peer->mxp_conn) {
 970                 if (create) {
 971                         write_lock(g_lock);
 972                         if (!peer->mxp_conn) { /* check again */
 973                                 /* create the conn */
 974                                 ret = mxlnd_conn_alloc_locked(&peer->mxp_conn, peer);
 975                                 if (ret != 0) {
 976                                         /* we tried, return the peer only.
 977                                          * the caller needs to see if the conn exists */
 978                                         CNETERR("%s: %s could not alloc conn\n",
 979                                         __func__, libcfs_nid2str(peer->mxp_nid));
 980                                 } else {
 981                                         /* drop extra conn ref */
 982                                         mxlnd_conn_decref(peer->mxp_conn);
 983                                 }
 984                         }
 985                         write_unlock(g_lock);
 986                 }
 987                 return peer;
 988         }
 989
 990         /* peer not found and we need to create one */
 991         hash = mxlnd_nid_to_hash(nid);
 992
 993         /* create peer (and conn) */
 994         /* adds conn ref for peer and one for this function */
 995         ret = mxlnd_peer_alloc(&peer, nid, *kmxlnd_tunables.kmx_board,
 996                                *kmxlnd_tunables.kmx_ep_id, 0ULL);
 997         if (ret != 0) /* no memory, peer is NULL */
 998                 return NULL;
 999
1000         write_lock(g_lock);
1001
1002         /* look again */
1003         old = mxlnd_find_peer_by_nid_locked(nid);
1004         if (old) {
1005                 /* someone already created one */
1006                 mxlnd_conn_decref(peer->mxp_conn); /* drop ref taken above.. */
1007                 mxlnd_conn_decref(peer->mxp_conn); /* drop peer's ref */
1008                 mxlnd_peer_decref(peer);
1009                 peer = old;
1010         } else {
1011                 /* no other peer, use this one */
1012                 cfs_list_add_tail(&peer->mxp_list,
1013                                   &kmxlnd_data.kmx_peers[hash]);
1014                 atomic_inc(&kmxlnd_data.kmx_npeers);
1015                 mxlnd_peer_addref(peer);
1016                 mxlnd_conn_decref(peer->mxp_conn); /* drop ref from peer_alloc */
1017         }
1018
1019         write_unlock(g_lock);
1020
1021         return peer;
1022 }
1023
1024 static inline int
1025 mxlnd_tx_requires_credit(kmx_ctx_t *tx)
1026 {
1027         return (tx->mxc_msg_type == MXLND_MSG_EAGER ||
1028                 tx->mxc_msg_type == MXLND_MSG_GET_REQ ||
1029                 tx->mxc_msg_type == MXLND_MSG_PUT_REQ ||
1030                 tx->mxc_msg_type == MXLND_MSG_NOOP);
1031 }
1032
1033 /**
1034  * mxlnd_init_msg - set type and number of bytes
1035  * @msg - msg pointer
1036  * @type - of message
1037  * @body_nob - bytes in msg body
1038  */
1039 static inline void
1040 mxlnd_init_msg(kmx_msg_t *msg, u8 type, int body_nob)
1041 {
1042         msg->mxm_type = type;
1043         msg->mxm_nob  = offsetof(kmx_msg_t, mxm_u) + body_nob;
1044 }
1045
1046 static inline void
1047 mxlnd_init_tx_msg (kmx_ctx_t *tx, u8 type, int body_nob, lnet_nid_t nid)
1048 {
1049         int             nob     = offsetof (kmx_msg_t, mxm_u) + body_nob;
1050         kmx_msg_t       *msg    = NULL;
1051
1052         LASSERT (tx != NULL);
1053         LASSERT (nob <= MXLND_MSG_SIZE);
1054
1055         tx->mxc_nid = nid;
1056         /* tx->mxc_peer should have already been set if we know it */
1057         tx->mxc_msg_type = type;
1058         tx->mxc_nseg = 1;
1059         /* tx->mxc_seg.segment_ptr is already pointing to mxc_page */
1060         tx->mxc_seg.segment_length = nob;
1061         tx->mxc_pin_type = MX_PIN_PHYSICAL;
1062
1063         msg = tx->mxc_msg;
1064         msg->mxm_type = type;
1065         msg->mxm_nob  = nob;
1066
1067         return;
1068 }
1069
1070 static inline __u32
1071 mxlnd_cksum (void *ptr, int nob)
1072 {
1073         char  *c  = ptr;
1074         __u32  sum = 0;
1075
1076         while (nob-- > 0)
1077                 sum = ((sum << 1) | (sum >> 31)) + *c++;
1078
1079         /* ensure I don't return 0 (== no checksum) */
1080         return (sum == 0) ? 1 : sum;
1081 }
1082
1083 /**
1084  * mxlnd_pack_msg_locked - complete msg info
1085  * @tx - msg to send
1086  */
1087 static inline void
1088 mxlnd_pack_msg_locked(kmx_ctx_t *tx)
1089 {
1090         kmx_msg_t       *msg    = tx->mxc_msg;
1091
1092         /* type and nob should already be set in init_msg() */
1093         msg->mxm_magic    = MXLND_MSG_MAGIC;
1094         msg->mxm_version  = MXLND_MSG_VERSION;
1095         /*   mxm_type */
1096         /* don't use mxlnd_tx_requires_credit() since we want PUT_ACK to
1097          * return credits as well */
1098         if (tx->mxc_msg_type != MXLND_MSG_CONN_REQ &&
1099             tx->mxc_msg_type != MXLND_MSG_CONN_ACK) {
1100                 msg->mxm_credits  = tx->mxc_conn->mxk_outstanding;
1101                 tx->mxc_conn->mxk_outstanding = 0;
1102         } else {
1103                 msg->mxm_credits  = 0;
1104         }
1105         /*   mxm_nob */
1106         msg->mxm_cksum    = 0;
1107         msg->mxm_srcnid   = kmxlnd_data.kmx_ni->ni_nid;
1108         msg->mxm_srcstamp = kmxlnd_data.kmx_incarnation;
1109         msg->mxm_dstnid   = tx->mxc_nid;
1110         /* if it is a new peer, the dststamp will be 0 */
1111         msg->mxm_dststamp = tx->mxc_conn->mxk_incarnation;
1112
1113         if (*kmxlnd_tunables.kmx_cksum) {
1114                 msg->mxm_cksum = mxlnd_cksum(msg, msg->mxm_nob);
1115         }
1116 }
1117
1118 int
1119 mxlnd_unpack_msg(kmx_msg_t *msg, int nob)
1120 {
1121         const int hdr_size      = offsetof(kmx_msg_t, mxm_u);
1122         __u32     msg_cksum     = 0;
1123         int       flip          = 0;
1124         int       msg_nob       = 0;
1125
1126         /* 6 bytes are enough to have received magic + version */
1127         if (nob < 6) {
1128                 CNETERR("not enough bytes for magic + hdr: %d\n", nob);
1129                 return -EPROTO;
1130         }
1131
1132         if (msg->mxm_magic == MXLND_MSG_MAGIC) {
1133                 flip = 0;
1134         } else if (msg->mxm_magic == __swab32(MXLND_MSG_MAGIC)) {
1135                 flip = 1;
1136         } else {
1137                 CNETERR("Bad magic: %08x\n", msg->mxm_magic);
1138                 return -EPROTO;
1139         }
1140
1141         if (msg->mxm_version !=
1142             (flip ? __swab16(MXLND_MSG_VERSION) : MXLND_MSG_VERSION)) {
1143                 CNETERR("Bad version: %d\n", msg->mxm_version);
1144                 return -EPROTO;
1145         }
1146
1147         if (nob < hdr_size) {
1148                 CNETERR("not enough for a header: %d\n", nob);
1149                 return -EPROTO;
1150         }
1151
1152         msg_nob = flip ? __swab32(msg->mxm_nob) : msg->mxm_nob;
1153         if (msg_nob > nob) {
1154                 CNETERR("Short message: got %d, wanted %d\n", nob, msg_nob);
1155                 return -EPROTO;
1156         }
1157
1158         /* checksum must be computed with mxm_cksum zero and BEFORE anything
1159          * gets flipped */
1160         msg_cksum = flip ? __swab32(msg->mxm_cksum) : msg->mxm_cksum;
1161         msg->mxm_cksum = 0;
1162         if (msg_cksum != 0 && msg_cksum != mxlnd_cksum(msg, msg_nob)) {
1163                 CNETERR("Bad checksum\n");
1164                 return -EPROTO;
1165         }
1166         msg->mxm_cksum = msg_cksum;
1167
1168         if (flip) {
1169                 /* leave magic unflipped as a clue to peer endianness */
1170                 __swab16s(&msg->mxm_version);
1171                 CLASSERT (sizeof(msg->mxm_type) == 1);
1172                 CLASSERT (sizeof(msg->mxm_credits) == 1);
1173                 msg->mxm_nob = msg_nob;
1174                 __swab64s(&msg->mxm_srcnid);
1175                 __swab64s(&msg->mxm_srcstamp);
1176                 __swab64s(&msg->mxm_dstnid);
1177                 __swab64s(&msg->mxm_dststamp);
1178         }
1179
1180         if (msg->mxm_srcnid == LNET_NID_ANY) {
1181                 CNETERR("Bad src nid: %s\n", libcfs_nid2str(msg->mxm_srcnid));
1182                 return -EPROTO;
1183         }
1184
1185         switch (msg->mxm_type) {
1186         default:
1187                 CNETERR("Unknown message type %x\n", msg->mxm_type);
1188                 return -EPROTO;
1189
1190         case MXLND_MSG_NOOP:
1191                 break;
1192
1193         case MXLND_MSG_EAGER:
1194                 if (msg_nob < offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[0])) {
1195                         CNETERR("Short EAGER: %d(%d)\n", msg_nob,
1196                                (int)offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[0]));
1197                         return -EPROTO;
1198                 }
1199                 break;
1200
1201         case MXLND_MSG_PUT_REQ:
1202                 if (msg_nob < hdr_size + sizeof(msg->mxm_u.put_req)) {
1203                         CNETERR("Short PUT_REQ: %d(%d)\n", msg_nob,
1204                                (int)(hdr_size + sizeof(msg->mxm_u.put_req)));
1205                         return -EPROTO;
1206                 }
1207                 if (flip)
1208                         __swab64s(&msg->mxm_u.put_req.mxprm_cookie);
1209                 break;
1210
1211         case MXLND_MSG_PUT_ACK:
1212                 if (msg_nob < hdr_size + sizeof(msg->mxm_u.put_ack)) {
1213                         CNETERR("Short PUT_ACK: %d(%d)\n", msg_nob,
1214                                (int)(hdr_size + sizeof(msg->mxm_u.put_ack)));
1215                         return -EPROTO;
1216                 }
1217                 if (flip) {
1218                         __swab64s(&msg->mxm_u.put_ack.mxpam_src_cookie);
1219                         __swab64s(&msg->mxm_u.put_ack.mxpam_dst_cookie);
1220                 }
1221                 break;
1222
1223         case MXLND_MSG_GET_REQ:
1224                 if (msg_nob < hdr_size + sizeof(msg->mxm_u.get_req)) {
1225                         CNETERR("Short GET_REQ: %d(%d)\n", msg_nob,
1226                                (int)(hdr_size + sizeof(msg->mxm_u.get_req)));
1227                         return -EPROTO;
1228                 }
1229                 if (flip) {
1230                         __swab64s(&msg->mxm_u.get_req.mxgrm_cookie);
1231                 }
1232                 break;
1233
1234         case MXLND_MSG_CONN_REQ:
1235         case MXLND_MSG_CONN_ACK:
1236                 if (msg_nob < hdr_size + sizeof(msg->mxm_u.conn_req)) {
1237                         CNETERR("Short connreq/ack: %d(%d)\n", msg_nob,
1238                                (int)(hdr_size + sizeof(msg->mxm_u.conn_req)));
1239                         return -EPROTO;
1240                 }
1241                 if (flip) {
1242                         __swab32s(&msg->mxm_u.conn_req.mxcrm_queue_depth);
1243                         __swab32s(&msg->mxm_u.conn_req.mxcrm_eager_size);
1244                 }
1245                 break;
1246         }
1247         return 0;
1248 }
1249
1250
1251 /**
1252  * mxlnd_recv_msg
1253  * @lntmsg - the LNET msg that this is continuing. If EAGER, then NULL.
1254  * @rx
1255  * @msg_type
1256  * @cookie
1257  * @length - length of incoming message
1258  * @pending - add to kmx_pending (0 is NO and 1 is YES)
1259  *
1260  * The caller gets the rx and sets nid, peer and conn if known.
1261  *
1262  * Returns 0 on success and -1 on failure
1263  */
1264 int
1265 mxlnd_recv_msg(lnet_msg_t *lntmsg, kmx_ctx_t *rx, u8 msg_type, u64 cookie, u32 length)
1266 {
1267         int             ret     = 0;
1268         mx_return_t     mxret   = MX_SUCCESS;
1269         uint64_t        mask    = ~(MXLND_ERROR_MASK);
1270
1271         rx->mxc_msg_type = msg_type;
1272         rx->mxc_lntmsg[0] = lntmsg; /* may be NULL if EAGER */
1273         rx->mxc_cookie = cookie;
1274         /* rx->mxc_match may already be set */
1275         /* rx->mxc_seg.segment_ptr is already set */
1276         rx->mxc_seg.segment_length = length;
1277         ret = mxlnd_q_pending_ctx(rx);
1278         if (ret == -1) {
1279                 /* the caller is responsible for calling conn_decref() if needed */
1280                 return -1;
1281         }
1282         mxret = mx_kirecv(kmxlnd_data.kmx_endpt, &rx->mxc_seg, 1, MX_PIN_PHYSICAL,
1283                           cookie, mask, (void *) rx, &rx->mxc_mxreq);
1284         if (mxret != MX_SUCCESS) {
1285                 mxlnd_deq_pending_ctx(rx);
1286                 CNETERR("mx_kirecv() failed with %s (%d)\n",
1287                         mx_strerror(mxret), (int) mxret);
1288                 return -1;
1289         }
1290         return 0;
1291 }
1292
1293
1294 /**
1295  * mxlnd_unexpected_recv - this is the callback function that will handle
1296  *                         unexpected receives
1297  * @context - NULL, ignore
1298  * @source - the peer's mx_endpoint_addr_t
1299  * @match_value - the msg's bits, should be MXLND_MSG_EAGER
1300  * @length - length of incoming message
1301  * @data_if_available - used for CONN_[REQ|ACK]
1302  *
1303  * If it is an eager-sized msg, we will call recv_msg() with the actual
1304  * length. If it is a large message, we will call recv_msg() with a
1305  * length of 0 bytes to drop it because we should never have a large,
1306  * unexpected message.
1307  *
1308  * NOTE - The MX library blocks until this function completes. Make it as fast as
1309  * possible. DO NOT allocate memory which can block!
1310  *
1311  * If we cannot get a rx or the conn is closed, drop the message on the floor
1312  * (i.e. recv 0 bytes and ignore).
1313  */
1314 mx_unexp_handler_action_t
1315 mxlnd_unexpected_recv(void *context, mx_endpoint_addr_t source,
1316                  uint64_t match_value, uint32_t length, void *data_if_available)
1317 {
1318         int             ret             = 0;
1319         kmx_ctx_t       *rx             = NULL;
1320         mx_ksegment_t   seg;
1321         u8              msg_type        = 0;
1322         u8              error           = 0;
1323         u64             cookie          = 0ULL;
1324         kmx_conn_t      *conn           = NULL;
1325         kmx_peer_t      *peer           = NULL;
1326         u64             nic_id          = 0ULL;
1327         u32             ep_id           = 0;
1328         u32             sid             = 0;
1329
1330         /* TODO this will change to the net struct */
1331         if (context != NULL) {
1332                 CNETERR("non-NULL context\n");
1333         }
1334
1335 #if MXLND_DEBUG
1336         CDEBUG(D_NET, "bits=0x%llx length=%d\n", match_value, length);
1337 #endif
1338
1339         mx_decompose_endpoint_addr2(source, &nic_id, &ep_id, &sid);
1340         mxlnd_parse_match(match_value, &msg_type, &error, &cookie);
1341         read_lock(&kmxlnd_data.kmx_global_lock);
1342         mx_get_endpoint_addr_context(source, (void **) &conn);
1343         if (conn) {
1344                 mxlnd_conn_addref(conn); /* add ref for this function */
1345                 peer = conn->mxk_peer;
1346         }
1347         read_unlock(&kmxlnd_data.kmx_global_lock);
1348
1349         if (msg_type == MXLND_MSG_BYE) {
1350                 if (conn) {
1351                         CDEBUG(D_NET, "peer %s sent BYE msg\n",
1352                                         libcfs_nid2str(peer->mxp_nid));
1353                         mxlnd_conn_disconnect(conn, 1, 0);
1354                         mxlnd_conn_decref(conn); /* drop ref taken above */
1355                 }
1356                 return MX_RECV_FINISHED;
1357         }
1358
1359         if (msg_type == MXLND_MSG_CONN_REQ) {
1360                 kmx_connparams_t       *cp      = NULL;
1361                 const int       expected        = offsetof(kmx_msg_t, mxm_u) +
1362                                                   sizeof(kmx_connreq_msg_t);
1363
1364                 if (conn) mxlnd_conn_decref(conn); /* drop ref taken above */
1365                 if (unlikely(length != expected || !data_if_available)) {
1366                         CNETERR("received invalid CONN_REQ from %llx "
1367                                 "length=%d (expected %d)\n", nic_id, length, expected);
1368                         mxlnd_send_message(source, MXLND_MSG_CONN_ACK, EPROTO, 0);
1369                         return MX_RECV_FINISHED;
1370                 }
1371
1372                 ret = mxlnd_connparams_alloc(&cp, context, source, match_value, length,
1373                                          conn, peer, data_if_available);
1374                 if (unlikely(ret != 0)) {
1375                         CNETERR("unable to alloc CONN_REQ from %llx:%d\n",
1376                                 nic_id, ep_id);
1377                         mxlnd_send_message(source, MXLND_MSG_CONN_ACK, ENOMEM, 0);
1378                         return MX_RECV_FINISHED;
1379                 }
1380                 spin_lock(&kmxlnd_data.kmx_conn_lock);
1381                 cfs_list_add_tail(&cp->mxr_list, &kmxlnd_data.kmx_conn_reqs);
1382                 spin_unlock(&kmxlnd_data.kmx_conn_lock);
1383                 up(&kmxlnd_data.kmx_conn_sem);
1384                 return MX_RECV_FINISHED;
1385         }
1386         if (msg_type == MXLND_MSG_CONN_ACK) {
1387                 kmx_connparams_t  *cp           = NULL;
1388                 const int       expected        = offsetof(kmx_msg_t, mxm_u) +
1389                                                   sizeof(kmx_connreq_msg_t);
1390
1391                 LASSERT(conn);
1392                 if (unlikely(error != 0)) {
1393                         CNETERR("received CONN_ACK from %s with error -%d\n",
1394                                libcfs_nid2str(peer->mxp_nid), (int) error);
1395                         mxlnd_conn_disconnect(conn, 1, 0);
1396                 } else if (unlikely(length != expected || !data_if_available)) {
1397                         CNETERR("received %s CONN_ACK from %s "
1398                                "length=%d (expected %d)\n",
1399                                data_if_available ? "short" : "missing",
1400                                libcfs_nid2str(peer->mxp_nid), length, expected);
1401                         mxlnd_conn_disconnect(conn, 1, 1);
1402                 } else {
1403                         /* peer is ready for messages */
1404                         ret = mxlnd_connparams_alloc(&cp, context, source, match_value, length,
1405                                          conn, peer, data_if_available);
1406                         if (unlikely(ret != 0)) {
1407                                 CNETERR("unable to alloc kmx_connparams_t"
1408                                                " from %llx:%d\n", nic_id, ep_id);
1409                                 mxlnd_conn_disconnect(conn, 1, 1);
1410                         } else {
1411                                 spin_lock(&kmxlnd_data.kmx_conn_lock);
1412                                 cfs_list_add_tail(&cp->mxr_list,
1413                                                   &kmxlnd_data.kmx_conn_reqs);
1414                                 spin_unlock(&kmxlnd_data.kmx_conn_lock);
1415                                 up(&kmxlnd_data.kmx_conn_sem);
1416                         }
1417                 }
1418                 mxlnd_conn_decref(conn); /* drop ref taken above */
1419
1420                 return MX_RECV_FINISHED;
1421         }
1422
1423         /* Handle unexpected messages (PUT_REQ and GET_REQ) */
1424
1425         LASSERT(peer != NULL && conn != NULL);
1426
1427         rx = mxlnd_get_idle_rx(conn);
1428         if (rx != NULL) {
1429                 if (length <= MXLND_MSG_SIZE) {
1430                         ret = mxlnd_recv_msg(NULL, rx, msg_type, match_value, length);
1431                 } else {
1432                         CNETERR("unexpected large receive with "
1433                                 "match_value=0x%llx length=%d\n",
1434                                 match_value, length);
1435                         ret = mxlnd_recv_msg(NULL, rx, msg_type, match_value, 0);
1436                 }
1437
1438                 if (ret == 0) {
1439                         /* hold conn ref until rx completes */
1440                         rx->mxc_conn = conn;
1441                         rx->mxc_peer = peer;
1442                         rx->mxc_nid = peer->mxp_nid;
1443                 } else {
1444                         CNETERR("could not post receive\n");
1445                         mxlnd_put_idle_rx(rx);
1446                 }
1447         }
1448
1449         /* Encountered error, drop incoming message on the floor */
1450         /* We could use MX_RECV_FINISHED but posting the receive of 0 bytes
1451          * uses the standard code path and acks the sender normally */
1452
1453         if (rx == NULL || ret != 0) {
1454                 mxlnd_conn_decref(conn); /* drop ref taken above */
1455                 if (rx == NULL) {
1456                         CNETERR("no idle rxs available - dropping rx"
1457                                 " 0x%llx from %s\n", match_value,
1458                                 libcfs_nid2str(peer->mxp_nid));
1459                 } else {
1460                         /* ret != 0 */
1461                         CNETERR("disconnected peer - dropping rx\n");
1462                 }
1463                 seg.segment_ptr = 0ULL;
1464                 seg.segment_length = 0;
1465                 mx_kirecv(kmxlnd_data.kmx_endpt, &seg, 1, MX_PIN_PHYSICAL,
1466                           match_value, ~0ULL, NULL, NULL);
1467         }
1468
1469         return MX_RECV_CONTINUE;
1470 }
1471
1472
1473 int
1474 mxlnd_get_peer_info(int index, lnet_nid_t *nidp, int *count)
1475 {
1476         int              i      = 0;
1477         int              ret    = -ENOENT;
1478         kmx_peer_t      *peer   = NULL;
1479
1480         read_lock(&kmxlnd_data.kmx_global_lock);
1481         for (i = 0; i < MXLND_HASH_SIZE; i++) {
1482                 cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],
1483                                         mxp_list) {
1484                         if (index-- == 0) {
1485                                 *nidp = peer->mxp_nid;
1486                                 *count = atomic_read(&peer->mxp_refcount);
1487                                 ret = 0;
1488                                 break;
1489                         }
1490                 }
1491         }
1492         read_unlock(&kmxlnd_data.kmx_global_lock);
1493
1494         return ret;
1495 }
1496
1497 void
1498 mxlnd_del_peer_locked(kmx_peer_t *peer)
1499 {
1500         if (peer->mxp_conn) {
1501                 mxlnd_conn_disconnect(peer->mxp_conn, 1, 1);
1502         } else {
1503                 cfs_list_del_init(&peer->mxp_list); /* remove from the global list */
1504                 mxlnd_peer_decref(peer); /* drop global list ref */
1505         }
1506         return;
1507 }
1508
1509 int
1510 mxlnd_del_peer(lnet_nid_t nid)
1511 {
1512         int             i       = 0;
1513         int             ret     = 0;
1514         kmx_peer_t      *peer   = NULL;
1515         kmx_peer_t      *next   = NULL;
1516
1517         if (nid != LNET_NID_ANY) {
1518                 peer = mxlnd_find_peer_by_nid(nid, 0); /* adds peer ref */
1519         }
1520         write_lock(&kmxlnd_data.kmx_global_lock);
1521         if (nid != LNET_NID_ANY) {
1522                 if (peer == NULL) {
1523                         ret = -ENOENT;
1524                 } else {
1525                         mxlnd_peer_decref(peer); /* and drops it */
1526                         mxlnd_del_peer_locked(peer);
1527                 }
1528         } else { /* LNET_NID_ANY */
1529                 for (i = 0; i < MXLND_HASH_SIZE; i++) {
1530                         cfs_list_for_each_entry_safe(peer, next,
1531                                                      &kmxlnd_data.kmx_peers[i],
1532                                                      mxp_list) {
1533                                 mxlnd_del_peer_locked(peer);
1534                         }
1535                 }
1536         }
1537         write_unlock(&kmxlnd_data.kmx_global_lock);
1538
1539         return ret;
1540 }
1541
1542 kmx_conn_t *
1543 mxlnd_get_conn_by_idx(int index)
1544 {
1545         int              i      = 0;
1546         kmx_peer_t      *peer   = NULL;
1547         kmx_conn_t      *conn   = NULL;
1548
1549         read_lock(&kmxlnd_data.kmx_global_lock);
1550         for (i = 0; i < MXLND_HASH_SIZE; i++) {
1551                 cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],
1552                                         mxp_list) {
1553                         cfs_list_for_each_entry(conn, &peer->mxp_conns,
1554                                                 mxk_list) {
1555                                 if (index-- > 0) {
1556                                         continue;
1557                                 }
1558
1559                                 mxlnd_conn_addref(conn); /* add ref here, dec in ctl() */
1560                                 read_unlock(&kmxlnd_data.kmx_global_lock);
1561                                 return conn;
1562                         }
1563                 }
1564         }
1565         read_unlock(&kmxlnd_data.kmx_global_lock);
1566
1567         return NULL;
1568 }
1569
1570 void
1571 mxlnd_close_matching_conns_locked(kmx_peer_t *peer)
1572 {
1573         kmx_conn_t      *conn   = NULL;
1574         kmx_conn_t      *next   = NULL;
1575
1576         cfs_list_for_each_entry_safe(conn, next, &peer->mxp_conns, mxk_list)
1577                 mxlnd_conn_disconnect(conn, 0, 1);
1578
1579         return;
1580 }
1581
1582 int
1583 mxlnd_close_matching_conns(lnet_nid_t nid)
1584 {
1585         int             i       = 0;
1586         int             ret     = 0;
1587         kmx_peer_t      *peer   = NULL;
1588
1589         write_lock(&kmxlnd_data.kmx_global_lock);
1590         if (nid != LNET_NID_ANY) {
1591                 peer = mxlnd_find_peer_by_nid_locked(nid); /* adds peer ref */
1592                 if (peer == NULL) {
1593                         ret = -ENOENT;
1594                 } else {
1595                         mxlnd_close_matching_conns_locked(peer);
1596                         mxlnd_peer_decref(peer); /* and drops it here */
1597                 }
1598         } else { /* LNET_NID_ANY */
1599                 for (i = 0; i < MXLND_HASH_SIZE; i++) {
1600                         cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],                                                mxp_list)
1601                                 mxlnd_close_matching_conns_locked(peer);
1602                 }
1603         }
1604         write_unlock(&kmxlnd_data.kmx_global_lock);
1605
1606         return ret;
1607 }
1608
1609 /**
1610  * mxlnd_ctl - modify MXLND parameters
1611  * @ni - LNET interface handle
1612  * @cmd - command to change
1613  * @arg - the ioctl data
1614  */
1615 int
1616 mxlnd_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
1617 {
1618         struct libcfs_ioctl_data *data  = arg;
1619         int                       ret   = -EINVAL;
1620
1621         LASSERT (ni == kmxlnd_data.kmx_ni);
1622
1623         switch (cmd) {
1624         case IOC_LIBCFS_GET_PEER: {
1625                 lnet_nid_t      nid     = 0;
1626                 int             count   = 0;
1627
1628                 ret = mxlnd_get_peer_info(data->ioc_count, &nid, &count);
1629                 data->ioc_nid    = nid;
1630                 data->ioc_count  = count;
1631                 break;
1632         }
1633         case IOC_LIBCFS_DEL_PEER: {
1634                 ret = mxlnd_del_peer(data->ioc_nid);
1635                 break;
1636         }
1637         case IOC_LIBCFS_GET_CONN: {
1638                 kmx_conn_t      *conn = NULL;
1639
1640                 conn = mxlnd_get_conn_by_idx(data->ioc_count);
1641                 if (conn == NULL) {
1642                         ret = -ENOENT;
1643                 } else {
1644                         ret = 0;
1645                         data->ioc_nid = conn->mxk_peer->mxp_nid;
1646                         mxlnd_conn_decref(conn); /* dec ref taken in get_conn_by_idx() */
1647                 }
1648                 break;
1649         }
1650         case IOC_LIBCFS_CLOSE_CONNECTION: {
1651                 ret = mxlnd_close_matching_conns(data->ioc_nid);
1652                 break;
1653         }
1654         default:
1655                 CNETERR("unknown ctl(%d)\n", cmd);
1656                 break;
1657         }
1658
1659         return ret;
1660 }
1661
1662 /**
1663  * mxlnd_peer_queue_tx_locked - add the tx to the peer's tx queue
1664  * @tx
1665  *
1666  * Add the tx to the peer's msg or data queue. The caller has locked the peer.
1667  */
1668 void
1669 mxlnd_peer_queue_tx_locked(kmx_ctx_t *tx)
1670 {
1671         u8              msg_type        = tx->mxc_msg_type;
1672         kmx_conn_t      *conn           = tx->mxc_conn;
1673
1674         LASSERT (msg_type != 0);
1675         LASSERT (tx->mxc_nid != 0);
1676         LASSERT (tx->mxc_peer != NULL);
1677         LASSERT (tx->mxc_conn != NULL);
1678
1679         tx->mxc_incarnation = conn->mxk_incarnation;
1680
1681         if (msg_type != MXLND_MSG_PUT_DATA &&
1682             msg_type != MXLND_MSG_GET_DATA) {
1683                 /* msg style tx */
1684                 if (mxlnd_tx_requires_credit(tx)) {
1685                         cfs_list_add_tail(&tx->mxc_list,
1686                                           &conn->mxk_tx_credit_queue);
1687                         conn->mxk_ntx_msgs++;
1688                 } else if (msg_type == MXLND_MSG_CONN_REQ ||
1689                            msg_type == MXLND_MSG_CONN_ACK) {
1690                         /* put conn msgs at the front of the queue */
1691                         cfs_list_add(&tx->mxc_list, &conn->mxk_tx_free_queue);
1692                 } else {
1693                         /* PUT_ACK, PUT_NAK */
1694                         cfs_list_add_tail(&tx->mxc_list,
1695                                           &conn->mxk_tx_free_queue);
1696                         conn->mxk_ntx_msgs++;
1697                 }
1698         } else {
1699                 /* data style tx */
1700                 cfs_list_add_tail(&tx->mxc_list, &conn->mxk_tx_free_queue);
1701                 conn->mxk_ntx_data++;
1702         }
1703
1704         return;
1705 }
1706
1707 /**
1708  * mxlnd_peer_queue_tx - add the tx to the global tx queue
1709  * @tx
1710  *
1711  * Add the tx to the peer's msg or data queue
1712  */
1713 static inline void
1714 mxlnd_peer_queue_tx(kmx_ctx_t *tx)
1715 {
1716         LASSERT(tx->mxc_peer != NULL);
1717         LASSERT(tx->mxc_conn != NULL);
1718         spin_lock(&tx->mxc_conn->mxk_lock);
1719         mxlnd_peer_queue_tx_locked(tx);
1720         spin_unlock(&tx->mxc_conn->mxk_lock);
1721
1722         return;
1723 }
1724
1725 /**
1726  * mxlnd_queue_tx - add the tx to the global tx queue
1727  * @tx
1728  *
1729  * Add the tx to the global queue and up the tx_queue_sem
1730  */
1731 void
1732 mxlnd_queue_tx(kmx_ctx_t *tx)
1733 {
1734         kmx_peer_t *peer   = tx->mxc_peer;
1735         LASSERT (tx->mxc_nid != 0);
1736
1737         if (peer != NULL) {
1738                 if (peer->mxp_incompatible &&
1739                     tx->mxc_msg_type != MXLND_MSG_CONN_ACK) {
1740                         /* let this fail now */
1741                         tx->mxc_errno = -ECONNABORTED;
1742                         mxlnd_conn_decref(peer->mxp_conn);
1743                         mxlnd_put_idle_tx(tx);
1744                         return;
1745                 }
1746                 if (tx->mxc_conn == NULL) {
1747                         int             ret     = 0;
1748                         kmx_conn_t      *conn   = NULL;
1749
1750                         ret = mxlnd_conn_alloc(&conn, peer); /* adds 2nd ref for tx... */
1751                         if (ret != 0) {
1752                                 tx->mxc_errno = ret;
1753                                 mxlnd_put_idle_tx(tx);
1754                                 goto done;
1755                         }
1756                         tx->mxc_conn = conn;
1757                         mxlnd_peer_decref(peer); /* and takes it from peer */
1758                 }
1759                 LASSERT(tx->mxc_conn != NULL);
1760                 mxlnd_peer_queue_tx(tx);
1761                 mxlnd_check_sends(peer);
1762         } else {
1763                 spin_lock(&kmxlnd_data.kmx_tx_queue_lock);
1764                 cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_tx_queue);
1765                 spin_unlock(&kmxlnd_data.kmx_tx_queue_lock);
1766                 up(&kmxlnd_data.kmx_tx_queue_sem);
1767         }
1768 done:
1769         return;
1770 }
1771
1772 int
1773 mxlnd_setup_iov(kmx_ctx_t *ctx, u32 niov, struct iovec *iov, u32 offset, u32 nob)
1774 {
1775         int             i                       = 0;
1776         int             sum                     = 0;
1777         int             old_sum                 = 0;
1778         int             nseg                    = 0;
1779         int             first_iov               = -1;
1780         int             first_iov_offset        = 0;
1781         int             first_found             = 0;
1782         int             last_iov                = -1;
1783         int             last_iov_length         = 0;
1784         mx_ksegment_t  *seg                     = NULL;
1785
1786         if (niov == 0) return 0;
1787         LASSERT(iov != NULL);
1788
1789         for (i = 0; i < niov; i++) {
1790                 sum = old_sum + (u32) iov[i].iov_len;
1791                 if (!first_found && (sum > offset)) {
1792                         first_iov = i;
1793                         first_iov_offset = offset - old_sum;
1794                         first_found = 1;
1795                         sum = (u32) iov[i].iov_len - first_iov_offset;
1796                         old_sum = 0;
1797                 }
1798                 if (sum >= nob) {
1799                         last_iov = i;
1800                         last_iov_length = (u32) iov[i].iov_len - (sum - nob);
1801                         if (first_iov == last_iov) last_iov_length -= first_iov_offset;
1802                         break;
1803                 }
1804                 old_sum = sum;
1805         }
1806         LASSERT(first_iov >= 0 && last_iov >= first_iov);
1807         nseg = last_iov - first_iov + 1;
1808         LASSERT(nseg > 0);
1809
1810         MXLND_ALLOC(seg, nseg * sizeof(*seg));
1811         if (seg == NULL) {
1812                 CNETERR("MXLND_ALLOC() failed\n");
1813                 return -1;
1814         }
1815         memset(seg, 0, nseg * sizeof(*seg));
1816         ctx->mxc_nseg = nseg;
1817         sum = 0;
1818         for (i = 0; i < nseg; i++) {
1819                 seg[i].segment_ptr = MX_PA_TO_U64(virt_to_phys(iov[first_iov + i].iov_base));
1820                 seg[i].segment_length = (u32) iov[first_iov + i].iov_len;
1821                 if (i == 0) {
1822                         seg[i].segment_ptr += (u64) first_iov_offset;
1823                         seg[i].segment_length -= (u32) first_iov_offset;
1824                 }
1825                 if (i == (nseg - 1)) {
1826                         seg[i].segment_length = (u32) last_iov_length;
1827                 }
1828                 sum += seg[i].segment_length;
1829         }
1830         ctx->mxc_seg_list = seg;
1831         ctx->mxc_pin_type = MX_PIN_PHYSICAL;
1832 #ifdef MX_PIN_FULLPAGES
1833         ctx->mxc_pin_type |= MX_PIN_FULLPAGES;
1834 #endif
1835         LASSERT(nob == sum);
1836         return 0;
1837 }
1838
1839 int
1840 mxlnd_setup_kiov(kmx_ctx_t *ctx, u32 niov, lnet_kiov_t *kiov, u32 offset, u32 nob)
1841 {
1842         int             i                       = 0;
1843         int             sum                     = 0;
1844         int             old_sum                 = 0;
1845         int             nseg                    = 0;
1846         int             first_kiov              = -1;
1847         int             first_kiov_offset       = 0;
1848         int             first_found             = 0;
1849         int             last_kiov               = -1;
1850         int             last_kiov_length        = 0;
1851         mx_ksegment_t  *seg                     = NULL;
1852
1853         if (niov == 0) return 0;
1854         LASSERT(kiov != NULL);
1855
1856         for (i = 0; i < niov; i++) {
1857                 sum = old_sum + kiov[i].kiov_len;
1858                 if (i == 0) sum -= kiov[i].kiov_offset;
1859                 if (!first_found && (sum > offset)) {
1860                         first_kiov = i;
1861                         first_kiov_offset = offset - old_sum;
1862                         if (i == 0) first_kiov_offset = kiov[i].kiov_offset;
1863                         first_found = 1;
1864                         sum = kiov[i].kiov_len - first_kiov_offset;
1865                         old_sum = 0;
1866                 }
1867                 if (sum >= nob) {
1868                         last_kiov = i;
1869                         last_kiov_length = kiov[i].kiov_len - (sum - nob);
1870                         if (first_kiov == last_kiov) last_kiov_length -= first_kiov_offset;
1871                         break;
1872                 }
1873                 old_sum = sum;
1874         }
1875         LASSERT(first_kiov >= 0 && last_kiov >= first_kiov);
1876         nseg = last_kiov - first_kiov + 1;
1877         LASSERT(nseg > 0);
1878
1879         MXLND_ALLOC(seg, nseg * sizeof(*seg));
1880         if (seg == NULL) {
1881                 CNETERR("MXLND_ALLOC() failed\n");
1882                 return -1;
1883         }
1884         memset(seg, 0, niov * sizeof(*seg));
1885         ctx->mxc_nseg = niov;
1886         sum = 0;
1887         for (i = 0; i < niov; i++) {
1888                 seg[i].segment_ptr = lnet_page2phys(kiov[first_kiov + i].kiov_page);
1889                 seg[i].segment_length = kiov[first_kiov + i].kiov_len;
1890                 if (i == 0) {
1891                         seg[i].segment_ptr += (u64) first_kiov_offset;
1892                         /* we have to add back the original kiov_offset */
1893                         seg[i].segment_length -= first_kiov_offset +
1894                                                  kiov[first_kiov].kiov_offset;
1895                 }
1896                 if (i == (nseg - 1)) {
1897                         seg[i].segment_length = last_kiov_length;
1898                 }
1899                 sum += seg[i].segment_length;
1900         }
1901         ctx->mxc_seg_list = seg;
1902         ctx->mxc_pin_type = MX_PIN_PHYSICAL;
1903 #ifdef MX_PIN_FULLPAGES
1904         ctx->mxc_pin_type |= MX_PIN_FULLPAGES;
1905 #endif
1906         LASSERT(nob == sum);
1907         return 0;
1908 }
1909
1910 void
1911 mxlnd_send_nak(kmx_ctx_t *tx, lnet_nid_t nid, int type, int status, __u64 cookie)
1912 {
1913         LASSERT(type == MXLND_MSG_PUT_ACK);
1914         mxlnd_init_tx_msg(tx, type, sizeof(kmx_putack_msg_t), tx->mxc_nid);
1915         tx->mxc_cookie = cookie;
1916         tx->mxc_msg->mxm_u.put_ack.mxpam_src_cookie = cookie;
1917         tx->mxc_msg->mxm_u.put_ack.mxpam_dst_cookie = ((u64) status << MXLND_ERROR_OFFSET); /* error code */
1918         tx->mxc_match = mxlnd_create_match(tx, status);
1919
1920         mxlnd_queue_tx(tx);
1921 }
1922
1923
1924 /**
1925  * mxlnd_send_data - get tx, map [k]iov, queue tx
1926  * @ni
1927  * @lntmsg
1928  * @peer
1929  * @msg_type
1930  * @cookie
1931  *
1932  * This setups the DATA send for PUT or GET.
1933  *
1934  * On success, it queues the tx, on failure it calls lnet_finalize()
1935  */
1936 void
1937 mxlnd_send_data(lnet_ni_t *ni, lnet_msg_t *lntmsg, kmx_peer_t *peer, u8 msg_type, u64 cookie)
1938 {
1939         int                     ret     = 0;
1940         lnet_process_id_t       target  = lntmsg->msg_target;
1941         unsigned int            niov    = lntmsg->msg_niov;
1942         struct iovec           *iov     = lntmsg->msg_iov;
1943         lnet_kiov_t            *kiov    = lntmsg->msg_kiov;
1944         unsigned int            offset  = lntmsg->msg_offset;
1945         unsigned int            nob     = lntmsg->msg_len;
1946         kmx_ctx_t              *tx      = NULL;
1947
1948         LASSERT(lntmsg != NULL);
1949         LASSERT(peer != NULL);
1950         LASSERT(msg_type == MXLND_MSG_PUT_DATA || msg_type == MXLND_MSG_GET_DATA);
1951         LASSERT((cookie>>MXLND_ERROR_OFFSET) == 0);
1952
1953         tx = mxlnd_get_idle_tx();
1954         if (tx == NULL) {
1955                 CNETERR("Can't allocate %s tx for %s\n",
1956                         msg_type == MXLND_MSG_PUT_DATA ? "PUT_DATA" : "GET_DATA",
1957                         libcfs_nid2str(target.nid));
1958                 goto failed_0;
1959         }
1960         tx->mxc_nid = target.nid;
1961         /* NOTE called when we have a ref on the conn, get one for this tx */
1962         mxlnd_conn_addref(peer->mxp_conn);
1963         tx->mxc_peer = peer;
1964         tx->mxc_conn = peer->mxp_conn;
1965         tx->mxc_msg_type = msg_type;
1966         tx->mxc_lntmsg[0] = lntmsg;
1967         tx->mxc_cookie = cookie;
1968         tx->mxc_match = mxlnd_create_match(tx, 0);
1969
1970         /* This setups up the mx_ksegment_t to send the DATA payload  */
1971         if (nob == 0) {
1972                 /* do not setup the segments */
1973                 CNETERR("nob = 0; why didn't we use an EAGER reply "
1974                         "to %s?\n", libcfs_nid2str(target.nid));
1975                 ret = 0;
1976         } else if (kiov == NULL) {
1977                 ret = mxlnd_setup_iov(tx, niov, iov, offset, nob);
1978         } else {
1979                 ret = mxlnd_setup_kiov(tx, niov, kiov, offset, nob);
1980         }
1981         if (ret != 0) {
1982                 CNETERR("Can't setup send DATA for %s\n",
1983                         libcfs_nid2str(target.nid));
1984                 tx->mxc_errno = -EIO;
1985                 goto failed_1;
1986         }
1987         mxlnd_queue_tx(tx);
1988         return;
1989
1990 failed_1:
1991         mxlnd_conn_decref(peer->mxp_conn);
1992         mxlnd_put_idle_tx(tx);
1993         return;
1994
1995 failed_0:
1996         CNETERR("no tx avail\n");
1997         lnet_finalize(ni, lntmsg, -EIO);
1998         return;
1999 }
2000
2001 /**
2002  * mxlnd_recv_data - map [k]iov, post rx
2003  * @ni
2004  * @lntmsg
2005  * @rx
2006  * @msg_type
2007  * @cookie
2008  *
2009  * This setups the DATA receive for PUT or GET.
2010  *
2011  * On success, it returns 0, on failure it returns -1
2012  */
2013 int
2014 mxlnd_recv_data(lnet_ni_t *ni, lnet_msg_t *lntmsg, kmx_ctx_t *rx, u8 msg_type, u64 cookie)
2015 {
2016         int                     ret     = 0;
2017         lnet_process_id_t       target  = lntmsg->msg_target;
2018         unsigned int            niov    = lntmsg->msg_niov;
2019         struct iovec           *iov     = lntmsg->msg_iov;
2020         lnet_kiov_t            *kiov    = lntmsg->msg_kiov;
2021         unsigned int            offset  = lntmsg->msg_offset;
2022         unsigned int            nob     = lntmsg->msg_len;
2023         mx_return_t             mxret   = MX_SUCCESS;
2024         u64                     mask    = ~(MXLND_ERROR_MASK);
2025
2026         /* above assumes MXLND_MSG_PUT_DATA */
2027         if (msg_type == MXLND_MSG_GET_DATA) {
2028                 niov = lntmsg->msg_md->md_niov;
2029                 iov = lntmsg->msg_md->md_iov.iov;
2030                 kiov = lntmsg->msg_md->md_iov.kiov;
2031                 offset = 0;
2032                 nob = lntmsg->msg_md->md_length;
2033         }
2034
2035         LASSERT(lntmsg != NULL);
2036         LASSERT(rx != NULL);
2037         LASSERT(msg_type == MXLND_MSG_PUT_DATA || msg_type == MXLND_MSG_GET_DATA);
2038         LASSERT((cookie>>MXLND_ERROR_OFFSET) == 0); /* ensure top 12 bits are 0 */
2039
2040         rx->mxc_msg_type = msg_type;
2041         rx->mxc_state = MXLND_CTX_PENDING;
2042         rx->mxc_nid = target.nid;
2043         /* if posting a GET_DATA, we may not yet know the peer */
2044         if (rx->mxc_peer != NULL) {
2045                 rx->mxc_conn = rx->mxc_peer->mxp_conn;
2046         }
2047         rx->mxc_lntmsg[0] = lntmsg;
2048         rx->mxc_cookie = cookie;
2049         rx->mxc_match = mxlnd_create_match(rx, 0);
2050         /* This setups up the mx_ksegment_t to receive the DATA payload  */
2051         if (kiov == NULL) {
2052                 ret = mxlnd_setup_iov(rx, niov, iov, offset, nob);
2053         } else {
2054                 ret = mxlnd_setup_kiov(rx, niov, kiov, offset, nob);
2055         }
2056         if (msg_type == MXLND_MSG_GET_DATA) {
2057                 rx->mxc_lntmsg[1] = lnet_create_reply_msg(kmxlnd_data.kmx_ni, lntmsg);
2058                 if (rx->mxc_lntmsg[1] == NULL) {
2059                         CNETERR("Can't create reply for GET -> %s\n",
2060                                 libcfs_nid2str(target.nid));
2061                         ret = -1;
2062                 }
2063         }
2064         if (ret != 0) {
2065                 CNETERR("Can't setup %s rx for %s\n",
2066                        msg_type == MXLND_MSG_PUT_DATA ? "PUT_DATA" : "GET_DATA",
2067                        libcfs_nid2str(target.nid));
2068                 return -1;
2069         }
2070         ret = mxlnd_q_pending_ctx(rx);
2071         if (ret == -1) {
2072                 return -1;
2073         }
2074         CDEBUG(D_NET, "receiving %s 0x%llx\n", mxlnd_msgtype_to_str(msg_type), rx->mxc_cookie);
2075         mxret = mx_kirecv(kmxlnd_data.kmx_endpt,
2076                           rx->mxc_seg_list, rx->mxc_nseg,
2077                           rx->mxc_pin_type, rx->mxc_match,
2078                           mask, (void *) rx,
2079                           &rx->mxc_mxreq);
2080         if (mxret != MX_SUCCESS) {
2081                 if (rx->mxc_conn != NULL) {
2082                         mxlnd_deq_pending_ctx(rx);
2083                 }
2084                 CNETERR("mx_kirecv() failed with %d for %s\n",
2085                         (int) mxret, libcfs_nid2str(target.nid));
2086                 return -1;
2087         }
2088
2089         return 0;
2090 }
2091
2092 /**
2093  * mxlnd_send - the LND required send function
2094  * @ni
2095  * @private
2096  * @lntmsg
2097  *
2098  * This must not block. Since we may not have a peer struct for the receiver,
2099  * it will append send messages on a global tx list. We will then up the
2100  * tx_queued's semaphore to notify it of the new send.
2101  */
2102 int
2103 mxlnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
2104 {
2105         int                     ret             = 0;
2106         int                     type            = lntmsg->msg_type;
2107         lnet_hdr_t             *hdr             = &lntmsg->msg_hdr;
2108         lnet_process_id_t       target          = lntmsg->msg_target;
2109         lnet_nid_t              nid             = target.nid;
2110         int                     target_is_router = lntmsg->msg_target_is_router;
2111         int                     routing         = lntmsg->msg_routing;
2112         unsigned int            payload_niov    = lntmsg->msg_niov;
2113         struct iovec           *payload_iov     = lntmsg->msg_iov;
2114         lnet_kiov_t            *payload_kiov    = lntmsg->msg_kiov;
2115         unsigned int            payload_offset  = lntmsg->msg_offset;
2116         unsigned int            payload_nob     = lntmsg->msg_len;
2117         kmx_ctx_t              *tx              = NULL;
2118         kmx_msg_t              *txmsg           = NULL;
2119         kmx_ctx_t              *rx              = (kmx_ctx_t *) private; /* for REPLY */
2120         kmx_ctx_t              *rx_data         = NULL;
2121         kmx_conn_t             *conn            = NULL;
2122         int                     nob             = 0;
2123         uint32_t                length          = 0;
2124         kmx_peer_t             *peer            = NULL;
2125         rwlock_t                *g_lock         =&kmxlnd_data.kmx_global_lock;
2126
2127         CDEBUG(D_NET, "sending %d bytes in %d frags to %s\n",
2128                        payload_nob, payload_niov, libcfs_id2str(target));
2129
2130         LASSERT (payload_nob == 0 || payload_niov > 0);
2131         LASSERT (payload_niov <= LNET_MAX_IOV);
2132         /* payload is either all vaddrs or all pages */
2133         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
2134
2135         /* private is used on LNET_GET_REPLY only, NULL for all other cases */
2136
2137         /* NOTE we may not know the peer if it is the very first PUT_REQ or GET_REQ
2138          * to a new peer, so create one if not found */
2139         peer = mxlnd_find_peer_by_nid(nid, 1); /* adds peer ref */
2140         if (peer == NULL || peer->mxp_conn == NULL) {
2141                 /* we could not find it nor could we create one or
2142                  * one exists but we cannot create a conn,
2143                  * fail this message */
2144                 if (peer) {
2145                         /* found peer without conn, drop ref taken above */
2146                         LASSERT(peer->mxp_conn == NULL);
2147                         mxlnd_peer_decref(peer);
2148                 }
2149                 return -ENOMEM;
2150         }
2151
2152         /* we have a peer with a conn */
2153
2154         if (unlikely(peer->mxp_incompatible)) {
2155                 mxlnd_peer_decref(peer); /* drop ref taken above */
2156         } else {
2157                 read_lock(g_lock);
2158                 conn = peer->mxp_conn;
2159                 if (conn && conn->mxk_status != MXLND_CONN_DISCONNECT)
2160                         mxlnd_conn_addref(conn);
2161                 else
2162                         conn = NULL;
2163                 read_unlock(g_lock);
2164                 mxlnd_peer_decref(peer); /* drop peer ref taken above */
2165                 if (!conn)
2166                         return -ENOTCONN;
2167         }
2168
2169         LASSERT(peer && conn);
2170
2171         CDEBUG(D_NET, "%s: peer 0x%llx is 0x%p\n", __func__, nid, peer);
2172
2173         switch (type) {
2174         case LNET_MSG_ACK:
2175                 LASSERT (payload_nob == 0);
2176                 break;
2177
2178         case LNET_MSG_REPLY:
2179         case LNET_MSG_PUT:
2180                 /* Is the payload small enough not to need DATA? */
2181                 nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[payload_nob]);
2182                 if (nob <= MXLND_MSG_SIZE)
2183                         break;                  /* send EAGER */
2184
2185                 tx = mxlnd_get_idle_tx();
2186                 if (unlikely(tx == NULL)) {
2187                         CNETERR("Can't allocate %s tx for %s\n",
2188                                 type == LNET_MSG_PUT ? "PUT" : "REPLY",
2189                                 libcfs_nid2str(nid));
2190                         if (conn) mxlnd_conn_decref(conn);
2191                         return -ENOMEM;
2192                 }
2193
2194                 tx->mxc_peer = peer;
2195                 tx->mxc_conn = conn;
2196                 /* we added a conn ref above */
2197                 mxlnd_init_tx_msg (tx, MXLND_MSG_PUT_REQ, sizeof(kmx_putreq_msg_t), nid);
2198                 txmsg = tx->mxc_msg;
2199                 txmsg->mxm_u.put_req.mxprm_hdr = *hdr;
2200                 txmsg->mxm_u.put_req.mxprm_cookie = tx->mxc_cookie;
2201                 tx->mxc_match = mxlnd_create_match(tx, 0);
2202
2203                 /* we must post a receive _before_ sending the request.
2204                  * we need to determine how much to receive, it will be either
2205                  * a put_ack or a put_nak. The put_ack is larger, so use it. */
2206
2207                 rx = mxlnd_get_idle_rx(conn);
2208                 if (unlikely(rx == NULL)) {
2209                         CNETERR("Can't allocate rx for PUT_ACK for %s\n",
2210                                 libcfs_nid2str(nid));
2211                         mxlnd_put_idle_tx(tx);
2212                         if (conn) mxlnd_conn_decref(conn); /* for the ref taken above */
2213                         return -ENOMEM;
2214                 }
2215                 rx->mxc_nid = nid;
2216                 rx->mxc_peer = peer;
2217                 mxlnd_conn_addref(conn); /* for this rx */
2218                 rx->mxc_conn = conn;
2219                 rx->mxc_msg_type = MXLND_MSG_PUT_ACK;
2220                 rx->mxc_cookie = tx->mxc_cookie;
2221                 rx->mxc_match = mxlnd_create_match(rx, 0);
2222
2223                 length = offsetof(kmx_msg_t, mxm_u) + sizeof(kmx_putack_msg_t);
2224                 ret = mxlnd_recv_msg(lntmsg, rx, MXLND_MSG_PUT_ACK, rx->mxc_match, length);
2225                 if (unlikely(ret != 0)) {
2226                         CNETERR("recv_msg() failed for PUT_ACK for %s\n",
2227                                            libcfs_nid2str(nid));
2228                         rx->mxc_lntmsg[0] = NULL;
2229                         mxlnd_put_idle_rx(rx);
2230                         mxlnd_put_idle_tx(tx);
2231                         mxlnd_conn_decref(conn); /* for the rx... */
2232                         mxlnd_conn_decref(conn); /* and for the tx */
2233                         return -EHOSTUNREACH;
2234                 }
2235
2236                 mxlnd_queue_tx(tx);
2237                 return 0;
2238
2239         case LNET_MSG_GET:
2240                 if (routing || target_is_router)
2241                         break;                  /* send EAGER */
2242
2243                 /* is the REPLY message too small for DATA? */
2244                 nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[lntmsg->msg_md->md_length]);
2245                 if (nob <= MXLND_MSG_SIZE)
2246                         break;                  /* send EAGER */
2247
2248                 /* get tx (we need the cookie) , post rx for incoming DATA,
2249                  * then post GET_REQ tx */
2250                 tx = mxlnd_get_idle_tx();
2251                 if (unlikely(tx == NULL)) {
2252                         CNETERR("Can't allocate GET tx for %s\n",
2253                                 libcfs_nid2str(nid));
2254                         mxlnd_conn_decref(conn); /* for the ref taken above */
2255                         return -ENOMEM;
2256                 }
2257                 rx_data = mxlnd_get_idle_rx(conn);
2258                 if (unlikely(rx_data == NULL)) {
2259                         CNETERR("Can't allocate DATA rx for %s\n",
2260                                 libcfs_nid2str(nid));
2261                         mxlnd_put_idle_tx(tx);
2262                         mxlnd_conn_decref(conn); /* for the ref taken above */
2263                         return -ENOMEM;
2264                 }
2265                 rx_data->mxc_peer = peer;
2266                 /* NOTE no need to lock peer before adding conn ref since we took
2267                  * a conn ref for the tx (it cannot be freed between there and here ) */
2268                 mxlnd_conn_addref(conn); /* for the rx_data */
2269                 rx_data->mxc_conn = conn;
2270
2271                 ret = mxlnd_recv_data(ni, lntmsg, rx_data, MXLND_MSG_GET_DATA, tx->mxc_cookie);
2272                 if (unlikely(ret != 0)) {
2273                         CNETERR("Can't setup GET sink for %s\n",
2274                                 libcfs_nid2str(nid));
2275                         mxlnd_put_idle_rx(rx_data);
2276                         mxlnd_put_idle_tx(tx);
2277                         mxlnd_conn_decref(conn); /* for the rx_data... */
2278                         mxlnd_conn_decref(conn); /* and for the tx */
2279                         return -EIO;
2280                 }
2281
2282                 tx->mxc_peer = peer;
2283                 tx->mxc_conn = conn;
2284                 /* conn ref taken above */
2285                 mxlnd_init_tx_msg(tx, MXLND_MSG_GET_REQ, sizeof(kmx_getreq_msg_t), nid);
2286                 txmsg = tx->mxc_msg;
2287                 txmsg->mxm_u.get_req.mxgrm_hdr = *hdr;
2288                 txmsg->mxm_u.get_req.mxgrm_cookie = tx->mxc_cookie;
2289                 tx->mxc_match = mxlnd_create_match(tx, 0);
2290
2291                 mxlnd_queue_tx(tx);
2292                 return 0;
2293
2294         default:
2295                 LBUG();
2296                 mxlnd_conn_decref(conn); /* drop ref taken above */
2297                 return -EIO;
2298         }
2299
2300         /* send EAGER */
2301
2302         LASSERT (offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[payload_nob])
2303                 <= MXLND_MSG_SIZE);
2304
2305         tx = mxlnd_get_idle_tx();
2306         if (unlikely(tx == NULL)) {
2307                 CNETERR("Can't send %s to %s: tx descs exhausted\n",
2308                         mxlnd_lnetmsg_to_str(type), libcfs_nid2str(nid));
2309                 mxlnd_conn_decref(conn); /* drop ref taken above */
2310                 return -ENOMEM;
2311         }
2312
2313         tx->mxc_peer = peer;
2314         tx->mxc_conn = conn;
2315         /* conn ref taken above */
2316         nob = offsetof(kmx_eager_msg_t, mxem_payload[payload_nob]);
2317         mxlnd_init_tx_msg (tx, MXLND_MSG_EAGER, nob, nid);
2318         tx->mxc_match = mxlnd_create_match(tx, 0);
2319
2320         txmsg = tx->mxc_msg;
2321         txmsg->mxm_u.eager.mxem_hdr = *hdr;
2322
2323         if (payload_kiov != NULL)
2324                 lnet_copy_kiov2flat(MXLND_MSG_SIZE, txmsg,
2325                             offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
2326                             payload_niov, payload_kiov, payload_offset, payload_nob);
2327         else
2328                 lnet_copy_iov2flat(MXLND_MSG_SIZE, txmsg,
2329                             offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
2330                             payload_niov, payload_iov, payload_offset, payload_nob);
2331
2332         tx->mxc_lntmsg[0] = lntmsg;              /* finalise lntmsg on completion */
2333         mxlnd_queue_tx(tx);
2334         return 0;
2335 }
2336
2337 /**
2338  * mxlnd_recv - the LND required recv function
2339  * @ni
2340  * @private
2341  * @lntmsg
2342  * @delayed
2343  * @niov
2344  * @kiov
2345  * @offset
2346  * @mlen
2347  * @rlen
2348  *
2349  * This must not block.
2350  */
2351 int
2352 mxlnd_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg, int delayed,
2353              unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
2354              unsigned int offset, unsigned int mlen, unsigned int rlen)
2355 {
2356         int             ret             = 0;
2357         int             nob             = 0;
2358         int             len             = 0;
2359         kmx_ctx_t       *rx             = private;
2360         kmx_msg_t       *rxmsg          = rx->mxc_msg;
2361         lnet_nid_t       nid            = rx->mxc_nid;
2362         kmx_ctx_t       *tx             = NULL;
2363         kmx_msg_t       *txmsg          = NULL;
2364         kmx_peer_t      *peer           = rx->mxc_peer;
2365         kmx_conn_t      *conn           = peer->mxp_conn;
2366         u64              cookie         = 0ULL;
2367         int              msg_type       = rxmsg->mxm_type;
2368         int              repost         = 1;
2369         int              credit         = 0;
2370         int              finalize       = 0;
2371
2372         LASSERT (mlen <= rlen);
2373         /* Either all pages or all vaddrs */
2374         LASSERT (!(kiov != NULL && iov != NULL));
2375         LASSERT (peer && conn);
2376
2377         /* conn_addref(conn) already taken for the primary rx */
2378
2379         switch (msg_type) {
2380         case MXLND_MSG_EAGER:
2381                 nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[rlen]);
2382                 len = rx->mxc_status.xfer_length;
2383                 if (unlikely(nob > len)) {
2384                         CNETERR("Eager message from %s too big: %d(%d)\n",
2385                                 libcfs_nid2str(nid), nob, len);
2386                         ret = -EPROTO;
2387                         break;
2388                 }
2389
2390                 if (kiov != NULL)
2391                         lnet_copy_flat2kiov(niov, kiov, offset,
2392                                 MXLND_MSG_SIZE, rxmsg,
2393                                 offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
2394                                 mlen);
2395                 else
2396                         lnet_copy_flat2iov(niov, iov, offset,
2397                                 MXLND_MSG_SIZE, rxmsg,
2398                                 offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
2399                                 mlen);
2400                 finalize = 1;
2401                 credit = 1;
2402                 break;
2403
2404         case MXLND_MSG_PUT_REQ:
2405                 /* we are going to reuse the rx, store the needed info */
2406                 cookie = rxmsg->mxm_u.put_req.mxprm_cookie;
2407
2408                 /* get tx, post rx, send PUT_ACK */
2409
2410                 tx = mxlnd_get_idle_tx();
2411                 if (unlikely(tx == NULL)) {
2412                         CNETERR("Can't allocate tx for %s\n", libcfs_nid2str(nid));
2413                         /* Not replying will break the connection */
2414                         ret = -ENOMEM;
2415                         break;
2416                 }
2417                 if (unlikely(mlen == 0)) {
2418                         finalize = 1;
2419                         tx->mxc_peer = peer;
2420                         tx->mxc_conn = conn;
2421                         mxlnd_send_nak(tx, nid, MXLND_MSG_PUT_ACK, 0, cookie);
2422                         /* repost = 1 */
2423                         break;
2424                 }
2425
2426                 mxlnd_init_tx_msg(tx, MXLND_MSG_PUT_ACK, sizeof(kmx_putack_msg_t), nid);
2427                 tx->mxc_peer = peer;
2428                 tx->mxc_conn = conn;
2429                 /* no need to lock peer first since we already have a ref */
2430                 mxlnd_conn_addref(conn); /* for the tx */
2431                 txmsg = tx->mxc_msg;
2432                 txmsg->mxm_u.put_ack.mxpam_src_cookie = cookie;
2433                 txmsg->mxm_u.put_ack.mxpam_dst_cookie = tx->mxc_cookie;
2434                 tx->mxc_cookie = cookie;
2435                 tx->mxc_match = mxlnd_create_match(tx, 0);
2436
2437                 /* we must post a receive _before_ sending the PUT_ACK */
2438                 mxlnd_ctx_init(rx);
2439                 rx->mxc_state = MXLND_CTX_PREP;
2440                 rx->mxc_peer = peer;
2441                 rx->mxc_conn = conn;
2442                 /* do not take another ref for this rx, it is already taken */
2443                 rx->mxc_nid = peer->mxp_nid;
2444                 ret = mxlnd_recv_data(ni, lntmsg, rx, MXLND_MSG_PUT_DATA,
2445                                       txmsg->mxm_u.put_ack.mxpam_dst_cookie);
2446
2447                 if (unlikely(ret != 0)) {
2448                         /* Notify peer that it's over */
2449                         CNETERR("Can't setup PUT_DATA rx for %s: %d\n",
2450                                 libcfs_nid2str(nid), ret);
2451                         mxlnd_ctx_init(tx);
2452                         tx->mxc_state = MXLND_CTX_PREP;
2453                         tx->mxc_peer = peer;
2454                         tx->mxc_conn = conn;
2455                         /* finalize = 0, let the PUT_ACK tx finalize this */
2456                         tx->mxc_lntmsg[0] = rx->mxc_lntmsg[0];
2457                         tx->mxc_lntmsg[1] = rx->mxc_lntmsg[1];
2458                         /* conn ref already taken above */
2459                         mxlnd_send_nak(tx, nid, MXLND_MSG_PUT_ACK, ret, cookie);
2460                         /* repost = 1 */
2461                         break;
2462                 }
2463
2464                 mxlnd_queue_tx(tx);
2465                 /* do not return a credit until after PUT_DATA returns */
2466                 repost = 0;
2467                 break;
2468
2469         case MXLND_MSG_GET_REQ:
2470                 cookie = rxmsg->mxm_u.get_req.mxgrm_cookie;
2471
2472                 if (likely(lntmsg != NULL)) {
2473                         mxlnd_send_data(ni, lntmsg, rx->mxc_peer, MXLND_MSG_GET_DATA,
2474                                         cookie);
2475                 } else {
2476                         /* GET didn't match anything */
2477                         /* The initiator has a rx mapped to [k]iov. We cannot send a nak.
2478                          * We have to embed the error code in the match bits.
2479                          * Send the error in bits 52-59 and the cookie in bits 0-51 */
2480                         tx = mxlnd_get_idle_tx();
2481                         if (unlikely(tx == NULL)) {
2482                                 CNETERR("Can't get tx for GET NAK for %s\n",
2483                                         libcfs_nid2str(nid));
2484                                 /* we can't get a tx, notify the peer that the GET failed */
2485                                 mxlnd_send_message(conn->mxk_epa, MXLND_MSG_GET_DATA,
2486                                                    ENODATA, cookie);
2487                                 ret = -ENOMEM;
2488                                 break;
2489                         }
2490                         tx->mxc_msg_type = MXLND_MSG_GET_DATA;
2491                         tx->mxc_state = MXLND_CTX_PENDING;
2492                         tx->mxc_nid = nid;
2493                         tx->mxc_peer = peer;
2494                         tx->mxc_conn = conn;
2495                         /* no need to lock peer first since we already have a ref */
2496                         mxlnd_conn_addref(conn); /* for this tx */
2497                         tx->mxc_cookie = cookie;
2498                         tx->mxc_match = mxlnd_create_match(tx, ENODATA);
2499                         tx->mxc_pin_type = MX_PIN_PHYSICAL;
2500                         mxlnd_queue_tx(tx);
2501                 }
2502                 /* finalize lntmsg after tx completes */
2503                 break;
2504
2505         default:
2506                 LBUG();
2507         }
2508
2509         if (repost) {
2510                 /* we received a message, increment peer's outstanding credits */
2511                 if (credit == 1) {
2512                         spin_lock(&conn->mxk_lock);
2513                         conn->mxk_outstanding++;
2514                         spin_unlock(&conn->mxk_lock);
2515                 }
2516                 /* we are done with the rx */
2517                 mxlnd_put_idle_rx(rx);
2518                 mxlnd_conn_decref(conn);
2519         }
2520
2521         if (finalize == 1) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg, 0);
2522
2523         /* we received a credit, see if we can use it to send a msg */
2524         if (credit) mxlnd_check_sends(peer);
2525
2526         return ret;
2527 }
2528
2529 void
2530 mxlnd_sleep(unsigned long timeout)
2531 {
2532         set_current_state(TASK_INTERRUPTIBLE);
2533         schedule_timeout(timeout);
2534         return;
2535 }
2536
2537 /**
2538  * mxlnd_tx_queued - the generic send queue thread
2539  * @arg - thread id (as a void *)
2540  *
2541  * This thread moves send messages from the global tx_queue to the owning
2542  * peer's tx_[msg|data]_queue. If the peer does not exist, it creates one and adds
2543  * it to the global peer list.
2544  */
2545 int
2546 mxlnd_tx_queued(void *arg)
2547 {
2548         long                    id      = (long) arg;
2549         int                     ret     = 0;
2550         int                     found   = 0;
2551         kmx_ctx_t              *tx      = NULL;
2552         kmx_peer_t             *peer    = NULL;
2553         cfs_list_t             *queue   = &kmxlnd_data.kmx_tx_queue;
2554         spinlock_t              *tx_q_lock = &kmxlnd_data.kmx_tx_queue_lock;
2555         rwlock_t                *g_lock  = &kmxlnd_data.kmx_global_lock;
2556
2557         while (!(atomic_read(&kmxlnd_data.kmx_shutdown))) {
2558                 ret = down_interruptible(&kmxlnd_data.kmx_tx_queue_sem);
2559                 if (atomic_read(&kmxlnd_data.kmx_shutdown))
2560                         break;
2561                 if (ret != 0) /* Should we check for -EINTR? */
2562                         continue;
2563                 spin_lock(tx_q_lock);
2564                 if (cfs_list_empty(&kmxlnd_data.kmx_tx_queue)) {
2565                         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                 spin_unlock(tx_q_lock);
2571
2572                 found = 0;
2573                 peer = mxlnd_find_peer_by_nid(tx->mxc_nid, 0); /* adds ref*/
2574                 if (peer != NULL) {
2575                         tx->mxc_peer = peer;
2576                         write_lock(g_lock);
2577                         if (peer->mxp_conn == NULL) {
2578                                 ret = mxlnd_conn_alloc_locked(&peer->mxp_conn,
2579                                                               peer);
2580                                 if (ret != 0) {
2581                                         /* out of memory: give up, fail tx */
2582                                         tx->mxc_errno = -ENOMEM;
2583                                         mxlnd_peer_decref(peer);
2584                                         write_unlock(g_lock);
2585                                         mxlnd_put_idle_tx(tx);
2586                                         continue;
2587                                 }
2588                         }
2589                         tx->mxc_conn = peer->mxp_conn;
2590                         mxlnd_conn_addref(tx->mxc_conn); /* for this tx */
2591                         mxlnd_peer_decref(peer); /* drop peer ref taken above */
2592                         write_unlock(g_lock);
2593                         mxlnd_queue_tx(tx);
2594                         found = 1;
2595                 }
2596                 if (found == 0) {
2597                         int             hash    = 0;
2598                         kmx_peer_t     *peer    = NULL;
2599                         kmx_peer_t     *old     = NULL;
2600
2601                         hash = mxlnd_nid_to_hash(tx->mxc_nid);
2602
2603                         LASSERT(tx->mxc_msg_type != MXLND_MSG_PUT_DATA &&
2604                                 tx->mxc_msg_type != MXLND_MSG_GET_DATA);
2605                         /* create peer */
2606                         /* adds conn ref for this function */
2607                         ret = mxlnd_peer_alloc(&peer, tx->mxc_nid,
2608                                         *kmxlnd_tunables.kmx_board,
2609                                         *kmxlnd_tunables.kmx_ep_id, 0ULL);
2610                         if (ret != 0) {
2611                                 /* finalize message */
2612                                 tx->mxc_errno = ret;
2613                                 mxlnd_put_idle_tx(tx);
2614                                 continue;
2615                         }
2616                         tx->mxc_peer = peer;
2617                         tx->mxc_conn = peer->mxp_conn;
2618                         /* this tx will keep the conn ref taken in peer_alloc() */
2619
2620                         /* add peer to global peer list, but look to see
2621                          * if someone already created it after we released
2622                          * the read lock */
2623                         write_lock(g_lock);
2624                         old = mxlnd_find_peer_by_nid_locked(peer->mxp_nid);
2625                         if (old) {
2626                                 /* we have a peer ref on old */
2627                                 if (old->mxp_conn) {
2628                                         found = 1;
2629                                 } else {
2630                                         /* no conn */
2631                                         /* drop our ref taken above... */
2632                                         mxlnd_peer_decref(old);
2633                                         /* and delete it */
2634                                         mxlnd_del_peer_locked(old);
2635                                 }
2636                         }
2637
2638                         if (found == 0) {
2639                                 cfs_list_add_tail(&peer->mxp_list,
2640                                                   &kmxlnd_data.kmx_peers[hash]);
2641                                 atomic_inc(&kmxlnd_data.kmx_npeers);
2642                         } else {
2643                                 tx->mxc_peer = old;
2644                                 tx->mxc_conn = old->mxp_conn;
2645                                 LASSERT(old->mxp_conn != NULL);
2646                                 mxlnd_conn_addref(old->mxp_conn);
2647                                 mxlnd_conn_decref(peer->mxp_conn); /* drop ref taken above.. */
2648                                 mxlnd_conn_decref(peer->mxp_conn); /* drop peer's ref */
2649                                 mxlnd_peer_decref(peer);
2650                         }
2651                         write_unlock(g_lock);
2652
2653                         mxlnd_queue_tx(tx);
2654                 }
2655         }
2656         mxlnd_thread_stop(id);
2657         return 0;
2658 }
2659
2660 /* When calling this, we must not have the peer lock. */
2661 void
2662 mxlnd_iconnect(kmx_peer_t *peer, u8 msg_type)
2663 {
2664         mx_return_t     mxret           = MX_SUCCESS;
2665         mx_request_t    request;
2666         kmx_conn_t      *conn           = peer->mxp_conn;
2667         u64             match           = ((u64) msg_type) << MXLND_MSG_OFFSET;
2668
2669         /* NOTE we are holding a conn ref every time we call this function,
2670          * we do not need to lock the peer before taking another ref */
2671         mxlnd_conn_addref(conn); /* hold until CONN_REQ or CONN_ACK completes */
2672
2673         LASSERT(msg_type == MXLND_MSG_ICON_REQ || msg_type == MXLND_MSG_ICON_ACK);
2674
2675         if (peer->mxp_reconnect_time == 0) {
2676                 peer->mxp_reconnect_time = jiffies;
2677         }
2678
2679         if (peer->mxp_nic_id == 0ULL) {
2680                 int     ret     = 0;
2681
2682                 ret = mxlnd_ip2nic_id(LNET_NIDADDR(peer->mxp_nid),
2683                                       &peer->mxp_nic_id, MXLND_LOOKUP_COUNT);
2684                 if (ret == 0) {
2685                         mx_nic_id_to_board_number(peer->mxp_nic_id, &peer->mxp_board);
2686                 }
2687                 if (peer->mxp_nic_id == 0ULL && conn->mxk_status == MXLND_CONN_WAIT) {
2688                         /* not mapped yet, return */
2689                         spin_lock(&conn->mxk_lock);
2690                         mxlnd_set_conn_status(conn, MXLND_CONN_INIT);
2691                         spin_unlock(&conn->mxk_lock);
2692                 }
2693         }
2694
2695         if (cfs_time_after(jiffies,
2696                            peer->mxp_reconnect_time + MXLND_CONNECT_TIMEOUT) &&
2697             conn->mxk_status != MXLND_CONN_DISCONNECT) {
2698                 /* give up and notify LNET */
2699                 CDEBUG(D_NET, "timeout trying to connect to %s\n",
2700                        libcfs_nid2str(peer->mxp_nid));
2701                 mxlnd_conn_disconnect(conn, 0, 0);
2702                 mxlnd_conn_decref(conn);
2703                 return;
2704         }
2705
2706         mxret = mx_iconnect(kmxlnd_data.kmx_endpt, peer->mxp_nic_id,
2707                             peer->mxp_ep_id, MXLND_MSG_MAGIC, match,
2708                             (void *) peer, &request);
2709         if (unlikely(mxret != MX_SUCCESS)) {
2710                 spin_lock(&conn->mxk_lock);
2711                 mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
2712                 spin_unlock(&conn->mxk_lock);
2713                 CNETERR("mx_iconnect() failed with %s (%d) to %s\n",
2714                        mx_strerror(mxret), mxret, libcfs_nid2str(peer->mxp_nid));
2715                 mxlnd_conn_decref(conn);
2716         }
2717         mx_set_request_timeout(kmxlnd_data.kmx_endpt, request,
2718                                MXLND_CONNECT_TIMEOUT/HZ*1000);
2719         return;
2720 }
2721
2722 #define MXLND_STATS 0
2723
2724 int
2725 mxlnd_check_sends(kmx_peer_t *peer)
2726 {
2727         int             ret             = 0;
2728         int             found           = 0;
2729         mx_return_t     mxret           = MX_SUCCESS;
2730         kmx_ctx_t       *tx             = NULL;
2731         kmx_conn_t      *conn           = NULL;
2732         u8              msg_type        = 0;
2733         int             credit          = 0;
2734         int             status          = 0;
2735         int             ntx_posted      = 0;
2736         int             credits         = 0;
2737 #if MXLND_STATS
2738         static unsigned long last       = 0;
2739 #endif
2740
2741         if (unlikely(peer == NULL)) {
2742                 LASSERT(peer != NULL);
2743                 return -1;
2744         }
2745         write_lock(&kmxlnd_data.kmx_global_lock);
2746         conn = peer->mxp_conn;
2747         /* NOTE take a ref for the duration of this function since it is
2748          * called when there might not be any queued txs for this peer */
2749         if (conn) {
2750                 if (conn->mxk_status == MXLND_CONN_DISCONNECT) {
2751                         write_unlock(&kmxlnd_data.kmx_global_lock);
2752                         return -1;
2753                 }
2754                 mxlnd_conn_addref(conn); /* for duration of this function */
2755         }
2756         write_unlock(&kmxlnd_data.kmx_global_lock);
2757
2758         /* do not add another ref for this tx */
2759
2760         if (conn == NULL) {
2761                 /* we do not have any conns */
2762                 CNETERR("peer %s has no conn\n", libcfs_nid2str(peer->mxp_nid));
2763                 return -1;
2764         }
2765
2766 #if MXLND_STATS
2767         if (cfs_time_after(jiffies, last)) {
2768                 last = jiffies + HZ;
2769                 CDEBUG(D_NET, "status= %s credits= %d outstanding= %d ntx_msgs= %d "
2770                               "ntx_posted= %d ntx_data= %d data_posted= %d\n",
2771                               mxlnd_connstatus_to_str(conn->mxk_status), conn->mxk_credits,
2772                               conn->mxk_outstanding, conn->mxk_ntx_msgs, conn->mxk_ntx_posted,
2773                               conn->mxk_ntx_data, conn->mxk_data_posted);
2774         }
2775 #endif
2776
2777         spin_lock(&conn->mxk_lock);
2778         ntx_posted = conn->mxk_ntx_posted;
2779         credits = conn->mxk_credits;
2780
2781         LASSERT(ntx_posted <= *kmxlnd_tunables.kmx_peercredits);
2782         LASSERT(ntx_posted >= 0);
2783
2784         LASSERT(credits <= *kmxlnd_tunables.kmx_peercredits);
2785         LASSERT(credits >= 0);
2786
2787         /* check number of queued msgs, ignore data */
2788         if (conn->mxk_outstanding >= MXLND_CREDIT_HIGHWATER()) {
2789                 /* check if any txs queued that could return credits... */
2790                 if (cfs_list_empty(&conn->mxk_tx_credit_queue) ||
2791                     conn->mxk_ntx_msgs == 0) {
2792                         /* if not, send a NOOP */
2793                         tx = mxlnd_get_idle_tx();
2794                         if (likely(tx != NULL)) {
2795                                 tx->mxc_peer = peer;
2796                                 tx->mxc_conn = peer->mxp_conn;
2797                                 mxlnd_conn_addref(conn); /* for this tx */
2798                                 mxlnd_init_tx_msg (tx, MXLND_MSG_NOOP, 0, peer->mxp_nid);
2799                                 tx->mxc_match = mxlnd_create_match(tx, 0);
2800                                 mxlnd_peer_queue_tx_locked(tx);
2801                                 found = 1;
2802                                 goto done_locked;
2803                         }
2804                 }
2805         }
2806
2807         /* if the peer is not ready, try to connect */
2808         if (unlikely(conn->mxk_status == MXLND_CONN_INIT ||
2809             conn->mxk_status == MXLND_CONN_FAIL)) {
2810                 CDEBUG(D_NET, "status=%s\n", mxlnd_connstatus_to_str(conn->mxk_status));
2811                 mxlnd_set_conn_status(conn, MXLND_CONN_WAIT);
2812                 spin_unlock(&conn->mxk_lock);
2813                 mxlnd_iconnect(peer, (u8) MXLND_MSG_ICON_REQ);
2814                 goto done;
2815         }
2816
2817         while (!cfs_list_empty(&conn->mxk_tx_free_queue) ||
2818                !cfs_list_empty(&conn->mxk_tx_credit_queue)) {
2819                 /* We have something to send. If we have a queued tx that does not
2820                  * require a credit (free), choose it since its completion will
2821                  * return a credit (here or at the peer), complete a DATA or
2822                  * CONN_REQ or CONN_ACK. */
2823                 cfs_list_t *tmp_tx = NULL;
2824                 if (!cfs_list_empty(&conn->mxk_tx_free_queue)) {
2825                         tmp_tx = &conn->mxk_tx_free_queue;
2826                 } else {
2827                         tmp_tx = &conn->mxk_tx_credit_queue;
2828                 }
2829                 tx = cfs_list_entry(tmp_tx->next, kmx_ctx_t, mxc_list);
2830
2831                 msg_type = tx->mxc_msg_type;
2832
2833                 /* don't try to send a rx */
2834                 LASSERT(tx->mxc_type == MXLND_REQ_TX);
2835
2836                 /* ensure that it is a valid msg type */
2837                 LASSERT(msg_type == MXLND_MSG_CONN_REQ ||
2838                         msg_type == MXLND_MSG_CONN_ACK ||
2839                         msg_type == MXLND_MSG_NOOP     ||
2840                         msg_type == MXLND_MSG_EAGER    ||
2841                         msg_type == MXLND_MSG_PUT_REQ  ||
2842                         msg_type == MXLND_MSG_PUT_ACK  ||
2843                         msg_type == MXLND_MSG_PUT_DATA ||
2844                         msg_type == MXLND_MSG_GET_REQ  ||
2845                         msg_type == MXLND_MSG_GET_DATA);
2846                 LASSERT(tx->mxc_peer == peer);
2847                 LASSERT(tx->mxc_nid == peer->mxp_nid);
2848
2849                 credit = mxlnd_tx_requires_credit(tx);
2850                 if (credit) {
2851
2852                         if (conn->mxk_ntx_posted == *kmxlnd_tunables.kmx_peercredits) {
2853                                 CDEBUG(D_NET, "%s: posted enough\n",
2854                                               libcfs_nid2str(peer->mxp_nid));
2855                                 goto done_locked;
2856                         }
2857
2858                         if (conn->mxk_credits == 0) {
2859                                 CDEBUG(D_NET, "%s: no credits\n",
2860                                               libcfs_nid2str(peer->mxp_nid));
2861                                 goto done_locked;
2862                         }
2863
2864                         if (conn->mxk_credits == 1 &&      /* last credit reserved for */
2865                             conn->mxk_outstanding == 0) {  /* giving back credits */
2866                                 CDEBUG(D_NET, "%s: not using last credit\n",
2867                                               libcfs_nid2str(peer->mxp_nid));
2868                                 goto done_locked;
2869                         }
2870                 }
2871
2872                 if (unlikely(conn->mxk_status != MXLND_CONN_READY)) {
2873                         if ( ! (msg_type == MXLND_MSG_CONN_REQ ||
2874                                 msg_type == MXLND_MSG_CONN_ACK)) {
2875                                 CDEBUG(D_NET, "peer status is %s for tx 0x%llx (%s)\n",
2876                                              mxlnd_connstatus_to_str(conn->mxk_status),
2877                                              tx->mxc_cookie,
2878                                              mxlnd_msgtype_to_str(tx->mxc_msg_type));
2879                                 if (conn->mxk_status == MXLND_CONN_DISCONNECT ||
2880                                     cfs_time_aftereq(jiffies, tx->mxc_deadline)) {
2881                                         cfs_list_del_init(&tx->mxc_list);
2882                                         tx->mxc_errno = -ECONNABORTED;
2883                                         spin_unlock(&conn->mxk_lock);
2884                                         mxlnd_put_idle_tx(tx);
2885                                         mxlnd_conn_decref(conn);
2886                                         goto done;
2887                                 }
2888                                 goto done_locked;
2889                         }
2890                 }
2891
2892                 cfs_list_del_init(&tx->mxc_list);
2893
2894                 /* handle credits, etc now while we have the lock to avoid races */
2895                 if (credit) {
2896                         conn->mxk_credits--;
2897                         conn->mxk_ntx_posted++;
2898                 }
2899                 if (msg_type != MXLND_MSG_PUT_DATA &&
2900                     msg_type != MXLND_MSG_GET_DATA) {
2901                         if (msg_type != MXLND_MSG_CONN_REQ &&
2902                             msg_type != MXLND_MSG_CONN_ACK) {
2903                                 conn->mxk_ntx_msgs--;
2904                         }
2905                 }
2906                 if (tx->mxc_incarnation == 0 &&
2907                     conn->mxk_incarnation != 0) {
2908                         tx->mxc_incarnation = conn->mxk_incarnation;
2909                 }
2910
2911                 /* if this is a NOOP and (1) mxp_conn->mxk_outstanding < CREDIT_HIGHWATER
2912                  * or (2) there is a non-DATA msg that can return credits in the
2913                  * queue, then drop this duplicate NOOP */
2914                 if (unlikely(msg_type == MXLND_MSG_NOOP)) {
2915                         if ((conn->mxk_outstanding < MXLND_CREDIT_HIGHWATER()) ||
2916                             (conn->mxk_ntx_msgs >= 1)) {
2917                                 conn->mxk_credits++;
2918                                 conn->mxk_ntx_posted--;
2919                                 spin_unlock(&conn->mxk_lock);
2920                                 /* redundant NOOP */
2921                                 mxlnd_put_idle_tx(tx);
2922                                 mxlnd_conn_decref(conn);
2923                                 CDEBUG(D_NET, "%s: redundant noop\n",
2924                                               libcfs_nid2str(peer->mxp_nid));
2925                                 found = 1;
2926                                 goto done;
2927                         }
2928                 }
2929
2930                 found = 1;
2931                 if (likely((msg_type != MXLND_MSG_PUT_DATA) &&
2932                     (msg_type != MXLND_MSG_GET_DATA))) {
2933                         mxlnd_pack_msg_locked(tx);
2934                 }
2935
2936                 mxret = MX_SUCCESS;
2937
2938                 status = conn->mxk_status;
2939                 spin_unlock(&conn->mxk_lock);
2940
2941                 if (likely((status == MXLND_CONN_READY) ||
2942                     (msg_type == MXLND_MSG_CONN_REQ) ||
2943                     (msg_type == MXLND_MSG_CONN_ACK))) {
2944                         ret = 0;
2945                         if (msg_type != MXLND_MSG_CONN_REQ &&
2946                             msg_type != MXLND_MSG_CONN_ACK) {
2947                                 /* add to the pending list */
2948                                 ret = mxlnd_q_pending_ctx(tx);
2949                         } else {
2950                                 /* CONN_REQ/ACK */
2951                                 tx->mxc_state = MXLND_CTX_PENDING;
2952                         }
2953
2954                         if (ret == 0) {
2955                                 if (likely(msg_type != MXLND_MSG_PUT_DATA &&
2956                                     msg_type != MXLND_MSG_GET_DATA)) {
2957                                         /* send a msg style tx */
2958                                         LASSERT(tx->mxc_nseg == 1);
2959                                         LASSERT(tx->mxc_pin_type == MX_PIN_PHYSICAL);
2960                                         CDEBUG(D_NET, "sending %s 0x%llx\n",
2961                                                mxlnd_msgtype_to_str(msg_type),
2962                                                tx->mxc_cookie);
2963                                         mxret = mx_kisend(kmxlnd_data.kmx_endpt,
2964                                                           &tx->mxc_seg,
2965                                                           tx->mxc_nseg,
2966                                                           tx->mxc_pin_type,
2967                                                           conn->mxk_epa,
2968                                                           tx->mxc_match,
2969                                                           (void *) tx,
2970                                                           &tx->mxc_mxreq);
2971                                 } else {
2972                                         /* send a DATA tx */
2973                                         spin_lock(&conn->mxk_lock);
2974                                         conn->mxk_ntx_data--;
2975                                         conn->mxk_data_posted++;
2976                                         spin_unlock(&conn->mxk_lock);
2977                                         CDEBUG(D_NET, "sending %s 0x%llx\n",
2978                                                mxlnd_msgtype_to_str(msg_type),
2979                                                tx->mxc_cookie);
2980                                         mxret = mx_kisend(kmxlnd_data.kmx_endpt,
2981                                                           tx->mxc_seg_list,
2982                                                           tx->mxc_nseg,
2983                                                           tx->mxc_pin_type,
2984                                                           conn->mxk_epa,
2985                                                           tx->mxc_match,
2986                                                           (void *) tx,
2987                                                           &tx->mxc_mxreq);
2988                                 }
2989                         } else {
2990                                 /* ret != 0 */
2991                                 mxret = MX_CONNECTION_FAILED;
2992                         }
2993                         if (likely(mxret == MX_SUCCESS)) {
2994                                 ret = 0;
2995                         } else {
2996                                 CNETERR("mx_kisend() failed with %s (%d) "
2997                                         "sending to %s\n", mx_strerror(mxret), (int) mxret,
2998                                        libcfs_nid2str(peer->mxp_nid));
2999                                 /* NOTE mx_kisend() only fails if there are not enough
3000                                 * resources. Do not change the connection status. */
3001                                 if (mxret == MX_NO_RESOURCES) {
3002                                         tx->mxc_errno = -ENOMEM;
3003                                 } else {
3004                                         tx->mxc_errno = -ECONNABORTED;
3005                                 }
3006                                 if (credit) {
3007                                         spin_lock(&conn->mxk_lock);
3008                                         conn->mxk_ntx_posted--;
3009                                         conn->mxk_credits++;
3010                                         spin_unlock(&conn->mxk_lock);
3011                                 } else if (msg_type == MXLND_MSG_PUT_DATA ||
3012                                            msg_type == MXLND_MSG_GET_DATA) {
3013                                         spin_lock(&conn->mxk_lock);
3014                                         conn->mxk_data_posted--;
3015                                         spin_unlock(&conn->mxk_lock);
3016                                 }
3017                                 if (msg_type != MXLND_MSG_PUT_DATA &&
3018                                     msg_type != MXLND_MSG_GET_DATA &&
3019                                     msg_type != MXLND_MSG_CONN_REQ &&
3020                                     msg_type != MXLND_MSG_CONN_ACK) {
3021                                         spin_lock(&conn->mxk_lock);
3022                                         conn->mxk_outstanding +=
3023                                                 tx->mxc_msg->mxm_credits;
3024                                         spin_unlock(&conn->mxk_lock);
3025                                 }
3026                                 if (msg_type != MXLND_MSG_CONN_REQ &&
3027                                     msg_type != MXLND_MSG_CONN_ACK) {
3028                                         /* remove from the pending list */
3029                                         mxlnd_deq_pending_ctx(tx);
3030                                 }
3031                                 mxlnd_put_idle_tx(tx);
3032                                 mxlnd_conn_decref(conn);
3033                         }
3034                 }
3035                 spin_lock(&conn->mxk_lock);
3036         }
3037 done_locked:
3038         spin_unlock(&conn->mxk_lock);
3039 done:
3040         mxlnd_conn_decref(conn); /* drop ref taken at start of function */
3041         return found;
3042 }
3043
3044
3045 /**
3046  * mxlnd_handle_tx_completion - a tx completed, progress or complete the msg
3047  * @ctx - the tx descriptor
3048  *
3049  * Determine which type of send request it was and start the next step, if needed,
3050  * or, if done, signal completion to LNET. After we are done, put back on the
3051  * idle tx list.
3052  */
3053 void
3054 mxlnd_handle_tx_completion(kmx_ctx_t *tx)
3055 {
3056         int             code    = tx->mxc_status.code;
3057         int             failed  = (code != MX_STATUS_SUCCESS || tx->mxc_errno != 0);
3058         kmx_msg_t       *msg    = tx->mxc_msg;
3059         kmx_peer_t      *peer   = tx->mxc_peer;
3060         kmx_conn_t      *conn   = tx->mxc_conn;
3061         u8              type    = tx->mxc_msg_type;
3062         int             credit  = mxlnd_tx_requires_credit(tx);
3063         u64             cookie  = tx->mxc_cookie;
3064
3065         CDEBUG(D_NET, "entering %s (0x%llx):\n",
3066                       mxlnd_msgtype_to_str(tx->mxc_msg_type), cookie);
3067
3068         LASSERT (peer != NULL);
3069         LASSERT (conn != NULL);
3070
3071         if (type != MXLND_MSG_PUT_DATA && type != MXLND_MSG_GET_DATA) {
3072                 LASSERT (type == msg->mxm_type);
3073         }
3074
3075         if (failed) {
3076                 if (tx->mxc_errno == 0) tx->mxc_errno = -EIO;
3077         } else {
3078                 spin_lock(&conn->mxk_lock);
3079                 conn->mxk_last_tx = cfs_time_current(); /* jiffies */
3080                 spin_unlock(&conn->mxk_lock);
3081         }
3082
3083         switch (type) {
3084
3085         case MXLND_MSG_GET_DATA:
3086                 spin_lock(&conn->mxk_lock);
3087                 if (conn->mxk_incarnation == tx->mxc_incarnation) {
3088                         conn->mxk_outstanding++;
3089                         conn->mxk_data_posted--;
3090                 }
3091                 spin_unlock(&conn->mxk_lock);
3092                 break;
3093
3094         case MXLND_MSG_PUT_DATA:
3095                 spin_lock(&conn->mxk_lock);
3096                 if (conn->mxk_incarnation == tx->mxc_incarnation) {
3097                         conn->mxk_data_posted--;
3098                 }
3099                 spin_unlock(&conn->mxk_lock);
3100                 break;
3101
3102         case MXLND_MSG_NOOP:
3103         case MXLND_MSG_PUT_REQ:
3104         case MXLND_MSG_PUT_ACK:
3105         case MXLND_MSG_GET_REQ:
3106         case MXLND_MSG_EAGER:
3107                 break;
3108
3109         case MXLND_MSG_CONN_ACK:
3110                 if (peer->mxp_incompatible) {
3111                         /* we sent our params, now close this conn */
3112                         mxlnd_conn_disconnect(conn, 0, 1);
3113                 }
3114         case MXLND_MSG_CONN_REQ:
3115                 if (failed) {
3116                         CNETERR("%s failed with %s (%d) (errno = %d) to %s\n",
3117                                type == MXLND_MSG_CONN_REQ ? "CONN_REQ" : "CONN_ACK",
3118                                mx_strstatus(code), code, tx->mxc_errno,
3119                                libcfs_nid2str(tx->mxc_nid));
3120                         if (!peer->mxp_incompatible) {
3121                                 spin_lock(&conn->mxk_lock);
3122                                 if (code == MX_STATUS_BAD_SESSION)
3123                                         mxlnd_set_conn_status(conn,
3124                                                               MXLND_CONN_INIT);
3125                                 else
3126                                         mxlnd_set_conn_status(conn,
3127                                                               MXLND_CONN_FAIL);
3128                                 spin_unlock(&conn->mxk_lock);
3129                         }
3130                 }
3131                 break;
3132
3133         default:
3134                 CNETERR("Unknown msg type of %d\n", type);
3135                 LBUG();
3136         }
3137
3138         if (credit) {
3139                 spin_lock(&conn->mxk_lock);
3140                 if (conn->mxk_incarnation == tx->mxc_incarnation) {
3141                         conn->mxk_ntx_posted--;
3142                 }
3143                 spin_unlock(&conn->mxk_lock);
3144         }
3145
3146         mxlnd_put_idle_tx(tx);
3147         mxlnd_conn_decref(conn);
3148
3149         mxlnd_check_sends(peer);
3150
3151         CDEBUG(D_NET, "leaving\n");
3152         return;
3153 }
3154
3155 /* Handle completion of MSG or DATA rx.
3156  * CONN_REQ and CONN_ACK are handled elsewhere. */
3157 void
3158 mxlnd_handle_rx_completion(kmx_ctx_t *rx)
3159 {
3160         int             ret             = 0;
3161         int             repost          = 1;
3162         int             credit          = 1;
3163         u32             nob             = rx->mxc_status.xfer_length;
3164         u64             bits            = rx->mxc_status.match_info;
3165         kmx_msg_t      *msg             = rx->mxc_msg;
3166         kmx_peer_t     *peer            = rx->mxc_peer;
3167         kmx_conn_t     *conn            = rx->mxc_conn;
3168         u8              type            = rx->mxc_msg_type;
3169         u64             seq             = bits;
3170         lnet_msg_t     *lntmsg[2];
3171         int             result          = 0;
3172         int             peer_ref        = 0;
3173         int             conn_ref        = 0;
3174
3175         /* NOTE We may only know the peer's nid if it is a PUT_REQ, GET_REQ,
3176          * failed GET reply */
3177
3178         /* NOTE peer may still be NULL if it is a new peer and
3179          *      conn may be NULL if this is a re-connect */
3180         if (likely(peer != NULL && conn != NULL)) {
3181                 /* we have a reference on the conn */
3182                 conn_ref = 1;
3183         } else if (peer != NULL && conn == NULL) {
3184                 /* we have a reference on the peer */
3185                 peer_ref = 1;
3186         } else if (peer == NULL && conn != NULL) {
3187                 /* fatal error */
3188                 CERROR("rx 0x%llx from %s has conn but no peer\n",
3189                        bits, libcfs_nid2str(rx->mxc_nid));
3190                 LBUG();
3191         } /* else peer and conn == NULL */
3192
3193         if (conn == NULL && peer != NULL) {
3194                 write_lock(&kmxlnd_data.kmx_global_lock);
3195                 conn = peer->mxp_conn;
3196                 if (conn) {
3197                         mxlnd_conn_addref(conn); /* conn takes ref... */
3198                         mxlnd_peer_decref(peer); /* from peer */
3199                         conn_ref = 1;
3200                         peer_ref = 0;
3201                 }
3202                 write_unlock(&kmxlnd_data.kmx_global_lock);
3203                 rx->mxc_conn = conn;
3204         }
3205
3206 #if MXLND_DEBUG
3207         CDEBUG(D_NET, "receiving msg bits=0x%llx nob=%d peer=0x%p\n", bits, nob, peer);
3208 #endif
3209
3210         lntmsg[0] = NULL;
3211         lntmsg[1] = NULL;
3212
3213         if (rx->mxc_status.code != MX_STATUS_SUCCESS &&
3214             rx->mxc_status.code != MX_STATUS_TRUNCATED) {
3215                 CNETERR("rx from %s failed with %s (%d)\n",
3216                         libcfs_nid2str(rx->mxc_nid),
3217                         mx_strstatus(rx->mxc_status.code),
3218                         rx->mxc_status.code);
3219                 credit = 0;
3220                 goto cleanup;
3221         }
3222
3223         if (nob == 0) {
3224                 /* this may be a failed GET reply */
3225                 if (type == MXLND_MSG_GET_DATA) {
3226                         /* get the error (52-59) bits from the match bits */
3227                         ret = (u32) MXLND_ERROR_VAL(rx->mxc_status.match_info);
3228                         lntmsg[0] = rx->mxc_lntmsg[0];
3229                         result = -ret;
3230                         goto cleanup;
3231                 } else {
3232                         /* we had a rx complete with 0 bytes (no hdr, nothing) */
3233                         CNETERR("rx from %s returned with 0 bytes\n",
3234                                 libcfs_nid2str(rx->mxc_nid));
3235                         goto cleanup;
3236                 }
3237         }
3238
3239         /* NOTE PUT_DATA and GET_DATA do not have mxc_msg, do not call unpack() */
3240         if (type == MXLND_MSG_PUT_DATA) {
3241                 /* result = 0; */
3242                 lntmsg[0] = rx->mxc_lntmsg[0];
3243                 goto cleanup;
3244         } else if (type == MXLND_MSG_GET_DATA) {
3245                 /* result = 0; */
3246                 lntmsg[0] = rx->mxc_lntmsg[0];
3247                 lntmsg[1] = rx->mxc_lntmsg[1];
3248                 goto cleanup;
3249         }
3250
3251         ret = mxlnd_unpack_msg(msg, nob);
3252         if (ret != 0) {
3253                 CNETERR("Error %d unpacking rx from %s\n",
3254                         ret, libcfs_nid2str(rx->mxc_nid));
3255                 goto cleanup;
3256         }
3257         rx->mxc_nob = nob;
3258         type = msg->mxm_type;
3259
3260         if (rx->mxc_nid != msg->mxm_srcnid ||
3261             kmxlnd_data.kmx_ni->ni_nid != msg->mxm_dstnid) {
3262                 CNETERR("rx with mismatched NID (type %s) (my nid is "
3263                        "0x%llx and rx msg dst is 0x%llx)\n",
3264                        mxlnd_msgtype_to_str(type), kmxlnd_data.kmx_ni->ni_nid,
3265                        msg->mxm_dstnid);
3266                 goto cleanup;
3267         }
3268
3269         if ((conn != NULL && msg->mxm_srcstamp != conn->mxk_incarnation) ||
3270             msg->mxm_dststamp != kmxlnd_data.kmx_incarnation) {
3271                 CNETERR("Stale rx from %s with type %s "
3272                        "(mxm_srcstamp (%lld) != mxk_incarnation (%lld) "
3273                        "|| mxm_dststamp (%lld) != kmx_incarnation (%lld))\n",
3274                        libcfs_nid2str(rx->mxc_nid), mxlnd_msgtype_to_str(type),
3275                        msg->mxm_srcstamp, conn->mxk_incarnation,
3276                        msg->mxm_dststamp, kmxlnd_data.kmx_incarnation);
3277                 credit = 0;
3278                 goto cleanup;
3279         }
3280
3281         CDEBUG(D_NET, "Received %s with %d credits\n",
3282                       mxlnd_msgtype_to_str(type), msg->mxm_credits);
3283
3284         LASSERT(peer != NULL && conn != NULL);
3285         if (msg->mxm_credits != 0) {
3286                 spin_lock(&conn->mxk_lock);
3287                 if (msg->mxm_srcstamp == conn->mxk_incarnation) {
3288                         if ((conn->mxk_credits + msg->mxm_credits) >
3289                              *kmxlnd_tunables.kmx_peercredits) {
3290                                 CNETERR("mxk_credits %d  mxm_credits %d\n",
3291                                         conn->mxk_credits, msg->mxm_credits);
3292                         }
3293                         conn->mxk_credits += msg->mxm_credits;
3294                         LASSERT(conn->mxk_credits >= 0);
3295                         LASSERT(conn->mxk_credits <= *kmxlnd_tunables.kmx_peercredits);
3296                 }
3297                 spin_unlock(&conn->mxk_lock);
3298         }
3299
3300         CDEBUG(D_NET, "switch %s for rx (0x%llx)\n", mxlnd_msgtype_to_str(type), seq);
3301         switch (type) {
3302         case MXLND_MSG_NOOP:
3303                 break;
3304
3305         case MXLND_MSG_EAGER:
3306                 ret = lnet_parse(kmxlnd_data.kmx_ni, &msg->mxm_u.eager.mxem_hdr,
3307                                         msg->mxm_srcnid, rx, 0);
3308                 repost = ret < 0;
3309                 break;
3310
3311         case MXLND_MSG_PUT_REQ:
3312                 ret = lnet_parse(kmxlnd_data.kmx_ni, &msg->mxm_u.put_req.mxprm_hdr,
3313                                         msg->mxm_srcnid, rx, 1);
3314                 repost = ret < 0;
3315                 break;
3316
3317         case MXLND_MSG_PUT_ACK: {
3318                 u64  cookie = (u64) msg->mxm_u.put_ack.mxpam_dst_cookie;
3319                 if (cookie > MXLND_MAX_COOKIE) {
3320                         CNETERR("NAK for msg_type %d from %s\n", rx->mxc_msg_type,
3321                                 libcfs_nid2str(rx->mxc_nid));
3322                         result = -((u32) MXLND_ERROR_VAL(cookie));
3323                         lntmsg[0] = rx->mxc_lntmsg[0];
3324                 } else {
3325                         mxlnd_send_data(kmxlnd_data.kmx_ni, rx->mxc_lntmsg[0],
3326                                         rx->mxc_peer, MXLND_MSG_PUT_DATA,
3327                                         rx->mxc_msg->mxm_u.put_ack.mxpam_dst_cookie);
3328                 }
3329                 /* repost == 1 */
3330                 break;
3331         }
3332         case MXLND_MSG_GET_REQ:
3333                 ret = lnet_parse(kmxlnd_data.kmx_ni, &msg->mxm_u.get_req.mxgrm_hdr,
3334                                         msg->mxm_srcnid, rx, 1);
3335                 repost = ret < 0;
3336                 break;
3337
3338         default:
3339                 CNETERR("Bad MXLND message type %x from %s\n", msg->mxm_type,
3340                         libcfs_nid2str(rx->mxc_nid));
3341                 ret = -EPROTO;
3342                 break;
3343         }
3344
3345         if (ret < 0) {
3346                 CDEBUG(D_NET, "setting PEER_CONN_FAILED\n");
3347                 spin_lock(&conn->mxk_lock);
3348                 mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
3349                 spin_unlock(&conn->mxk_lock);
3350         }
3351
3352 cleanup:
3353         if (conn != NULL) {
3354                 spin_lock(&conn->mxk_lock);
3355                 conn->mxk_last_rx = cfs_time_current(); /* jiffies */
3356                 spin_unlock(&conn->mxk_lock);
3357         }
3358
3359         if (repost) {
3360                 /* lnet_parse() failed, etc., repost now */
3361                 mxlnd_put_idle_rx(rx);
3362                 if (conn != NULL && credit == 1) {
3363                         if (type == MXLND_MSG_PUT_DATA ||
3364                             type == MXLND_MSG_EAGER ||
3365                             type == MXLND_MSG_PUT_REQ ||
3366                             type == MXLND_MSG_NOOP) {
3367                                 spin_lock(&conn->mxk_lock);
3368                                 conn->mxk_outstanding++;
3369                                 spin_unlock(&conn->mxk_lock);
3370                         }
3371                 }
3372                 if (conn_ref) mxlnd_conn_decref(conn);
3373                 LASSERT(peer_ref == 0);
3374         }
3375
3376         if (type == MXLND_MSG_PUT_DATA || type == MXLND_MSG_GET_DATA) {
3377                 CDEBUG(D_NET, "leaving for rx (0x%llx)\n", bits);
3378         } else {
3379                 CDEBUG(D_NET, "leaving for rx (0x%llx)\n", seq);
3380         }
3381
3382         if (lntmsg[0] != NULL) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[0], result);
3383         if (lntmsg[1] != NULL) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[1], result);
3384
3385         if (conn != NULL && credit == 1) mxlnd_check_sends(peer);
3386
3387         return;
3388 }
3389
3390 void
3391 mxlnd_handle_connect_msg(kmx_peer_t *peer, u8 msg_type, mx_status_t status)
3392 {
3393         kmx_ctx_t       *tx     = NULL;
3394         kmx_msg_t       *txmsg  = NULL;
3395         kmx_conn_t      *conn   = peer->mxp_conn;
3396         u64             nic_id  = 0ULL;
3397         u32             ep_id   = 0;
3398         u32             sid     = 0;
3399         u8              type    = (msg_type == MXLND_MSG_ICON_REQ ?
3400                                    MXLND_MSG_CONN_REQ : MXLND_MSG_CONN_ACK);
3401
3402         /* a conn ref was taken when calling mx_iconnect(),
3403          * hold it until CONN_REQ or CONN_ACK completes */
3404
3405         CDEBUG(D_NET, "entering\n");
3406         if (status.code != MX_STATUS_SUCCESS) {
3407                 int send_bye    = (msg_type == MXLND_MSG_ICON_REQ ? 0 : 1);
3408
3409                 CNETERR("mx_iconnect() failed for %s with %s (%d) "
3410                         "to %s mxp_nid = 0x%llx mxp_nic_id = 0x%0llx mxp_ep_id = %d\n",
3411                         mxlnd_msgtype_to_str(msg_type),
3412                         mx_strstatus(status.code), status.code,
3413                         libcfs_nid2str(peer->mxp_nid),
3414                         peer->mxp_nid,
3415                         peer->mxp_nic_id,
3416                         peer->mxp_ep_id);
3417                 spin_lock(&conn->mxk_lock);
3418                 mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
3419                 spin_unlock(&conn->mxk_lock);
3420
3421                 if (cfs_time_after(jiffies, peer->mxp_reconnect_time +
3422                                    MXLND_CONNECT_TIMEOUT)) {
3423                         CNETERR("timeout, calling conn_disconnect()\n");
3424                         mxlnd_conn_disconnect(conn, 0, send_bye);
3425                 }
3426
3427                 mxlnd_conn_decref(conn);
3428                 return;
3429         }
3430         mx_decompose_endpoint_addr2(status.source, &nic_id, &ep_id, &sid);
3431         write_lock(&kmxlnd_data.kmx_global_lock);
3432         spin_lock(&conn->mxk_lock);
3433         conn->mxk_epa = status.source;
3434         mx_set_endpoint_addr_context(conn->mxk_epa, (void *) conn);
3435         if (msg_type == MXLND_MSG_ICON_ACK && likely(!peer->mxp_incompatible)) {
3436                 mxlnd_set_conn_status(conn, MXLND_CONN_READY);
3437         }
3438         spin_unlock(&conn->mxk_lock);
3439         write_unlock(&kmxlnd_data.kmx_global_lock);
3440
3441         /* mx_iconnect() succeeded, reset delay to 0 */
3442         write_lock(&kmxlnd_data.kmx_global_lock);
3443         peer->mxp_reconnect_time = 0;
3444         peer->mxp_conn->mxk_sid = sid;
3445         write_unlock(&kmxlnd_data.kmx_global_lock);
3446
3447         /* marshal CONN_REQ or CONN_ACK msg */
3448         /* we are still using the conn ref from iconnect() - do not take another */
3449         tx = mxlnd_get_idle_tx();
3450         if (tx == NULL) {
3451                 CNETERR("Can't obtain %s tx for %s\n",
3452                        mxlnd_msgtype_to_str(type),
3453                        libcfs_nid2str(peer->mxp_nid));
3454                 spin_lock(&conn->mxk_lock);
3455                 mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
3456                 spin_unlock(&conn->mxk_lock);
3457                 mxlnd_conn_decref(conn);
3458                 return;
3459         }
3460
3461         tx->mxc_peer = peer;
3462         tx->mxc_conn = conn;
3463         tx->mxc_deadline = jiffies + MXLND_CONNECT_TIMEOUT;
3464         CDEBUG(D_NET, "sending %s\n", mxlnd_msgtype_to_str(type));
3465         mxlnd_init_tx_msg (tx, type, sizeof(kmx_connreq_msg_t), peer->mxp_nid);
3466         txmsg = tx->mxc_msg;
3467         txmsg->mxm_u.conn_req.mxcrm_queue_depth = *kmxlnd_tunables.kmx_peercredits;
3468         txmsg->mxm_u.conn_req.mxcrm_eager_size = MXLND_MSG_SIZE;
3469         tx->mxc_match = mxlnd_create_match(tx, 0);
3470
3471         mxlnd_queue_tx(tx);
3472         return;
3473 }
3474
3475 /**
3476  * mxlnd_request_waitd - the MX request completion thread(s)
3477  * @arg - thread id (as a void *)
3478  *
3479  * This thread waits for a MX completion and then completes the request.
3480  * We will create one thread per CPU.
3481  */
3482 int
3483 mxlnd_request_waitd(void *arg)
3484 {
3485         long                    id              = (long) arg;
3486         __u32                   result          = 0;
3487         mx_return_t             mxret           = MX_SUCCESS;
3488         mx_status_t             status;
3489         kmx_ctx_t              *ctx             = NULL;
3490         enum kmx_req_state      req_type        = MXLND_REQ_TX;
3491         kmx_peer_t             *peer            = NULL;
3492         kmx_conn_t             *conn            = NULL;
3493 #if MXLND_POLLING
3494         int                     count           = 0;
3495 #endif
3496
3497         memset(&status, 0, sizeof(status));
3498
3499         CDEBUG(D_NET, "%s starting\n", name);
3500
3501         while (!(atomic_read(&kmxlnd_data.kmx_shutdown))) {
3502                 u8      msg_type        = 0;
3503
3504                 mxret = MX_SUCCESS;
3505                 result = 0;
3506 #if MXLND_POLLING
3507                 if (id == 0 && count++ < *kmxlnd_tunables.kmx_polling) {
3508                         mxret = mx_test_any(kmxlnd_data.kmx_endpt, 0ULL, 0ULL,
3509                                             &status, &result);
3510                 } else {
3511                         count = 0;
3512                         mxret = mx_wait_any(kmxlnd_data.kmx_endpt, MXLND_WAIT_TIMEOUT,
3513                                             0ULL, 0ULL, &status, &result);
3514                 }
3515 #else
3516                 mxret = mx_wait_any(kmxlnd_data.kmx_endpt, MXLND_WAIT_TIMEOUT,
3517                                     0ULL, 0ULL, &status, &result);
3518 #endif
3519                 if (unlikely(atomic_read(&kmxlnd_data.kmx_shutdown)))
3520                         break;
3521
3522                 if (result != 1) {
3523                         /* nothing completed... */
3524                         continue;
3525                 }
3526
3527                 CDEBUG(D_NET, "wait_any() returned with %s (%d) with "
3528                        "match_info 0x%llx and length %d\n",
3529                        mx_strstatus(status.code), status.code,
3530                        (u64) status.match_info, status.msg_length);
3531
3532                 if (status.code != MX_STATUS_SUCCESS) {
3533                         CNETERR("wait_any() failed with %s (%d) with "
3534                                 "match_info 0x%llx and length %d\n",
3535                                 mx_strstatus(status.code), status.code,
3536                                 (u64) status.match_info, status.msg_length);
3537                 }
3538
3539                 msg_type = MXLND_MSG_TYPE(status.match_info);
3540
3541                 /* This may be a mx_iconnect() request completing,
3542                  * check the bit mask for CONN_REQ and CONN_ACK */
3543                 if (msg_type == MXLND_MSG_ICON_REQ ||
3544                     msg_type == MXLND_MSG_ICON_ACK) {
3545                         peer = (kmx_peer_t*) status.context;
3546                         mxlnd_handle_connect_msg(peer, msg_type, status);
3547                         continue;
3548                 }
3549
3550                 /* This must be a tx or rx */
3551
3552                 /* NOTE: if this is a RX from the unexpected callback, it may
3553                  * have very little info. If we dropped it in unexpected_recv(),
3554                  * it will not have a context. If so, ignore it. */
3555                 ctx = (kmx_ctx_t *) status.context;
3556                 if (ctx != NULL) {
3557
3558                         req_type = ctx->mxc_type;
3559                         conn = ctx->mxc_conn; /* this may be NULL */
3560                         mxlnd_deq_pending_ctx(ctx);
3561
3562                         /* copy status to ctx->mxc_status */
3563                         ctx->mxc_status = status;
3564
3565                         switch (req_type) {
3566                         case MXLND_REQ_TX:
3567                                 mxlnd_handle_tx_completion(ctx);
3568                                 break;
3569                         case MXLND_REQ_RX:
3570                                 mxlnd_handle_rx_completion(ctx);
3571                                 break;
3572                         default:
3573                                 CNETERR("Unknown ctx type %d\n", req_type);
3574                                 LBUG();
3575                                 break;
3576                         }
3577
3578                         /* conn is always set except for the first CONN_REQ rx
3579                          * from a new peer */
3580                         if (status.code != MX_STATUS_SUCCESS && conn != NULL) {
3581                                 mxlnd_conn_disconnect(conn, 1, 1);
3582                         }
3583                 }
3584                 CDEBUG(D_NET, "waitd() completed task\n");
3585         }
3586         CDEBUG(D_NET, "%s stopping\n", name);
3587         mxlnd_thread_stop(id);
3588         return 0;
3589 }
3590
3591
3592 unsigned long
3593 mxlnd_check_timeouts(unsigned long now)
3594 {
3595         int             i               = 0;
3596         int             disconnect      = 0;
3597         unsigned long   next            = 0; /* jiffies */
3598         kmx_peer_t      *peer           = NULL;
3599         kmx_conn_t      *conn           = NULL;
3600         rwlock_t        *g_lock         = &kmxlnd_data.kmx_global_lock;
3601
3602         read_lock(g_lock);
3603         for (i = 0; i < MXLND_HASH_SIZE; i++) {
3604                 cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],
3605                                         mxp_list) {
3606
3607                         if (unlikely(atomic_read(&kmxlnd_data.kmx_shutdown))) {
3608                                 read_unlock(g_lock);
3609                                 return next;
3610                         }
3611
3612                         conn = peer->mxp_conn;
3613                         if (conn) {
3614                                 mxlnd_conn_addref(conn);
3615                         } else {
3616                                 continue;
3617                         }
3618
3619                         spin_lock(&conn->mxk_lock);
3620
3621                         /* if nothing pending (timeout == 0) or
3622                          * if conn is already disconnected,
3623                          * skip this conn */
3624                         if (conn->mxk_timeout == 0 ||
3625                             conn->mxk_status == MXLND_CONN_DISCONNECT) {
3626                                 spin_unlock(&conn->mxk_lock);
3627                                 mxlnd_conn_decref(conn);
3628                                 continue;
3629                         }
3630
3631                         /* we want to find the timeout that will occur first.
3632                          * if it is in the future, we will sleep until then.
3633                          * if it is in the past, then we will sleep one
3634                          * second and repeat the process. */
3635                         if ((next == 0) ||
3636                             (cfs_time_before(conn->mxk_timeout, next))) {
3637                                 next = conn->mxk_timeout;
3638                         }
3639
3640                         disconnect = 0;
3641
3642                         if (cfs_time_aftereq(now, conn->mxk_timeout))
3643                                 disconnect = 1;
3644                         spin_unlock(&conn->mxk_lock);
3645
3646                         if (disconnect)
3647                                 mxlnd_conn_disconnect(conn, 1, 1);
3648                         mxlnd_conn_decref(conn);
3649                 }
3650         }
3651         read_unlock(g_lock);
3652         if (next == 0)
3653                 next = now + MXLND_COMM_TIMEOUT;
3654
3655         return next;
3656 }
3657
3658 void
3659 mxlnd_passive_connect(kmx_connparams_t *cp)
3660 {
3661         int             ret             = 0;
3662         int             incompatible    = 0;
3663         u64             nic_id          = 0ULL;
3664         u32             ep_id           = 0;
3665         u32             sid             = 0;
3666         int             conn_ref        = 0;
3667         kmx_msg_t       *msg            = &cp->mxr_msg;
3668         kmx_peer_t      *peer           = cp->mxr_peer;
3669         kmx_conn_t      *conn           = NULL;
3670         rwlock_t        *g_lock         = &kmxlnd_data.kmx_global_lock;
3671
3672         mx_decompose_endpoint_addr2(cp->mxr_epa, &nic_id, &ep_id, &sid);
3673
3674         ret = mxlnd_unpack_msg(msg, cp->mxr_nob);
3675         if (ret != 0) {
3676                 if (peer) {
3677                         CNETERR("Error %d unpacking CONN_REQ from %s\n",
3678                                ret, libcfs_nid2str(peer->mxp_nid));
3679                 } else {
3680                         CNETERR("Error %d unpacking CONN_REQ from "
3681                                "unknown host with nic_id 0x%llx\n", ret, nic_id);
3682                 }
3683                 goto cleanup;
3684         }
3685         if (kmxlnd_data.kmx_ni->ni_nid != msg->mxm_dstnid) {
3686                 CNETERR("Can't accept %s: bad dst nid %s\n",
3687                                 libcfs_nid2str(msg->mxm_srcnid),
3688                                 libcfs_nid2str(msg->mxm_dstnid));
3689                 goto cleanup;
3690         }
3691         if (msg->mxm_u.conn_req.mxcrm_queue_depth != *kmxlnd_tunables.kmx_peercredits) {
3692                 CNETERR("Can't accept %s: incompatible queue depth "
3693                             "%d (%d wanted)\n",
3694                                 libcfs_nid2str(msg->mxm_srcnid),
3695                                 msg->mxm_u.conn_req.mxcrm_queue_depth,
3696                                 *kmxlnd_tunables.kmx_peercredits);
3697                 incompatible = 1;
3698         }
3699         if (msg->mxm_u.conn_req.mxcrm_eager_size != MXLND_MSG_SIZE) {
3700                 CNETERR("Can't accept %s: incompatible EAGER size "
3701                             "%d (%d wanted)\n",
3702                                 libcfs_nid2str(msg->mxm_srcnid),
3703                                 msg->mxm_u.conn_req.mxcrm_eager_size,
3704                                 (int) MXLND_MSG_SIZE);
3705                 incompatible = 1;
3706         }
3707
3708         if (peer == NULL) {
3709                 peer = mxlnd_find_peer_by_nid(msg->mxm_srcnid, 0); /* adds peer ref */
3710                 if (peer == NULL) {
3711                         int             hash    = 0;
3712                         u32             board   = 0;
3713                         kmx_peer_t      *existing_peer    = NULL;
3714
3715                         hash = mxlnd_nid_to_hash(msg->mxm_srcnid);
3716
3717                         mx_nic_id_to_board_number(nic_id, &board);
3718
3719                         /* adds conn ref for peer and one for this function */
3720                         ret = mxlnd_peer_alloc(&peer, msg->mxm_srcnid,
3721                                                board, ep_id, 0ULL);
3722                         if (ret != 0) {
3723                                 goto cleanup;
3724                         }
3725                         peer->mxp_conn->mxk_sid = sid;
3726                         LASSERT(peer->mxp_ep_id == ep_id);
3727                         write_lock(g_lock);
3728                         existing_peer = mxlnd_find_peer_by_nid_locked(msg->mxm_srcnid);
3729                         if (existing_peer) {
3730                                 mxlnd_conn_decref(peer->mxp_conn);
3731                                 mxlnd_peer_decref(peer);
3732                                 peer = existing_peer;
3733                                 mxlnd_conn_addref(peer->mxp_conn);
3734                                 conn = peer->mxp_conn;
3735                         } else {
3736                                 cfs_list_add_tail(&peer->mxp_list,
3737                                                   &kmxlnd_data.kmx_peers[hash]);
3738                                 atomic_inc(&kmxlnd_data.kmx_npeers);
3739                         }
3740                         write_unlock(g_lock);
3741                 } else {
3742                         ret = mxlnd_conn_alloc(&conn, peer); /* adds 2nd ref */
3743                         write_lock(g_lock);
3744                         mxlnd_peer_decref(peer); /* drop ref taken above */
3745                         write_unlock(g_lock);
3746                         if (ret != 0) {
3747                                 CNETERR("Cannot allocate mxp_conn\n");
3748                                 goto cleanup;
3749                         }
3750                 }
3751                 conn_ref = 1; /* peer/conn_alloc() added ref for this function */
3752                 conn = peer->mxp_conn;
3753         } else { /* unexpected handler found peer */
3754                 kmx_conn_t      *old_conn       = peer->mxp_conn;
3755
3756                 if (sid != peer->mxp_conn->mxk_sid) {
3757                         /* do not call mx_disconnect() or send a BYE */
3758                         mxlnd_conn_disconnect(old_conn, 0, 0);
3759
3760                         /* This allocs a conn, points peer->mxp_conn to this one.
3761                         * The old conn is still on the peer->mxp_conns list.
3762                         * As the pending requests complete, they will call
3763                         * conn_decref() which will eventually free it. */
3764                         ret = mxlnd_conn_alloc(&conn, peer);
3765                         if (ret != 0) {
3766                                 CNETERR("Cannot allocate peer->mxp_conn\n");
3767                                 goto cleanup;
3768                         }
3769                         /* conn_alloc() adds one ref for the peer and one
3770                          * for this function */
3771                         conn_ref = 1;
3772
3773                         peer->mxp_conn->mxk_sid = sid;
3774                 } else {
3775                         /* same sid */
3776                         conn = peer->mxp_conn;
3777                 }
3778         }
3779         write_lock(g_lock);
3780         peer->mxp_incompatible = incompatible;
3781         write_unlock(g_lock);
3782         spin_lock(&conn->mxk_lock);
3783         conn->mxk_incarnation = msg->mxm_srcstamp;
3784         mxlnd_set_conn_status(conn, MXLND_CONN_WAIT);
3785         spin_unlock(&conn->mxk_lock);
3786
3787         /* handle_conn_ack() will create the CONN_ACK msg */
3788         mxlnd_iconnect(peer, (u8) MXLND_MSG_ICON_ACK);
3789
3790 cleanup:
3791         if (conn_ref) mxlnd_conn_decref(conn);
3792
3793         mxlnd_connparams_free(cp);
3794         return;
3795 }
3796
3797 void
3798 mxlnd_check_conn_ack(kmx_connparams_t *cp)
3799 {
3800         int             ret             = 0;
3801         int             incompatible    = 0;
3802         u64             nic_id          = 0ULL;
3803         u32             ep_id           = 0;
3804         u32             sid             = 0;
3805         kmx_msg_t       *msg            = &cp->mxr_msg;
3806         kmx_peer_t      *peer           = cp->mxr_peer;
3807         kmx_conn_t      *conn           = cp->mxr_conn;
3808
3809         mx_decompose_endpoint_addr2(cp->mxr_epa, &nic_id, &ep_id, &sid);
3810
3811         ret = mxlnd_unpack_msg(msg, cp->mxr_nob);
3812         if (ret != 0) {
3813                 if (peer) {
3814                         CNETERR("Error %d unpacking CONN_ACK from %s\n",
3815                                ret, libcfs_nid2str(peer->mxp_nid));
3816                 } else {
3817                         CNETERR("Error %d unpacking CONN_ACK from "
3818                                "unknown host with nic_id 0x%llx\n", ret, nic_id);
3819                 }
3820                 ret = -1;
3821                 incompatible = 1;
3822                 goto failed;
3823         }
3824         if (kmxlnd_data.kmx_ni->ni_nid != msg->mxm_dstnid) {
3825                 CNETERR("Can't accept CONN_ACK from %s: "
3826                        "bad dst nid %s\n", libcfs_nid2str(msg->mxm_srcnid),
3827                         libcfs_nid2str(msg->mxm_dstnid));
3828                 ret = -1;
3829                 goto failed;
3830         }
3831         if (msg->mxm_u.conn_req.mxcrm_queue_depth != *kmxlnd_tunables.kmx_peercredits) {
3832                 CNETERR("Can't accept CONN_ACK from %s: "
3833                        "incompatible queue depth %d (%d wanted)\n",
3834                         libcfs_nid2str(msg->mxm_srcnid),
3835                         msg->mxm_u.conn_req.mxcrm_queue_depth,
3836                         *kmxlnd_tunables.kmx_peercredits);
3837                 incompatible = 1;
3838                 ret = -1;
3839                 goto failed;
3840         }
3841         if (msg->mxm_u.conn_req.mxcrm_eager_size != MXLND_MSG_SIZE) {
3842                 CNETERR("Can't accept CONN_ACK from %s: "
3843                         "incompatible EAGER size %d (%d wanted)\n",
3844                         libcfs_nid2str(msg->mxm_srcnid),
3845                         msg->mxm_u.conn_req.mxcrm_eager_size,
3846                         (int) MXLND_MSG_SIZE);
3847                 incompatible = 1;
3848                 ret = -1;
3849                 goto failed;
3850         }
3851         write_lock(&kmxlnd_data.kmx_global_lock);
3852         peer->mxp_incompatible = incompatible;
3853         write_unlock(&kmxlnd_data.kmx_global_lock);
3854         spin_lock(&conn->mxk_lock);
3855         conn->mxk_credits = *kmxlnd_tunables.kmx_peercredits;
3856         conn->mxk_outstanding = 0;
3857         conn->mxk_incarnation = msg->mxm_srcstamp;
3858         conn->mxk_timeout = 0;
3859         if (!incompatible) {
3860                 CDEBUG(D_NET, "setting peer %s CONN_READY\n",
3861                        libcfs_nid2str(msg->mxm_srcnid));
3862                 mxlnd_set_conn_status(conn, MXLND_CONN_READY);
3863         }
3864         spin_unlock(&conn->mxk_lock);
3865
3866         if (!incompatible)
3867                 mxlnd_check_sends(peer);
3868
3869 failed:
3870         if (ret < 0) {
3871                 spin_lock(&conn->mxk_lock);
3872                 mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
3873                 spin_unlock(&conn->mxk_lock);
3874         }
3875
3876         if (incompatible) mxlnd_conn_disconnect(conn, 0, 0);
3877
3878         mxlnd_connparams_free(cp);
3879         return;
3880 }
3881
3882 int
3883 mxlnd_abort_msgs(void)
3884 {
3885         int                     count           = 0;
3886         cfs_list_t              *orphans        = &kmxlnd_data.kmx_orphan_msgs;
3887         spinlock_t              *g_conn_lock    = &kmxlnd_data.kmx_conn_lock;
3888
3889         /* abort orphans */
3890         spin_lock(g_conn_lock);
3891         while (!cfs_list_empty(orphans)) {
3892                 kmx_ctx_t       *ctx     = NULL;
3893                 kmx_conn_t      *conn   = NULL;
3894
3895                 ctx = cfs_list_entry(orphans->next, kmx_ctx_t, mxc_list);
3896                 cfs_list_del_init(&ctx->mxc_list);
3897                 spin_unlock(g_conn_lock);
3898
3899                 ctx->mxc_errno = -ECONNABORTED;
3900                 conn = ctx->mxc_conn;
3901                 CDEBUG(D_NET, "aborting %s %s %s\n",
3902                        mxlnd_msgtype_to_str(ctx->mxc_msg_type),
3903                        ctx->mxc_type == MXLND_REQ_TX ? "(TX) to" : "(RX) from",
3904                        libcfs_nid2str(ctx->mxc_nid));
3905                 if (ctx->mxc_type == MXLND_REQ_TX) {
3906                         mxlnd_put_idle_tx(ctx); /* do not hold any locks */
3907                         if (conn) mxlnd_conn_decref(conn); /* for this tx */
3908                 } else {
3909                         ctx->mxc_state = MXLND_CTX_CANCELED;
3910                         mxlnd_handle_rx_completion(ctx);
3911                 }
3912
3913                 count++;
3914                 spin_lock(g_conn_lock);
3915         }
3916         spin_unlock(g_conn_lock);
3917
3918         return count;
3919 }
3920
3921 int
3922 mxlnd_free_conn_zombies(void)
3923 {
3924         int             count           = 0;
3925         cfs_list_t      *zombies        = &kmxlnd_data.kmx_conn_zombies;
3926         spinlock_t      *g_conn_lock    = &kmxlnd_data.kmx_conn_lock;
3927         rwlock_t        *g_lock         = &kmxlnd_data.kmx_global_lock;
3928
3929         /* cleanup any zombies */
3930         spin_lock(g_conn_lock);
3931         while (!cfs_list_empty(zombies)) {
3932                 kmx_conn_t      *conn   = NULL;
3933
3934                 conn = cfs_list_entry(zombies->next, kmx_conn_t, mxk_zombie);
3935                 cfs_list_del_init(&conn->mxk_zombie);
3936                 spin_unlock(g_conn_lock);
3937
3938                 write_lock(g_lock);
3939                 mxlnd_conn_free_locked(conn);
3940                 write_unlock(g_lock);
3941
3942                 count++;
3943                 spin_lock(g_conn_lock);
3944         }
3945         spin_unlock(g_conn_lock);
3946         CDEBUG(D_NET, "%s: freed %d zombies\n", __func__, count);
3947         return count;
3948 }
3949
3950 /**
3951  * mxlnd_connd - handles incoming connection requests
3952  * @arg - thread id (as a void *)
3953  *
3954  * This thread handles incoming connection requests
3955  */
3956 int
3957 mxlnd_connd(void *arg)
3958 {
3959         long                    id              = (long) arg;
3960
3961         CDEBUG(D_NET, "connd starting\n");
3962
3963         while (!(atomic_read(&kmxlnd_data.kmx_shutdown))) {
3964                 int                ret             = 0;
3965                 kmx_connparams_t  *cp              = NULL;
3966                 spinlock_t        *g_conn_lock  = &kmxlnd_data.kmx_conn_lock;
3967                 cfs_list_t        *conn_reqs    = &kmxlnd_data.kmx_conn_reqs;
3968
3969                 ret = down_interruptible(&kmxlnd_data.kmx_conn_sem);
3970
3971                 if (atomic_read(&kmxlnd_data.kmx_shutdown))
3972                         break;
3973
3974                 if (ret != 0)
3975                         continue;
3976
3977                 ret = mxlnd_abort_msgs();
3978                 ret += mxlnd_free_conn_zombies();
3979
3980                 spin_lock(g_conn_lock);
3981                 if (cfs_list_empty(conn_reqs)) {
3982                         if (ret == 0)
3983                                 CNETERR("connd woke up but did not find a "
3984                                         "kmx_connparams_t or zombie conn\n");
3985                         spin_unlock(g_conn_lock);
3986                         continue;
3987                 }
3988                 cp = cfs_list_entry(conn_reqs->next, kmx_connparams_t,
3989                                     mxr_list);
3990                 cfs_list_del_init(&cp->mxr_list);
3991                 spin_unlock(g_conn_lock);
3992
3993                 switch (MXLND_MSG_TYPE(cp->mxr_match)) {
3994                 case MXLND_MSG_CONN_REQ:
3995                         /* We have a connection request. Handle it. */
3996                         mxlnd_passive_connect(cp);
3997                         break;
3998                 case MXLND_MSG_CONN_ACK:
3999                         /* The peer is ready for messages */
4000                         mxlnd_check_conn_ack(cp);
4001                         break;
4002                 }
4003         }
4004
4005         mxlnd_free_conn_zombies();
4006
4007         CDEBUG(D_NET, "connd stopping\n");
4008         mxlnd_thread_stop(id);
4009         return 0;
4010 }
4011
4012 /**
4013  * mxlnd_timeoutd - enforces timeouts on messages
4014  * @arg - thread id (as a void *)
4015  *
4016  * This thread queries each peer for its earliest timeout. If a peer has timed out,
4017  * it calls mxlnd_conn_disconnect().
4018  *
4019  * After checking for timeouts, try progressing sends (call check_sends()).
4020  */
4021 int
4022 mxlnd_timeoutd(void *arg)
4023 {
4024         int             i       = 0;
4025         long            id      = (long) arg;
4026         unsigned long   now     = 0;
4027         unsigned long   next    = 0;
4028         unsigned long   delay   = HZ;
4029         kmx_peer_t     *peer    = NULL;
4030         kmx_peer_t     *temp    = NULL;
4031         kmx_conn_t     *conn    = NULL;
4032         rwlock_t   *g_lock  = &kmxlnd_data.kmx_global_lock;
4033
4034         CDEBUG(D_NET, "timeoutd starting\n");
4035
4036         while (!(atomic_read(&kmxlnd_data.kmx_shutdown))) {
4037
4038                 now = jiffies;
4039                 /* if the next timeout has not arrived, go back to sleep */
4040                 if (cfs_time_after(now, next)) {
4041                         next = mxlnd_check_timeouts(now);
4042                 }
4043
4044                 /* try to progress peers' txs */
4045                 write_lock(g_lock);
4046                 for (i = 0; i < MXLND_HASH_SIZE; i++) {
4047                         cfs_list_t *peers = &kmxlnd_data.kmx_peers[i];
4048
4049                         /* NOTE we are safe against the removal of peer, but
4050                          * not against the removal of temp */
4051                         cfs_list_for_each_entry_safe(peer, temp, peers,
4052                                                      mxp_list) {
4053                                 if (atomic_read(&kmxlnd_data.kmx_shutdown))
4054                                         break;
4055                                 mxlnd_peer_addref(peer); /* add ref... */
4056                                 conn = peer->mxp_conn;
4057                                 if (conn && conn->mxk_status != MXLND_CONN_DISCONNECT) {
4058                                         mxlnd_conn_addref(conn); /* take ref... */
4059                                 } else {
4060                                         CDEBUG(D_NET, "ignoring %s\n",
4061                                                libcfs_nid2str(peer->mxp_nid));
4062                                         mxlnd_peer_decref(peer); /* ...to here */
4063                                         continue;
4064                                 }
4065
4066                                 if ((conn->mxk_status == MXLND_CONN_READY ||
4067                                     conn->mxk_status == MXLND_CONN_FAIL) &&
4068                                     cfs_time_after(now,
4069                                                    conn->mxk_last_tx +
4070                                                    HZ)) {
4071                                         write_unlock(g_lock);
4072                                         mxlnd_check_sends(peer);
4073                                         write_lock(g_lock);
4074                                 }
4075                                 mxlnd_conn_decref(conn); /* until here */
4076                                 mxlnd_peer_decref(peer); /* ...to here */
4077                         }
4078                 }
4079                 write_unlock(g_lock);
4080
4081                 mxlnd_sleep(delay);
4082         }
4083         CDEBUG(D_NET, "timeoutd stopping\n");
4084         mxlnd_thread_stop(id);
4085         return 0;
4086 }