xprt.c 28.2 KB
Newer Older
L
Linus Torvalds 已提交
1 2 3 4 5 6 7 8 9 10 11 12
/*
 *  linux/net/sunrpc/xprt.c
 *
 *  This is a generic RPC call interface supporting congestion avoidance,
 *  and asynchronous calls.
 *
 *  The interface works like this:
 *
 *  -	When a process places a call, it allocates a request slot if
 *	one is available. Otherwise, it sleeps on the backlog queue
 *	(xprt_reserve).
 *  -	Next, the caller puts together the RPC message, stuffs it into
13 14 15
 *	the request struct, and calls xprt_transmit().
 *  -	xprt_transmit sends the message and installs the caller on the
 *	transport's wait list. At the same time, it installs a timer that
L
Linus Torvalds 已提交
16 17
 *	is run after the packet's timeout has expired.
 *  -	When a packet arrives, the data_ready handler walks the list of
18
 *	pending requests for that transport. If a matching XID is found, the
L
Linus Torvalds 已提交
19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
 *	caller is woken up, and the timer removed.
 *  -	When no reply arrives within the timeout interval, the timer is
 *	fired by the kernel and runs xprt_timer(). It either adjusts the
 *	timeout values (minor timeout) or wakes up the caller with a status
 *	of -ETIMEDOUT.
 *  -	When the caller receives a notification from RPC that a reply arrived,
 *	it should release the RPC slot, and process the reply.
 *	If the call timed out, it may choose to retry the operation by
 *	adjusting the initial timeout value, and simply calling rpc_call
 *	again.
 *
 *  Support for async RPC is done through a set of RPC-specific scheduling
 *  primitives that `transparently' work for processes as well as async
 *  tasks that rely on callbacks.
 *
 *  Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de>
35 36
 *
 *  Transport switch API copyright (C) 2005, Chuck Lever <cel@netapp.com>
L
Linus Torvalds 已提交
37 38
 */

39 40
#include <linux/module.h>

L
Linus Torvalds 已提交
41
#include <linux/types.h>
42
#include <linux/interrupt.h>
L
Linus Torvalds 已提交
43
#include <linux/workqueue.h>
44
#include <linux/net.h>
L
Linus Torvalds 已提交
45

46
#include <linux/sunrpc/clnt.h>
47
#include <linux/sunrpc/metrics.h>
L
Linus Torvalds 已提交
48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64

/*
 * Local variables
 */

#ifdef RPC_DEBUG
# define RPCDBG_FACILITY	RPCDBG_XPRT
#endif

/*
 * Local functions
 */
static void	xprt_request_init(struct rpc_task *, struct rpc_xprt *);
static inline void	do_xprt_reserve(struct rpc_task *);
static void	xprt_connect_status(struct rpc_task *task);
static int      __xprt_get_cong(struct rpc_xprt *, struct rpc_task *);

J
Jiri Slaby 已提交
65
static DEFINE_SPINLOCK(xprt_list_lock);
66 67
static LIST_HEAD(xprt_list);

68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84
/*
 * The transport code maintains an estimate on the maximum number of out-
 * standing RPC requests, using a smoothed version of the congestion
 * avoidance implemented in 44BSD. This is basically the Van Jacobson
 * congestion algorithm: If a retransmit occurs, the congestion window is
 * halved; otherwise, it is incremented by 1/cwnd when
 *
 *	-	a reply is received and
 *	-	a full number of requests are outstanding and
 *	-	the congestion window hasn't been updated recently.
 */
#define RPC_CWNDSHIFT		(8U)
#define RPC_CWNDSCALE		(1U << RPC_CWNDSHIFT)
#define RPC_INITCWND		RPC_CWNDSCALE
#define RPC_MAXCWND(xprt)	((xprt)->max_reqs << RPC_CWNDSHIFT)

#define RPCXPRT_CONGESTED(xprt) ((xprt)->cong >= (xprt)->cwnd)
L
Linus Torvalds 已提交
85

86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106
/**
 * xprt_register_transport - register a transport implementation
 * @transport: transport to register
 *
 * If a transport implementation is loaded as a kernel module, it can
 * call this interface to make itself known to the RPC client.
 *
 * Returns:
 * 0:		transport successfully registered
 * -EEXIST:	transport already registered
 * -EINVAL:	transport module being unloaded
 */
int xprt_register_transport(struct xprt_class *transport)
{
	struct xprt_class *t;
	int result;

	result = -EEXIST;
	spin_lock(&xprt_list_lock);
	list_for_each_entry(t, &xprt_list, list) {
		/* don't register the same transport class twice */
107
		if (t->ident == transport->ident)
108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157
			goto out;
	}

	result = -EINVAL;
	if (try_module_get(THIS_MODULE)) {
		list_add_tail(&transport->list, &xprt_list);
		printk(KERN_INFO "RPC: Registered %s transport module.\n",
			transport->name);
		result = 0;
	}

out:
	spin_unlock(&xprt_list_lock);
	return result;
}
EXPORT_SYMBOL_GPL(xprt_register_transport);

/**
 * xprt_unregister_transport - unregister a transport implementation
 * transport: transport to unregister
 *
 * Returns:
 * 0:		transport successfully unregistered
 * -ENOENT:	transport never registered
 */
