rtrs-srv.c

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