verbs.c 34.3 KB
Newer Older
1
/*
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37
 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
 *
 * This software is available to you under a choice of one of two
 * licenses.  You may choose to be licensed under the terms of the GNU
 * General Public License (GPL) Version 2, available from the file
 * COPYING in the main directory of this source tree, or the BSD-type
 * license below:
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 *      Redistributions of source code must retain the above copyright
 *      notice, this list of conditions and the following disclaimer.
 *
 *      Redistributions in binary form must reproduce the above
 *      copyright notice, this list of conditions and the following
 *      disclaimer in the documentation and/or other materials provided
 *      with the distribution.
 *
 *      Neither the name of the Network Appliance, Inc. nor the names of
 *      its contributors may be used to endorse or promote products
 *      derived from this software without specific prior written
 *      permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
38 39
 */

40 41 42 43 44 45 46 47 48 49
/*
 * verbs.c
 *
 * Encapsulates the major functions managing:
 *  o adapters
 *  o endpoints
 *  o connections
 *  o buffer memory
 */

50
#include <linux/interrupt.h>
51
#include <linux/slab.h>
52
#include <linux/prefetch.h>
53
#include <linux/sunrpc/addr.h>
54
#include <linux/sunrpc/svc_rdma.h>
55
#include <asm/bitops.h>
56
#include <linux/module.h> /* try_module_get()/module_put() */
57
#include <rdma/ib_cm.h>
58

59 60
#include "xprt_rdma.h"

61 62 63 64
/*
 * Globals/Macros
 */

J
Jeff Layton 已提交
65
#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
66 67 68 69 70 71 72
# define RPCDBG_FACILITY	RPCDBG_TRANS
#endif

/*
 * internal functions
 */

73
static struct workqueue_struct *rpcrdma_receive_wq;
74

75 76
int
rpcrdma_alloc_wq(void)
77
{
78
	struct workqueue_struct *recv_wq;
79

80 81 82 83 84
	recv_wq = alloc_workqueue("xprtrdma_receive",
				  WQ_MEM_RECLAIM | WQ_UNBOUND | WQ_HIGHPRI,
				  0);
	if (!recv_wq)
		return -ENOMEM;
85

86 87
	rpcrdma_receive_wq = recv_wq;
	return 0;
88 89
}

90 91
void
rpcrdma_destroy_wq(void)
92
{
93
	struct workqueue_struct *wq;
94

95 96 97 98 99
	if (rpcrdma_receive_wq) {
		wq = rpcrdma_receive_wq;
		rpcrdma_receive_wq = NULL;
		destroy_workqueue(wq);
	}
100 101
}

102 103 104 105 106
static void
rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context)
{
	struct rpcrdma_ep *ep = context;

107 108 109
	pr_err("rpcrdma: %s on device %s ep %p\n",
	       ib_event_msg(event->event), event->device->name, context);

110 111
	if (ep->rep_connected == 1) {
		ep->rep_connected = -EIO;
112
		rpcrdma_conn_func(ep);
113 114 115 116
		wake_up_all(&ep->rep_connect_wait);
	}
}

117 118 119 120 121
/**
 * rpcrdma_wc_send - Invoked by RDMA provider for each polled Send WC
 * @cq:	completion queue (ignored)
 * @wc:	completed WR
 *
122 123
 */
static void
124
rpcrdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
125
{
126 127 128 129 130
	/* WARNING: Only wr_cqe and status are reliable at this point */
	if (wc->status != IB_WC_SUCCESS && wc->status != IB_WC_WR_FLUSH_ERR)
		pr_err("rpcrdma: Send: %s (%u/0x%x)\n",
		       ib_wc_status_msg(wc->status),
		       wc->status, wc->vendor_err);
131
}
132

133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154
/* Perform basic sanity checking to avoid using garbage
 * to update the credit grant value.
 */
static void
rpcrdma_update_granted_credits(struct rpcrdma_rep *rep)
{
	struct rpcrdma_msg *rmsgp = rdmab_to_msg(rep->rr_rdmabuf);
	struct rpcrdma_buffer *buffer = &rep->rr_rxprt->rx_buf;
	u32 credits;

	if (rep->rr_len < RPCRDMA_HDRLEN_ERR)
		return;

	credits = be32_to_cpu(rmsgp->rm_credit);
	if (credits == 0)
		credits = 1;	/* don't deadlock */
	else if (credits > buffer->rb_max_requests)
		credits = buffer->rb_max_requests;

	atomic_set(&buffer->rb_credits, credits);
}

155
/**
156
 * rpcrdma_wc_receive - Invoked by RDMA provider for each polled Receive WC
157 158 159 160
 * @cq:	completion queue (ignored)
 * @wc:	completed WR
 *
 */
161
static void
162
rpcrdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
163
{
164 165 166
	struct ib_cqe *cqe = wc->wr_cqe;
	struct rpcrdma_rep *rep = container_of(cqe, struct rpcrdma_rep,
					       rr_cqe);
167

168 169 170
	/* WARNING: Only wr_id and status are reliable at this point */
	if (wc->status != IB_WC_SUCCESS)
		goto out_fail;
171

172
	/* status == SUCCESS means all fields in wc are trustworthy */
173 174 175
	if (wc->opcode != IB_WC_RECV)
		return;

176 177 178
	dprintk("RPC:       %s: rep %p opcode 'recv', length %u: success\n",
		__func__, rep, wc->byte_len);

179
	rep->rr_len = wc->byte_len;
180 181 182
	rep->rr_wc_flags = wc->wc_flags;
	rep->rr_inv_rkey = wc->ex.invalidate_rkey;

183
	ib_dma_sync_single_for_cpu(rep->rr_device,
184 185
				   rdmab_addr(rep->rr_rdmabuf),
				   rep->rr_len, DMA_FROM_DEVICE);
186 187

	rpcrdma_update_granted_credits(rep);
188 189

out_schedule:
190
	queue_work(rpcrdma_receive_wq, &rep->rr_work);
191
	return;
192

193 194
out_fail:
	if (wc->status != IB_WC_WR_FLUSH_ERR)
195 196 197
		pr_err("rpcrdma: Recv: %s (%u/0x%x)\n",
		       ib_wc_status_msg(wc->status),
		       wc->status, wc->vendor_err);
198
	rep->rr_len = RPCRDMA_BAD_LEN;
199
	goto out_schedule;
200 201
}