int xprt_unregister_transport(struct xprt_class *transport)
{
	struct xprt_class *t;
	int result;

	result = 0;
	spin_lock(&xprt_list_lock);
	list_for_each_entry(t, &xprt_list, list) {
		if (t == transport) {
			printk(KERN_INFO
				"RPC: Unregistered %s transport module.\n",
				transport->name);
			list_del_init(&transport->list);
			module_put(THIS_MODULE);
			goto out;
		}
	}
	result = -ENOENT;

out:
	spin_unlock(&xprt_list_lock);
	return result;
}
EXPORT_SYMBOL_GPL(xprt_unregister_transport);

158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185
/**
 * xprt_reserve_xprt - serialize write access to transports
 * @task: task that is requesting access to the transport
 *
 * This prevents mixing the payload of separate requests, and prevents
 * transport connects from colliding with writes.  No congestion control
 * is provided.
 */
int xprt_reserve_xprt(struct rpc_task *task)
{
	struct rpc_xprt	*xprt = task->tk_xprt;
	struct rpc_rqst *req = task->tk_rqstp;

	if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
		if (task == xprt->snd_task)
			return 1;
		if (task == NULL)
			return 0;
		goto out_sleep;
	}
	xprt->snd_task = task;
	if (req) {
		req->rq_bytes_sent = 0;
		req->rq_ntrans++;
	}
	return 1;

out_sleep:
186
	dprintk("RPC: %5u failed to lock transport %p\n",
187 188 189 190 191 192 193 194 195
			task->tk_pid, xprt);
	task->tk_timeout = 0;
	task->tk_status = -EAGAIN;
	if (req && req->rq_ntrans)
		rpc_sleep_on(&xprt->resend, task, NULL, NULL);
	else
		rpc_sleep_on(&xprt->sending, task, NULL, NULL);
	return 0;
}
196
EXPORT_SYMBOL_GPL(xprt_reserve_xprt);
197

198 199 200 201 202 203 204 205
static void xprt_clear_locked(struct rpc_xprt *xprt)
{
	xprt->snd_task = NULL;
	if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state) || xprt->shutdown) {
		smp_mb__before_clear_bit();
		clear_bit(XPRT_LOCKED, &xprt->state);
		smp_mb__after_clear_bit();
	} else
206
		queue_work(rpciod_workqueue, &xprt->task_cleanup);
207 208
}

L
Linus Torvalds 已提交
209
/*
210 211 212 213 214 215
 * xprt_reserve_xprt_cong - serialize write access to transports
 * @task: task that is requesting access to the transport
 *
 * Same as xprt_reserve_xprt, but Van Jacobson congestion control is
 * integrated into the decision of whether a request is allowed to be
 * woken up and given access to the transport.
L
Linus Torvalds 已提交
216
 */
217
int xprt_reserve_xprt_cong(struct rpc_task *task)
L
Linus Torvalds 已提交
218
{
219
	struct rpc_xprt	*xprt = task->tk_xprt;
L
Linus Torvalds 已提交
220 221
	struct rpc_rqst *req = task->tk_rqstp;

222
	if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
L
Linus Torvalds 已提交
223 224 225 226
		if (task == xprt->snd_task)
			return 1;
		goto out_sleep;
	}
227
	if (__xprt_get_cong(xprt, task)) {
L
Linus Torvalds 已提交
228 229 230 231 232 233 234
		xprt->snd_task = task;
		if (req) {
			req->rq_bytes_sent = 0;
			req->rq_ntrans++;
		}
		return 1;
	}
235
	xprt_clear_locked(xprt);
L
Linus Torvalds 已提交
236
out_sleep:
237
	dprintk("RPC: %5u failed to lock transport %p\n", task->tk_pid, xprt);
L
Linus Torvalds 已提交
238 239 240 241 242 243 244 245
	task->tk_timeout = 0;
	task->tk_status = -EAGAIN;
	if (req && req->rq_ntrans)
		rpc_sleep_on(&xprt->resend, task, NULL, NULL);
	else
		rpc_sleep_on(&xprt->sending, task, NULL, NULL);
	return 0;
}
246
EXPORT_SYMBOL_GPL(xprt_reserve_xprt_cong);
L
Linus Torvalds 已提交
247

248
static inline int xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
L
Linus Torvalds 已提交
249 250 251
{
	int retval;

C
Chuck Lever 已提交
252
	spin_lock_bh(&xprt->transport_lock);
253
	retval = xprt->ops->reserve_xprt(task);
C
Chuck Lever 已提交
254
	spin_unlock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
255 256 257
	return retval;
}

258
static void __xprt_lock_write_next(struct rpc_xprt *xprt)
259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281
{
	struct rpc_task *task;
	struct rpc_rqst *req;

	if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
		return;

	task = rpc_wake_up_next(&xprt->resend);
	if (!task) {
		task = rpc_wake_up_next(&xprt->sending);
		if (!task)
			goto out_unlock;
	}

	req = task->tk_rqstp;
	xprt->snd_task = task;
	if (req) {
		req->rq_bytes_sent = 0;
		req->rq_ntrans++;
	}
	return;

out_unlock:
282
	xprt_clear_locked(xprt);
283 284 285
}

static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt)
L
Linus Torvalds 已提交
286 287 288
{
	struct rpc_task *task;

289
	if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
L
Linus Torvalds 已提交
290
		return;
291
	if (RPCXPRT_CONGESTED(xprt))
L
Linus Torvalds 已提交
292 293 294 295 296 297 298
		goto out_unlock;
	task = rpc_wake_up_next(&xprt->resend);
	if (!task) {
		task = rpc_wake_up_next(&xprt->sending);
		if (!task)
			goto out_unlock;
	}
299
	if (__xprt_get_cong(xprt, task)) {
L
Linus Torvalds 已提交
300 301 302 303 304 305 306 307 308
		struct rpc_rqst *req = task->tk_rqstp;
		xprt->snd_task = task;
		if (req) {
			req->rq_bytes_sent = 0;
			req->rq_ntrans++;
		}
		return;
	}
out_unlock:
309
	xprt_clear_locked(xprt);
L
Linus Torvalds 已提交
310 311
}

