verbs.c 37.1 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

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
# define RPCDBG_FACILITY	RPCDBG_TRANS
#endif

/*
 * internal functions
 */
72
static void rpcrdma_create_mrs(struct rpcrdma_xprt *r_xprt);
73 74
static void rpcrdma_destroy_mrs(struct rpcrdma_buffer *buf);
static void rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb);
75

76
static struct workqueue_struct *rpcrdma_receive_wq __read_mostly;
77

78 79
int
rpcrdma_alloc_wq(void)
80
{
81
	struct workqueue_struct *recv_wq;
82

83 84 85 86 87
	recv_wq = alloc_workqueue("xprtrdma_receive",
				  WQ_MEM_RECLAIM | WQ_UNBOUND | WQ_HIGHPRI,
				  0);
	if (!recv_wq)
		return -ENOMEM;
88

89 90
	rpcrdma_receive_wq = recv_wq;
	return 0;
91 92
}

93 94
void
rpcrdma_destroy_wq(void)
95
{
96
	struct workqueue_struct *wq;
97

98 99 100 101 102
	if (rpcrdma_receive_wq) {
		wq = rpcrdma_receive_wq;
		rpcrdma_receive_wq = NULL;
		destroy_workqueue(wq);
	}
103 104
}

105 106 107 108 109
static void
rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context)
{
	struct rpcrdma_ep *ep = context;

110 111 112
	pr_err("rpcrdma: %s on device %s ep %p\n",
	       ib_event_msg(event->event), event->device->name, context);

113 114
	if (ep->rep_connected == 1) {
		ep->rep_connected = -EIO;
115
		rpcrdma_conn_func(ep);
116 117 118 119
		wake_up_all(&ep->rep_connect_wait);
	}
}

120 121 122 123 124
/**
 * rpcrdma_wc_send - Invoked by RDMA provider for each polled Send WC
 * @cq:	completion queue (ignored)
 * @wc:	completed WR
 *
125 126
 */
static void
127
rpcrdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
128
{
129 130 131 132 133
	/* 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);
134
}
135

136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157
/* 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);
}

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

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

175
	/* status == SUCCESS means all fields in wc are trustworthy */
176 177 178
	if (wc->opcode != IB_WC_RECV)
		return;

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

182
	rep->rr_len = wc->byte_len;
183 184 185
	rep->rr_wc_flags = wc->wc_flags;
	rep->rr_inv_rkey = wc->ex.invalidate_rkey;

186
	ib_dma_sync_single_for_cpu(rdmab_device(rep->rr_rdmabuf),
187 188
				   rdmab_addr(rep->rr_rdmabuf),
				   rep->rr_len, DMA_FROM_DEVICE);
189 190

	rpcrdma_update_granted_credits(rep);
191 192

out_schedule:
193
	queue_work(rpcrdma_receive_wq, &rep->rr_work);
194
	return;
195

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

205 206 207 208 209 210 211 212
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;

213 214
	/* Default settings for RPC-over-RDMA Version One */
	r_xprt->rx_ia.ri_reminv_expected = false;
215
	r_xprt->rx_ia.ri_implicit_roundup = xprt_rdma_pad_optimize;
216 217 218 219 220 221
	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) {
222
		r_xprt->rx_ia.ri_reminv_expected = true;
223
		r_xprt->rx_ia.ri_implicit_roundup = true;
224 225 226 227 228 229 230 231
		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;
232 233
	dprintk("RPC:       %s: max send %u, max recv %u\n",
		__func__, cdata->inline_wsize, cdata->inline_rsize);
234 235 236
	rpcrdma_set_max_header_sizes(r_xprt);
}

237 238 239 240 241 242
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 已提交
243
#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
244
	struct sockaddr *sap = (struct sockaddr *)&ep->rep_remote_addr;
245
#endif
246 247
	struct ib_qp_attr *attr = &ia->ri_qp_attr;
	struct ib_qp_init_attr *iattr = &ia->ri_qp_init_attr;