202 203 204 205 206 207 208 209
static void
rpcrdma_update_connect_private(struct rpcrdma_xprt *r_xprt,
			       struct rdma_conn_param *param)
{
	struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
	const struct rpcrdma_connect_private *pmsg = param->private_data;
	unsigned int rsize, wsize;

210 211
	/* Default settings for RPC-over-RDMA Version One */
	r_xprt->rx_ia.ri_reminv_expected = false;
212
	r_xprt->rx_ia.ri_implicit_roundup = xprt_rdma_pad_optimize;
213 214 215 216 217 218
	rsize = RPCRDMA_V1_DEF_INLINE_SIZE;
	wsize = RPCRDMA_V1_DEF_INLINE_SIZE;

	if (pmsg &&
	    pmsg->cp_magic == rpcrdma_cmp_magic &&
	    pmsg->cp_version == RPCRDMA_CMP_VERSION) {
219
		r_xprt->rx_ia.ri_reminv_expected = true;
220
		r_xprt->rx_ia.ri_implicit_roundup = true;
221 222 223 224 225 226 227 228
		rsize = rpcrdma_decode_buffer_size(pmsg->cp_send_size);
		wsize = rpcrdma_decode_buffer_size(pmsg->cp_recv_size);
	}

	if (rsize < cdata->inline_rsize)
		cdata->inline_rsize = rsize;
	if (wsize < cdata->inline_wsize)
		cdata->inline_wsize = wsize;
229 230
	dprintk("RPC:       %s: max send %u, max recv %u\n",
		__func__, cdata->inline_wsize, cdata->inline_rsize);
231 232 233
	rpcrdma_set_max_header_sizes(r_xprt);
}

234 235 236 237 238 239
static int
rpcrdma_conn_upcall(struct rdma_cm_id *id, struct rdma_cm_event *event)
{
	struct rpcrdma_xprt *xprt = id->context;
	struct rpcrdma_ia *ia = &xprt->rx_ia;
	struct rpcrdma_ep *ep = &xprt->rx_ep;
J
Jeff Layton 已提交
240
#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
241
	struct sockaddr *sap = (struct sockaddr *)&ep->rep_remote_addr;
242
#endif
243 244
	struct ib_qp_attr *attr = &ia->ri_qp_attr;
	struct ib_qp_init_attr *iattr = &ia->ri_qp_init_attr;
245 246 247 248 249
	int connstate = 0;

	switch (event->event) {
	case RDMA_CM_EVENT_ADDR_RESOLVED:
	case RDMA_CM_EVENT_ROUTE_RESOLVED:
250
		ia->ri_async_rc = 0;
251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266
		complete(&ia->ri_done);
		break;
	case RDMA_CM_EVENT_ADDR_ERROR:
		ia->ri_async_rc = -EHOSTUNREACH;
		dprintk("RPC:       %s: CM address resolution error, ep 0x%p\n",
			__func__, ep);
		complete(&ia->ri_done);
		break;
	case RDMA_CM_EVENT_ROUTE_ERROR:
		ia->ri_async_rc = -ENETUNREACH;
		dprintk("RPC:       %s: CM route resolution error, ep 0x%p\n",
			__func__, ep);
		complete(&ia->ri_done);
		break;
	case RDMA_CM_EVENT_ESTABLISHED:
		connstate = 1;
267 268 269
		ib_query_qp(ia->ri_id->qp, attr,
			    IB_QP_MAX_QP_RD_ATOMIC | IB_QP_MAX_DEST_RD_ATOMIC,
			    iattr);
270 271
		dprintk("RPC:       %s: %d responder resources"
			" (%d initiator)\n",
272 273
			__func__, attr->max_dest_rd_atomic,
			attr->max_rd_atomic);
274
		rpcrdma_update_connect_private(xprt, &event->param.conn);
275 276 277 278 279 280 281 282
		goto connected;
	case RDMA_CM_EVENT_CONNECT_ERROR:
		connstate = -ENOTCONN;
		goto connected;
	case RDMA_CM_EVENT_UNREACHABLE:
		connstate = -ENETDOWN;
		goto connected;
	case RDMA_CM_EVENT_REJECTED:
283 284 285 286 287
#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
		pr_info("rpcrdma: connection to %pIS:%u on %s rejected: %s\n",
			sap, rpc_get_port(sap), ia->ri_device->name,
			rdma_reject_msg(id, event->status));
#endif
288
		connstate = -ECONNREFUSED;
289 290
		if (event->status == IB_CM_REJ_STALE_CONN)
			connstate = -EAGAIN;
291 292 293 294 295 296 297 298 299
		goto connected;
	case RDMA_CM_EVENT_DISCONNECTED:
		connstate = -ECONNABORTED;
		goto connected;
	case RDMA_CM_EVENT_DEVICE_REMOVAL:
		connstate = -ENODEV;
connected:
		dprintk("RPC:       %s: %sconnected\n",
					__func__, connstate > 0 ? "" : "dis");
300
		atomic_set(&xprt->rx_buf.rb_credits, 1);
301
		ep->rep_connected = connstate;
302
		rpcrdma_conn_func(ep);
303
		wake_up_all(&ep->rep_connect_wait);
304
		/*FALLTHROUGH*/
305
	default:
306 307
		dprintk("RPC:       %s: %pIS:%u (ep 0x%p): %s\n",
			__func__, sap, rpc_get_port(sap), ep,
308
			rdma_event_msg(event->event));
309 310 311
		break;
	}

J
Jeff Layton 已提交
312
#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
313
	if (connstate == 1) {
314
		int ird = attr->max_dest_rd_atomic;
315
		int tird = ep->rep_remote_cma.responder_resources;
316

317
		pr_info("rpcrdma: connection to %pIS:%u on %s, memreg '%s', %d credits, %d responders%s\n",
318
			sap, rpc_get_port(sap),
319
			ia->ri_device->name,
320
			ia->ri_ops->ro_displayname,
321 322 323
			xprt->rx_buf.rb_max_requests,
			ird, ird < 4 && ird < tird / 2 ? " (low!)" : "");
	} else if (connstate < 0) {
324 325
		pr_info("rpcrdma: connection to %pIS:%u closed (%d)\n",
			sap, rpc_get_port(sap), connstate);
326 327 328
	}
#endif

329 330 331
	return 0;
}

332 333 334 335 336 337 338 339
static void rpcrdma_destroy_id(struct rdma_cm_id *id)
{
	if (id) {
		module_put(id->device->owner);
		rdma_destroy_id(id);
	}
}