312 313 314 315 316 317
/**
 * xprt_release_xprt - allow other requests to use a transport
 * @xprt: transport with other tasks potentially waiting
 * @task: task that is releasing access to the transport
 *
 * Note that "task" can be NULL.  No congestion control is provided.
L
Linus Torvalds 已提交
318
 */
319
void xprt_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
L
Linus Torvalds 已提交
320 321
{
	if (xprt->snd_task == task) {
322
		xprt_clear_locked(xprt);
L
Linus Torvalds 已提交
323 324 325
		__xprt_lock_write_next(xprt);
	}
}
326
EXPORT_SYMBOL_GPL(xprt_release_xprt);
L
Linus Torvalds 已提交
327

328 329 330 331 332 333 334 335 336 337 338
/**
 * xprt_release_xprt_cong - allow other requests to use a transport
 * @xprt: transport with other tasks potentially waiting
 * @task: task that is releasing access to the transport
 *
 * Note that "task" can be NULL.  Another task is awoken to use the
 * transport if the transport's congestion window allows it.
 */
void xprt_release_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task)
{
	if (xprt->snd_task == task) {
339
		xprt_clear_locked(xprt);
340 341 342
		__xprt_lock_write_next_cong(xprt);
	}
}
343
EXPORT_SYMBOL_GPL(xprt_release_xprt_cong);
344 345

static inline void xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task)
L
Linus Torvalds 已提交
346
{
C
Chuck Lever 已提交
347
	spin_lock_bh(&xprt->transport_lock);
348
	xprt->ops->release_xprt(xprt, task);
C
Chuck Lever 已提交
349
	spin_unlock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
350 351 352 353 354 355 356 357 358 359 360 361 362
}

/*
 * Van Jacobson congestion avoidance. Check if the congestion window
 * overflowed. Put the task to sleep if this is the case.
 */
static int
__xprt_get_cong(struct rpc_xprt *xprt, struct rpc_task *task)
{
	struct rpc_rqst *req = task->tk_rqstp;

	if (req->rq_cong)
		return 1;
363
	dprintk("RPC: %5u xprt_cwnd_limited cong = %lu cwnd = %lu\n",
L
Linus Torvalds 已提交
364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382
			task->tk_pid, xprt->cong, xprt->cwnd);
	if (RPCXPRT_CONGESTED(xprt))
		return 0;
	req->rq_cong = 1;
	xprt->cong += RPC_CWNDSCALE;
	return 1;
}

/*
 * Adjust the congestion window, and wake up the next task
 * that has been sleeping due to congestion
 */
static void
__xprt_put_cong(struct rpc_xprt *xprt, struct rpc_rqst *req)
{
	if (!req->rq_cong)
		return;
	req->rq_cong = 0;
	xprt->cong -= RPC_CWNDSCALE;
383
	__xprt_lock_write_next_cong(xprt);
L
Linus Torvalds 已提交
384 385
}

386 387 388 389 390 391 392 393 394 395
/**
 * xprt_release_rqst_cong - housekeeping when request is complete
 * @task: RPC request that recently completed
 *
 * Useful for transports that require congestion control.
 */
void xprt_release_rqst_cong(struct rpc_task *task)
{
	__xprt_put_cong(task->tk_xprt, task->tk_rqstp);
}
396
EXPORT_SYMBOL_GPL(xprt_release_rqst_cong);
397

398 399 400 401 402
/**
 * xprt_adjust_cwnd - adjust transport congestion window
 * @task: recently completed RPC request used to adjust window
 * @result: result code of completed RPC request
 *
L
Linus Torvalds 已提交
403 404
 * We use a time-smoothed congestion estimator to avoid heavy oscillation.
 */
405
void xprt_adjust_cwnd(struct rpc_task *task, int result)
L
Linus Torvalds 已提交
406
{
407 408 409
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = task->tk_xprt;
	unsigned long cwnd = xprt->cwnd;
L
Linus Torvalds 已提交
410 411 412 413 414 415 416

	if (result >= 0 && cwnd <= xprt->cong) {
		/* The (cwnd >> 1) term makes sure
		 * the result gets rounded properly. */
		cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd;
		if (cwnd > RPC_MAXCWND(xprt))
			cwnd = RPC_MAXCWND(xprt);
417
		__xprt_lock_write_next_cong(xprt);
L
Linus Torvalds 已提交
418 419 420 421 422
	} else if (result == -ETIMEDOUT) {
		cwnd >>= 1;
		if (cwnd < RPC_CWNDSCALE)
			cwnd = RPC_CWNDSCALE;
	}
423
	dprintk("RPC:       cong %ld, cwnd was %ld, now %ld\n",
L
Linus Torvalds 已提交
424 425
			xprt->cong, xprt->cwnd, cwnd);
	xprt->cwnd = cwnd;
426
	__xprt_put_cong(xprt, req);
L
Linus Torvalds 已提交
427
}
428
EXPORT_SYMBOL_GPL(xprt_adjust_cwnd);
L
Linus Torvalds 已提交
429

