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