340 341 342 343
static struct rdma_cm_id *
rpcrdma_create_id(struct rpcrdma_xprt *xprt,
			struct rpcrdma_ia *ia, struct sockaddr *addr)
{
344
	unsigned long wtimeout = msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1;
345 346 347
	struct rdma_cm_id *id;
	int rc;

348 349
	init_completion(&ia->ri_done);

350 351
	id = rdma_create_id(&init_net, rpcrdma_conn_upcall, xprt, RDMA_PS_TCP,
			    IB_QPT_RC);
352 353 354 355 356 357 358
	if (IS_ERR(id)) {
		rc = PTR_ERR(id);
		dprintk("RPC:       %s: rdma_create_id() failed %i\n",
			__func__, rc);
		return id;
	}

359
	ia->ri_async_rc = -ETIMEDOUT;
360 361 362 363 364 365
	rc = rdma_resolve_addr(id, NULL, addr, RDMA_RESOLVE_TIMEOUT);
	if (rc) {
		dprintk("RPC:       %s: rdma_resolve_addr() failed %i\n",
			__func__, rc);
		goto out;
	}
366 367 368 369 370 371
	rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout);
	if (rc < 0) {
		dprintk("RPC:       %s: wait() exited: %i\n",
			__func__, rc);
		goto out;
	}
372 373 374 375 376 377 378 379 380 381 382

	/* FIXME:
	 * Until xprtrdma supports DEVICE_REMOVAL, the provider must
	 * be pinned while there are active NFS/RDMA mounts to prevent
	 * hangs and crashes at umount time.
	 */
	if (!ia->ri_async_rc && !try_module_get(id->device->owner)) {
		dprintk("RPC:       %s: Failed to get device module\n",
			__func__);
		ia->ri_async_rc = -ENODEV;
	}
383 384 385 386
	rc = ia->ri_async_rc;
	if (rc)
		goto out;

387
	ia->ri_async_rc = -ETIMEDOUT;
388 389 390 391
	rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
	if (rc) {
		dprintk("RPC:       %s: rdma_resolve_route() failed %i\n",
			__func__, rc);
392
		goto put;
393
	}
394 395 396 397 398 399
	rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout);
	if (rc < 0) {
		dprintk("RPC:       %s: wait() exited: %i\n",
			__func__, rc);
		goto put;
	}
400 401
	rc = ia->ri_async_rc;
	if (rc)
402
		goto put;
403 404

	return id;
405 406
put:
	module_put(id->device->owner);
407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424
out:
	rdma_destroy_id(id);
	return ERR_PTR(rc);
}

/*
 * Exported functions.
 */

/*
 * Open and initialize an Interface Adapter.
 *  o initializes fields of struct rpcrdma_ia, including
 *    interface and provider attributes and protection zone.
 */
int
rpcrdma_ia_open(struct rpcrdma_xprt *xprt, struct sockaddr *addr, int memreg)
{
	struct rpcrdma_ia *ia = &xprt->rx_ia;
425 426
	int rc;

427 428 429 430 431
	ia->ri_id = rpcrdma_create_id(xprt, ia, addr);
	if (IS_ERR(ia->ri_id)) {
		rc = PTR_ERR(ia->ri_id);
		goto out1;
	}
432
	ia->ri_device = ia->ri_id->device;
433

434
	ia->ri_pd = ib_alloc_pd(ia->ri_device, 0);
435 436
	if (IS_ERR(ia->ri_pd)) {
		rc = PTR_ERR(ia->ri_pd);
437
		pr_err("rpcrdma: ib_alloc_pd() returned %d\n", rc);
438 439 440
		goto out2;
	}

441
	switch (memreg) {
442
	case RPCRDMA_FRMR:
443 444 445 446 447
		if (frwr_is_supported(ia)) {
			ia->ri_ops = &rpcrdma_frwr_memreg_ops;
			break;
		}
		/*FALLTHROUGH*/
448
	case RPCRDMA_MTHCAFMR:
449 450 451 452 453
		if (fmr_is_supported(ia)) {
			ia->ri_ops = &rpcrdma_fmr_memreg_ops;
			break;
		}
		/*FALLTHROUGH*/
454
	default:
455 456 457
		pr_err("rpcrdma: Unsupported memory registration mode: %d\n",
		       memreg);
		rc = -EINVAL;
458
		goto out3;
459 460 461
	}

	return 0;
462 463 464 465

out3:
	ib_dealloc_pd(ia->ri_pd);
	ia->ri_pd = NULL;
466
out2:
467
	rpcrdma_destroy_id(ia->ri_id);
468
	ia->ri_id = NULL;
469 470 471 472 473 474 475 476 477 478 479 480 481
out1:
	return rc;
}

/*
 * Clean up/close an IA.
 *   o if event handles and PD have been initialized, free them.
 *   o close the IA
 */
void
rpcrdma_ia_close(struct rpcrdma_ia *ia)
{
	dprintk("RPC:       %s: entering\n", __func__);
482 483 484
	if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) {
		if (ia->ri_id->qp)
			rdma_destroy_qp(ia->ri_id);
485
		rpcrdma_destroy_id(ia->ri_id);
486 487
		ia->ri_id = NULL;
	}
488 489 490

	/* If the pd is still busy, xprtrdma missed freeing a resource */
	if (ia->ri_pd && !IS_ERR(ia->ri_pd))
491
		ib_dealloc_pd(ia->ri_pd);
492 493 494 495 496 497 498
}

/*
 * Create unconnected endpoint.
 */
int
rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia,
499
		  struct rpcrdma_create_data_internal *cdata)