430 431 432 433 434 435 436 437 438 439 440 441 442
/**
 * xprt_wake_pending_tasks - wake all tasks on a transport's pending queue
 * @xprt: transport with waiting tasks
 * @status: result code to plant in each task before waking it
 *
 */
void xprt_wake_pending_tasks(struct rpc_xprt *xprt, int status)
{
	if (status < 0)
		rpc_wake_up_status(&xprt->pending, status);
	else
		rpc_wake_up(&xprt->pending);
}
443
EXPORT_SYMBOL_GPL(xprt_wake_pending_tasks);
444

445 446 447 448 449 450 451 452 453 454 455 456 457
/**
 * xprt_wait_for_buffer_space - wait for transport output buffer to clear
 * @task: task to be put to sleep
 *
 */
void xprt_wait_for_buffer_space(struct rpc_task *task)
{
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;

	task->tk_timeout = req->rq_timeout;
	rpc_sleep_on(&xprt->pending, task, NULL, NULL);
}
458
EXPORT_SYMBOL_GPL(xprt_wait_for_buffer_space);
459 460 461 462 463 464 465 466 467 468 469 470 471 472

/**
 * xprt_write_space - wake the task waiting for transport output buffer space
 * @xprt: transport with waiting tasks
 *
 * Can be called in a soft IRQ context, so xprt_write_space never sleeps.
 */
void xprt_write_space(struct rpc_xprt *xprt)
{
	if (unlikely(xprt->shutdown))
		return;

	spin_lock_bh(&xprt->transport_lock);
	if (xprt->snd_task) {
473 474
		dprintk("RPC:       write space: waking waiting task on "
				"xprt %p\n", xprt);
475 476 477 478
		rpc_wake_up_task(xprt->snd_task);
	}
	spin_unlock_bh(&xprt->transport_lock);
}
479
EXPORT_SYMBOL_GPL(xprt_write_space);
480

481 482 483 484 485 486 487 488 489 490 491 492
/**
 * xprt_set_retrans_timeout_def - set a request's retransmit timeout
 * @task: task whose timeout is to be set
 *
 * Set a request's retransmit timeout based on the transport's
 * default timeout parameters.  Used by transports that don't adjust
 * the retransmit timeout based on round-trip time estimation.
 */
void xprt_set_retrans_timeout_def(struct rpc_task *task)
{
	task->tk_timeout = task->tk_rqstp->rq_timeout;
}
493
EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_def);
494 495 496 497

/*
 * xprt_set_retrans_timeout_rtt - set a request's retransmit timeout
 * @task: task whose timeout is to be set
498
 *
499 500 501 502 503 504 505 506 507 508 509 510 511 512
 * Set a request's retransmit timeout using the RTT estimator.
 */
void xprt_set_retrans_timeout_rtt(struct rpc_task *task)
{
	int timer = task->tk_msg.rpc_proc->p_timer;
	struct rpc_rtt *rtt = task->tk_client->cl_rtt;
	struct rpc_rqst *req = task->tk_rqstp;
	unsigned long max_timeout = req->rq_xprt->timeout.to_maxval;

	task->tk_timeout = rpc_calc_rto(rtt, timer);
	task->tk_timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries;
	if (task->tk_timeout > max_timeout || task->tk_timeout == 0)
		task->tk_timeout = max_timeout;
}
513
EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_rtt);
514

L
Linus Torvalds 已提交
515 516 517 518 519 520 521 522 523 524 525 526 527 528
static void xprt_reset_majortimeo(struct rpc_rqst *req)
{
	struct rpc_timeout *to = &req->rq_xprt->timeout;

	req->rq_majortimeo = req->rq_timeout;
	if (to->to_exponential)
		req->rq_majortimeo <<= to->to_retries;
	else
		req->rq_majortimeo += to->to_increment * to->to_retries;
	if (req->rq_majortimeo > to->to_maxval || req->rq_majortimeo == 0)
		req->rq_majortimeo = to->to_maxval;
	req->rq_majortimeo += jiffies;
}

529 530 531 532
/**
 * xprt_adjust_timeout - adjust timeout values for next retransmit
 * @req: RPC request containing parameters to use for the adjustment
 *
L
Linus Torvalds 已提交
533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552
 */
int xprt_adjust_timeout(struct rpc_rqst *req)
{
	struct rpc_xprt *xprt = req->rq_xprt;
	struct rpc_timeout *to = &xprt->timeout;
	int status = 0;

	if (time_before(jiffies, req->rq_majortimeo)) {
		if (to->to_exponential)
			req->rq_timeout <<= 1;
		else
			req->rq_timeout += to->to_increment;
		if (to->to_maxval && req->rq_timeout >= to->to_maxval)
			req->rq_timeout = to->to_maxval;
		req->rq_retries++;
	} else {
		req->rq_timeout = to->to_initval;
		req->rq_retries = 0;
		xprt_reset_majortimeo(req);
		/* Reset the RTT counters == "slow start" */
C
Chuck Lever 已提交
553
		spin_lock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
554
		rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval);
C
Chuck Lever 已提交
555
		spin_unlock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
556 557 558 559 560 561 562 563 564 565
		status = -ETIMEDOUT;
	}

	if (req->rq_timeout == 0) {
		printk(KERN_WARNING "xprt_adjust_timeout: rq_timeout = 0!\n");
		req->rq_timeout = 5 * HZ;
	}
	return status;
}

