lnet/klnds/o2iblnd/o2iblnd_cb.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lnet/klnds/o2iblnd/o2iblnd_cb.c
  32  *
  33  * Author: Eric Barton <eric@bartonsoftware.com>
  34  */
  35
  36 #include "o2iblnd.h"
  37
  38 #define MAX_CONN_RACES_BEFORE_ABORT 20
  39
  40 static void kiblnd_peer_alive(struct kib_peer_ni *peer_ni);
  41 static void kiblnd_peer_connect_failed(struct kib_peer_ni *peer_ni, int active,
  42                                        int error);
  43 static struct ib_rdma_wr *
  44 kiblnd_init_tx_msg_payload(struct lnet_ni *ni, struct kib_tx *tx,
  45                                int type, int body_nob, int payload_nob);
  46 #define kiblnd_init_tx_msg(ni, tx, type, body) \
  47         kiblnd_init_tx_msg_payload(ni, tx, type, body, 0)
  48 static int kiblnd_init_rdma(struct kib_conn *conn, struct kib_tx *tx, int type,
  49                             int resid, struct kib_rdma_desc *dstrd, u64 dstcookie);
  50 static void kiblnd_queue_tx_locked(struct kib_tx *tx, struct kib_conn *conn);
  51 static void kiblnd_queue_tx(struct kib_tx *tx, struct kib_conn *conn);
  52
  53 static void kiblnd_unmap_tx(struct kib_tx *tx);
  54 static void kiblnd_check_sends_locked(struct kib_conn *conn);
  55
  56 static void
  57 kiblnd_tx_done(struct kib_tx *tx)
  58 {
  59         struct lnet_msg *lntmsg[2];
  60         int         rc;
  61         int         i;
  62
  63         LASSERT (!in_interrupt());
  64         LASSERT (!tx->tx_queued);               /* mustn't be queued for sending */
  65         LASSERT (tx->tx_sending == 0);          /* mustn't be awaiting sent callback */
  66         LASSERT (!tx->tx_waiting);              /* mustn't be awaiting peer_ni response */
  67         LASSERT (tx->tx_pool != NULL);
  68
  69         kiblnd_unmap_tx(tx);
  70
  71         /* tx may have up to 2 lnet msgs to finalise */
  72         lntmsg[0] = tx->tx_lntmsg[0]; tx->tx_lntmsg[0] = NULL;
  73         lntmsg[1] = tx->tx_lntmsg[1]; tx->tx_lntmsg[1] = NULL;
  74         rc = tx->tx_status;
  75
  76         if (tx->tx_conn != NULL) {
  77                 kiblnd_conn_decref(tx->tx_conn);
  78                 tx->tx_conn = NULL;
  79         }
  80
  81         tx->tx_nwrq = tx->tx_nsge = 0;
  82         tx->tx_status = 0;
  83
  84         kiblnd_pool_free_node(&tx->tx_pool->tpo_pool, &tx->tx_list);
  85
  86         /* delay finalize until my descs have been freed */
  87         for (i = 0; i < 2; i++) {
  88                 if (lntmsg[i] == NULL)
  89                         continue;
  90
  91                 /* propagate health status to LNet for requests */
  92                 if (i == 0 && lntmsg[i])
  93                         lntmsg[i]->msg_health_status = tx->tx_hstatus;
  94
  95                 lnet_finalize(lntmsg[i], rc);
  96         }
  97 }
  98
  99 void
 100 kiblnd_txlist_done(struct list_head *txlist, int status,
 101                    enum lnet_msg_hstatus hstatus)
 102 {
 103         struct kib_tx *tx;
 104
 105         while ((tx = list_first_entry_or_null(txlist,
 106                                               struct kib_tx,
 107                                               tx_list)) != NULL) {
 108                 list_del(&tx->tx_list);
 109                 /* complete now */
 110                 tx->tx_waiting = 0;
 111                 tx->tx_status = status;
 112                 if (hstatus != LNET_MSG_STATUS_OK)
 113                         tx->tx_hstatus = hstatus;
 114                 kiblnd_tx_done(tx);
 115         }
 116 }
 117
 118 static struct kib_tx *
 119 kiblnd_get_idle_tx(struct lnet_ni *ni, lnet_nid_t target)
 120 {
 121         struct kib_net *net = ni->ni_data;
 122         struct list_head *node;
 123         struct kib_tx *tx;
 124         struct kib_tx_poolset *tps;
 125
 126         tps = net->ibn_tx_ps[lnet_cpt_of_nid(target, ni)];
 127         node = kiblnd_pool_alloc_node(&tps->tps_poolset);
 128         if (node == NULL)
 129                 return NULL;
 130         tx = container_of(node, struct kib_tx, tx_list);
 131
 132         LASSERT (tx->tx_nwrq == 0);
 133         LASSERT (!tx->tx_queued);
 134         LASSERT (tx->tx_sending == 0);
 135         LASSERT (!tx->tx_waiting);
 136         LASSERT (tx->tx_status == 0);
 137         LASSERT (tx->tx_conn == NULL);
 138         LASSERT (tx->tx_lntmsg[0] == NULL);
 139         LASSERT (tx->tx_lntmsg[1] == NULL);
 140         LASSERT (tx->tx_nfrags == 0);
 141
 142         tx->tx_gpu = 0;
 143         tx->tx_gaps = false;
 144         tx->tx_hstatus = LNET_MSG_STATUS_OK;
 145
 146         return tx;
 147 }
 148
 149 static void
 150 kiblnd_drop_rx(struct kib_rx *rx)
 151 {
 152         struct kib_conn *conn = rx->rx_conn;
 153         struct kib_sched_info *sched = conn->ibc_sched;
 154         unsigned long flags;
 155
 156         spin_lock_irqsave(&sched->ibs_lock, flags);
 157         LASSERT(conn->ibc_nrx > 0);
 158         conn->ibc_nrx--;
 159         spin_unlock_irqrestore(&sched->ibs_lock, flags);
 160
 161         kiblnd_conn_decref(conn);
 162 }
 163
 164 int
 165 kiblnd_post_rx(struct kib_rx *rx, int credit)
 166 {
 167         struct kib_conn *conn = rx->rx_conn;
 168         struct kib_net *net = conn->ibc_peer->ibp_ni->ni_data;
 169         struct ib_recv_wr *bad_wrq = NULL;
 170 #ifdef HAVE_OFED_IB_GET_DMA_MR
 171         struct ib_mr *mr = conn->ibc_hdev->ibh_mrs;
 172 #endif
 173         int rc;
 174
 175         LASSERT (net != NULL);
 176         LASSERT (!in_interrupt());
 177         LASSERT (credit == IBLND_POSTRX_NO_CREDIT ||
 178                  credit == IBLND_POSTRX_PEER_CREDIT ||
 179                  credit == IBLND_POSTRX_RSRVD_CREDIT);
 180 #ifdef HAVE_OFED_IB_GET_DMA_MR
 181         LASSERT(mr != NULL);
 182
 183         rx->rx_sge.lkey   = mr->lkey;
 184 #else
 185         rx->rx_sge.lkey   = conn->ibc_hdev->ibh_pd->local_dma_lkey;
 186 #endif
 187         rx->rx_sge.addr   = rx->rx_msgaddr;
 188         rx->rx_sge.length = IBLND_MSG_SIZE;
 189
 190         rx->rx_wrq.next = NULL;
 191         rx->rx_wrq.sg_list = &rx->rx_sge;
 192         rx->rx_wrq.num_sge = 1;
 193         rx->rx_wrq.wr_id = kiblnd_ptr2wreqid(rx, IBLND_WID_RX);
 194
 195         LASSERT (conn->ibc_state >= IBLND_CONN_INIT);
 196         LASSERT (rx->rx_nob >= 0);              /* not posted */
 197
 198         if (conn->ibc_state > IBLND_CONN_ESTABLISHED) {
 199                 kiblnd_drop_rx(rx);             /* No more posts for this rx */
 200                 return 0;
 201         }
 202
 203         rx->rx_nob = -1;                        /* flag posted */
 204
 205         /* NB: need an extra reference after ib_post_recv because we don't
 206          * own this rx (and rx::rx_conn) anymore, LU-5678.
 207          */
 208         kiblnd_conn_addref(conn);
 209 #ifdef HAVE_OFED_IB_POST_SEND_RECV_CONST
 210         rc = ib_post_recv(conn->ibc_cmid->qp, &rx->rx_wrq,
 211                           (const struct ib_recv_wr **)&bad_wrq);
 212 #else
 213         rc = ib_post_recv(conn->ibc_cmid->qp, &rx->rx_wrq, &bad_wrq);
 214 #endif
 215         if (unlikely(rc != 0)) {
 216                 CERROR("Can't post rx for %s: %d, bad_wrq: %p\n",
 217                        libcfs_nid2str(conn->ibc_peer->ibp_nid), rc, bad_wrq);
 218                 rx->rx_nob = 0;
 219         }
 220
 221         if (conn->ibc_state < IBLND_CONN_ESTABLISHED) /* Initial post */
 222                 goto out;
 223
 224         if (unlikely(rc != 0)) {
 225                 kiblnd_close_conn(conn, rc);
 226                 kiblnd_drop_rx(rx);     /* No more posts for this rx */
 227                 goto out;
 228         }
 229
 230         if (credit == IBLND_POSTRX_NO_CREDIT)
 231                 goto out;
 232
 233         spin_lock(&conn->ibc_lock);
 234         if (credit == IBLND_POSTRX_PEER_CREDIT)
 235                 conn->ibc_outstanding_credits++;
 236         else
 237                 conn->ibc_reserved_credits++;
 238         kiblnd_check_sends_locked(conn);
 239         spin_unlock(&conn->ibc_lock);
 240
 241 out:
 242         kiblnd_conn_decref(conn);
 243         return rc;
 244 }
 245
 246 static struct kib_tx *
 247 kiblnd_find_waiting_tx_locked(struct kib_conn *conn, int txtype, u64 cookie)
 248 {
 249         struct kib_tx *tx;
 250
 251         list_for_each_entry(tx, &conn->ibc_active_txs, tx_list) {
 252                 LASSERT(!tx->tx_queued);
 253                 LASSERT(tx->tx_sending != 0 || tx->tx_waiting);
 254
 255                 if (tx->tx_cookie != cookie)
 256                         continue;
 257
 258                 if (tx->tx_waiting &&
 259                     tx->tx_msg->ibm_type == txtype)
 260                         return tx;
 261
 262                 CWARN("Bad completion: %swaiting, type %x (wanted %x)\n",
 263                       tx->tx_waiting ? "" : "NOT ",
 264                       tx->tx_msg->ibm_type, txtype);
 265         }
 266         return NULL;
 267 }
 268
 269 static void
 270 kiblnd_handle_completion(struct kib_conn *conn, int txtype, int status, u64 cookie)
 271 {
 272         struct kib_tx *tx;
 273         struct lnet_ni *ni = conn->ibc_peer->ibp_ni;
 274         int idle;
 275
 276         spin_lock(&conn->ibc_lock);
 277
 278         tx = kiblnd_find_waiting_tx_locked(conn, txtype, cookie);
 279         if (tx == NULL) {
 280                 spin_unlock(&conn->ibc_lock);
 281
 282                 CWARN("Unmatched completion type %x cookie %#llx from %s\n",
 283                       txtype, cookie, libcfs_nid2str(conn->ibc_peer->ibp_nid));
 284                 kiblnd_close_conn(conn, -EPROTO);
 285                 return;
 286         }
 287
 288         if (tx->tx_status == 0) {               /* success so far */
 289                 if (status < 0) {               /* failed? */
 290                         if (status == -ECONNABORTED) {
 291                                 CDEBUG(D_NET, "bad status for connection to %s "
 292                                               "with completion type %x\n",
 293                                        libcfs_nid2str(conn->ibc_peer->ibp_nid),
 294                                        txtype);
 295                         }
 296
 297                         tx->tx_status = status;
 298                         tx->tx_hstatus = LNET_MSG_STATUS_REMOTE_ERROR;
 299                 } else if (txtype == IBLND_MSG_GET_REQ) {
 300                         lnet_set_reply_msg_len(ni, tx->tx_lntmsg[1], status);
 301                 }
 302         }
 303
 304         tx->tx_waiting = 0;
 305
 306         idle = !tx->tx_queued && (tx->tx_sending == 0);
 307         if (idle)
 308                 list_del(&tx->tx_list);
 309
 310         spin_unlock(&conn->ibc_lock);
 311
 312         if (idle)
 313                 kiblnd_tx_done(tx);
 314 }
 315
 316 static void
 317 kiblnd_send_completion(struct kib_conn *conn, int type, int status, u64 cookie)
 318 {
 319         struct lnet_ni *ni = conn->ibc_peer->ibp_ni;
 320         struct kib_tx *tx = kiblnd_get_idle_tx(ni, conn->ibc_peer->ibp_nid);
 321
 322         if (tx == NULL) {
 323                 CERROR("Can't get tx for completion %x for %s\n",
 324                        type, libcfs_nid2str(conn->ibc_peer->ibp_nid));
 325                 return;
 326         }
 327
 328         tx->tx_msg->ibm_u.completion.ibcm_status = status;
 329         tx->tx_msg->ibm_u.completion.ibcm_cookie = cookie;
 330         kiblnd_init_tx_msg(ni, tx, type, sizeof(struct kib_completion_msg));
 331
 332         kiblnd_queue_tx(tx, conn);
 333 }
 334
 335 static void
 336 kiblnd_handle_rx(struct kib_rx *rx)
 337 {
 338         struct kib_msg *msg = rx->rx_msg;
 339         struct kib_conn *conn = rx->rx_conn;
 340         struct lnet_ni *ni = conn->ibc_peer->ibp_ni;
 341         int credits = msg->ibm_credits;
 342         struct kib_tx *tx;
 343         int rc = 0;
 344         int rc2;
 345         int post_credit;
 346         struct lnet_hdr hdr;
 347         struct lnet_nid srcnid;
 348
 349         LASSERT (conn->ibc_state >= IBLND_CONN_ESTABLISHED);
 350
 351         CDEBUG(D_NET, "Received %x[%d] nob %u cm_id %p qp_num 0x%x\n",
 352                 msg->ibm_type, credits,
 353                 msg->ibm_nob,
 354                 conn->ibc_cmid,
 355                 conn->ibc_cmid->qp ? conn->ibc_cmid->qp->qp_num : 0);
 356         kiblnd_dump_conn_dbg(conn);
 357
 358         if (credits != 0) {
 359                 /* Have I received credits that will let me send? */
 360                 spin_lock(&conn->ibc_lock);
 361
 362                 if (conn->ibc_credits + credits >
 363                     conn->ibc_queue_depth) {
 364                         rc2 = conn->ibc_credits;
 365                         spin_unlock(&conn->ibc_lock);
 366
 367                         CERROR("Bad credits from %s: %d + %d > %d\n",
 368                                libcfs_nid2str(conn->ibc_peer->ibp_nid),
 369                                rc2, credits,
 370                                conn->ibc_queue_depth);
 371
 372                         kiblnd_close_conn(conn, -EPROTO);
 373                         kiblnd_post_rx(rx, IBLND_POSTRX_NO_CREDIT);
 374                         return;
 375                 }
 376
 377                 conn->ibc_credits += credits;
 378
 379                 /* This ensures the credit taken by NOOP can be returned */
 380                 if (msg->ibm_type == IBLND_MSG_NOOP &&
 381                     !IBLND_OOB_CAPABLE(conn->ibc_version)) /* v1 only */
 382                         conn->ibc_outstanding_credits++;
 383
 384                 kiblnd_check_sends_locked(conn);
 385                 spin_unlock(&conn->ibc_lock);
 386         }
 387
 388         switch (msg->ibm_type) {
 389         default:
 390                 CERROR("Bad IBLND message type %x from %s\n",
 391                        msg->ibm_type, libcfs_nid2str(conn->ibc_peer->ibp_nid));
 392                 post_credit = IBLND_POSTRX_NO_CREDIT;
 393                 rc = -EPROTO;
 394                 break;
 395
 396         case IBLND_MSG_NOOP:
 397                 if (IBLND_OOB_CAPABLE(conn->ibc_version)) {
 398                         post_credit = IBLND_POSTRX_NO_CREDIT;
 399                         break;
 400                 }
 401
 402                 if (credits != 0) /* credit already posted */
 403                         post_credit = IBLND_POSTRX_NO_CREDIT;
 404                 else              /* a keepalive NOOP */
 405                         post_credit = IBLND_POSTRX_PEER_CREDIT;
 406                 break;
 407
 408         case IBLND_MSG_IMMEDIATE:
 409                 post_credit = IBLND_POSTRX_DONT_POST;
 410                 lnet_hdr_from_nid4(&hdr, &msg->ibm_u.immediate.ibim_hdr);
 411                 lnet_nid4_to_nid(msg->ibm_srcnid, &srcnid);
 412                 rc = lnet_parse(ni, &hdr, &srcnid, rx, 0);
 413                 if (rc < 0)                     /* repost on error */
 414                         post_credit = IBLND_POSTRX_PEER_CREDIT;
 415                 break;
 416
 417         case IBLND_MSG_PUT_REQ:
 418                 post_credit = IBLND_POSTRX_DONT_POST;
 419                 lnet_hdr_from_nid4(&hdr, &msg->ibm_u.putreq.ibprm_hdr);
 420                 lnet_nid4_to_nid(msg->ibm_srcnid, &srcnid);
 421                 rc = lnet_parse(ni, &hdr, &srcnid, rx, 1);
 422                 if (rc < 0)                     /* repost on error */
 423                         post_credit = IBLND_POSTRX_PEER_CREDIT;
 424                 break;
 425
 426         case IBLND_MSG_PUT_NAK:
 427                 CWARN ("PUT_NACK from %s\n",
 428                        libcfs_nid2str(conn->ibc_peer->ibp_nid));
 429                 post_credit = IBLND_POSTRX_RSRVD_CREDIT;
 430                 kiblnd_handle_completion(conn, IBLND_MSG_PUT_REQ,
 431                                          msg->ibm_u.completion.ibcm_status,
 432                                          msg->ibm_u.completion.ibcm_cookie);
 433                 break;
 434
 435         case IBLND_MSG_PUT_ACK:
 436                 post_credit = IBLND_POSTRX_RSRVD_CREDIT;
 437
 438                 spin_lock(&conn->ibc_lock);
 439                 tx = kiblnd_find_waiting_tx_locked(conn, IBLND_MSG_PUT_REQ,
 440                                         msg->ibm_u.putack.ibpam_src_cookie);
 441                 if (tx != NULL)
 442                         list_del(&tx->tx_list);
 443                 spin_unlock(&conn->ibc_lock);
 444
 445                 if (tx == NULL) {
 446                         CERROR("Unmatched PUT_ACK from %s\n",
 447                                libcfs_nid2str(conn->ibc_peer->ibp_nid));
 448                         rc = -EPROTO;
 449                         break;
 450                 }
 451
 452                 LASSERT (tx->tx_waiting);
 453                 /* CAVEAT EMPTOR: I could be racing with tx_complete, but...
 454                  * (a) I can overwrite tx_msg since my peer_ni has received it!
 455                  * (b) tx_waiting set tells tx_complete() it's not done. */
 456
 457                 tx->tx_nwrq = tx->tx_nsge = 0;  /* overwrite PUT_REQ */
 458
 459                 rc2 = kiblnd_init_rdma(conn, tx, IBLND_MSG_PUT_DONE,
 460                                        kiblnd_rd_size(&msg->ibm_u.putack.ibpam_rd),
 461                                        &msg->ibm_u.putack.ibpam_rd,
 462                                        msg->ibm_u.putack.ibpam_dst_cookie);
 463                 if (rc2 < 0)
 464                         CERROR("Can't setup rdma for PUT to %s: %d\n",
 465                                libcfs_nid2str(conn->ibc_peer->ibp_nid), rc2);
 466
 467                 spin_lock(&conn->ibc_lock);
 468                 tx->tx_waiting = 0;     /* clear waiting and queue atomically */
 469                 kiblnd_queue_tx_locked(tx, conn);
 470                 spin_unlock(&conn->ibc_lock);
 471                 break;
 472
 473         case IBLND_MSG_PUT_DONE:
 474                 post_credit = IBLND_POSTRX_PEER_CREDIT;
 475                 kiblnd_handle_completion(conn, IBLND_MSG_PUT_ACK,
 476                                          msg->ibm_u.completion.ibcm_status,
 477                                          msg->ibm_u.completion.ibcm_cookie);
 478                 break;
 479
 480         case IBLND_MSG_GET_REQ:
 481                 post_credit = IBLND_POSTRX_DONT_POST;
 482                 lnet_hdr_from_nid4(&hdr, &msg->ibm_u.get.ibgm_hdr);
 483                 lnet_nid4_to_nid(msg->ibm_srcnid, &srcnid);
 484                 rc = lnet_parse(ni, &hdr, &srcnid, rx, 1);
 485                 if (rc < 0)                     /* repost on error */
 486                         post_credit = IBLND_POSTRX_PEER_CREDIT;
 487                 break;
 488
 489         case IBLND_MSG_GET_DONE:
 490                 post_credit = IBLND_POSTRX_RSRVD_CREDIT;
 491                 kiblnd_handle_completion(conn, IBLND_MSG_GET_REQ,
 492                                          msg->ibm_u.completion.ibcm_status,
 493                                          msg->ibm_u.completion.ibcm_cookie);
 494                 break;
 495         }
 496
 497         if (rc < 0)                             /* protocol error */
 498                 kiblnd_close_conn(conn, rc);
 499
 500         if (post_credit != IBLND_POSTRX_DONT_POST)
 501                 kiblnd_post_rx(rx, post_credit);
 502 }
 503
 504 static void
 505 kiblnd_rx_complete(struct kib_rx *rx, int status, int nob)
 506 {
 507         struct kib_msg *msg = rx->rx_msg;
 508         struct kib_conn   *conn = rx->rx_conn;
 509         struct lnet_ni *ni = conn->ibc_peer->ibp_ni;
 510         struct kib_net *net = ni->ni_data;
 511         int rc;
 512         int err = -EIO;
 513
 514         LASSERT(net);
 515         LASSERT(rx->rx_nob < 0);        /* was posted */
 516         rx->rx_nob = 0;                 /* isn't now */
 517
 518         if (conn->ibc_state > IBLND_CONN_ESTABLISHED)
 519                 goto ignore;
 520
 521         if (status != IB_WC_SUCCESS) {
 522                 CNETERR("Rx from %s failed: %d\n",
 523                         libcfs_nid2str(conn->ibc_peer->ibp_nid), status);
 524                 goto failed;
 525         }
 526
 527         LASSERT(nob >= 0);
 528         rx->rx_nob = nob;
 529
 530         rc = kiblnd_unpack_msg(msg, rx->rx_nob);
 531         if (rc != 0) {
 532                 CERROR("Error %d unpacking rx from %s\n",
 533                        rc, libcfs_nid2str(conn->ibc_peer->ibp_nid));
 534                 goto failed;
 535         }
 536
 537         if (msg->ibm_srcnid != conn->ibc_peer->ibp_nid ||
 538             msg->ibm_dstnid != lnet_nid_to_nid4(&ni->ni_nid) ||
 539             msg->ibm_srcstamp != conn->ibc_incarnation ||
 540             msg->ibm_dststamp != net->ibn_incarnation) {
 541                 CERROR("Stale rx from %s\n",
 542                        libcfs_nid2str(conn->ibc_peer->ibp_nid));
 543                 err = -ESTALE;
 544                 goto failed;
 545         }
 546
 547         /* set time last known alive */
 548         kiblnd_peer_alive(conn->ibc_peer);
 549
 550         /* racing with connection establishment/teardown! */
 551
 552         if (conn->ibc_state < IBLND_CONN_ESTABLISHED) {
 553                 rwlock_t  *g_lock = &kiblnd_data.kib_global_lock;
 554                 unsigned long  flags;
 555
 556                 write_lock_irqsave(g_lock, flags);
 557                 /* must check holding global lock to eliminate race */
 558                 if (conn->ibc_state < IBLND_CONN_ESTABLISHED) {
 559                         list_add_tail(&rx->rx_list, &conn->ibc_early_rxs);
 560                         write_unlock_irqrestore(g_lock, flags);
 561                         return;
 562                 }
 563                 write_unlock_irqrestore(g_lock, flags);
 564         }
 565         kiblnd_handle_rx(rx);
 566         return;
 567
 568 failed:
 569         CDEBUG(D_NET, "rx %p conn %p\n", rx, conn);
 570         kiblnd_close_conn(conn, err);
 571 ignore:
 572         kiblnd_drop_rx(rx);                     /* Don't re-post rx. */
 573 }
 574
 575 static int
 576 kiblnd_fmr_map_tx(struct kib_net *net, struct kib_tx *tx,
 577                   struct kib_rdma_desc *rd, u32 nob)
 578 {
 579         struct kib_hca_dev *hdev;
 580         struct kib_dev *dev;
 581         struct kib_fmr_poolset *fps;
 582         int                     cpt;
 583         int                     rc;
 584         int i;
 585
 586         LASSERT(tx->tx_pool != NULL);
 587         LASSERT(tx->tx_pool->tpo_pool.po_owner != NULL);
 588
 589         dev = net->ibn_dev;
 590         hdev = tx->tx_pool->tpo_hdev;
 591         cpt = tx->tx_pool->tpo_pool.po_owner->ps_cpt;
 592
 593         /*
 594          * If we're dealing with FastReg, but the device doesn't
 595          * support GAPS and the tx has GAPS, then there is no real point
 596          * in trying to map the memory, because it'll just fail. So
 597          * preemptively fail with an appropriate message
 598          */
 599         if (IS_FAST_REG_DEV(dev) &&
 600             !(dev->ibd_dev_caps & IBLND_DEV_CAPS_FASTREG_GAPS_SUPPORT) &&
 601             tx->tx_gaps) {
 602                 CERROR("Using FastReg with no GAPS support, but tx has gaps. "
 603                        "Try setting use_fastreg_gaps to 1\n");
 604                 return -EPROTONOSUPPORT;
 605         }
 606
 607 #ifdef HAVE_OFED_FMR_POOL_API
 608         /*
 609          * FMR does not support gaps but the tx has gaps then
 610          * we should make sure that the number of fragments we'll be sending
 611          * over fits within the number of fragments negotiated on the
 612          * connection, otherwise, we won't be able to RDMA the data.
 613          * We need to maintain the number of fragments negotiation on the
 614          * connection for backwards compatibility.
 615          */
 616         if (tx->tx_gaps && (dev->ibd_dev_caps & IBLND_DEV_CAPS_FMR_ENABLED)) {
 617                 if (tx->tx_conn &&
 618                     tx->tx_conn->ibc_max_frags <= rd->rd_nfrags) {
 619                         CERROR("TX number of frags (%d) is <= than connection"
 620                                " number of frags (%d). Consider setting peer's"
 621                                " map_on_demand to 256\n", tx->tx_nfrags,
 622                                tx->tx_conn->ibc_max_frags);
 623                         return -EFBIG;
 624                 }
 625         }
 626 #endif
 627
 628         fps = net->ibn_fmr_ps[cpt];
 629         rc = kiblnd_fmr_pool_map(fps, tx, rd, nob, 0, &tx->tx_fmr);
 630         if (rc != 0) {
 631                 CERROR("Can't map %u bytes (%u/%u)s: %d\n", nob,
 632                        tx->tx_nfrags, rd->rd_nfrags, rc);
 633                 return rc;
 634         }
 635
 636         /*
 637          * If rd is not tx_rd, it's going to get sent to a peer_ni, who will
 638          * need the rkey
 639          */
 640         rd->rd_key = tx->tx_fmr.fmr_key;
 641         /*
 642          * for FastReg or FMR with no gaps we can accumulate all
 643          * the fragments in one FastReg or FMR fragment.
 644          */
 645         if (
 646 #ifdef HAVE_OFED_FMR_POOL_API
 647             ((dev->ibd_dev_caps & IBLND_DEV_CAPS_FMR_ENABLED)
 648              && !tx->tx_gaps) ||
 649 #endif
 650             IS_FAST_REG_DEV(dev)) {
 651                 /* FMR requires zero based address */
 652 #ifdef HAVE_OFED_FMR_POOL_API
 653                 if (dev->ibd_dev_caps & IBLND_DEV_CAPS_FMR_ENABLED)
 654                         rd->rd_frags[0].rf_addr &= ~hdev->ibh_page_mask;
 655 #endif
 656                 rd->rd_frags[0].rf_nob = nob;
 657                 rd->rd_nfrags = 1;
 658         } else {
 659                 /*
 660                  * We're transmitting with gaps using FMR.
 661                  * We'll need to use multiple fragments and identify the
 662                  * zero based address of each fragment.
 663                  */
 664                 for (i = 0; i < rd->rd_nfrags; i++) {
 665                         rd->rd_frags[i].rf_addr &= ~hdev->ibh_page_mask;
 666                         rd->rd_frags[i].rf_addr += i << hdev->ibh_page_shift;
 667                 }
 668         }
 669
 670         return 0;
 671 }
 672
 673 static void
 674 kiblnd_unmap_tx(struct kib_tx *tx)
 675 {
 676         if (
 677 #ifdef HAVE_OFED_FMR_POOL_API
 678                 tx->tx_fmr.fmr_pfmr ||
 679 #endif
 680                 tx->tx_fmr.fmr_frd)
 681                 kiblnd_fmr_pool_unmap(&tx->tx_fmr, tx->tx_status);
 682
 683         if (tx->tx_nfrags != 0) {
 684                 kiblnd_dma_unmap_sg(tx->tx_pool->tpo_hdev, tx);
 685                 tx->tx_nfrags = 0;
 686         }
 687 }
 688
 689 #ifdef HAVE_OFED_IB_GET_DMA_MR
 690 static struct ib_mr *
 691 kiblnd_find_rd_dma_mr(struct lnet_ni *ni, struct kib_rdma_desc *rd)
 692 {
 693         struct kib_net *net = ni->ni_data;
 694         struct kib_hca_dev *hdev = net->ibn_dev->ibd_hdev;
 695         struct lnet_ioctl_config_o2iblnd_tunables *tunables;
 696
 697         tunables = &ni->ni_lnd_tunables.lnd_tun_u.lnd_o2ib;
 698
 699         /*
 700          * if map-on-demand is turned on and the device supports
 701          * either FMR or FastReg then use that. Otherwise use global
 702          * memory regions. If that's not available either, then you're
 703          * dead in the water and fail the operation.
 704          */
 705         if (tunables->lnd_map_on_demand && (IS_FAST_REG_DEV(net->ibn_dev)
 706 #ifdef HAVE_OFED_FMR_POOL_API
 707              || net->ibn_dev->ibd_dev_caps & IBLND_DEV_CAPS_FMR_ENABLED
 708 #endif
 709         ))
 710                 return NULL;
 711
 712         /*
 713          * hdev->ibh_mrs can be NULL. This case is dealt with gracefully
 714          * in the call chain. The mapping will fail with appropriate error
 715          * message.
 716          */
 717         return hdev->ibh_mrs;
 718 }
 719 #endif
 720
 721 static int kiblnd_map_tx(struct lnet_ni *ni, struct kib_tx *tx,
 722                          struct kib_rdma_desc *rd, int nfrags)
 723 {
 724         struct kib_net *net = ni->ni_data;
 725         struct kib_hca_dev *hdev = net->ibn_dev->ibd_hdev;
 726 #ifdef HAVE_OFED_IB_GET_DMA_MR
 727         struct ib_mr *mr = NULL;
 728 #endif
 729         __u32 nob;
 730         int i;
 731
 732         /* If rd is not tx_rd, it's going to get sent to a peer_ni and I'm the
 733          * RDMA sink */
 734         tx->tx_dmadir = (rd != tx->tx_rd) ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
 735         tx->tx_nfrags = nfrags;
 736
 737         rd->rd_nfrags = kiblnd_dma_map_sg(hdev, tx);
 738         for (i = 0, nob = 0; i < rd->rd_nfrags; i++) {
 739                 rd->rd_frags[i].rf_nob  = kiblnd_sg_dma_len(
 740                         hdev->ibh_ibdev, &tx->tx_frags[i]);
 741                 rd->rd_frags[i].rf_addr = kiblnd_sg_dma_address(
 742                         hdev->ibh_ibdev, &tx->tx_frags[i]);
 743                 nob += rd->rd_frags[i].rf_nob;
 744         }
 745
 746 #ifdef HAVE_OFED_IB_GET_DMA_MR
 747         mr = kiblnd_find_rd_dma_mr(ni, rd);
 748         if (mr != NULL) {
 749                 /* found pre-mapping MR */
 750                 rd->rd_key = (rd != tx->tx_rd) ? mr->rkey : mr->lkey;
 751                 return 0;
 752         }
 753 #endif
 754
 755         if (net->ibn_fmr_ps != NULL)
 756                 return kiblnd_fmr_map_tx(net, tx, rd, nob);
 757
 758         return -EINVAL;
 759 }
 760
 761 static int kiblnd_setup_rd_kiov(struct lnet_ni *ni, struct kib_tx *tx,
 762                                 struct kib_rdma_desc *rd, int nkiov,
 763                                 struct bio_vec *kiov, int offset, int nob)
 764 {
 765         struct kib_net *net = ni->ni_data;
 766         struct scatterlist *sg;
 767         int fragnob;
 768         int max_nkiov;
 769         int sg_count = 0;
 770
 771         CDEBUG(D_NET, "niov %d offset %d nob %d gpu %d\n",
 772                nkiov, offset, nob, tx->tx_gpu);
 773
 774         LASSERT(nob > 0);
 775         LASSERT(nkiov > 0);
 776         LASSERT(net != NULL);
 777
 778         while (offset >= kiov->bv_len) {
 779                 offset -= kiov->bv_len;
 780                 nkiov--;
 781                 kiov++;
 782                 LASSERT(nkiov > 0);
 783         }
 784
 785         max_nkiov = nkiov;
 786
 787         sg = tx->tx_frags;
 788         do {
 789                 LASSERT(nkiov > 0);
 790
 791                 if (!sg) {
 792                         CERROR("lacking enough sg entries to map tx\n");
 793                         return -EFAULT;
 794                 }
 795                 sg_count++;
 796
 797                 fragnob = min((int)(kiov->bv_len - offset), nob);
 798
 799                 /*
 800                  * We're allowed to start at a non-aligned page offset in
 801                  * the first fragment and end at a non-aligned page offset
 802                  * in the last fragment.
 803                  */
 804                 if ((fragnob < (int)(kiov->bv_len - offset)) &&
 805                     nkiov < max_nkiov && nob > fragnob) {
 806                         CDEBUG(D_NET, "fragnob %d < available page %d: with"
 807                                       " remaining %d kiovs with %d nob left\n",
 808                                fragnob, (int)(kiov->bv_len - offset),
 809                                nkiov, nob);
 810                         tx->tx_gaps = true;
 811                 }
 812
 813                 sg_set_page(sg, kiov->bv_page, fragnob,
 814                             kiov->bv_offset + offset);
 815                 sg = sg_next(sg);
 816
 817                 offset = 0;
 818                 kiov++;
 819                 nkiov--;
 820                 nob -= fragnob;
 821         } while (nob > 0);
 822
 823         return kiblnd_map_tx(ni, tx, rd, sg_count);
 824 }
 825
 826 static int
 827 kiblnd_post_tx_locked(struct kib_conn *conn, struct kib_tx *tx, int credit)
 828 __must_hold(&conn->ibc_lock)
 829 {
 830         struct kib_msg *msg = tx->tx_msg;
 831         struct kib_peer_ni *peer_ni = conn->ibc_peer;
 832         struct lnet_ni *ni = peer_ni->ibp_ni;
 833         struct kib_fast_reg_descriptor *frd = tx->tx_fmr.fmr_frd;
 834         int ver = conn->ibc_version;
 835         int rc;
 836         int done;
 837
 838         LASSERT(tx->tx_queued);
 839         /* We rely on this for QP sizing */
 840         LASSERT(tx->tx_nwrq > 0 && tx->tx_nsge >= 0);
 841         LASSERT(tx->tx_nwrq <= 1 + conn->ibc_max_frags);
 842
 843         LASSERT(credit == 0 || credit == 1);
 844         LASSERT(conn->ibc_outstanding_credits >= 0);
 845         LASSERT(conn->ibc_outstanding_credits <= conn->ibc_queue_depth);
 846         LASSERT(conn->ibc_credits >= 0);
 847         LASSERT(conn->ibc_credits <= conn->ibc_queue_depth);
 848
 849         if (conn->ibc_nsends_posted ==
 850             kiblnd_concurrent_sends(ver, ni)) {
 851                 /* tx completions outstanding... */
 852                 CDEBUG(D_NET, "%s: posted enough\n",
 853                        libcfs_nid2str(peer_ni->ibp_nid));
 854                 return -EAGAIN;
 855         }
 856
 857         if (credit != 0 && conn->ibc_credits == 0) {   /* no credits */
 858                 CDEBUG(D_NET, "%s: no credits cm_id %p qp_num 0x%x\n",
 859                         libcfs_nid2str(peer_ni->ibp_nid),
 860                         conn->ibc_cmid,
 861                         conn->ibc_cmid->qp ? conn->ibc_cmid->qp->qp_num : 0);
 862                 kiblnd_dump_conn_dbg(conn);
 863                 return -EAGAIN;
 864         }
 865
 866         if (credit != 0 && !IBLND_OOB_CAPABLE(ver) &&
 867             conn->ibc_credits == 1 &&   /* last credit reserved */
 868             msg->ibm_type != IBLND_MSG_NOOP) {      /* for NOOP */
 869                 CDEBUG(D_NET, "%s: not using last credit\n",
 870                        libcfs_nid2str(peer_ni->ibp_nid));
 871                 return -EAGAIN;
 872         }
 873
 874         /* NB don't drop ibc_lock before bumping tx_sending */
 875         list_del(&tx->tx_list);
 876         tx->tx_queued = 0;
 877
 878         if (msg->ibm_type == IBLND_MSG_NOOP &&
 879             (!kiblnd_need_noop(conn) ||     /* redundant NOOP */
 880              (IBLND_OOB_CAPABLE(ver) && /* posted enough NOOP */
 881               conn->ibc_noops_posted == IBLND_OOB_MSGS(ver)))) {
 882                 /* OK to drop when posted enough NOOPs, since
 883                  * kiblnd_check_sends_locked will queue NOOP again when
 884                  * posted NOOPs complete */
 885                 spin_unlock(&conn->ibc_lock);
 886                 tx->tx_hstatus = LNET_MSG_STATUS_LOCAL_ERROR;
 887                 kiblnd_tx_done(tx);
 888                 spin_lock(&conn->ibc_lock);
 889                 CDEBUG(D_NET, "%s(%d): redundant or enough NOOP cm_id %p qp_num 0x%x\n",
 890                         libcfs_nid2str(peer_ni->ibp_nid),
 891                         conn->ibc_noops_posted,
 892                         conn->ibc_cmid,
 893                         conn->ibc_cmid->qp ? conn->ibc_cmid->qp->qp_num : 0);
 894                 kiblnd_dump_conn_dbg(conn);
 895                 return 0;
 896         }
 897
 898         CDEBUG(D_NET, "Transmit %x[%d] nob %u cm_id %p qp_num 0x%x\n",
 899                 msg->ibm_type, credit,
 900                 msg->ibm_nob,
 901                 conn->ibc_cmid,
 902                 conn->ibc_cmid->qp ? conn->ibc_cmid->qp->qp_num : 0);
 903         kiblnd_dump_conn_dbg(conn);
 904
 905         kiblnd_pack_msg(peer_ni->ibp_ni, msg, ver, conn->ibc_outstanding_credits,
 906                         peer_ni->ibp_nid, conn->ibc_incarnation);
 907
 908         conn->ibc_credits -= credit;
 909         conn->ibc_outstanding_credits = 0;
 910         conn->ibc_nsends_posted++;
 911         if (msg->ibm_type == IBLND_MSG_NOOP)
 912                 conn->ibc_noops_posted++;
 913
 914         /* CAVEAT EMPTOR!  This tx could be the PUT_DONE of an RDMA
 915          * PUT.  If so, it was first queued here as a PUT_REQ, sent and
 916          * stashed on ibc_active_txs, matched by an incoming PUT_ACK,
 917          * and then re-queued here.  It's (just) possible that
 918          * tx_sending is non-zero if we've not done the tx_complete()
 919          * from the first send; hence the ++ rather than = below. */
 920         tx->tx_sending++;
 921         list_add(&tx->tx_list, &conn->ibc_active_txs);
 922
 923         /* I'm still holding ibc_lock! */
 924         if (conn->ibc_state != IBLND_CONN_ESTABLISHED) {
 925                 CDEBUG(D_NET, "connection to %s is not established\n",
 926                                 conn->ibc_peer? libcfs_nid2str(conn->ibc_peer->ibp_nid): "NULL");
 927                 rc = -ECONNABORTED;
 928         } else if (tx->tx_pool->tpo_pool.po_failed ||
 929                  conn->ibc_hdev != tx->tx_pool->tpo_hdev) {
 930                 /* close_conn will launch failover */
 931                 rc = -ENETDOWN;
 932         } else {
 933                 struct ib_send_wr *bad = &tx->tx_wrq[tx->tx_nwrq - 1].wr;
 934                 struct ib_send_wr *wr  = &tx->tx_wrq[0].wr;
 935
 936                 if (frd != NULL && !frd->frd_posted) {
 937                         wr = &frd->frd_inv_wr.wr;
 938                         wr->next = &frd->frd_fastreg_wr.wr;
 939                         frd->frd_fastreg_wr.wr.next = &tx->tx_wrq[0].wr;
 940                 }
 941
 942                 LASSERTF(bad->wr_id == kiblnd_ptr2wreqid(tx, IBLND_WID_TX),
 943                          "bad wr_id %#llx, opc %d, flags %d, peer_ni: %s\n",
 944                          bad->wr_id, bad->opcode, bad->send_flags,
 945                          libcfs_nid2str(conn->ibc_peer->ibp_nid));
 946
 947                 bad = NULL;
 948                 if (lnet_send_error_simulation(tx->tx_lntmsg[0], &tx->tx_hstatus))
 949                         rc = -EINVAL;
 950                 else
 951 #ifdef HAVE_OFED_IB_POST_SEND_RECV_CONST
 952                         rc = ib_post_send(conn->ibc_cmid->qp, wr,
 953                                           (const struct ib_send_wr **)&bad);
 954 #else
 955                         rc = ib_post_send(conn->ibc_cmid->qp, wr, &bad);
 956 #endif
 957                 if (frd && !frd->frd_posted) {
 958                         /* The local invalidate becomes invalid (has been
 959                          * successfully used) if the post succeeds or the
 960                          * failing wr was not the invalidate. */
 961                         frd->frd_valid =
 962                                 !(rc == 0 || (bad != &frd->frd_inv_wr.wr));
 963                 }
 964         }
 965
 966         conn->ibc_last_send = ktime_get();
 967
 968         if (rc == 0) {
 969                 if (frd != NULL)
 970                         frd->frd_posted = true;
 971                 return 0;
 972         }
 973
 974         /* NB credits are transferred in the actual
 975          * message, which can only be the last work item */
 976         conn->ibc_credits += credit;
 977         conn->ibc_outstanding_credits += msg->ibm_credits;
 978         conn->ibc_nsends_posted--;
 979         if (msg->ibm_type == IBLND_MSG_NOOP)
 980                 conn->ibc_noops_posted--;
 981
 982         tx->tx_status = rc;
 983         tx->tx_waiting = 0;
 984         tx->tx_sending--;
 985
 986         done = (tx->tx_sending == 0);
 987         if (done)
 988                 list_del(&tx->tx_list);
 989
 990         spin_unlock(&conn->ibc_lock);
 991
 992         if (conn->ibc_state == IBLND_CONN_ESTABLISHED)
 993                 CERROR("Error %d posting transmit to %s\n",
 994                        rc, libcfs_nid2str(peer_ni->ibp_nid));
 995         else
 996                 CDEBUG(D_NET, "Error %d posting transmit to %s\n",
 997                        rc, libcfs_nid2str(peer_ni->ibp_nid));
 998
 999         kiblnd_close_conn(conn, rc);
1000
1001         if (done)
1002                 kiblnd_tx_done(tx);
1003
1004         spin_lock(&conn->ibc_lock);
1005
1006         return -EIO;
1007 }
1008
1009 static void
1010 kiblnd_check_sends_locked(struct kib_conn *conn)
1011 {
1012         int ver = conn->ibc_version;
1013         struct lnet_ni *ni = conn->ibc_peer->ibp_ni;
1014         struct kib_tx *tx;
1015
1016         /* Don't send anything until after the connection is established */
1017         if (conn->ibc_state < IBLND_CONN_ESTABLISHED) {
1018                 CDEBUG(D_NET, "%s too soon\n",
1019                        libcfs_nid2str(conn->ibc_peer->ibp_nid));
1020                 return;
1021         }
1022
1023         LASSERT(conn->ibc_nsends_posted <=
1024                 kiblnd_concurrent_sends(ver, ni));
1025         LASSERT (!IBLND_OOB_CAPABLE(ver) ||
1026                  conn->ibc_noops_posted <= IBLND_OOB_MSGS(ver));
1027         LASSERT (conn->ibc_reserved_credits >= 0);
1028
1029         while (conn->ibc_reserved_credits > 0 &&
1030                (tx = list_first_entry_or_null(&conn->ibc_tx_queue_rsrvd,
1031                                               struct kib_tx, tx_list)) != NULL) {
1032                 list_move_tail(&tx->tx_list, &conn->ibc_tx_queue);
1033                 conn->ibc_reserved_credits--;
1034         }
1035
1036         if (kiblnd_need_noop(conn)) {
1037                 spin_unlock(&conn->ibc_lock);
1038
1039                 tx = kiblnd_get_idle_tx(ni, conn->ibc_peer->ibp_nid);
1040                 if (tx != NULL)
1041                         kiblnd_init_tx_msg(ni, tx, IBLND_MSG_NOOP, 0);
1042
1043                 spin_lock(&conn->ibc_lock);
1044                 if (tx != NULL)
1045                         kiblnd_queue_tx_locked(tx, conn);
1046         }
1047
1048         for (;;) {
1049                 int credit;
1050
1051                 if (!list_empty(&conn->ibc_tx_queue_nocred)) {
1052                         credit = 0;
1053                         tx = list_first_entry(&conn->ibc_tx_queue_nocred,
1054                                               struct kib_tx, tx_list);
1055                 } else if (!list_empty(&conn->ibc_tx_noops)) {
1056                         LASSERT (!IBLND_OOB_CAPABLE(ver));
1057                         credit = 1;
1058                         tx = list_first_entry(&conn->ibc_tx_noops,
1059                                               struct kib_tx, tx_list);
1060                 } else if (!list_empty(&conn->ibc_tx_queue)) {
1061                         credit = 1;
1062                         tx = list_first_entry(&conn->ibc_tx_queue,
1063                                               struct kib_tx, tx_list);
1064                 } else
1065                         break;
1066
1067                 if (kiblnd_post_tx_locked(conn, tx, credit) != 0)
1068                         break;
1069         }
1070 }
1071
1072 static void
1073 kiblnd_tx_complete(struct kib_tx *tx, int status)
1074 {
1075         int           failed = (status != IB_WC_SUCCESS);
1076         struct kib_conn   *conn = tx->tx_conn;
1077         int           idle;
1078
1079         if (tx->tx_sending <= 0) {
1080                 CERROR("Received an event on a freed tx: %p status %d\n",
1081                        tx, tx->tx_status);
1082                 return;
1083         }
1084
1085         if (failed) {
1086                 if (conn->ibc_state == IBLND_CONN_ESTABLISHED)
1087                         CNETERR("Tx -> %s cookie %#llx"
1088                                 " sending %d waiting %d: failed %d\n",
1089                                 libcfs_nid2str(conn->ibc_peer->ibp_nid),
1090                                 tx->tx_cookie, tx->tx_sending, tx->tx_waiting,
1091                                 status);
1092
1093                 kiblnd_close_conn(conn, -EIO);
1094         } else {
1095                 kiblnd_peer_alive(conn->ibc_peer);
1096         }
1097
1098         spin_lock(&conn->ibc_lock);
1099
1100         /* I could be racing with rdma completion.  Whoever makes 'tx' idle
1101          * gets to free it, which also drops its ref on 'conn'. */
1102
1103         tx->tx_sending--;
1104         conn->ibc_nsends_posted--;
1105         if (tx->tx_msg->ibm_type == IBLND_MSG_NOOP)
1106                 conn->ibc_noops_posted--;
1107
1108         if (failed) {
1109                 tx->tx_hstatus = LNET_MSG_STATUS_REMOTE_DROPPED;
1110                 tx->tx_waiting = 0;             /* don't wait for peer_ni */
1111                 tx->tx_status = -EIO;
1112 #ifdef O2IBLND_CONN_STATE_DEBUG
1113                 kiblnd_dump_conn_dbg(conn);
1114 #endif
1115         }
1116
1117         idle = (tx->tx_sending == 0) &&         /* This is the final callback */
1118                !tx->tx_waiting &&               /* Not waiting for peer_ni */
1119                !tx->tx_queued;                  /* Not re-queued (PUT_DONE) */
1120         if (idle)
1121                 list_del(&tx->tx_list);
1122
1123         kiblnd_check_sends_locked(conn);
1124         spin_unlock(&conn->ibc_lock);
1125
1126         if (idle)
1127                 kiblnd_tx_done(tx);
1128 }
1129
1130
1131 static void
1132 kiblnd_init_tx_sge(struct kib_tx *tx, u64 addr, unsigned int len)
1133 {
1134         struct ib_sge *sge = &tx->tx_sge[tx->tx_nsge];
1135         struct kib_hca_dev *hdev = tx->tx_pool->tpo_hdev;
1136 #ifdef HAVE_OFED_IB_GET_DMA_MR
1137         struct ib_mr *mr = hdev->ibh_mrs;
1138 #endif
1139
1140         *sge = (struct ib_sge) {
1141 #ifdef HAVE_OFED_IB_GET_DMA_MR
1142                 .lkey   = mr->lkey,
1143 #else
1144                 .lkey   = hdev->ibh_pd->local_dma_lkey,
1145 #endif
1146                 .addr   = addr,
1147                 .length = len,
1148         };
1149
1150         tx->tx_nsge++;
1151 }
1152
1153 static struct ib_rdma_wr *
1154 kiblnd_init_tx_msg_payload(struct lnet_ni *ni, struct kib_tx *tx, int type,
1155                    int body_nob, int payload)
1156 {
1157         struct ib_rdma_wr *wrq;
1158         int nob = offsetof(struct kib_msg, ibm_u) + body_nob;
1159
1160         LASSERT(tx->tx_nwrq >= 0);
1161         LASSERT(tx->tx_nwrq < IBLND_MAX_RDMA_FRAGS + 1);
1162         LASSERT(nob <= IBLND_MSG_SIZE);
1163
1164         kiblnd_init_msg(tx->tx_msg, type, body_nob + payload);
1165
1166         wrq = &tx->tx_wrq[tx->tx_nwrq];
1167
1168         *wrq = (struct ib_rdma_wr) {
1169                 .wr = {
1170                         .wr_id          = kiblnd_ptr2wreqid(tx, IBLND_WID_TX),
1171                         .num_sge        = 1,
1172                         .sg_list        = &tx->tx_sge[tx->tx_nsge],
1173                         .opcode         = IB_WR_SEND,
1174                         .send_flags     = IB_SEND_SIGNALED,
1175                 },
1176         };
1177
1178         kiblnd_init_tx_sge(tx, tx->tx_msgaddr, nob);
1179
1180         tx->tx_nwrq++;
1181         return wrq;
1182 }
1183
1184 static int
1185 kiblnd_init_rdma(struct kib_conn *conn, struct kib_tx *tx, int type,
1186                  int resid, struct kib_rdma_desc *dstrd, u64 dstcookie)
1187 {
1188         struct kib_msg *ibmsg = tx->tx_msg;
1189         struct kib_rdma_desc *srcrd = tx->tx_rd;
1190         struct ib_rdma_wr *wrq = NULL;
1191         struct ib_sge     *sge;
1192         int                rc  = resid;
1193         int                srcidx;
1194         int                dstidx;
1195         int                sge_nob;
1196         int                wrq_sge;
1197
1198         LASSERT(!in_interrupt());
1199         LASSERT(tx->tx_nwrq == 0 && tx->tx_nsge == 0);
1200         LASSERT(type == IBLND_MSG_GET_DONE || type == IBLND_MSG_PUT_DONE);
1201
1202         for (srcidx = dstidx = wrq_sge = sge_nob = 0;
1203              resid > 0; resid -= sge_nob) {
1204                 int     prev = dstidx;
1205
1206                 if (srcidx >= srcrd->rd_nfrags) {
1207                         CERROR("Src buffer exhausted: %d frags %px\n",
1208                                 srcidx, tx);
1209                         rc = -EPROTO;
1210                         break;
1211                 }
1212
1213                 if (dstidx >= dstrd->rd_nfrags) {
1214                         CERROR("Dst buffer exhausted: %d frags\n", dstidx);
1215                         rc = -EPROTO;
1216                         break;
1217                 }
1218
1219                 if (tx->tx_nwrq >= conn->ibc_max_frags) {
1220                         CERROR("RDMA has too many fragments for peer_ni %s (%d), "
1221                                "src idx/frags: %d/%d dst idx/frags: %d/%d\n",
1222                                libcfs_nid2str(conn->ibc_peer->ibp_nid),
1223                                conn->ibc_max_frags,
1224                                srcidx, srcrd->rd_nfrags,
1225                                dstidx, dstrd->rd_nfrags);
1226                         rc = -EMSGSIZE;
1227                         break;
1228                 }
1229
1230                 sge_nob = min3(kiblnd_rd_frag_size(srcrd, srcidx),
1231                                kiblnd_rd_frag_size(dstrd, dstidx),
1232                                resid);
1233
1234                 sge = &tx->tx_sge[tx->tx_nsge];
1235                 sge->addr   = kiblnd_rd_frag_addr(srcrd, srcidx);
1236                 sge->lkey   = kiblnd_rd_frag_key(srcrd, srcidx);
1237                 sge->length = sge_nob;
1238
1239                 if (wrq_sge == 0) {
1240                         wrq = &tx->tx_wrq[tx->tx_nwrq];
1241
1242                         wrq->wr.next    = &(wrq + 1)->wr;
1243                         wrq->wr.wr_id   = kiblnd_ptr2wreqid(tx, IBLND_WID_RDMA);
1244                         wrq->wr.sg_list = sge;
1245                         wrq->wr.opcode  = IB_WR_RDMA_WRITE;
1246                         wrq->wr.send_flags = 0;
1247
1248 #ifdef HAVE_OFED_IB_RDMA_WR
1249                         wrq->remote_addr        = kiblnd_rd_frag_addr(dstrd,
1250                                                                       dstidx);
1251                         wrq->rkey               = kiblnd_rd_frag_key(dstrd,
1252                                                                      dstidx);
1253 #else
1254                         wrq->wr.wr.rdma.remote_addr = kiblnd_rd_frag_addr(dstrd,
1255                                                                         dstidx);
1256                         wrq->wr.wr.rdma.rkey    = kiblnd_rd_frag_key(dstrd,
1257                                                                      dstidx);
1258 #endif
1259                 }
1260
1261                 srcidx = kiblnd_rd_consume_frag(srcrd, srcidx, sge_nob);
1262                 dstidx = kiblnd_rd_consume_frag(dstrd, dstidx, sge_nob);
1263
1264                 wrq_sge++;
1265                 if (wrq_sge == *kiblnd_tunables.kib_wrq_sge || dstidx != prev) {
1266                         tx->tx_nwrq++;
1267                         wrq->wr.num_sge = wrq_sge;
1268                         wrq_sge = 0;
1269                 }
1270                 tx->tx_nsge++;
1271         }
1272
1273         if (rc < 0)     /* no RDMA if completing with failure */
1274                 tx->tx_nwrq = tx->tx_nsge = 0;
1275
1276         ibmsg->ibm_u.completion.ibcm_status = rc;
1277         ibmsg->ibm_u.completion.ibcm_cookie = dstcookie;
1278         kiblnd_init_tx_msg(conn->ibc_peer->ibp_ni, tx,
1279                            type, sizeof(struct kib_completion_msg));
1280
1281         return rc;
1282 }
1283
1284 static void
1285 kiblnd_queue_tx_locked(struct kib_tx *tx, struct kib_conn *conn)
1286 {
1287         struct list_head *q;
1288         s64 timeout_ns;
1289
1290         LASSERT(tx->tx_nwrq > 0);       /* work items set up */
1291         LASSERT(!tx->tx_queued);        /* not queued for sending already */
1292         LASSERT(conn->ibc_state >= IBLND_CONN_ESTABLISHED);
1293
1294         if (conn->ibc_state >= IBLND_CONN_DISCONNECTED) {
1295                 CDEBUG(D_NET, "connection with %s is disconnected\n",
1296                                 conn->ibc_peer? libcfs_nid2str(conn->ibc_peer->ibp_nid): "NULL");
1297
1298                 tx->tx_status = -ECONNABORTED;
1299                 tx->tx_waiting = 0;
1300                 if (tx->tx_conn != NULL) {
1301                         /* PUT_DONE first attached to conn as a PUT_REQ */
1302                         LASSERT(tx->tx_conn == conn);
1303                         LASSERT(tx->tx_msg->ibm_type == IBLND_MSG_PUT_DONE);
1304                         tx->tx_conn = NULL;
1305                         kiblnd_conn_decref(conn);
1306                 }
1307                 list_add(&tx->tx_list, &conn->ibc_zombie_txs);
1308
1309                 return;
1310         }
1311
1312         timeout_ns = kiblnd_timeout() * NSEC_PER_SEC;
1313         tx->tx_queued = 1;
1314         tx->tx_deadline = ktime_add_ns(ktime_get(), timeout_ns);
1315
1316         if (tx->tx_conn == NULL) {
1317                 kiblnd_conn_addref(conn);
1318                 tx->tx_conn = conn;
1319                 LASSERT (tx->tx_msg->ibm_type != IBLND_MSG_PUT_DONE);
1320         } else {
1321                 /* PUT_DONE first attached to conn as a PUT_REQ */
1322                 LASSERT (tx->tx_conn == conn);
1323                 LASSERT (tx->tx_msg->ibm_type == IBLND_MSG_PUT_DONE);
1324         }
1325
1326         switch (tx->tx_msg->ibm_type) {
1327         default:
1328                 LBUG();
1329
1330         case IBLND_MSG_PUT_REQ:
1331         case IBLND_MSG_GET_REQ:
1332                 q = &conn->ibc_tx_queue_rsrvd;
1333                 break;
1334
1335         case IBLND_MSG_PUT_NAK:
1336         case IBLND_MSG_PUT_ACK:
1337         case IBLND_MSG_PUT_DONE:
1338         case IBLND_MSG_GET_DONE:
1339                 q = &conn->ibc_tx_queue_nocred;
1340                 break;
1341
1342         case IBLND_MSG_NOOP:
1343                 if (IBLND_OOB_CAPABLE(conn->ibc_version))
1344                         q = &conn->ibc_tx_queue_nocred;
1345                 else
1346                         q = &conn->ibc_tx_noops;
1347                 break;
1348
1349         case IBLND_MSG_IMMEDIATE:
1350                 q = &conn->ibc_tx_queue;
1351                 break;
1352         }
1353
1354         list_add_tail(&tx->tx_list, q);
1355 }
1356
1357 static void
1358 kiblnd_queue_tx(struct kib_tx *tx, struct kib_conn *conn)
1359 {
1360         spin_lock(&conn->ibc_lock);
1361         kiblnd_queue_tx_locked(tx, conn);
1362         kiblnd_check_sends_locked(conn);
1363         spin_unlock(&conn->ibc_lock);
1364 }
1365
1366 static int
1367 kiblnd_resolve_addr_cap(struct rdma_cm_id *cmid,
1368                         struct sockaddr_in *srcaddr,
1369                         struct sockaddr_in *dstaddr,
1370                         int timeout_ms)
1371 {
1372         unsigned short port;
1373         int rc;
1374
1375         /* allow the port to be reused */
1376         rc = rdma_set_reuseaddr(cmid, 1);
1377         if (rc != 0) {
1378                 CERROR("Unable to set reuse on cmid: %d\n", rc);
1379                 return rc;
1380         }
1381
1382         /* look for a free privileged port */
1383         for (port = PROT_SOCK-1; port > 0; port--) {
1384                 srcaddr->sin_port = htons(port);
1385                 rc = rdma_resolve_addr(cmid,
1386                                        (struct sockaddr *)srcaddr,
1387                                        (struct sockaddr *)dstaddr,
1388                                        timeout_ms);
1389                 if (rc == 0) {
1390                         CDEBUG(D_NET, "bound to port %hu\n", port);
1391                         return 0;
1392                 } else if (rc == -EADDRINUSE || rc == -EADDRNOTAVAIL) {
1393                         CDEBUG(D_NET, "bind to port %hu failed: %d\n",
1394                                port, rc);
1395                 } else {
1396                         return rc;
1397                 }
1398         }
1399
1400         CERROR("cannot bind to a free privileged port: rc = %d\n", rc);
1401
1402         return rc;
1403 }
1404
1405 static int
1406 kiblnd_resolve_addr(struct rdma_cm_id *cmid,
1407                     struct sockaddr_in *srcaddr,
1408                     struct sockaddr_in *dstaddr,
1409                     int timeout_ms)
1410 {
1411         const struct cred *old_creds = NULL;
1412         struct cred *new_creds;
1413         int rc;
1414
1415         if (!capable(CAP_NET_BIND_SERVICE)) {
1416                 new_creds = prepare_kernel_cred(NULL);
1417                 if (!new_creds)
1418                         return -ENOMEM;
1419
1420                 cap_raise(new_creds->cap_effective, CAP_NET_BIND_SERVICE);
1421                 old_creds = override_creds(new_creds);
1422         }
1423
1424         rc = kiblnd_resolve_addr_cap(cmid, srcaddr, dstaddr, timeout_ms);
1425
1426         if (old_creds)
1427                 revert_creds(old_creds);
1428
1429         return rc;
1430 }
1431
1432 static void
1433 kiblnd_connect_peer(struct kib_peer_ni *peer_ni)
1434 {
1435         struct rdma_cm_id *cmid;
1436         struct kib_dev *dev;
1437         struct kib_net *net = peer_ni->ibp_ni->ni_data;
1438         struct sockaddr_in srcaddr;
1439         struct sockaddr_in dstaddr;
1440         int rc;
1441
1442         LASSERT (net != NULL);
1443         LASSERT (peer_ni->ibp_connecting > 0);
1444
1445         cmid = kiblnd_rdma_create_id(peer_ni->ibp_ni->ni_net_ns,
1446                                      kiblnd_cm_callback, peer_ni,
1447                                      RDMA_PS_TCP, IB_QPT_RC);
1448
1449         if (IS_ERR(cmid)) {
1450                 CERROR("Can't create CMID for %s: %ld\n",
1451                        libcfs_nid2str(peer_ni->ibp_nid), PTR_ERR(cmid));
1452                 rc = PTR_ERR(cmid);
1453                 goto failed;
1454         }
1455
1456         dev = net->ibn_dev;
1457         memset(&srcaddr, 0, sizeof(srcaddr));
1458         srcaddr.sin_family = AF_INET;
1459         srcaddr.sin_addr.s_addr = htonl(dev->ibd_ifip);
1460
1461         memset(&dstaddr, 0, sizeof(dstaddr));
1462         dstaddr.sin_family = AF_INET;
1463         dstaddr.sin_port = htons(*kiblnd_tunables.kib_service);
1464         dstaddr.sin_addr.s_addr = htonl(LNET_NIDADDR(peer_ni->ibp_nid));
1465
1466         kiblnd_peer_addref(peer_ni);               /* cmid's ref */
1467
1468         if (*kiblnd_tunables.kib_use_priv_port) {
1469                 rc = kiblnd_resolve_addr(cmid, &srcaddr, &dstaddr,
1470                                          kiblnd_timeout() * 1000);
1471         } else {
1472                 rc = rdma_resolve_addr(cmid,
1473                                        (struct sockaddr *)&srcaddr,
1474                                        (struct sockaddr *)&dstaddr,
1475                                        kiblnd_timeout() * 1000);
1476         }
1477         if (rc != 0) {
1478                 /* Can't initiate address resolution:  */
1479                 CERROR("Can't resolve addr for %s: %d\n",
1480                        libcfs_nid2str(peer_ni->ibp_nid), rc);
1481                 goto failed2;
1482         }
1483
1484         return;
1485
1486  failed2:
1487         kiblnd_peer_connect_failed(peer_ni, 1, rc);
1488         kiblnd_peer_decref(peer_ni);               /* cmid's ref */
1489         rdma_destroy_id(cmid);
1490         return;
1491  failed:
1492         kiblnd_peer_connect_failed(peer_ni, 1, rc);
1493 }
1494
1495 bool
1496 kiblnd_reconnect_peer(struct kib_peer_ni *peer_ni)
1497 {
1498         rwlock_t *glock = &kiblnd_data.kib_global_lock;
1499         char *reason = NULL;
1500         LIST_HEAD(txs);
1501         unsigned long flags;
1502
1503         write_lock_irqsave(glock, flags);
1504         if (peer_ni->ibp_reconnecting == 0) {
1505                 if (peer_ni->ibp_accepting)
1506                         reason = "accepting";
1507                 else if (peer_ni->ibp_connecting)
1508                         reason = "connecting";
1509                 else if (!list_empty(&peer_ni->ibp_conns))
1510                         reason = "connected";
1511                 else /* connected then closed */
1512                         reason = "closed";
1513
1514                 goto no_reconnect;
1515         }
1516
1517         if (peer_ni->ibp_accepting)
1518                 CNETERR("Detecting race between accepting and reconnecting\n");
1519         peer_ni->ibp_reconnecting--;
1520
1521         if (!kiblnd_peer_active(peer_ni)) {
1522                 list_splice_init(&peer_ni->ibp_tx_queue, &txs);
1523                 reason = "unlinked";
1524                 goto no_reconnect;
1525         }
1526
1527         peer_ni->ibp_connecting++;
1528         peer_ni->ibp_reconnected++;
1529
1530         write_unlock_irqrestore(glock, flags);
1531
1532         kiblnd_connect_peer(peer_ni);
1533         return true;
1534
1535  no_reconnect:
1536         write_unlock_irqrestore(glock, flags);
1537
1538         CWARN("Abort reconnection of %s: %s\n",
1539               libcfs_nid2str(peer_ni->ibp_nid), reason);
1540         kiblnd_txlist_done(&txs, -ECONNABORTED,
1541                            LNET_MSG_STATUS_LOCAL_ABORTED);
1542         return false;
1543 }
1544
1545 void
1546 kiblnd_launch_tx(struct lnet_ni *ni, struct kib_tx *tx, lnet_nid_t nid)
1547 {
1548         struct kib_peer_ni *peer_ni;
1549         struct kib_peer_ni *peer2;
1550         struct kib_conn *conn;
1551         rwlock_t *g_lock = &kiblnd_data.kib_global_lock;
1552         unsigned long flags;
1553         int rc;
1554         int i;
1555         struct lnet_ioctl_config_o2iblnd_tunables *tunables;
1556         s64 timeout_ns;
1557
1558         /* If I get here, I've committed to send, so I complete the tx with
1559          * failure on any problems
1560          */
1561
1562         LASSERT(!tx || !tx->tx_conn);     /* only set when assigned a conn */
1563         LASSERT(!tx || tx->tx_nwrq > 0);  /* work items have been set up */
1564
1565         /* First time, just use a read lock since I expect to find my peer_ni
1566          * connected
1567          */
1568         read_lock_irqsave(g_lock, flags);
1569
1570         peer_ni = kiblnd_find_peer_locked(ni, nid);
1571         if (peer_ni != NULL && !list_empty(&peer_ni->ibp_conns)) {
1572                 /* Found a peer_ni with an established connection */
1573                 conn = kiblnd_get_conn_locked(peer_ni);
1574                 kiblnd_conn_addref(conn); /* 1 ref for me... */
1575
1576                 read_unlock_irqrestore(g_lock, flags);
1577
1578                 if (tx != NULL)
1579                         kiblnd_queue_tx(tx, conn);
1580                 kiblnd_conn_decref(conn); /* ...to here */
1581                 return;
1582         }
1583
1584         timeout_ns = kiblnd_timeout() * NSEC_PER_SEC;
1585         read_unlock(g_lock);
1586         /* Re-try with a write lock */
1587         write_lock(g_lock);
1588
1589         peer_ni = kiblnd_find_peer_locked(ni, nid);
1590         if (peer_ni != NULL) {
1591                 if (list_empty(&peer_ni->ibp_conns)) {
1592                         /* found a peer_ni, but it's still connecting... */
1593                         LASSERT(kiblnd_peer_connecting(peer_ni));
1594                         if (tx != NULL) {
1595                                 tx->tx_deadline = ktime_add_ns(ktime_get(),
1596                                                                timeout_ns);
1597                                 list_add_tail(&tx->tx_list,
1598                                               &peer_ni->ibp_tx_queue);
1599                         }
1600                         write_unlock_irqrestore(g_lock, flags);
1601                 } else {
1602                         conn = kiblnd_get_conn_locked(peer_ni);
1603                         kiblnd_conn_addref(conn); /* 1 ref for me... */
1604
1605                         write_unlock_irqrestore(g_lock, flags);
1606
1607                         if (tx != NULL)
1608                                 kiblnd_queue_tx(tx, conn);
1609                         kiblnd_conn_decref(conn); /* ...to here */
1610                 }
1611                 return;
1612         }
1613
1614         write_unlock_irqrestore(g_lock, flags);
1615
1616         /* Allocate a peer_ni ready to add to the peer_ni table and retry */
1617         rc = kiblnd_create_peer(ni, &peer_ni, nid);
1618         if (rc != 0) {
1619                 CERROR("Can't create peer_ni %s\n", libcfs_nid2str(nid));
1620                 if (tx != NULL) {
1621                         tx->tx_status = -EHOSTUNREACH;
1622                         tx->tx_waiting = 0;
1623                         tx->tx_hstatus = LNET_MSG_STATUS_LOCAL_ERROR;
1624                         kiblnd_tx_done(tx);
1625                 }
1626                 return;
1627         }
1628
1629         write_lock_irqsave(g_lock, flags);
1630
1631         peer2 = kiblnd_find_peer_locked(ni, nid);
1632         if (peer2 != NULL) {
1633                 if (list_empty(&peer2->ibp_conns)) {
1634                         /* found a peer_ni, but it's still connecting... */
1635                         LASSERT(kiblnd_peer_connecting(peer2));
1636                         if (tx != NULL) {
1637                                 tx->tx_deadline = ktime_add_ns(ktime_get(),
1638                                                                timeout_ns);
1639                                 list_add_tail(&tx->tx_list,
1640                                               &peer2->ibp_tx_queue);
1641                         }
1642                         write_unlock_irqrestore(g_lock, flags);
1643                 } else {
1644                         conn = kiblnd_get_conn_locked(peer2);
1645                         kiblnd_conn_addref(conn); /* 1 ref for me... */
1646
1647                         write_unlock_irqrestore(g_lock, flags);
1648
1649                         if (tx != NULL)
1650                                 kiblnd_queue_tx(tx, conn);
1651                         kiblnd_conn_decref(conn); /* ...to here */
1652                 }
1653
1654                 kiblnd_peer_decref(peer_ni);
1655                 return;
1656         }
1657
1658         /* Brand new peer_ni */
1659         LASSERT(peer_ni->ibp_connecting == 0);
1660         tunables = &peer_ni->ibp_ni->ni_lnd_tunables.lnd_tun_u.lnd_o2ib;
1661         peer_ni->ibp_connecting = tunables->lnd_conns_per_peer;
1662
1663         /* always called with a ref on ni, which prevents ni being shutdown */
1664         LASSERT(((struct kib_net *)ni->ni_data)->ibn_shutdown == 0);
1665
1666         if (tx != NULL) {
1667                 tx->tx_deadline = ktime_add_ns(ktime_get(), timeout_ns);
1668                 list_add_tail(&tx->tx_list, &peer_ni->ibp_tx_queue);
1669         }
1670
1671         kiblnd_peer_addref(peer_ni);
1672         hash_add(kiblnd_data.kib_peers, &peer_ni->ibp_list, nid);
1673
1674         write_unlock_irqrestore(g_lock, flags);
1675
1676         for (i = 0; i < tunables->lnd_conns_per_peer; i++)
1677                 kiblnd_connect_peer(peer_ni);
1678         kiblnd_peer_decref(peer_ni);
1679 }
1680
1681 int
1682 kiblnd_send(struct lnet_ni *ni, void *private, struct lnet_msg *lntmsg)
1683 {
1684         struct kib_dev *dev = ((struct kib_net *)ni->ni_data)->ibn_dev;
1685         struct lnet_hdr *hdr = &lntmsg->msg_hdr;
1686         int type = lntmsg->msg_type;
1687         struct lnet_processid *target = &lntmsg->msg_target;
1688         int target_is_router = lntmsg->msg_target_is_router;
1689         int routing = lntmsg->msg_routing;
1690         unsigned int payload_niov = lntmsg->msg_niov;
1691         struct bio_vec *payload_kiov = lntmsg->msg_kiov;
1692         unsigned int payload_offset = lntmsg->msg_offset;
1693         unsigned int payload_nob = lntmsg->msg_len;
1694         struct lnet_libmd *msg_md = lntmsg->msg_md;
1695         bool gpu;
1696         struct kib_msg *ibmsg;
1697         struct kib_rdma_desc *rd;
1698         struct kib_tx *tx;
1699         int nob;
1700         int rc;
1701
1702         /* NB 'private' is different depending on what we're sending.... */
1703
1704         CDEBUG(D_NET, "sending %d bytes in %d frags to %s\n",
1705                payload_nob, payload_niov, libcfs_idstr(target));
1706
1707         LASSERT(payload_nob == 0 || payload_niov > 0);
1708
1709         /* Thread context */
1710         LASSERT(!in_interrupt());
1711
1712         tx = kiblnd_get_idle_tx(ni, lnet_nid_to_nid4(&target->nid));
1713         if (tx == NULL) {
1714                 CERROR("Can't allocate %s txd for %s\n",
1715                         lnet_msgtyp2str(type),
1716                         libcfs_nidstr(&target->nid));
1717                 return -ENOMEM;
1718         }
1719         ibmsg = tx->tx_msg;
1720         gpu = lnet_md_is_gpu(msg_md);
1721
1722         switch (type) {
1723         default:
1724                 LBUG();
1725                 return (-EIO);
1726
1727         case LNET_MSG_ACK:
1728                 LASSERT(payload_nob == 0);
1729                 break;
1730
1731         case LNET_MSG_GET:
1732                 if (routing || target_is_router)
1733                         break;                  /* send IMMEDIATE */
1734
1735                 /* is the REPLY message too small for RDMA? */
1736                 nob = offsetof(struct kib_msg,
1737                                ibm_u.immediate.ibim_payload[msg_md->md_length]);
1738                 if (nob <= IBLND_MSG_SIZE && !gpu)
1739                         break;                  /* send IMMEDIATE */
1740
1741                 rd = &ibmsg->ibm_u.get.ibgm_rd;
1742                 tx->tx_gpu = gpu;
1743                 rc = kiblnd_setup_rd_kiov(ni, tx, rd,
1744                                           msg_md->md_niov,
1745                                           msg_md->md_kiov,
1746                                           0, msg_md->md_length);
1747                 if (rc != 0) {
1748                         CERROR("Can't setup GET sink for %s: %d\n",
1749                                libcfs_nidstr(&target->nid), rc);
1750                         tx->tx_hstatus = LNET_MSG_STATUS_LOCAL_ERROR;
1751                         kiblnd_tx_done(tx);
1752                         return -EIO;
1753                 }
1754
1755                 nob = offsetof(struct kib_get_msg, ibgm_rd.rd_frags[rd->rd_nfrags]);
1756                 ibmsg->ibm_u.get.ibgm_cookie = tx->tx_cookie;
1757                 lnet_hdr_to_nid4(hdr, &ibmsg->ibm_u.get.ibgm_hdr);
1758
1759                 kiblnd_init_tx_msg(ni, tx, IBLND_MSG_GET_REQ, nob);
1760
1761                 tx->tx_lntmsg[1] = lnet_create_reply_msg(ni, lntmsg);
1762                 if (tx->tx_lntmsg[1] == NULL) {
1763                         CERROR("Can't create reply for GET -> %s\n",
1764                                libcfs_nidstr(&target->nid));
1765                         kiblnd_tx_done(tx);
1766                         return -EIO;
1767                 }
1768
1769                 /* finalise lntmsg[0,1] on completion */
1770                 tx->tx_lntmsg[0] = lntmsg;
1771                 tx->tx_waiting = 1;             /* waiting for GET_DONE */
1772                 kiblnd_launch_tx(ni, tx, lnet_nid_to_nid4(&target->nid));
1773                 return 0;
1774
1775         case LNET_MSG_REPLY:
1776         case LNET_MSG_PUT:
1777                 /* Is the payload small enough not to need RDMA? */
1778                 nob = offsetof(struct kib_msg,
1779                                 ibm_u.immediate.ibim_payload[payload_nob]);
1780                 if (nob <= IBLND_MSG_SIZE && !gpu)
1781                         break;                  /* send IMMEDIATE */
1782
1783                 tx->tx_gpu = gpu;
1784
1785                 rc = kiblnd_setup_rd_kiov(ni, tx, tx->tx_rd,
1786                                           payload_niov, payload_kiov,
1787                                           payload_offset, payload_nob);
1788                 if (rc != 0) {
1789                         CERROR("Can't setup PUT src for %s: %d\n",
1790                                libcfs_nidstr(&target->nid), rc);
1791                         kiblnd_tx_done(tx);
1792                         return -EIO;
1793                 }
1794
1795                 lnet_hdr_to_nid4(hdr, &ibmsg->ibm_u.putreq.ibprm_hdr);
1796                 ibmsg->ibm_u.putreq.ibprm_cookie = tx->tx_cookie;
1797                 kiblnd_init_tx_msg(ni, tx, IBLND_MSG_PUT_REQ,
1798                                    sizeof(struct kib_putreq_msg));
1799
1800                 /* finalise lntmsg[0,1] on completion */
1801                 tx->tx_lntmsg[0] = lntmsg;
1802                 tx->tx_waiting = 1;             /* waiting for PUT_{ACK,NAK} */
1803                 kiblnd_launch_tx(ni, tx, lnet_nid_to_nid4(&target->nid));
1804                 return 0;
1805         }
1806
1807         /* send IMMEDIATE */
1808         LASSERT(offsetof(struct kib_msg, ibm_u.immediate.ibim_payload[payload_nob])
1809                 <= IBLND_MSG_SIZE);
1810
1811         ibmsg = tx->tx_msg;
1812         lnet_hdr_to_nid4(hdr, &ibmsg->ibm_u.immediate.ibim_hdr);
1813
1814         if (IS_FAST_REG_DEV(dev) && payload_nob)  {
1815                 struct ib_rdma_wr *wrq;
1816                 int i;
1817
1818                 nob = offsetof(struct kib_immediate_msg, ibim_payload[0]);
1819                 wrq = kiblnd_init_tx_msg_payload(ni, tx, IBLND_MSG_IMMEDIATE,
1820                                                  nob, payload_nob);
1821
1822                 rd = tx->tx_rd;
1823                 rc = kiblnd_setup_rd_kiov(ni, tx, rd,
1824                                           payload_niov, payload_kiov,
1825                                           payload_offset, payload_nob);
1826                 if (rc != 0) {
1827                         CERROR("Can't setup IMMEDIATE src for %s: %d\n",
1828                                libcfs_nidstr(&target->nid), rc);
1829                         kiblnd_tx_done(tx);
1830                         return -EIO;
1831                 }
1832
1833                 /* lets generate a SGE chain */
1834                 for (i = 0; i < rd->rd_nfrags; i++) {
1835                         kiblnd_init_tx_sge(tx, rd->rd_frags[i].rf_addr,
1836                                            rd->rd_frags[i].rf_nob);
1837                         wrq->wr.num_sge++;
1838                 }
1839         } else {
1840                 lnet_copy_kiov2flat(IBLND_MSG_SIZE, ibmsg,
1841                                     offsetof(struct kib_msg,
1842                                              ibm_u.immediate.ibim_payload),
1843                                     payload_niov, payload_kiov,
1844                                     payload_offset, payload_nob);
1845
1846                 nob = offsetof(struct kib_immediate_msg,
1847                                ibim_payload[payload_nob]);
1848
1849                 kiblnd_init_tx_msg(ni, tx, IBLND_MSG_IMMEDIATE, nob);
1850         }
1851
1852         /* finalise lntmsg on completion */
1853         tx->tx_lntmsg[0] = lntmsg;
1854
1855         kiblnd_launch_tx(ni, tx, lnet_nid_to_nid4(&target->nid));
1856         return 0;
1857 }
1858
1859 static void
1860 kiblnd_reply(struct lnet_ni *ni, struct kib_rx *rx, struct lnet_msg *lntmsg)
1861 {
1862         struct lnet_processid *target = &lntmsg->msg_target;
1863         unsigned int niov = lntmsg->msg_niov;
1864         struct bio_vec *kiov = lntmsg->msg_kiov;
1865         unsigned int offset = lntmsg->msg_offset;
1866         unsigned int nob = lntmsg->msg_len;
1867         struct lnet_libmd *msg_md = lntmsg->msg_md;
1868         struct kib_tx *tx;
1869         int rc;
1870
1871         tx = kiblnd_get_idle_tx(ni, rx->rx_conn->ibc_peer->ibp_nid);
1872         if (tx == NULL) {
1873                 CERROR("Can't get tx for REPLY to %s\n",
1874                        libcfs_nidstr(&target->nid));
1875                 goto failed_0;
1876         }
1877
1878         tx->tx_gpu = lnet_md_is_gpu(msg_md);
1879
1880         if (nob == 0)
1881                 rc = 0;
1882         else
1883                 rc = kiblnd_setup_rd_kiov(ni, tx, tx->tx_rd,
1884                                           niov, kiov, offset, nob);
1885
1886         if (rc != 0) {
1887                 CERROR("Can't setup GET src for %s: %d\n",
1888                        libcfs_nidstr(&target->nid), rc);
1889                 goto failed_1;
1890         }
1891
1892         rc = kiblnd_init_rdma(rx->rx_conn, tx,
1893                               IBLND_MSG_GET_DONE, nob,
1894                               &rx->rx_msg->ibm_u.get.ibgm_rd,
1895                               rx->rx_msg->ibm_u.get.ibgm_cookie);
1896         if (rc < 0) {
1897                 CERROR("Can't setup rdma for GET from %s: %d\n",
1898                        libcfs_nidstr(&target->nid), rc);
1899                 goto failed_1;
1900         }
1901
1902         if (nob == 0) {
1903                 /* No RDMA: local completion may happen now! */
1904                 lnet_finalize(lntmsg, 0);
1905         } else {
1906                 /* RDMA: lnet_finalize(lntmsg) when it
1907                  * completes */
1908                 tx->tx_lntmsg[0] = lntmsg;
1909         }
1910
1911         kiblnd_queue_tx(tx, rx->rx_conn);
1912         return;
1913
1914
1915 failed_1:
1916         tx->tx_hstatus = LNET_MSG_STATUS_LOCAL_ERROR;
1917         kiblnd_tx_done(tx);
1918 failed_0:
1919         lnet_finalize(lntmsg, -EIO);
1920 }
1921
1922 unsigned int
1923 kiblnd_get_dev_prio(struct lnet_ni *ni, unsigned int dev_idx)
1924 {
1925         struct kib_net *net = ni->ni_data;
1926         struct device *dev = NULL;
1927
1928         if (net)
1929                 dev = net->ibn_dev->ibd_hdev->ibh_ibdev->dma_device;
1930
1931         return lnet_get_dev_prio(dev, dev_idx);
1932
1933 }
1934
1935 int
1936 kiblnd_recv(struct lnet_ni *ni, void *private, struct lnet_msg *lntmsg,
1937             int delayed, unsigned int niov, struct bio_vec *kiov,
1938             unsigned int offset, unsigned int mlen, unsigned int rlen)
1939 {
1940         struct kib_rx *rx = private;
1941         struct kib_msg *rxmsg = rx->rx_msg;
1942         struct kib_conn *conn = rx->rx_conn;
1943         struct kib_tx *tx;
1944         __u64        ibprm_cookie;
1945         int          nob;
1946         int          post_credit = IBLND_POSTRX_PEER_CREDIT;
1947         int          rc = 0;
1948
1949         LASSERT (mlen <= rlen);
1950         LASSERT (!in_interrupt());
1951
1952         switch (rxmsg->ibm_type) {
1953         default:
1954                 LBUG();
1955                 /* fallthrough */
1956         case IBLND_MSG_IMMEDIATE:
1957                 nob = offsetof(struct kib_msg, ibm_u.immediate.ibim_payload[rlen]);
1958                 if (nob > rx->rx_nob) {
1959                         CERROR("Immediate message from %s too big: %d(%d)\n",
1960                                libcfs_nidstr(&lntmsg->msg_hdr.src_nid),
1961                                nob, rx->rx_nob);
1962                         rc = -EPROTO;
1963                         break;
1964                 }
1965
1966                 lnet_copy_flat2kiov(niov, kiov, offset,
1967                                     IBLND_MSG_SIZE, rxmsg,
1968                                     offsetof(struct kib_msg,
1969                                              ibm_u.immediate.ibim_payload),
1970                                     mlen);
1971                 lnet_finalize(lntmsg, 0);
1972                 break;
1973
1974         case IBLND_MSG_PUT_REQ: {
1975                 struct kib_msg  *txmsg;
1976                 struct kib_rdma_desc *rd;
1977                 struct lnet_libmd *msg_md = lntmsg->msg_md;
1978
1979                 ibprm_cookie = rxmsg->ibm_u.putreq.ibprm_cookie;
1980                 if (mlen == 0) {
1981                         lnet_finalize(lntmsg, 0);
1982                         kiblnd_send_completion(rx->rx_conn, IBLND_MSG_PUT_NAK,
1983                                                0, ibprm_cookie);
1984                         break;
1985                 }
1986
1987                 tx = kiblnd_get_idle_tx(ni, conn->ibc_peer->ibp_nid);
1988                 if (tx == NULL) {
1989                         CERROR("Can't allocate tx for %s\n",
1990                                 libcfs_nid2str(conn->ibc_peer->ibp_nid));
1991                         /* Not replying will break the connection */
1992                         rc = -ENOMEM;
1993                         break;
1994                 }
1995
1996                 tx->tx_gpu = lnet_md_is_gpu(msg_md);
1997
1998                 txmsg = tx->tx_msg;
1999                 rd = &txmsg->ibm_u.putack.ibpam_rd;
2000                 rc = kiblnd_setup_rd_kiov(ni, tx, rd,
2001                                           niov, kiov, offset, mlen);
2002                 if (rc != 0) {
2003                         CERROR("Can't setup PUT sink for %s: %d\n",
2004                                libcfs_nid2str(conn->ibc_peer->ibp_nid), rc);
2005                         tx->tx_hstatus = LNET_MSG_STATUS_LOCAL_ERROR;
2006                         kiblnd_tx_done(tx);
2007                         /* tell peer_ni it's over */
2008                         kiblnd_send_completion(rx->rx_conn, IBLND_MSG_PUT_NAK,
2009                                                rc, ibprm_cookie);
2010                         break;
2011                 }
2012
2013                 nob = offsetof(struct kib_putack_msg, ibpam_rd.rd_frags[rd->rd_nfrags]);
2014                 txmsg->ibm_u.putack.ibpam_src_cookie = ibprm_cookie;
2015                 txmsg->ibm_u.putack.ibpam_dst_cookie = tx->tx_cookie;
2016
2017                 kiblnd_init_tx_msg(ni, tx, IBLND_MSG_PUT_ACK, nob);
2018
2019                 tx->tx_lntmsg[0] = lntmsg;      /* finalise lntmsg on completion */
2020                 tx->tx_waiting = 1;             /* waiting for PUT_DONE */
2021                 kiblnd_queue_tx(tx, conn);
2022
2023                 /* reposted buffer reserved for PUT_DONE */
2024                 post_credit = IBLND_POSTRX_NO_CREDIT;
2025                 break;
2026                 }
2027
2028         case IBLND_MSG_GET_REQ:
2029                 if (lntmsg != NULL) {
2030                         /* Optimized GET; RDMA lntmsg's payload */
2031                         kiblnd_reply(ni, rx, lntmsg);
2032                 } else {
2033                         /* GET didn't match anything */
2034                         kiblnd_send_completion(rx->rx_conn, IBLND_MSG_GET_DONE,
2035                                                -ENODATA,
2036                                                rxmsg->ibm_u.get.ibgm_cookie);
2037                 }
2038                 break;
2039         }
2040
2041         kiblnd_post_rx(rx, post_credit);
2042         return rc;
2043 }
2044
2045 static void
2046 kiblnd_thread_fini (void)
2047 {
2048         atomic_dec (&kiblnd_data.kib_nthreads);
2049 }
2050
2051 static void
2052 kiblnd_peer_alive(struct kib_peer_ni *peer_ni)
2053 {
2054         /* This is racy, but everyone's only writing ktime_get_seconds() */
2055         peer_ni->ibp_last_alive = ktime_get_seconds();
2056         smp_mb();
2057 }
2058
2059 static void
2060 kiblnd_peer_notify(struct kib_peer_ni *peer_ni)
2061 {
2062         int error = 0;
2063         time64_t last_alive = 0;
2064         unsigned long flags;
2065
2066         read_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
2067
2068         if (kiblnd_peer_idle(peer_ni) && peer_ni->ibp_error != 0) {
2069                 error = peer_ni->ibp_error;
2070                 peer_ni->ibp_error = 0;
2071
2072                 last_alive = peer_ni->ibp_last_alive;
2073         }
2074
2075         read_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
2076
2077         if (error != 0) {
2078                 struct lnet_nid nid;
2079
2080                 lnet_nid4_to_nid(peer_ni->ibp_nid, &nid);
2081                 lnet_notify(peer_ni->ibp_ni, &nid,
2082                             false, false, last_alive);
2083         }
2084 }
2085
2086 void
2087 kiblnd_close_conn_locked(struct kib_conn *conn, int error)
2088 {
2089         /* This just does the immediate housekeeping.  'error' is zero for a
2090          * normal shutdown which can happen only after the connection has been
2091          * established.  If the connection is established, schedule the
2092          * connection to be finished off by the connd.  Otherwise the connd is
2093          * already dealing with it (either to set it up or tear it down).
2094          * Caller holds kib_global_lock exclusively in irq context */
2095         struct kib_peer_ni *peer_ni = conn->ibc_peer;
2096         struct kib_dev *dev;
2097         unsigned long flags;
2098
2099         LASSERT (error != 0 || conn->ibc_state >= IBLND_CONN_ESTABLISHED);
2100
2101         if (error != 0 && conn->ibc_comms_error == 0)
2102                 conn->ibc_comms_error = error;
2103
2104         if (conn->ibc_state != IBLND_CONN_ESTABLISHED)
2105                 return; /* already being handled  */
2106
2107         if (error == 0 &&
2108             list_empty(&conn->ibc_tx_noops) &&
2109             list_empty(&conn->ibc_tx_queue) &&
2110             list_empty(&conn->ibc_tx_queue_rsrvd) &&
2111             list_empty(&conn->ibc_tx_queue_nocred) &&
2112             list_empty(&conn->ibc_active_txs))
2113                 CDEBUG(D_NET, "closing conn %p to %s\n",
2114                        conn,
2115                        libcfs_nid2str(peer_ni->ibp_nid));
2116         else
2117                 CNETERR("Closing conn %p to %s: error %d%s%s%s%s%s\n",
2118                         conn,
2119                         libcfs_nid2str(peer_ni->ibp_nid), error,
2120                         list_empty(&conn->ibc_tx_queue) ? "" : "(sending)",
2121                         list_empty(&conn->ibc_tx_noops) ?
2122                                                 "" : "(sending_noops)",
2123                         list_empty(&conn->ibc_tx_queue_rsrvd) ?
2124                                                 "" : "(sending_rsrvd)",
2125                         list_empty(&conn->ibc_tx_queue_nocred) ?
2126                                                 "" : "(sending_nocred)",
2127                         list_empty(&conn->ibc_active_txs) ? "" : "(waiting)");
2128
2129         dev = ((struct kib_net *)peer_ni->ibp_ni->ni_data)->ibn_dev;
2130         if (peer_ni->ibp_next_conn == conn)
2131                 /* clear next_conn so it won't be used */
2132                 peer_ni->ibp_next_conn = NULL;
2133         list_del(&conn->ibc_list);
2134         /* connd (see below) takes over ibc_list's ref */
2135
2136         if (list_empty(&peer_ni->ibp_conns) &&    /* no more conns */
2137             kiblnd_peer_active(peer_ni)) {         /* still in peer_ni table */
2138                 kiblnd_unlink_peer_locked(peer_ni);
2139
2140                 /* set/clear error on last conn */
2141                 peer_ni->ibp_error = conn->ibc_comms_error;
2142         }
2143
2144         kiblnd_set_conn_state(conn, IBLND_CONN_CLOSING);
2145
2146         if (error != 0 &&
2147             kiblnd_dev_can_failover(dev)) {
2148                 list_add_tail(&dev->ibd_fail_list,
2149                               &kiblnd_data.kib_failed_devs);
2150                 wake_up(&kiblnd_data.kib_failover_waitq);
2151         }
2152
2153         spin_lock_irqsave(&kiblnd_data.kib_connd_lock, flags);
2154
2155         list_add_tail(&conn->ibc_list, &kiblnd_data.kib_connd_conns);
2156         wake_up(&kiblnd_data.kib_connd_waitq);
2157
2158         spin_unlock_irqrestore(&kiblnd_data.kib_connd_lock, flags);
2159 }
2160
2161 void
2162 kiblnd_close_conn(struct kib_conn *conn, int error)
2163 {
2164         unsigned long flags;
2165
2166         write_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
2167
2168         kiblnd_close_conn_locked(conn, error);
2169
2170         write_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
2171 }
2172
2173 static void
2174 kiblnd_handle_early_rxs(struct kib_conn *conn)
2175 {
2176         unsigned long flags;
2177         struct kib_rx *rx;
2178
2179         LASSERT(!in_interrupt());
2180         LASSERT(conn->ibc_state >= IBLND_CONN_ESTABLISHED);
2181
2182         write_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
2183         while ((rx = list_first_entry_or_null(&conn->ibc_early_rxs,
2184                                               struct kib_rx,
2185                                               rx_list)) != NULL) {
2186                 list_del(&rx->rx_list);
2187                 write_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
2188
2189                 kiblnd_handle_rx(rx);
2190
2191                 write_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
2192         }
2193         write_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
2194 }
2195
2196 void
2197 kiblnd_abort_txs(struct kib_conn *conn, struct list_head *txs)
2198 {
2199         LIST_HEAD(zombies);
2200         struct kib_tx *nxt;
2201         struct kib_tx *tx;
2202
2203         spin_lock(&conn->ibc_lock);
2204
2205         list_for_each_entry_safe(tx, nxt, txs, tx_list) {
2206                 if (txs == &conn->ibc_active_txs) {
2207                         LASSERT(!tx->tx_queued);
2208                         LASSERT(tx->tx_waiting ||
2209                                 tx->tx_sending != 0);
2210                         if (conn->ibc_comms_error == -ETIMEDOUT) {
2211                                 if (tx->tx_waiting && !tx->tx_sending)
2212                                         tx->tx_hstatus =
2213                                           LNET_MSG_STATUS_REMOTE_TIMEOUT;
2214                                 else if (tx->tx_sending)
2215                                         tx->tx_hstatus =
2216                                           LNET_MSG_STATUS_NETWORK_TIMEOUT;
2217                         }
2218                 } else {
2219                         LASSERT(tx->tx_queued);
2220                         if (conn->ibc_comms_error == -ETIMEDOUT)
2221                                 tx->tx_hstatus = LNET_MSG_STATUS_LOCAL_TIMEOUT;
2222                         else
2223                                 tx->tx_hstatus = LNET_MSG_STATUS_LOCAL_ERROR;
2224                 }
2225
2226                 tx->tx_status = -ECONNABORTED;
2227                 tx->tx_waiting = 0;
2228
2229                 /*
2230                  * TODO: This makes an assumption that
2231                  * kiblnd_tx_complete() will be called for each tx. If
2232                  * that event is dropped we could end up with stale
2233                  * connections floating around. We'd like to deal with
2234                  * that in a better way.
2235                  *
2236                  * Also that means we can exceed the timeout by many
2237                  * seconds.
2238                  */
2239                 if (tx->tx_sending == 0) {
2240                         tx->tx_queued = 0;
2241                         list_move(&tx->tx_list, &zombies);
2242                 } else {
2243                         /* keep tx until cq destroy */
2244                         list_move(&tx->tx_list, &conn->ibc_zombie_txs);
2245                         conn->ibc_waits ++;
2246                 }
2247         }
2248
2249         spin_unlock(&conn->ibc_lock);
2250
2251         /*
2252          * aborting transmits occurs when finalizing the connection.
2253          * The connection is finalized on error.
2254          * Passing LNET_MSG_STATUS_OK to txlist_done() will not
2255          * override the value already set in tx->tx_hstatus above.
2256          */
2257         kiblnd_txlist_done(&zombies, -ECONNABORTED, LNET_MSG_STATUS_OK);
2258 }
2259
2260 static bool
2261 kiblnd_tx_may_discard(struct kib_conn *conn)
2262 {
2263         bool rc = false;
2264         struct kib_tx *nxt;
2265         struct kib_tx *tx;
2266
2267         spin_lock(&conn->ibc_lock);
2268
2269         list_for_each_entry_safe(tx, nxt, &conn->ibc_zombie_txs, tx_list) {
2270                 if (tx->tx_sending > 0 && tx->tx_lntmsg[0] &&
2271                     lnet_md_discarded(tx->tx_lntmsg[0]->msg_md)) {
2272                         tx->tx_sending --;
2273                         if (tx->tx_sending == 0) {
2274                                 kiblnd_conn_decref(tx->tx_conn);
2275                                 tx->tx_conn = NULL;
2276                                 rc = true;
2277                         }
2278                 }
2279         }
2280
2281         spin_unlock(&conn->ibc_lock);
2282         return rc;
2283 }
2284
2285 static void
2286 kiblnd_finalise_conn(struct kib_conn *conn)
2287 {
2288         LASSERT (!in_interrupt());
2289         LASSERT (conn->ibc_state > IBLND_CONN_INIT);
2290
2291         /* abort_receives moves QP state to IB_QPS_ERR.  This is only required
2292          * for connections that didn't get as far as being connected, because
2293          * rdma_disconnect() does this for free. */
2294         kiblnd_abort_receives(conn);
2295
2296         kiblnd_set_conn_state(conn, IBLND_CONN_DISCONNECTED);
2297
2298         /* Complete all tx descs not waiting for sends to complete.
2299          * NB we should be safe from RDMA now that the QP has changed state */
2300
2301         CDEBUG(D_NET, "abort connection with %s\n", libcfs_nid2str(conn->ibc_peer->ibp_nid));
2302
2303         kiblnd_abort_txs(conn, &conn->ibc_tx_noops);
2304         kiblnd_abort_txs(conn, &conn->ibc_tx_queue);
2305         kiblnd_abort_txs(conn, &conn->ibc_tx_queue_rsrvd);
2306         kiblnd_abort_txs(conn, &conn->ibc_tx_queue_nocred);
2307         kiblnd_abort_txs(conn, &conn->ibc_active_txs);
2308
2309         kiblnd_handle_early_rxs(conn);
2310 }
2311
2312 static void
2313 kiblnd_peer_connect_failed(struct kib_peer_ni *peer_ni, int active,
2314                            int error)
2315 {
2316         LIST_HEAD(zombies);
2317         unsigned long flags;
2318         enum lnet_msg_hstatus hstatus;
2319
2320         LASSERT(error != 0);
2321         LASSERT(!in_interrupt());
2322
2323         write_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
2324
2325         if (active) {
2326                 LASSERT(peer_ni->ibp_connecting > 0);
2327                 peer_ni->ibp_connecting--;
2328         } else {
2329                 LASSERT (peer_ni->ibp_accepting > 0);
2330                 peer_ni->ibp_accepting--;
2331         }
2332
2333         if (kiblnd_peer_connecting(peer_ni)) {
2334                 /* another connection attempt under way... */
2335                 write_unlock_irqrestore(&kiblnd_data.kib_global_lock,
2336                                         flags);
2337                 return;
2338         }
2339
2340         peer_ni->ibp_reconnected = 0;
2341         if (list_empty(&peer_ni->ibp_conns)) {
2342                 /* Take peer_ni's blocked transmits to complete with error */
2343                 list_splice_init(&peer_ni->ibp_tx_queue, &zombies);
2344
2345                 if (kiblnd_peer_active(peer_ni))
2346                         kiblnd_unlink_peer_locked(peer_ni);
2347
2348                 peer_ni->ibp_error = error;
2349         } else {
2350                 /* Can't have blocked transmits if there are connections */
2351                 LASSERT(list_empty(&peer_ni->ibp_tx_queue));
2352         }
2353
2354         write_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
2355
2356         kiblnd_peer_notify(peer_ni);
2357
2358         if (list_empty(&zombies))
2359                 return;
2360
2361         CNETERR("Deleting messages for %s: connection failed\n",
2362                 libcfs_nid2str(peer_ni->ibp_nid));
2363
2364         switch (error) {
2365         case -EHOSTUNREACH:
2366         case -ETIMEDOUT:
2367                 hstatus = LNET_MSG_STATUS_NETWORK_TIMEOUT;
2368                 break;
2369         case -ECONNREFUSED:
2370                 hstatus = LNET_MSG_STATUS_REMOTE_DROPPED;
2371                 break;
2372         default:
2373                 hstatus = LNET_MSG_STATUS_LOCAL_DROPPED;
2374                 break;
2375         }
2376
2377         kiblnd_txlist_done(&zombies, error, hstatus);
2378 }
2379
2380 static void
2381 kiblnd_connreq_done(struct kib_conn *conn, int status)
2382 {
2383         struct kib_peer_ni *peer_ni = conn->ibc_peer;
2384         struct kib_tx *tx;
2385         LIST_HEAD(txs);
2386         unsigned long    flags;
2387         int              active;
2388
2389         active = (conn->ibc_state == IBLND_CONN_ACTIVE_CONNECT);
2390
2391         CDEBUG(D_NET,"%s: active(%d), version(%x), status(%d)\n",
2392                libcfs_nid2str(peer_ni->ibp_nid), active,
2393                conn->ibc_version, status);
2394
2395         LASSERT (!in_interrupt());
2396         LASSERT ((conn->ibc_state == IBLND_CONN_ACTIVE_CONNECT &&
2397                   peer_ni->ibp_connecting > 0) ||
2398                  (conn->ibc_state == IBLND_CONN_PASSIVE_WAIT &&
2399                   peer_ni->ibp_accepting > 0));
2400
2401         LIBCFS_FREE(conn->ibc_connvars, sizeof(*conn->ibc_connvars));
2402         conn->ibc_connvars = NULL;
2403
2404         if (status != 0) {
2405                 /* failed to establish connection */
2406                 kiblnd_peer_connect_failed(peer_ni, active, status);
2407                 kiblnd_finalise_conn(conn);
2408                 return;
2409         }
2410
2411         /* connection established */
2412         write_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
2413
2414         conn->ibc_last_send = ktime_get();
2415         kiblnd_set_conn_state(conn, IBLND_CONN_ESTABLISHED);
2416         kiblnd_peer_alive(peer_ni);
2417
2418         /* Add conn to peer_ni's list and nuke any dangling conns from a different
2419          * peer_ni instance... */
2420         kiblnd_conn_addref(conn);       /* +1 ref for ibc_list */
2421         list_add(&conn->ibc_list, &peer_ni->ibp_conns);
2422         peer_ni->ibp_reconnected = 0;
2423         if (active)
2424                 peer_ni->ibp_connecting--;
2425         else
2426                 peer_ni->ibp_accepting--;
2427
2428         if (peer_ni->ibp_version == 0) {
2429                 peer_ni->ibp_version     = conn->ibc_version;
2430                 peer_ni->ibp_incarnation = conn->ibc_incarnation;
2431         }
2432
2433         if (peer_ni->ibp_version     != conn->ibc_version ||
2434             peer_ni->ibp_incarnation != conn->ibc_incarnation) {
2435                 kiblnd_close_stale_conns_locked(peer_ni, conn->ibc_version,
2436                                                 conn->ibc_incarnation);
2437                 peer_ni->ibp_version     = conn->ibc_version;
2438                 peer_ni->ibp_incarnation = conn->ibc_incarnation;
2439         }
2440
2441         /* grab pending txs while I have the lock */
2442         list_splice_init(&peer_ni->ibp_tx_queue, &txs);
2443
2444         if (!kiblnd_peer_active(peer_ni) ||        /* peer_ni has been deleted */
2445             conn->ibc_comms_error != 0) {       /* error has happened already */
2446
2447                 /* start to shut down connection */
2448                 kiblnd_close_conn_locked(conn, -ECONNABORTED);
2449                 write_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
2450
2451                 kiblnd_txlist_done(&txs, -ECONNABORTED,
2452                                    LNET_MSG_STATUS_LOCAL_ERROR);
2453
2454                 return;
2455         }
2456
2457         /* +1 ref for myself, this connection is visible to other threads
2458          * now, refcount of peer:ibp_conns can be released by connection
2459          * close from either a different thread, or the calling of
2460          * kiblnd_check_sends_locked() below. See bz21911 for details.
2461          */
2462         kiblnd_conn_addref(conn);
2463         write_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
2464
2465         /* Schedule blocked txs
2466          * Note: if we are running with conns_per_peer > 1, these blocked
2467          * txs will all get scheduled to the first connection which gets
2468          * scheduled.  We won't be using round robin on this first batch.
2469          */
2470         spin_lock(&conn->ibc_lock);
2471         while ((tx = list_first_entry_or_null(&txs, struct kib_tx,
2472                                               tx_list)) != NULL) {
2473                 list_del(&tx->tx_list);
2474
2475                 kiblnd_queue_tx_locked(tx, conn);
2476         }
2477         kiblnd_check_sends_locked(conn);
2478         spin_unlock(&conn->ibc_lock);
2479
2480         /* schedule blocked rxs */
2481         kiblnd_handle_early_rxs(conn);
2482         kiblnd_conn_decref(conn);
2483 }
2484
2485 static void
2486 kiblnd_reject(struct rdma_cm_id *cmid, struct kib_rej *rej)
2487 {
2488         int          rc;
2489
2490 #ifdef HAVE_OFED_RDMA_REJECT_4ARGS
2491         rc = rdma_reject(cmid, rej, sizeof(*rej), IB_CM_REJ_CONSUMER_DEFINED);
2492 #else
2493         rc = rdma_reject(cmid, rej, sizeof(*rej));
2494 #endif
2495
2496         if (rc != 0)
2497                 CWARN("Error %d sending reject\n", rc);
2498 }
2499
2500 static int
2501 kiblnd_passive_connect(struct rdma_cm_id *cmid, void *priv, int priv_nob)
2502 {
2503         rwlock_t *g_lock = &kiblnd_data.kib_global_lock;
2504         struct kib_msg *reqmsg = priv;
2505         struct kib_msg *ackmsg;
2506         struct kib_dev *ibdev;
2507         struct kib_peer_ni *peer_ni;
2508         struct kib_peer_ni *peer2;
2509         struct kib_conn *conn;
2510         struct lnet_ni *ni = NULL;
2511         struct kib_net *net = NULL;
2512         struct lnet_nid destnid;
2513         lnet_nid_t nid;
2514         struct rdma_conn_param cp;
2515         struct kib_rej rej;
2516         int version = IBLND_MSG_VERSION;
2517         unsigned long flags;
2518         int rc;
2519         struct sockaddr_in *peer_addr;
2520
2521         LASSERT(!in_interrupt());
2522         /* cmid inherits 'context' from the corresponding listener id */
2523         ibdev = cmid->context;
2524         LASSERT(ibdev);
2525
2526         memset(&rej, 0, sizeof(rej));
2527         rej.ibr_magic                = IBLND_MSG_MAGIC;
2528         rej.ibr_why                  = IBLND_REJECT_FATAL;
2529         rej.ibr_cp.ibcp_max_msg_size = IBLND_MSG_SIZE;
2530
2531         peer_addr = (struct sockaddr_in *)&(cmid->route.addr.dst_addr);
2532         if (*kiblnd_tunables.kib_require_priv_port &&
2533             ntohs(peer_addr->sin_port) >= PROT_SOCK) {
2534                 __u32 ip = ntohl(peer_addr->sin_addr.s_addr);
2535                 CERROR("peer_ni's port (%pI4h:%hu) is not privileged\n",
2536                        &ip, ntohs(peer_addr->sin_port));
2537                 goto failed;
2538         }
2539
2540         if (priv_nob < offsetof(struct kib_msg, ibm_type)) {
2541                 CERROR("Short connection request\n");
2542                 goto failed;
2543         }
2544
2545         /* Future protocol version compatibility support!  If the
2546          * o2iblnd-specific protocol changes, or when LNET unifies
2547          * protocols over all LNDs, the initial connection will
2548          * negotiate a protocol version.  I trap this here to avoid
2549          * console errors; the reject tells the peer_ni which protocol I
2550          * speak. */
2551         if (reqmsg->ibm_magic == LNET_PROTO_MAGIC ||
2552             reqmsg->ibm_magic == __swab32(LNET_PROTO_MAGIC))
2553                 goto failed;
2554         if (reqmsg->ibm_magic == IBLND_MSG_MAGIC &&
2555             reqmsg->ibm_version != IBLND_MSG_VERSION &&
2556             reqmsg->ibm_version != IBLND_MSG_VERSION_1)
2557                 goto failed;
2558         if (reqmsg->ibm_magic == __swab32(IBLND_MSG_MAGIC) &&
2559             reqmsg->ibm_version != __swab16(IBLND_MSG_VERSION) &&
2560             reqmsg->ibm_version != __swab16(IBLND_MSG_VERSION_1))
2561                 goto failed;
2562
2563         rc = kiblnd_unpack_msg(reqmsg, priv_nob);
2564         if (rc != 0) {
2565                 CERROR("Can't parse connection request: %d\n", rc);
2566                 goto failed;
2567         }
2568
2569         nid = reqmsg->ibm_srcnid;
2570         lnet_nid4_to_nid(reqmsg->ibm_dstnid, &destnid);
2571         ni  = lnet_nid_to_ni_addref(&destnid);
2572
2573         if (ni != NULL) {
2574                 net = (struct kib_net *)ni->ni_data;
2575                 rej.ibr_incarnation = net->ibn_incarnation;
2576         } else {
2577                 if (ibdev->ibd_nnets == 0) {
2578                         rej.ibr_why = IBLND_REJECT_EARLY;
2579                         CNETERR("Can't accept conn from %s (%s:%d:%pI4h): net for nid %s not added yet\n",
2580                                 libcfs_nid2str(nid),
2581                                 ibdev->ibd_ifname, ibdev->ibd_nnets,
2582                                 &ibdev->ibd_ifip,
2583                                 libcfs_nid2str(reqmsg->ibm_dstnid));
2584                         goto failed;
2585                 }
2586                 list_for_each_entry(net, &ibdev->ibd_nets, ibn_list) {
2587                         if ((net->ibn_dev == ibdev) &&
2588                             (net->ibn_ni != NULL) &&
2589                             (net->ibn_ni->ni_state != LNET_NI_STATE_ACTIVE)) {
2590                                 rej.ibr_why = IBLND_REJECT_EARLY;
2591                                 CNETERR("Can't accept conn from %s on %s (%s:%d:%pI4h): nid %s not ready\n",
2592                                        libcfs_nid2str(nid),
2593                                        libcfs_nidstr(&net->ibn_ni->ni_nid),
2594                                        ibdev->ibd_ifname, ibdev->ibd_nnets,
2595                                        &ibdev->ibd_ifip,
2596                                        libcfs_nid2str(reqmsg->ibm_dstnid));
2597                                 goto failed;
2598                         }
2599                 }
2600         }
2601
2602         if (ni == NULL ||                       /* no matching net */
2603             !nid_same(&ni->ni_nid, &destnid) || /* right NET, wrong NID! */
2604             net->ibn_dev != ibdev) {            /* wrong device */
2605                 CERROR("Can't accept conn from %s on %s (%s:%d:%pI4h): bad dst nid %s\n",
2606                        libcfs_nid2str(nid),
2607                        ni ? libcfs_nidstr(&ni->ni_nid) : "NA",
2608                        ibdev->ibd_ifname, ibdev->ibd_nnets,
2609                        &ibdev->ibd_ifip,
2610                        libcfs_nid2str(reqmsg->ibm_dstnid));
2611
2612                 goto failed;
2613         }
2614
2615         /* check time stamp as soon as possible */
2616         if (reqmsg->ibm_dststamp != 0 &&
2617             reqmsg->ibm_dststamp != net->ibn_incarnation) {
2618                 CWARN("Stale connection request\n");
2619                 rej.ibr_why = IBLND_REJECT_CONN_STALE;
2620                 goto failed;
2621         }
2622
2623         /* I can accept peer_ni's version */
2624         version = reqmsg->ibm_version;
2625
2626         if (reqmsg->ibm_type != IBLND_MSG_CONNREQ) {
2627                 CERROR("Unexpected connreq msg type: %x from %s\n",
2628                        reqmsg->ibm_type, libcfs_nid2str(nid));
2629                 goto failed;
2630         }
2631
2632         if (reqmsg->ibm_u.connparams.ibcp_queue_depth >
2633             kiblnd_msg_queue_size(version, ni)) {
2634                 CERROR("Can't accept conn from %s, queue depth too large:  %d (<=%d wanted)\n",
2635                        libcfs_nid2str(nid),
2636                        reqmsg->ibm_u.connparams.ibcp_queue_depth,
2637                        kiblnd_msg_queue_size(version, ni));
2638
2639                 if (version == IBLND_MSG_VERSION)
2640                         rej.ibr_why = IBLND_REJECT_MSG_QUEUE_SIZE;
2641
2642                 goto failed;
2643         }
2644
2645         if (reqmsg->ibm_u.connparams.ibcp_max_frags >
2646             IBLND_MAX_RDMA_FRAGS) {
2647                 CWARN("Can't accept conn from %s (version %x): max_frags %d too large (%d wanted)\n",
2648                       libcfs_nid2str(nid), version,
2649                       reqmsg->ibm_u.connparams.ibcp_max_frags,
2650                       IBLND_MAX_RDMA_FRAGS);
2651
2652                 if (version >= IBLND_MSG_VERSION)
2653                         rej.ibr_why = IBLND_REJECT_RDMA_FRAGS;
2654
2655                 goto failed;
2656         } else if (reqmsg->ibm_u.connparams.ibcp_max_frags <
2657                    IBLND_MAX_RDMA_FRAGS &&
2658                    net->ibn_fmr_ps == NULL) {
2659                 CWARN("Can't accept conn from %s (version %x): max_frags %d incompatible without FMR pool (%d wanted)\n",
2660                       libcfs_nid2str(nid), version,
2661                       reqmsg->ibm_u.connparams.ibcp_max_frags,
2662                       IBLND_MAX_RDMA_FRAGS);
2663
2664                 if (version == IBLND_MSG_VERSION)
2665                         rej.ibr_why = IBLND_REJECT_RDMA_FRAGS;
2666
2667                 goto failed;
2668         }
2669
2670         if (reqmsg->ibm_u.connparams.ibcp_max_msg_size > IBLND_MSG_SIZE) {
2671                 CERROR("Can't accept %s: message size %d too big (%d max)\n",
2672                        libcfs_nid2str(nid),
2673                        reqmsg->ibm_u.connparams.ibcp_max_msg_size,
2674                        IBLND_MSG_SIZE);
2675                 goto failed;
2676         }
2677
2678         /* assume 'nid' is a new peer_ni; create  */
2679         rc = kiblnd_create_peer(ni, &peer_ni, nid);
2680         if (rc != 0) {
2681                 CERROR("Can't create peer_ni for %s\n", libcfs_nid2str(nid));
2682                 rej.ibr_why = IBLND_REJECT_NO_RESOURCES;
2683                 goto failed;
2684         }
2685
2686         /* We have validated the peer's parameters so use those */
2687         peer_ni->ibp_max_frags = reqmsg->ibm_u.connparams.ibcp_max_frags;
2688         peer_ni->ibp_queue_depth = reqmsg->ibm_u.connparams.ibcp_queue_depth;
2689
2690         write_lock_irqsave(g_lock, flags);
2691
2692         peer2 = kiblnd_find_peer_locked(ni, nid);
2693         if (peer2 != NULL) {
2694                 if (peer2->ibp_version == 0) {
2695                         peer2->ibp_version     = version;
2696                         peer2->ibp_incarnation = reqmsg->ibm_srcstamp;
2697                 }
2698
2699                 /* not the guy I've talked with */
2700                 if (peer2->ibp_incarnation != reqmsg->ibm_srcstamp ||
2701                     peer2->ibp_version     != version) {
2702                         kiblnd_close_peer_conns_locked(peer2, -ESTALE);
2703
2704                         if (kiblnd_peer_active(peer2)) {
2705                                 peer2->ibp_incarnation = reqmsg->ibm_srcstamp;
2706                                 peer2->ibp_version = version;
2707                         }
2708                         write_unlock_irqrestore(g_lock, flags);
2709
2710                         CWARN("Conn stale %s version %x/%x incarnation %llu/%llu\n",
2711                               libcfs_nid2str(nid), peer2->ibp_version, version,
2712                               peer2->ibp_incarnation, reqmsg->ibm_srcstamp);
2713
2714                         kiblnd_peer_decref(peer_ni);
2715                         rej.ibr_why = IBLND_REJECT_CONN_STALE;
2716                         goto failed;
2717                 }
2718
2719                 /* Tie-break connection race in favour of the higher NID.
2720                  * If we keep running into a race condition multiple times,
2721                  * we have to assume that the connection attempt with the
2722                  * higher NID is stuck in a connecting state and will never
2723                  * recover.  As such, we pass through this if-block and let
2724                  * the lower NID connection win so we can move forward.
2725                  */
2726                 if (peer2->ibp_connecting != 0 &&
2727                     nid < lnet_nid_to_nid4(&ni->ni_nid) &&
2728                     peer2->ibp_races < MAX_CONN_RACES_BEFORE_ABORT) {
2729                         peer2->ibp_races++;
2730                         write_unlock_irqrestore(g_lock, flags);
2731
2732                         CDEBUG(D_NET, "Conn race %s\n",
2733                                libcfs_nid2str(peer2->ibp_nid));
2734
2735                         kiblnd_peer_decref(peer_ni);
2736                         rej.ibr_why = IBLND_REJECT_CONN_RACE;
2737                         goto failed;
2738                 }
2739                 if (peer2->ibp_races >= MAX_CONN_RACES_BEFORE_ABORT)
2740                         CNETERR("Conn race %s: unresolved after %d attempts, letting lower NID win\n",
2741                                 libcfs_nid2str(peer2->ibp_nid),
2742                                 MAX_CONN_RACES_BEFORE_ABORT);
2743                 /*
2744                  * passive connection is allowed even this peer_ni is waiting for
2745                  * reconnection.
2746                  */
2747                 peer2->ibp_reconnecting = 0;
2748                 peer2->ibp_races = 0;
2749                 peer2->ibp_accepting++;
2750                 kiblnd_peer_addref(peer2);
2751
2752                 /* Race with kiblnd_launch_tx (active connect) to create peer_ni
2753                  * so copy validated parameters since we now know what the
2754                  * peer_ni's limits are */
2755                 peer2->ibp_max_frags = peer_ni->ibp_max_frags;
2756                 peer2->ibp_queue_depth = peer_ni->ibp_queue_depth;
2757
2758                 write_unlock_irqrestore(g_lock, flags);
2759                 kiblnd_peer_decref(peer_ni);
2760                 peer_ni = peer2;
2761         } else {
2762                 /* Brand new peer_ni */
2763                 LASSERT(peer_ni->ibp_accepting == 0);
2764                 LASSERT(peer_ni->ibp_version == 0 &&
2765                         peer_ni->ibp_incarnation == 0);
2766
2767                 peer_ni->ibp_accepting   = 1;
2768                 peer_ni->ibp_version     = version;
2769                 peer_ni->ibp_incarnation = reqmsg->ibm_srcstamp;
2770
2771                 /* I have a ref on ni that prevents it being shutdown */
2772                 LASSERT(net->ibn_shutdown == 0);
2773
2774                 kiblnd_peer_addref(peer_ni);
2775                 hash_add(kiblnd_data.kib_peers, &peer_ni->ibp_list, nid);
2776
2777                 write_unlock_irqrestore(g_lock, flags);
2778         }
2779
2780         conn = kiblnd_create_conn(peer_ni, cmid, IBLND_CONN_PASSIVE_WAIT,
2781                                   version);
2782         if (!conn) {
2783                 kiblnd_peer_connect_failed(peer_ni, 0, -ENOMEM);
2784                 kiblnd_peer_decref(peer_ni);
2785                 rej.ibr_why = IBLND_REJECT_NO_RESOURCES;
2786                 goto failed;
2787         }
2788
2789         /* conn now "owns" cmid, so I return success from here on to ensure the
2790          * CM callback doesn't destroy cmid.
2791          */
2792         conn->ibc_incarnation      = reqmsg->ibm_srcstamp;
2793         conn->ibc_credits          = conn->ibc_queue_depth;
2794         conn->ibc_reserved_credits = conn->ibc_queue_depth;
2795         LASSERT(conn->ibc_credits + conn->ibc_reserved_credits +
2796                 IBLND_OOB_MSGS(version) <= IBLND_RX_MSGS(conn));
2797
2798         ackmsg = &conn->ibc_connvars->cv_msg;
2799         memset(ackmsg, 0, sizeof(*ackmsg));
2800
2801         kiblnd_init_msg(ackmsg, IBLND_MSG_CONNACK,
2802                         sizeof(ackmsg->ibm_u.connparams));
2803         ackmsg->ibm_u.connparams.ibcp_queue_depth  = conn->ibc_queue_depth;
2804         ackmsg->ibm_u.connparams.ibcp_max_frags    = conn->ibc_max_frags;
2805         ackmsg->ibm_u.connparams.ibcp_max_msg_size = IBLND_MSG_SIZE;
2806
2807         kiblnd_pack_msg(ni, ackmsg, version, 0, nid, reqmsg->ibm_srcstamp);
2808
2809         memset(&cp, 0, sizeof(cp));
2810         cp.private_data        = ackmsg;
2811         cp.private_data_len    = ackmsg->ibm_nob;
2812         cp.responder_resources = 0;            /* No atomic ops or RDMA reads */
2813         cp.initiator_depth     = 0;
2814         cp.flow_control        = 1;
2815         cp.retry_count         = *kiblnd_tunables.kib_retry_count;
2816         cp.rnr_retry_count     = *kiblnd_tunables.kib_rnr_retry_count;
2817
2818         CDEBUG(D_NET, "Accept %s conn %p\n", libcfs_nid2str(nid), conn);
2819
2820         rc = rdma_accept(cmid, &cp);
2821         if (rc != 0) {
2822                 CNETERR("Can't accept %s: %d cm_id %p\n", libcfs_nid2str(nid), rc, cmid);
2823                 rej.ibr_version = version;
2824                 rej.ibr_why     = IBLND_REJECT_FATAL;
2825
2826                 kiblnd_reject(cmid, &rej);
2827                 kiblnd_connreq_done(conn, rc);
2828                 kiblnd_conn_decref(conn);
2829         }
2830
2831         lnet_ni_decref(ni);
2832         return 0;
2833
2834  failed:
2835         if (ni != NULL) {
2836                 rej.ibr_cp.ibcp_queue_depth =
2837                         kiblnd_msg_queue_size(version, ni);
2838                 rej.ibr_cp.ibcp_max_frags   = IBLND_MAX_RDMA_FRAGS;
2839                 lnet_ni_decref(ni);
2840         }
2841
2842         rej.ibr_version = version;
2843         kiblnd_reject(cmid, &rej);
2844
2845         return -ECONNREFUSED;
2846 }
2847
2848 static void
2849 kiblnd_check_reconnect(struct kib_conn *conn, int version,
2850                        u64 incarnation, int why, struct kib_connparams *cp)
2851 {
2852         rwlock_t        *glock = &kiblnd_data.kib_global_lock;
2853         struct kib_peer_ni *peer_ni = conn->ibc_peer;
2854         char            *reason;
2855         int              msg_size = IBLND_MSG_SIZE;
2856         int              frag_num = -1;
2857         int              queue_dep = -1;
2858         bool             reconnect;
2859         unsigned long    flags;
2860
2861         LASSERT(conn->ibc_state == IBLND_CONN_ACTIVE_CONNECT);
2862         LASSERT(peer_ni->ibp_connecting > 0);   /* 'conn' at least */
2863
2864         if (cp) {
2865                 msg_size        = cp->ibcp_max_msg_size;
2866                 frag_num        = cp->ibcp_max_frags;
2867                 queue_dep       = cp->ibcp_queue_depth;
2868         }
2869
2870         write_lock_irqsave(glock, flags);
2871         /* retry connection if it's still needed and no other connection
2872          * attempts (active or passive) are in progress
2873          * NB: reconnect is still needed even when ibp_tx_queue is
2874          * empty if ibp_version != version because reconnect may be
2875          * initiated.
2876          */
2877         reconnect = (!list_empty(&peer_ni->ibp_tx_queue) ||
2878                      peer_ni->ibp_version != version) &&
2879                     peer_ni->ibp_connecting &&
2880                     peer_ni->ibp_accepting == 0;
2881         if (!reconnect) {
2882                 reason = "no need";
2883                 goto out;
2884         }
2885
2886         switch (why) {
2887         default:
2888                 reason = "Unknown";
2889                 break;
2890
2891         case IBLND_REJECT_RDMA_FRAGS: {
2892                 if (!cp) {
2893                         reason = "can't negotiate max frags";
2894                         goto out;
2895                 }
2896
2897                 if (conn->ibc_max_frags <= frag_num) {
2898                         reason = "unsupported max frags";
2899                         goto out;
2900                 }
2901
2902                 peer_ni->ibp_max_frags = frag_num;
2903                 reason = "rdma fragments";
2904                 break;
2905         }
2906         case IBLND_REJECT_MSG_QUEUE_SIZE:
2907                 if (!cp) {
2908                         reason = "can't negotiate queue depth";
2909                         goto out;
2910                 }
2911                 if (conn->ibc_queue_depth <= queue_dep) {
2912                         reason = "unsupported queue depth";
2913                         goto out;
2914                 }
2915
2916                 peer_ni->ibp_queue_depth = queue_dep;
2917                 reason = "queue depth";
2918                 break;
2919
2920         case IBLND_REJECT_CONN_STALE:
2921                 reason = "stale";
2922                 break;
2923
2924         case IBLND_REJECT_CONN_RACE:
2925                 reason = "conn race";
2926                 break;
2927
2928         case IBLND_REJECT_CONN_UNCOMPAT:
2929                 reason = "version negotiation";
2930                 break;
2931         }
2932
2933         conn->ibc_reconnect = 1;
2934         peer_ni->ibp_reconnecting++;
2935         peer_ni->ibp_version = version;
2936         if (incarnation != 0)
2937                 peer_ni->ibp_incarnation = incarnation;
2938  out:
2939         write_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
2940
2941         CNETERR("%s: %s (%s), %x, %x, msg_size: %d, queue_depth: %d/%d, max_frags: %d/%d\n",
2942                 libcfs_nid2str(peer_ni->ibp_nid),
2943                 reconnect ? "reconnect" : "don't reconnect",
2944                 reason, IBLND_MSG_VERSION, version, msg_size,
2945                 conn->ibc_queue_depth, queue_dep,
2946                 conn->ibc_max_frags, frag_num);
2947         /*
2948          * if conn::ibc_reconnect is TRUE, connd will reconnect to the peer_ni
2949          * while destroying the zombie
2950          */
2951 }
2952
2953 static void
2954 kiblnd_rejected(struct kib_conn *conn, int reason, void *priv, int priv_nob)
2955 {
2956         struct kib_peer_ni *peer_ni = conn->ibc_peer;
2957         int status = -ECONNREFUSED;
2958
2959         LASSERT (!in_interrupt());
2960         LASSERT (conn->ibc_state == IBLND_CONN_ACTIVE_CONNECT);
2961
2962         switch (reason) {
2963         case IB_CM_REJ_STALE_CONN:
2964                 kiblnd_check_reconnect(conn, IBLND_MSG_VERSION, 0,
2965                                        IBLND_REJECT_CONN_STALE, NULL);
2966                 break;
2967
2968         case IB_CM_REJ_INVALID_SERVICE_ID:
2969                 status = -EHOSTUNREACH;
2970                 CNETERR("%s rejected: no listener at %d\n",
2971                         libcfs_nid2str(peer_ni->ibp_nid),
2972                         *kiblnd_tunables.kib_service);
2973                 break;
2974
2975         case IB_CM_REJ_CONSUMER_DEFINED:
2976                 if (priv_nob >= offsetof(struct kib_rej, ibr_padding)) {
2977                         struct kib_rej *rej = priv;
2978                         struct kib_connparams *cp = NULL;
2979                         bool flip = false;
2980                         __u64 incarnation = -1;
2981
2982                         /* NB. default incarnation is -1 because:
2983                          * a) V1 will ignore dst incarnation in connreq.
2984                          * b) V2 will provide incarnation while rejecting me,
2985                          *    -1 will be overwrote.
2986                          *
2987                          * if I try to connect to a V1 peer_ni with V2 protocol,
2988                          * it rejected me then upgrade to V2, I have no idea
2989                          * about the upgrading and try to reconnect with V1,
2990                          * in this case upgraded V2 can find out I'm trying to
2991                          * talk to the old guy and reject me(incarnation is -1).
2992                          */
2993
2994                         if (rej->ibr_magic == __swab32(IBLND_MSG_MAGIC) ||
2995                             rej->ibr_magic == __swab32(LNET_PROTO_MAGIC)) {
2996                                 __swab32s(&rej->ibr_magic);
2997                                 __swab16s(&rej->ibr_version);
2998                                 flip = true;
2999                         }
3000
3001                         if (priv_nob >= sizeof(struct kib_rej) &&
3002                             rej->ibr_version > IBLND_MSG_VERSION_1) {
3003                                 /* priv_nob is always 148 in current version
3004                                  * of OFED, so we still need to check version.
3005                                  * (define of IB_CM_REJ_PRIVATE_DATA_SIZE)
3006                                  */
3007                                 cp = &rej->ibr_cp;
3008
3009                                 if (flip) {
3010                                         __swab64s(&rej->ibr_incarnation);
3011                                         __swab16s(&cp->ibcp_queue_depth);
3012                                         __swab16s(&cp->ibcp_max_frags);
3013                                         __swab32s(&cp->ibcp_max_msg_size);
3014                                 }
3015
3016                                 incarnation = rej->ibr_incarnation;
3017                         }
3018
3019                         if (rej->ibr_magic != IBLND_MSG_MAGIC &&
3020                             rej->ibr_magic != LNET_PROTO_MAGIC) {
3021                                 CERROR("%s rejected: consumer defined fatal error\n",
3022                                        libcfs_nid2str(peer_ni->ibp_nid));
3023                                 break;
3024                         }
3025
3026                         if (rej->ibr_version != IBLND_MSG_VERSION &&
3027                             rej->ibr_version != IBLND_MSG_VERSION_1) {
3028                                 CERROR("%s rejected: o2iblnd version %x error\n",
3029                                        libcfs_nid2str(peer_ni->ibp_nid),
3030                                        rej->ibr_version);
3031                                 break;
3032                         }
3033
3034                         if (rej->ibr_why     == IBLND_REJECT_FATAL &&
3035                             rej->ibr_version == IBLND_MSG_VERSION_1) {
3036                                 CDEBUG(D_NET, "rejected by old version peer_ni %s: %x\n",
3037                                        libcfs_nid2str(peer_ni->ibp_nid),
3038                                        rej->ibr_version);
3039
3040                                 if (conn->ibc_version != IBLND_MSG_VERSION_1)
3041                                         rej->ibr_why = IBLND_REJECT_CONN_UNCOMPAT;
3042                         }
3043
3044                         switch (rej->ibr_why) {
3045                         case IBLND_REJECT_CONN_RACE:
3046                         case IBLND_REJECT_CONN_STALE:
3047                         case IBLND_REJECT_CONN_UNCOMPAT:
3048                         case IBLND_REJECT_MSG_QUEUE_SIZE:
3049                         case IBLND_REJECT_RDMA_FRAGS:
3050                                 kiblnd_check_reconnect(conn, rej->ibr_version,
3051                                                        incarnation,
3052                                                        rej->ibr_why, cp);
3053                                 break;
3054
3055                         case IBLND_REJECT_NO_RESOURCES:
3056                                 CERROR("%s rejected: o2iblnd no resources\n",
3057                                        libcfs_nid2str(peer_ni->ibp_nid));
3058                                 break;
3059
3060                         case IBLND_REJECT_FATAL:
3061                                 CERROR("%s rejected: o2iblnd fatal error\n",
3062                                        libcfs_nid2str(peer_ni->ibp_nid));
3063                                 break;
3064
3065                         case IBLND_REJECT_EARLY:
3066                                 CNETERR("%s rejected: tried too early\n",
3067                                        libcfs_nid2str(peer_ni->ibp_nid));
3068                                 break;
3069
3070                         default:
3071                                 CERROR("%s rejected: o2iblnd reason %d\n",
3072                                        libcfs_nid2str(peer_ni->ibp_nid),
3073                                        rej->ibr_why);
3074                                 break;
3075                         }
3076                         break;
3077                 }
3078                 fallthrough;
3079         default:
3080                 CNETERR("%s rejected: reason %d, size %d\n",
3081                         libcfs_nid2str(peer_ni->ibp_nid), reason, priv_nob);
3082                 break;
3083         }
3084
3085         kiblnd_connreq_done(conn, status);
3086 }
3087
3088 static void
3089 kiblnd_check_connreply(struct kib_conn *conn, void *priv, int priv_nob)
3090 {
3091         struct kib_peer_ni *peer_ni = conn->ibc_peer;
3092         struct lnet_ni *ni = peer_ni->ibp_ni;
3093         struct kib_net *net = ni->ni_data;
3094         struct kib_msg *msg = priv;
3095         int            ver  = conn->ibc_version;
3096         int            rc   = kiblnd_unpack_msg(msg, priv_nob);
3097         unsigned long  flags;
3098
3099         LASSERT (net != NULL);
3100
3101         if (rc != 0) {
3102                 CERROR("Can't unpack connack from %s: %d\n",
3103                        libcfs_nid2str(peer_ni->ibp_nid), rc);
3104                 goto failed;
3105         }
3106
3107         if (msg->ibm_type != IBLND_MSG_CONNACK) {
3108                 CERROR("Unexpected message %d from %s\n",
3109                        msg->ibm_type, libcfs_nid2str(peer_ni->ibp_nid));
3110                 rc = -EPROTO;
3111                 goto failed;
3112         }
3113
3114         if (ver != msg->ibm_version) {
3115                 CERROR("%s replied version %x is different with "
3116                        "requested version %x\n",
3117                        libcfs_nid2str(peer_ni->ibp_nid), msg->ibm_version, ver);
3118                 rc = -EPROTO;
3119                 goto failed;
3120         }
3121
3122         if (msg->ibm_u.connparams.ibcp_queue_depth >
3123             conn->ibc_queue_depth) {
3124                 CERROR("%s has incompatible queue depth %d (<=%d wanted)\n",
3125                        libcfs_nid2str(peer_ni->ibp_nid),
3126                        msg->ibm_u.connparams.ibcp_queue_depth,
3127                        conn->ibc_queue_depth);
3128                 rc = -EPROTO;
3129                 goto failed;
3130         }
3131
3132         if (msg->ibm_u.connparams.ibcp_max_frags >
3133             conn->ibc_max_frags) {
3134                 CERROR("%s has incompatible max_frags %d (<=%d wanted)\n",
3135                        libcfs_nid2str(peer_ni->ibp_nid),
3136                        msg->ibm_u.connparams.ibcp_max_frags,
3137                        conn->ibc_max_frags);
3138                 rc = -EPROTO;
3139                 goto failed;
3140         }
3141
3142         if (msg->ibm_u.connparams.ibcp_max_msg_size > IBLND_MSG_SIZE) {
3143                 CERROR("%s max message size %d too big (%d max)\n",
3144                        libcfs_nid2str(peer_ni->ibp_nid),
3145                        msg->ibm_u.connparams.ibcp_max_msg_size,
3146                        IBLND_MSG_SIZE);
3147                 rc = -EPROTO;
3148                 goto failed;
3149         }
3150
3151         read_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
3152         if (msg->ibm_dstnid == lnet_nid_to_nid4(&ni->ni_nid) &&
3153             msg->ibm_dststamp == net->ibn_incarnation)
3154                 rc = 0;
3155         else
3156                 rc = -ESTALE;
3157         read_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
3158
3159         if (rc != 0) {
3160                 CERROR("Bad connection reply from %s, rc = %d, "
3161                        "version: %x max_frags: %d\n",
3162                        libcfs_nid2str(peer_ni->ibp_nid), rc,
3163                        msg->ibm_version, msg->ibm_u.connparams.ibcp_max_frags);
3164                 goto failed;
3165         }
3166
3167         conn->ibc_incarnation      = msg->ibm_srcstamp;
3168         conn->ibc_credits          = msg->ibm_u.connparams.ibcp_queue_depth;
3169         conn->ibc_reserved_credits = msg->ibm_u.connparams.ibcp_queue_depth;
3170         conn->ibc_queue_depth      = msg->ibm_u.connparams.ibcp_queue_depth;
3171         conn->ibc_max_frags        = msg->ibm_u.connparams.ibcp_max_frags;
3172         LASSERT(conn->ibc_credits + conn->ibc_reserved_credits +
3173                 IBLND_OOB_MSGS(ver) <= IBLND_RX_MSGS(conn));
3174
3175         kiblnd_connreq_done(conn, 0);
3176         return;
3177
3178  failed:
3179         /* NB My QP has already established itself, so I handle anything going
3180          * wrong here by setting ibc_comms_error.
3181          * kiblnd_connreq_done(0) moves the conn state to ESTABLISHED, but then
3182          * immediately tears it down. */
3183
3184         LASSERT (rc != 0);
3185         conn->ibc_comms_error = rc;
3186         kiblnd_connreq_done(conn, 0);
3187 }
3188
3189 static int
3190 kiblnd_active_connect(struct rdma_cm_id *cmid)
3191 {
3192         struct kib_peer_ni *peer_ni = cmid->context;
3193         struct kib_conn *conn;
3194         struct kib_msg *msg;
3195         struct rdma_conn_param cp;
3196         int                      version;
3197         __u64                    incarnation;
3198         unsigned long            flags;
3199         int                      rc;
3200
3201         read_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
3202
3203         incarnation = peer_ni->ibp_incarnation;
3204         version     = (peer_ni->ibp_version == 0) ? IBLND_MSG_VERSION :
3205                                                  peer_ni->ibp_version;
3206
3207         read_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
3208
3209         conn = kiblnd_create_conn(peer_ni, cmid, IBLND_CONN_ACTIVE_CONNECT,
3210                                   version);
3211         if (conn == NULL) {
3212                 kiblnd_peer_connect_failed(peer_ni, 1, -ENOMEM);
3213                 kiblnd_peer_decref(peer_ni); /* lose cmid's ref */
3214                 return -ENOMEM;
3215         }
3216
3217         /* conn "owns" cmid now, so I return success from here on to ensure the
3218          * CM callback doesn't destroy cmid. conn also takes over cmid's ref
3219          * on peer_ni */
3220
3221         msg = &conn->ibc_connvars->cv_msg;
3222
3223         memset(msg, 0, sizeof(*msg));
3224         kiblnd_init_msg(msg, IBLND_MSG_CONNREQ, sizeof(msg->ibm_u.connparams));
3225         msg->ibm_u.connparams.ibcp_queue_depth  = conn->ibc_queue_depth;
3226         msg->ibm_u.connparams.ibcp_max_frags    = conn->ibc_max_frags;
3227         msg->ibm_u.connparams.ibcp_max_msg_size = IBLND_MSG_SIZE;
3228
3229         kiblnd_pack_msg(peer_ni->ibp_ni, msg, version,
3230                         0, peer_ni->ibp_nid, incarnation);
3231
3232         memset(&cp, 0, sizeof(cp));
3233         cp.private_data        = msg;
3234         cp.private_data_len    = msg->ibm_nob;
3235         cp.responder_resources = 0;             /* No atomic ops or RDMA reads */
3236         cp.initiator_depth     = 0;
3237         cp.flow_control        = 1;
3238         cp.retry_count         = *kiblnd_tunables.kib_retry_count;
3239         cp.rnr_retry_count     = *kiblnd_tunables.kib_rnr_retry_count;
3240
3241         LASSERT(cmid->context == (void *)conn);
3242         LASSERT(conn->ibc_cmid == cmid);
3243         rc = rdma_connect_locked(cmid, &cp);
3244         if (rc != 0) {
3245                 CNETERR("Can't connect to %s: %d cm_id %p\n",
3246                         libcfs_nid2str(peer_ni->ibp_nid), rc, cmid);
3247                 kiblnd_connreq_done(conn, rc);
3248                 kiblnd_conn_decref(conn);
3249         } else {
3250                 CDEBUG(D_NET, "Connected to %s: cm_id %p\n",
3251                         libcfs_nid2str(peer_ni->ibp_nid), cmid);
3252         }
3253
3254         return 0;
3255 }
3256
3257 /* set the IP ToS ("Type of Service") used by the RoCE QoS */
3258 static void
3259 kiblnd_set_tos(struct rdma_cm_id *cmid)
3260 {
3261         struct kib_peer_ni *peer_ni = cmid->context;
3262         struct lnet_ioctl_config_o2iblnd_tunables *t;
3263
3264         t = &peer_ni->ibp_ni->ni_lnd_tunables.lnd_tun_u.lnd_o2ib;
3265         if (t->lnd_tos < 0)
3266                 return;
3267
3268         rdma_set_service_type(cmid, t->lnd_tos);
3269 }
3270
3271 int
3272 kiblnd_cm_callback(struct rdma_cm_id *cmid, struct rdma_cm_event *event)
3273 {
3274         struct kib_peer_ni *peer_ni;
3275         struct kib_conn *conn;
3276         int rc;
3277
3278         switch (event->event) {
3279         default:
3280                 CERROR("Unexpected event: %d, status: %d\n",
3281                        event->event, event->status);
3282                 LBUG();
3283
3284         case RDMA_CM_EVENT_CONNECT_REQUEST:
3285                 /* destroy cmid on failure */
3286                 rc = kiblnd_passive_connect(cmid,
3287                                             (void *)KIBLND_CONN_PARAM(event),
3288                                             KIBLND_CONN_PARAM_LEN(event));
3289                 CDEBUG(D_NET, "connreq: %d cm_id %p\n", rc, cmid);
3290                 return rc;
3291
3292         case RDMA_CM_EVENT_ADDR_ERROR:
3293                 peer_ni = cmid->context;
3294                 CNETERR("%s: ADDR ERROR %d cm_id %p\n",
3295                         libcfs_nid2str(peer_ni->ibp_nid), event->status, cmid);
3296                 kiblnd_peer_connect_failed(peer_ni, 1, -EHOSTUNREACH);
3297                 kiblnd_peer_decref(peer_ni);
3298                 return -EHOSTUNREACH;      /* rc != 0 destroys cmid */
3299
3300         case RDMA_CM_EVENT_ADDR_RESOLVED:
3301                 peer_ni = cmid->context;
3302
3303                 CDEBUG(D_NET, "%s Addr resolved: %d cm_id %p\n",
3304                         libcfs_nid2str(peer_ni->ibp_nid), event->status, cmid);
3305
3306                 if (event->status != 0) {
3307                         CNETERR("Can't resolve address for %s: %d cm_id %p\n",
3308                                 libcfs_nid2str(peer_ni->ibp_nid),
3309                                 event->status, cmid);
3310                         rc = event->status;
3311                 } else {
3312                         kiblnd_set_tos(cmid);
3313                         rc = rdma_resolve_route(
3314                                 cmid, kiblnd_timeout() * 1000);
3315                         if (rc == 0) {
3316                                 struct kib_net *net = peer_ni->ibp_ni->ni_data;
3317                                 struct kib_dev *dev = net->ibn_dev;
3318
3319                                 CDEBUG(D_NET, "%s: connection bound to "\
3320                                        "%s:%pI4h:%s\n",
3321                                        libcfs_nid2str(peer_ni->ibp_nid),
3322                                        dev->ibd_ifname,
3323                                        &dev->ibd_ifip, cmid->device->name);
3324
3325                                 return 0;
3326                         }
3327
3328                         /* Can't initiate route resolution */
3329                         CNETERR("Can't resolve route for %s: %d cm_id %p\n",
3330                                 libcfs_nid2str(peer_ni->ibp_nid), rc, cmid);
3331                 }
3332                 kiblnd_peer_connect_failed(peer_ni, 1, rc);
3333                 kiblnd_peer_decref(peer_ni);
3334                 return rc;                      /* rc != 0 destroys cmid */
3335
3336         case RDMA_CM_EVENT_ROUTE_ERROR:
3337                 peer_ni = cmid->context;
3338                 CNETERR("%s: ROUTE ERROR %d cm_id %p\n",
3339                         libcfs_nid2str(peer_ni->ibp_nid), event->status, cmid);
3340                 kiblnd_peer_connect_failed(peer_ni, 1, -EHOSTUNREACH);
3341                 kiblnd_peer_decref(peer_ni);
3342                 return -EHOSTUNREACH;           /* rc != 0 destroys cmid */
3343
3344         case RDMA_CM_EVENT_ROUTE_RESOLVED:
3345                 peer_ni = cmid->context;
3346                 CDEBUG(D_NET,"%s Route resolved: %d\n",
3347                        libcfs_nid2str(peer_ni->ibp_nid), event->status);
3348
3349                 if (event->status == 0)
3350                         return kiblnd_active_connect(cmid);
3351
3352                 CNETERR("Can't resolve route for %s: %d cm_id %p\n",
3353                         libcfs_nid2str(peer_ni->ibp_nid), event->status, cmid);
3354                 kiblnd_peer_connect_failed(peer_ni, 1, event->status);
3355                 kiblnd_peer_decref(peer_ni);
3356                 return event->status;           /* rc != 0 destroys cmid */
3357
3358         case RDMA_CM_EVENT_UNREACHABLE:
3359                 conn = cmid->context;
3360                 CNETERR("%s: UNREACHABLE %d cm_id %p conn %p ibc_state: %d\n",
3361                         libcfs_nid2str(conn->ibc_peer->ibp_nid),
3362                         event->status, cmid, conn, conn->ibc_state);
3363                 LASSERT(conn->ibc_state != IBLND_CONN_INIT);
3364                 if (conn->ibc_state == IBLND_CONN_ACTIVE_CONNECT ||
3365                     conn->ibc_state == IBLND_CONN_PASSIVE_WAIT) {
3366                         kiblnd_connreq_done(conn, -ENETDOWN);
3367                         kiblnd_conn_decref(conn);
3368                 }
3369                 return 0;
3370
3371         case RDMA_CM_EVENT_CONNECT_ERROR:
3372                 conn = cmid->context;
3373                 CNETERR("%s: CONNECT ERROR %d cm_id %p conn %p state: %d\n",
3374                         libcfs_nid2str(conn->ibc_peer->ibp_nid),
3375                         event->status, cmid, conn, conn->ibc_state);
3376                 if (conn->ibc_state == IBLND_CONN_ACTIVE_CONNECT ||
3377                     conn->ibc_state == IBLND_CONN_PASSIVE_WAIT) {
3378                         kiblnd_connreq_done(conn, -ENOTCONN);
3379                         kiblnd_conn_decref(conn);
3380                 }
3381                 return 0;
3382
3383         case RDMA_CM_EVENT_REJECTED:
3384                 conn = cmid->context;
3385                 switch (conn->ibc_state) {
3386                 default:
3387                         LBUG();
3388
3389                 case IBLND_CONN_PASSIVE_WAIT:
3390                         CERROR("%s: REJECTED %d cm_id %p\n",
3391                                 libcfs_nid2str(conn->ibc_peer->ibp_nid),
3392                                 event->status, cmid);
3393                         kiblnd_connreq_done(conn, -ECONNRESET);
3394                         break;
3395
3396                 case IBLND_CONN_ACTIVE_CONNECT:
3397                         kiblnd_rejected(conn, event->status,
3398                                         (void *)KIBLND_CONN_PARAM(event),
3399                                         KIBLND_CONN_PARAM_LEN(event));
3400                         break;
3401                 }
3402                 kiblnd_conn_decref(conn);
3403                 return 0;
3404
3405         case RDMA_CM_EVENT_ESTABLISHED:
3406                 conn = cmid->context;
3407                 switch (conn->ibc_state) {
3408                 default:
3409                         LBUG();
3410
3411                 case IBLND_CONN_PASSIVE_WAIT:
3412                         CDEBUG(D_NET, "ESTABLISHED (passive): %s cm_id %p conn %p\n",
3413                                 libcfs_nid2str(conn->ibc_peer->ibp_nid), cmid, conn);
3414                         kiblnd_connreq_done(conn, 0);
3415                         break;
3416
3417                 case IBLND_CONN_ACTIVE_CONNECT:
3418                         CDEBUG(D_NET, "ESTABLISHED(active): %s cm_id %p conn %p\n",
3419                                 libcfs_nid2str(conn->ibc_peer->ibp_nid), cmid, conn);
3420                         kiblnd_check_connreply(conn,
3421                                                (void *)KIBLND_CONN_PARAM(event),
3422                                                KIBLND_CONN_PARAM_LEN(event));
3423                         break;
3424                 }
3425                 /* net keeps its ref on conn! */
3426                 return 0;
3427
3428         case RDMA_CM_EVENT_TIMEWAIT_EXIT:
3429                 CDEBUG(D_NET, "Ignore TIMEWAIT_EXIT event\n");
3430                 return 0;
3431
3432         case RDMA_CM_EVENT_DISCONNECTED:
3433                 conn = cmid->context;
3434                 if (conn->ibc_state < IBLND_CONN_ESTABLISHED) {
3435                         CERROR("%s DISCONNECTED cm_id %p conn %p\n",
3436                                 libcfs_nid2str(conn->ibc_peer->ibp_nid), cmid, conn);
3437                         kiblnd_connreq_done(conn, -ECONNRESET);
3438                 } else {
3439                         kiblnd_close_conn(conn, 0);
3440                 }
3441                 kiblnd_conn_decref(conn);
3442                 cmid->context = NULL;
3443                 return 0;
3444
3445         case RDMA_CM_EVENT_DEVICE_REMOVAL:
3446                 LCONSOLE_ERROR_MSG(0x131,
3447                                    "Received notification of device removal\n"
3448                                    "Please shutdown LNET to allow this to proceed\n");
3449                 /* Can't remove network from underneath LNET for now, so I have
3450                  * to ignore this */
3451                 return 0;
3452
3453         case RDMA_CM_EVENT_ADDR_CHANGE:
3454                 LCONSOLE_INFO("Physical link changed (eg hca/port)\n");
3455                 return 0;
3456         }
3457 }
3458
3459 static int
3460 kiblnd_check_txs_locked(struct kib_conn *conn, struct list_head *txs)
3461 {
3462         struct kib_tx *tx;
3463
3464         list_for_each_entry(tx, txs, tx_list) {
3465                 if (txs != &conn->ibc_active_txs) {
3466                         LASSERT(tx->tx_queued);
3467                 } else {
3468                         LASSERT(!tx->tx_queued);
3469                         LASSERT(tx->tx_waiting || tx->tx_sending != 0);
3470                 }
3471
3472                 if (ktime_compare(ktime_get(), tx->tx_deadline) >= 0) {
3473                         CERROR("Timed out tx: %s(WSQ:%d%d%d), %lld seconds\n",
3474                                kiblnd_queue2str(conn, txs),
3475                                tx->tx_waiting, tx->tx_sending, tx->tx_queued,
3476                                kiblnd_timeout() +
3477                                ktime_ms_delta(ktime_get(),
3478                                               tx->tx_deadline) / MSEC_PER_SEC);
3479                         return 1;
3480                 }
3481         }
3482
3483         return 0;
3484 }
3485
3486 static int
3487 kiblnd_conn_timed_out_locked(struct kib_conn *conn)
3488 {
3489         return  kiblnd_check_txs_locked(conn, &conn->ibc_tx_queue) ||
3490                 kiblnd_check_txs_locked(conn, &conn->ibc_tx_noops) ||
3491                 kiblnd_check_txs_locked(conn, &conn->ibc_tx_queue_rsrvd) ||
3492                 kiblnd_check_txs_locked(conn, &conn->ibc_tx_queue_nocred) ||
3493                 kiblnd_check_txs_locked(conn, &conn->ibc_active_txs);
3494 }
3495
3496 static void
3497 kiblnd_check_conns (int idx)
3498 {
3499         LIST_HEAD(closes);
3500         LIST_HEAD(checksends);
3501         LIST_HEAD(timedout_txs);
3502         struct hlist_head *peers = &kiblnd_data.kib_peers[idx];
3503         struct kib_peer_ni *peer_ni;
3504         struct kib_conn *conn;
3505         struct kib_tx *tx, *tx_tmp;
3506         unsigned long flags;
3507
3508         /* NB. We expect to have a look at all the peers and not find any
3509          * RDMAs to time out, so we just use a shared lock while we
3510          * take a look...
3511          */
3512         write_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
3513
3514         hlist_for_each_entry(peer_ni, peers, ibp_list) {
3515                 /* Check tx_deadline */
3516                 list_for_each_entry_safe(tx, tx_tmp, &peer_ni->ibp_tx_queue, tx_list) {
3517                         if (ktime_compare(ktime_get(), tx->tx_deadline) >= 0) {
3518                                 CWARN("Timed out tx for %s: %lld seconds\n",
3519                                       libcfs_nid2str(peer_ni->ibp_nid),
3520                                       ktime_ms_delta(ktime_get(),
3521                                                      tx->tx_deadline) / MSEC_PER_SEC);
3522                                 list_move(&tx->tx_list, &timedout_txs);
3523                         }
3524                 }
3525
3526                 list_for_each_entry(conn, &peer_ni->ibp_conns, ibc_list) {
3527                         int timedout;
3528                         int sendnoop;
3529
3530                         LASSERT(conn->ibc_state == IBLND_CONN_ESTABLISHED);
3531
3532                         spin_lock(&conn->ibc_lock);
3533
3534                         sendnoop = kiblnd_need_noop(conn);
3535                         timedout = kiblnd_conn_timed_out_locked(conn);
3536                         if (!sendnoop && !timedout) {
3537                                 spin_unlock(&conn->ibc_lock);
3538                                 continue;
3539                         }
3540
3541                         if (timedout) {
3542                                 CERROR("Timed out RDMA with %s (%lld): c: %u, oc: %u, rc: %u\n",
3543                                        libcfs_nid2str(peer_ni->ibp_nid),
3544                                        ktime_get_seconds()
3545                                        - peer_ni->ibp_last_alive,
3546                                        conn->ibc_credits,
3547                                        conn->ibc_outstanding_credits,
3548                                        conn->ibc_reserved_credits);
3549 #ifdef O2IBLND_CONN_STATE_DEBUG
3550                                 kiblnd_dump_conn_dbg(conn);
3551 #endif
3552                                 list_add(&conn->ibc_connd_list, &closes);
3553                         } else {
3554                                 list_add(&conn->ibc_connd_list, &checksends);
3555                         }
3556                         /* +ref for 'closes' or 'checksends' */
3557                         kiblnd_conn_addref(conn);
3558
3559                         spin_unlock(&conn->ibc_lock);
3560                 }
3561         }
3562
3563         write_unlock_irqrestore(&kiblnd_data.kib_global_lock, flags);
3564
3565         if (!list_empty(&timedout_txs))
3566                 kiblnd_txlist_done(&timedout_txs, -ETIMEDOUT,
3567                                    LNET_MSG_STATUS_NETWORK_TIMEOUT);
3568
3569         /* Handle timeout by closing the whole
3570          * connection. We can only be sure RDMA activity
3571          * has ceased once the QP has been modified.
3572          */
3573         while ((conn = list_first_entry_or_null(&closes,
3574                                                 struct kib_conn,
3575                                                 ibc_connd_list)) != NULL) {
3576                 list_del(&conn->ibc_connd_list);
3577                 kiblnd_close_conn(conn, -ETIMEDOUT);
3578                 kiblnd_conn_decref(conn);
3579         }
3580
3581         /* In case we have enough credits to return via a
3582          * NOOP, but there were no non-blocking tx descs
3583          * free to do it last time...
3584          */
3585         while ((conn = list_first_entry_or_null(&checksends,
3586                                                 struct kib_conn,
3587                                                 ibc_connd_list)) != NULL) {
3588                 list_del(&conn->ibc_connd_list);
3589
3590                 spin_lock(&conn->ibc_lock);
3591                 kiblnd_check_sends_locked(conn);
3592                 spin_unlock(&conn->ibc_lock);
3593
3594                 kiblnd_conn_decref(conn);
3595         }
3596 }
3597
3598 static void
3599 kiblnd_disconnect_conn(struct kib_conn *conn)
3600 {
3601         LASSERT (!in_interrupt());
3602         LASSERT (current == kiblnd_data.kib_connd);
3603         LASSERT (conn->ibc_state == IBLND_CONN_CLOSING);
3604 #ifdef O2IBLND_CONN_STATE_DEBUG
3605         kiblnd_dump_conn_dbg(conn);
3606 #endif
3607         rdma_disconnect(conn->ibc_cmid);
3608         kiblnd_finalise_conn(conn);
3609
3610         kiblnd_peer_notify(conn->ibc_peer);
3611 }
3612
3613 /*
3614  * High-water for reconnection to the same peer_ni, reconnection attempt should
3615  * be delayed after trying more than KIB_RECONN_HIGH_RACE.
3616  */
3617 #define KIB_RECONN_HIGH_RACE    10
3618 /*
3619  * Allow connd to take a break and handle other things after consecutive
3620  * reconnection attemps.
3621  */
3622 #define KIB_RECONN_BREAK        100
3623
3624 int
3625 kiblnd_connd (void *arg)
3626 {
3627         spinlock_t *lock = &kiblnd_data.kib_connd_lock;
3628         wait_queue_entry_t wait;
3629         unsigned long flags;
3630         struct kib_conn *conn;
3631         int timeout;
3632         int i;
3633         bool dropped_lock;
3634         int peer_index = 0;
3635         unsigned long deadline = jiffies;
3636
3637         init_wait(&wait);
3638         kiblnd_data.kib_connd = current;
3639
3640         spin_lock_irqsave(lock, flags);
3641
3642         while (!kiblnd_data.kib_shutdown) {
3643                 int reconn = 0;
3644
3645                 dropped_lock = false;
3646
3647                 conn = list_first_entry_or_null(&kiblnd_data.kib_connd_zombies,
3648                                                 struct kib_conn, ibc_list);
3649                 if (conn) {
3650                         struct kib_peer_ni *peer_ni = NULL;
3651
3652                         list_del(&conn->ibc_list);
3653                         if (conn->ibc_reconnect) {
3654                                 peer_ni = conn->ibc_peer;
3655                                 kiblnd_peer_addref(peer_ni);
3656                         }
3657
3658                         spin_unlock_irqrestore(lock, flags);
3659                         dropped_lock = true;
3660
3661                         kiblnd_destroy_conn(conn);
3662
3663                         spin_lock_irqsave(lock, flags);
3664                         if (!peer_ni) {
3665                                 LIBCFS_FREE(conn, sizeof(*conn));
3666                                 continue;
3667                         }
3668
3669                         conn->ibc_peer = peer_ni;
3670                         if (peer_ni->ibp_reconnected < KIB_RECONN_HIGH_RACE)
3671                                 list_add_tail(&conn->ibc_list,
3672                                               &kiblnd_data.kib_reconn_list);
3673                         else
3674                                 list_add_tail(&conn->ibc_list,
3675                                               &kiblnd_data.kib_reconn_wait);
3676                 }
3677
3678                 conn = list_first_entry_or_null(&kiblnd_data.kib_connd_conns,
3679                                                 struct kib_conn, ibc_list);
3680                 if (conn) {
3681                         int wait;
3682
3683                         list_del(&conn->ibc_list);
3684
3685                         spin_unlock_irqrestore(lock, flags);
3686                         dropped_lock = true;
3687
3688                         kiblnd_disconnect_conn(conn);
3689                         wait = conn->ibc_waits;
3690                         if (wait == 0) /* keep ref for connd_wait, see below */
3691                                 kiblnd_conn_decref(conn);
3692
3693                         spin_lock_irqsave(lock, flags);
3694
3695                         if (wait)
3696                                 list_add_tail(&conn->ibc_list,
3697                                               &kiblnd_data.kib_connd_waits);
3698                 }
3699
3700                 while (reconn < KIB_RECONN_BREAK) {
3701                         if (kiblnd_data.kib_reconn_sec !=
3702                             ktime_get_real_seconds()) {
3703                                 kiblnd_data.kib_reconn_sec = ktime_get_real_seconds();
3704                                 list_splice_init(&kiblnd_data.kib_reconn_wait,
3705                                                  &kiblnd_data.kib_reconn_list);
3706                         }
3707
3708                         conn = list_first_entry_or_null(&kiblnd_data.kib_reconn_list,
3709                                                         struct kib_conn, ibc_list);
3710                         if (!conn)
3711                                 break;
3712
3713                         list_del(&conn->ibc_list);
3714
3715                         spin_unlock_irqrestore(lock, flags);
3716                         dropped_lock = true;
3717
3718                         reconn += kiblnd_reconnect_peer(conn->ibc_peer);
3719                         kiblnd_peer_decref(conn->ibc_peer);
3720                         LIBCFS_FREE(conn, sizeof(*conn));
3721
3722                         spin_lock_irqsave(lock, flags);
3723                 }
3724
3725                 conn = list_first_entry_or_null(&kiblnd_data.kib_connd_waits,
3726                                                 struct kib_conn, ibc_list);
3727                 if (conn) {
3728                         list_del(&conn->ibc_list);
3729                         spin_unlock_irqrestore(lock, flags);
3730
3731                         dropped_lock = kiblnd_tx_may_discard(conn);
3732                         if (dropped_lock)
3733                                 kiblnd_conn_decref(conn);
3734
3735                         spin_lock_irqsave(lock, flags);
3736                         if (!dropped_lock)
3737                                 list_add_tail(&conn->ibc_list,
3738                                               &kiblnd_data.kib_connd_waits);
3739                 }
3740
3741                 /* careful with the jiffy wrap... */
3742                 timeout = (int)(deadline - jiffies);
3743                 if (timeout <= 0) {
3744                         const int n = 4;
3745                         const int p = 1;
3746                         int chunk = HASH_SIZE(kiblnd_data.kib_peers);
3747                         unsigned int lnd_timeout;
3748
3749                         spin_unlock_irqrestore(lock, flags);
3750                         dropped_lock = true;
3751
3752                         /* Time to check for RDMA timeouts on a few more
3753                          * peers: I do checks every 'p' seconds on a
3754                          * proportion of the peer_ni table and I need to check
3755                          * every connection 'n' times within a timeout
3756                          * interval, to ensure I detect a timeout on any
3757                          * connection within (n+1)/n times the timeout
3758                          * interval.
3759                          */
3760
3761                         lnd_timeout = kiblnd_timeout();
3762                         if (lnd_timeout > n * p)
3763                                 chunk = (chunk * n * p) / lnd_timeout;
3764                         if (chunk == 0)
3765                                 chunk = 1;
3766
3767                         for (i = 0; i < chunk; i++) {
3768                                 kiblnd_check_conns(peer_index);
3769                                 peer_index = (peer_index + 1) %
3770                                         HASH_SIZE(kiblnd_data.kib_peers);
3771                         }
3772
3773                         deadline += cfs_time_seconds(p);
3774                         spin_lock_irqsave(lock, flags);
3775                 }
3776
3777                 if (dropped_lock)
3778                         continue;
3779
3780                 /* Nothing to do for 'timeout'  */
3781                 set_current_state(TASK_INTERRUPTIBLE);
3782                 add_wait_queue(&kiblnd_data.kib_connd_waitq, &wait);
3783                 spin_unlock_irqrestore(lock, flags);
3784
3785                 schedule_timeout(timeout);
3786
3787                 remove_wait_queue(&kiblnd_data.kib_connd_waitq, &wait);
3788                 spin_lock_irqsave(lock, flags);
3789         }
3790
3791         spin_unlock_irqrestore(lock, flags);
3792
3793         kiblnd_thread_fini();
3794         return 0;
3795 }
3796
3797 void
3798 kiblnd_qp_event(struct ib_event *event, void *arg)
3799 {
3800         struct kib_conn *conn = arg;
3801
3802         switch (event->event) {
3803         case IB_EVENT_COMM_EST:
3804                 CDEBUG(D_NET, "%s established\n",
3805                        libcfs_nid2str(conn->ibc_peer->ibp_nid));
3806                 /* We received a packet but connection isn't established
3807                  * probably handshake packet was lost, so free to
3808                  * force make connection established */
3809                 rdma_notify(conn->ibc_cmid, IB_EVENT_COMM_EST);
3810                 return;
3811
3812         case IB_EVENT_PORT_ERR:
3813         case IB_EVENT_DEVICE_FATAL:
3814                 CERROR("Fatal device error for NI %s\n",
3815                        libcfs_nidstr(&conn->ibc_peer->ibp_ni->ni_nid));
3816                 atomic_set(&conn->ibc_peer->ibp_ni->ni_fatal_error_on, 1);
3817                 return;
3818
3819         case IB_EVENT_PORT_ACTIVE:
3820                 CERROR("Port reactivated for NI %s\n",
3821                        libcfs_nidstr(&conn->ibc_peer->ibp_ni->ni_nid));
3822                 atomic_set(&conn->ibc_peer->ibp_ni->ni_fatal_error_on, 0);
3823                 return;
3824
3825         default:
3826                 CERROR("%s: Async QP event type %d\n",
3827                        libcfs_nid2str(conn->ibc_peer->ibp_nid), event->event);
3828                 return;
3829         }
3830 }
3831
3832 static void
3833 kiblnd_complete (struct ib_wc *wc)
3834 {
3835         switch (kiblnd_wreqid2type(wc->wr_id)) {
3836         default:
3837                 LBUG();
3838
3839         case IBLND_WID_MR:
3840                 if (wc->status != IB_WC_SUCCESS &&
3841                     wc->status != IB_WC_WR_FLUSH_ERR)
3842                         CNETERR("FastReg failed: %d\n", wc->status);
3843                 return;
3844
3845         case IBLND_WID_RDMA:
3846                 /* We only get RDMA completion notification if it fails.  All
3847                  * subsequent work items, including the final SEND will fail
3848                  * too.  However we can't print out any more info about the
3849                  * failing RDMA because 'tx' might be back on the idle list or
3850                  * even reused already if we didn't manage to post all our work
3851                  * items */
3852                 CNETERR("RDMA (tx: %p) failed: %d\n",
3853                         kiblnd_wreqid2ptr(wc->wr_id), wc->status);
3854                 return;
3855
3856         case IBLND_WID_TX:
3857                 kiblnd_tx_complete(kiblnd_wreqid2ptr(wc->wr_id), wc->status);
3858                 return;
3859
3860         case IBLND_WID_RX:
3861                 kiblnd_rx_complete(kiblnd_wreqid2ptr(wc->wr_id), wc->status,
3862                                    wc->byte_len);
3863                 return;
3864         }
3865 }
3866
3867 void
3868 kiblnd_cq_completion(struct ib_cq *cq, void *arg)
3869 {
3870         /* NB I'm not allowed to schedule this conn once its refcount has
3871          * reached 0.  Since fundamentally I'm racing with scheduler threads
3872          * consuming my CQ I could be called after all completions have
3873          * occurred.  But in this case, ibc_nrx == 0 && ibc_nsends_posted == 0
3874          * and this CQ is about to be destroyed so I NOOP. */
3875         struct kib_conn *conn = arg;
3876         struct kib_sched_info *sched = conn->ibc_sched;
3877         unsigned long flags;
3878
3879         LASSERT(cq == conn->ibc_cq);
3880
3881         spin_lock_irqsave(&sched->ibs_lock, flags);
3882
3883         conn->ibc_ready = 1;
3884
3885         if (!conn->ibc_scheduled &&
3886             (conn->ibc_nrx > 0 ||
3887              conn->ibc_nsends_posted > 0)) {
3888                 kiblnd_conn_addref(conn); /* +1 ref for sched_conns */
3889                 kiblnd_dump_conn_dbg(conn);
3890                 conn->ibc_scheduled = 1;
3891                 list_add_tail(&conn->ibc_sched_list, &sched->ibs_conns);
3892
3893                 if (waitqueue_active(&sched->ibs_waitq))
3894                         wake_up(&sched->ibs_waitq);
3895         }
3896
3897         spin_unlock_irqrestore(&sched->ibs_lock, flags);
3898 }
3899
3900 void
3901 kiblnd_cq_event(struct ib_event *event, void *arg)
3902 {
3903         struct kib_conn *conn = arg;
3904
3905         CERROR("%s: async CQ event type %d\n",
3906                libcfs_nid2str(conn->ibc_peer->ibp_nid), event->event);
3907 }
3908
3909 int
3910 kiblnd_scheduler(void *arg)
3911 {
3912         long id = (long)arg;
3913         struct kib_sched_info *sched;
3914         struct kib_conn *conn;
3915         wait_queue_entry_t wait;
3916         unsigned long flags;
3917         struct ib_wc wc;
3918         bool did_something;
3919         int rc;
3920
3921         init_wait(&wait);
3922
3923         sched = kiblnd_data.kib_scheds[KIB_THREAD_CPT(id)];
3924
3925         rc = cfs_cpt_bind(lnet_cpt_table(), sched->ibs_cpt);
3926         if (rc != 0) {
3927                 CWARN("Unable to bind on CPU partition %d, please verify whether all CPUs are healthy and reload modules if necessary, otherwise your system might under risk of low performance\n", sched->ibs_cpt);
3928         }
3929
3930         spin_lock_irqsave(&sched->ibs_lock, flags);
3931
3932         while (!kiblnd_data.kib_shutdown) {
3933                 if (need_resched()) {
3934                         spin_unlock_irqrestore(&sched->ibs_lock, flags);
3935
3936                         cond_resched();
3937
3938                         spin_lock_irqsave(&sched->ibs_lock, flags);
3939                 }
3940
3941                 did_something = false;
3942
3943                 conn = list_first_entry_or_null(&sched->ibs_conns,
3944                                                 struct kib_conn,
3945                                                 ibc_sched_list);
3946                 if (conn) {
3947                         /* take over kib_sched_conns' ref on conn... */
3948                         LASSERT(conn->ibc_scheduled);
3949                         list_del(&conn->ibc_sched_list);
3950                         conn->ibc_ready = 0;
3951
3952                         spin_unlock_irqrestore(&sched->ibs_lock, flags);
3953
3954                         wc.wr_id = IBLND_WID_INVAL;
3955
3956                         rc = ib_poll_cq(conn->ibc_cq, 1, &wc);
3957                         if (rc == 0) {
3958                                 rc = ib_req_notify_cq(conn->ibc_cq,
3959                                                       IB_CQ_NEXT_COMP);
3960                                 if (rc < 0) {
3961                                         CWARN("%s: ib_req_notify_cq failed: %d, closing connection %p\n",
3962                                               libcfs_nid2str(conn->ibc_peer->ibp_nid),
3963                                               rc, conn);
3964                                         kiblnd_close_conn(conn, -EIO);
3965                                         kiblnd_conn_decref(conn);
3966                                         spin_lock_irqsave(&sched->ibs_lock,
3967                                                           flags);
3968                                         continue;
3969                                 }
3970
3971                                 rc = ib_poll_cq(conn->ibc_cq, 1, &wc);
3972                         }
3973
3974                         if (unlikely(rc > 0 && wc.wr_id == IBLND_WID_INVAL)) {
3975                                 LCONSOLE_ERROR(
3976                                         "ib_poll_cq (rc: %d) returned invalid "
3977                                         "wr_id, opcode %d, status: %d, "
3978                                         "vendor_err: %d, conn: %s status: %d\n"
3979                                         "please upgrade firmware and OFED or "
3980                                         "contact vendor.\n", rc,
3981                                         wc.opcode, wc.status, wc.vendor_err,
3982                                         libcfs_nid2str(conn->ibc_peer->ibp_nid),
3983                                         conn->ibc_state);
3984                                 rc = -EINVAL;
3985                         }
3986
3987                         if (rc < 0) {
3988                                 CWARN("%s: ib_poll_cq failed: %d, closing connection %p\n",
3989                                       libcfs_nid2str(conn->ibc_peer->ibp_nid),
3990                                       rc, conn);
3991                                 kiblnd_close_conn(conn, -EIO);
3992                                 kiblnd_conn_decref(conn);
3993                                 spin_lock_irqsave(&sched->ibs_lock, flags);
3994                                 continue;
3995                         }
3996
3997                         spin_lock_irqsave(&sched->ibs_lock, flags);
3998
3999                         if (rc != 0 || conn->ibc_ready) {
4000                                 /* There may be another completion waiting; get
4001                                  * another scheduler to check while I handle
4002                                  * this one... */
4003                                 /* +1 ref for sched_conns */
4004                                 kiblnd_conn_addref(conn);
4005                                 list_add_tail(&conn->ibc_sched_list,
4006                                               &sched->ibs_conns);
4007                                 if (waitqueue_active(&sched->ibs_waitq))
4008                                         wake_up(&sched->ibs_waitq);
4009                         } else {
4010                                 conn->ibc_scheduled = 0;
4011                         }
4012
4013                         if (rc != 0) {
4014                                 spin_unlock_irqrestore(&sched->ibs_lock, flags);
4015                                 kiblnd_complete(&wc);
4016
4017                                 spin_lock_irqsave(&sched->ibs_lock, flags);
4018                         }
4019
4020                         kiblnd_conn_decref(conn); /* ..drop my ref from above */
4021                         did_something = true;
4022                 }
4023
4024                 if (did_something)
4025                         continue;
4026
4027                 set_current_state(TASK_INTERRUPTIBLE);
4028                 add_wait_queue_exclusive(&sched->ibs_waitq, &wait);
4029                 spin_unlock_irqrestore(&sched->ibs_lock, flags);
4030
4031                 schedule();
4032
4033                 remove_wait_queue(&sched->ibs_waitq, &wait);
4034                 set_current_state(TASK_RUNNING);
4035                 spin_lock_irqsave(&sched->ibs_lock, flags);
4036         }
4037
4038         spin_unlock_irqrestore(&sched->ibs_lock, flags);
4039
4040         kiblnd_thread_fini();
4041         return 0;
4042 }
4043
4044 int
4045 kiblnd_failover_thread(void *arg)
4046 {
4047         rwlock_t *glock = &kiblnd_data.kib_global_lock;
4048         struct kib_dev *dev;
4049         struct net *ns = arg;
4050         wait_queue_entry_t wait;
4051         unsigned long flags;
4052         int rc;
4053
4054         LASSERT(*kiblnd_tunables.kib_dev_failover != 0);
4055
4056         init_wait(&wait);
4057         write_lock_irqsave(glock, flags);
4058
4059         while (!kiblnd_data.kib_shutdown) {
4060                 bool do_failover = false;
4061                 int long_sleep;
4062
4063                 list_for_each_entry(dev, &kiblnd_data.kib_failed_devs,
4064                                     ibd_fail_list) {
4065                         if (ktime_get_seconds() < dev->ibd_next_failover)
4066                                 continue;
4067                         do_failover = true;
4068                         break;
4069                 }
4070
4071                 if (do_failover) {
4072                         list_del_init(&dev->ibd_fail_list);
4073                         dev->ibd_failover = 1;
4074                         write_unlock_irqrestore(glock, flags);
4075
4076                         rc = kiblnd_dev_failover(dev, ns);
4077
4078                         write_lock_irqsave(glock, flags);
4079
4080                         LASSERT(dev->ibd_failover);
4081                         dev->ibd_failover = 0;
4082                         if (rc >= 0) { /* Device is OK or failover succeed */
4083                                 dev->ibd_next_failover = ktime_get_seconds() + 3;
4084                                 continue;
4085                         }
4086
4087                         /* failed to failover, retry later */
4088                         dev->ibd_next_failover = ktime_get_seconds() +
4089                                 min(dev->ibd_failed_failover, 10);
4090                         if (kiblnd_dev_can_failover(dev)) {
4091                                 list_add_tail(&dev->ibd_fail_list,
4092                                               &kiblnd_data.kib_failed_devs);
4093                         }
4094
4095                         continue;
4096                 }
4097
4098                 /* long sleep if no more pending failover */
4099                 long_sleep = list_empty(&kiblnd_data.kib_failed_devs);
4100
4101                 set_current_state(TASK_INTERRUPTIBLE);
4102                 add_wait_queue(&kiblnd_data.kib_failover_waitq, &wait);
4103                 write_unlock_irqrestore(glock, flags);
4104
4105                 rc = schedule_timeout(long_sleep ? cfs_time_seconds(10) :
4106                                       cfs_time_seconds(1));
4107                 set_current_state(TASK_RUNNING);
4108                 remove_wait_queue(&kiblnd_data.kib_failover_waitq, &wait);
4109                 write_lock_irqsave(glock, flags);
4110
4111                 if (!long_sleep || rc != 0)
4112                         continue;
4113
4114                 /* have a long sleep, routine check all active devices,
4115                  * we need checking like this because if there is not active
4116                  * connection on the dev and no SEND from local, we may listen
4117                  * on wrong HCA for ever while there is a bonding failover
4118                  */
4119                 list_for_each_entry(dev, &kiblnd_data.kib_devs, ibd_list) {
4120                         if (kiblnd_dev_can_failover(dev)) {
4121                                 list_add_tail(&dev->ibd_fail_list,
4122                                               &kiblnd_data.kib_failed_devs);
4123                         }
4124                 }
4125         }
4126
4127         write_unlock_irqrestore(glock, flags);
4128
4129         kiblnd_thread_fini();
4130         return 0;
4131 }