248 249 250 251 252
	int connstate = 0;

	switch (event->event) {
	case RDMA_CM_EVENT_ADDR_RESOLVED:
	case RDMA_CM_EVENT_ROUTE_RESOLVED:
253
		ia->ri_async_rc = 0;
254 255 256 257 258 259 260 261 262 263 264 265 266 267
		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;
268 269 270 271 272 273 274 275 276 277 278 279 280 281 282
	case RDMA_CM_EVENT_DEVICE_REMOVAL:
#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
		pr_info("rpcrdma: removing device for %pIS:%u\n",
			sap, rpc_get_port(sap));
#endif
		set_bit(RPCRDMA_IAF_REMOVING, &ia->ri_flags);
		ep->rep_connected = -ENODEV;
		xprt_force_disconnect(&xprt->rx_xprt);
		wait_for_completion(&ia->ri_remove_done);

		ia->ri_id = NULL;
		ia->ri_pd = NULL;
		ia->ri_device = NULL;
		/* Return 1 to ensure the core destroys the id. */
		return 1;
283 284
	case RDMA_CM_EVENT_ESTABLISHED:
		connstate = 1;
285 286 287
		ib_query_qp(ia->ri_id->qp, attr,
			    IB_QP_MAX_QP_RD_ATOMIC | IB_QP_MAX_DEST_RD_ATOMIC,
			    iattr);
288 289
		dprintk("RPC:       %s: %d responder resources"
			" (%d initiator)\n",
290 291
			__func__, attr->max_dest_rd_atomic,
			attr->max_rd_atomic);
292
		rpcrdma_update_connect_private(xprt, &event->param.conn);
293 294 295 296 297 298 299 300
		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:
301 302 303 304 305
#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
306
		connstate = -ECONNREFUSED;
307 308
		if (event->status == IB_CM_REJ_STALE_CONN)
			connstate = -EAGAIN;
309 310 311 312 313 314
		goto connected;
	case RDMA_CM_EVENT_DISCONNECTED:
		connstate = -ECONNABORTED;
connected:
		dprintk("RPC:       %s: %sconnected\n",
					__func__, connstate > 0 ? "" : "dis");
315
		atomic_set(&xprt->rx_buf.rb_credits, 1);
316
		ep->rep_connected = connstate;
317
		rpcrdma_conn_func(ep);
318
		wake_up_all(&ep->rep_connect_wait);
319
		/*FALLTHROUGH*/
320
	default:
321 322
		dprintk("RPC:       %s: %pIS:%u (ep 0x%p): %s\n",
			__func__, sap, rpc_get_port(sap), ep,
323
			rdma_event_msg(event->event));
324 325 326
		break;
	}

J
Jeff Layton 已提交
327
#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
328
	if (connstate == 1) {
329
		int ird = attr->max_dest_rd_atomic;
330
		int tird = ep->rep_remote_cma.responder_resources;
331

332
		pr_info("rpcrdma: connection to %pIS:%u on %s, memreg '%s', %d credits, %d responders%s\n",
333
			sap, rpc_get_port(sap),
334
			ia->ri_device->name,
335
			ia->ri_ops->ro_displayname,
336 337 338
			xprt->rx_buf.rb_max_requests,
			ird, ird < 4 && ird < tird / 2 ? " (low!)" : "");
	} else if (connstate < 0) {
339 340
		pr_info("rpcrdma: connection to %pIS:%u closed (%d)\n",
			sap, rpc_get_port(sap), connstate);
341 342 343
	}
#endif

344 345 346 347 348 349 350
	return 0;
}

static struct rdma_cm_id *
rpcrdma_create_id(struct rpcrdma_xprt *xprt,
			struct rpcrdma_ia *ia, struct sockaddr *addr)
{
351
	unsigned long wtimeout = msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1;
352 353 354
	struct rdma_cm_id *id;
	int rc;

355
	init_completion(&ia->ri_done);
356
	init_completion(&ia->ri_remove_done);
357

358 359
	id = rdma_create_id(&init_net, rpcrdma_conn_upcall, xprt, RDMA_PS_TCP,
			    IB_QPT_RC);
360 361 362 363 364 365 366
	if (IS_ERR(id)) {
		rc = PTR_ERR(id);
		dprintk("RPC:       %s: rdma_create_id() failed %i\n",
			__func__, rc);
		return id;
	}

367
	ia->ri_async_rc = -ETIMEDOUT;
368 369 370 371 372 373
	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;
	}
374 375 376 377 378 379
	rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout);
	if (rc < 0) {
		dprintk("RPC:       %s: wait() exited: %i\n",
			__func__, rc);
		goto out;
	}
380

381 382 383 384
	rc = ia->ri_async_rc;
	if (rc)
		goto out;

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

	return id;
403

404 405 406 407 408 409 410 411 412
out:
	rdma_destroy_id(id);
	return ERR_PTR(rc);
}

/*
 * Exported functions.
 */

413 414 415 416 417 418 419
/**
 * rpcrdma_ia_open - Open and initialize an Interface Adapter.
 * @xprt: controlling transport
 * @addr: IP address of remote peer
 *
 * Returns 0 on success, negative errno if an appropriate
 * Interface Adapter could not be found and opened.
420 421
 */
int
422
rpcrdma_ia_open(struct rpcrdma_xprt *xprt, struct sockaddr *addr)
423 424
{
	struct rpcrdma_ia *ia = &xprt->rx_ia;
425 426
	int rc;

427 428 429
	ia->ri_id = rpcrdma_create_id(xprt, ia, addr);
	if (IS_ERR(ia->ri_id)) {
		rc = PTR_ERR(ia->ri_id);
430
		goto out_err;
431
	}
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
		goto out_err;
439 440
	}