566
static void xprt_autoclose(struct work_struct *work)
L
Linus Torvalds 已提交
567
{
568 569
	struct rpc_xprt *xprt =
		container_of(work, struct rpc_xprt, task_cleanup);
L
Linus Torvalds 已提交
570

571
	xprt->ops->close(xprt);
572
	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
L
Linus Torvalds 已提交
573 574 575
	xprt_release_write(xprt, NULL);
}

576
/**
577
 * xprt_disconnect_done - mark a transport as disconnected
578 579
 * @xprt: transport to flag for disconnect
 *
L
Linus Torvalds 已提交
580
 */
581
void xprt_disconnect_done(struct rpc_xprt *xprt)
L
Linus Torvalds 已提交
582
{
583
	dprintk("RPC:       disconnected transport %p\n", xprt);
C
Chuck Lever 已提交
584
	spin_lock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
585
	xprt_clear_connected(xprt);
586
	xprt_wake_pending_tasks(xprt, -ENOTCONN);
C
Chuck Lever 已提交
587
	spin_unlock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
588
}
589
EXPORT_SYMBOL_GPL(xprt_disconnect_done);
L
Linus Torvalds 已提交
590

591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609
/**
 * xprt_force_disconnect - force a transport to disconnect
 * @xprt: transport to disconnect
 *
 */
void xprt_force_disconnect(struct rpc_xprt *xprt)
{
	/* Don't race with the test_bit() in xprt_clear_locked() */
	spin_lock_bh(&xprt->transport_lock);
	set_bit(XPRT_CLOSE_WAIT, &xprt->state);
	/* Try to schedule an autoclose RPC call */
	if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
		queue_work(rpciod_workqueue, &xprt->task_cleanup);
	else if (xprt->snd_task != NULL)
		rpc_wake_up_task(xprt->snd_task);
	spin_unlock_bh(&xprt->transport_lock);
}
EXPORT_SYMBOL_GPL(xprt_force_disconnect);

L
Linus Torvalds 已提交
610 611 612 613 614
static void
xprt_init_autodisconnect(unsigned long data)
{
	struct rpc_xprt *xprt = (struct rpc_xprt *)data;

C
Chuck Lever 已提交
615
	spin_lock(&xprt->transport_lock);
L
Linus Torvalds 已提交
616 617
	if (!list_empty(&xprt->recv) || xprt->shutdown)
		goto out_abort;
618
	if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
L
Linus Torvalds 已提交
619
		goto out_abort;
C
Chuck Lever 已提交
620
	spin_unlock(&xprt->transport_lock);
621
	if (xprt_connecting(xprt))
L
Linus Torvalds 已提交
622 623
		xprt_release_write(xprt, NULL);
	else
624
		queue_work(rpciod_workqueue, &xprt->task_cleanup);
L
Linus Torvalds 已提交
625 626
	return;
out_abort:
C
Chuck Lever 已提交
627
	spin_unlock(&xprt->transport_lock);
L
Linus Torvalds 已提交
628 629
}

630 631 632
/**
 * xprt_connect - schedule a transport connect operation
 * @task: RPC task that is requesting the connect
L
Linus Torvalds 已提交
633 634 635 636 637 638
 *
 */
void xprt_connect(struct rpc_task *task)
{
	struct rpc_xprt	*xprt = task->tk_xprt;

639
	dprintk("RPC: %5u xprt_connect xprt %p %s connected\n", task->tk_pid,
L
Linus Torvalds 已提交
640 641
			xprt, (xprt_connected(xprt) ? "is" : "is not"));

642
	if (!xprt_bound(xprt)) {
L
Linus Torvalds 已提交
643 644 645 646 647 648
		task->tk_status = -EIO;
		return;
	}
	if (!xprt_lock_write(xprt, task))
		return;
	if (xprt_connected(xprt))
649 650 651 652
		xprt_release_write(xprt, task);
	else {
		if (task->tk_rqstp)
			task->tk_rqstp->rq_bytes_sent = 0;
L
Linus Torvalds 已提交
653

654
		task->tk_timeout = xprt->connect_timeout;
655
		rpc_sleep_on(&xprt->pending, task, xprt_connect_status, NULL);
656
		xprt->stat.connect_start = jiffies;
657
		xprt->ops->connect(task);
L
Linus Torvalds 已提交
658 659 660 661
	}
	return;
}

662
static void xprt_connect_status(struct rpc_task *task)
L
Linus Torvalds 已提交
663 664 665 666
{
	struct rpc_xprt	*xprt = task->tk_xprt;

	if (task->tk_status >= 0) {
667 668
		xprt->stat.connect_count++;
		xprt->stat.connect_time += (long)jiffies - xprt->stat.connect_start;
669
		dprintk("RPC: %5u xprt_connect_status: connection established\n",
L
Linus Torvalds 已提交
670 671 672 673 674 675 676
				task->tk_pid);
		return;
	}

	switch (task->tk_status) {
	case -ECONNREFUSED:
	case -ECONNRESET:
677 678 679
		dprintk("RPC: %5u xprt_connect_status: server %s refused "
				"connection\n", task->tk_pid,
				task->tk_client->cl_server);
680
		break;
L
Linus Torvalds 已提交
681
	case -ENOTCONN:
682
		dprintk("RPC: %5u xprt_connect_status: connection broken\n",
683 684
				task->tk_pid);
		break;
L
Linus Torvalds 已提交
685
	case -ETIMEDOUT:
686 687
		dprintk("RPC: %5u xprt_connect_status: connect attempt timed "
				"out\n", task->tk_pid);
L
Linus Torvalds 已提交
688 689
		break;
	default:
690 691 692
		dprintk("RPC: %5u xprt_connect_status: error %d connecting to "
				"server %s\n", task->tk_pid, -task->tk_status,
				task->tk_client->cl_server);
693 694
		xprt_release_write(xprt, task);
		task->tk_status = -EIO;
L
Linus Torvalds 已提交
695 696 697
	}
}

