lnet/klnds/mxlnd/mxlnd_cb.c

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