441
	switch (xprt_rdma_memreg_strategy) {
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
		pr_err("rpcrdma: Device %s does not support memreg mode %d\n",
		       ia->ri_device->name, xprt_rdma_memreg_strategy);
457
		rc = -EINVAL;
458
		goto out_err;
459 460 461
	}

	return 0;
462

463 464
out_err:
	rpcrdma_ia_close(ia);
465 466 467
	return rc;
}

468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517
/**
 * rpcrdma_ia_remove - Handle device driver unload
 * @ia: interface adapter being removed
 *
 * Divest transport H/W resources associated with this adapter,
 * but allow it to be restored later.
 */
void
rpcrdma_ia_remove(struct rpcrdma_ia *ia)
{
	struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt,
						   rx_ia);
	struct rpcrdma_ep *ep = &r_xprt->rx_ep;
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
	struct rpcrdma_req *req;
	struct rpcrdma_rep *rep;

	cancel_delayed_work_sync(&buf->rb_refresh_worker);

	/* This is similar to rpcrdma_ep_destroy, but:
	 * - Don't cancel the connect worker.
	 * - Don't call rpcrdma_ep_disconnect, which waits
	 *   for another conn upcall, which will deadlock.
	 * - rdma_disconnect is unneeded, the underlying
	 *   connection is already gone.
	 */
	if (ia->ri_id->qp) {
		ib_drain_qp(ia->ri_id->qp);
		rdma_destroy_qp(ia->ri_id);
		ia->ri_id->qp = NULL;
	}
	ib_free_cq(ep->rep_attr.recv_cq);
	ib_free_cq(ep->rep_attr.send_cq);

	/* The ULP is responsible for ensuring all DMA
	 * mappings and MRs are gone.
	 */
	list_for_each_entry(rep, &buf->rb_recv_bufs, rr_list)
		rpcrdma_dma_unmap_regbuf(rep->rr_rdmabuf);
	list_for_each_entry(req, &buf->rb_allreqs, rl_all) {
		rpcrdma_dma_unmap_regbuf(req->rl_rdmabuf);
		rpcrdma_dma_unmap_regbuf(req->rl_sendbuf);
		rpcrdma_dma_unmap_regbuf(req->rl_recvbuf);
	}
	rpcrdma_destroy_mrs(buf);

	/* Allow waiters to continue */
	complete(&ia->ri_remove_done);
}

518 519 520 521
/**
 * rpcrdma_ia_close - Clean up/close an IA.
 * @ia: interface adapter to close
 *
522 523 524 525 526
 */
void
rpcrdma_ia_close(struct rpcrdma_ia *ia)
{
	dprintk("RPC:       %s: entering\n", __func__);
527 528 529
	if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) {
		if (ia->ri_id->qp)
			rdma_destroy_qp(ia->ri_id);
530
		rdma_destroy_id(ia->ri_id);
531
	}
532 533
	ia->ri_id = NULL;
	ia->ri_device = NULL;
534 535 536

	/* If the pd is still busy, xprtrdma missed freeing a resource */
	if (ia->ri_pd && !IS_ERR(ia->ri_pd))
537
		ib_dealloc_pd(ia->ri_pd);
538
	ia->ri_pd = NULL;
539 540 541 542 543 544 545
}

/*
 * Create unconnected endpoint.
 */
int
rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia,
546
		  struct rpcrdma_create_data_internal *cdata)
547
{
548
	struct rpcrdma_connect_private *pmsg = &ep->rep_cm_private;
549
	unsigned int max_qp_wr, max_sge;
550
	struct ib_cq *sendcq, *recvcq;
551
	int rc;
552

553 554
	max_sge = min_t(unsigned int, ia->ri_device->attrs.max_sge,
			RPCRDMA_MAX_SEND_SGES);
555 556
	if (max_sge < RPCRDMA_MIN_SEND_SGES) {
		pr_warn("rpcrdma: HCA provides only %d send SGEs\n", max_sge);
557 558
		return -ENOMEM;
	}
559
	ia->ri_max_send_sges = max_sge - RPCRDMA_MIN_SEND_SGES;
560

561
	if (ia->ri_device->attrs.max_qp_wr <= RPCRDMA_BACKWARD_WRS) {
562 563 564 565
		dprintk("RPC:       %s: insufficient wqe's available\n",
			__func__);
		return -ENOMEM;
	}
566
	max_qp_wr = ia->ri_device->attrs.max_qp_wr - RPCRDMA_BACKWARD_WRS - 1;
567

568
	/* check provider's send/recv wr limits */
569 570
	if (cdata->max_requests > max_qp_wr)
		cdata->max_requests = max_qp_wr;
571 572 573 574 575

	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;
576
	ep->rep_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS;
577
	ep->rep_attr.cap.max_send_wr += 1;	/* drain cqe */
C
Chuck Lever 已提交
578 579 580
	rc = ia->ri_ops->ro_open(ia, ep, cdata);
	if (rc)
		return rc;
581
	ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
582
	ep->rep_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS;
583
	ep->rep_attr.cap.max_recv_wr += 1;	/* drain cqe */
584
	ep->rep_attr.cap.max_send_sge = max_sge;
585 586 587 588 589 590 591 592 593 594 595 596 597 598 599
	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 */
600
	ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1;
601 602
	if (ep->rep_cqinit <= 2)
		ep->rep_cqinit = 0;	/* always signal? */
603
	rpcrdma_init_cqcount(ep, 0);
604
	init_waitqueue_head(&ep->rep_connect_wait);
605
	INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker);