698 699 700 701 702
/**
 * xprt_lookup_rqst - find an RPC request corresponding to an XID
 * @xprt: transport on which the original request was transmitted
 * @xid: RPC XID of incoming reply
 *
L
Linus Torvalds 已提交
703
 */
704
struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, __be32 xid)
L
Linus Torvalds 已提交
705 706 707 708 709
{
	struct list_head *pos;

	list_for_each(pos, &xprt->recv) {
		struct rpc_rqst *entry = list_entry(pos, struct rpc_rqst, rq_list);
710 711
		if (entry->rq_xid == xid)
			return entry;
L
Linus Torvalds 已提交
712
	}
713 714 715

	dprintk("RPC:       xprt_lookup_rqst did not find xid %08x\n",
			ntohl(xid));
716 717
	xprt->stat.bad_xids++;
	return NULL;
L
Linus Torvalds 已提交
718
}
719
EXPORT_SYMBOL_GPL(xprt_lookup_rqst);
L
Linus Torvalds 已提交
720

721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738
/**
 * xprt_update_rtt - update an RPC client's RTT state after receiving a reply
 * @task: RPC request that recently completed
 *
 */
void xprt_update_rtt(struct rpc_task *task)
{
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_rtt *rtt = task->tk_client->cl_rtt;
	unsigned timer = task->tk_msg.rpc_proc->p_timer;

	if (timer) {
		if (req->rq_ntrans == 1)
			rpc_update_rtt(rtt, timer,
					(long)jiffies - req->rq_xtime);
		rpc_set_timeo(rtt, timer, req->rq_ntrans - 1);
	}
}
739
EXPORT_SYMBOL_GPL(xprt_update_rtt);
740

741 742
/**
 * xprt_complete_rqst - called when reply processing is complete
743
 * @task: RPC request that recently completed
744 745
 * @copied: actual number of bytes received from the transport
 *
746
 * Caller holds transport lock.
L
Linus Torvalds 已提交
747
 */
748
void xprt_complete_rqst(struct rpc_task *task, int copied)
L
Linus Torvalds 已提交
749
{
750
	struct rpc_rqst *req = task->tk_rqstp;
L
Linus Torvalds 已提交
751

752 753
	dprintk("RPC: %5u xid %08x complete (%d bytes received)\n",
			task->tk_pid, ntohl(req->rq_xid), copied);
L
Linus Torvalds 已提交
754

755
	task->tk_xprt->stat.recvs++;
756 757
	task->tk_rtt = (long)jiffies - req->rq_xtime;

L
Linus Torvalds 已提交
758
	list_del_init(&req->rq_list);
759 760
	/* Ensure all writes are done before we update req->rq_received */
	smp_wmb();
L
Linus Torvalds 已提交
761 762 763
	req->rq_received = req->rq_private_buf.len = copied;
	rpc_wake_up_task(task);
}
764
EXPORT_SYMBOL_GPL(xprt_complete_rqst);
L
Linus Torvalds 已提交
765

766
static void xprt_timer(struct rpc_task *task)
L
Linus Torvalds 已提交
767
{
768
	struct rpc_rqst *req = task->tk_rqstp;
L
Linus Torvalds 已提交
769 770
	struct rpc_xprt *xprt = req->rq_xprt;

771
	dprintk("RPC: %5u xprt_timer\n", task->tk_pid);
L
Linus Torvalds 已提交
772

773 774 775 776 777 778
	spin_lock(&xprt->transport_lock);
	if (!req->rq_received) {
		if (xprt->ops->timer)
			xprt->ops->timer(task);
		task->tk_status = -ETIMEDOUT;
	}
L
Linus Torvalds 已提交
779 780
	task->tk_timeout = 0;
	rpc_wake_up_task(task);
C
Chuck Lever 已提交
781
	spin_unlock(&xprt->transport_lock);
L
Linus Torvalds 已提交
782 783
}

784 785 786 787
/**
 * xprt_prepare_transmit - reserve the transport before sending a request
 * @task: RPC task about to send a request
 *
L
Linus Torvalds 已提交
788
 */
789
int xprt_prepare_transmit(struct rpc_task *task)
L
Linus Torvalds 已提交
790 791 792 793 794
{
	struct rpc_rqst	*req = task->tk_rqstp;
	struct rpc_xprt	*xprt = req->rq_xprt;
	int err = 0;

795
	dprintk("RPC: %5u xprt_prepare_transmit\n", task->tk_pid);
L
Linus Torvalds 已提交
796

C
Chuck Lever 已提交
797
	spin_lock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
798 799 800 801
	if (req->rq_received && !req->rq_bytes_sent) {
		err = req->rq_received;
		goto out_unlock;
	}
802
	if (!xprt->ops->reserve_xprt(task)) {
L
Linus Torvalds 已提交
803 804 805 806 807 808 809 810 811
		err = -EAGAIN;
		goto out_unlock;
	}

	if (!xprt_connected(xprt)) {
		err = -ENOTCONN;
		goto out_unlock;
	}
out_unlock:
C
Chuck Lever 已提交
812
	spin_unlock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
813 814 815
	return err;
}

