drivers/infiniband/ulp/rtrs/rtrs-clt.c

   1 // SPDX-License-Identifier: GPL-2.0-or-later
   2 /*
   3  * RDMA Transport Layer
   4  *
   5  * Copyright (c) 2014 - 2018 ProfitBricks GmbH. All rights reserved.
   6  * Copyright (c) 2018 - 2019 1&1 IONOS Cloud GmbH. All rights reserved.
   7  * Copyright (c) 2019 - 2020 1&1 IONOS SE. All rights reserved.
   8  */
   9
  10 #undef pr_fmt
  11 #define pr_fmt(fmt) KBUILD_MODNAME " L" __stringify(__LINE__) ": " fmt
  12
  13 #include <linux/module.h>
  14 #include <linux/rculist.h>
  15 #include <linux/random.h>
  16
  17 #include "rtrs-clt.h"
  18 #include "rtrs-log.h"
  19
  20 #define RTRS_CONNECT_TIMEOUT_MS 30000
  21 /*
  22  * Wait a bit before trying to reconnect after a failure
  23  * in order to give server time to finish clean up which
  24  * leads to "false positives" failed reconnect attempts
  25  */
  26 #define RTRS_RECONNECT_BACKOFF 1000
  27 /*
  28  * Wait for additional random time between 0 and 8 seconds
  29  * before starting to reconnect to avoid clients reconnecting
  30  * all at once in case of a major network outage
  31  */
  32 #define RTRS_RECONNECT_SEED 8
  33
  34 #define FIRST_CONN 0x01
  35
  36 MODULE_DESCRIPTION("RDMA Transport Client");
  37 MODULE_LICENSE("GPL");
  38
  39 static const struct rtrs_rdma_dev_pd_ops dev_pd_ops;
  40 static struct rtrs_rdma_dev_pd dev_pd = {
  41         .ops = &dev_pd_ops
  42 };
  43
  44 static struct workqueue_struct *rtrs_wq;
  45 static struct class *rtrs_clt_dev_class;
  46
  47 static inline bool rtrs_clt_is_connected(const struct rtrs_clt *clt)
  48 {
  49         struct rtrs_clt_sess *sess;
  50         bool connected = false;
  51
  52         rcu_read_lock();
  53         list_for_each_entry_rcu(sess, &clt->paths_list, s.entry)
  54                 connected |= READ_ONCE(sess->state) == RTRS_CLT_CONNECTED;
  55         rcu_read_unlock();
  56
  57         return connected;
  58 }
  59
  60 static struct rtrs_permit *
  61 __rtrs_get_permit(struct rtrs_clt *clt, enum rtrs_clt_con_type con_type)
  62 {
  63         size_t max_depth = clt->queue_depth;
  64         struct rtrs_permit *permit;
  65         int bit;
  66
  67         /*
  68          * Adapted from null_blk get_tag(). Callers from different cpus may
  69          * grab the same bit, since find_first_zero_bit is not atomic.
  70          * But then the test_and_set_bit_lock will fail for all the
  71          * callers but one, so that they will loop again.
  72          * This way an explicit spinlock is not required.
  73          */
  74         do {
  75                 bit = find_first_zero_bit(clt->permits_map, max_depth);
  76                 if (unlikely(bit >= max_depth))
  77                         return NULL;
  78         } while (unlikely(test_and_set_bit_lock(bit, clt->permits_map)));
  79
  80         permit = get_permit(clt, bit);
  81         WARN_ON(permit->mem_id != bit);
  82         permit->cpu_id = raw_smp_processor_id();
  83         permit->con_type = con_type;
  84
  85         return permit;
  86 }
  87
  88 static inline void __rtrs_put_permit(struct rtrs_clt *clt,
  89                                       struct rtrs_permit *permit)
  90 {
  91         clear_bit_unlock(permit->mem_id, clt->permits_map);
  92 }
  93
  94 /**
  95  * rtrs_clt_get_permit() - allocates permit for future RDMA operation
  96  * @clt:        Current session
  97  * @con_type:   Type of connection to use with the permit
  98  * @can_wait:   Wait type
  99  *
 100  * Description:
 101  *    Allocates permit for the following RDMA operation.  Permit is used
 102  *    to preallocate all resources and to propagate memory pressure
 103  *    up earlier.
 104  *
 105  * Context:
 106  *    Can sleep if @wait == RTRS_PERMIT_WAIT
 107  */
 108 struct rtrs_permit *rtrs_clt_get_permit(struct rtrs_clt *clt,
 109                                           enum rtrs_clt_con_type con_type,
 110                                           enum wait_type can_wait)
 111 {
 112         struct rtrs_permit *permit;
 113         DEFINE_WAIT(wait);
 114
 115         permit = __rtrs_get_permit(clt, con_type);
 116         if (likely(permit) || !can_wait)
 117                 return permit;
 118
 119         do {
 120                 prepare_to_wait(&clt->permits_wait, &wait,
 121                                 TASK_UNINTERRUPTIBLE);
 122                 permit = __rtrs_get_permit(clt, con_type);
 123                 if (likely(permit))
 124                         break;
 125
 126                 io_schedule();
 127         } while (1);
 128
 129         finish_wait(&clt->permits_wait, &wait);
 130
 131         return permit;
 132 }
 133 EXPORT_SYMBOL(rtrs_clt_get_permit);
 134
 135 /**
 136  * rtrs_clt_put_permit() - puts allocated permit
 137  * @clt:        Current session
 138  * @permit:     Permit to be freed
 139  *
 140  * Context:
 141  *    Does not matter
 142  */
 143 void rtrs_clt_put_permit(struct rtrs_clt *clt, struct rtrs_permit *permit)
 144 {
 145         if (WARN_ON(!test_bit(permit->mem_id, clt->permits_map)))
 146                 return;
 147
 148         __rtrs_put_permit(clt, permit);
 149
 150         /*
 151          * rtrs_clt_get_permit() adds itself to the &clt->permits_wait list
 152          * before calling schedule(). So if rtrs_clt_get_permit() is sleeping
 153          * it must have added itself to &clt->permits_wait before
 154          * __rtrs_put_permit() finished.
 155          * Hence it is safe to guard wake_up() with a waitqueue_active() test.
 156          */
 157         if (waitqueue_active(&clt->permits_wait))
 158                 wake_up(&clt->permits_wait);
 159 }
 160 EXPORT_SYMBOL(rtrs_clt_put_permit);
 161
 162 /**
 163  * rtrs_permit_to_clt_con() - returns RDMA connection pointer by the permit
 164  * @sess: client session pointer
 165  * @permit: permit for the allocation of the RDMA buffer
 166  * Note:
 167  *     IO connection starts from 1.
 168  *     0 connection is for user messages.
 169  */
 170 static
 171 struct rtrs_clt_con *rtrs_permit_to_clt_con(struct rtrs_clt_sess *sess,
 172                                             struct rtrs_permit *permit)
 173 {
 174         int id = 0;
 175
 176         if (likely(permit->con_type == RTRS_IO_CON))
 177                 id = (permit->cpu_id % (sess->s.irq_con_num - 1)) + 1;
 178
 179         return to_clt_con(sess->s.con[id]);
 180 }
 181
 182 /**
 183  * rtrs_clt_change_state() - change the session state through session state
 184  * machine.
 185  *
 186  * @sess: client session to change the state of.
 187  * @new_state: state to change to.
 188  *
 189  * returns true if sess's state is changed to new state, otherwise return false.
 190  *
 191  * Locks:
 192  * state_wq lock must be hold.
 193  */
 194 static bool rtrs_clt_change_state(struct rtrs_clt_sess *sess,
 195                                      enum rtrs_clt_state new_state)
 196 {
 197         enum rtrs_clt_state old_state;
 198         bool changed = false;
 199
 200         lockdep_assert_held(&sess->state_wq.lock);
 201
 202         old_state = sess->state;
 203         switch (new_state) {
 204         case RTRS_CLT_CONNECTING:
 205                 switch (old_state) {
 206                 case RTRS_CLT_RECONNECTING:
 207                         changed = true;
 208                         fallthrough;
 209                 default:
 210                         break;
 211                 }
 212                 break;
 213         case RTRS_CLT_RECONNECTING:
 214                 switch (old_state) {
 215                 case RTRS_CLT_CONNECTED:
 216                 case RTRS_CLT_CONNECTING_ERR:
 217                 case RTRS_CLT_CLOSED:
 218                         changed = true;
 219                         fallthrough;
 220                 default:
 221                         break;
 222                 }
 223                 break;
 224         case RTRS_CLT_CONNECTED:
 225                 switch (old_state) {
 226                 case RTRS_CLT_CONNECTING:
 227                         changed = true;
 228                         fallthrough;
 229                 default:
 230                         break;
 231                 }
 232                 break;
 233         case RTRS_CLT_CONNECTING_ERR:
 234                 switch (old_state) {
 235                 case RTRS_CLT_CONNECTING:
 236                         changed = true;
 237                         fallthrough;
 238                 default:
 239                         break;
 240                 }
 241                 break;
 242         case RTRS_CLT_CLOSING:
 243                 switch (old_state) {
 244                 case RTRS_CLT_CONNECTING:
 245                 case RTRS_CLT_CONNECTING_ERR:
 246                 case RTRS_CLT_RECONNECTING:
 247                 case RTRS_CLT_CONNECTED:
 248                         changed = true;
 249                         fallthrough;
 250                 default:
 251                         break;
 252                 }
 253                 break;
 254         case RTRS_CLT_CLOSED:
 255                 switch (old_state) {
 256                 case RTRS_CLT_CLOSING:
 257                         changed = true;
 258                         fallthrough;
 259                 default:
 260                         break;
 261                 }
 262                 break;
 263         case RTRS_CLT_DEAD:
 264                 switch (old_state) {
 265                 case RTRS_CLT_CLOSED:
 266                         changed = true;
 267                         fallthrough;
 268                 default:
 269                         break;
 270                 }
 271                 break;
 272         default:
 273                 break;
 274         }
 275         if (changed) {
 276                 sess->state = new_state;
 277                 wake_up_locked(&sess->state_wq);
 278         }
 279
 280         return changed;
 281 }
 282
 283 static bool rtrs_clt_change_state_from_to(struct rtrs_clt_sess *sess,
 284                                            enum rtrs_clt_state old_state,
 285                                            enum rtrs_clt_state new_state)
 286 {
 287         bool changed = false;
 288
 289         spin_lock_irq(&sess->state_wq.lock);
 290         if (sess->state == old_state)
 291                 changed = rtrs_clt_change_state(sess, new_state);
 292         spin_unlock_irq(&sess->state_wq.lock);
 293
 294         return changed;
 295 }
 296
 297 static void rtrs_rdma_error_recovery(struct rtrs_clt_con *con)
 298 {
 299         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
 300
 301         if (rtrs_clt_change_state_from_to(sess,
 302                                            RTRS_CLT_CONNECTED,
 303                                            RTRS_CLT_RECONNECTING)) {
 304                 struct rtrs_clt *clt = sess->clt;
 305                 unsigned int delay_ms;
 306
 307                 /*
 308                  * Normal scenario, reconnect if we were successfully connected
 309                  */
 310                 delay_ms = clt->reconnect_delay_sec * 1000;
 311                 queue_delayed_work(rtrs_wq, &sess->reconnect_dwork,
 312                                    msecs_to_jiffies(delay_ms +
 313                                                     prandom_u32() % RTRS_RECONNECT_SEED));
 314         } else {
 315                 /*
 316                  * Error can happen just on establishing new connection,
 317                  * so notify waiter with error state, waiter is responsible
 318                  * for cleaning the rest and reconnect if needed.
 319                  */
 320                 rtrs_clt_change_state_from_to(sess,
 321                                                RTRS_CLT_CONNECTING,
 322                                                RTRS_CLT_CONNECTING_ERR);
 323         }
 324 }
 325
 326 static void rtrs_clt_fast_reg_done(struct ib_cq *cq, struct ib_wc *wc)
 327 {
 328         struct rtrs_clt_con *con = to_clt_con(wc->qp->qp_context);
 329
 330         if (unlikely(wc->status != IB_WC_SUCCESS)) {
 331                 rtrs_err(con->c.sess, "Failed IB_WR_REG_MR: %s\n",
 332                           ib_wc_status_msg(wc->status));
 333                 rtrs_rdma_error_recovery(con);
 334         }
 335 }
 336
 337 static struct ib_cqe fast_reg_cqe = {
 338         .done = rtrs_clt_fast_reg_done
 339 };
 340
 341 static void complete_rdma_req(struct rtrs_clt_io_req *req, int errno,
 342                               bool notify, bool can_wait);
 343
 344 static void rtrs_clt_inv_rkey_done(struct ib_cq *cq, struct ib_wc *wc)
 345 {
 346         struct rtrs_clt_io_req *req =
 347                 container_of(wc->wr_cqe, typeof(*req), inv_cqe);
 348         struct rtrs_clt_con *con = to_clt_con(wc->qp->qp_context);
 349
 350         if (unlikely(wc->status != IB_WC_SUCCESS)) {
 351                 rtrs_err(con->c.sess, "Failed IB_WR_LOCAL_INV: %s\n",
 352                           ib_wc_status_msg(wc->status));
 353                 rtrs_rdma_error_recovery(con);
 354         }
 355         req->need_inv = false;
 356         if (likely(req->need_inv_comp))
 357                 complete(&req->inv_comp);
 358         else
 359                 /* Complete request from INV callback */
 360                 complete_rdma_req(req, req->inv_errno, true, false);
 361 }
 362
 363 static int rtrs_inv_rkey(struct rtrs_clt_io_req *req)
 364 {
 365         struct rtrs_clt_con *con = req->con;
 366         struct ib_send_wr wr = {
 367                 .opcode             = IB_WR_LOCAL_INV,
 368                 .wr_cqe             = &req->inv_cqe,
 369                 .send_flags         = IB_SEND_SIGNALED,
 370                 .ex.invalidate_rkey = req->mr->rkey,
 371         };
 372         req->inv_cqe.done = rtrs_clt_inv_rkey_done;
 373
 374         return ib_post_send(con->c.qp, &wr, NULL);
 375 }
 376
 377 static void complete_rdma_req(struct rtrs_clt_io_req *req, int errno,
 378                               bool notify, bool can_wait)
 379 {
 380         struct rtrs_clt_con *con = req->con;
 381         struct rtrs_clt_sess *sess;
 382         int err;
 383
 384         if (WARN_ON(!req->in_use))
 385                 return;
 386         if (WARN_ON(!req->con))
 387                 return;
 388         sess = to_clt_sess(con->c.sess);
 389
 390         if (req->sg_cnt) {
 391                 if (unlikely(req->dir == DMA_FROM_DEVICE && req->need_inv)) {
 392                         /*
 393                          * We are here to invalidate read requests
 394                          * ourselves.  In normal scenario server should
 395                          * send INV for all read requests, but
 396                          * we are here, thus two things could happen:
 397                          *
 398                          *    1.  this is failover, when errno != 0
 399                          *        and can_wait == 1,
 400                          *
 401                          *    2.  something totally bad happened and
 402                          *        server forgot to send INV, so we
 403                          *        should do that ourselves.
 404                          */
 405
 406                         if (likely(can_wait)) {
 407                                 req->need_inv_comp = true;
 408                         } else {
 409                                 /* This should be IO path, so always notify */
 410                                 WARN_ON(!notify);
 411                                 /* Save errno for INV callback */
 412                                 req->inv_errno = errno;
 413                         }
 414
 415                         err = rtrs_inv_rkey(req);
 416                         if (unlikely(err)) {
 417                                 rtrs_err(con->c.sess, "Send INV WR key=%#x: %d\n",
 418                                           req->mr->rkey, err);
 419                         } else if (likely(can_wait)) {
 420                                 wait_for_completion(&req->inv_comp);
 421                         } else {
 422                                 /*
 423                                  * Something went wrong, so request will be
 424                                  * completed from INV callback.
 425                                  */
 426                                 WARN_ON_ONCE(1);
 427
 428                                 return;
 429                         }
 430                 }
 431                 ib_dma_unmap_sg(sess->s.dev->ib_dev, req->sglist,
 432                                 req->sg_cnt, req->dir);
 433         }
 434         if (sess->clt->mp_policy == MP_POLICY_MIN_INFLIGHT)
 435                 atomic_dec(&sess->stats->inflight);
 436
 437         req->in_use = false;
 438         req->con = NULL;
 439
 440         if (errno) {
 441                 rtrs_err_rl(con->c.sess,
 442                             "IO request failed: error=%d path=%s [%s:%u]\n",
 443                             errno, kobject_name(&sess->kobj), sess->hca_name,
 444                             sess->hca_port);
 445         }
 446
 447         if (notify)
 448                 req->conf(req->priv, errno);
 449 }
 450
 451 static int rtrs_post_send_rdma(struct rtrs_clt_con *con,
 452                                 struct rtrs_clt_io_req *req,
 453                                 struct rtrs_rbuf *rbuf, u32 off,
 454                                 u32 imm, struct ib_send_wr *wr)
 455 {
 456         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
 457         enum ib_send_flags flags;
 458         struct ib_sge sge;
 459
 460         if (unlikely(!req->sg_size)) {
 461                 rtrs_wrn(con->c.sess,
 462                          "Doing RDMA Write failed, no data supplied\n");
 463                 return -EINVAL;
 464         }
 465
 466         /* user data and user message in the first list element */
 467         sge.addr   = req->iu->dma_addr;
 468         sge.length = req->sg_size;
 469         sge.lkey   = sess->s.dev->ib_pd->local_dma_lkey;
 470
 471         /*
 472          * From time to time we have to post signalled sends,
 473          * or send queue will fill up and only QP reset can help.
 474          */
 475         flags = atomic_inc_return(&con->io_cnt) % sess->queue_depth ?
 476                         0 : IB_SEND_SIGNALED;
 477
 478         ib_dma_sync_single_for_device(sess->s.dev->ib_dev, req->iu->dma_addr,
 479                                       req->sg_size, DMA_TO_DEVICE);
 480
 481         return rtrs_iu_post_rdma_write_imm(&con->c, req->iu, &sge, 1,
 482                                             rbuf->rkey, rbuf->addr + off,
 483                                             imm, flags, wr);
 484 }
 485
 486 static void process_io_rsp(struct rtrs_clt_sess *sess, u32 msg_id,
 487                            s16 errno, bool w_inval)
 488 {
 489         struct rtrs_clt_io_req *req;
 490
 491         if (WARN_ON(msg_id >= sess->queue_depth))
 492                 return;
 493
 494         req = &sess->reqs[msg_id];
 495         /* Drop need_inv if server responded with send with invalidation */
 496         req->need_inv &= !w_inval;
 497         complete_rdma_req(req, errno, true, false);
 498 }
 499
 500 static void rtrs_clt_recv_done(struct rtrs_clt_con *con, struct ib_wc *wc)
 501 {
 502         struct rtrs_iu *iu;
 503         int err;
 504         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
 505
 506         WARN_ON((sess->flags & RTRS_MSG_NEW_RKEY_F) == 0);
 507         iu = container_of(wc->wr_cqe, struct rtrs_iu,
 508                           cqe);
 509         err = rtrs_iu_post_recv(&con->c, iu);
 510         if (unlikely(err)) {
 511                 rtrs_err(con->c.sess, "post iu failed %d\n", err);
 512                 rtrs_rdma_error_recovery(con);
 513         }
 514 }
 515
 516 static void rtrs_clt_rkey_rsp_done(struct rtrs_clt_con *con, struct ib_wc *wc)
 517 {
 518         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
 519         struct rtrs_msg_rkey_rsp *msg;
 520         u32 imm_type, imm_payload;
 521         bool w_inval = false;
 522         struct rtrs_iu *iu;
 523         u32 buf_id;
 524         int err;
 525
 526         WARN_ON((sess->flags & RTRS_MSG_NEW_RKEY_F) == 0);
 527
 528         iu = container_of(wc->wr_cqe, struct rtrs_iu, cqe);
 529
 530         if (unlikely(wc->byte_len < sizeof(*msg))) {
 531                 rtrs_err(con->c.sess, "rkey response is malformed: size %d\n",
 532                           wc->byte_len);
 533                 goto out;
 534         }
 535         ib_dma_sync_single_for_cpu(sess->s.dev->ib_dev, iu->dma_addr,
 536                                    iu->size, DMA_FROM_DEVICE);
 537         msg = iu->buf;
 538         if (unlikely(le16_to_cpu(msg->type) != RTRS_MSG_RKEY_RSP)) {
 539                 rtrs_err(sess->clt, "rkey response is malformed: type %d\n",
 540                           le16_to_cpu(msg->type));
 541                 goto out;
 542         }
 543         buf_id = le16_to_cpu(msg->buf_id);
 544         if (WARN_ON(buf_id >= sess->queue_depth))
 545                 goto out;
 546
 547         rtrs_from_imm(be32_to_cpu(wc->ex.imm_data), &imm_type, &imm_payload);
 548         if (likely(imm_type == RTRS_IO_RSP_IMM ||
 549                    imm_type == RTRS_IO_RSP_W_INV_IMM)) {
 550                 u32 msg_id;
 551
 552                 w_inval = (imm_type == RTRS_IO_RSP_W_INV_IMM);
 553                 rtrs_from_io_rsp_imm(imm_payload, &msg_id, &err);
 554
 555                 if (WARN_ON(buf_id != msg_id))
 556                         goto out;
 557                 sess->rbufs[buf_id].rkey = le32_to_cpu(msg->rkey);
 558                 process_io_rsp(sess, msg_id, err, w_inval);
 559         }
 560         ib_dma_sync_single_for_device(sess->s.dev->ib_dev, iu->dma_addr,
 561                                       iu->size, DMA_FROM_DEVICE);
 562         return rtrs_clt_recv_done(con, wc);
 563 out:
 564         rtrs_rdma_error_recovery(con);
 565 }
 566
 567 static void rtrs_clt_rdma_done(struct ib_cq *cq, struct ib_wc *wc);
 568
 569 static struct ib_cqe io_comp_cqe = {
 570         .done = rtrs_clt_rdma_done
 571 };
 572
 573 /*
 574  * Post x2 empty WRs: first is for this RDMA with IMM,
 575  * second is for RECV with INV, which happened earlier.
 576  */
 577 static int rtrs_post_recv_empty_x2(struct rtrs_con *con, struct ib_cqe *cqe)
 578 {
 579         struct ib_recv_wr wr_arr[2], *wr;
 580         int i;
 581
 582         memset(wr_arr, 0, sizeof(wr_arr));
 583         for (i = 0; i < ARRAY_SIZE(wr_arr); i++) {
 584                 wr = &wr_arr[i];
 585                 wr->wr_cqe  = cqe;
 586                 if (i)
 587                         /* Chain backwards */
 588                         wr->next = &wr_arr[i - 1];
 589         }
 590
 591         return ib_post_recv(con->qp, wr, NULL);
 592 }
 593
 594 static void rtrs_clt_rdma_done(struct ib_cq *cq, struct ib_wc *wc)
 595 {
 596         struct rtrs_clt_con *con = to_clt_con(wc->qp->qp_context);
 597         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
 598         u32 imm_type, imm_payload;
 599         bool w_inval = false;
 600         int err;
 601
 602         if (unlikely(wc->status != IB_WC_SUCCESS)) {
 603                 if (wc->status != IB_WC_WR_FLUSH_ERR) {
 604                         rtrs_err(sess->clt, "RDMA failed: %s\n",
 605                                   ib_wc_status_msg(wc->status));
 606                         rtrs_rdma_error_recovery(con);
 607                 }
 608                 return;
 609         }
 610         rtrs_clt_update_wc_stats(con);
 611
 612         switch (wc->opcode) {
 613         case IB_WC_RECV_RDMA_WITH_IMM:
 614                 /*
 615                  * post_recv() RDMA write completions of IO reqs (read/write)
 616                  * and hb
 617                  */
 618                 if (WARN_ON(wc->wr_cqe->done != rtrs_clt_rdma_done))
 619                         return;
 620                 rtrs_from_imm(be32_to_cpu(wc->ex.imm_data),
 621                                &imm_type, &imm_payload);
 622                 if (likely(imm_type == RTRS_IO_RSP_IMM ||
 623                            imm_type == RTRS_IO_RSP_W_INV_IMM)) {
 624                         u32 msg_id;
 625
 626                         w_inval = (imm_type == RTRS_IO_RSP_W_INV_IMM);
 627                         rtrs_from_io_rsp_imm(imm_payload, &msg_id, &err);
 628
 629                         process_io_rsp(sess, msg_id, err, w_inval);
 630                 } else if (imm_type == RTRS_HB_MSG_IMM) {
 631                         WARN_ON(con->c.cid);
 632                         rtrs_send_hb_ack(&sess->s);
 633                         if (sess->flags & RTRS_MSG_NEW_RKEY_F)
 634                                 return  rtrs_clt_recv_done(con, wc);
 635                 } else if (imm_type == RTRS_HB_ACK_IMM) {
 636                         WARN_ON(con->c.cid);
 637                         sess->s.hb_missed_cnt = 0;
 638                         sess->s.hb_cur_latency =
 639                                 ktime_sub(ktime_get(), sess->s.hb_last_sent);
 640                         if (sess->flags & RTRS_MSG_NEW_RKEY_F)
 641                                 return  rtrs_clt_recv_done(con, wc);
 642                 } else {
 643                         rtrs_wrn(con->c.sess, "Unknown IMM type %u\n",
 644                                   imm_type);
 645                 }
 646                 if (w_inval)
 647                         /*
 648                          * Post x2 empty WRs: first is for this RDMA with IMM,
 649                          * second is for RECV with INV, which happened earlier.
 650                          */
 651                         err = rtrs_post_recv_empty_x2(&con->c, &io_comp_cqe);
 652                 else
 653                         err = rtrs_post_recv_empty(&con->c, &io_comp_cqe);
 654                 if (unlikely(err)) {
 655                         rtrs_err(con->c.sess, "rtrs_post_recv_empty(): %d\n",
 656                                   err);
 657                         rtrs_rdma_error_recovery(con);
 658                         break;
 659                 }
 660                 break;
 661         case IB_WC_RECV:
 662                 /*
 663                  * Key invalidations from server side
 664                  */
 665                 WARN_ON(!(wc->wc_flags & IB_WC_WITH_INVALIDATE ||
 666                           wc->wc_flags & IB_WC_WITH_IMM));
 667                 WARN_ON(wc->wr_cqe->done != rtrs_clt_rdma_done);
 668                 if (sess->flags & RTRS_MSG_NEW_RKEY_F) {
 669                         if (wc->wc_flags & IB_WC_WITH_INVALIDATE)
 670                                 return  rtrs_clt_recv_done(con, wc);
 671
 672                         return  rtrs_clt_rkey_rsp_done(con, wc);
 673                 }
 674                 break;
 675         case IB_WC_RDMA_WRITE:
 676                 /*
 677                  * post_send() RDMA write completions of IO reqs (read/write)
 678                  */
 679                 break;
 680
 681         default:
 682                 rtrs_wrn(sess->clt, "Unexpected WC type: %d\n", wc->opcode);
 683                 return;
 684         }
 685 }
 686
 687 static int post_recv_io(struct rtrs_clt_con *con, size_t q_size)
 688 {
 689         int err, i;
 690         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
 691
 692         for (i = 0; i < q_size; i++) {
 693                 if (sess->flags & RTRS_MSG_NEW_RKEY_F) {
 694                         struct rtrs_iu *iu = &con->rsp_ius[i];
 695
 696                         err = rtrs_iu_post_recv(&con->c, iu);
 697                 } else {
 698                         err = rtrs_post_recv_empty(&con->c, &io_comp_cqe);
 699                 }
 700                 if (unlikely(err))
 701                         return err;
 702         }
 703
 704         return 0;
 705 }
 706
 707 static int post_recv_sess(struct rtrs_clt_sess *sess)
 708 {
 709         size_t q_size = 0;
 710         int err, cid;
 711
 712         for (cid = 0; cid < sess->s.con_num; cid++) {
 713                 if (cid == 0)
 714                         q_size = SERVICE_CON_QUEUE_DEPTH;
 715                 else
 716                         q_size = sess->queue_depth;
 717
 718                 /*
 719                  * x2 for RDMA read responses + FR key invalidations,
 720                  * RDMA writes do not require any FR registrations.
 721                  */
 722                 q_size *= 2;
 723
 724                 err = post_recv_io(to_clt_con(sess->s.con[cid]), q_size);
 725                 if (unlikely(err)) {
 726                         rtrs_err(sess->clt, "post_recv_io(), err: %d\n", err);
 727                         return err;
 728                 }
 729         }
 730
 731         return 0;
 732 }
 733
 734 struct path_it {
 735         int i;
 736         struct list_head skip_list;
 737         struct rtrs_clt *clt;
 738         struct rtrs_clt_sess *(*next_path)(struct path_it *it);
 739 };
 740
 741 /**
 742  * list_next_or_null_rr_rcu - get next list element in round-robin fashion.
 743  * @head:       the head for the list.
 744  * @ptr:        the list head to take the next element from.
 745  * @type:       the type of the struct this is embedded in.
 746  * @memb:       the name of the list_head within the struct.
 747  *
 748  * Next element returned in round-robin fashion, i.e. head will be skipped,
 749  * but if list is observed as empty, NULL will be returned.
 750  *
 751  * This primitive may safely run concurrently with the _rcu list-mutation
 752  * primitives such as list_add_rcu() as long as it's guarded by rcu_read_lock().
 753  */
 754 #define list_next_or_null_rr_rcu(head, ptr, type, memb) \
 755 ({ \
 756         list_next_or_null_rcu(head, ptr, type, memb) ?: \
 757                 list_next_or_null_rcu(head, READ_ONCE((ptr)->next), \
 758                                       type, memb); \
 759 })
 760
 761 /**
 762  * get_next_path_rr() - Returns path in round-robin fashion.
 763  * @it: the path pointer
 764  *
 765  * Related to @MP_POLICY_RR
 766  *
 767  * Locks:
 768  *    rcu_read_lock() must be hold.
 769  */
 770 static struct rtrs_clt_sess *get_next_path_rr(struct path_it *it)
 771 {
 772         struct rtrs_clt_sess __rcu **ppcpu_path;
 773         struct rtrs_clt_sess *path;
 774         struct rtrs_clt *clt;
 775
 776         clt = it->clt;
 777
 778         /*
 779          * Here we use two RCU objects: @paths_list and @pcpu_path
 780          * pointer.  See rtrs_clt_remove_path_from_arr() for details
 781          * how that is handled.
 782          */
 783
 784         ppcpu_path = this_cpu_ptr(clt->pcpu_path);
 785         path = rcu_dereference(*ppcpu_path);
 786         if (unlikely(!path))
 787                 path = list_first_or_null_rcu(&clt->paths_list,
 788                                               typeof(*path), s.entry);
 789         else
 790                 path = list_next_or_null_rr_rcu(&clt->paths_list,
 791                                                 &path->s.entry,
 792                                                 typeof(*path),
 793                                                 s.entry);
 794         rcu_assign_pointer(*ppcpu_path, path);
 795
 796         return path;
 797 }
 798
 799 /**
 800  * get_next_path_min_inflight() - Returns path with minimal inflight count.
 801  * @it: the path pointer
 802  *
 803  * Related to @MP_POLICY_MIN_INFLIGHT
 804  *
 805  * Locks:
 806  *    rcu_read_lock() must be hold.
 807  */
 808 static struct rtrs_clt_sess *get_next_path_min_inflight(struct path_it *it)
 809 {
 810         struct rtrs_clt_sess *min_path = NULL;
 811         struct rtrs_clt *clt = it->clt;
 812         struct rtrs_clt_sess *sess;
 813         int min_inflight = INT_MAX;
 814         int inflight;
 815
 816         list_for_each_entry_rcu(sess, &clt->paths_list, s.entry) {
 817                 if (unlikely(!list_empty(raw_cpu_ptr(sess->mp_skip_entry))))
 818                         continue;
 819
 820                 inflight = atomic_read(&sess->stats->inflight);
 821
 822                 if (inflight < min_inflight) {
 823                         min_inflight = inflight;
 824                         min_path = sess;
 825                 }
 826         }
 827
 828         /*
 829          * add the path to the skip list, so that next time we can get
 830          * a different one
 831          */
 832         if (min_path)
 833                 list_add(raw_cpu_ptr(min_path->mp_skip_entry), &it->skip_list);
 834
 835         return min_path;
 836 }
 837
 838 /**
 839  * get_next_path_min_latency() - Returns path with minimal latency.
 840  * @it: the path pointer
 841  *
 842  * Return: a path with the lowest latency or NULL if all paths are tried
 843  *
 844  * Locks:
 845  *    rcu_read_lock() must be hold.
 846  *
 847  * Related to @MP_POLICY_MIN_LATENCY
 848  *
 849  * This DOES skip an already-tried path.
 850  * There is a skip-list to skip a path if the path has tried but failed.
 851  * It will try the minimum latency path and then the second minimum latency
 852  * path and so on. Finally it will return NULL if all paths are tried.
 853  * Therefore the caller MUST check the returned
 854  * path is NULL and trigger the IO error.
 855  */
 856 static struct rtrs_clt_sess *get_next_path_min_latency(struct path_it *it)
 857 {
 858         struct rtrs_clt_sess *min_path = NULL;
 859         struct rtrs_clt *clt = it->clt;
 860         struct rtrs_clt_sess *sess;
 861         ktime_t min_latency = INT_MAX;
 862         ktime_t latency;
 863
 864         list_for_each_entry_rcu(sess, &clt->paths_list, s.entry) {
 865                 if (unlikely(READ_ONCE(sess->state) != RTRS_CLT_CONNECTED))
 866                         continue;
 867
 868                 if (unlikely(!list_empty(raw_cpu_ptr(sess->mp_skip_entry))))
 869                         continue;
 870
 871                 latency = sess->s.hb_cur_latency;
 872
 873                 if (latency < min_latency) {
 874                         min_latency = latency;
 875                         min_path = sess;
 876                 }
 877         }
 878
 879         /*
 880          * add the path to the skip list, so that next time we can get
 881          * a different one
 882          */
 883         if (min_path)
 884                 list_add(raw_cpu_ptr(min_path->mp_skip_entry), &it->skip_list);
 885
 886         return min_path;
 887 }
 888
 889 static inline void path_it_init(struct path_it *it, struct rtrs_clt *clt)
 890 {
 891         INIT_LIST_HEAD(&it->skip_list);
 892         it->clt = clt;
 893         it->i = 0;
 894
 895         if (clt->mp_policy == MP_POLICY_RR)
 896                 it->next_path = get_next_path_rr;
 897         else if (clt->mp_policy == MP_POLICY_MIN_INFLIGHT)
 898                 it->next_path = get_next_path_min_inflight;
 899         else
 900                 it->next_path = get_next_path_min_latency;
 901 }
 902
 903 static inline void path_it_deinit(struct path_it *it)
 904 {
 905         struct list_head *skip, *tmp;
 906         /*
 907          * The skip_list is used only for the MIN_INFLIGHT policy.
 908          * We need to remove paths from it, so that next IO can insert
 909          * paths (->mp_skip_entry) into a skip_list again.
 910          */
 911         list_for_each_safe(skip, tmp, &it->skip_list)
 912                 list_del_init(skip);
 913 }
 914
 915 /**
 916  * rtrs_clt_init_req() Initialize an rtrs_clt_io_req holding information
 917  * about an inflight IO.
 918  * The user buffer holding user control message (not data) is copied into
 919  * the corresponding buffer of rtrs_iu (req->iu->buf), which later on will
 920  * also hold the control message of rtrs.
 921  * @req: an io request holding information about IO.
 922  * @sess: client session
 923  * @conf: conformation callback function to notify upper layer.
 924  * @permit: permit for allocation of RDMA remote buffer
 925  * @priv: private pointer
 926  * @vec: kernel vector containing control message
 927  * @usr_len: length of the user message
 928  * @sg: scater list for IO data
 929  * @sg_cnt: number of scater list entries
 930  * @data_len: length of the IO data
 931  * @dir: direction of the IO.
 932  */
 933 static void rtrs_clt_init_req(struct rtrs_clt_io_req *req,
 934                               struct rtrs_clt_sess *sess,
 935                               void (*conf)(void *priv, int errno),
 936                               struct rtrs_permit *permit, void *priv,
 937                               const struct kvec *vec, size_t usr_len,
 938                               struct scatterlist *sg, size_t sg_cnt,
 939                               size_t data_len, int dir)
 940 {
 941         struct iov_iter iter;
 942         size_t len;
 943
 944         req->permit = permit;
 945         req->in_use = true;
 946         req->usr_len = usr_len;
 947         req->data_len = data_len;
 948         req->sglist = sg;
 949         req->sg_cnt = sg_cnt;
 950         req->priv = priv;
 951         req->dir = dir;
 952         req->con = rtrs_permit_to_clt_con(sess, permit);
 953         req->conf = conf;
 954         req->need_inv = false;
 955         req->need_inv_comp = false;
 956         req->inv_errno = 0;
 957
 958         iov_iter_kvec(&iter, READ, vec, 1, usr_len);
 959         len = _copy_from_iter(req->iu->buf, usr_len, &iter);
 960         WARN_ON(len != usr_len);
 961
 962         reinit_completion(&req->inv_comp);
 963 }
 964
 965 static struct rtrs_clt_io_req *
 966 rtrs_clt_get_req(struct rtrs_clt_sess *sess,
 967                  void (*conf)(void *priv, int errno),
 968                  struct rtrs_permit *permit, void *priv,
 969                  const struct kvec *vec, size_t usr_len,
 970                  struct scatterlist *sg, size_t sg_cnt,
 971                  size_t data_len, int dir)
 972 {
 973         struct rtrs_clt_io_req *req;
 974
 975         req = &sess->reqs[permit->mem_id];
 976         rtrs_clt_init_req(req, sess, conf, permit, priv, vec, usr_len,
 977                            sg, sg_cnt, data_len, dir);
 978         return req;
 979 }
 980
 981 static struct rtrs_clt_io_req *
 982 rtrs_clt_get_copy_req(struct rtrs_clt_sess *alive_sess,
 983                        struct rtrs_clt_io_req *fail_req)
 984 {
 985         struct rtrs_clt_io_req *req;
 986         struct kvec vec = {
 987                 .iov_base = fail_req->iu->buf,
 988                 .iov_len  = fail_req->usr_len
 989         };
 990
 991         req = &alive_sess->reqs[fail_req->permit->mem_id];
 992         rtrs_clt_init_req(req, alive_sess, fail_req->conf, fail_req->permit,
 993                            fail_req->priv, &vec, fail_req->usr_len,
 994                            fail_req->sglist, fail_req->sg_cnt,
 995                            fail_req->data_len, fail_req->dir);
 996         return req;
 997 }
 998
 999 static int rtrs_post_rdma_write_sg(struct rtrs_clt_con *con,
1000                                     struct rtrs_clt_io_req *req,
1001                                     struct rtrs_rbuf *rbuf,
1002                                     u32 size, u32 imm)
1003 {
1004         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
1005         struct ib_sge *sge = req->sge;
1006         enum ib_send_flags flags;
1007         struct scatterlist *sg;
1008         size_t num_sge;
1009         int i;
1010
1011         for_each_sg(req->sglist, sg, req->sg_cnt, i) {
1012                 sge[i].addr   = sg_dma_address(sg);
1013                 sge[i].length = sg_dma_len(sg);
1014                 sge[i].lkey   = sess->s.dev->ib_pd->local_dma_lkey;
1015         }
1016         sge[i].addr   = req->iu->dma_addr;
1017         sge[i].length = size;
1018         sge[i].lkey   = sess->s.dev->ib_pd->local_dma_lkey;
1019
1020         num_sge = 1 + req->sg_cnt;
1021
1022         /*
1023          * From time to time we have to post signalled sends,
1024          * or send queue will fill up and only QP reset can help.
1025          */
1026         flags = atomic_inc_return(&con->io_cnt) % sess->queue_depth ?
1027                         0 : IB_SEND_SIGNALED;
1028
1029         ib_dma_sync_single_for_device(sess->s.dev->ib_dev, req->iu->dma_addr,
1030                                       size, DMA_TO_DEVICE);
1031
1032         return rtrs_iu_post_rdma_write_imm(&con->c, req->iu, sge, num_sge,
1033                                             rbuf->rkey, rbuf->addr, imm,
1034                                             flags, NULL);
1035 }
1036
1037 static int rtrs_clt_write_req(struct rtrs_clt_io_req *req)
1038 {
1039         struct rtrs_clt_con *con = req->con;
1040         struct rtrs_sess *s = con->c.sess;
1041         struct rtrs_clt_sess *sess = to_clt_sess(s);
1042         struct rtrs_msg_rdma_write *msg;
1043
1044         struct rtrs_rbuf *rbuf;
1045         int ret, count = 0;
1046         u32 imm, buf_id;
1047
1048         const size_t tsize = sizeof(*msg) + req->data_len + req->usr_len;
1049
1050         if (unlikely(tsize > sess->chunk_size)) {
1051                 rtrs_wrn(s, "Write request failed, size too big %zu > %d\n",
1052                           tsize, sess->chunk_size);
1053                 return -EMSGSIZE;
1054         }
1055         if (req->sg_cnt) {
1056                 count = ib_dma_map_sg(sess->s.dev->ib_dev, req->sglist,
1057                                       req->sg_cnt, req->dir);
1058                 if (unlikely(!count)) {
1059                         rtrs_wrn(s, "Write request failed, map failed\n");
1060                         return -EINVAL;
1061                 }
1062         }
1063         /* put rtrs msg after sg and user message */
1064         msg = req->iu->buf + req->usr_len;
1065         msg->type = cpu_to_le16(RTRS_MSG_WRITE);
1066         msg->usr_len = cpu_to_le16(req->usr_len);
1067
1068         /* rtrs message on server side will be after user data and message */
1069         imm = req->permit->mem_off + req->data_len + req->usr_len;
1070         imm = rtrs_to_io_req_imm(imm);
1071         buf_id = req->permit->mem_id;
1072         req->sg_size = tsize;
1073         rbuf = &sess->rbufs[buf_id];
1074
1075         /*
1076          * Update stats now, after request is successfully sent it is not
1077          * safe anymore to touch it.
1078          */
1079         rtrs_clt_update_all_stats(req, WRITE);
1080
1081         ret = rtrs_post_rdma_write_sg(req->con, req, rbuf,
1082                                        req->usr_len + sizeof(*msg),
1083                                        imm);
1084         if (unlikely(ret)) {
1085                 rtrs_err_rl(s,
1086                             "Write request failed: error=%d path=%s [%s:%u]\n",
1087                             ret, kobject_name(&sess->kobj), sess->hca_name,
1088                             sess->hca_port);
1089                 if (sess->clt->mp_policy == MP_POLICY_MIN_INFLIGHT)
1090                         atomic_dec(&sess->stats->inflight);
1091                 if (req->sg_cnt)
1092                         ib_dma_unmap_sg(sess->s.dev->ib_dev, req->sglist,
1093                                         req->sg_cnt, req->dir);
1094         }
1095
1096         return ret;
1097 }
1098
1099 static int rtrs_map_sg_fr(struct rtrs_clt_io_req *req, size_t count)
1100 {
1101         int nr;
1102
1103         /* Align the MR to a 4K page size to match the block virt boundary */
1104         nr = ib_map_mr_sg(req->mr, req->sglist, count, NULL, SZ_4K);
1105         if (nr < 0)
1106                 return nr;
1107         if (unlikely(nr < req->sg_cnt))
1108                 return -EINVAL;
1109         ib_update_fast_reg_key(req->mr, ib_inc_rkey(req->mr->rkey));
1110
1111         return nr;
1112 }
1113
1114 static int rtrs_clt_read_req(struct rtrs_clt_io_req *req)
1115 {
1116         struct rtrs_clt_con *con = req->con;
1117         struct rtrs_sess *s = con->c.sess;
1118         struct rtrs_clt_sess *sess = to_clt_sess(s);
1119         struct rtrs_msg_rdma_read *msg;
1120         struct rtrs_ib_dev *dev = sess->s.dev;
1121
1122         struct ib_reg_wr rwr;
1123         struct ib_send_wr *wr = NULL;
1124
1125         int ret, count = 0;
1126         u32 imm, buf_id;
1127
1128         const size_t tsize = sizeof(*msg) + req->data_len + req->usr_len;
1129
1130         if (unlikely(tsize > sess->chunk_size)) {
1131                 rtrs_wrn(s,
1132                           "Read request failed, message size is %zu, bigger than CHUNK_SIZE %d\n",
1133                           tsize, sess->chunk_size);
1134                 return -EMSGSIZE;
1135         }
1136
1137         if (req->sg_cnt) {
1138                 count = ib_dma_map_sg(dev->ib_dev, req->sglist, req->sg_cnt,
1139                                       req->dir);
1140                 if (unlikely(!count)) {
1141                         rtrs_wrn(s,
1142                                   "Read request failed, dma map failed\n");
1143                         return -EINVAL;
1144                 }
1145         }
1146         /* put our message into req->buf after user message*/
1147         msg = req->iu->buf + req->usr_len;
1148         msg->type = cpu_to_le16(RTRS_MSG_READ);
1149         msg->usr_len = cpu_to_le16(req->usr_len);
1150
1151         if (count) {
1152                 ret = rtrs_map_sg_fr(req, count);
1153                 if (ret < 0) {
1154                         rtrs_err_rl(s,
1155                                      "Read request failed, failed to map  fast reg. data, err: %d\n",
1156                                      ret);
1157                         ib_dma_unmap_sg(dev->ib_dev, req->sglist, req->sg_cnt,
1158                                         req->dir);
1159                         return ret;
1160                 }
1161                 rwr = (struct ib_reg_wr) {
1162                         .wr.opcode = IB_WR_REG_MR,
1163                         .wr.wr_cqe = &fast_reg_cqe,
1164                         .mr = req->mr,
1165                         .key = req->mr->rkey,
1166                         .access = (IB_ACCESS_LOCAL_WRITE |
1167                                    IB_ACCESS_REMOTE_WRITE),
1168                 };
1169                 wr = &rwr.wr;
1170
1171                 msg->sg_cnt = cpu_to_le16(1);
1172                 msg->flags = cpu_to_le16(RTRS_MSG_NEED_INVAL_F);
1173
1174                 msg->desc[0].addr = cpu_to_le64(req->mr->iova);
1175                 msg->desc[0].key = cpu_to_le32(req->mr->rkey);
1176                 msg->desc[0].len = cpu_to_le32(req->mr->length);
1177
1178                 /* Further invalidation is required */
1179                 req->need_inv = !!RTRS_MSG_NEED_INVAL_F;
1180
1181         } else {
1182                 msg->sg_cnt = 0;
1183                 msg->flags = 0;
1184         }
1185         /*
1186          * rtrs message will be after the space reserved for disk data and
1187          * user message
1188          */
1189         imm = req->permit->mem_off + req->data_len + req->usr_len;
1190         imm = rtrs_to_io_req_imm(imm);
1191         buf_id = req->permit->mem_id;
1192
1193         req->sg_size  = sizeof(*msg);
1194         req->sg_size += le16_to_cpu(msg->sg_cnt) * sizeof(struct rtrs_sg_desc);
1195         req->sg_size += req->usr_len;
1196
1197         /*
1198          * Update stats now, after request is successfully sent it is not
1199          * safe anymore to touch it.
1200          */
1201         rtrs_clt_update_all_stats(req, READ);
1202
1203         ret = rtrs_post_send_rdma(req->con, req, &sess->rbufs[buf_id],
1204                                    req->data_len, imm, wr);
1205         if (unlikely(ret)) {
1206                 rtrs_err_rl(s,
1207                             "Read request failed: error=%d path=%s [%s:%u]\n",
1208                             ret, kobject_name(&sess->kobj), sess->hca_name,
1209                             sess->hca_port);
1210                 if (sess->clt->mp_policy == MP_POLICY_MIN_INFLIGHT)
1211                         atomic_dec(&sess->stats->inflight);
1212                 req->need_inv = false;
1213                 if (req->sg_cnt)
1214                         ib_dma_unmap_sg(dev->ib_dev, req->sglist,
1215                                         req->sg_cnt, req->dir);
1216         }
1217
1218         return ret;
1219 }
1220
1221 /**
1222  * rtrs_clt_failover_req() Try to find an active path for a failed request
1223  * @clt: clt context
1224  * @fail_req: a failed io request.
1225  */
1226 static int rtrs_clt_failover_req(struct rtrs_clt *clt,
1227                                  struct rtrs_clt_io_req *fail_req)
1228 {
1229         struct rtrs_clt_sess *alive_sess;
1230         struct rtrs_clt_io_req *req;
1231         int err = -ECONNABORTED;
1232         struct path_it it;
1233
1234         rcu_read_lock();
1235         for (path_it_init(&it, clt);
1236              (alive_sess = it.next_path(&it)) && it.i < it.clt->paths_num;
1237              it.i++) {
1238                 if (unlikely(READ_ONCE(alive_sess->state) !=
1239                              RTRS_CLT_CONNECTED))
1240                         continue;
1241                 req = rtrs_clt_get_copy_req(alive_sess, fail_req);
1242                 if (req->dir == DMA_TO_DEVICE)
1243                         err = rtrs_clt_write_req(req);
1244                 else
1245                         err = rtrs_clt_read_req(req);
1246                 if (unlikely(err)) {
1247                         req->in_use = false;
1248                         continue;
1249                 }
1250                 /* Success path */
1251                 rtrs_clt_inc_failover_cnt(alive_sess->stats);
1252                 break;
1253         }
1254         path_it_deinit(&it);
1255         rcu_read_unlock();
1256
1257         return err;
1258 }
1259
1260 static void fail_all_outstanding_reqs(struct rtrs_clt_sess *sess)
1261 {
1262         struct rtrs_clt *clt = sess->clt;
1263         struct rtrs_clt_io_req *req;
1264         int i, err;
1265
1266         if (!sess->reqs)
1267                 return;
1268         for (i = 0; i < sess->queue_depth; ++i) {
1269                 req = &sess->reqs[i];
1270                 if (!req->in_use)
1271                         continue;
1272
1273                 /*
1274                  * Safely (without notification) complete failed request.
1275                  * After completion this request is still useble and can
1276                  * be failovered to another path.
1277                  */
1278                 complete_rdma_req(req, -ECONNABORTED, false, true);
1279
1280                 err = rtrs_clt_failover_req(clt, req);
1281                 if (unlikely(err))
1282                         /* Failover failed, notify anyway */
1283                         req->conf(req->priv, err);
1284         }
1285 }
1286
1287 static void free_sess_reqs(struct rtrs_clt_sess *sess)
1288 {
1289         struct rtrs_clt_io_req *req;
1290         int i;
1291
1292         if (!sess->reqs)
1293                 return;
1294         for (i = 0; i < sess->queue_depth; ++i) {
1295                 req = &sess->reqs[i];
1296                 if (req->mr)
1297                         ib_dereg_mr(req->mr);
1298                 kfree(req->sge);
1299                 rtrs_iu_free(req->iu, sess->s.dev->ib_dev, 1);
1300         }
1301         kfree(sess->reqs);
1302         sess->reqs = NULL;
1303 }
1304
1305 static int alloc_sess_reqs(struct rtrs_clt_sess *sess)
1306 {
1307         struct rtrs_clt_io_req *req;
1308         struct rtrs_clt *clt = sess->clt;
1309         int i, err = -ENOMEM;
1310
1311         sess->reqs = kcalloc(sess->queue_depth, sizeof(*sess->reqs),
1312                              GFP_KERNEL);
1313         if (!sess->reqs)
1314                 return -ENOMEM;
1315
1316         for (i = 0; i < sess->queue_depth; ++i) {
1317                 req = &sess->reqs[i];
1318                 req->iu = rtrs_iu_alloc(1, sess->max_hdr_size, GFP_KERNEL,
1319                                          sess->s.dev->ib_dev,
1320                                          DMA_TO_DEVICE,
1321                                          rtrs_clt_rdma_done);
1322                 if (!req->iu)
1323                         goto out;
1324
1325                 req->sge = kmalloc_array(clt->max_segments + 1,
1326                                          sizeof(*req->sge), GFP_KERNEL);
1327                 if (!req->sge)
1328                         goto out;
1329
1330                 req->mr = ib_alloc_mr(sess->s.dev->ib_pd, IB_MR_TYPE_MEM_REG,
1331                                       sess->max_pages_per_mr);
1332                 if (IS_ERR(req->mr)) {
1333                         err = PTR_ERR(req->mr);
1334                         req->mr = NULL;
1335                         pr_err("Failed to alloc sess->max_pages_per_mr %d\n",
1336                                sess->max_pages_per_mr);
1337                         goto out;
1338                 }
1339
1340                 init_completion(&req->inv_comp);
1341         }
1342
1343         return 0;
1344
1345 out:
1346         free_sess_reqs(sess);
1347
1348         return err;
1349 }
1350
1351 static int alloc_permits(struct rtrs_clt *clt)
1352 {
1353         unsigned int chunk_bits;
1354         int err, i;
1355
1356         clt->permits_map = kcalloc(BITS_TO_LONGS(clt->queue_depth),
1357                                    sizeof(long), GFP_KERNEL);
1358         if (!clt->permits_map) {
1359                 err = -ENOMEM;
1360                 goto out_err;
1361         }
1362         clt->permits = kcalloc(clt->queue_depth, permit_size(clt), GFP_KERNEL);
1363         if (!clt->permits) {
1364                 err = -ENOMEM;
1365                 goto err_map;
1366         }
1367         chunk_bits = ilog2(clt->queue_depth - 1) + 1;
1368         for (i = 0; i < clt->queue_depth; i++) {
1369                 struct rtrs_permit *permit;
1370
1371                 permit = get_permit(clt, i);
1372                 permit->mem_id = i;
1373                 permit->mem_off = i << (MAX_IMM_PAYL_BITS - chunk_bits);
1374         }
1375
1376         return 0;
1377
1378 err_map:
1379         kfree(clt->permits_map);
1380         clt->permits_map = NULL;
1381 out_err:
1382         return err;
1383 }
1384
1385 static void free_permits(struct rtrs_clt *clt)
1386 {
1387         if (clt->permits_map) {
1388                 size_t sz = clt->queue_depth;
1389
1390                 wait_event(clt->permits_wait,
1391                            find_first_bit(clt->permits_map, sz) >= sz);
1392         }
1393         kfree(clt->permits_map);
1394         clt->permits_map = NULL;
1395         kfree(clt->permits);
1396         clt->permits = NULL;
1397 }
1398
1399 static void query_fast_reg_mode(struct rtrs_clt_sess *sess)
1400 {
1401         struct ib_device *ib_dev;
1402         u64 max_pages_per_mr;
1403         int mr_page_shift;
1404
1405         ib_dev = sess->s.dev->ib_dev;
1406
1407         /*
1408          * Use the smallest page size supported by the HCA, down to a
1409          * minimum of 4096 bytes. We're unlikely to build large sglists
1410          * out of smaller entries.
1411          */
1412         mr_page_shift      = max(12, ffs(ib_dev->attrs.page_size_cap) - 1);
1413         max_pages_per_mr   = ib_dev->attrs.max_mr_size;
1414         do_div(max_pages_per_mr, (1ull << mr_page_shift));
1415         sess->max_pages_per_mr =
1416                 min3(sess->max_pages_per_mr, (u32)max_pages_per_mr,
1417                      ib_dev->attrs.max_fast_reg_page_list_len);
1418         sess->max_send_sge = ib_dev->attrs.max_send_sge;
1419 }
1420
1421 static bool rtrs_clt_change_state_get_old(struct rtrs_clt_sess *sess,
1422                                            enum rtrs_clt_state new_state,
1423                                            enum rtrs_clt_state *old_state)
1424 {
1425         bool changed;
1426
1427         spin_lock_irq(&sess->state_wq.lock);
1428         if (old_state)
1429                 *old_state = sess->state;
1430         changed = rtrs_clt_change_state(sess, new_state);
1431         spin_unlock_irq(&sess->state_wq.lock);
1432
1433         return changed;
1434 }
1435
1436 static void rtrs_clt_hb_err_handler(struct rtrs_con *c)
1437 {
1438         struct rtrs_clt_con *con = container_of(c, typeof(*con), c);
1439
1440         rtrs_rdma_error_recovery(con);
1441 }
1442
1443 static void rtrs_clt_init_hb(struct rtrs_clt_sess *sess)
1444 {
1445         rtrs_init_hb(&sess->s, &io_comp_cqe,
1446                       RTRS_HB_INTERVAL_MS,
1447                       RTRS_HB_MISSED_MAX,
1448                       rtrs_clt_hb_err_handler,
1449                       rtrs_wq);
1450 }
1451
1452 static void rtrs_clt_start_hb(struct rtrs_clt_sess *sess)
1453 {
1454         rtrs_start_hb(&sess->s);
1455 }
1456
1457 static void rtrs_clt_stop_hb(struct rtrs_clt_sess *sess)
1458 {
1459         rtrs_stop_hb(&sess->s);
1460 }
1461
1462 static void rtrs_clt_reconnect_work(struct work_struct *work);
1463 static void rtrs_clt_close_work(struct work_struct *work);
1464
1465 static struct rtrs_clt_sess *alloc_sess(struct rtrs_clt *clt,
1466                                          const struct rtrs_addr *path,
1467                                          size_t con_num, u16 max_segments,
1468                                          u32 nr_poll_queues)
1469 {
1470         struct rtrs_clt_sess *sess;
1471         int err = -ENOMEM;
1472         int cpu;
1473         size_t total_con;
1474
1475         sess = kzalloc(sizeof(*sess), GFP_KERNEL);
1476         if (!sess)
1477                 goto err;
1478
1479         /*
1480          * irqmode and poll
1481          * +1: Extra connection for user messages
1482          */
1483         total_con = con_num + nr_poll_queues + 1;
1484         sess->s.con = kcalloc(total_con, sizeof(*sess->s.con), GFP_KERNEL);
1485         if (!sess->s.con)
1486                 goto err_free_sess;
1487
1488         sess->s.con_num = total_con;
1489         sess->s.irq_con_num = con_num + 1;
1490
1491         sess->stats = kzalloc(sizeof(*sess->stats), GFP_KERNEL);
1492         if (!sess->stats)
1493                 goto err_free_con;
1494
1495         mutex_init(&sess->init_mutex);
1496         uuid_gen(&sess->s.uuid);
1497         memcpy(&sess->s.dst_addr, path->dst,
1498                rdma_addr_size((struct sockaddr *)path->dst));
1499
1500         /*
1501          * rdma_resolve_addr() passes src_addr to cma_bind_addr, which
1502          * checks the sa_family to be non-zero. If user passed src_addr=NULL
1503          * the sess->src_addr will contain only zeros, which is then fine.
1504          */
1505         if (path->src)
1506                 memcpy(&sess->s.src_addr, path->src,
1507                        rdma_addr_size((struct sockaddr *)path->src));
1508         strlcpy(sess->s.sessname, clt->sessname, sizeof(sess->s.sessname));
1509         sess->clt = clt;
1510         sess->max_pages_per_mr = max_segments;
1511         init_waitqueue_head(&sess->state_wq);
1512         sess->state = RTRS_CLT_CONNECTING;
1513         atomic_set(&sess->connected_cnt, 0);
1514         INIT_WORK(&sess->close_work, rtrs_clt_close_work);
1515         INIT_DELAYED_WORK(&sess->reconnect_dwork, rtrs_clt_reconnect_work);
1516         rtrs_clt_init_hb(sess);
1517
1518         sess->mp_skip_entry = alloc_percpu(typeof(*sess->mp_skip_entry));
1519         if (!sess->mp_skip_entry)
1520                 goto err_free_stats;
1521
1522         for_each_possible_cpu(cpu)
1523                 INIT_LIST_HEAD(per_cpu_ptr(sess->mp_skip_entry, cpu));
1524
1525         err = rtrs_clt_init_stats(sess->stats);
1526         if (err)
1527                 goto err_free_percpu;
1528
1529         return sess;
1530
1531 err_free_percpu:
1532         free_percpu(sess->mp_skip_entry);
1533 err_free_stats:
1534         kfree(sess->stats);
1535 err_free_con:
1536         kfree(sess->s.con);
1537 err_free_sess:
1538         kfree(sess);
1539 err:
1540         return ERR_PTR(err);
1541 }
1542
1543 void free_sess(struct rtrs_clt_sess *sess)
1544 {
1545         free_percpu(sess->mp_skip_entry);
1546         mutex_destroy(&sess->init_mutex);
1547         kfree(sess->s.con);
1548         kfree(sess->rbufs);
1549         kfree(sess);
1550 }
1551
1552 static int create_con(struct rtrs_clt_sess *sess, unsigned int cid)
1553 {
1554         struct rtrs_clt_con *con;
1555
1556         con = kzalloc(sizeof(*con), GFP_KERNEL);
1557         if (!con)
1558                 return -ENOMEM;
1559
1560         /* Map first two connections to the first CPU */
1561         con->cpu  = (cid ? cid - 1 : 0) % nr_cpu_ids;
1562         con->c.cid = cid;
1563         con->c.sess = &sess->s;
1564         atomic_set(&con->io_cnt, 0);
1565         mutex_init(&con->con_mutex);
1566
1567         sess->s.con[cid] = &con->c;
1568
1569         return 0;
1570 }
1571
1572 static void destroy_con(struct rtrs_clt_con *con)
1573 {
1574         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
1575
1576         sess->s.con[con->c.cid] = NULL;
1577         mutex_destroy(&con->con_mutex);
1578         kfree(con);
1579 }
1580
1581 static int create_con_cq_qp(struct rtrs_clt_con *con)
1582 {
1583         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
1584         u32 max_send_wr, max_recv_wr, cq_size;
1585         int err, cq_vector;
1586         struct rtrs_msg_rkey_rsp *rsp;
1587
1588         lockdep_assert_held(&con->con_mutex);
1589         if (con->c.cid == 0) {
1590                 /*
1591                  * One completion for each receive and two for each send
1592                  * (send request + registration)
1593                  * + 2 for drain and heartbeat
1594                  * in case qp gets into error state
1595                  */
1596                 max_send_wr = SERVICE_CON_QUEUE_DEPTH * 2 + 2;
1597                 max_recv_wr = SERVICE_CON_QUEUE_DEPTH * 2 + 2;
1598                 /* We must be the first here */
1599                 if (WARN_ON(sess->s.dev))
1600                         return -EINVAL;
1601
1602                 /*
1603                  * The whole session uses device from user connection.
1604                  * Be careful not to close user connection before ib dev
1605                  * is gracefully put.
1606                  */
1607                 sess->s.dev = rtrs_ib_dev_find_or_add(con->c.cm_id->device,
1608                                                        &dev_pd);
1609                 if (!sess->s.dev) {
1610                         rtrs_wrn(sess->clt,
1611                                   "rtrs_ib_dev_find_get_or_add(): no memory\n");
1612                         return -ENOMEM;
1613                 }
1614                 sess->s.dev_ref = 1;
1615                 query_fast_reg_mode(sess);
1616         } else {
1617                 /*
1618                  * Here we assume that session members are correctly set.
1619                  * This is always true if user connection (cid == 0) is
1620                  * established first.
1621                  */
1622                 if (WARN_ON(!sess->s.dev))
1623                         return -EINVAL;
1624                 if (WARN_ON(!sess->queue_depth))
1625                         return -EINVAL;
1626
1627                 /* Shared between connections */
1628                 sess->s.dev_ref++;
1629                 max_send_wr =
1630                         min_t(int, sess->s.dev->ib_dev->attrs.max_qp_wr,
1631                               /* QD * (REQ + RSP + FR REGS or INVS) + drain */
1632                               sess->queue_depth * 3 + 1);
1633                 max_recv_wr =
1634                         min_t(int, sess->s.dev->ib_dev->attrs.max_qp_wr,
1635                               sess->queue_depth * 3 + 1);
1636         }
1637         /* alloc iu to recv new rkey reply when server reports flags set */
1638         if (sess->flags & RTRS_MSG_NEW_RKEY_F || con->c.cid == 0) {
1639                 con->rsp_ius = rtrs_iu_alloc(max_recv_wr, sizeof(*rsp),
1640                                               GFP_KERNEL, sess->s.dev->ib_dev,
1641                                               DMA_FROM_DEVICE,
1642                                               rtrs_clt_rdma_done);
1643                 if (!con->rsp_ius)
1644                         return -ENOMEM;
1645                 con->queue_size = max_recv_wr;
1646         }
1647         cq_size = max_send_wr + max_recv_wr;
1648         cq_vector = con->cpu % sess->s.dev->ib_dev->num_comp_vectors;
1649         if (con->c.cid >= sess->s.irq_con_num)
1650                 err = rtrs_cq_qp_create(&sess->s, &con->c, sess->max_send_sge,
1651                                         cq_vector, cq_size, max_send_wr,
1652                                         max_recv_wr, IB_POLL_DIRECT);
1653         else
1654                 err = rtrs_cq_qp_create(&sess->s, &con->c, sess->max_send_sge,
1655                                         cq_vector, cq_size, max_send_wr,
1656                                         max_recv_wr, IB_POLL_SOFTIRQ);
1657         /*
1658          * In case of error we do not bother to clean previous allocations,
1659          * since destroy_con_cq_qp() must be called.
1660          */
1661         return err;
1662 }
1663
1664 static void destroy_con_cq_qp(struct rtrs_clt_con *con)
1665 {
1666         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
1667
1668         /*
1669          * Be careful here: destroy_con_cq_qp() can be called even
1670          * create_con_cq_qp() failed, see comments there.
1671          */
1672         lockdep_assert_held(&con->con_mutex);
1673         rtrs_cq_qp_destroy(&con->c);
1674         if (con->rsp_ius) {
1675                 rtrs_iu_free(con->rsp_ius, sess->s.dev->ib_dev, con->queue_size);
1676                 con->rsp_ius = NULL;
1677                 con->queue_size = 0;
1678         }
1679         if (sess->s.dev_ref && !--sess->s.dev_ref) {
1680                 rtrs_ib_dev_put(sess->s.dev);
1681                 sess->s.dev = NULL;
1682         }
1683 }
1684
1685 static void stop_cm(struct rtrs_clt_con *con)
1686 {
1687         rdma_disconnect(con->c.cm_id);
1688         if (con->c.qp)
1689                 ib_drain_qp(con->c.qp);
1690 }
1691
1692 static void destroy_cm(struct rtrs_clt_con *con)
1693 {
1694         rdma_destroy_id(con->c.cm_id);
1695         con->c.cm_id = NULL;
1696 }
1697
1698 static int rtrs_rdma_addr_resolved(struct rtrs_clt_con *con)
1699 {
1700         struct rtrs_sess *s = con->c.sess;
1701         int err;
1702
1703         mutex_lock(&con->con_mutex);
1704         err = create_con_cq_qp(con);
1705         mutex_unlock(&con->con_mutex);
1706         if (err) {
1707                 rtrs_err(s, "create_con_cq_qp(), err: %d\n", err);
1708                 return err;
1709         }
1710         err = rdma_resolve_route(con->c.cm_id, RTRS_CONNECT_TIMEOUT_MS);
1711         if (err)
1712                 rtrs_err(s, "Resolving route failed, err: %d\n", err);
1713
1714         return err;
1715 }
1716
1717 static int rtrs_rdma_route_resolved(struct rtrs_clt_con *con)
1718 {
1719         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
1720         struct rtrs_clt *clt = sess->clt;
1721         struct rtrs_msg_conn_req msg;
1722         struct rdma_conn_param param;
1723
1724         int err;
1725
1726         param = (struct rdma_conn_param) {
1727                 .retry_count = 7,
1728                 .rnr_retry_count = 7,
1729                 .private_data = &msg,
1730                 .private_data_len = sizeof(msg),
1731         };
1732
1733         msg = (struct rtrs_msg_conn_req) {
1734                 .magic = cpu_to_le16(RTRS_MAGIC),
1735                 .version = cpu_to_le16(RTRS_PROTO_VER),
1736                 .cid = cpu_to_le16(con->c.cid),
1737                 .cid_num = cpu_to_le16(sess->s.con_num),
1738                 .recon_cnt = cpu_to_le16(sess->s.recon_cnt),
1739         };
1740         msg.first_conn = sess->for_new_clt ? FIRST_CONN : 0;
1741         uuid_copy(&msg.sess_uuid, &sess->s.uuid);
1742         uuid_copy(&msg.paths_uuid, &clt->paths_uuid);
1743
1744         err = rdma_connect_locked(con->c.cm_id, &param);
1745         if (err)
1746                 rtrs_err(clt, "rdma_connect_locked(): %d\n", err);
1747
1748         return err;
1749 }
1750
1751 static int rtrs_rdma_conn_established(struct rtrs_clt_con *con,
1752                                        struct rdma_cm_event *ev)
1753 {
1754         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
1755         struct rtrs_clt *clt = sess->clt;
1756         const struct rtrs_msg_conn_rsp *msg;
1757         u16 version, queue_depth;
1758         int errno;
1759         u8 len;
1760
1761         msg = ev->param.conn.private_data;
1762         len = ev->param.conn.private_data_len;
1763         if (len < sizeof(*msg)) {
1764                 rtrs_err(clt, "Invalid RTRS connection response\n");
1765                 return -ECONNRESET;
1766         }
1767         if (le16_to_cpu(msg->magic) != RTRS_MAGIC) {
1768                 rtrs_err(clt, "Invalid RTRS magic\n");
1769                 return -ECONNRESET;
1770         }
1771         version = le16_to_cpu(msg->version);
1772         if (version >> 8 != RTRS_PROTO_VER_MAJOR) {
1773                 rtrs_err(clt, "Unsupported major RTRS version: %d, expected %d\n",
1774                           version >> 8, RTRS_PROTO_VER_MAJOR);
1775                 return -ECONNRESET;
1776         }
1777         errno = le16_to_cpu(msg->errno);
1778         if (errno) {
1779                 rtrs_err(clt, "Invalid RTRS message: errno %d\n",
1780                           errno);
1781                 return -ECONNRESET;
1782         }
1783         if (con->c.cid == 0) {
1784                 queue_depth = le16_to_cpu(msg->queue_depth);
1785
1786                 if (queue_depth > MAX_SESS_QUEUE_DEPTH) {
1787                         rtrs_err(clt, "Invalid RTRS message: queue=%d\n",
1788                                   queue_depth);
1789                         return -ECONNRESET;
1790                 }
1791                 if (!sess->rbufs || sess->queue_depth < queue_depth) {
1792                         kfree(sess->rbufs);
1793                         sess->rbufs = kcalloc(queue_depth, sizeof(*sess->rbufs),
1794                                               GFP_KERNEL);
1795                         if (!sess->rbufs)
1796                                 return -ENOMEM;
1797                 }
1798                 sess->queue_depth = queue_depth;
1799                 sess->max_hdr_size = le32_to_cpu(msg->max_hdr_size);
1800                 sess->max_io_size = le32_to_cpu(msg->max_io_size);
1801                 sess->flags = le32_to_cpu(msg->flags);
1802                 sess->chunk_size = sess->max_io_size + sess->max_hdr_size;
1803
1804                 /*
1805                  * Global queue depth and IO size is always a minimum.
1806                  * If while a reconnection server sends us a value a bit
1807                  * higher - client does not care and uses cached minimum.
1808                  *
1809                  * Since we can have several sessions (paths) restablishing
1810                  * connections in parallel, use lock.
1811                  */
1812                 mutex_lock(&clt->paths_mutex);
1813                 clt->queue_depth = min_not_zero(sess->queue_depth,
1814                                                 clt->queue_depth);
1815                 clt->max_io_size = min_not_zero(sess->max_io_size,
1816                                                 clt->max_io_size);
1817                 mutex_unlock(&clt->paths_mutex);
1818
1819                 /*
1820                  * Cache the hca_port and hca_name for sysfs
1821                  */
1822                 sess->hca_port = con->c.cm_id->port_num;
1823                 scnprintf(sess->hca_name, sizeof(sess->hca_name),
1824                           sess->s.dev->ib_dev->name);
1825                 sess->s.src_addr = con->c.cm_id->route.addr.src_addr;
1826                 /* set for_new_clt, to allow future reconnect on any path */
1827                 sess->for_new_clt = 1;
1828         }
1829
1830         return 0;
1831 }
1832
1833 static inline void flag_success_on_conn(struct rtrs_clt_con *con)
1834 {
1835         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
1836
1837         atomic_inc(&sess->connected_cnt);
1838         con->cm_err = 1;
1839 }
1840
1841 static int rtrs_rdma_conn_rejected(struct rtrs_clt_con *con,
1842                                     struct rdma_cm_event *ev)
1843 {
1844         struct rtrs_sess *s = con->c.sess;
1845         const struct rtrs_msg_conn_rsp *msg;
1846         const char *rej_msg;
1847         int status, errno;
1848         u8 data_len;
1849
1850         status = ev->status;
1851         rej_msg = rdma_reject_msg(con->c.cm_id, status);
1852         msg = rdma_consumer_reject_data(con->c.cm_id, ev, &data_len);
1853
1854         if (msg && data_len >= sizeof(*msg)) {
1855                 errno = (int16_t)le16_to_cpu(msg->errno);
1856                 if (errno == -EBUSY)
1857                         rtrs_err(s,
1858                                   "Previous session is still exists on the server, please reconnect later\n");
1859                 else
1860                         rtrs_err(s,
1861                                   "Connect rejected: status %d (%s), rtrs errno %d\n",
1862                                   status, rej_msg, errno);
1863         } else {
1864                 rtrs_err(s,
1865                           "Connect rejected but with malformed message: status %d (%s)\n",
1866                           status, rej_msg);
1867         }
1868
1869         return -ECONNRESET;
1870 }
1871
1872 static void rtrs_clt_close_conns(struct rtrs_clt_sess *sess, bool wait)
1873 {
1874         if (rtrs_clt_change_state_get_old(sess, RTRS_CLT_CLOSING, NULL))
1875                 queue_work(rtrs_wq, &sess->close_work);
1876         if (wait)
1877                 flush_work(&sess->close_work);
1878 }
1879
1880 static inline void flag_error_on_conn(struct rtrs_clt_con *con, int cm_err)
1881 {
1882         if (con->cm_err == 1) {
1883                 struct rtrs_clt_sess *sess;
1884
1885                 sess = to_clt_sess(con->c.sess);
1886                 if (atomic_dec_and_test(&sess->connected_cnt))
1887
1888                         wake_up(&sess->state_wq);
1889         }
1890         con->cm_err = cm_err;
1891 }
1892
1893 static int rtrs_clt_rdma_cm_handler(struct rdma_cm_id *cm_id,
1894                                      struct rdma_cm_event *ev)
1895 {
1896         struct rtrs_clt_con *con = cm_id->context;
1897         struct rtrs_sess *s = con->c.sess;
1898         struct rtrs_clt_sess *sess = to_clt_sess(s);
1899         int cm_err = 0;
1900
1901         switch (ev->event) {
1902         case RDMA_CM_EVENT_ADDR_RESOLVED:
1903                 cm_err = rtrs_rdma_addr_resolved(con);
1904                 break;
1905         case RDMA_CM_EVENT_ROUTE_RESOLVED:
1906                 cm_err = rtrs_rdma_route_resolved(con);
1907                 break;
1908         case RDMA_CM_EVENT_ESTABLISHED:
1909                 cm_err = rtrs_rdma_conn_established(con, ev);
1910                 if (likely(!cm_err)) {
1911                         /*
1912                          * Report success and wake up. Here we abuse state_wq,
1913                          * i.e. wake up without state change, but we set cm_err.
1914                          */
1915                         flag_success_on_conn(con);
1916                         wake_up(&sess->state_wq);
1917                         return 0;
1918                 }
1919                 break;
1920         case RDMA_CM_EVENT_REJECTED:
1921                 cm_err = rtrs_rdma_conn_rejected(con, ev);
1922                 break;
1923         case RDMA_CM_EVENT_DISCONNECTED:
1924                 /* No message for disconnecting */
1925                 cm_err = -ECONNRESET;
1926                 break;
1927         case RDMA_CM_EVENT_CONNECT_ERROR:
1928         case RDMA_CM_EVENT_UNREACHABLE:
1929         case RDMA_CM_EVENT_ADDR_CHANGE:
1930         case RDMA_CM_EVENT_TIMEWAIT_EXIT:
1931                 rtrs_wrn(s, "CM error (CM event: %s, err: %d)\n",
1932                          rdma_event_msg(ev->event), ev->status);
1933                 cm_err = -ECONNRESET;
1934                 break;
1935         case RDMA_CM_EVENT_ADDR_ERROR:
1936         case RDMA_CM_EVENT_ROUTE_ERROR:
1937                 rtrs_wrn(s, "CM error (CM event: %s, err: %d)\n",
1938                          rdma_event_msg(ev->event), ev->status);
1939                 cm_err = -EHOSTUNREACH;
1940                 break;
1941         case RDMA_CM_EVENT_DEVICE_REMOVAL:
1942                 /*
1943                  * Device removal is a special case.  Queue close and return 0.
1944                  */
1945                 rtrs_clt_close_conns(sess, false);
1946                 return 0;
1947         default:
1948                 rtrs_err(s, "Unexpected RDMA CM error (CM event: %s, err: %d)\n",
1949                          rdma_event_msg(ev->event), ev->status);
1950                 cm_err = -ECONNRESET;
1951                 break;
1952         }
1953
1954         if (cm_err) {
1955                 /*
1956                  * cm error makes sense only on connection establishing,
1957                  * in other cases we rely on normal procedure of reconnecting.
1958                  */
1959                 flag_error_on_conn(con, cm_err);
1960                 rtrs_rdma_error_recovery(con);
1961         }
1962
1963         return 0;
1964 }
1965
1966 static int create_cm(struct rtrs_clt_con *con)
1967 {
1968         struct rtrs_sess *s = con->c.sess;
1969         struct rtrs_clt_sess *sess = to_clt_sess(s);
1970         struct rdma_cm_id *cm_id;
1971         int err;
1972
1973         cm_id = rdma_create_id(&init_net, rtrs_clt_rdma_cm_handler, con,
1974                                sess->s.dst_addr.ss_family == AF_IB ?
1975                                RDMA_PS_IB : RDMA_PS_TCP, IB_QPT_RC);
1976         if (IS_ERR(cm_id)) {
1977                 err = PTR_ERR(cm_id);
1978                 rtrs_err(s, "Failed to create CM ID, err: %d\n", err);
1979
1980                 return err;
1981         }
1982         con->c.cm_id = cm_id;
1983         con->cm_err = 0;
1984         /* allow the port to be reused */
1985         err = rdma_set_reuseaddr(cm_id, 1);
1986         if (err != 0) {
1987                 rtrs_err(s, "Set address reuse failed, err: %d\n", err);
1988                 goto destroy_cm;
1989         }
1990         err = rdma_resolve_addr(cm_id, (struct sockaddr *)&sess->s.src_addr,
1991                                 (struct sockaddr *)&sess->s.dst_addr,
1992                                 RTRS_CONNECT_TIMEOUT_MS);
1993         if (err) {
1994                 rtrs_err(s, "Failed to resolve address, err: %d\n", err);
1995                 goto destroy_cm;
1996         }
1997         /*
1998          * Combine connection status and session events. This is needed
1999          * for waiting two possible cases: cm_err has something meaningful
2000          * or session state was really changed to error by device removal.
2001          */
2002         err = wait_event_interruptible_timeout(
2003                         sess->state_wq,
2004                         con->cm_err || sess->state != RTRS_CLT_CONNECTING,
2005                         msecs_to_jiffies(RTRS_CONNECT_TIMEOUT_MS));
2006         if (err == 0 || err == -ERESTARTSYS) {
2007                 if (err == 0)
2008                         err = -ETIMEDOUT;
2009                 /* Timedout or interrupted */
2010                 goto errr;
2011         }
2012         if (con->cm_err < 0) {
2013                 err = con->cm_err;
2014                 goto errr;
2015         }
2016         if (READ_ONCE(sess->state) != RTRS_CLT_CONNECTING) {
2017                 /* Device removal */
2018                 err = -ECONNABORTED;
2019                 goto errr;
2020         }
2021
2022         return 0;
2023
2024 errr:
2025         stop_cm(con);
2026         mutex_lock(&con->con_mutex);
2027         destroy_con_cq_qp(con);
2028         mutex_unlock(&con->con_mutex);
2029 destroy_cm:
2030         destroy_cm(con);
2031
2032         return err;
2033 }
2034
2035 static void rtrs_clt_sess_up(struct rtrs_clt_sess *sess)
2036 {
2037         struct rtrs_clt *clt = sess->clt;
2038         int up;
2039
2040         /*
2041          * We can fire RECONNECTED event only when all paths were
2042          * connected on rtrs_clt_open(), then each was disconnected
2043          * and the first one connected again.  That's why this nasty
2044          * game with counter value.
2045          */
2046
2047         mutex_lock(&clt->paths_ev_mutex);
2048         up = ++clt->paths_up;
2049         /*
2050          * Here it is safe to access paths num directly since up counter
2051          * is greater than MAX_PATHS_NUM only while rtrs_clt_open() is
2052          * in progress, thus paths removals are impossible.
2053          */
2054         if (up > MAX_PATHS_NUM && up == MAX_PATHS_NUM + clt->paths_num)
2055                 clt->paths_up = clt->paths_num;
2056         else if (up == 1)
2057                 clt->link_ev(clt->priv, RTRS_CLT_LINK_EV_RECONNECTED);
2058         mutex_unlock(&clt->paths_ev_mutex);
2059
2060         /* Mark session as established */
2061         sess->established = true;
2062         sess->reconnect_attempts = 0;
2063         sess->stats->reconnects.successful_cnt++;
2064 }
2065
2066 static void rtrs_clt_sess_down(struct rtrs_clt_sess *sess)
2067 {
2068         struct rtrs_clt *clt = sess->clt;
2069
2070         if (!sess->established)
2071                 return;
2072
2073         sess->established = false;
2074         mutex_lock(&clt->paths_ev_mutex);
2075         WARN_ON(!clt->paths_up);
2076         if (--clt->paths_up == 0)
2077                 clt->link_ev(clt->priv, RTRS_CLT_LINK_EV_DISCONNECTED);
2078         mutex_unlock(&clt->paths_ev_mutex);
2079 }
2080
2081 static void rtrs_clt_stop_and_destroy_conns(struct rtrs_clt_sess *sess)
2082 {
2083         struct rtrs_clt_con *con;
2084         unsigned int cid;
2085
2086         WARN_ON(READ_ONCE(sess->state) == RTRS_CLT_CONNECTED);
2087
2088         /*
2089          * Possible race with rtrs_clt_open(), when DEVICE_REMOVAL comes
2090          * exactly in between.  Start destroying after it finishes.
2091          */
2092         mutex_lock(&sess->init_mutex);
2093         mutex_unlock(&sess->init_mutex);
2094
2095         /*
2096          * All IO paths must observe !CONNECTED state before we
2097          * free everything.
2098          */
2099         synchronize_rcu();
2100
2101         rtrs_clt_stop_hb(sess);
2102
2103         /*
2104          * The order it utterly crucial: firstly disconnect and complete all
2105          * rdma requests with error (thus set in_use=false for requests),
2106          * then fail outstanding requests checking in_use for each, and
2107          * eventually notify upper layer about session disconnection.
2108          */
2109
2110         for (cid = 0; cid < sess->s.con_num; cid++) {
2111                 if (!sess->s.con[cid])
2112                         break;
2113                 con = to_clt_con(sess->s.con[cid]);
2114                 stop_cm(con);
2115         }
2116         fail_all_outstanding_reqs(sess);
2117         free_sess_reqs(sess);
2118         rtrs_clt_sess_down(sess);
2119
2120         /*
2121          * Wait for graceful shutdown, namely when peer side invokes
2122          * rdma_disconnect(). 'connected_cnt' is decremented only on
2123          * CM events, thus if other side had crashed and hb has detected
2124          * something is wrong, here we will stuck for exactly timeout ms,
2125          * since CM does not fire anything.  That is fine, we are not in
2126          * hurry.
2127          */
2128         wait_event_timeout(sess->state_wq, !atomic_read(&sess->connected_cnt),
2129                            msecs_to_jiffies(RTRS_CONNECT_TIMEOUT_MS));
2130
2131         for (cid = 0; cid < sess->s.con_num; cid++) {
2132                 if (!sess->s.con[cid])
2133                         break;
2134                 con = to_clt_con(sess->s.con[cid]);
2135                 mutex_lock(&con->con_mutex);
2136                 destroy_con_cq_qp(con);
2137                 mutex_unlock(&con->con_mutex);
2138                 destroy_cm(con);
2139                 destroy_con(con);
2140         }
2141 }
2142
2143 static inline bool xchg_sessions(struct rtrs_clt_sess __rcu **rcu_ppcpu_path,
2144                                  struct rtrs_clt_sess *sess,
2145                                  struct rtrs_clt_sess *next)
2146 {
2147         struct rtrs_clt_sess **ppcpu_path;
2148
2149         /* Call cmpxchg() without sparse warnings */
2150         ppcpu_path = (typeof(ppcpu_path))rcu_ppcpu_path;
2151         return sess == cmpxchg(ppcpu_path, sess, next);
2152 }
2153
2154 static void rtrs_clt_remove_path_from_arr(struct rtrs_clt_sess *sess)
2155 {
2156         struct rtrs_clt *clt = sess->clt;
2157         struct rtrs_clt_sess *next;
2158         bool wait_for_grace = false;
2159         int cpu;
2160
2161         mutex_lock(&clt->paths_mutex);
2162         list_del_rcu(&sess->s.entry);
2163
2164         /* Make sure everybody observes path removal. */
2165         synchronize_rcu();
2166
2167         /*
2168          * At this point nobody sees @sess in the list, but still we have
2169          * dangling pointer @pcpu_path which _can_ point to @sess.  Since
2170          * nobody can observe @sess in the list, we guarantee that IO path
2171          * will not assign @sess to @pcpu_path, i.e. @pcpu_path can be equal
2172          * to @sess, but can never again become @sess.
2173          */
2174
2175         /*
2176          * Decrement paths number only after grace period, because
2177          * caller of do_each_path() must firstly observe list without
2178          * path and only then decremented paths number.
2179          *
2180          * Otherwise there can be the following situation:
2181          *    o Two paths exist and IO is coming.
2182          *    o One path is removed:
2183          *      CPU#0                          CPU#1
2184          *      do_each_path():                rtrs_clt_remove_path_from_arr():
2185          *          path = get_next_path()
2186          *          ^^^                            list_del_rcu(path)
2187          *          [!CONNECTED path]              clt->paths_num--
2188          *                                              ^^^^^^^^^
2189          *          load clt->paths_num                 from 2 to 1
2190          *                    ^^^^^^^^^
2191          *                    sees 1
2192          *
2193          *      path is observed as !CONNECTED, but do_each_path() loop
2194          *      ends, because expression i < clt->paths_num is false.
2195          */
2196         clt->paths_num--;
2197
2198         /*
2199          * Get @next connection from current @sess which is going to be
2200          * removed.  If @sess is the last element, then @next is NULL.
2201          */
2202         rcu_read_lock();
2203         next = list_next_or_null_rr_rcu(&clt->paths_list, &sess->s.entry,
2204                                         typeof(*next), s.entry);
2205         rcu_read_unlock();
2206
2207         /*
2208          * @pcpu paths can still point to the path which is going to be
2209          * removed, so change the pointer manually.
2210          */
2211         for_each_possible_cpu(cpu) {
2212                 struct rtrs_clt_sess __rcu **ppcpu_path;
2213
2214                 ppcpu_path = per_cpu_ptr(clt->pcpu_path, cpu);
2215                 if (rcu_dereference_protected(*ppcpu_path,
2216                         lockdep_is_held(&clt->paths_mutex)) != sess)
2217                         /*
2218                          * synchronize_rcu() was called just after deleting
2219                          * entry from the list, thus IO code path cannot
2220                          * change pointer back to the pointer which is going
2221                          * to be removed, we are safe here.
2222                          */
2223                         continue;
2224
2225                 /*
2226                  * We race with IO code path, which also changes pointer,
2227                  * thus we have to be careful not to overwrite it.
2228                  */
2229                 if (xchg_sessions(ppcpu_path, sess, next))
2230                         /*
2231                          * @ppcpu_path was successfully replaced with @next,
2232                          * that means that someone could also pick up the
2233                          * @sess and dereferencing it right now, so wait for
2234                          * a grace period is required.
2235                          */
2236                         wait_for_grace = true;
2237         }
2238         if (wait_for_grace)
2239                 synchronize_rcu();
2240
2241         mutex_unlock(&clt->paths_mutex);
2242 }
2243
2244 static void rtrs_clt_add_path_to_arr(struct rtrs_clt_sess *sess)
2245 {
2246         struct rtrs_clt *clt = sess->clt;
2247
2248         mutex_lock(&clt->paths_mutex);
2249         clt->paths_num++;
2250
2251         list_add_tail_rcu(&sess->s.entry, &clt->paths_list);
2252         mutex_unlock(&clt->paths_mutex);
2253 }
2254
2255 static void rtrs_clt_close_work(struct work_struct *work)
2256 {
2257         struct rtrs_clt_sess *sess;
2258
2259         sess = container_of(work, struct rtrs_clt_sess, close_work);
2260
2261         cancel_delayed_work_sync(&sess->reconnect_dwork);
2262         rtrs_clt_stop_and_destroy_conns(sess);
2263         rtrs_clt_change_state_get_old(sess, RTRS_CLT_CLOSED, NULL);
2264 }
2265
2266 static int init_conns(struct rtrs_clt_sess *sess)
2267 {
2268         unsigned int cid;
2269         int err;
2270
2271         /*
2272          * On every new session connections increase reconnect counter
2273          * to avoid clashes with previous sessions not yet closed
2274          * sessions on a server side.
2275          */
2276         sess->s.recon_cnt++;
2277
2278         /* Establish all RDMA connections  */
2279         for (cid = 0; cid < sess->s.con_num; cid++) {
2280                 err = create_con(sess, cid);
2281                 if (err)
2282                         goto destroy;
2283
2284                 err = create_cm(to_clt_con(sess->s.con[cid]));
2285                 if (err) {
2286                         destroy_con(to_clt_con(sess->s.con[cid]));
2287                         goto destroy;
2288                 }
2289         }
2290         err = alloc_sess_reqs(sess);
2291         if (err)
2292                 goto destroy;
2293
2294         rtrs_clt_start_hb(sess);
2295
2296         return 0;
2297
2298 destroy:
2299         while (cid--) {
2300                 struct rtrs_clt_con *con = to_clt_con(sess->s.con[cid]);
2301
2302                 stop_cm(con);
2303
2304                 mutex_lock(&con->con_mutex);
2305                 destroy_con_cq_qp(con);
2306                 mutex_unlock(&con->con_mutex);
2307                 destroy_cm(con);
2308                 destroy_con(con);
2309         }
2310         /*
2311          * If we've never taken async path and got an error, say,
2312          * doing rdma_resolve_addr(), switch to CONNECTION_ERR state
2313          * manually to keep reconnecting.
2314          */
2315         rtrs_clt_change_state_get_old(sess, RTRS_CLT_CONNECTING_ERR, NULL);
2316
2317         return err;
2318 }
2319
2320 static void rtrs_clt_info_req_done(struct ib_cq *cq, struct ib_wc *wc)
2321 {
2322         struct rtrs_clt_con *con = to_clt_con(wc->qp->qp_context);
2323         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
2324         struct rtrs_iu *iu;
2325
2326         iu = container_of(wc->wr_cqe, struct rtrs_iu, cqe);
2327         rtrs_iu_free(iu, sess->s.dev->ib_dev, 1);
2328
2329         if (unlikely(wc->status != IB_WC_SUCCESS)) {
2330                 rtrs_err(sess->clt, "Sess info request send failed: %s\n",
2331                           ib_wc_status_msg(wc->status));
2332                 rtrs_clt_change_state_get_old(sess, RTRS_CLT_CONNECTING_ERR, NULL);
2333                 return;
2334         }
2335
2336         rtrs_clt_update_wc_stats(con);
2337 }
2338
2339 static int process_info_rsp(struct rtrs_clt_sess *sess,
2340                             const struct rtrs_msg_info_rsp *msg)
2341 {
2342         unsigned int sg_cnt, total_len;
2343         int i, sgi;
2344
2345         sg_cnt = le16_to_cpu(msg->sg_cnt);
2346         if (unlikely(!sg_cnt || (sess->queue_depth % sg_cnt))) {
2347                 rtrs_err(sess->clt, "Incorrect sg_cnt %d, is not multiple\n",
2348                           sg_cnt);
2349                 return -EINVAL;
2350         }
2351
2352         /*
2353          * Check if IB immediate data size is enough to hold the mem_id and
2354          * the offset inside the memory chunk.
2355          */
2356         if (unlikely((ilog2(sg_cnt - 1) + 1) +
2357                      (ilog2(sess->chunk_size - 1) + 1) >
2358                      MAX_IMM_PAYL_BITS)) {
2359                 rtrs_err(sess->clt,
2360                           "RDMA immediate size (%db) not enough to encode %d buffers of size %dB\n",
2361                           MAX_IMM_PAYL_BITS, sg_cnt, sess->chunk_size);
2362                 return -EINVAL;
2363         }
2364         total_len = 0;
2365         for (sgi = 0, i = 0; sgi < sg_cnt && i < sess->queue_depth; sgi++) {
2366                 const struct rtrs_sg_desc *desc = &msg->desc[sgi];
2367                 u32 len, rkey;
2368                 u64 addr;
2369
2370                 addr = le64_to_cpu(desc->addr);
2371                 rkey = le32_to_cpu(desc->key);
2372                 len  = le32_to_cpu(desc->len);
2373
2374                 total_len += len;
2375
2376                 if (unlikely(!len || (len % sess->chunk_size))) {
2377                         rtrs_err(sess->clt, "Incorrect [%d].len %d\n", sgi,
2378                                   len);
2379                         return -EINVAL;
2380                 }
2381                 for ( ; len && i < sess->queue_depth; i++) {
2382                         sess->rbufs[i].addr = addr;
2383                         sess->rbufs[i].rkey = rkey;
2384
2385                         len  -= sess->chunk_size;
2386                         addr += sess->chunk_size;
2387                 }
2388         }
2389         /* Sanity check */
2390         if (unlikely(sgi != sg_cnt || i != sess->queue_depth)) {
2391                 rtrs_err(sess->clt, "Incorrect sg vector, not fully mapped\n");
2392                 return -EINVAL;
2393         }
2394         if (unlikely(total_len != sess->chunk_size * sess->queue_depth)) {
2395                 rtrs_err(sess->clt, "Incorrect total_len %d\n", total_len);
2396                 return -EINVAL;
2397         }
2398
2399         return 0;
2400 }
2401
2402 static void rtrs_clt_info_rsp_done(struct ib_cq *cq, struct ib_wc *wc)
2403 {
2404         struct rtrs_clt_con *con = to_clt_con(wc->qp->qp_context);
2405         struct rtrs_clt_sess *sess = to_clt_sess(con->c.sess);
2406         struct rtrs_msg_info_rsp *msg;
2407         enum rtrs_clt_state state;
2408         struct rtrs_iu *iu;
2409         size_t rx_sz;
2410         int err;
2411
2412         state = RTRS_CLT_CONNECTING_ERR;
2413
2414         WARN_ON(con->c.cid);
2415         iu = container_of(wc->wr_cqe, struct rtrs_iu, cqe);
2416         if (unlikely(wc->status != IB_WC_SUCCESS)) {
2417                 rtrs_err(sess->clt, "Sess info response recv failed: %s\n",
2418                           ib_wc_status_msg(wc->status));
2419                 goto out;
2420         }
2421         WARN_ON(wc->opcode != IB_WC_RECV);
2422
2423         if (unlikely(wc->byte_len < sizeof(*msg))) {
2424                 rtrs_err(sess->clt, "Sess info response is malformed: size %d\n",
2425                           wc->byte_len);
2426                 goto out;
2427         }
2428         ib_dma_sync_single_for_cpu(sess->s.dev->ib_dev, iu->dma_addr,
2429                                    iu->size, DMA_FROM_DEVICE);
2430         msg = iu->buf;
2431         if (unlikely(le16_to_cpu(msg->type) != RTRS_MSG_INFO_RSP)) {
2432                 rtrs_err(sess->clt, "Sess info response is malformed: type %d\n",
2433                           le16_to_cpu(msg->type));
2434                 goto out;
2435         }
2436         rx_sz  = sizeof(*msg);
2437         rx_sz += sizeof(msg->desc[0]) * le16_to_cpu(msg->sg_cnt);
2438         if (unlikely(wc->byte_len < rx_sz)) {
2439                 rtrs_err(sess->clt, "Sess info response is malformed: size %d\n",
2440                           wc->byte_len);
2441                 goto out;
2442         }
2443         err = process_info_rsp(sess, msg);
2444         if (unlikely(err))
2445                 goto out;
2446
2447         err = post_recv_sess(sess);
2448         if (unlikely(err))
2449                 goto out;
2450
2451         state = RTRS_CLT_CONNECTED;
2452
2453 out:
2454         rtrs_clt_update_wc_stats(con);
2455         rtrs_iu_free(iu, sess->s.dev->ib_dev, 1);
2456         rtrs_clt_change_state_get_old(sess, state, NULL);
2457 }
2458
2459 static int rtrs_send_sess_info(struct rtrs_clt_sess *sess)
2460 {
2461         struct rtrs_clt_con *usr_con = to_clt_con(sess->s.con[0]);
2462         struct rtrs_msg_info_req *msg;
2463         struct rtrs_iu *tx_iu, *rx_iu;
2464         size_t rx_sz;
2465         int err;
2466
2467         rx_sz  = sizeof(struct rtrs_msg_info_rsp);
2468         rx_sz += sizeof(u64) * MAX_SESS_QUEUE_DEPTH;
2469
2470         tx_iu = rtrs_iu_alloc(1, sizeof(struct rtrs_msg_info_req), GFP_KERNEL,
2471                                sess->s.dev->ib_dev, DMA_TO_DEVICE,
2472                                rtrs_clt_info_req_done);
2473         rx_iu = rtrs_iu_alloc(1, rx_sz, GFP_KERNEL, sess->s.dev->ib_dev,
2474                                DMA_FROM_DEVICE, rtrs_clt_info_rsp_done);
2475         if (unlikely(!tx_iu || !rx_iu)) {
2476                 err = -ENOMEM;
2477                 goto out;
2478         }
2479         /* Prepare for getting info response */
2480         err = rtrs_iu_post_recv(&usr_con->c, rx_iu);
2481         if (unlikely(err)) {
2482                 rtrs_err(sess->clt, "rtrs_iu_post_recv(), err: %d\n", err);
2483                 goto out;
2484         }
2485         rx_iu = NULL;
2486
2487         msg = tx_iu->buf;
2488         msg->type = cpu_to_le16(RTRS_MSG_INFO_REQ);
2489         memcpy(msg->sessname, sess->s.sessname, sizeof(msg->sessname));
2490
2491         ib_dma_sync_single_for_device(sess->s.dev->ib_dev, tx_iu->dma_addr,
2492                                       tx_iu->size, DMA_TO_DEVICE);
2493
2494         /* Send info request */
2495         err = rtrs_iu_post_send(&usr_con->c, tx_iu, sizeof(*msg), NULL);
2496         if (unlikely(err)) {
2497                 rtrs_err(sess->clt, "rtrs_iu_post_send(), err: %d\n", err);
2498                 goto out;
2499         }
2500         tx_iu = NULL;
2501
2502         /* Wait for state change */
2503         wait_event_interruptible_timeout(sess->state_wq,
2504                                          sess->state != RTRS_CLT_CONNECTING,
2505                                          msecs_to_jiffies(
2506                                                  RTRS_CONNECT_TIMEOUT_MS));
2507         if (unlikely(READ_ONCE(sess->state) != RTRS_CLT_CONNECTED)) {
2508                 if (READ_ONCE(sess->state) == RTRS_CLT_CONNECTING_ERR)
2509                         err = -ECONNRESET;
2510                 else
2511                         err = -ETIMEDOUT;
2512         }
2513
2514 out:
2515         if (tx_iu)
2516                 rtrs_iu_free(tx_iu, sess->s.dev->ib_dev, 1);
2517         if (rx_iu)
2518                 rtrs_iu_free(rx_iu, sess->s.dev->ib_dev, 1);
2519         if (unlikely(err))
2520                 /* If we've never taken async path because of malloc problems */
2521                 rtrs_clt_change_state_get_old(sess, RTRS_CLT_CONNECTING_ERR, NULL);
2522
2523         return err;
2524 }
2525
2526 /**
2527  * init_sess() - establishes all session connections and does handshake
2528  * @sess: client session.
2529  * In case of error full close or reconnect procedure should be taken,
2530  * because reconnect or close async works can be started.
2531  */
2532 static int init_sess(struct rtrs_clt_sess *sess)
2533 {
2534         int err;
2535         char str[NAME_MAX];
2536         struct rtrs_addr path = {
2537                 .src = &sess->s.src_addr,
2538                 .dst = &sess->s.dst_addr,
2539         };
2540
2541         rtrs_addr_to_str(&path, str, sizeof(str));
2542
2543         mutex_lock(&sess->init_mutex);
2544         err = init_conns(sess);
2545         if (err) {
2546                 rtrs_err(sess->clt,
2547                          "init_conns() failed: err=%d path=%s [%s:%u]\n", err,
2548                          str, sess->hca_name, sess->hca_port);
2549                 goto out;
2550         }
2551         err = rtrs_send_sess_info(sess);
2552         if (err) {
2553                 rtrs_err(
2554                         sess->clt,
2555                         "rtrs_send_sess_info() failed: err=%d path=%s [%s:%u]\n",
2556                         err, str, sess->hca_name, sess->hca_port);
2557                 goto out;
2558         }
2559         rtrs_clt_sess_up(sess);
2560 out:
2561         mutex_unlock(&sess->init_mutex);
2562
2563         return err;
2564 }
2565
2566 static void rtrs_clt_reconnect_work(struct work_struct *work)
2567 {
2568         struct rtrs_clt_sess *sess;
2569         struct rtrs_clt *clt;
2570         unsigned int delay_ms;
2571         int err;
2572
2573         sess = container_of(to_delayed_work(work), struct rtrs_clt_sess,
2574                             reconnect_dwork);
2575         clt = sess->clt;
2576
2577         if (READ_ONCE(sess->state) != RTRS_CLT_RECONNECTING)
2578                 return;
2579
2580         if (sess->reconnect_attempts >= clt->max_reconnect_attempts) {
2581                 /* Close a session completely if max attempts is reached */
2582                 rtrs_clt_close_conns(sess, false);
2583                 return;
2584         }
2585         sess->reconnect_attempts++;
2586
2587         /* Stop everything */
2588         rtrs_clt_stop_and_destroy_conns(sess);
2589         msleep(RTRS_RECONNECT_BACKOFF);
2590         if (rtrs_clt_change_state_get_old(sess, RTRS_CLT_CONNECTING, NULL)) {
2591                 err = init_sess(sess);
2592                 if (err)
2593                         goto reconnect_again;
2594         }
2595
2596         return;
2597
2598 reconnect_again:
2599         if (rtrs_clt_change_state_get_old(sess, RTRS_CLT_RECONNECTING, NULL)) {
2600                 sess->stats->reconnects.fail_cnt++;
2601                 delay_ms = clt->reconnect_delay_sec * 1000;
2602                 queue_delayed_work(rtrs_wq, &sess->reconnect_dwork,
2603                                    msecs_to_jiffies(delay_ms +
2604                                                     prandom_u32() %
2605                                                     RTRS_RECONNECT_SEED));
2606         }
2607 }
2608
2609 static void rtrs_clt_dev_release(struct device *dev)
2610 {
2611         struct rtrs_clt *clt = container_of(dev, struct rtrs_clt, dev);
2612
2613         kfree(clt);
2614 }
2615
2616 static struct rtrs_clt *alloc_clt(const char *sessname, size_t paths_num,
2617                                   u16 port, size_t pdu_sz, void *priv,
2618                                   void  (*link_ev)(void *priv,
2619                                                    enum rtrs_clt_link_ev ev),
2620                                   unsigned int max_segments,
2621                                   unsigned int reconnect_delay_sec,
2622                                   unsigned int max_reconnect_attempts)
2623 {
2624         struct rtrs_clt *clt;
2625         int err;
2626
2627         if (!paths_num || paths_num > MAX_PATHS_NUM)
2628                 return ERR_PTR(-EINVAL);
2629
2630         if (strlen(sessname) >= sizeof(clt->sessname))
2631                 return ERR_PTR(-EINVAL);
2632
2633         clt = kzalloc(sizeof(*clt), GFP_KERNEL);
2634         if (!clt)
2635                 return ERR_PTR(-ENOMEM);
2636
2637         clt->pcpu_path = alloc_percpu(typeof(*clt->pcpu_path));
2638         if (!clt->pcpu_path) {
2639                 kfree(clt);
2640                 return ERR_PTR(-ENOMEM);
2641         }
2642
2643         uuid_gen(&clt->paths_uuid);
2644         INIT_LIST_HEAD_RCU(&clt->paths_list);
2645         clt->paths_num = paths_num;
2646         clt->paths_up = MAX_PATHS_NUM;
2647         clt->port = port;
2648         clt->pdu_sz = pdu_sz;
2649         clt->max_segments = max_segments;
2650         clt->reconnect_delay_sec = reconnect_delay_sec;
2651         clt->max_reconnect_attempts = max_reconnect_attempts;
2652         clt->priv = priv;
2653         clt->link_ev = link_ev;
2654         clt->mp_policy = MP_POLICY_MIN_INFLIGHT;
2655         strlcpy(clt->sessname, sessname, sizeof(clt->sessname));
2656         init_waitqueue_head(&clt->permits_wait);
2657         mutex_init(&clt->paths_ev_mutex);
2658         mutex_init(&clt->paths_mutex);
2659
2660         clt->dev.class = rtrs_clt_dev_class;
2661         clt->dev.release = rtrs_clt_dev_release;
2662         err = dev_set_name(&clt->dev, "%s", sessname);
2663         if (err)
2664                 goto err;
2665         /*
2666          * Suppress user space notification until
2667          * sysfs files are created
2668          */
2669         dev_set_uevent_suppress(&clt->dev, true);
2670         err = device_register(&clt->dev);
2671         if (err) {
2672                 put_device(&clt->dev);
2673                 goto err;
2674         }
2675
2676         clt->kobj_paths = kobject_create_and_add("paths", &clt->dev.kobj);
2677         if (!clt->kobj_paths) {
2678                 err = -ENOMEM;
2679                 goto err_dev;
2680         }
2681         err = rtrs_clt_create_sysfs_root_files(clt);
2682         if (err) {
2683                 kobject_del(clt->kobj_paths);
2684                 kobject_put(clt->kobj_paths);
2685                 goto err_dev;
2686         }
2687         dev_set_uevent_suppress(&clt->dev, false);
2688         kobject_uevent(&clt->dev.kobj, KOBJ_ADD);
2689
2690         return clt;
2691 err_dev:
2692         device_unregister(&clt->dev);
2693 err:
2694         free_percpu(clt->pcpu_path);
2695         kfree(clt);
2696         return ERR_PTR(err);
2697 }
2698
2699 static void free_clt(struct rtrs_clt *clt)
2700 {
2701         free_permits(clt);
2702         free_percpu(clt->pcpu_path);
2703         mutex_destroy(&clt->paths_ev_mutex);
2704         mutex_destroy(&clt->paths_mutex);
2705         /* release callback will free clt in last put */
2706         device_unregister(&clt->dev);
2707 }
2708
2709 /**
2710  * rtrs_clt_open() - Open a session to an RTRS server
2711  * @ops: holds the link event callback and the private pointer.
2712  * @sessname: name of the session
2713  * @paths: Paths to be established defined by their src and dst addresses
2714  * @paths_num: Number of elements in the @paths array
2715  * @port: port to be used by the RTRS session
2716  * @pdu_sz: Size of extra payload which can be accessed after permit allocation.
2717  * @reconnect_delay_sec: time between reconnect tries
2718  * @max_segments: Max. number of segments per IO request
2719  * @max_reconnect_attempts: Number of times to reconnect on error before giving
2720  *                          up, 0 for * disabled, -1 for forever
2721  * @nr_poll_queues: number of polling mode connection using IB_POLL_DIRECT flag
2722  *
2723  * Starts session establishment with the rtrs_server. The function can block
2724  * up to ~2000ms before it returns.
2725  *
2726  * Return a valid pointer on success otherwise PTR_ERR.
2727  */
2728 struct rtrs_clt *rtrs_clt_open(struct rtrs_clt_ops *ops,
2729                                  const char *sessname,
2730                                  const struct rtrs_addr *paths,
2731                                  size_t paths_num, u16 port,
2732                                  size_t pdu_sz, u8 reconnect_delay_sec,
2733                                  u16 max_segments,
2734                                  s16 max_reconnect_attempts, u32 nr_poll_queues)
2735 {
2736         struct rtrs_clt_sess *sess, *tmp;
2737         struct rtrs_clt *clt;
2738         int err, i;
2739
2740         clt = alloc_clt(sessname, paths_num, port, pdu_sz, ops->priv,
2741                         ops->link_ev,
2742                         max_segments, reconnect_delay_sec,
2743                         max_reconnect_attempts);
2744         if (IS_ERR(clt)) {
2745                 err = PTR_ERR(clt);
2746                 goto out;
2747         }
2748         for (i = 0; i < paths_num; i++) {
2749                 struct rtrs_clt_sess *sess;
2750
2751                 sess = alloc_sess(clt, &paths[i], nr_cpu_ids,
2752                                   max_segments, nr_poll_queues);
2753                 if (IS_ERR(sess)) {
2754                         err = PTR_ERR(sess);
2755                         goto close_all_sess;
2756                 }
2757                 if (!i)
2758                         sess->for_new_clt = 1;
2759                 list_add_tail_rcu(&sess->s.entry, &clt->paths_list);
2760
2761                 err = init_sess(sess);
2762                 if (err) {
2763                         list_del_rcu(&sess->s.entry);
2764                         rtrs_clt_close_conns(sess, true);
2765                         free_sess(sess);
2766                         goto close_all_sess;
2767                 }
2768
2769                 err = rtrs_clt_create_sess_files(sess);
2770                 if (err) {
2771                         list_del_rcu(&sess->s.entry);
2772                         rtrs_clt_close_conns(sess, true);
2773                         free_sess(sess);
2774                         goto close_all_sess;
2775                 }
2776         }
2777         err = alloc_permits(clt);
2778         if (err)
2779                 goto close_all_sess;
2780
2781         return clt;
2782
2783 close_all_sess:
2784         list_for_each_entry_safe(sess, tmp, &clt->paths_list, s.entry) {
2785                 rtrs_clt_destroy_sess_files(sess, NULL);
2786                 rtrs_clt_close_conns(sess, true);
2787                 kobject_put(&sess->kobj);
2788         }
2789         rtrs_clt_destroy_sysfs_root(clt);
2790         free_clt(clt);
2791
2792 out:
2793         return ERR_PTR(err);
2794 }
2795 EXPORT_SYMBOL(rtrs_clt_open);
2796
2797 /**
2798  * rtrs_clt_close() - Close a session
2799  * @clt: Session handle. Session is freed upon return.
2800  */
2801 void rtrs_clt_close(struct rtrs_clt *clt)
2802 {
2803         struct rtrs_clt_sess *sess, *tmp;
2804
2805         /* Firstly forbid sysfs access */
2806         rtrs_clt_destroy_sysfs_root(clt);
2807
2808         /* Now it is safe to iterate over all paths without locks */
2809         list_for_each_entry_safe(sess, tmp, &clt->paths_list, s.entry) {
2810                 rtrs_clt_close_conns(sess, true);
2811                 rtrs_clt_destroy_sess_files(sess, NULL);
2812                 kobject_put(&sess->kobj);
2813         }
2814         free_clt(clt);
2815 }
2816 EXPORT_SYMBOL(rtrs_clt_close);
2817
2818 int rtrs_clt_reconnect_from_sysfs(struct rtrs_clt_sess *sess)
2819 {
2820         enum rtrs_clt_state old_state;
2821         int err = -EBUSY;
2822         bool changed;
2823
2824         changed = rtrs_clt_change_state_get_old(sess, RTRS_CLT_RECONNECTING,
2825                                                  &old_state);
2826         if (changed) {
2827                 sess->reconnect_attempts = 0;
2828                 queue_delayed_work(rtrs_wq, &sess->reconnect_dwork, 0);
2829         }
2830         if (changed || old_state == RTRS_CLT_RECONNECTING) {
2831                 /*
2832                  * flush_delayed_work() queues pending work for immediate
2833                  * execution, so do the flush if we have queued something
2834                  * right now or work is pending.
2835                  */
2836                 flush_delayed_work(&sess->reconnect_dwork);
2837                 err = (READ_ONCE(sess->state) ==
2838                        RTRS_CLT_CONNECTED ? 0 : -ENOTCONN);
2839         }
2840
2841         return err;
2842 }
2843
2844 int rtrs_clt_disconnect_from_sysfs(struct rtrs_clt_sess *sess)
2845 {
2846         rtrs_clt_close_conns(sess, true);
2847
2848         return 0;
2849 }
2850
2851 int rtrs_clt_remove_path_from_sysfs(struct rtrs_clt_sess *sess,
2852                                      const struct attribute *sysfs_self)
2853 {
2854         enum rtrs_clt_state old_state;
2855         bool changed;
2856
2857         /*
2858          * Continue stopping path till state was changed to DEAD or
2859          * state was observed as DEAD:
2860          * 1. State was changed to DEAD - we were fast and nobody
2861          *    invoked rtrs_clt_reconnect(), which can again start
2862          *    reconnecting.
2863          * 2. State was observed as DEAD - we have someone in parallel
2864          *    removing the path.
2865          */
2866         do {
2867                 rtrs_clt_close_conns(sess, true);
2868                 changed = rtrs_clt_change_state_get_old(sess,
2869                                                         RTRS_CLT_DEAD,
2870                                                         &old_state);
2871         } while (!changed && old_state != RTRS_CLT_DEAD);
2872
2873         if (likely(changed)) {
2874                 rtrs_clt_remove_path_from_arr(sess);
2875                 rtrs_clt_destroy_sess_files(sess, sysfs_self);
2876                 kobject_put(&sess->kobj);
2877         }
2878
2879         return 0;
2880 }
2881
2882 void rtrs_clt_set_max_reconnect_attempts(struct rtrs_clt *clt, int value)
2883 {
2884         clt->max_reconnect_attempts = (unsigned int)value;
2885 }
2886
2887 int rtrs_clt_get_max_reconnect_attempts(const struct rtrs_clt *clt)
2888 {
2889         return (int)clt->max_reconnect_attempts;
2890 }
2891
2892 /**
2893  * rtrs_clt_request() - Request data transfer to/from server via RDMA.
2894  *
2895  * @dir:        READ/WRITE
2896  * @ops:        callback function to be called as confirmation, and the pointer.
2897  * @clt:        Session
2898  * @permit:     Preallocated permit
2899  * @vec:        Message that is sent to server together with the request.
2900  *              Sum of len of all @vec elements limited to <= IO_MSG_SIZE.
2901  *              Since the msg is copied internally it can be allocated on stack.
2902  * @nr:         Number of elements in @vec.
2903  * @data_len:   length of data sent to/from server
2904  * @sg:         Pages to be sent/received to/from server.
2905  * @sg_cnt:     Number of elements in the @sg
2906  *
2907  * Return:
2908  * 0:           Success
2909  * <0:          Error
2910  *
2911  * On dir=READ rtrs client will request a data transfer from Server to client.
2912  * The data that the server will respond with will be stored in @sg when
2913  * the user receives an %RTRS_CLT_RDMA_EV_RDMA_REQUEST_WRITE_COMPL event.
2914  * On dir=WRITE rtrs client will rdma write data in sg to server side.
2915  */
2916 int rtrs_clt_request(int dir, struct rtrs_clt_req_ops *ops,
2917                      struct rtrs_clt *clt, struct rtrs_permit *permit,
2918                       const struct kvec *vec, size_t nr, size_t data_len,
2919                       struct scatterlist *sg, unsigned int sg_cnt)
2920 {
2921         struct rtrs_clt_io_req *req;
2922         struct rtrs_clt_sess *sess;
2923
2924         enum dma_data_direction dma_dir;
2925         int err = -ECONNABORTED, i;
2926         size_t usr_len, hdr_len;
2927         struct path_it it;
2928
2929         /* Get kvec length */
2930         for (i = 0, usr_len = 0; i < nr; i++)
2931                 usr_len += vec[i].iov_len;
2932
2933         if (dir == READ) {
2934                 hdr_len = sizeof(struct rtrs_msg_rdma_read) +
2935                           sg_cnt * sizeof(struct rtrs_sg_desc);
2936                 dma_dir = DMA_FROM_DEVICE;
2937         } else {
2938                 hdr_len = sizeof(struct rtrs_msg_rdma_write);
2939                 dma_dir = DMA_TO_DEVICE;
2940         }
2941
2942         rcu_read_lock();
2943         for (path_it_init(&it, clt);
2944              (sess = it.next_path(&it)) && it.i < it.clt->paths_num; it.i++) {
2945                 if (unlikely(READ_ONCE(sess->state) != RTRS_CLT_CONNECTED))
2946                         continue;
2947
2948                 if (unlikely(usr_len + hdr_len > sess->max_hdr_size)) {
2949                         rtrs_wrn_rl(sess->clt,
2950                                      "%s request failed, user message size is %zu and header length %zu, but max size is %u\n",
2951                                      dir == READ ? "Read" : "Write",
2952                                      usr_len, hdr_len, sess->max_hdr_size);
2953                         err = -EMSGSIZE;
2954                         break;
2955                 }
2956                 req = rtrs_clt_get_req(sess, ops->conf_fn, permit, ops->priv,
2957                                        vec, usr_len, sg, sg_cnt, data_len,
2958                                        dma_dir);
2959                 if (dir == READ)
2960                         err = rtrs_clt_read_req(req);
2961                 else
2962                         err = rtrs_clt_write_req(req);
2963                 if (unlikely(err)) {
2964                         req->in_use = false;
2965                         continue;
2966                 }
2967                 /* Success path */
2968                 break;
2969         }
2970         path_it_deinit(&it);
2971         rcu_read_unlock();
2972
2973         return err;
2974 }
2975 EXPORT_SYMBOL(rtrs_clt_request);
2976
2977 int rtrs_clt_rdma_cq_direct(struct rtrs_clt *clt, unsigned int index)
2978 {
2979         int cnt;
2980         struct rtrs_con *con;
2981         struct rtrs_clt_sess *sess;
2982         struct path_it it;
2983
2984         rcu_read_lock();
2985         for (path_it_init(&it, clt);
2986              (sess = it.next_path(&it)) && it.i < it.clt->paths_num; it.i++) {
2987                 if (READ_ONCE(sess->state) != RTRS_CLT_CONNECTED)
2988                         continue;
2989
2990                 con = sess->s.con[index + 1];
2991                 cnt = ib_process_cq_direct(con->cq, -1);
2992                 if (cnt)
2993                         break;
2994         }
2995         path_it_deinit(&it);
2996         rcu_read_unlock();
2997
2998         return cnt;
2999 }
3000 EXPORT_SYMBOL(rtrs_clt_rdma_cq_direct);
3001
3002 /**
3003  * rtrs_clt_query() - queries RTRS session attributes
3004  *@clt: session pointer
3005  *@attr: query results for session attributes.
3006  * Returns:
3007  *    0 on success
3008  *    -ECOMM            no connection to the server
3009  */
3010 int rtrs_clt_query(struct rtrs_clt *clt, struct rtrs_attrs *attr)
3011 {
3012         if (!rtrs_clt_is_connected(clt))
3013                 return -ECOMM;
3014
3015         attr->queue_depth      = clt->queue_depth;
3016         /* Cap max_io_size to min of remote buffer size and the fr pages */
3017         attr->max_io_size = min_t(int, clt->max_io_size,
3018                                   clt->max_segments * SZ_4K);
3019
3020         return 0;
3021 }
3022 EXPORT_SYMBOL(rtrs_clt_query);
3023
3024 int rtrs_clt_create_path_from_sysfs(struct rtrs_clt *clt,
3025                                      struct rtrs_addr *addr)
3026 {
3027         struct rtrs_clt_sess *sess;
3028         int err;
3029
3030         sess = alloc_sess(clt, addr, nr_cpu_ids, clt->max_segments, 0);
3031         if (IS_ERR(sess))
3032                 return PTR_ERR(sess);
3033
3034         /*
3035          * It is totally safe to add path in CONNECTING state: coming
3036          * IO will never grab it.  Also it is very important to add
3037          * path before init, since init fires LINK_CONNECTED event.
3038          */
3039         rtrs_clt_add_path_to_arr(sess);
3040
3041         err = init_sess(sess);
3042         if (err)
3043                 goto close_sess;
3044
3045         err = rtrs_clt_create_sess_files(sess);
3046         if (err)
3047                 goto close_sess;
3048
3049         return 0;
3050
3051 close_sess:
3052         rtrs_clt_remove_path_from_arr(sess);
3053         rtrs_clt_close_conns(sess, true);
3054         free_sess(sess);
3055
3056         return err;
3057 }
3058
3059 static int rtrs_clt_ib_dev_init(struct rtrs_ib_dev *dev)
3060 {
3061         if (!(dev->ib_dev->attrs.device_cap_flags &
3062               IB_DEVICE_MEM_MGT_EXTENSIONS)) {
3063                 pr_err("Memory registrations not supported.\n");
3064                 return -ENOTSUPP;
3065         }
3066
3067         return 0;
3068 }
3069
3070 static const struct rtrs_rdma_dev_pd_ops dev_pd_ops = {
3071         .init = rtrs_clt_ib_dev_init
3072 };
3073
3074 static int __init rtrs_client_init(void)
3075 {
3076         rtrs_rdma_dev_pd_init(0, &dev_pd);
3077
3078         rtrs_clt_dev_class = class_create(THIS_MODULE, "rtrs-client");
3079         if (IS_ERR(rtrs_clt_dev_class)) {
3080                 pr_err("Failed to create rtrs-client dev class\n");
3081                 return PTR_ERR(rtrs_clt_dev_class);
3082         }
3083         rtrs_wq = alloc_workqueue("rtrs_client_wq", 0, 0);
3084         if (!rtrs_wq) {
3085                 class_destroy(rtrs_clt_dev_class);
3086                 return -ENOMEM;
3087         }
3088
3089         return 0;
3090 }
3091
3092 static void __exit rtrs_client_exit(void)
3093 {
3094         destroy_workqueue(rtrs_wq);
3095         class_destroy(rtrs_clt_dev_class);
3096         rtrs_rdma_dev_pd_deinit(&dev_pd);
3097 }
3098
3099 module_init(rtrs_client_init);
3100 module_exit(rtrs_client_exit);