606

607 608 609
	sendcq = ib_alloc_cq(ia->ri_device, NULL,
			     ep->rep_attr.cap.max_send_wr + 1,
			     0, IB_POLL_SOFTIRQ);
610 611 612
	if (IS_ERR(sendcq)) {
		rc = PTR_ERR(sendcq);
		dprintk("RPC:       %s: failed to create send CQ: %i\n",
613 614 615 616
			__func__, rc);
		goto out1;
	}

617 618 619
	recvcq = ib_alloc_cq(ia->ri_device, NULL,
			     ep->rep_attr.cap.max_recv_wr + 1,
			     0, IB_POLL_SOFTIRQ);
620 621 622 623 624 625 626 627 628
	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;
629 630

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

633 634 635
	/* Prepare RDMA-CM private message */
	pmsg->cp_magic = rpcrdma_cmp_magic;
	pmsg->cp_version = RPCRDMA_CMP_VERSION;
636
	pmsg->cp_flags |= ia->ri_ops->ro_send_w_inv_ok;
637 638 639 640
	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);
641 642

	/* Client offers RDMA Read but does not initiate */
643
	ep->rep_remote_cma.initiator_depth = 0;
644
	if (ia->ri_device->attrs.max_qp_rd_atom > 32)	/* arbitrary but <= 255 */
645 646
		ep->rep_remote_cma.responder_resources = 32;
	else
647
		ep->rep_remote_cma.responder_resources =
648
						ia->ri_device->attrs.max_qp_rd_atom;
649

650 651 652 653 654 655 656 657 658 659
	/* 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).
	 */
660 661 662 663 664 665
	ep->rep_remote_cma.flow_control = 0;
	ep->rep_remote_cma.rnr_retry_count = 0;

	return 0;

out2:
666
	ib_free_cq(sendcq);
667 668 669 670 671 672 673 674 675 676 677
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.
 */
678
void
679 680 681 682 683
rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
{
	dprintk("RPC:       %s: entering, connected is %d\n",
		__func__, ep->rep_connected);

684 685
	cancel_delayed_work_sync(&ep->rep_connect_worker);

686
	if (ia->ri_id->qp) {
687
		rpcrdma_ep_disconnect(ep, ia);
688 689
		rdma_destroy_qp(ia->ri_id);
		ia->ri_id->qp = NULL;
690 691
	}

692
	ib_free_cq(ep->rep_attr.recv_cq);
693
	ib_free_cq(ep->rep_attr.send_cq);
694 695
}

696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737
/* Re-establish a connection after a device removal event.
 * Unlike a normal reconnection, a fresh PD and a new set
 * of MRs and buffers is needed.
 */
static int
rpcrdma_ep_recreate_xprt(struct rpcrdma_xprt *r_xprt,
			 struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
{
	struct sockaddr *sap = (struct sockaddr *)&r_xprt->rx_data.addr;
	int rc, err;

	pr_info("%s: r_xprt = %p\n", __func__, r_xprt);

	rc = -EHOSTUNREACH;
	if (rpcrdma_ia_open(r_xprt, sap))
		goto out1;

	rc = -ENOMEM;
	err = rpcrdma_ep_create(ep, ia, &r_xprt->rx_data);
	if (err) {
		pr_err("rpcrdma: rpcrdma_ep_create returned %d\n", err);
		goto out2;
	}

	rc = -ENETUNREACH;
	err = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr);
	if (err) {
		pr_err("rpcrdma: rdma_create_qp returned %d\n", err);
		goto out3;
	}

	rpcrdma_create_mrs(r_xprt);
	return 0;

out3:
	rpcrdma_ep_destroy(ep, ia);
out2:
	rpcrdma_ia_close(ia);
out1:
	return rc;
}

738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783
static int
rpcrdma_ep_reconnect(struct rpcrdma_xprt *r_xprt, struct rpcrdma_ep *ep,
		     struct rpcrdma_ia *ia)
{
	struct sockaddr *sap = (struct sockaddr *)&r_xprt->rx_data.addr;
	struct rdma_cm_id *id, *old;
	int err, rc;

