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