816
void xprt_end_transmit(struct rpc_task *task)
817
{
818
	xprt_release_write(task->tk_xprt, task);
819 820
}

821 822 823 824 825 826 827
/**
 * xprt_transmit - send an RPC request on a transport
 * @task: controlling RPC task
 *
 * We have to copy the iovec because sendmsg fiddles with its contents.
 */
void xprt_transmit(struct rpc_task *task)
L
Linus Torvalds 已提交
828 829 830
{
	struct rpc_rqst	*req = task->tk_rqstp;
	struct rpc_xprt	*xprt = req->rq_xprt;
831
	int status;
L
Linus Torvalds 已提交
832

833
	dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
L
Linus Torvalds 已提交
834 835 836

	if (!req->rq_received) {
		if (list_empty(&req->rq_list)) {
C
Chuck Lever 已提交
837
			spin_lock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
838 839 840 841 842
			/* Update the softirq receive buffer */
			memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
					sizeof(req->rq_private_buf));
			/* Add request to the receive list */
			list_add_tail(&req->rq_list, &xprt->recv);
C
Chuck Lever 已提交
843
			spin_unlock_bh(&xprt->transport_lock);
L
Linus Torvalds 已提交
844
			xprt_reset_majortimeo(req);
845 846
			/* Turn off autodisconnect */
			del_singleshot_timer_sync(&xprt->timer);
L
Linus Torvalds 已提交
847 848 849 850
		}
	} else if (!req->rq_bytes_sent)
		return;

851
	status = xprt->ops->send_request(task);
852
	if (status == 0) {
853
		dprintk("RPC: %5u xmit complete\n", task->tk_pid);
854
		spin_lock_bh(&xprt->transport_lock);
855

856
		xprt->ops->set_retrans_timeout(task);
857 858 859 860 861

		xprt->stat.sends++;
		xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs;
		xprt->stat.bklog_u += xprt->backlog.qlen;

862 863 864 865 866 867 868 869
		/* Don't race with disconnect */
		if (!xprt_connected(xprt))
			task->tk_status = -ENOTCONN;
		else if (!req->rq_received)
			rpc_sleep_on(&xprt->pending, task, NULL, xprt_timer);
		spin_unlock_bh(&xprt->transport_lock);
		return;
	}
L
Linus Torvalds 已提交
870 871 872 873 874 875

	/* Note: at this point, task->tk_sleeping has not yet been set,
	 *	 hence there is no danger of the waking up task being put on
	 *	 schedq, and being picked up by a parallel run of rpciod().
	 */
	task->tk_status = status;
876
	if (status == -ECONNREFUSED)
L
Linus Torvalds 已提交
877 878 879
		rpc_sleep_on(&xprt->sending, task, NULL, NULL);
}

880
static inline void do_xprt_reserve(struct rpc_task *task)
L
Linus Torvalds 已提交
881 882 883 884 885 886 887 888 889 890 891 892 893
{
	struct rpc_xprt	*xprt = task->tk_xprt;

	task->tk_status = 0;
	if (task->tk_rqstp)
		return;
	if (!list_empty(&xprt->free)) {
		struct rpc_rqst	*req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
		list_del_init(&req->rq_list);
		task->tk_rqstp = req;
		xprt_request_init(task, xprt);
		return;
	}
894
	dprintk("RPC:       waiting for request slot\n");
L
Linus Torvalds 已提交
895 896 897 898 899
	task->tk_status = -EAGAIN;
	task->tk_timeout = 0;
	rpc_sleep_on(&xprt->backlog, task, NULL, NULL);
}

900 901 902 903 904 905 906 907
/**
 * xprt_reserve - allocate an RPC request slot
 * @task: RPC task requesting a slot allocation
 *
 * If no more slots are available, place the task on the transport's
 * backlog queue.
 */
void xprt_reserve(struct rpc_task *task)
L
Linus Torvalds 已提交
908 909 910 911
{
	struct rpc_xprt	*xprt = task->tk_xprt;

	task->tk_status = -EIO;
912 913 914
	spin_lock(&xprt->reserve_lock);
	do_xprt_reserve(task);
	spin_unlock(&xprt->reserve_lock);
L
Linus Torvalds 已提交
915 916
}

917
static inline __be32 xprt_alloc_xid(struct rpc_xprt *xprt)
L
Linus Torvalds 已提交
918 919 920 921 922 923
{
	return xprt->xid++;
}

static inline void xprt_init_xid(struct rpc_xprt *xprt)
{
924
	xprt->xid = net_random();
L
Linus Torvalds 已提交
925 926
}

927
static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
L
Linus Torvalds 已提交
928 929 930 931 932 933
{
	struct rpc_rqst	*req = task->tk_rqstp;

	req->rq_timeout = xprt->timeout.to_initval;
	req->rq_task	= task;
	req->rq_xprt    = xprt;
934
	req->rq_buffer  = NULL;
L
Linus Torvalds 已提交
935
	req->rq_xid     = xprt_alloc_xid(xprt);
936
	req->rq_release_snd_buf = NULL;
937
	xprt_reset_majortimeo(req);
938
	dprintk("RPC: %5u reserved req %p xid %08x\n", task->tk_pid,
L
Linus Torvalds 已提交
939 940 941
			req, ntohl(req->rq_xid));
}

942 943 944 945
/**
 * xprt_release - release an RPC request slot
 * @task: task which is finished with the slot
 *
L
Linus Torvalds 已提交
946
 */