	dprintk("RPC:       %s: reconnecting...\n", __func__);

	rpcrdma_ep_disconnect(ep, ia);

	rc = -EHOSTUNREACH;
	id = rpcrdma_create_id(r_xprt, ia, sap);
	if (IS_ERR(id))
		goto out;

	/* As long as the new ID points to the same device as the
	 * old ID, we can reuse the transport's existing PD and all
	 * previously allocated MRs. Also, the same device means
	 * the transport's previous DMA mappings are still valid.
	 *
	 * This is a sanity check only. There should be no way these
	 * point to two different devices here.
	 */
	old = id;
	rc = -ENETUNREACH;
	if (ia->ri_device != id->device) {
		pr_err("rpcrdma: can't reconnect on different device!\n");
		goto out_destroy;
	}

	err = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr);
	if (err) {
		dprintk("RPC:       %s: rdma_create_qp returned %d\n",
			__func__, err);
		goto out_destroy;
	}

	/* Atomically replace the transport's ID and QP. */
	rc = 0;
	old = ia->ri_id;
	ia->ri_id = id;
	rdma_destroy_qp(old);

out_destroy:
784
	rdma_destroy_id(old);
785 786 787 788
out:
	return rc;
}

789 790 791 792 793 794
/*
 * Connect unconnected endpoint.
 */
int
rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
{
795 796 797
	struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt,
						   rx_ia);
	unsigned int extras;
798
	int rc;
799 800

retry:
801 802
	switch (ep->rep_connected) {
	case 0:
803 804 805 806 807
		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);
808 809
			rc = -ENETUNREACH;
			goto out_noupdate;
810
		}
811
		break;
812 813 814 815 816
	case -ENODEV:
		rc = rpcrdma_ep_recreate_xprt(r_xprt, ep, ia);
		if (rc)
			goto out_noupdate;
		break;
817 818 819 820
	default:
		rc = rpcrdma_ep_reconnect(r_xprt, ep, ia);
		if (rc)
			goto out;
821 822 823 824 825 826 827 828 829 830 831 832 833
	}

	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) {
834
		if (ep->rep_connected == -EAGAIN)
835 836
			goto retry;
		rc = ep->rep_connected;
837
		goto out;
838 839
	}

840 841 842 843 844
	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);

845 846 847
out:
	if (rc)
		ep->rep_connected = rc;
848 849

out_noupdate:
850 851 852 853 854 855 856 857 858 859 860 861
	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.
 */
862
void
863 864 865 866 867 868 869 870 871 872 873 874 875 876 877
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;
	}
878 879

	ib_drain_qp(ia->ri_id->qp);
880 881
}

882 883 884 885 886 887 888 889 890
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)) {
891
		mw = rpcrdma_pop_mw(&buf->rb_stale_mrs);
892 893 894 895 896 897
		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);
898
	}
899 900 901 902 903 904 905 906 907 908
	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);
909
	rpcrdma_push_mw(mw, &buf->rb_stale_mrs);
910 911 912 913 914
	spin_unlock(&buf->rb_recovery_lock);

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

C
Chuck Lever 已提交
915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963
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);
}

964
struct rpcrdma_req *
965 966
rpcrdma_create_req(struct rpcrdma_xprt *r_xprt)
{
967
	struct rpcrdma_buffer *buffer = &r_xprt->rx_buf;
968 969
	struct rpcrdma_req *req;

970
	req = kzalloc(sizeof(*req), GFP_KERNEL);
971
	if (req == NULL)
972
		return ERR_PTR(-ENOMEM);
973

974 975 976
	spin_lock(&buffer->rb_reqslock);
	list_add(&req->rl_all, &buffer->rb_allreqs);
	spin_unlock(&buffer->rb_reqslock);
977
	req->rl_cqe.done = rpcrdma_wc_send;
978
	req->rl_buffer = &r_xprt->rx_buf;
979
	INIT_LIST_HEAD(&req->rl_registered);
980 981
	req->rl_send_wr.next = NULL;
	req->rl_send_wr.wr_cqe = &req->rl_cqe;
982
	req->rl_send_wr.sg_list = req->rl_send_sge;
983
	req->rl_send_wr.opcode = IB_WR_SEND;
984 985 986
	return req;
}

