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