500
{
501
	struct rpcrdma_connect_private *pmsg = &ep->rep_cm_private;
502
	unsigned int max_qp_wr, max_sge;
503
	struct ib_cq *sendcq, *recvcq;
504
	int rc;
505

506 507
	max_sge = min_t(unsigned int, ia->ri_device->attrs.max_sge,
			RPCRDMA_MAX_SEND_SGES);
508 509
	if (max_sge < RPCRDMA_MIN_SEND_SGES) {
		pr_warn("rpcrdma: HCA provides only %d send SGEs\n", max_sge);
510 511
		return -ENOMEM;
	}
512
	ia->ri_max_send_sges = max_sge - RPCRDMA_MIN_SEND_SGES;
513

514
	if (ia->ri_device->attrs.max_qp_wr <= RPCRDMA_BACKWARD_WRS) {
515 516 517 518
		dprintk("RPC:       %s: insufficient wqe's available\n",
			__func__);
		return -ENOMEM;
	}
519
	max_qp_wr = ia->ri_device->attrs.max_qp_wr - RPCRDMA_BACKWARD_WRS - 1;
520

521
	/* check provider's send/recv wr limits */
522 523
	if (cdata->max_requests > max_qp_wr)
		cdata->max_requests = max_qp_wr;
524 525 526 527 528

	ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall;
	ep->rep_attr.qp_context = ep;
	ep->rep_attr.srq = NULL;
	ep->rep_attr.cap.max_send_wr = cdata->max_requests;
529
	ep->rep_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS;
530
	ep->rep_attr.cap.max_send_wr += 1;	/* drain cqe */
C
Chuck Lever 已提交
531 532 533
	rc = ia->ri_ops->ro_open(ia, ep, cdata);
	if (rc)
		return rc;
534
	ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
535
	ep->rep_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS;
536
	ep->rep_attr.cap.max_recv_wr += 1;	/* drain cqe */
537
	ep->rep_attr.cap.max_send_sge = max_sge;
538 539 540 541 542 543 544 545 546 547 548 549 550 551 552
	ep->rep_attr.cap.max_recv_sge = 1;
	ep->rep_attr.cap.max_inline_data = 0;
	ep->rep_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
	ep->rep_attr.qp_type = IB_QPT_RC;
	ep->rep_attr.port_num = ~0;

	dprintk("RPC:       %s: requested max: dtos: send %d recv %d; "
		"iovs: send %d recv %d\n",
		__func__,
		ep->rep_attr.cap.max_send_wr,
		ep->rep_attr.cap.max_recv_wr,
		ep->rep_attr.cap.max_send_sge,
		ep->rep_attr.cap.max_recv_sge);

	/* set trigger for requesting send completion */
553
	ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1;
554 555
	if (ep->rep_cqinit <= 2)
		ep->rep_cqinit = 0;	/* always signal? */
556
	rpcrdma_init_cqcount(ep, 0);
557
	init_waitqueue_head(&ep->rep_connect_wait);
558
	INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker);
559

560 561 562
	sendcq = ib_alloc_cq(ia->ri_device, NULL,
			     ep->rep_attr.cap.max_send_wr + 1,
			     0, IB_POLL_SOFTIRQ);
563 564 565
	if (IS_ERR(sendcq)) {
		rc = PTR_ERR(sendcq);
		dprintk("RPC:       %s: failed to create send CQ: %i\n",
566 567 568 569
			__func__, rc);
		goto out1;
	}

570 571 572
	recvcq = ib_alloc_cq(ia->ri_device, NULL,
			     ep->rep_attr.cap.max_recv_wr + 1,
			     0, IB_POLL_SOFTIRQ);
573 574 575 576 577 578 579 580 581
	if (IS_ERR(recvcq)) {
		rc = PTR_ERR(recvcq);
		dprintk("RPC:       %s: failed to create recv CQ: %i\n",
			__func__, rc);
		goto out2;
	}

	ep->rep_attr.send_cq = sendcq;
	ep->rep_attr.recv_cq = recvcq;
582 583

	/* Initialize cma parameters */
584
	memset(&ep->rep_remote_cma, 0, sizeof(ep->rep_remote_cma));
585

586 587 588
	/* Prepare RDMA-CM private message */
	pmsg->cp_magic = rpcrdma_cmp_magic;
	pmsg->cp_version = RPCRDMA_CMP_VERSION;
589
	pmsg->cp_flags |= ia->ri_ops->ro_send_w_inv_ok;
590 591 592 593
	pmsg->cp_send_size = rpcrdma_encode_buffer_size(cdata->inline_wsize);
	pmsg->cp_recv_size = rpcrdma_encode_buffer_size(cdata->inline_rsize);
	ep->rep_remote_cma.private_data = pmsg;
	ep->rep_remote_cma.private_data_len = sizeof(*pmsg);
594 595

	/* Client offers RDMA Read but does not initiate */
596
	ep->rep_remote_cma.initiator_depth = 0;
597
	if (ia->ri_device->attrs.max_qp_rd_atom > 32)	/* arbitrary but <= 255 */
598 599
		ep->rep_remote_cma.responder_resources = 32;
	else
600
		ep->rep_remote_cma.responder_resources =
601
						ia->ri_device->attrs.max_qp_rd_atom;
602

603 604 605 606 607 608 609 610 611 612
	/* Limit transport retries so client can detect server
	 * GID changes quickly. RPC layer handles re-establishing
	 * transport connection and retransmission.
	 */
	ep->rep_remote_cma.retry_count = 6;

	/* RPC-over-RDMA handles its own flow control. In addition,
	 * make all RNR NAKs visible so we know that RPC-over-RDMA
	 * flow control is working correctly (no NAKs should be seen).
	 */
613 614 615 616 617 618
	ep->rep_remote_cma.flow_control = 0;
	ep->rep_remote_cma.rnr_retry_count = 0;

	return 0;

out2:
619
	ib_free_cq(sendcq);
620 621 622 623 624 625 626 627 628 629 630
out1:
	return rc;
}

/*
 * rpcrdma_ep_destroy
 *
 * Disconnect and destroy endpoint. After this, the only
 * valid operations on the ep are to free it (if dynamically
 * allocated) or re-create it.
 */
631
void
632 633 634 635 636
rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
{
	dprintk("RPC:       %s: entering, connected is %d\n",
		__func__, ep->rep_connected);

637 638
	cancel_delayed_work_sync(&ep->rep_connect_worker);

639
	if (ia->ri_id->qp) {
640
		rpcrdma_ep_disconnect(ep, ia);
641 642
		rdma_destroy_qp(ia->ri_id);
		ia->ri_id->qp = NULL;
643 644
	}

645
	ib_free_cq(ep->rep_attr.recv_cq);
646
	ib_free_cq(ep->rep_attr.send_cq);
647 648 649 650 651 652 653 654
}

/*
 * Connect unconnected endpoint.
 */
int
rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
{
655 656
	struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt,
						   rx_ia);
657
	struct rdma_cm_id *id, *old;
658 659
	struct sockaddr *sap;
	unsigned int extras;
660 661
	int rc = 0;