947
void xprt_release(struct rpc_task *task)
L
Linus Torvalds 已提交
948 949 950 951 952 953
{
	struct rpc_xprt	*xprt = task->tk_xprt;
	struct rpc_rqst	*req;

	if (!(req = task->tk_rqstp))
		return;
954
	rpc_count_iostats(task);
C
Chuck Lever 已提交
955
	spin_lock_bh(&xprt->transport_lock);
956
	xprt->ops->release_xprt(xprt, task);
957 958
	if (xprt->ops->release_request)
		xprt->ops->release_request(task);
L
Linus Torvalds 已提交
959 960 961
	if (!list_empty(&req->rq_list))
		list_del(&req->rq_list);
	xprt->last_used = jiffies;
962
	if (list_empty(&xprt->recv))
963
		mod_timer(&xprt->timer,
964
				xprt->last_used + xprt->idle_timeout);
C
Chuck Lever 已提交
965
	spin_unlock_bh(&xprt->transport_lock);
966
	xprt->ops->buf_free(req->rq_buffer);
L
Linus Torvalds 已提交
967
	task->tk_rqstp = NULL;
968 969
	if (req->rq_release_snd_buf)
		req->rq_release_snd_buf(req);
L
Linus Torvalds 已提交
970 971
	memset(req, 0, sizeof(*req));	/* mark unused */

972
	dprintk("RPC: %5u release request %p\n", task->tk_pid, req);
L
Linus Torvalds 已提交
973

C
Chuck Lever 已提交
974
	spin_lock(&xprt->reserve_lock);
L
Linus Torvalds 已提交
975
	list_add(&req->rq_list, &xprt->free);
976
	rpc_wake_up_next(&xprt->backlog);
C
Chuck Lever 已提交
977
	spin_unlock(&xprt->reserve_lock);
L
Linus Torvalds 已提交
978 979
}

980 981
/**
 * xprt_create_transport - create an RPC transport
982
 * @args: rpc transport creation arguments
983 984
 *
 */
985
struct rpc_xprt *xprt_create_transport(struct xprt_create *args)
986 987 988
{
	struct rpc_xprt	*xprt;
	struct rpc_rqst	*req;
989
	struct xprt_class *t;
990

991 992
	spin_lock(&xprt_list_lock);
	list_for_each_entry(t, &xprt_list, list) {
993
		if (t->ident == args->ident) {
994 995 996
			spin_unlock(&xprt_list_lock);
			goto found;
		}
997
	}
998
	spin_unlock(&xprt_list_lock);
999
	printk(KERN_ERR "RPC: transport (%d) not supported\n", args->ident);
1000 1001 1002 1003
	return ERR_PTR(-EIO);

found:
	xprt = t->setup(args);
1004
	if (IS_ERR(xprt)) {
1005
		dprintk("RPC:       xprt_create_transport: failed, %ld\n",
1006 1007
				-PTR_ERR(xprt));
		return xprt;
1008 1009
	}

1010
	kref_init(&xprt->kref);
1011 1012 1013 1014 1015
	spin_lock_init(&xprt->transport_lock);
	spin_lock_init(&xprt->reserve_lock);

	INIT_LIST_HEAD(&xprt->free);
	INIT_LIST_HEAD(&xprt->recv);
1016
	INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
1017 1018
	setup_timer(&xprt->timer, xprt_init_autodisconnect,
			(unsigned long)xprt);
1019 1020
	xprt->last_used = jiffies;
	xprt->cwnd = RPC_INITCWND;
1021
	xprt->bind_index = 0;
1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034

	rpc_init_wait_queue(&xprt->binding, "xprt_binding");
	rpc_init_wait_queue(&xprt->pending, "xprt_pending");
	rpc_init_wait_queue(&xprt->sending, "xprt_sending");
	rpc_init_wait_queue(&xprt->resend, "xprt_resend");
	rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog");

	/* initialize free list */
	for (req = &xprt->slot[xprt->max_reqs-1]; req >= &xprt->slot[0]; req--)
		list_add(&req->rq_list, &xprt->free);

	xprt_init_xid(xprt);

1035
	dprintk("RPC:       created transport %p with %u slots\n", xprt,
1036 1037 1038 1039 1040
			xprt->max_reqs);

	return xprt;
}

1041 1042
/**
 * xprt_destroy - destroy an RPC transport, killing off all requests.
1043
 * @kref: kref for the transport to destroy
1044
 *
L
Linus Torvalds 已提交
1045
 */
1046
static void xprt_destroy(struct kref *kref)
L
Linus Torvalds 已提交
1047
{
1048 1049
	struct rpc_xprt *xprt = container_of(kref, struct rpc_xprt, kref);

1050
	dprintk("RPC:       destroying transport %p\n", xprt);
1051 1052
	xprt->shutdown = 1;
	del_timer_sync(&xprt->timer);
1053 1054 1055 1056

	/*
	 * Tear down transport state and free the rpc_xprt
	 */
1057
	xprt->ops->destroy(xprt);
1058
}
L
Linus Torvalds 已提交
1059

1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078
/**
 * xprt_put - release a reference to an RPC transport.
 * @xprt: pointer to the transport
 *
 */
void xprt_put(struct rpc_xprt *xprt)
{
	kref_put(&xprt->kref, xprt_destroy);
}

/**
 * xprt_get - return a reference to an RPC transport.
 * @xprt: pointer to the transport
 *
 */
struct rpc_xprt *xprt_get(struct rpc_xprt *xprt)
{
	kref_get(&xprt->kref);
	return xprt;
L
Linus Torvalds 已提交
1079
}