987
struct rpcrdma_rep *
988 989 990 991 992 993 994
rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt)
{
	struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
	struct rpcrdma_rep *rep;
	int rc;

	rc = -ENOMEM;
995
	rep = kzalloc(sizeof(*rep), GFP_KERNEL);
996 997 998
	if (rep == NULL)
		goto out;

999
	rep->rr_rdmabuf = rpcrdma_alloc_regbuf(cdata->inline_rsize,
1000
					       DMA_FROM_DEVICE, GFP_KERNEL);
1001 1002
	if (IS_ERR(rep->rr_rdmabuf)) {
		rc = PTR_ERR(rep->rr_rdmabuf);
1003
		goto out_free;
1004
	}
1005

1006
	rep->rr_cqe.done = rpcrdma_wc_receive;
1007
	rep->rr_rxprt = r_xprt;
1008
	INIT_WORK(&rep->rr_work, rpcrdma_reply_handler);
1009 1010 1011 1012
	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;
1013 1014 1015 1016 1017 1018 1019 1020
	return rep;

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

1021
int
1022
rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
1023
{
1024
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1025 1026
	int i, rc;

1027
	buf->rb_max_requests = r_xprt->rx_data.max_requests;
1028
	buf->rb_bc_srv_max_requests = 0;
1029
	atomic_set(&buf->rb_credits, 1);
C
Chuck Lever 已提交
1030
	spin_lock_init(&buf->rb_mwlock);
1031 1032
	spin_lock_init(&buf->rb_lock);
	spin_lock_init(&buf->rb_recovery_lock);
C
Chuck Lever 已提交
1033 1034
	INIT_LIST_HEAD(&buf->rb_mws);
	INIT_LIST_HEAD(&buf->rb_all);
1035
	INIT_LIST_HEAD(&buf->rb_pending);
1036
	INIT_LIST_HEAD(&buf->rb_stale_mrs);
C
Chuck Lever 已提交
1037 1038
	INIT_DELAYED_WORK(&buf->rb_refresh_worker,
			  rpcrdma_mr_refresh_worker);
1039 1040
	INIT_DELAYED_WORK(&buf->rb_recovery_worker,
			  rpcrdma_mr_recovery_worker);
1041

C
Chuck Lever 已提交
1042
	rpcrdma_create_mrs(r_xprt);
1043

1044
	INIT_LIST_HEAD(&buf->rb_send_bufs);
1045 1046
	INIT_LIST_HEAD(&buf->rb_allreqs);
	spin_lock_init(&buf->rb_reqslock);
1047 1048 1049
	for (i = 0; i < buf->rb_max_requests; i++) {
		struct rpcrdma_req *req;

1050 1051
		req = rpcrdma_create_req(r_xprt);
		if (IS_ERR(req)) {
1052 1053
			dprintk("RPC:       %s: request buffer %d alloc"
				" failed\n", __func__, i);
1054
			rc = PTR_ERR(req);
1055 1056
			goto out;
		}
1057
		req->rl_backchannel = false;
1058
		list_add(&req->rl_list, &buf->rb_send_bufs);
1059 1060 1061
	}

	INIT_LIST_HEAD(&buf->rb_recv_bufs);
1062
	for (i = 0; i < buf->rb_max_requests + RPCRDMA_MAX_BC_REQUESTS; i++) {
1063
		struct rpcrdma_rep *rep;
1064

1065 1066
		rep = rpcrdma_create_rep(r_xprt);
		if (IS_ERR(rep)) {
1067 1068
			dprintk("RPC:       %s: reply buffer %d alloc failed\n",
				__func__, i);
1069
			rc = PTR_ERR(rep);
1070 1071
			goto out;
		}
1072
		list_add(&rep->rr_list, &buf->rb_recv_bufs);
1073
	}
1074

1075 1076 1077 1078 1079 1080
	return 0;
out:
	rpcrdma_buffer_destroy(buf);
	return rc;
}

1081 1082 1083 1084 1085 1086
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,
1087
			       struct rpcrdma_req, rl_list);
1088
	list_del_init(&req->rl_list);
1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102
	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;
}

1103
static void
1104
rpcrdma_destroy_rep(struct rpcrdma_rep *rep)
1105
{
1106
	rpcrdma_free_regbuf(rep->rr_rdmabuf);
1107 1108 1109
	kfree(rep);
}

1110
void
1111
rpcrdma_destroy_req(struct rpcrdma_req *req)
1112
{
1113 1114 1115
	rpcrdma_free_regbuf(req->rl_recvbuf);
	rpcrdma_free_regbuf(req->rl_sendbuf);
	rpcrdma_free_regbuf(req->rl_rdmabuf);
1116 1117 1118
	kfree(req);
}

C
Chuck Lever 已提交
1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144
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);
}