662
	if (ep->rep_connected != 0) {
663
retry:
664
		dprintk("RPC:       %s: reconnecting...\n", __func__);
665 666

		rpcrdma_ep_disconnect(ep, ia);
667

668 669
		sap = (struct sockaddr *)&r_xprt->rx_data.addr;
		id = rpcrdma_create_id(r_xprt, ia, sap);
670
		if (IS_ERR(id)) {
671
			rc = -EHOSTUNREACH;
672 673 674 675 676 677 678 679 680
			goto out;
		}
		/* TEMP TEMP TEMP - fail if new device:
		 * Deregister/remarshal *all* requests!
		 * Close and recreate adapter, pd, etc!
		 * Re-determine all attributes still sane!
		 * More stuff I haven't thought of!
		 * Rrrgh!
		 */
681
		if (ia->ri_device != id->device) {
682 683
			printk("RPC:       %s: can't reconnect on "
				"different device!\n", __func__);
684
			rpcrdma_destroy_id(id);
685
			rc = -ENETUNREACH;
686 687 688
			goto out;
		}
		/* END TEMP */
689 690 691 692
		rc = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr);
		if (rc) {
			dprintk("RPC:       %s: rdma_create_qp failed %i\n",
				__func__, rc);
693
			rpcrdma_destroy_id(id);
694 695 696
			rc = -ENETUNREACH;
			goto out;
		}
697 698

		old = ia->ri_id;
699
		ia->ri_id = id;
700 701

		rdma_destroy_qp(old);
702
		rpcrdma_destroy_id(old);
703 704 705 706 707 708 709 710 711
	} else {
		dprintk("RPC:       %s: connecting...\n", __func__);
		rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr);
		if (rc) {
			dprintk("RPC:       %s: rdma_create_qp failed %i\n",
				__func__, rc);
			/* do not update ep->rep_connected */
			return -ENETUNREACH;
		}
712 713 714 715 716 717 718 719 720 721 722 723 724
	}

	ep->rep_connected = 0;

	rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma);
	if (rc) {
		dprintk("RPC:       %s: rdma_connect() failed with %i\n",
				__func__, rc);
		goto out;
	}

	wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0);
	if (ep->rep_connected <= 0) {
725
		if (ep->rep_connected == -EAGAIN)
726 727
			goto retry;
		rc = ep->rep_connected;
728
		goto out;
729 730
	}

731 732 733 734 735
	dprintk("RPC:       %s: connected\n", __func__);
	extras = r_xprt->rx_buf.rb_bc_srv_max_requests;
	if (extras)
		rpcrdma_ep_post_extra_recv(r_xprt, extras);

736 737 738 739 740 741 742 743 744 745 746 747 748 749 750
out:
	if (rc)
		ep->rep_connected = rc;
	return rc;
}

/*
 * rpcrdma_ep_disconnect
 *
 * This is separate from destroy to facilitate the ability
 * to reconnect without recreating the endpoint.
 *
 * This call is not reentrant, and must not be made in parallel
 * on the same endpoint.
 */
751
void
752 753 754 755 756 757 758 759 760 761 762 763 764 765 766
rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
{
	int rc;

	rc = rdma_disconnect(ia->ri_id);
	if (!rc) {
		/* returns without wait if not connected */
		wait_event_interruptible(ep->rep_connect_wait,
							ep->rep_connected != 1);
		dprintk("RPC:       %s: after wait, %sconnected\n", __func__,
			(ep->rep_connected == 1) ? "still " : "dis");
	} else {
		dprintk("RPC:       %s: rdma_disconnect %i\n", __func__, rc);
		ep->rep_connected = rc;
	}
767 768

	ib_drain_qp(ia->ri_id->qp);
769 770
}

771 772 773 774 775 776 777 778 779
static void
rpcrdma_mr_recovery_worker(struct work_struct *work)
{
	struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer,
						  rb_recovery_worker.work);
	struct rpcrdma_mw *mw;

	spin_lock(&buf->rb_recovery_lock);
	while (!list_empty(&buf->rb_stale_mrs)) {
780
		mw = rpcrdma_pop_mw(&buf->rb_stale_mrs);
781 782 783 784 785 786
		spin_unlock(&buf->rb_recovery_lock);

		dprintk("RPC:       %s: recovering MR %p\n", __func__, mw);
		mw->mw_xprt->rx_ia.ri_ops->ro_recover_mr(mw);

		spin_lock(&buf->rb_recovery_lock);
787
	}
788 789 790 791 792 793 794 795 796 797
	spin_unlock(&buf->rb_recovery_lock);
}

void
rpcrdma_defer_mr_recovery(struct rpcrdma_mw *mw)
{
	struct rpcrdma_xprt *r_xprt = mw->mw_xprt;
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;

	spin_lock(&buf->rb_recovery_lock);
798
	rpcrdma_push_mw(mw, &buf->rb_stale_mrs);
799 800 801 802 803
	spin_unlock(&buf->rb_recovery_lock);

	schedule_delayed_work(&buf->rb_recovery_worker, 0);
}

C
Chuck Lever 已提交
804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852
static void
rpcrdma_create_mrs(struct rpcrdma_xprt *r_xprt)
{
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
	struct rpcrdma_ia *ia = &r_xprt->rx_ia;
	unsigned int count;
	LIST_HEAD(free);
	LIST_HEAD(all);

	for (count = 0; count < 32; count++) {
		struct rpcrdma_mw *mw;
		int rc;

		mw = kzalloc(sizeof(*mw), GFP_KERNEL);
		if (!mw)
			break;

		rc = ia->ri_ops->ro_init_mr(ia, mw);
		if (rc) {
			kfree(mw);
			break;
		}

		mw->mw_xprt = r_xprt;

		list_add(&mw->mw_list, &free);
		list_add(&mw->mw_all, &all);
	}

	spin_lock(&buf->rb_mwlock);
	list_splice(&free, &buf->rb_mws);
	list_splice(&all, &buf->rb_all);
	r_xprt->rx_stats.mrs_allocated += count;
	spin_unlock(&buf->rb_mwlock);

	dprintk("RPC:       %s: created %u MRs\n", __func__, count);
}

static void
rpcrdma_mr_refresh_worker(struct work_struct *work)
{
	struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer,
						  rb_refresh_worker.work);
	struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt,
						   rx_buf);

	rpcrdma_create_mrs(r_xprt);
}

