drivers/infiniband/ulp/rtrs/rtrs-srv.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/mempool.h>
  15
  16 #include "rtrs-srv.h"
  17 #include "rtrs-log.h"
  18
  19 MODULE_DESCRIPTION("RDMA Transport Server");
  20 MODULE_LICENSE("GPL");
  21
  22 /* Must be power of 2, see mask from mr->page_size in ib_sg_to_pages() */
  23 #define DEFAULT_MAX_CHUNK_SIZE (128 << 10)
  24 #define DEFAULT_SESS_QUEUE_DEPTH 512
  25 #define MAX_HDR_SIZE PAGE_SIZE
  26
  27 /* We guarantee to serve 10 paths at least */
  28 #define CHUNK_POOL_SZ 10
  29
  30 static struct rtrs_rdma_dev_pd dev_pd;
  31 static mempool_t *chunk_pool;
  32 struct class *rtrs_dev_class;
  33
  34 static int __read_mostly max_chunk_size = DEFAULT_MAX_CHUNK_SIZE;
  35 static int __read_mostly sess_queue_depth = DEFAULT_SESS_QUEUE_DEPTH;
  36
  37 static bool always_invalidate = true;
  38 module_param(always_invalidate, bool, 0444);
  39 MODULE_PARM_DESC(always_invalidate,
  40                  "Invalidate memory registration for contiguous memory regions before accessing.");
  41
  42 module_param_named(max_chunk_size, max_chunk_size, int, 0444);
  43 MODULE_PARM_DESC(max_chunk_size,
  44                  "Max size for each IO request, when change the unit is in byte (default: "
  45                  __stringify(DEFAULT_MAX_CHUNK_SIZE) "KB)");
  46
  47 module_param_named(sess_queue_depth, sess_queue_depth, int, 0444);
  48 MODULE_PARM_DESC(sess_queue_depth,
  49                  "Number of buffers for pending I/O requests to allocate per session. Maximum: "
  50                  __stringify(MAX_SESS_QUEUE_DEPTH) " (default: "
  51                  __stringify(DEFAULT_SESS_QUEUE_DEPTH) ")");
  52
  53 static cpumask_t cq_affinity_mask = { CPU_BITS_ALL };
  54
  55 static struct workqueue_struct *rtrs_wq;
  56
  57 static inline struct rtrs_srv_con *to_srv_con(struct rtrs_con *c)
  58 {
  59         return container_of(c, struct rtrs_srv_con, c);
  60 }
  61
  62 static inline struct rtrs_srv_sess *to_srv_sess(struct rtrs_sess *s)
  63 {
  64         return container_of(s, struct rtrs_srv_sess, s);
  65 }
  66
  67 static bool __rtrs_srv_change_state(struct rtrs_srv_sess *sess,
  68                                      enum rtrs_srv_state new_state)
  69 {
  70         enum rtrs_srv_state old_state;
  71         bool changed = false;
  72
  73         lockdep_assert_held(&sess->state_lock);
  74         old_state = sess->state;
  75         switch (new_state) {
  76         case RTRS_SRV_CONNECTED:
  77                 switch (old_state) {
  78                 case RTRS_SRV_CONNECTING:
  79                         changed = true;
  80                         fallthrough;
  81                 default:
  82                         break;
  83                 }
  84                 break;
  85         case RTRS_SRV_CLOSING:
  86                 switch (old_state) {
  87                 case RTRS_SRV_CONNECTING:
  88                 case RTRS_SRV_CONNECTED:
  89                         changed = true;
  90                         fallthrough;
  91                 default:
  92                         break;
  93                 }
  94                 break;
  95         case RTRS_SRV_CLOSED:
  96                 switch (old_state) {
  97                 case RTRS_SRV_CLOSING:
  98                         changed = true;
  99                         fallthrough;
 100                 default:
 101                         break;
 102                 }
 103                 break;
 104         default:
 105                 break;
 106         }
 107         if (changed)
 108                 sess->state = new_state;
 109
 110         return changed;
 111 }
 112
 113 static bool rtrs_srv_change_state_get_old(struct rtrs_srv_sess *sess,
 114                                            enum rtrs_srv_state new_state,
 115                                            enum rtrs_srv_state *old_state)
 116 {
 117         bool changed;
 118
 119         spin_lock_irq(&sess->state_lock);
 120         *old_state = sess->state;
 121         changed = __rtrs_srv_change_state(sess, new_state);
 122         spin_unlock_irq(&sess->state_lock);
 123
 124         return changed;
 125 }
 126
 127 static bool rtrs_srv_change_state(struct rtrs_srv_sess *sess,
 128                                    enum rtrs_srv_state new_state)
 129 {
 130         enum rtrs_srv_state old_state;
 131
 132         return rtrs_srv_change_state_get_old(sess, new_state, &old_state);
 133 }
 134
 135 static void free_id(struct rtrs_srv_op *id)
 136 {
 137         if (!id)
 138                 return;
 139         kfree(id);
 140 }
 141
 142 static void rtrs_srv_free_ops_ids(struct rtrs_srv_sess *sess)
 143 {
 144         struct rtrs_srv *srv = sess->srv;
 145         int i;
 146
 147         WARN_ON(atomic_read(&sess->ids_inflight));
 148         if (sess->ops_ids) {
 149                 for (i = 0; i < srv->queue_depth; i++)
 150                         free_id(sess->ops_ids[i]);
 151                 kfree(sess->ops_ids);
 152                 sess->ops_ids = NULL;
 153         }
 154 }
 155
 156 static void rtrs_srv_rdma_done(struct ib_cq *cq, struct ib_wc *wc);
 157
 158 static struct ib_cqe io_comp_cqe = {
 159         .done = rtrs_srv_rdma_done
 160 };
 161
 162 static int rtrs_srv_alloc_ops_ids(struct rtrs_srv_sess *sess)
 163 {
 164         struct rtrs_srv *srv = sess->srv;
 165         struct rtrs_srv_op *id;
 166         int i;
 167
 168         sess->ops_ids = kcalloc(srv->queue_depth, sizeof(*sess->ops_ids),
 169                                 GFP_KERNEL);
 170         if (!sess->ops_ids)
 171                 goto err;
 172
 173         for (i = 0; i < srv->queue_depth; ++i) {
 174                 id = kzalloc(sizeof(*id), GFP_KERNEL);
 175                 if (!id)
 176                         goto err;
 177
 178                 sess->ops_ids[i] = id;
 179         }
 180         init_waitqueue_head(&sess->ids_waitq);
 181         atomic_set(&sess->ids_inflight, 0);
 182
 183         return 0;
 184
 185 err:
 186         rtrs_srv_free_ops_ids(sess);
 187         return -ENOMEM;
 188 }
 189
 190 static inline void rtrs_srv_get_ops_ids(struct rtrs_srv_sess *sess)
 191 {
 192         atomic_inc(&sess->ids_inflight);
 193 }
 194
 195 static inline void rtrs_srv_put_ops_ids(struct rtrs_srv_sess *sess)
 196 {
 197         if (atomic_dec_and_test(&sess->ids_inflight))
 198                 wake_up(&sess->ids_waitq);
 199 }
 200
 201 static void rtrs_srv_wait_ops_ids(struct rtrs_srv_sess *sess)
 202 {
 203         wait_event(sess->ids_waitq, !atomic_read(&sess->ids_inflight));
 204 }
 205
 206
 207 static void rtrs_srv_reg_mr_done(struct ib_cq *cq, struct ib_wc *wc)
 208 {
 209         struct rtrs_srv_con *con = cq->cq_context;
 210         struct rtrs_sess *s = con->c.sess;
 211         struct rtrs_srv_sess *sess = to_srv_sess(s);
 212
 213         if (unlikely(wc->status != IB_WC_SUCCESS)) {
 214                 rtrs_err(s, "REG MR failed: %s\n",
 215                           ib_wc_status_msg(wc->status));
 216                 close_sess(sess);
 217                 return;
 218         }
 219 }
 220
 221 static struct ib_cqe local_reg_cqe = {
 222         .done = rtrs_srv_reg_mr_done
 223 };
 224
 225 static int rdma_write_sg(struct rtrs_srv_op *id)
 226 {
 227         struct rtrs_sess *s = id->con->c.sess;
 228         struct rtrs_srv_sess *sess = to_srv_sess(s);
 229         dma_addr_t dma_addr = sess->dma_addr[id->msg_id];
 230         struct rtrs_srv_mr *srv_mr;
 231         struct rtrs_srv *srv = sess->srv;
 232         struct ib_send_wr inv_wr, imm_wr;
 233         struct ib_rdma_wr *wr = NULL;
 234         enum ib_send_flags flags;
 235         size_t sg_cnt;
 236         int err, offset;
 237         bool need_inval;
 238         u32 rkey = 0;
 239         struct ib_reg_wr rwr;
 240         struct ib_sge *plist;
 241         struct ib_sge list;
 242
 243         sg_cnt = le16_to_cpu(id->rd_msg->sg_cnt);
 244         need_inval = le16_to_cpu(id->rd_msg->flags) & RTRS_MSG_NEED_INVAL_F;
 245         if (unlikely(sg_cnt != 1))
 246                 return -EINVAL;
 247
 248         offset = 0;
 249
 250         wr              = &id->tx_wr;
 251         plist           = &id->tx_sg;
 252         plist->addr     = dma_addr + offset;
 253         plist->length   = le32_to_cpu(id->rd_msg->desc[0].len);
 254
 255         /* WR will fail with length error
 256          * if this is 0
 257          */
 258         if (unlikely(plist->length == 0)) {
 259                 rtrs_err(s, "Invalid RDMA-Write sg list length 0\n");
 260                 return -EINVAL;
 261         }
 262
 263         plist->lkey = sess->s.dev->ib_pd->local_dma_lkey;
 264         offset += plist->length;
 265
 266         wr->wr.sg_list  = plist;
 267         wr->wr.num_sge  = 1;
 268         wr->remote_addr = le64_to_cpu(id->rd_msg->desc[0].addr);
 269         wr->rkey        = le32_to_cpu(id->rd_msg->desc[0].key);
 270         if (rkey == 0)
 271                 rkey = wr->rkey;
 272         else
 273                 /* Only one key is actually used */
 274                 WARN_ON_ONCE(rkey != wr->rkey);
 275
 276         wr->wr.opcode = IB_WR_RDMA_WRITE;
 277         wr->wr.ex.imm_data = 0;
 278         wr->wr.send_flags  = 0;
 279
 280         if (need_inval && always_invalidate) {
 281                 wr->wr.next = &rwr.wr;
 282                 rwr.wr.next = &inv_wr;
 283                 inv_wr.next = &imm_wr;
 284         } else if (always_invalidate) {
 285                 wr->wr.next = &rwr.wr;
 286                 rwr.wr.next = &imm_wr;
 287         } else if (need_inval) {
 288                 wr->wr.next = &inv_wr;
 289                 inv_wr.next = &imm_wr;
 290         } else {
 291                 wr->wr.next = &imm_wr;
 292         }
 293         /*
 294          * From time to time we have to post signaled sends,
 295          * or send queue will fill up and only QP reset can help.
 296          */
 297         flags = (atomic_inc_return(&id->con->wr_cnt) % srv->queue_depth) ?
 298                 0 : IB_SEND_SIGNALED;
 299
 300         if (need_inval) {
 301                 inv_wr.sg_list = NULL;
 302                 inv_wr.num_sge = 0;
 303                 inv_wr.opcode = IB_WR_SEND_WITH_INV;
 304                 inv_wr.send_flags = 0;
 305                 inv_wr.ex.invalidate_rkey = rkey;
 306         }
 307
 308         imm_wr.next = NULL;
 309         if (always_invalidate) {
 310                 struct rtrs_msg_rkey_rsp *msg;
 311
 312                 srv_mr = &sess->mrs[id->msg_id];
 313                 rwr.wr.opcode = IB_WR_REG_MR;
 314                 rwr.wr.num_sge = 0;
 315                 rwr.mr = srv_mr->mr;
 316                 rwr.wr.send_flags = 0;
 317                 rwr.key = srv_mr->mr->rkey;
 318                 rwr.access = (IB_ACCESS_LOCAL_WRITE |
 319                               IB_ACCESS_REMOTE_WRITE);
 320                 msg = srv_mr->iu->buf;
 321                 msg->buf_id = cpu_to_le16(id->msg_id);
 322                 msg->type = cpu_to_le16(RTRS_MSG_RKEY_RSP);
 323                 msg->rkey = cpu_to_le32(srv_mr->mr->rkey);
 324
 325                 list.addr   = srv_mr->iu->dma_addr;
 326                 list.length = sizeof(*msg);
 327                 list.lkey   = sess->s.dev->ib_pd->local_dma_lkey;
 328                 imm_wr.sg_list = &list;
 329                 imm_wr.num_sge = 1;
 330                 imm_wr.opcode = IB_WR_SEND_WITH_IMM;
 331                 ib_dma_sync_single_for_device(sess->s.dev->ib_dev,
 332                                               srv_mr->iu->dma_addr,
 333                                               srv_mr->iu->size, DMA_TO_DEVICE);
 334         } else {
 335                 imm_wr.sg_list = NULL;
 336                 imm_wr.num_sge = 0;
 337                 imm_wr.opcode = IB_WR_RDMA_WRITE_WITH_IMM;
 338         }
 339         imm_wr.send_flags = flags;
 340         imm_wr.ex.imm_data = cpu_to_be32(rtrs_to_io_rsp_imm(id->msg_id,
 341                                                              0, need_inval));
 342
 343         imm_wr.wr_cqe   = &io_comp_cqe;
 344         ib_dma_sync_single_for_device(sess->s.dev->ib_dev, dma_addr,
 345                                       offset, DMA_BIDIRECTIONAL);
 346
 347         err = ib_post_send(id->con->c.qp, &id->tx_wr.wr, NULL);
 348         if (unlikely(err))
 349                 rtrs_err(s,
 350                           "Posting RDMA-Write-Request to QP failed, err: %d\n",
 351                           err);
 352
 353         return err;
 354 }
 355
 356 /**
 357  * send_io_resp_imm() - respond to client with empty IMM on failed READ/WRITE
 358  *                      requests or on successful WRITE request.
 359  * @con:        the connection to send back result
 360  * @id:         the id associated with the IO
 361  * @errno:      the error number of the IO.
 362  *
 363  * Return 0 on success, errno otherwise.
 364  */
 365 static int send_io_resp_imm(struct rtrs_srv_con *con, struct rtrs_srv_op *id,
 366                             int errno)
 367 {
 368         struct rtrs_sess *s = con->c.sess;
 369         struct rtrs_srv_sess *sess = to_srv_sess(s);
 370         struct ib_send_wr inv_wr, imm_wr, *wr = NULL;
 371         struct ib_reg_wr rwr;
 372         struct rtrs_srv *srv = sess->srv;
 373         struct rtrs_srv_mr *srv_mr;
 374         bool need_inval = false;
 375         enum ib_send_flags flags;
 376         u32 imm;
 377         int err;
 378
 379         if (id->dir == READ) {
 380                 struct rtrs_msg_rdma_read *rd_msg = id->rd_msg;
 381                 size_t sg_cnt;
 382
 383                 need_inval = le16_to_cpu(rd_msg->flags) &
 384                                 RTRS_MSG_NEED_INVAL_F;
 385                 sg_cnt = le16_to_cpu(rd_msg->sg_cnt);
 386
 387                 if (need_inval) {
 388                         if (likely(sg_cnt)) {
 389                                 inv_wr.sg_list = NULL;
 390                                 inv_wr.num_sge = 0;
 391                                 inv_wr.opcode = IB_WR_SEND_WITH_INV;
 392                                 inv_wr.send_flags = 0;
 393                                 /* Only one key is actually used */
 394                                 inv_wr.ex.invalidate_rkey =
 395                                         le32_to_cpu(rd_msg->desc[0].key);
 396                         } else {
 397                                 WARN_ON_ONCE(1);
 398                                 need_inval = false;
 399                         }
 400                 }
 401         }
 402
 403         if (need_inval && always_invalidate) {
 404                 wr = &inv_wr;
 405                 inv_wr.next = &rwr.wr;
 406                 rwr.wr.next = &imm_wr;
 407         } else if (always_invalidate) {
 408                 wr = &rwr.wr;
 409                 rwr.wr.next = &imm_wr;
 410         } else if (need_inval) {
 411                 wr = &inv_wr;
 412                 inv_wr.next = &imm_wr;
 413         } else {
 414                 wr = &imm_wr;
 415         }
 416         /*
 417          * From time to time we have to post signalled sends,
 418          * or send queue will fill up and only QP reset can help.
 419          */
 420         flags = (atomic_inc_return(&con->wr_cnt) % srv->queue_depth) ?
 421                 0 : IB_SEND_SIGNALED;
 422         imm = rtrs_to_io_rsp_imm(id->msg_id, errno, need_inval);
 423         imm_wr.next = NULL;
 424         if (always_invalidate) {
 425                 struct ib_sge list;
 426                 struct rtrs_msg_rkey_rsp *msg;
 427
 428                 srv_mr = &sess->mrs[id->msg_id];
 429                 rwr.wr.next = &imm_wr;
 430                 rwr.wr.opcode = IB_WR_REG_MR;
 431                 rwr.wr.num_sge = 0;
 432                 rwr.wr.send_flags = 0;
 433                 rwr.mr = srv_mr->mr;
 434                 rwr.key = srv_mr->mr->rkey;
 435                 rwr.access = (IB_ACCESS_LOCAL_WRITE |
 436                               IB_ACCESS_REMOTE_WRITE);
 437                 msg = srv_mr->iu->buf;
 438                 msg->buf_id = cpu_to_le16(id->msg_id);
 439                 msg->type = cpu_to_le16(RTRS_MSG_RKEY_RSP);
 440                 msg->rkey = cpu_to_le32(srv_mr->mr->rkey);
 441
 442                 list.addr   = srv_mr->iu->dma_addr;
 443                 list.length = sizeof(*msg);
 444                 list.lkey   = sess->s.dev->ib_pd->local_dma_lkey;
 445                 imm_wr.sg_list = &list;
 446                 imm_wr.num_sge = 1;
 447                 imm_wr.opcode = IB_WR_SEND_WITH_IMM;
 448                 ib_dma_sync_single_for_device(sess->s.dev->ib_dev,
 449                                               srv_mr->iu->dma_addr,
 450                                               srv_mr->iu->size, DMA_TO_DEVICE);
 451         } else {
 452                 imm_wr.sg_list = NULL;
 453                 imm_wr.num_sge = 0;
 454                 imm_wr.opcode = IB_WR_RDMA_WRITE_WITH_IMM;
 455         }
 456         imm_wr.send_flags = flags;
 457         imm_wr.wr_cqe   = &io_comp_cqe;
 458
 459         imm_wr.ex.imm_data = cpu_to_be32(imm);
 460
 461         err = ib_post_send(id->con->c.qp, wr, NULL);
 462         if (unlikely(err))
 463                 rtrs_err_rl(s, "Posting RDMA-Reply to QP failed, err: %d\n",
 464                              err);
 465
 466         return err;
 467 }
 468
 469 void close_sess(struct rtrs_srv_sess *sess)
 470 {
 471         enum rtrs_srv_state old_state;
 472
 473         if (rtrs_srv_change_state_get_old(sess, RTRS_SRV_CLOSING,
 474                                            &old_state))
 475                 queue_work(rtrs_wq, &sess->close_work);
 476         WARN_ON(sess->state != RTRS_SRV_CLOSING);
 477 }
 478
 479 static inline const char *rtrs_srv_state_str(enum rtrs_srv_state state)
 480 {
 481         switch (state) {
 482         case RTRS_SRV_CONNECTING:
 483                 return "RTRS_SRV_CONNECTING";
 484         case RTRS_SRV_CONNECTED:
 485                 return "RTRS_SRV_CONNECTED";
 486         case RTRS_SRV_CLOSING:
 487                 return "RTRS_SRV_CLOSING";
 488         case RTRS_SRV_CLOSED:
 489                 return "RTRS_SRV_CLOSED";
 490         default:
 491                 return "UNKNOWN";
 492         }
 493 }
 494
 495 /**
 496  * rtrs_srv_resp_rdma() - Finish an RDMA request
 497  *
 498  * @id:         Internal RTRS operation identifier
 499  * @status:     Response Code sent to the other side for this operation.
 500  *              0 = success, <=0 error
 501  * Context: any
 502  *
 503  * Finish a RDMA operation. A message is sent to the client and the
 504  * corresponding memory areas will be released.
 505  */
 506 bool rtrs_srv_resp_rdma(struct rtrs_srv_op *id, int status)
 507 {
 508         struct rtrs_srv_sess *sess;
 509         struct rtrs_srv_con *con;
 510         struct rtrs_sess *s;
 511         int err;
 512
 513         if (WARN_ON(!id))
 514                 return true;
 515
 516         con = id->con;
 517         s = con->c.sess;
 518         sess = to_srv_sess(s);
 519
 520         id->status = status;
 521
 522         if (unlikely(sess->state != RTRS_SRV_CONNECTED)) {
 523                 rtrs_err_rl(s,
 524                              "Sending I/O response failed,  session is disconnected, sess state %s\n",
 525                              rtrs_srv_state_str(sess->state));
 526                 goto out;
 527         }
 528         if (always_invalidate) {
 529                 struct rtrs_srv_mr *mr = &sess->mrs[id->msg_id];
 530
 531                 ib_update_fast_reg_key(mr->mr, ib_inc_rkey(mr->mr->rkey));
 532         }
 533         if (unlikely(atomic_sub_return(1,
 534                                        &con->sq_wr_avail) < 0)) {
 535                 pr_err("IB send queue full\n");
 536                 atomic_add(1, &con->sq_wr_avail);
 537                 spin_lock(&con->rsp_wr_wait_lock);
 538                 list_add_tail(&id->wait_list, &con->rsp_wr_wait_list);
 539                 spin_unlock(&con->rsp_wr_wait_lock);
 540                 return false;
 541         }
 542
 543         if (status || id->dir == WRITE || !id->rd_msg->sg_cnt)
 544                 err = send_io_resp_imm(con, id, status);
 545         else
 546                 err = rdma_write_sg(id);
 547
 548         if (unlikely(err)) {
 549                 rtrs_err_rl(s, "IO response failed: %d\n", err);
 550                 close_sess(sess);
 551         }
 552 out:
 553         rtrs_srv_put_ops_ids(sess);
 554         return true;
 555 }
 556 EXPORT_SYMBOL(rtrs_srv_resp_rdma);
 557
 558 /**
 559  * rtrs_srv_set_sess_priv() - Set private pointer in rtrs_srv.
 560  * @srv:        Session pointer
 561  * @priv:       The private pointer that is associated with the session.
 562  */
 563 void rtrs_srv_set_sess_priv(struct rtrs_srv *srv, void *priv)
 564 {
 565         srv->priv = priv;
 566 }
 567 EXPORT_SYMBOL(rtrs_srv_set_sess_priv);
 568
 569 static void unmap_cont_bufs(struct rtrs_srv_sess *sess)
 570 {
 571         int i;
 572
 573         for (i = 0; i < sess->mrs_num; i++) {
 574                 struct rtrs_srv_mr *srv_mr;
 575
 576                 srv_mr = &sess->mrs[i];
 577                 rtrs_iu_free(srv_mr->iu, DMA_TO_DEVICE,
 578                               sess->s.dev->ib_dev, 1);
 579                 ib_dereg_mr(srv_mr->mr);
 580                 ib_dma_unmap_sg(sess->s.dev->ib_dev, srv_mr->sgt.sgl,
 581                                 srv_mr->sgt.nents, DMA_BIDIRECTIONAL);
 582                 sg_free_table(&srv_mr->sgt);
 583         }
 584         kfree(sess->mrs);
 585 }
 586
 587 static int map_cont_bufs(struct rtrs_srv_sess *sess)
 588 {
 589         struct rtrs_srv *srv = sess->srv;
 590         struct rtrs_sess *ss = &sess->s;
 591         int i, mri, err, mrs_num;
 592         unsigned int chunk_bits;
 593         int chunks_per_mr = 1;
 594
 595         /*
 596          * Here we map queue_depth chunks to MR.  Firstly we have to
 597          * figure out how many chunks can we map per MR.
 598          */
 599         if (always_invalidate) {
 600                 /*
 601                  * in order to do invalidate for each chunks of memory, we needs
 602                  * more memory regions.
 603                  */
 604                 mrs_num = srv->queue_depth;
 605         } else {
 606                 chunks_per_mr =
 607                         sess->s.dev->ib_dev->attrs.max_fast_reg_page_list_len;
 608                 mrs_num = DIV_ROUND_UP(srv->queue_depth, chunks_per_mr);
 609                 chunks_per_mr = DIV_ROUND_UP(srv->queue_depth, mrs_num);
 610         }
 611
 612         sess->mrs = kcalloc(mrs_num, sizeof(*sess->mrs), GFP_KERNEL);
 613         if (!sess->mrs)
 614                 return -ENOMEM;
 615
 616         sess->mrs_num = mrs_num;
 617
 618         for (mri = 0; mri < mrs_num; mri++) {
 619                 struct rtrs_srv_mr *srv_mr = &sess->mrs[mri];
 620                 struct sg_table *sgt = &srv_mr->sgt;
 621                 struct scatterlist *s;
 622                 struct ib_mr *mr;
 623                 int nr, chunks;
 624
 625                 chunks = chunks_per_mr * mri;
 626                 if (!always_invalidate)
 627                         chunks_per_mr = min_t(int, chunks_per_mr,
 628                                               srv->queue_depth - chunks);
 629
 630                 err = sg_alloc_table(sgt, chunks_per_mr, GFP_KERNEL);
 631                 if (err)
 632                         goto err;
 633
 634                 for_each_sg(sgt->sgl, s, chunks_per_mr, i)
 635                         sg_set_page(s, srv->chunks[chunks + i],
 636                                     max_chunk_size, 0);
 637
 638                 nr = ib_dma_map_sg(sess->s.dev->ib_dev, sgt->sgl,
 639                                    sgt->nents, DMA_BIDIRECTIONAL);
 640                 if (nr < sgt->nents) {
 641                         err = nr < 0 ? nr : -EINVAL;
 642                         goto free_sg;
 643                 }
 644                 mr = ib_alloc_mr(sess->s.dev->ib_pd, IB_MR_TYPE_MEM_REG,
 645                                  sgt->nents);
 646                 if (IS_ERR(mr)) {
 647                         err = PTR_ERR(mr);
 648                         goto unmap_sg;
 649                 }
 650                 nr = ib_map_mr_sg(mr, sgt->sgl, sgt->nents,
 651                                   NULL, max_chunk_size);
 652                 if (nr < sgt->nents) {
 653                         err = nr < 0 ? nr : -EINVAL;
 654                         goto dereg_mr;
 655                 }
 656
 657                 if (always_invalidate) {
 658                         srv_mr->iu = rtrs_iu_alloc(1,
 659                                         sizeof(struct rtrs_msg_rkey_rsp),
 660                                         GFP_KERNEL, sess->s.dev->ib_dev,
 661                                         DMA_TO_DEVICE, rtrs_srv_rdma_done);
 662                         if (!srv_mr->iu) {
 663                                 rtrs_err(ss, "rtrs_iu_alloc(), err: %d\n",
 664                                           -ENOMEM);
 665                                 goto free_iu;
 666                         }
 667                 }
 668                 /* Eventually dma addr for each chunk can be cached */
 669                 for_each_sg(sgt->sgl, s, sgt->orig_nents, i)
 670                         sess->dma_addr[chunks + i] = sg_dma_address(s);
 671
 672                 ib_update_fast_reg_key(mr, ib_inc_rkey(mr->rkey));
 673                 srv_mr->mr = mr;
 674
 675                 continue;
 676 err:
 677                 while (mri--) {
 678                         srv_mr = &sess->mrs[mri];
 679                         sgt = &srv_mr->sgt;
 680                         mr = srv_mr->mr;
 681 free_iu:
 682                         rtrs_iu_free(srv_mr->iu, DMA_TO_DEVICE,
 683                                       sess->s.dev->ib_dev, 1);
 684 dereg_mr:
 685                         ib_dereg_mr(mr);
 686 unmap_sg:
 687                         ib_dma_unmap_sg(sess->s.dev->ib_dev, sgt->sgl,
 688                                         sgt->nents, DMA_BIDIRECTIONAL);
 689 free_sg:
 690                         sg_free_table(sgt);
 691                 }
 692                 kfree(sess->mrs);
 693
 694                 return err;
 695         }
 696
 697         chunk_bits = ilog2(srv->queue_depth - 1) + 1;
 698         sess->mem_bits = (MAX_IMM_PAYL_BITS - chunk_bits);
 699
 700         return 0;
 701 }
 702
 703 static void rtrs_srv_hb_err_handler(struct rtrs_con *c)
 704 {
 705         close_sess(to_srv_sess(c->sess));
 706 }
 707
 708 static void rtrs_srv_init_hb(struct rtrs_srv_sess *sess)
 709 {
 710         rtrs_init_hb(&sess->s, &io_comp_cqe,
 711                       RTRS_HB_INTERVAL_MS,
 712                       RTRS_HB_MISSED_MAX,
 713                       rtrs_srv_hb_err_handler,
 714                       rtrs_wq);
 715 }
 716
 717 static void rtrs_srv_start_hb(struct rtrs_srv_sess *sess)
 718 {
 719         rtrs_start_hb(&sess->s);
 720 }
 721
 722 static void rtrs_srv_stop_hb(struct rtrs_srv_sess *sess)
 723 {
 724         rtrs_stop_hb(&sess->s);
 725 }
 726
 727 static void rtrs_srv_info_rsp_done(struct ib_cq *cq, struct ib_wc *wc)
 728 {
 729         struct rtrs_srv_con *con = cq->cq_context;
 730         struct rtrs_sess *s = con->c.sess;
 731         struct rtrs_srv_sess *sess = to_srv_sess(s);
 732         struct rtrs_iu *iu;
 733
 734         iu = container_of(wc->wr_cqe, struct rtrs_iu, cqe);
 735         rtrs_iu_free(iu, DMA_TO_DEVICE, sess->s.dev->ib_dev, 1);
 736
 737         if (unlikely(wc->status != IB_WC_SUCCESS)) {
 738                 rtrs_err(s, "Sess info response send failed: %s\n",
 739                           ib_wc_status_msg(wc->status));
 740                 close_sess(sess);
 741                 return;
 742         }
 743         WARN_ON(wc->opcode != IB_WC_SEND);
 744 }
 745
 746 static void rtrs_srv_sess_up(struct rtrs_srv_sess *sess)
 747 {
 748         struct rtrs_srv *srv = sess->srv;
 749         struct rtrs_srv_ctx *ctx = srv->ctx;
 750         int up;
 751
 752         mutex_lock(&srv->paths_ev_mutex);
 753         up = ++srv->paths_up;
 754         if (up == 1)
 755                 ctx->ops.link_ev(srv, RTRS_SRV_LINK_EV_CONNECTED, NULL);
 756         mutex_unlock(&srv->paths_ev_mutex);
 757
 758         /* Mark session as established */
 759         sess->established = true;
 760 }
 761
 762 static void rtrs_srv_sess_down(struct rtrs_srv_sess *sess)
 763 {
 764         struct rtrs_srv *srv = sess->srv;
 765         struct rtrs_srv_ctx *ctx = srv->ctx;
 766
 767         if (!sess->established)
 768                 return;
 769
 770         sess->established = false;
 771         mutex_lock(&srv->paths_ev_mutex);
 772         WARN_ON(!srv->paths_up);
 773         if (--srv->paths_up == 0)
 774                 ctx->ops.link_ev(srv, RTRS_SRV_LINK_EV_DISCONNECTED, srv->priv);
 775         mutex_unlock(&srv->paths_ev_mutex);
 776 }
 777
 778 static int post_recv_sess(struct rtrs_srv_sess *sess);
 779
 780 static int process_info_req(struct rtrs_srv_con *con,
 781                             struct rtrs_msg_info_req *msg)
 782 {
 783         struct rtrs_sess *s = con->c.sess;
 784         struct rtrs_srv_sess *sess = to_srv_sess(s);
 785         struct ib_send_wr *reg_wr = NULL;
 786         struct rtrs_msg_info_rsp *rsp;
 787         struct rtrs_iu *tx_iu;
 788         struct ib_reg_wr *rwr;
 789         int mri, err;
 790         size_t tx_sz;
 791
 792         err = post_recv_sess(sess);
 793         if (unlikely(err)) {
 794                 rtrs_err(s, "post_recv_sess(), err: %d\n", err);
 795                 return err;
 796         }
 797         rwr = kcalloc(sess->mrs_num, sizeof(*rwr), GFP_KERNEL);
 798         if (unlikely(!rwr))
 799                 return -ENOMEM;
 800         strlcpy(sess->s.sessname, msg->sessname, sizeof(sess->s.sessname));
 801
 802         tx_sz  = sizeof(*rsp);
 803         tx_sz += sizeof(rsp->desc[0]) * sess->mrs_num;
 804         tx_iu = rtrs_iu_alloc(1, tx_sz, GFP_KERNEL, sess->s.dev->ib_dev,
 805                                DMA_TO_DEVICE, rtrs_srv_info_rsp_done);
 806         if (unlikely(!tx_iu)) {
 807                 err = -ENOMEM;
 808                 goto rwr_free;
 809         }
 810
 811         rsp = tx_iu->buf;
 812         rsp->type = cpu_to_le16(RTRS_MSG_INFO_RSP);
 813         rsp->sg_cnt = cpu_to_le16(sess->mrs_num);
 814
 815         for (mri = 0; mri < sess->mrs_num; mri++) {
 816                 struct ib_mr *mr = sess->mrs[mri].mr;
 817
 818                 rsp->desc[mri].addr = cpu_to_le64(mr->iova);
 819                 rsp->desc[mri].key  = cpu_to_le32(mr->rkey);
 820                 rsp->desc[mri].len  = cpu_to_le32(mr->length);
 821
 822                 /*
 823                  * Fill in reg MR request and chain them *backwards*
 824                  */
 825                 rwr[mri].wr.next = mri ? &rwr[mri - 1].wr : NULL;
 826                 rwr[mri].wr.opcode = IB_WR_REG_MR;
 827                 rwr[mri].wr.wr_cqe = &local_reg_cqe;
 828                 rwr[mri].wr.num_sge = 0;
 829                 rwr[mri].wr.send_flags = mri ? 0 : IB_SEND_SIGNALED;
 830                 rwr[mri].mr = mr;
 831                 rwr[mri].key = mr->rkey;
 832                 rwr[mri].access = (IB_ACCESS_LOCAL_WRITE |
 833                                    IB_ACCESS_REMOTE_WRITE);
 834                 reg_wr = &rwr[mri].wr;
 835         }
 836
 837         err = rtrs_srv_create_sess_files(sess);
 838         if (unlikely(err))
 839                 goto iu_free;
 840         kobject_get(&sess->kobj);
 841         get_device(&sess->srv->dev);
 842         rtrs_srv_change_state(sess, RTRS_SRV_CONNECTED);
 843         rtrs_srv_start_hb(sess);
 844
 845         /*
 846          * We do not account number of established connections at the current
 847          * moment, we rely on the client, which should send info request when
 848          * all connections are successfully established.  Thus, simply notify
 849          * listener with a proper event if we are the first path.
 850          */
 851         rtrs_srv_sess_up(sess);
 852
 853         ib_dma_sync_single_for_device(sess->s.dev->ib_dev, tx_iu->dma_addr,
 854                                       tx_iu->size, DMA_TO_DEVICE);
 855
 856         /* Send info response */
 857         err = rtrs_iu_post_send(&con->c, tx_iu, tx_sz, reg_wr);
 858         if (unlikely(err)) {
 859                 rtrs_err(s, "rtrs_iu_post_send(), err: %d\n", err);
 860 iu_free:
 861                 rtrs_iu_free(tx_iu, DMA_TO_DEVICE, sess->s.dev->ib_dev, 1);
 862         }
 863 rwr_free:
 864         kfree(rwr);
 865
 866         return err;
 867 }
 868
 869 static void rtrs_srv_info_req_done(struct ib_cq *cq, struct ib_wc *wc)
 870 {
 871         struct rtrs_srv_con *con = cq->cq_context;
 872         struct rtrs_sess *s = con->c.sess;
 873         struct rtrs_srv_sess *sess = to_srv_sess(s);
 874         struct rtrs_msg_info_req *msg;
 875         struct rtrs_iu *iu;
 876         int err;
 877
 878         WARN_ON(con->c.cid);
 879
 880         iu = container_of(wc->wr_cqe, struct rtrs_iu, cqe);
 881         if (unlikely(wc->status != IB_WC_SUCCESS)) {
 882                 rtrs_err(s, "Sess info request receive failed: %s\n",
 883                           ib_wc_status_msg(wc->status));
 884                 goto close;
 885         }
 886         WARN_ON(wc->opcode != IB_WC_RECV);
 887
 888         if (unlikely(wc->byte_len < sizeof(*msg))) {
 889                 rtrs_err(s, "Sess info request is malformed: size %d\n",
 890                           wc->byte_len);
 891                 goto close;
 892         }
 893         ib_dma_sync_single_for_cpu(sess->s.dev->ib_dev, iu->dma_addr,
 894                                    iu->size, DMA_FROM_DEVICE);
 895         msg = iu->buf;
 896         if (unlikely(le16_to_cpu(msg->type) != RTRS_MSG_INFO_REQ)) {
 897                 rtrs_err(s, "Sess info request is malformed: type %d\n",
 898                           le16_to_cpu(msg->type));
 899                 goto close;
 900         }
 901         err = process_info_req(con, msg);
 902         if (unlikely(err))
 903                 goto close;
 904
 905 out:
 906         rtrs_iu_free(iu, DMA_FROM_DEVICE, sess->s.dev->ib_dev, 1);
 907         return;
 908 close:
 909         close_sess(sess);
 910         goto out;
 911 }
 912
 913 static int post_recv_info_req(struct rtrs_srv_con *con)
 914 {
 915         struct rtrs_sess *s = con->c.sess;
 916         struct rtrs_srv_sess *sess = to_srv_sess(s);
 917         struct rtrs_iu *rx_iu;
 918         int err;
 919
 920         rx_iu = rtrs_iu_alloc(1, sizeof(struct rtrs_msg_info_req),
 921                                GFP_KERNEL, sess->s.dev->ib_dev,
 922                                DMA_FROM_DEVICE, rtrs_srv_info_req_done);
 923         if (unlikely(!rx_iu))
 924                 return -ENOMEM;
 925         /* Prepare for getting info response */
 926         err = rtrs_iu_post_recv(&con->c, rx_iu);
 927         if (unlikely(err)) {
 928                 rtrs_err(s, "rtrs_iu_post_recv(), err: %d\n", err);
 929                 rtrs_iu_free(rx_iu, DMA_FROM_DEVICE, sess->s.dev->ib_dev, 1);
 930                 return err;
 931         }
 932
 933         return 0;
 934 }
 935
 936 static int post_recv_io(struct rtrs_srv_con *con, size_t q_size)
 937 {
 938         int i, err;
 939
 940         for (i = 0; i < q_size; i++) {
 941                 err = rtrs_post_recv_empty(&con->c, &io_comp_cqe);
 942                 if (unlikely(err))
 943                         return err;
 944         }
 945
 946         return 0;
 947 }
 948
 949 static int post_recv_sess(struct rtrs_srv_sess *sess)
 950 {
 951         struct rtrs_srv *srv = sess->srv;
 952         struct rtrs_sess *s = &sess->s;
 953         size_t q_size;
 954         int err, cid;
 955
 956         for (cid = 0; cid < sess->s.con_num; cid++) {
 957                 if (cid == 0)
 958                         q_size = SERVICE_CON_QUEUE_DEPTH;
 959                 else
 960                         q_size = srv->queue_depth;
 961
 962                 err = post_recv_io(to_srv_con(sess->s.con[cid]), q_size);
 963                 if (unlikely(err)) {
 964                         rtrs_err(s, "post_recv_io(), err: %d\n", err);
 965                         return err;
 966                 }
 967         }
 968
 969         return 0;
 970 }
 971
 972 static void process_read(struct rtrs_srv_con *con,
 973                          struct rtrs_msg_rdma_read *msg,
 974                          u32 buf_id, u32 off)
 975 {
 976         struct rtrs_sess *s = con->c.sess;
 977         struct rtrs_srv_sess *sess = to_srv_sess(s);
 978         struct rtrs_srv *srv = sess->srv;
 979         struct rtrs_srv_ctx *ctx = srv->ctx;
 980         struct rtrs_srv_op *id;
 981
 982         size_t usr_len, data_len;
 983         void *data;
 984         int ret;
 985
 986         if (unlikely(sess->state != RTRS_SRV_CONNECTED)) {
 987                 rtrs_err_rl(s,
 988                              "Processing read request failed,  session is disconnected, sess state %s\n",
 989                              rtrs_srv_state_str(sess->state));
 990                 return;
 991         }
 992         if (unlikely(msg->sg_cnt != 1 && msg->sg_cnt != 0)) {
 993                 rtrs_err_rl(s,
 994                             "Processing read request failed, invalid message\n");
 995                 return;
 996         }
 997         rtrs_srv_get_ops_ids(sess);
 998         rtrs_srv_update_rdma_stats(sess->stats, off, READ);
 999         id = sess->ops_ids[buf_id];
1000         id->con         = con;
1001         id->dir         = READ;
1002         id->msg_id      = buf_id;
1003         id->rd_msg      = msg;
1004         usr_len = le16_to_cpu(msg->usr_len);
1005         data_len = off - usr_len;
1006         data = page_address(srv->chunks[buf_id]);
1007         ret = ctx->ops.rdma_ev(srv, srv->priv, id, READ, data, data_len,
1008                            data + data_len, usr_len);
1009
1010         if (unlikely(ret)) {
1011                 rtrs_err_rl(s,
1012                              "Processing read request failed, user module cb reported for msg_id %d, err: %d\n",
1013                              buf_id, ret);
1014                 goto send_err_msg;
1015         }
1016
1017         return;
1018
1019 send_err_msg:
1020         ret = send_io_resp_imm(con, id, ret);
1021         if (ret < 0) {
1022                 rtrs_err_rl(s,
1023                              "Sending err msg for failed RDMA-Write-Req failed, msg_id %d, err: %d\n",
1024                              buf_id, ret);
1025                 close_sess(sess);
1026         }
1027         rtrs_srv_put_ops_ids(sess);
1028 }
1029
1030 static void process_write(struct rtrs_srv_con *con,
1031                           struct rtrs_msg_rdma_write *req,
1032                           u32 buf_id, u32 off)
1033 {
1034         struct rtrs_sess *s = con->c.sess;
1035         struct rtrs_srv_sess *sess = to_srv_sess(s);
1036         struct rtrs_srv *srv = sess->srv;
1037         struct rtrs_srv_ctx *ctx = srv->ctx;
1038         struct rtrs_srv_op *id;
1039
1040         size_t data_len, usr_len;
1041         void *data;
1042         int ret;
1043
1044         if (unlikely(sess->state != RTRS_SRV_CONNECTED)) {
1045                 rtrs_err_rl(s,
1046                              "Processing write request failed,  session is disconnected, sess state %s\n",
1047                              rtrs_srv_state_str(sess->state));
1048                 return;
1049         }
1050         rtrs_srv_get_ops_ids(sess);
1051         rtrs_srv_update_rdma_stats(sess->stats, off, WRITE);
1052         id = sess->ops_ids[buf_id];
1053         id->con    = con;
1054         id->dir    = WRITE;
1055         id->msg_id = buf_id;
1056
1057         usr_len = le16_to_cpu(req->usr_len);
1058         data_len = off - usr_len;
1059         data = page_address(srv->chunks[buf_id]);
1060         ret = ctx->ops.rdma_ev(srv, srv->priv, id, WRITE, data, data_len,
1061                            data + data_len, usr_len);
1062         if (unlikely(ret)) {
1063                 rtrs_err_rl(s,
1064                              "Processing write request failed, user module callback reports err: %d\n",
1065                              ret);
1066                 goto send_err_msg;
1067         }
1068
1069         return;
1070
1071 send_err_msg:
1072         ret = send_io_resp_imm(con, id, ret);
1073         if (ret < 0) {
1074                 rtrs_err_rl(s,
1075                              "Processing write request failed, sending I/O response failed, msg_id %d, err: %d\n",
1076                              buf_id, ret);
1077                 close_sess(sess);
1078         }
1079         rtrs_srv_put_ops_ids(sess);
1080 }
1081
1082 static void process_io_req(struct rtrs_srv_con *con, void *msg,
1083                            u32 id, u32 off)
1084 {
1085         struct rtrs_sess *s = con->c.sess;
1086         struct rtrs_srv_sess *sess = to_srv_sess(s);
1087         struct rtrs_msg_rdma_hdr *hdr;
1088         unsigned int type;
1089
1090         ib_dma_sync_single_for_cpu(sess->s.dev->ib_dev, sess->dma_addr[id],
1091                                    max_chunk_size, DMA_BIDIRECTIONAL);
1092         hdr = msg;
1093         type = le16_to_cpu(hdr->type);
1094
1095         switch (type) {
1096         case RTRS_MSG_WRITE:
1097                 process_write(con, msg, id, off);
1098                 break;
1099         case RTRS_MSG_READ:
1100                 process_read(con, msg, id, off);
1101                 break;
1102         default:
1103                 rtrs_err(s,
1104                           "Processing I/O request failed, unknown message type received: 0x%02x\n",
1105                           type);
1106                 goto err;
1107         }
1108
1109         return;
1110
1111 err:
1112         close_sess(sess);
1113 }
1114
1115 static void rtrs_srv_inv_rkey_done(struct ib_cq *cq, struct ib_wc *wc)
1116 {
1117         struct rtrs_srv_mr *mr =
1118                 container_of(wc->wr_cqe, typeof(*mr), inv_cqe);
1119         struct rtrs_srv_con *con = cq->cq_context;
1120         struct rtrs_sess *s = con->c.sess;
1121         struct rtrs_srv_sess *sess = to_srv_sess(s);
1122         struct rtrs_srv *srv = sess->srv;
1123         u32 msg_id, off;
1124         void *data;
1125
1126         if (unlikely(wc->status != IB_WC_SUCCESS)) {
1127                 rtrs_err(s, "Failed IB_WR_LOCAL_INV: %s\n",
1128                           ib_wc_status_msg(wc->status));
1129                 close_sess(sess);
1130         }
1131         msg_id = mr->msg_id;
1132         off = mr->msg_off;
1133         data = page_address(srv->chunks[msg_id]) + off;
1134         process_io_req(con, data, msg_id, off);
1135 }
1136
1137 static int rtrs_srv_inv_rkey(struct rtrs_srv_con *con,
1138                               struct rtrs_srv_mr *mr)
1139 {
1140         struct ib_send_wr wr = {
1141                 .opcode             = IB_WR_LOCAL_INV,
1142                 .wr_cqe             = &mr->inv_cqe,
1143                 .send_flags         = IB_SEND_SIGNALED,
1144                 .ex.invalidate_rkey = mr->mr->rkey,
1145         };
1146         mr->inv_cqe.done = rtrs_srv_inv_rkey_done;
1147
1148         return ib_post_send(con->c.qp, &wr, NULL);
1149 }
1150
1151 static void rtrs_rdma_process_wr_wait_list(struct rtrs_srv_con *con)
1152 {
1153         spin_lock(&con->rsp_wr_wait_lock);
1154         while (!list_empty(&con->rsp_wr_wait_list)) {
1155                 struct rtrs_srv_op *id;
1156                 int ret;
1157
1158                 id = list_entry(con->rsp_wr_wait_list.next,
1159                                 struct rtrs_srv_op, wait_list);
1160                 list_del(&id->wait_list);
1161
1162                 spin_unlock(&con->rsp_wr_wait_lock);
1163                 ret = rtrs_srv_resp_rdma(id, id->status);
1164                 spin_lock(&con->rsp_wr_wait_lock);
1165
1166                 if (!ret) {
1167                         list_add(&id->wait_list, &con->rsp_wr_wait_list);
1168                         break;
1169                 }
1170         }
1171         spin_unlock(&con->rsp_wr_wait_lock);
1172 }
1173
1174 static void rtrs_srv_rdma_done(struct ib_cq *cq, struct ib_wc *wc)
1175 {
1176         struct rtrs_srv_con *con = cq->cq_context;
1177         struct rtrs_sess *s = con->c.sess;
1178         struct rtrs_srv_sess *sess = to_srv_sess(s);
1179         struct rtrs_srv *srv = sess->srv;
1180         u32 imm_type, imm_payload;
1181         int err;
1182
1183         if (unlikely(wc->status != IB_WC_SUCCESS)) {
1184                 if (wc->status != IB_WC_WR_FLUSH_ERR) {
1185                         rtrs_err(s,
1186                                   "%s (wr_cqe: %p, type: %d, vendor_err: 0x%x, len: %u)\n",
1187                                   ib_wc_status_msg(wc->status), wc->wr_cqe,
1188                                   wc->opcode, wc->vendor_err, wc->byte_len);
1189                         close_sess(sess);
1190                 }
1191                 return;
1192         }
1193
1194         switch (wc->opcode) {
1195         case IB_WC_RECV_RDMA_WITH_IMM:
1196                 /*
1197                  * post_recv() RDMA write completions of IO reqs (read/write)
1198                  * and hb
1199                  */
1200                 if (WARN_ON(wc->wr_cqe != &io_comp_cqe))
1201                         return;
1202                 err = rtrs_post_recv_empty(&con->c, &io_comp_cqe);
1203                 if (unlikely(err)) {
1204                         rtrs_err(s, "rtrs_post_recv(), err: %d\n", err);
1205                         close_sess(sess);
1206                         break;
1207                 }
1208                 rtrs_from_imm(be32_to_cpu(wc->ex.imm_data),
1209                                &imm_type, &imm_payload);
1210                 if (likely(imm_type == RTRS_IO_REQ_IMM)) {
1211                         u32 msg_id, off;
1212                         void *data;
1213
1214                         msg_id = imm_payload >> sess->mem_bits;
1215                         off = imm_payload & ((1 << sess->mem_bits) - 1);
1216                         if (unlikely(msg_id > srv->queue_depth ||
1217                                      off > max_chunk_size)) {
1218                                 rtrs_err(s, "Wrong msg_id %u, off %u\n",
1219                                           msg_id, off);
1220                                 close_sess(sess);
1221                                 return;
1222                         }
1223                         if (always_invalidate) {
1224                                 struct rtrs_srv_mr *mr = &sess->mrs[msg_id];
1225
1226                                 mr->msg_off = off;
1227                                 mr->msg_id = msg_id;
1228                                 err = rtrs_srv_inv_rkey(con, mr);
1229                                 if (unlikely(err)) {
1230                                         rtrs_err(s, "rtrs_post_recv(), err: %d\n",
1231                                                   err);
1232                                         close_sess(sess);
1233                                         break;
1234                                 }
1235                         } else {
1236                                 data = page_address(srv->chunks[msg_id]) + off;
1237                                 process_io_req(con, data, msg_id, off);
1238                         }
1239                 } else if (imm_type == RTRS_HB_MSG_IMM) {
1240                         WARN_ON(con->c.cid);
1241                         rtrs_send_hb_ack(&sess->s);
1242                 } else if (imm_type == RTRS_HB_ACK_IMM) {
1243                         WARN_ON(con->c.cid);
1244                         sess->s.hb_missed_cnt = 0;
1245                 } else {
1246                         rtrs_wrn(s, "Unknown IMM type %u\n", imm_type);
1247                 }
1248                 break;
1249         case IB_WC_RDMA_WRITE:
1250         case IB_WC_SEND:
1251                 /*
1252                  * post_send() RDMA write completions of IO reqs (read/write)
1253                  * and hb
1254                  */
1255                 atomic_add(srv->queue_depth, &con->sq_wr_avail);
1256
1257                 if (unlikely(!list_empty_careful(&con->rsp_wr_wait_list)))
1258                         rtrs_rdma_process_wr_wait_list(con);
1259
1260                 break;
1261         default:
1262                 rtrs_wrn(s, "Unexpected WC type: %d\n", wc->opcode);
1263                 return;
1264         }
1265 }
1266
1267 /**
1268  * rtrs_srv_get_sess_name() - Get rtrs_srv peer hostname.
1269  * @srv:        Session
1270  * @sessname:   Sessname buffer
1271  * @len:        Length of sessname buffer
1272  */
1273 int rtrs_srv_get_sess_name(struct rtrs_srv *srv, char *sessname, size_t len)
1274 {
1275         struct rtrs_srv_sess *sess;
1276         int err = -ENOTCONN;
1277
1278         mutex_lock(&srv->paths_mutex);
1279         list_for_each_entry(sess, &srv->paths_list, s.entry) {
1280                 if (sess->state != RTRS_SRV_CONNECTED)
1281                         continue;
1282                 strlcpy(sessname, sess->s.sessname,
1283                        min_t(size_t, sizeof(sess->s.sessname), len));
1284                 err = 0;
1285                 break;
1286         }
1287         mutex_unlock(&srv->paths_mutex);
1288
1289         return err;
1290 }
1291 EXPORT_SYMBOL(rtrs_srv_get_sess_name);
1292
1293 /**
1294  * rtrs_srv_get_sess_qdepth() - Get rtrs_srv qdepth.
1295  * @srv:        Session
1296  */
1297 int rtrs_srv_get_queue_depth(struct rtrs_srv *srv)
1298 {
1299         return srv->queue_depth;
1300 }
1301 EXPORT_SYMBOL(rtrs_srv_get_queue_depth);
1302
1303 static int find_next_bit_ring(struct rtrs_srv_sess *sess)
1304 {
1305         struct ib_device *ib_dev = sess->s.dev->ib_dev;
1306         int v;
1307
1308         v = cpumask_next(sess->cur_cq_vector, &cq_affinity_mask);
1309         if (v >= nr_cpu_ids || v >= ib_dev->num_comp_vectors)
1310                 v = cpumask_first(&cq_affinity_mask);
1311         return v;
1312 }
1313
1314 static int rtrs_srv_get_next_cq_vector(struct rtrs_srv_sess *sess)
1315 {
1316         sess->cur_cq_vector = find_next_bit_ring(sess);
1317
1318         return sess->cur_cq_vector;
1319 }
1320
1321 static struct rtrs_srv *__alloc_srv(struct rtrs_srv_ctx *ctx,
1322                                      const uuid_t *paths_uuid)
1323 {
1324         struct rtrs_srv *srv;
1325         int i;
1326
1327         srv = kzalloc(sizeof(*srv), GFP_KERNEL);
1328         if  (!srv)
1329                 return NULL;
1330
1331         refcount_set(&srv->refcount, 1);
1332         INIT_LIST_HEAD(&srv->paths_list);
1333         mutex_init(&srv->paths_mutex);
1334         mutex_init(&srv->paths_ev_mutex);
1335         uuid_copy(&srv->paths_uuid, paths_uuid);
1336         srv->queue_depth = sess_queue_depth;
1337         srv->ctx = ctx;
1338
1339         srv->chunks = kcalloc(srv->queue_depth, sizeof(*srv->chunks),
1340                               GFP_KERNEL);
1341         if (!srv->chunks)
1342                 goto err_free_srv;
1343
1344         for (i = 0; i < srv->queue_depth; i++) {
1345                 srv->chunks[i] = mempool_alloc(chunk_pool, GFP_KERNEL);
1346                 if (!srv->chunks[i])
1347                         goto err_free_chunks;
1348         }
1349         list_add(&srv->ctx_list, &ctx->srv_list);
1350
1351         return srv;
1352
1353 err_free_chunks:
1354         while (i--)
1355                 mempool_free(srv->chunks[i], chunk_pool);
1356         kfree(srv->chunks);
1357
1358 err_free_srv:
1359         kfree(srv);
1360
1361         return NULL;
1362 }
1363
1364 static void free_srv(struct rtrs_srv *srv)
1365 {
1366         int i;
1367
1368         WARN_ON(refcount_read(&srv->refcount));
1369         for (i = 0; i < srv->queue_depth; i++)
1370                 mempool_free(srv->chunks[i], chunk_pool);
1371         kfree(srv->chunks);
1372         mutex_destroy(&srv->paths_mutex);
1373         mutex_destroy(&srv->paths_ev_mutex);
1374         /* last put to release the srv structure */
1375         put_device(&srv->dev);
1376 }
1377
1378 static inline struct rtrs_srv *__find_srv_and_get(struct rtrs_srv_ctx *ctx,
1379                                                    const uuid_t *paths_uuid)
1380 {
1381         struct rtrs_srv *srv;
1382
1383         list_for_each_entry(srv, &ctx->srv_list, ctx_list) {
1384                 if (uuid_equal(&srv->paths_uuid, paths_uuid) &&
1385                     refcount_inc_not_zero(&srv->refcount))
1386                         return srv;
1387         }
1388
1389         return NULL;
1390 }
1391
1392 static struct rtrs_srv *get_or_create_srv(struct rtrs_srv_ctx *ctx,
1393                                            const uuid_t *paths_uuid)
1394 {
1395         struct rtrs_srv *srv;
1396
1397         mutex_lock(&ctx->srv_mutex);
1398         srv = __find_srv_and_get(ctx, paths_uuid);
1399         if (!srv)
1400                 srv = __alloc_srv(ctx, paths_uuid);
1401         mutex_unlock(&ctx->srv_mutex);
1402
1403         return srv;
1404 }
1405
1406 static void put_srv(struct rtrs_srv *srv)
1407 {
1408         if (refcount_dec_and_test(&srv->refcount)) {
1409                 struct rtrs_srv_ctx *ctx = srv->ctx;
1410
1411                 WARN_ON(srv->dev.kobj.state_in_sysfs);
1412
1413                 mutex_lock(&ctx->srv_mutex);
1414                 list_del(&srv->ctx_list);
1415                 mutex_unlock(&ctx->srv_mutex);
1416                 free_srv(srv);
1417         }
1418 }
1419
1420 static void __add_path_to_srv(struct rtrs_srv *srv,
1421                               struct rtrs_srv_sess *sess)
1422 {
1423         list_add_tail(&sess->s.entry, &srv->paths_list);
1424         srv->paths_num++;
1425         WARN_ON(srv->paths_num >= MAX_PATHS_NUM);
1426 }
1427
1428 static void del_path_from_srv(struct rtrs_srv_sess *sess)
1429 {
1430         struct rtrs_srv *srv = sess->srv;
1431
1432         if (WARN_ON(!srv))
1433                 return;
1434
1435         mutex_lock(&srv->paths_mutex);
1436         list_del(&sess->s.entry);
1437         WARN_ON(!srv->paths_num);
1438         srv->paths_num--;
1439         mutex_unlock(&srv->paths_mutex);
1440 }
1441
1442 /* return true if addresses are the same, error other wise */
1443 static int sockaddr_cmp(const struct sockaddr *a, const struct sockaddr *b)
1444 {
1445         switch (a->sa_family) {
1446         case AF_IB:
1447                 return memcmp(&((struct sockaddr_ib *)a)->sib_addr,
1448                               &((struct sockaddr_ib *)b)->sib_addr,
1449                               sizeof(struct ib_addr)) &&
1450                         (b->sa_family == AF_IB);
1451         case AF_INET:
1452                 return memcmp(&((struct sockaddr_in *)a)->sin_addr,
1453                               &((struct sockaddr_in *)b)->sin_addr,
1454                               sizeof(struct in_addr)) &&
1455                         (b->sa_family == AF_INET);
1456         case AF_INET6:
1457                 return memcmp(&((struct sockaddr_in6 *)a)->sin6_addr,
1458                               &((struct sockaddr_in6 *)b)->sin6_addr,
1459                               sizeof(struct in6_addr)) &&
1460                         (b->sa_family == AF_INET6);
1461         default:
1462                 return -ENOENT;
1463         }
1464 }
1465
1466 static bool __is_path_w_addr_exists(struct rtrs_srv *srv,
1467                                     struct rdma_addr *addr)
1468 {
1469         struct rtrs_srv_sess *sess;
1470
1471         list_for_each_entry(sess, &srv->paths_list, s.entry)
1472                 if (!sockaddr_cmp((struct sockaddr *)&sess->s.dst_addr,
1473                                   (struct sockaddr *)&addr->dst_addr) &&
1474                     !sockaddr_cmp((struct sockaddr *)&sess->s.src_addr,
1475                                   (struct sockaddr *)&addr->src_addr))
1476                         return true;
1477
1478         return false;
1479 }
1480
1481 static void free_sess(struct rtrs_srv_sess *sess)
1482 {
1483         if (sess->kobj.state_in_sysfs)
1484                 kobject_put(&sess->kobj);
1485         else
1486                 kfree(sess);
1487 }
1488
1489 static void rtrs_srv_close_work(struct work_struct *work)
1490 {
1491         struct rtrs_srv_sess *sess;
1492         struct rtrs_srv_con *con;
1493         int i;
1494
1495         sess = container_of(work, typeof(*sess), close_work);
1496
1497         rtrs_srv_destroy_sess_files(sess);
1498         rtrs_srv_stop_hb(sess);
1499
1500         for (i = 0; i < sess->s.con_num; i++) {
1501                 if (!sess->s.con[i])
1502                         continue;
1503                 con = to_srv_con(sess->s.con[i]);
1504                 rdma_disconnect(con->c.cm_id);
1505                 ib_drain_qp(con->c.qp);
1506         }
1507         /* Wait for all inflights */
1508         rtrs_srv_wait_ops_ids(sess);
1509
1510         /* Notify upper layer if we are the last path */
1511         rtrs_srv_sess_down(sess);
1512
1513         unmap_cont_bufs(sess);
1514         rtrs_srv_free_ops_ids(sess);
1515
1516         for (i = 0; i < sess->s.con_num; i++) {
1517                 if (!sess->s.con[i])
1518                         continue;
1519                 con = to_srv_con(sess->s.con[i]);
1520                 rtrs_cq_qp_destroy(&con->c);
1521                 rdma_destroy_id(con->c.cm_id);
1522                 kfree(con);
1523         }
1524         rtrs_ib_dev_put(sess->s.dev);
1525
1526         del_path_from_srv(sess);
1527         put_srv(sess->srv);
1528         sess->srv = NULL;
1529         rtrs_srv_change_state(sess, RTRS_SRV_CLOSED);
1530
1531         kfree(sess->dma_addr);
1532         kfree(sess->s.con);
1533         free_sess(sess);
1534 }
1535
1536 static int rtrs_rdma_do_accept(struct rtrs_srv_sess *sess,
1537                                struct rdma_cm_id *cm_id)
1538 {
1539         struct rtrs_srv *srv = sess->srv;
1540         struct rtrs_msg_conn_rsp msg;
1541         struct rdma_conn_param param;
1542         int err;
1543
1544         param = (struct rdma_conn_param) {
1545                 .rnr_retry_count = 7,
1546                 .private_data = &msg,
1547                 .private_data_len = sizeof(msg),
1548         };
1549
1550         msg = (struct rtrs_msg_conn_rsp) {
1551                 .magic = cpu_to_le16(RTRS_MAGIC),
1552                 .version = cpu_to_le16(RTRS_PROTO_VER),
1553                 .queue_depth = cpu_to_le16(srv->queue_depth),
1554                 .max_io_size = cpu_to_le32(max_chunk_size - MAX_HDR_SIZE),
1555                 .max_hdr_size = cpu_to_le32(MAX_HDR_SIZE),
1556         };
1557
1558         if (always_invalidate)
1559                 msg.flags = cpu_to_le32(RTRS_MSG_NEW_RKEY_F);
1560
1561         err = rdma_accept(cm_id, &param);
1562         if (err)
1563                 pr_err("rdma_accept(), err: %d\n", err);
1564
1565         return err;
1566 }
1567
1568 static int rtrs_rdma_do_reject(struct rdma_cm_id *cm_id, int errno)
1569 {
1570         struct rtrs_msg_conn_rsp msg;
1571         int err;
1572
1573         msg = (struct rtrs_msg_conn_rsp) {
1574                 .magic = cpu_to_le16(RTRS_MAGIC),
1575                 .version = cpu_to_le16(RTRS_PROTO_VER),
1576                 .errno = cpu_to_le16(errno),
1577         };
1578
1579         err = rdma_reject(cm_id, &msg, sizeof(msg));
1580         if (err)
1581                 pr_err("rdma_reject(), err: %d\n", err);
1582
1583         /* Bounce errno back */
1584         return errno;
1585 }
1586
1587 static struct rtrs_srv_sess *
1588 __find_sess(struct rtrs_srv *srv, const uuid_t *sess_uuid)
1589 {
1590         struct rtrs_srv_sess *sess;
1591
1592         list_for_each_entry(sess, &srv->paths_list, s.entry) {
1593                 if (uuid_equal(&sess->s.uuid, sess_uuid))
1594                         return sess;
1595         }
1596
1597         return NULL;
1598 }
1599
1600 static int create_con(struct rtrs_srv_sess *sess,
1601                       struct rdma_cm_id *cm_id,
1602                       unsigned int cid)
1603 {
1604         struct rtrs_srv *srv = sess->srv;
1605         struct rtrs_sess *s = &sess->s;
1606         struct rtrs_srv_con *con;
1607
1608         u16 cq_size, wr_queue_size;
1609         int err, cq_vector;
1610
1611         con = kzalloc(sizeof(*con), GFP_KERNEL);
1612         if (!con) {
1613                 err = -ENOMEM;
1614                 goto err;
1615         }
1616
1617         spin_lock_init(&con->rsp_wr_wait_lock);
1618         INIT_LIST_HEAD(&con->rsp_wr_wait_list);
1619         con->c.cm_id = cm_id;
1620         con->c.sess = &sess->s;
1621         con->c.cid = cid;
1622         atomic_set(&con->wr_cnt, 0);
1623
1624         if (con->c.cid == 0) {
1625                 /*
1626                  * All receive and all send (each requiring invalidate)
1627                  * + 2 for drain and heartbeat
1628                  */
1629                 wr_queue_size = SERVICE_CON_QUEUE_DEPTH * 3 + 2;
1630                 cq_size = wr_queue_size;
1631         } else {
1632                 /*
1633                  * If we have all receive requests posted and
1634                  * all write requests posted and each read request
1635                  * requires an invalidate request + drain
1636                  * and qp gets into error state.
1637                  */
1638                 cq_size = srv->queue_depth * 3 + 1;
1639                 /*
1640                  * In theory we might have queue_depth * 32
1641                  * outstanding requests if an unsafe global key is used
1642                  * and we have queue_depth read requests each consisting
1643                  * of 32 different addresses. div 3 for mlx5.
1644                  */
1645                 wr_queue_size = sess->s.dev->ib_dev->attrs.max_qp_wr / 3;
1646         }
1647         atomic_set(&con->sq_wr_avail, wr_queue_size);
1648         cq_vector = rtrs_srv_get_next_cq_vector(sess);
1649
1650         /* TODO: SOFTIRQ can be faster, but be careful with softirq context */
1651         err = rtrs_cq_qp_create(&sess->s, &con->c, 1, cq_vector, cq_size,
1652                                  wr_queue_size, IB_POLL_WORKQUEUE);
1653         if (err) {
1654                 rtrs_err(s, "rtrs_cq_qp_create(), err: %d\n", err);
1655                 goto free_con;
1656         }
1657         if (con->c.cid == 0) {
1658                 err = post_recv_info_req(con);
1659                 if (err)
1660                         goto free_cqqp;
1661         }
1662         WARN_ON(sess->s.con[cid]);
1663         sess->s.con[cid] = &con->c;
1664
1665         /*
1666          * Change context from server to current connection.  The other
1667          * way is to use cm_id->qp->qp_context, which does not work on OFED.
1668          */
1669         cm_id->context = &con->c;
1670
1671         return 0;
1672
1673 free_cqqp:
1674         rtrs_cq_qp_destroy(&con->c);
1675 free_con:
1676         kfree(con);
1677
1678 err:
1679         return err;
1680 }
1681
1682 static struct rtrs_srv_sess *__alloc_sess(struct rtrs_srv *srv,
1683                                            struct rdma_cm_id *cm_id,
1684                                            unsigned int con_num,
1685                                            unsigned int recon_cnt,
1686                                            const uuid_t *uuid)
1687 {
1688         struct rtrs_srv_sess *sess;
1689         int err = -ENOMEM;
1690
1691         if (srv->paths_num >= MAX_PATHS_NUM) {
1692                 err = -ECONNRESET;
1693                 goto err;
1694         }
1695         if (__is_path_w_addr_exists(srv, &cm_id->route.addr)) {
1696                 err = -EEXIST;
1697                 pr_err("Path with same addr exists\n");
1698                 goto err;
1699         }
1700         sess = kzalloc(sizeof(*sess), GFP_KERNEL);
1701         if (!sess)
1702                 goto err;
1703
1704         sess->stats = kzalloc(sizeof(*sess->stats), GFP_KERNEL);
1705         if (!sess->stats)
1706                 goto err_free_sess;
1707
1708         sess->stats->sess = sess;
1709
1710         sess->dma_addr = kcalloc(srv->queue_depth, sizeof(*sess->dma_addr),
1711                                  GFP_KERNEL);
1712         if (!sess->dma_addr)
1713                 goto err_free_stats;
1714
1715         sess->s.con = kcalloc(con_num, sizeof(*sess->s.con), GFP_KERNEL);
1716         if (!sess->s.con)
1717                 goto err_free_dma_addr;
1718
1719         sess->state = RTRS_SRV_CONNECTING;
1720         sess->srv = srv;
1721         sess->cur_cq_vector = -1;
1722         sess->s.dst_addr = cm_id->route.addr.dst_addr;
1723         sess->s.src_addr = cm_id->route.addr.src_addr;
1724         sess->s.con_num = con_num;
1725         sess->s.recon_cnt = recon_cnt;
1726         uuid_copy(&sess->s.uuid, uuid);
1727         spin_lock_init(&sess->state_lock);
1728         INIT_WORK(&sess->close_work, rtrs_srv_close_work);
1729         rtrs_srv_init_hb(sess);
1730
1731         sess->s.dev = rtrs_ib_dev_find_or_add(cm_id->device, &dev_pd);
1732         if (!sess->s.dev) {
1733                 err = -ENOMEM;
1734                 goto err_free_con;
1735         }
1736         err = map_cont_bufs(sess);
1737         if (err)
1738                 goto err_put_dev;
1739
1740         err = rtrs_srv_alloc_ops_ids(sess);
1741         if (err)
1742                 goto err_unmap_bufs;
1743
1744         __add_path_to_srv(srv, sess);
1745
1746         return sess;
1747
1748 err_unmap_bufs:
1749         unmap_cont_bufs(sess);
1750 err_put_dev:
1751         rtrs_ib_dev_put(sess->s.dev);
1752 err_free_con:
1753         kfree(sess->s.con);
1754 err_free_dma_addr:
1755         kfree(sess->dma_addr);
1756 err_free_stats:
1757         kfree(sess->stats);
1758 err_free_sess:
1759         kfree(sess);
1760 err:
1761         return ERR_PTR(err);
1762 }
1763
1764 static int rtrs_rdma_connect(struct rdma_cm_id *cm_id,
1765                               const struct rtrs_msg_conn_req *msg,
1766                               size_t len)
1767 {
1768         struct rtrs_srv_ctx *ctx = cm_id->context;
1769         struct rtrs_srv_sess *sess;
1770         struct rtrs_srv *srv;
1771
1772         u16 version, con_num, cid;
1773         u16 recon_cnt;
1774         int err;
1775
1776         if (len < sizeof(*msg)) {
1777                 pr_err("Invalid RTRS connection request\n");
1778                 goto reject_w_econnreset;
1779         }
1780         if (le16_to_cpu(msg->magic) != RTRS_MAGIC) {
1781                 pr_err("Invalid RTRS magic\n");
1782                 goto reject_w_econnreset;
1783         }
1784         version = le16_to_cpu(msg->version);
1785         if (version >> 8 != RTRS_PROTO_VER_MAJOR) {
1786                 pr_err("Unsupported major RTRS version: %d, expected %d\n",
1787                        version >> 8, RTRS_PROTO_VER_MAJOR);
1788                 goto reject_w_econnreset;
1789         }
1790         con_num = le16_to_cpu(msg->cid_num);
1791         if (con_num > 4096) {
1792                 /* Sanity check */
1793                 pr_err("Too many connections requested: %d\n", con_num);
1794                 goto reject_w_econnreset;
1795         }
1796         cid = le16_to_cpu(msg->cid);
1797         if (cid >= con_num) {
1798                 /* Sanity check */
1799                 pr_err("Incorrect cid: %d >= %d\n", cid, con_num);
1800                 goto reject_w_econnreset;
1801         }
1802         recon_cnt = le16_to_cpu(msg->recon_cnt);
1803         srv = get_or_create_srv(ctx, &msg->paths_uuid);
1804         if (!srv) {
1805                 err = -ENOMEM;
1806                 goto reject_w_err;
1807         }
1808         mutex_lock(&srv->paths_mutex);
1809         sess = __find_sess(srv, &msg->sess_uuid);
1810         if (sess) {
1811                 struct rtrs_sess *s = &sess->s;
1812
1813                 /* Session already holds a reference */
1814                 put_srv(srv);
1815
1816                 if (sess->state != RTRS_SRV_CONNECTING) {
1817                         rtrs_err(s, "Session in wrong state: %s\n",
1818                                   rtrs_srv_state_str(sess->state));
1819                         mutex_unlock(&srv->paths_mutex);
1820                         goto reject_w_econnreset;
1821                 }
1822                 /*
1823                  * Sanity checks
1824                  */
1825                 if (con_num != sess->s.con_num || cid >= sess->s.con_num) {
1826                         rtrs_err(s, "Incorrect request: %d, %d\n",
1827                                   cid, con_num);
1828                         mutex_unlock(&srv->paths_mutex);
1829                         goto reject_w_econnreset;
1830                 }
1831                 if (sess->s.con[cid]) {
1832                         rtrs_err(s, "Connection already exists: %d\n",
1833                                   cid);
1834                         mutex_unlock(&srv->paths_mutex);
1835                         goto reject_w_econnreset;
1836                 }
1837         } else {
1838                 sess = __alloc_sess(srv, cm_id, con_num, recon_cnt,
1839                                     &msg->sess_uuid);
1840                 if (IS_ERR(sess)) {
1841                         mutex_unlock(&srv->paths_mutex);
1842                         put_srv(srv);
1843                         err = PTR_ERR(sess);
1844                         goto reject_w_err;
1845                 }
1846         }
1847         err = create_con(sess, cm_id, cid);
1848         if (err) {
1849                 (void)rtrs_rdma_do_reject(cm_id, err);
1850                 /*
1851                  * Since session has other connections we follow normal way
1852                  * through workqueue, but still return an error to tell cma.c
1853                  * to call rdma_destroy_id() for current connection.
1854                  */
1855                 goto close_and_return_err;
1856         }
1857         err = rtrs_rdma_do_accept(sess, cm_id);
1858         if (err) {
1859                 (void)rtrs_rdma_do_reject(cm_id, err);
1860                 /*
1861                  * Since current connection was successfully added to the
1862                  * session we follow normal way through workqueue to close the
1863                  * session, thus return 0 to tell cma.c we call
1864                  * rdma_destroy_id() ourselves.
1865                  */
1866                 err = 0;
1867                 goto close_and_return_err;
1868         }
1869         mutex_unlock(&srv->paths_mutex);
1870
1871         return 0;
1872
1873 reject_w_err:
1874         return rtrs_rdma_do_reject(cm_id, err);
1875
1876 reject_w_econnreset:
1877         return rtrs_rdma_do_reject(cm_id, -ECONNRESET);
1878
1879 close_and_return_err:
1880         close_sess(sess);
1881         mutex_unlock(&srv->paths_mutex);
1882
1883         return err;
1884 }
1885
1886 static int rtrs_srv_rdma_cm_handler(struct rdma_cm_id *cm_id,
1887                                      struct rdma_cm_event *ev)
1888 {
1889         struct rtrs_srv_sess *sess = NULL;
1890         struct rtrs_sess *s = NULL;
1891
1892         if (ev->event != RDMA_CM_EVENT_CONNECT_REQUEST) {
1893                 struct rtrs_con *c = cm_id->context;
1894
1895                 s = c->sess;
1896                 sess = to_srv_sess(s);
1897         }
1898
1899         switch (ev->event) {
1900         case RDMA_CM_EVENT_CONNECT_REQUEST:
1901                 /*
1902                  * In case of error cma.c will destroy cm_id,
1903                  * see cma_process_remove()
1904                  */
1905                 return rtrs_rdma_connect(cm_id, ev->param.conn.private_data,
1906                                           ev->param.conn.private_data_len);
1907         case RDMA_CM_EVENT_ESTABLISHED:
1908                 /* Nothing here */
1909                 break;
1910         case RDMA_CM_EVENT_REJECTED:
1911         case RDMA_CM_EVENT_CONNECT_ERROR:
1912         case RDMA_CM_EVENT_UNREACHABLE:
1913                 rtrs_err(s, "CM error (CM event: %s, err: %d)\n",
1914                           rdma_event_msg(ev->event), ev->status);
1915                 close_sess(sess);
1916                 break;
1917         case RDMA_CM_EVENT_DISCONNECTED:
1918         case RDMA_CM_EVENT_ADDR_CHANGE:
1919         case RDMA_CM_EVENT_TIMEWAIT_EXIT:
1920                 close_sess(sess);
1921                 break;
1922         case RDMA_CM_EVENT_DEVICE_REMOVAL:
1923                 close_sess(sess);
1924                 break;
1925         default:
1926                 pr_err("Ignoring unexpected CM event %s, err %d\n",
1927                        rdma_event_msg(ev->event), ev->status);
1928                 break;
1929         }
1930
1931         return 0;
1932 }
1933
1934 static struct rdma_cm_id *rtrs_srv_cm_init(struct rtrs_srv_ctx *ctx,
1935                                             struct sockaddr *addr,
1936                                             enum rdma_ucm_port_space ps)
1937 {
1938         struct rdma_cm_id *cm_id;
1939         int ret;
1940
1941         cm_id = rdma_create_id(&init_net, rtrs_srv_rdma_cm_handler,
1942                                ctx, ps, IB_QPT_RC);
1943         if (IS_ERR(cm_id)) {
1944                 ret = PTR_ERR(cm_id);
1945                 pr_err("Creating id for RDMA connection failed, err: %d\n",
1946                        ret);
1947                 goto err_out;
1948         }
1949         ret = rdma_bind_addr(cm_id, addr);
1950         if (ret) {
1951                 pr_err("Binding RDMA address failed, err: %d\n", ret);
1952                 goto err_cm;
1953         }
1954         ret = rdma_listen(cm_id, 64);
1955         if (ret) {
1956                 pr_err("Listening on RDMA connection failed, err: %d\n",
1957                        ret);
1958                 goto err_cm;
1959         }
1960
1961         return cm_id;
1962
1963 err_cm:
1964         rdma_destroy_id(cm_id);
1965 err_out:
1966
1967         return ERR_PTR(ret);
1968 }
1969
1970 static int rtrs_srv_rdma_init(struct rtrs_srv_ctx *ctx, u16 port)
1971 {
1972         struct sockaddr_in6 sin = {
1973                 .sin6_family    = AF_INET6,
1974                 .sin6_addr      = IN6ADDR_ANY_INIT,
1975                 .sin6_port      = htons(port),
1976         };
1977         struct sockaddr_ib sib = {
1978                 .sib_family                     = AF_IB,
1979                 .sib_sid        = cpu_to_be64(RDMA_IB_IP_PS_IB | port),
1980                 .sib_sid_mask   = cpu_to_be64(0xffffffffffffffffULL),
1981                 .sib_pkey       = cpu_to_be16(0xffff),
1982         };
1983         struct rdma_cm_id *cm_ip, *cm_ib;
1984         int ret;
1985
1986         /*
1987          * We accept both IPoIB and IB connections, so we need to keep
1988          * two cm id's, one for each socket type and port space.
1989          * If the cm initialization of one of the id's fails, we abort
1990          * everything.
1991          */
1992         cm_ip = rtrs_srv_cm_init(ctx, (struct sockaddr *)&sin, RDMA_PS_TCP);
1993         if (IS_ERR(cm_ip))
1994                 return PTR_ERR(cm_ip);
1995
1996         cm_ib = rtrs_srv_cm_init(ctx, (struct sockaddr *)&sib, RDMA_PS_IB);
1997         if (IS_ERR(cm_ib)) {
1998                 ret = PTR_ERR(cm_ib);
1999                 goto free_cm_ip;
2000         }
2001
2002         ctx->cm_id_ip = cm_ip;
2003         ctx->cm_id_ib = cm_ib;
2004
2005         return 0;
2006
2007 free_cm_ip:
2008         rdma_destroy_id(cm_ip);
2009
2010         return ret;
2011 }
2012
2013 static struct rtrs_srv_ctx *alloc_srv_ctx(struct rtrs_srv_ops *ops)
2014 {
2015         struct rtrs_srv_ctx *ctx;
2016
2017         ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
2018         if (!ctx)
2019                 return NULL;
2020
2021         ctx->ops = *ops;
2022         mutex_init(&ctx->srv_mutex);
2023         INIT_LIST_HEAD(&ctx->srv_list);
2024
2025         return ctx;
2026 }
2027
2028 static void free_srv_ctx(struct rtrs_srv_ctx *ctx)
2029 {
2030         WARN_ON(!list_empty(&ctx->srv_list));
2031         mutex_destroy(&ctx->srv_mutex);
2032         kfree(ctx);
2033 }
2034
2035 /**
2036  * rtrs_srv_open() - open RTRS server context
2037  * @ops:                callback functions
2038  * @port:               port to listen on
2039  *
2040  * Creates server context with specified callbacks.
2041  *
2042  * Return a valid pointer on success otherwise PTR_ERR.
2043  */
2044 struct rtrs_srv_ctx *rtrs_srv_open(struct rtrs_srv_ops *ops, u16 port)
2045 {
2046         struct rtrs_srv_ctx *ctx;
2047         int err;
2048
2049         ctx = alloc_srv_ctx(ops);
2050         if (!ctx)
2051                 return ERR_PTR(-ENOMEM);
2052
2053         err = rtrs_srv_rdma_init(ctx, port);
2054         if (err) {
2055                 free_srv_ctx(ctx);
2056                 return ERR_PTR(err);
2057         }
2058
2059         return ctx;
2060 }
2061 EXPORT_SYMBOL(rtrs_srv_open);
2062
2063 static void close_sessions(struct rtrs_srv *srv)
2064 {
2065         struct rtrs_srv_sess *sess;
2066
2067         mutex_lock(&srv->paths_mutex);
2068         list_for_each_entry(sess, &srv->paths_list, s.entry)
2069                 close_sess(sess);
2070         mutex_unlock(&srv->paths_mutex);
2071 }
2072
2073 static void close_ctx(struct rtrs_srv_ctx *ctx)
2074 {
2075         struct rtrs_srv *srv;
2076
2077         mutex_lock(&ctx->srv_mutex);
2078         list_for_each_entry(srv, &ctx->srv_list, ctx_list)
2079                 close_sessions(srv);
2080         mutex_unlock(&ctx->srv_mutex);
2081         flush_workqueue(rtrs_wq);
2082 }
2083
2084 /**
2085  * rtrs_srv_close() - close RTRS server context
2086  * @ctx: pointer to server context
2087  *
2088  * Closes RTRS server context with all client sessions.
2089  */
2090 void rtrs_srv_close(struct rtrs_srv_ctx *ctx)
2091 {
2092         rdma_destroy_id(ctx->cm_id_ip);
2093         rdma_destroy_id(ctx->cm_id_ib);
2094         close_ctx(ctx);
2095         free_srv_ctx(ctx);
2096 }
2097 EXPORT_SYMBOL(rtrs_srv_close);
2098
2099 static int check_module_params(void)
2100 {
2101         if (sess_queue_depth < 1 || sess_queue_depth > MAX_SESS_QUEUE_DEPTH) {
2102                 pr_err("Invalid sess_queue_depth value %d, has to be >= %d, <= %d.\n",
2103                        sess_queue_depth, 1, MAX_SESS_QUEUE_DEPTH);
2104                 return -EINVAL;
2105         }
2106         if (max_chunk_size < 4096 || !is_power_of_2(max_chunk_size)) {
2107                 pr_err("Invalid max_chunk_size value %d, has to be >= %d and should be power of two.\n",
2108                        max_chunk_size, 4096);
2109                 return -EINVAL;
2110         }
2111
2112         /*
2113          * Check if IB immediate data size is enough to hold the mem_id and the
2114          * offset inside the memory chunk
2115          */
2116         if ((ilog2(sess_queue_depth - 1) + 1) +
2117             (ilog2(max_chunk_size - 1) + 1) > MAX_IMM_PAYL_BITS) {
2118                 pr_err("RDMA immediate size (%db) not enough to encode %d buffers of size %dB. Reduce 'sess_queue_depth' or 'max_chunk_size' parameters.\n",
2119                        MAX_IMM_PAYL_BITS, sess_queue_depth, max_chunk_size);
2120                 return -EINVAL;
2121         }
2122
2123         return 0;
2124 }
2125
2126 static int __init rtrs_server_init(void)
2127 {
2128         int err;
2129
2130         pr_info("Loading module %s, proto %s: (max_chunk_size: %d (pure IO %ld, headers %ld) , sess_queue_depth: %d, always_invalidate: %d)\n",
2131                 KBUILD_MODNAME, RTRS_PROTO_VER_STRING,
2132                 max_chunk_size, max_chunk_size - MAX_HDR_SIZE, MAX_HDR_SIZE,
2133                 sess_queue_depth, always_invalidate);
2134
2135         rtrs_rdma_dev_pd_init(0, &dev_pd);
2136
2137         err = check_module_params();
2138         if (err) {
2139                 pr_err("Failed to load module, invalid module parameters, err: %d\n",
2140                        err);
2141                 return err;
2142         }
2143         chunk_pool = mempool_create_page_pool(sess_queue_depth * CHUNK_POOL_SZ,
2144                                               get_order(max_chunk_size));
2145         if (!chunk_pool)
2146                 return -ENOMEM;
2147         rtrs_dev_class = class_create(THIS_MODULE, "rtrs-server");
2148         if (IS_ERR(rtrs_dev_class)) {
2149                 err = PTR_ERR(rtrs_dev_class);
2150                 goto out_chunk_pool;
2151         }
2152         rtrs_wq = alloc_workqueue("rtrs_server_wq", WQ_MEM_RECLAIM, 0);
2153         if (!rtrs_wq)
2154                 goto out_dev_class;
2155
2156         return 0;
2157
2158 out_dev_class:
2159         class_destroy(rtrs_dev_class);
2160 out_chunk_pool:
2161         mempool_destroy(chunk_pool);
2162
2163         return err;
2164 }
2165
2166 static void __exit rtrs_server_exit(void)
2167 {
2168         destroy_workqueue(rtrs_wq);
2169         class_destroy(rtrs_dev_class);
2170         mempool_destroy(chunk_pool);
2171         rtrs_rdma_dev_pd_deinit(&dev_pd);
2172 }
2173
2174 module_init(rtrs_server_init);
2175 module_exit(rtrs_server_exit);