1145 1146 1147
void
rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf)
{
1148
	cancel_delayed_work_sync(&buf->rb_recovery_worker);
1149
	cancel_delayed_work_sync(&buf->rb_refresh_worker);
1150

1151 1152
	while (!list_empty(&buf->rb_recv_bufs)) {
		struct rpcrdma_rep *rep;
1153

1154
		rep = rpcrdma_buffer_get_rep_locked(buf);
1155
		rpcrdma_destroy_rep(rep);
1156
	}
1157
	buf->rb_send_count = 0;
1158

1159 1160
	spin_lock(&buf->rb_reqslock);
	while (!list_empty(&buf->rb_allreqs)) {
1161
		struct rpcrdma_req *req;
A
Allen Andrews 已提交
1162

1163 1164 1165 1166 1167
		req = list_first_entry(&buf->rb_allreqs,
				       struct rpcrdma_req, rl_all);
		list_del(&req->rl_all);

		spin_unlock(&buf->rb_reqslock);
1168
		rpcrdma_destroy_req(req);
1169
		spin_lock(&buf->rb_reqslock);
1170
	}
1171
	spin_unlock(&buf->rb_reqslock);
1172
	buf->rb_recv_count = 0;
A
Allen Andrews 已提交
1173

C
Chuck Lever 已提交
1174
	rpcrdma_destroy_mrs(buf);
1175 1176
}

1177 1178
struct rpcrdma_mw *
rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt)
1179
{
1180 1181 1182
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
	struct rpcrdma_mw *mw = NULL;

C
Chuck Lever 已提交
1183
	spin_lock(&buf->rb_mwlock);
1184 1185
	if (!list_empty(&buf->rb_mws))
		mw = rpcrdma_pop_mw(&buf->rb_mws);
C
Chuck Lever 已提交
1186
	spin_unlock(&buf->rb_mwlock);
1187 1188

	if (!mw)
C
Chuck Lever 已提交
1189
		goto out_nomws;
1190
	mw->mw_flags = 0;
1191
	return mw;
C
Chuck Lever 已提交
1192 1193 1194

out_nomws:
	dprintk("RPC:       %s: no MWs available\n", __func__);
1195 1196
	if (r_xprt->rx_ep.rep_connected != -ENODEV)
		schedule_delayed_work(&buf->rb_refresh_worker, 0);
C
Chuck Lever 已提交
1197 1198 1199 1200 1201

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

	return NULL;
1202 1203
}

1204 1205
void
rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw)
1206
{
1207
	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1208

C
Chuck Lever 已提交
1209
	spin_lock(&buf->rb_mwlock);
1210
	rpcrdma_push_mw(mw, &buf->rb_mws);
C
Chuck Lever 已提交
1211
	spin_unlock(&buf->rb_mwlock);
1212 1213
}

1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230
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);
}

1231 1232
/*
 * Get a set of request/reply buffers.
1233 1234
 *
 * Reply buffer (if available) is attached to send buffer upon return.
1235 1236 1237 1238 1239
 */
struct rpcrdma_req *
rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
{
	struct rpcrdma_req *req;
1240

1241
	spin_lock(&buffers->rb_lock);
1242 1243
	if (list_empty(&buffers->rb_send_bufs))
		goto out_reqbuf;
1244
	buffers->rb_send_count++;
1245
	req = rpcrdma_buffer_get_req_locked(buffers);
1246
	req->rl_reply = rpcrdma_buffer_get_rep(buffers);
1247
	spin_unlock(&buffers->rb_lock);
1248
	return req;
1249

1250
out_reqbuf:
1251
	spin_unlock(&buffers->rb_lock);
1252
	pr_warn("RPC:       %s: out of request buffers\n", __func__);
1253
	return NULL;
1254 1255 1256 1257 1258 1259 1260 1261 1262 1263
}

/*
 * 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;
1264
	struct rpcrdma_rep *rep = req->rl_reply;
1265

1266
	req->rl_send_wr.num_sge = 0;
1267 1268
	req->rl_reply = NULL;

1269
	spin_lock(&buffers->rb_lock);
1270
	buffers->rb_send_count--;
1271
	list_add_tail(&req->rl_list, &buffers->rb_send_bufs);
1272 1273
	if (rep) {
		buffers->rb_recv_count--;
1274
		list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);
1275
	}
1276
	spin_unlock(&buffers->rb_lock);
1277 1278 1279 1280
}

/*
 * Recover reply buffers from pool.
1281
 * This happens when recovering from disconnect.
1282 1283 1284 1285 1286 1287
 */
void
rpcrdma_recv_buffer_get(struct rpcrdma_req *req)
{
	struct rpcrdma_buffer *buffers = req->rl_buffer;

1288
	spin_lock(&buffers->rb_lock);
1289
	req->rl_reply = rpcrdma_buffer_get_rep(buffers);
1290
	spin_unlock(&buffers->rb_lock);
1291 1292 1293 1294
}

/*
 * Put reply buffers back into pool when not attached to
1295
 * request. This happens in error conditions.
1296 1297 1298 1299
 */
void
rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep)
{
1300
	struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf;
1301

1302
	spin_lock(&buffers->rb_lock);
1303
	buffers->rb_recv_count--;
1304
	list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);
1305
	spin_unlock(&buffers->rb_lock);
1306 1307
}