853
struct rpcrdma_req *
854 855
rpcrdma_create_req(struct rpcrdma_xprt *r_xprt)
{
856
	struct rpcrdma_buffer *buffer = &r_xprt->rx_buf;
857 858
	struct rpcrdma_req *req;

859
	req = kzalloc(sizeof(*req), GFP_KERNEL);
860
	if (req == NULL)
861
		return ERR_PTR(-ENOMEM);
862

863 864 865 866
	INIT_LIST_HEAD(&req->rl_free);
	spin_lock(&buffer->rb_reqslock);
	list_add(&req->rl_all, &buffer->rb_allreqs);
	spin_unlock(&buffer->rb_reqslock);
867
	req->rl_cqe.done = rpcrdma_wc_send;
868
	req->rl_buffer = &r_xprt->rx_buf;
869
	INIT_LIST_HEAD(&req->rl_registered);
870 871
	req->rl_send_wr.next = NULL;
	req->rl_send_wr.wr_cqe = &req->rl_cqe;
872
	req->rl_send_wr.sg_list = req->rl_send_sge;
873
	req->rl_send_wr.opcode = IB_WR_SEND;
874 875 876
	return req;
}

877
struct rpcrdma_rep *
878 879 880 881 882 883 884 885
rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt)
{
	struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
	struct rpcrdma_ia *ia = &r_xprt->rx_ia;
	struct rpcrdma_rep *rep;
	int rc;

	rc = -ENOMEM;
886
	rep = kzalloc(sizeof(*rep), GFP_KERNEL);
887 888 889
	if (rep == NULL)
		goto out;

890
	rep->rr_rdmabuf = rpcrdma_alloc_regbuf(cdata->inline_rsize,
891
					       DMA_FROM_DEVICE, GFP_KERNEL);
892 893
	if (IS_ERR(rep->rr_rdmabuf)) {
		rc = PTR_ERR(rep->rr_rdmabuf);
894
		goto out_free;
895
	}
896

897
	rep->rr_device = ia->ri_device;
898
	rep->rr_cqe.done = rpcrdma_wc_receive;
899
	rep->rr_rxprt = r_xprt;
900
	INIT_WORK(&rep->rr_work, rpcrdma_reply_handler);
901 902 903 904
	rep->rr_recv_wr.next = NULL;
	rep->rr_recv_wr.wr_cqe = &rep->rr_cqe;
	rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;
	rep->rr_recv_wr.num_sge = 1;
905 906 907 908 909 910 911 912
	return rep;

out_free:
	kfree(rep);
out:
	return ERR_PTR(rc);
}

913
int
914
rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
915
{
916
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
917 918
	int i, rc;

919
	buf->rb_max_requests = r_xprt->rx_data.max_requests;
920
	buf->rb_bc_srv_max_requests = 0;
921
	atomic_set(&buf->rb_credits, 1);
C
Chuck Lever 已提交
922
	spin_lock_init(&buf->rb_mwlock);
923 924
	spin_lock_init(&buf->rb_lock);
	spin_lock_init(&buf->rb_recovery_lock);
C
Chuck Lever 已提交
925 926
	INIT_LIST_HEAD(&buf->rb_mws);
	INIT_LIST_HEAD(&buf->rb_all);
927
	INIT_LIST_HEAD(&buf->rb_stale_mrs);
C
Chuck Lever 已提交
928 929
	INIT_DELAYED_WORK(&buf->rb_refresh_worker,
			  rpcrdma_mr_refresh_worker);
930 931
	INIT_DELAYED_WORK(&buf->rb_recovery_worker,
			  rpcrdma_mr_recovery_worker);
932

C
Chuck Lever 已提交
933
	rpcrdma_create_mrs(r_xprt);
934

935
	INIT_LIST_HEAD(&buf->rb_send_bufs);
936 937
	INIT_LIST_HEAD(&buf->rb_allreqs);
	spin_lock_init(&buf->rb_reqslock);
938 939 940
	for (i = 0; i < buf->rb_max_requests; i++) {
		struct rpcrdma_req *req;

941 942
		req = rpcrdma_create_req(r_xprt);
		if (IS_ERR(req)) {
943 944
			dprintk("RPC:       %s: request buffer %d alloc"
				" failed\n", __func__, i);
945
			rc = PTR_ERR(req);
946 947
			goto out;
		}
948
		req->rl_backchannel = false;
949 950 951 952
		list_add(&req->rl_free, &buf->rb_send_bufs);
	}

	INIT_LIST_HEAD(&buf->rb_recv_bufs);
953
	for (i = 0; i < buf->rb_max_requests + RPCRDMA_MAX_BC_REQUESTS; i++) {
954
		struct rpcrdma_rep *rep;
955

956 957
		rep = rpcrdma_create_rep(r_xprt);
		if (IS_ERR(rep)) {
958 959
			dprintk("RPC:       %s: reply buffer %d alloc failed\n",
				__func__, i);
960
			rc = PTR_ERR(rep);
961 962
			goto out;
		}
963
		list_add(&rep->rr_list, &buf->rb_recv_bufs);
964
	}
965

966 967 968 969 970 971
	return 0;
out:
	rpcrdma_buffer_destroy(buf);
	return rc;
}

972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993
static struct rpcrdma_req *
rpcrdma_buffer_get_req_locked(struct rpcrdma_buffer *buf)
{
	struct rpcrdma_req *req;

	req = list_first_entry(&buf->rb_send_bufs,
			       struct rpcrdma_req, rl_free);
	list_del(&req->rl_free);
	return req;
}

static struct rpcrdma_rep *
rpcrdma_buffer_get_rep_locked(struct rpcrdma_buffer *buf)
{
	struct rpcrdma_rep *rep;

	rep = list_first_entry(&buf->rb_recv_bufs,
			       struct rpcrdma_rep, rr_list);
	list_del(&rep->rr_list);
	return rep;
}

994
static void
995
rpcrdma_destroy_rep(struct rpcrdma_rep *rep)
996
{
997
	rpcrdma_free_regbuf(rep->rr_rdmabuf);
998 999 1000
	kfree(rep);
}

1001
void
1002
rpcrdma_destroy_req(struct rpcrdma_req *req)
1003
{
1004 1005 1006
	rpcrdma_free_regbuf(req->rl_recvbuf);
	rpcrdma_free_regbuf(req->rl_sendbuf);
	rpcrdma_free_regbuf(req->rl_rdmabuf);
1007 1008 1009
	kfree(req);
}