1308
/**
1309
 * rpcrdma_alloc_regbuf - allocate and DMA-map memory for SEND/RECV buffers
1310
 * @size: size of buffer to be allocated, in bytes
1311
 * @direction: direction of data movement
1312 1313
 * @flags: GFP flags
 *
1314 1315
 * Returns an ERR_PTR, or a pointer to a regbuf, a buffer that
 * can be persistently DMA-mapped for I/O.
1316 1317
 *
 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for
1318 1319
 * receiving the payload of RDMA RECV operations. During Long Calls
 * or Replies they may be registered externally via ro_map.
1320 1321
 */
struct rpcrdma_regbuf *
1322 1323
rpcrdma_alloc_regbuf(size_t size, enum dma_data_direction direction,
		     gfp_t flags)
1324 1325 1326 1327 1328
{
	struct rpcrdma_regbuf *rb;

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

1331
	rb->rg_device = NULL;
1332
	rb->rg_direction = direction;
1333
	rb->rg_iov.length = size;
1334 1335

	return rb;
1336
}
1337

1338 1339 1340 1341 1342 1343 1344 1345
/**
 * __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)
{
1346 1347
	struct ib_device *device = ia->ri_device;

1348 1349 1350
	if (rb->rg_direction == DMA_NONE)
		return false;

1351
	rb->rg_iov.addr = ib_dma_map_single(device,
1352 1353 1354
					    (void *)rb->rg_base,
					    rdmab_length(rb),
					    rb->rg_direction);
1355
	if (ib_dma_mapping_error(device, rdmab_addr(rb)))
1356 1357
		return false;

1358
	rb->rg_device = device;
1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371
	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;
1372 1373 1374 1375 1376 1377 1378
}

/**
 * rpcrdma_free_regbuf - deregister and free registered buffer
 * @rb: regbuf to be deregistered and freed
 */
void
1379
rpcrdma_free_regbuf(struct rpcrdma_regbuf *rb)
1380
{
1381 1382 1383
	if (!rb)
		return;

1384
	rpcrdma_dma_unmap_regbuf(rb);
1385
	kfree(rb);
1386 1387
}

1388 1389 1390 1391 1392 1393 1394 1395 1396 1397
/*
 * 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)
{
1398 1399
	struct ib_send_wr *send_wr = &req->rl_send_wr;
	struct ib_send_wr *send_wr_fail;
1400
	int rc;
1401

1402 1403
	if (req->rl_reply) {
		rc = rpcrdma_ep_post_recv(ia, req->rl_reply);
1404
		if (rc)
1405
			return rc;
1406 1407 1408
		req->rl_reply = NULL;
	}

1409
	dprintk("RPC:       %s: posting %d s/g entries\n",
1410
		__func__, send_wr->num_sge);
1411

1412
	rpcrdma_set_signaled(ep, send_wr);
1413
	rc = ib_post_send(ia->ri_id->qp, send_wr, &send_wr_fail);
1414
	if (rc)
1415 1416 1417 1418 1419 1420
		goto out_postsend_err;
	return 0;

out_postsend_err:
	pr_err("rpcrdma: RDMA Send ib_post_send returned %i\n", rc);
	return -ENOTCONN;
1421 1422 1423 1424 1425 1426
}

int
rpcrdma_ep_post_recv(struct rpcrdma_ia *ia,
		     struct rpcrdma_rep *rep)
{
1427
	struct ib_recv_wr *recv_wr_fail;
1428 1429
	int rc;

1430 1431
	if (!rpcrdma_dma_map_regbuf(ia, rep->rr_rdmabuf))
		goto out_map;
1432
	rc = ib_post_recv(ia->ri_id->qp, &rep->rr_recv_wr, &recv_wr_fail);
1433
	if (rc)
1434 1435 1436
		goto out_postrecv;
	return 0;

1437 1438 1439 1440
out_map:
	pr_err("rpcrdma: failed to DMA map the Receive buffer\n");
	return -EIO;

1441 1442 1443
out_postrecv:
	pr_err("rpcrdma: ib_post_recv returned %i\n", rc);
	return -ENOTCONN;
1444
}
1445

1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461
/**
 * 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--) {
1462
		spin_lock(&buffers->rb_lock);
1463 1464 1465
		if (list_empty(&buffers->rb_recv_bufs))
			goto out_reqbuf;
		rep = rpcrdma_buffer_get_rep_locked(buffers);
1466
		spin_unlock(&buffers->rb_lock);
1467

1468
		rc = rpcrdma_ep_post_recv(ia, rep);
1469 1470 1471 1472 1473 1474 1475
		if (rc)
			goto out_rc;
	}

	return 0;

out_reqbuf:
1476
	spin_unlock(&buffers->rb_lock);
1477 1478 1479 1480 1481 1482 1483
	pr_warn("%s: no extra receive buffers\n", __func__);
	return -ENOMEM;

out_rc:
	rpcrdma_recv_buffer_put(rep);
	return rc;
}