C
Chuck Lever 已提交
1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035
static void
rpcrdma_destroy_mrs(struct rpcrdma_buffer *buf)
{
	struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt,
						   rx_buf);
	struct rpcrdma_ia *ia = rdmab_to_ia(buf);
	struct rpcrdma_mw *mw;
	unsigned int count;

	count = 0;
	spin_lock(&buf->rb_mwlock);
	while (!list_empty(&buf->rb_all)) {
		mw = list_entry(buf->rb_all.next, struct rpcrdma_mw, mw_all);
		list_del(&mw->mw_all);

		spin_unlock(&buf->rb_mwlock);
		ia->ri_ops->ro_release_mr(mw);
		count++;
		spin_lock(&buf->rb_mwlock);
	}
	spin_unlock(&buf->rb_mwlock);
	r_xprt->rx_stats.mrs_allocated = 0;

	dprintk("RPC:       %s: released %u MRs\n", __func__, count);
}

1036 1037 1038
void
rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf)
{
1039
	cancel_delayed_work_sync(&buf->rb_recovery_worker);
1040
	cancel_delayed_work_sync(&buf->rb_refresh_worker);
1041

1042 1043
	while (!list_empty(&buf->rb_recv_bufs)) {
		struct rpcrdma_rep *rep;
1044

1045
		rep = rpcrdma_buffer_get_rep_locked(buf);
1046
		rpcrdma_destroy_rep(rep);
1047
	}
1048
	buf->rb_send_count = 0;
1049

1050 1051
	spin_lock(&buf->rb_reqslock);
	while (!list_empty(&buf->rb_allreqs)) {
1052
		struct rpcrdma_req *req;
A
Allen Andrews 已提交
1053

1054 1055 1056 1057 1058
		req = list_first_entry(&buf->rb_allreqs,
				       struct rpcrdma_req, rl_all);
		list_del(&req->rl_all);

		spin_unlock(&buf->rb_reqslock);
1059
		rpcrdma_destroy_req(req);
1060
		spin_lock(&buf->rb_reqslock);
1061
	}
1062
	spin_unlock(&buf->rb_reqslock);
1063
	buf->rb_recv_count = 0;
A
Allen Andrews 已提交
1064

C
Chuck Lever 已提交
1065
	rpcrdma_destroy_mrs(buf);
1066 1067
}

1068 1069
struct rpcrdma_mw *
rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt)
1070
{
1071 1072 1073
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
	struct rpcrdma_mw *mw = NULL;

C
Chuck Lever 已提交
1074
	spin_lock(&buf->rb_mwlock);
1075 1076
	if (!list_empty(&buf->rb_mws))
		mw = rpcrdma_pop_mw(&buf->rb_mws);
C
Chuck Lever 已提交
1077
	spin_unlock(&buf->rb_mwlock);
1078 1079

	if (!mw)
C
Chuck Lever 已提交
1080
		goto out_nomws;
1081
	return mw;
C
Chuck Lever 已提交
1082 1083 1084 1085 1086 1087 1088 1089 1090

out_nomws:
	dprintk("RPC:       %s: no MWs available\n", __func__);
	schedule_delayed_work(&buf->rb_refresh_worker, 0);

	/* Allow the reply handler and refresh worker to run */
	cond_resched();

	return NULL;
1091 1092
}

1093 1094
void
rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw)
1095
{
1096
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1097

C
Chuck Lever 已提交
1098
	spin_lock(&buf->rb_mwlock);
1099
	rpcrdma_push_mw(mw, &buf->rb_mws);
C
Chuck Lever 已提交
1100
	spin_unlock(&buf->rb_mwlock);
1101 1102
}

1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119
static struct rpcrdma_rep *
rpcrdma_buffer_get_rep(struct rpcrdma_buffer *buffers)
{
	/* If an RPC previously completed without a reply (say, a
	 * credential problem or a soft timeout occurs) then hold off
	 * on supplying more Receive buffers until the number of new
	 * pending RPCs catches up to the number of posted Receives.
	 */
	if (unlikely(buffers->rb_send_count < buffers->rb_recv_count))
		return NULL;

	if (unlikely(list_empty(&buffers->rb_recv_bufs)))
		return NULL;
	buffers->rb_recv_count++;
	return rpcrdma_buffer_get_rep_locked(buffers);
}

1120 1121
/*
 * Get a set of request/reply buffers.
1122 1123
 *
 * Reply buffer (if available) is attached to send buffer upon return.
1124 1125 1126 1127 1128
 */
struct rpcrdma_req *
rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
{
	struct rpcrdma_req *req;
1129

1130
	spin_lock(&buffers->rb_lock);
1131 1132
	if (list_empty(&buffers->rb_send_bufs))
		goto out_reqbuf;
1133
	buffers->rb_send_count++;
1134
	req = rpcrdma_buffer_get_req_locked(buffers);
1135
	req->rl_reply = rpcrdma_buffer_get_rep(buffers);
1136
	spin_unlock(&buffers->rb_lock);
1137
	return req;
1138

1139
out_reqbuf:
1140
	spin_unlock(&buffers->rb_lock);
1141
	pr_warn("RPC:       %s: out of request buffers\n", __func__);
1142
	return NULL;
1143 1144 1145 1146 1147 1148 1149 1150 1151 1152
}

/*
 * Put request/reply buffers back into pool.
 * Pre-decrement counter/array index.
 */
void
rpcrdma_buffer_put(struct rpcrdma_req *req)
{
	struct rpcrdma_buffer *buffers = req->rl_buffer;
1153
	struct rpcrdma_rep *rep = req->rl_reply;
1154

1155
	req->rl_send_wr.num_sge = 0;
1156 1157
	req->rl_reply = NULL;

1158
	spin_lock(&buffers->rb_lock);
1159
	buffers->rb_send_count--;
1160
	list_add_tail(&req->rl_free, &buffers->rb_send_bufs);
1161 1162
	if (rep) {
		buffers->rb_recv_count--;
1163
		list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);
1164
	}
1165
	spin_unlock(&buffers->rb_lock);
1166 1167 1168 1169
}

/*
 * Recover reply buffers from pool.
1170
 * This happens when recovering from disconnect.
1171 1172 1173 1174 1175 1176
 */
void
rpcrdma_recv_buffer_get(struct rpcrdma_req *req)
{
	struct rpcrdma_buffer *buffers = req->rl_buffer;

1177
	spin_lock(&buffers->rb_lock);
1178
	req->rl_reply = rpcrdma_buffer_get_rep(buffers);
1179
	spin_unlock(&buffers->rb_lock);
1180 1181 1182 1183
}

/*
 * Put reply buffers back into pool when not attached to
1184
 * request. This happens in error conditions.
1185 1186 1187 1188
 */
void
rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep)
{
1189
	struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf;
1190

1191
	spin_lock(&buffers->rb_lock);
1192
	buffers->rb_recv_count--;
1193
	list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);
1194
	spin_unlock(&buffers->rb_lock);
1195 1196
}

1197
/**
1198
 * rpcrdma_alloc_regbuf - allocate and DMA-map memory for SEND/RECV buffers
1199
 * @size: size of buffer to be allocated, in bytes
1200
 * @direction: direction of data movement
1201 1202
 * @flags: GFP flags
 *
1203 1204
 * Returns an ERR_PTR, or a pointer to a regbuf, a buffer that
 * can be persistently DMA-mapped for I/O.
1205 1206
 *
 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for
1207 1208
 * receiving the payload of RDMA RECV operations. During Long Calls
 * or Replies they may be registered externally via ro_map.
1209 1210
 */
struct rpcrdma_regbuf *
1211 1212
rpcrdma_alloc_regbuf(size_t size, enum dma_data_direction direction,
		     gfp_t flags)
1213 1214 1215 1216 1217
{
	struct rpcrdma_regbuf *rb;

	rb = kmalloc(sizeof(*rb) + size, flags);
	if (rb == NULL)
1218
		return ERR_PTR(-ENOMEM);
1219

1220
	rb->rg_device = NULL;
1221
	rb->rg_direction = direction;
1222
	rb->rg_iov.length = size;
1223 1224

	return rb;
1225
}
1226

1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258
/**
 * __rpcrdma_map_regbuf - DMA-map a regbuf
 * @ia: controlling rpcrdma_ia
 * @rb: regbuf to be mapped
 */
bool
__rpcrdma_dma_map_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb)
{
	if (rb->rg_direction == DMA_NONE)
		return false;

	rb->rg_iov.addr = ib_dma_map_single(ia->ri_device,
					    (void *)rb->rg_base,
					    rdmab_length(rb),
					    rb->rg_direction);
	if (ib_dma_mapping_error(ia->ri_device, rdmab_addr(rb)))
		return false;

	rb->rg_device = ia->ri_device;
	rb->rg_iov.lkey = ia->ri_pd->local_dma_lkey;
	return true;
}

static void
rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb)
{
	if (!rpcrdma_regbuf_is_mapped(rb))
		return;

	ib_dma_unmap_single(rb->rg_device, rdmab_addr(rb),
			    rdmab_length(rb), rb->rg_direction);
	rb->rg_device = NULL;
1259 1260 1261 1262 1263 1264 1265
}

/**
 * rpcrdma_free_regbuf - deregister and free registered buffer
 * @rb: regbuf to be deregistered and freed
 */
void
1266
rpcrdma_free_regbuf(struct rpcrdma_regbuf *rb)
1267
{
1268 1269 1270
	if (!rb)
		return;

1271
	rpcrdma_dma_unmap_regbuf(rb);
1272
	kfree(rb);
1273 1274
}

1275 1276 1277 1278 1279 1280 1281 1282 1283 1284
/*
 * Prepost any receive buffer, then post send.
 *
 * Receive buffer is donated to hardware, reclaimed upon recv completion.
 */
int
rpcrdma_ep_post(struct rpcrdma_ia *ia,
		struct rpcrdma_ep *ep,
		struct rpcrdma_req *req)
{
1285 1286
	struct ib_send_wr *send_wr = &req->rl_send_wr;
	struct ib_send_wr *send_wr_fail;
1287
	int rc;
1288

1289 1290
	if (req->rl_reply) {
		rc = rpcrdma_ep_post_recv(ia, req->rl_reply);
1291
		if (rc)
1292
			return rc;
1293 1294 1295
		req->rl_reply = NULL;
	}

1296
	dprintk("RPC:       %s: posting %d s/g entries\n",
1297
		__func__, send_wr->num_sge);
1298

1299
	rpcrdma_set_signaled(ep, send_wr);
1300
	rc = ib_post_send(ia->ri_id->qp, send_wr, &send_wr_fail);
1301
	if (rc)
1302 1303 1304 1305 1306 1307
		goto out_postsend_err;
	return 0;

out_postsend_err:
	pr_err("rpcrdma: RDMA Send ib_post_send returned %i\n", rc);
	return -ENOTCONN;
1308 1309 1310 1311 1312 1313
}

int
rpcrdma_ep_post_recv(struct rpcrdma_ia *ia,
		     struct rpcrdma_rep *rep)
{
1314
	struct ib_recv_wr *recv_wr_fail;
1315 1316
	int rc;

1317 1318
	if (!rpcrdma_dma_map_regbuf(ia, rep->rr_rdmabuf))
		goto out_map;
1319
	rc = ib_post_recv(ia->ri_id->qp, &rep->rr_recv_wr, &recv_wr_fail);
1320
	if (rc)
1321 1322 1323
		goto out_postrecv;
	return 0;

1324 1325 1326 1327
out_map:
	pr_err("rpcrdma: failed to DMA map the Receive buffer\n");
	return -EIO;

1328 1329 1330
out_postrecv:
	pr_err("rpcrdma: ib_post_recv returned %i\n", rc);
	return -ENOTCONN;
1331
}
1332

1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348
/**
 * rpcrdma_ep_post_extra_recv - Post buffers for incoming backchannel requests
 * @r_xprt: transport associated with these backchannel resources
 * @min_reqs: minimum number of incoming requests expected
 *
 * Returns zero if all requested buffers were posted, or a negative errno.
 */
int
rpcrdma_ep_post_extra_recv(struct rpcrdma_xprt *r_xprt, unsigned int count)
{
	struct rpcrdma_buffer *buffers = &r_xprt->rx_buf;
	struct rpcrdma_ia *ia = &r_xprt->rx_ia;
	struct rpcrdma_rep *rep;
	int rc;

	while (count--) {
1349
		spin_lock(&buffers->rb_lock);
1350 1351 1352
		if (list_empty(&buffers->rb_recv_bufs))
			goto out_reqbuf;
		rep = rpcrdma_buffer_get_rep_locked(buffers);
1353
		spin_unlock(&buffers->rb_lock);
1354

1355
		rc = rpcrdma_ep_post_recv(ia, rep);
1356 1357 1358 1359 1360 1361 1362
		if (rc)
			goto out_rc;
	}

	return 0;

out_reqbuf:
1363
	spin_unlock(&buffers->rb_lock);
1364 1365 1366 1367 1368 1369 1370
	pr_warn("%s: no extra receive buffers\n", __func__);
	return -ENOMEM;

out_rc:
	rpcrdma_recv_buffer_put(rep);
	return rc;
}