xprt.c revision 0f9dc2b16884bb5957d010ed8e9114e771a05916
1/* 2 * linux/net/sunrpc/xprt.c 3 * 4 * This is a generic RPC call interface supporting congestion avoidance, 5 * and asynchronous calls. 6 * 7 * The interface works like this: 8 * 9 * - When a process places a call, it allocates a request slot if 10 * one is available. Otherwise, it sleeps on the backlog queue 11 * (xprt_reserve). 12 * - Next, the caller puts together the RPC message, stuffs it into 13 * the request struct, and calls xprt_call(). 14 * - xprt_call transmits the message and installs the caller on the 15 * socket's wait list. At the same time, it installs a timer that 16 * is run after the packet's timeout has expired. 17 * - When a packet arrives, the data_ready handler walks the list of 18 * pending requests for that socket. If a matching XID is found, the 19 * caller is woken up, and the timer removed. 20 * - When no reply arrives within the timeout interval, the timer is 21 * fired by the kernel and runs xprt_timer(). It either adjusts the 22 * timeout values (minor timeout) or wakes up the caller with a status 23 * of -ETIMEDOUT. 24 * - When the caller receives a notification from RPC that a reply arrived, 25 * it should release the RPC slot, and process the reply. 26 * If the call timed out, it may choose to retry the operation by 27 * adjusting the initial timeout value, and simply calling rpc_call 28 * again. 29 * 30 * Support for async RPC is done through a set of RPC-specific scheduling 31 * primitives that `transparently' work for processes as well as async 32 * tasks that rely on callbacks. 33 * 34 * Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de> 35 * 36 * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com> 37 * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com> 38 * TCP NFS related read + write fixes 39 * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie> 40 * 41 * Rewrite of larges part of the code in order to stabilize TCP stuff. 42 * Fix behaviour when socket buffer is full. 43 * (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no> 44 */ 45 46#include <linux/types.h> 47#include <linux/slab.h> 48#include <linux/capability.h> 49#include <linux/sched.h> 50#include <linux/errno.h> 51#include <linux/socket.h> 52#include <linux/in.h> 53#include <linux/net.h> 54#include <linux/mm.h> 55#include <linux/udp.h> 56#include <linux/tcp.h> 57#include <linux/sunrpc/clnt.h> 58#include <linux/file.h> 59#include <linux/workqueue.h> 60#include <linux/random.h> 61 62#include <net/sock.h> 63#include <net/checksum.h> 64#include <net/udp.h> 65#include <net/tcp.h> 66 67/* 68 * Local variables 69 */ 70 71#ifdef RPC_DEBUG 72# undef RPC_DEBUG_DATA 73# define RPCDBG_FACILITY RPCDBG_XPRT 74#endif 75 76#define XPRT_MAX_BACKOFF (8) 77#define XPRT_IDLE_TIMEOUT (5*60*HZ) 78#define XPRT_MAX_RESVPORT (800) 79 80/* 81 * Local functions 82 */ 83static void xprt_request_init(struct rpc_task *, struct rpc_xprt *); 84static inline void do_xprt_reserve(struct rpc_task *); 85static void xprt_disconnect(struct rpc_xprt *); 86static void xprt_connect_status(struct rpc_task *task); 87static struct rpc_xprt * xprt_setup(int proto, struct sockaddr_in *ap, 88 struct rpc_timeout *to); 89static struct socket *xprt_create_socket(struct rpc_xprt *, int, int); 90static void xprt_bind_socket(struct rpc_xprt *, struct socket *); 91static int __xprt_get_cong(struct rpc_xprt *, struct rpc_task *); 92 93static int xprt_clear_backlog(struct rpc_xprt *xprt); 94 95#ifdef RPC_DEBUG_DATA 96/* 97 * Print the buffer contents (first 128 bytes only--just enough for 98 * diropres return). 99 */ 100static void 101xprt_pktdump(char *msg, u32 *packet, unsigned int count) 102{ 103 u8 *buf = (u8 *) packet; 104 int j; 105 106 dprintk("RPC: %s\n", msg); 107 for (j = 0; j < count && j < 128; j += 4) { 108 if (!(j & 31)) { 109 if (j) 110 dprintk("\n"); 111 dprintk("0x%04x ", j); 112 } 113 dprintk("%02x%02x%02x%02x ", 114 buf[j], buf[j+1], buf[j+2], buf[j+3]); 115 } 116 dprintk("\n"); 117} 118#else 119static inline void 120xprt_pktdump(char *msg, u32 *packet, unsigned int count) 121{ 122 /* NOP */ 123} 124#endif 125 126/* 127 * Look up RPC transport given an INET socket 128 */ 129static inline struct rpc_xprt * 130xprt_from_sock(struct sock *sk) 131{ 132 return (struct rpc_xprt *) sk->sk_user_data; 133} 134 135/* 136 * Serialize write access to sockets, in order to prevent different 137 * requests from interfering with each other. 138 * Also prevents TCP socket connects from colliding with writes. 139 */ 140static int 141__xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task) 142{ 143 struct rpc_rqst *req = task->tk_rqstp; 144 145 if (test_and_set_bit(XPRT_LOCKED, &xprt->sockstate)) { 146 if (task == xprt->snd_task) 147 return 1; 148 if (task == NULL) 149 return 0; 150 goto out_sleep; 151 } 152 if (xprt->nocong || __xprt_get_cong(xprt, task)) { 153 xprt->snd_task = task; 154 if (req) { 155 req->rq_bytes_sent = 0; 156 req->rq_ntrans++; 157 } 158 return 1; 159 } 160 smp_mb__before_clear_bit(); 161 clear_bit(XPRT_LOCKED, &xprt->sockstate); 162 smp_mb__after_clear_bit(); 163out_sleep: 164 dprintk("RPC: %4d failed to lock socket %p\n", task->tk_pid, xprt); 165 task->tk_timeout = 0; 166 task->tk_status = -EAGAIN; 167 if (req && req->rq_ntrans) 168 rpc_sleep_on(&xprt->resend, task, NULL, NULL); 169 else 170 rpc_sleep_on(&xprt->sending, task, NULL, NULL); 171 return 0; 172} 173 174static inline int 175xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task) 176{ 177 int retval; 178 179 spin_lock_bh(&xprt->sock_lock); 180 retval = __xprt_lock_write(xprt, task); 181 spin_unlock_bh(&xprt->sock_lock); 182 return retval; 183} 184 185 186static void 187__xprt_lock_write_next(struct rpc_xprt *xprt) 188{ 189 struct rpc_task *task; 190 191 if (test_and_set_bit(XPRT_LOCKED, &xprt->sockstate)) 192 return; 193 if (!xprt->nocong && RPCXPRT_CONGESTED(xprt)) 194 goto out_unlock; 195 task = rpc_wake_up_next(&xprt->resend); 196 if (!task) { 197 task = rpc_wake_up_next(&xprt->sending); 198 if (!task) 199 goto out_unlock; 200 } 201 if (xprt->nocong || __xprt_get_cong(xprt, task)) { 202 struct rpc_rqst *req = task->tk_rqstp; 203 xprt->snd_task = task; 204 if (req) { 205 req->rq_bytes_sent = 0; 206 req->rq_ntrans++; 207 } 208 return; 209 } 210out_unlock: 211 smp_mb__before_clear_bit(); 212 clear_bit(XPRT_LOCKED, &xprt->sockstate); 213 smp_mb__after_clear_bit(); 214} 215 216/* 217 * Releases the socket for use by other requests. 218 */ 219static void 220__xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task) 221{ 222 if (xprt->snd_task == task) { 223 xprt->snd_task = NULL; 224 smp_mb__before_clear_bit(); 225 clear_bit(XPRT_LOCKED, &xprt->sockstate); 226 smp_mb__after_clear_bit(); 227 __xprt_lock_write_next(xprt); 228 } 229} 230 231static inline void 232xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task) 233{ 234 spin_lock_bh(&xprt->sock_lock); 235 __xprt_release_write(xprt, task); 236 spin_unlock_bh(&xprt->sock_lock); 237} 238 239/* 240 * Write data to socket. 241 */ 242static inline int 243xprt_sendmsg(struct rpc_xprt *xprt, struct rpc_rqst *req) 244{ 245 struct socket *sock = xprt->sock; 246 struct xdr_buf *xdr = &req->rq_snd_buf; 247 struct sockaddr *addr = NULL; 248 int addrlen = 0; 249 unsigned int skip; 250 int result; 251 252 if (!sock) 253 return -ENOTCONN; 254 255 xprt_pktdump("packet data:", 256 req->rq_svec->iov_base, 257 req->rq_svec->iov_len); 258 259 /* For UDP, we need to provide an address */ 260 if (!xprt->stream) { 261 addr = (struct sockaddr *) &xprt->addr; 262 addrlen = sizeof(xprt->addr); 263 } 264 /* Dont repeat bytes */ 265 skip = req->rq_bytes_sent; 266 267 clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags); 268 result = xdr_sendpages(sock, addr, addrlen, xdr, skip, MSG_DONTWAIT); 269 270 dprintk("RPC: xprt_sendmsg(%d) = %d\n", xdr->len - skip, result); 271 272 if (result >= 0) 273 return result; 274 275 switch (result) { 276 case -ECONNREFUSED: 277 /* When the server has died, an ICMP port unreachable message 278 * prompts ECONNREFUSED. 279 */ 280 case -EAGAIN: 281 break; 282 case -ECONNRESET: 283 case -ENOTCONN: 284 case -EPIPE: 285 /* connection broken */ 286 if (xprt->stream) 287 result = -ENOTCONN; 288 break; 289 default: 290 printk(KERN_NOTICE "RPC: sendmsg returned error %d\n", -result); 291 } 292 return result; 293} 294 295/* 296 * Van Jacobson congestion avoidance. Check if the congestion window 297 * overflowed. Put the task to sleep if this is the case. 298 */ 299static int 300__xprt_get_cong(struct rpc_xprt *xprt, struct rpc_task *task) 301{ 302 struct rpc_rqst *req = task->tk_rqstp; 303 304 if (req->rq_cong) 305 return 1; 306 dprintk("RPC: %4d xprt_cwnd_limited cong = %ld cwnd = %ld\n", 307 task->tk_pid, xprt->cong, xprt->cwnd); 308 if (RPCXPRT_CONGESTED(xprt)) 309 return 0; 310 req->rq_cong = 1; 311 xprt->cong += RPC_CWNDSCALE; 312 return 1; 313} 314 315/* 316 * Adjust the congestion window, and wake up the next task 317 * that has been sleeping due to congestion 318 */ 319static void 320__xprt_put_cong(struct rpc_xprt *xprt, struct rpc_rqst *req) 321{ 322 if (!req->rq_cong) 323 return; 324 req->rq_cong = 0; 325 xprt->cong -= RPC_CWNDSCALE; 326 __xprt_lock_write_next(xprt); 327} 328 329/* 330 * Adjust RPC congestion window 331 * We use a time-smoothed congestion estimator to avoid heavy oscillation. 332 */ 333static void 334xprt_adjust_cwnd(struct rpc_xprt *xprt, int result) 335{ 336 unsigned long cwnd; 337 338 cwnd = xprt->cwnd; 339 if (result >= 0 && cwnd <= xprt->cong) { 340 /* The (cwnd >> 1) term makes sure 341 * the result gets rounded properly. */ 342 cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd; 343 if (cwnd > RPC_MAXCWND(xprt)) 344 cwnd = RPC_MAXCWND(xprt); 345 __xprt_lock_write_next(xprt); 346 } else if (result == -ETIMEDOUT) { 347 cwnd >>= 1; 348 if (cwnd < RPC_CWNDSCALE) 349 cwnd = RPC_CWNDSCALE; 350 } 351 dprintk("RPC: cong %ld, cwnd was %ld, now %ld\n", 352 xprt->cong, xprt->cwnd, cwnd); 353 xprt->cwnd = cwnd; 354} 355 356/* 357 * Reset the major timeout value 358 */ 359static void xprt_reset_majortimeo(struct rpc_rqst *req) 360{ 361 struct rpc_timeout *to = &req->rq_xprt->timeout; 362 363 req->rq_majortimeo = req->rq_timeout; 364 if (to->to_exponential) 365 req->rq_majortimeo <<= to->to_retries; 366 else 367 req->rq_majortimeo += to->to_increment * to->to_retries; 368 if (req->rq_majortimeo > to->to_maxval || req->rq_majortimeo == 0) 369 req->rq_majortimeo = to->to_maxval; 370 req->rq_majortimeo += jiffies; 371} 372 373/* 374 * Adjust timeout values etc for next retransmit 375 */ 376int xprt_adjust_timeout(struct rpc_rqst *req) 377{ 378 struct rpc_xprt *xprt = req->rq_xprt; 379 struct rpc_timeout *to = &xprt->timeout; 380 int status = 0; 381 382 if (time_before(jiffies, req->rq_majortimeo)) { 383 if (to->to_exponential) 384 req->rq_timeout <<= 1; 385 else 386 req->rq_timeout += to->to_increment; 387 if (to->to_maxval && req->rq_timeout >= to->to_maxval) 388 req->rq_timeout = to->to_maxval; 389 req->rq_retries++; 390 pprintk("RPC: %lu retrans\n", jiffies); 391 } else { 392 req->rq_timeout = to->to_initval; 393 req->rq_retries = 0; 394 xprt_reset_majortimeo(req); 395 /* Reset the RTT counters == "slow start" */ 396 spin_lock_bh(&xprt->sock_lock); 397 rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval); 398 spin_unlock_bh(&xprt->sock_lock); 399 pprintk("RPC: %lu timeout\n", jiffies); 400 status = -ETIMEDOUT; 401 } 402 403 if (req->rq_timeout == 0) { 404 printk(KERN_WARNING "xprt_adjust_timeout: rq_timeout = 0!\n"); 405 req->rq_timeout = 5 * HZ; 406 } 407 return status; 408} 409 410/* 411 * Close down a transport socket 412 */ 413static void 414xprt_close(struct rpc_xprt *xprt) 415{ 416 struct socket *sock = xprt->sock; 417 struct sock *sk = xprt->inet; 418 419 if (!sk) 420 return; 421 422 write_lock_bh(&sk->sk_callback_lock); 423 xprt->inet = NULL; 424 xprt->sock = NULL; 425 426 sk->sk_user_data = NULL; 427 sk->sk_data_ready = xprt->old_data_ready; 428 sk->sk_state_change = xprt->old_state_change; 429 sk->sk_write_space = xprt->old_write_space; 430 write_unlock_bh(&sk->sk_callback_lock); 431 432 sk->sk_no_check = 0; 433 434 sock_release(sock); 435} 436 437static void 438xprt_socket_autoclose(void *args) 439{ 440 struct rpc_xprt *xprt = (struct rpc_xprt *)args; 441 442 xprt_disconnect(xprt); 443 xprt_close(xprt); 444 xprt_release_write(xprt, NULL); 445} 446 447/* 448 * Mark a transport as disconnected 449 */ 450static void 451xprt_disconnect(struct rpc_xprt *xprt) 452{ 453 dprintk("RPC: disconnected transport %p\n", xprt); 454 spin_lock_bh(&xprt->sock_lock); 455 xprt_clear_connected(xprt); 456 rpc_wake_up_status(&xprt->pending, -ENOTCONN); 457 spin_unlock_bh(&xprt->sock_lock); 458} 459 460/* 461 * Used to allow disconnection when we've been idle 462 */ 463static void 464xprt_init_autodisconnect(unsigned long data) 465{ 466 struct rpc_xprt *xprt = (struct rpc_xprt *)data; 467 468 spin_lock(&xprt->sock_lock); 469 if (!list_empty(&xprt->recv) || xprt->shutdown) 470 goto out_abort; 471 if (test_and_set_bit(XPRT_LOCKED, &xprt->sockstate)) 472 goto out_abort; 473 spin_unlock(&xprt->sock_lock); 474 /* Let keventd close the socket */ 475 if (test_bit(XPRT_CONNECTING, &xprt->sockstate) != 0) 476 xprt_release_write(xprt, NULL); 477 else 478 schedule_work(&xprt->task_cleanup); 479 return; 480out_abort: 481 spin_unlock(&xprt->sock_lock); 482} 483 484static void xprt_socket_connect(void *args) 485{ 486 struct rpc_xprt *xprt = (struct rpc_xprt *)args; 487 struct socket *sock = xprt->sock; 488 int status = -EIO; 489 490 if (xprt->shutdown || xprt->addr.sin_port == 0) 491 goto out; 492 493 /* 494 * Start by resetting any existing state 495 */ 496 xprt_close(xprt); 497 sock = xprt_create_socket(xprt, xprt->prot, xprt->resvport); 498 if (sock == NULL) { 499 /* couldn't create socket or bind to reserved port; 500 * this is likely a permanent error, so cause an abort */ 501 goto out; 502 } 503 xprt_bind_socket(xprt, sock); 504 xprt_sock_setbufsize(xprt); 505 506 status = 0; 507 if (!xprt->stream) 508 goto out; 509 510 /* 511 * Tell the socket layer to start connecting... 512 */ 513 status = sock->ops->connect(sock, (struct sockaddr *) &xprt->addr, 514 sizeof(xprt->addr), O_NONBLOCK); 515 dprintk("RPC: %p connect status %d connected %d sock state %d\n", 516 xprt, -status, xprt_connected(xprt), sock->sk->sk_state); 517 if (status < 0) { 518 switch (status) { 519 case -EINPROGRESS: 520 case -EALREADY: 521 goto out_clear; 522 } 523 } 524out: 525 if (status < 0) 526 rpc_wake_up_status(&xprt->pending, status); 527 else 528 rpc_wake_up(&xprt->pending); 529out_clear: 530 smp_mb__before_clear_bit(); 531 clear_bit(XPRT_CONNECTING, &xprt->sockstate); 532 smp_mb__after_clear_bit(); 533} 534 535/* 536 * Attempt to connect a TCP socket. 537 * 538 */ 539void xprt_connect(struct rpc_task *task) 540{ 541 struct rpc_xprt *xprt = task->tk_xprt; 542 543 dprintk("RPC: %4d xprt_connect xprt %p %s connected\n", task->tk_pid, 544 xprt, (xprt_connected(xprt) ? "is" : "is not")); 545 546 if (xprt->shutdown) { 547 task->tk_status = -EIO; 548 return; 549 } 550 if (!xprt->addr.sin_port) { 551 task->tk_status = -EIO; 552 return; 553 } 554 if (!xprt_lock_write(xprt, task)) 555 return; 556 if (xprt_connected(xprt)) 557 goto out_write; 558 559 if (task->tk_rqstp) 560 task->tk_rqstp->rq_bytes_sent = 0; 561 562 task->tk_timeout = RPC_CONNECT_TIMEOUT; 563 rpc_sleep_on(&xprt->pending, task, xprt_connect_status, NULL); 564 if (!test_and_set_bit(XPRT_CONNECTING, &xprt->sockstate)) { 565 /* Note: if we are here due to a dropped connection 566 * we delay reconnecting by RPC_REESTABLISH_TIMEOUT/HZ 567 * seconds 568 */ 569 if (xprt->sock != NULL) 570 schedule_delayed_work(&xprt->sock_connect, 571 RPC_REESTABLISH_TIMEOUT); 572 else 573 schedule_work(&xprt->sock_connect); 574 } 575 return; 576 out_write: 577 xprt_release_write(xprt, task); 578} 579 580/* 581 * We arrive here when awoken from waiting on connection establishment. 582 */ 583static void 584xprt_connect_status(struct rpc_task *task) 585{ 586 struct rpc_xprt *xprt = task->tk_xprt; 587 588 if (task->tk_status >= 0) { 589 dprintk("RPC: %4d xprt_connect_status: connection established\n", 590 task->tk_pid); 591 return; 592 } 593 594 /* if soft mounted, just cause this RPC to fail */ 595 if (RPC_IS_SOFT(task)) 596 task->tk_status = -EIO; 597 598 switch (task->tk_status) { 599 case -ECONNREFUSED: 600 case -ECONNRESET: 601 case -ENOTCONN: 602 return; 603 case -ETIMEDOUT: 604 dprintk("RPC: %4d xprt_connect_status: timed out\n", 605 task->tk_pid); 606 break; 607 default: 608 printk(KERN_ERR "RPC: error %d connecting to server %s\n", 609 -task->tk_status, task->tk_client->cl_server); 610 } 611 xprt_release_write(xprt, task); 612} 613 614/* 615 * Look up the RPC request corresponding to a reply, and then lock it. 616 */ 617static inline struct rpc_rqst * 618xprt_lookup_rqst(struct rpc_xprt *xprt, u32 xid) 619{ 620 struct list_head *pos; 621 struct rpc_rqst *req = NULL; 622 623 list_for_each(pos, &xprt->recv) { 624 struct rpc_rqst *entry = list_entry(pos, struct rpc_rqst, rq_list); 625 if (entry->rq_xid == xid) { 626 req = entry; 627 break; 628 } 629 } 630 return req; 631} 632 633/* 634 * Complete reply received. 635 * The TCP code relies on us to remove the request from xprt->pending. 636 */ 637static void 638xprt_complete_rqst(struct rpc_xprt *xprt, struct rpc_rqst *req, int copied) 639{ 640 struct rpc_task *task = req->rq_task; 641 struct rpc_clnt *clnt = task->tk_client; 642 643 /* Adjust congestion window */ 644 if (!xprt->nocong) { 645 unsigned timer = task->tk_msg.rpc_proc->p_timer; 646 xprt_adjust_cwnd(xprt, copied); 647 __xprt_put_cong(xprt, req); 648 if (timer) { 649 if (req->rq_ntrans == 1) 650 rpc_update_rtt(clnt->cl_rtt, timer, 651 (long)jiffies - req->rq_xtime); 652 rpc_set_timeo(clnt->cl_rtt, timer, req->rq_ntrans - 1); 653 } 654 } 655 656#ifdef RPC_PROFILE 657 /* Profile only reads for now */ 658 if (copied > 1024) { 659 static unsigned long nextstat; 660 static unsigned long pkt_rtt, pkt_len, pkt_cnt; 661 662 pkt_cnt++; 663 pkt_len += req->rq_slen + copied; 664 pkt_rtt += jiffies - req->rq_xtime; 665 if (time_before(nextstat, jiffies)) { 666 printk("RPC: %lu %ld cwnd\n", jiffies, xprt->cwnd); 667 printk("RPC: %ld %ld %ld %ld stat\n", 668 jiffies, pkt_cnt, pkt_len, pkt_rtt); 669 pkt_rtt = pkt_len = pkt_cnt = 0; 670 nextstat = jiffies + 5 * HZ; 671 } 672 } 673#endif 674 675 dprintk("RPC: %4d has input (%d bytes)\n", task->tk_pid, copied); 676 list_del_init(&req->rq_list); 677 req->rq_received = req->rq_private_buf.len = copied; 678 679 /* ... and wake up the process. */ 680 rpc_wake_up_task(task); 681 return; 682} 683 684static size_t 685skb_read_bits(skb_reader_t *desc, void *to, size_t len) 686{ 687 if (len > desc->count) 688 len = desc->count; 689 if (skb_copy_bits(desc->skb, desc->offset, to, len)) 690 return 0; 691 desc->count -= len; 692 desc->offset += len; 693 return len; 694} 695 696static size_t 697skb_read_and_csum_bits(skb_reader_t *desc, void *to, size_t len) 698{ 699 unsigned int csum2, pos; 700 701 if (len > desc->count) 702 len = desc->count; 703 pos = desc->offset; 704 csum2 = skb_copy_and_csum_bits(desc->skb, pos, to, len, 0); 705 desc->csum = csum_block_add(desc->csum, csum2, pos); 706 desc->count -= len; 707 desc->offset += len; 708 return len; 709} 710 711/* 712 * We have set things up such that we perform the checksum of the UDP 713 * packet in parallel with the copies into the RPC client iovec. -DaveM 714 */ 715int 716csum_partial_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb) 717{ 718 skb_reader_t desc; 719 720 desc.skb = skb; 721 desc.offset = sizeof(struct udphdr); 722 desc.count = skb->len - desc.offset; 723 724 if (skb->ip_summed == CHECKSUM_UNNECESSARY) 725 goto no_checksum; 726 727 desc.csum = csum_partial(skb->data, desc.offset, skb->csum); 728 if (xdr_partial_copy_from_skb(xdr, 0, &desc, skb_read_and_csum_bits) < 0) 729 return -1; 730 if (desc.offset != skb->len) { 731 unsigned int csum2; 732 csum2 = skb_checksum(skb, desc.offset, skb->len - desc.offset, 0); 733 desc.csum = csum_block_add(desc.csum, csum2, desc.offset); 734 } 735 if (desc.count) 736 return -1; 737 if ((unsigned short)csum_fold(desc.csum)) 738 return -1; 739 return 0; 740no_checksum: 741 if (xdr_partial_copy_from_skb(xdr, 0, &desc, skb_read_bits) < 0) 742 return -1; 743 if (desc.count) 744 return -1; 745 return 0; 746} 747 748/* 749 * Input handler for RPC replies. Called from a bottom half and hence 750 * atomic. 751 */ 752static void 753udp_data_ready(struct sock *sk, int len) 754{ 755 struct rpc_task *task; 756 struct rpc_xprt *xprt; 757 struct rpc_rqst *rovr; 758 struct sk_buff *skb; 759 int err, repsize, copied; 760 u32 _xid, *xp; 761 762 read_lock(&sk->sk_callback_lock); 763 dprintk("RPC: udp_data_ready...\n"); 764 if (!(xprt = xprt_from_sock(sk))) { 765 printk("RPC: udp_data_ready request not found!\n"); 766 goto out; 767 } 768 769 dprintk("RPC: udp_data_ready client %p\n", xprt); 770 771 if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL) 772 goto out; 773 774 if (xprt->shutdown) 775 goto dropit; 776 777 repsize = skb->len - sizeof(struct udphdr); 778 if (repsize < 4) { 779 printk("RPC: impossible RPC reply size %d!\n", repsize); 780 goto dropit; 781 } 782 783 /* Copy the XID from the skb... */ 784 xp = skb_header_pointer(skb, sizeof(struct udphdr), 785 sizeof(_xid), &_xid); 786 if (xp == NULL) 787 goto dropit; 788 789 /* Look up and lock the request corresponding to the given XID */ 790 spin_lock(&xprt->sock_lock); 791 rovr = xprt_lookup_rqst(xprt, *xp); 792 if (!rovr) 793 goto out_unlock; 794 task = rovr->rq_task; 795 796 dprintk("RPC: %4d received reply\n", task->tk_pid); 797 798 if ((copied = rovr->rq_private_buf.buflen) > repsize) 799 copied = repsize; 800 801 /* Suck it into the iovec, verify checksum if not done by hw. */ 802 if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) 803 goto out_unlock; 804 805 /* Something worked... */ 806 dst_confirm(skb->dst); 807 808 xprt_complete_rqst(xprt, rovr, copied); 809 810 out_unlock: 811 spin_unlock(&xprt->sock_lock); 812 dropit: 813 skb_free_datagram(sk, skb); 814 out: 815 read_unlock(&sk->sk_callback_lock); 816} 817 818/* 819 * Copy from an skb into memory and shrink the skb. 820 */ 821static inline size_t 822tcp_copy_data(skb_reader_t *desc, void *p, size_t len) 823{ 824 if (len > desc->count) 825 len = desc->count; 826 if (skb_copy_bits(desc->skb, desc->offset, p, len)) { 827 dprintk("RPC: failed to copy %zu bytes from skb. %zu bytes remain\n", 828 len, desc->count); 829 return 0; 830 } 831 desc->offset += len; 832 desc->count -= len; 833 dprintk("RPC: copied %zu bytes from skb. %zu bytes remain\n", 834 len, desc->count); 835 return len; 836} 837 838/* 839 * TCP read fragment marker 840 */ 841static inline void 842tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc) 843{ 844 size_t len, used; 845 char *p; 846 847 p = ((char *) &xprt->tcp_recm) + xprt->tcp_offset; 848 len = sizeof(xprt->tcp_recm) - xprt->tcp_offset; 849 used = tcp_copy_data(desc, p, len); 850 xprt->tcp_offset += used; 851 if (used != len) 852 return; 853 xprt->tcp_reclen = ntohl(xprt->tcp_recm); 854 if (xprt->tcp_reclen & 0x80000000) 855 xprt->tcp_flags |= XPRT_LAST_FRAG; 856 else 857 xprt->tcp_flags &= ~XPRT_LAST_FRAG; 858 xprt->tcp_reclen &= 0x7fffffff; 859 xprt->tcp_flags &= ~XPRT_COPY_RECM; 860 xprt->tcp_offset = 0; 861 /* Sanity check of the record length */ 862 if (xprt->tcp_reclen < 4) { 863 printk(KERN_ERR "RPC: Invalid TCP record fragment length\n"); 864 xprt_disconnect(xprt); 865 } 866 dprintk("RPC: reading TCP record fragment of length %d\n", 867 xprt->tcp_reclen); 868} 869 870static void 871tcp_check_recm(struct rpc_xprt *xprt) 872{ 873 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u, tcp_flags = %lx\n", 874 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen, xprt->tcp_flags); 875 if (xprt->tcp_offset == xprt->tcp_reclen) { 876 xprt->tcp_flags |= XPRT_COPY_RECM; 877 xprt->tcp_offset = 0; 878 if (xprt->tcp_flags & XPRT_LAST_FRAG) { 879 xprt->tcp_flags &= ~XPRT_COPY_DATA; 880 xprt->tcp_flags |= XPRT_COPY_XID; 881 xprt->tcp_copied = 0; 882 } 883 } 884} 885 886/* 887 * TCP read xid 888 */ 889static inline void 890tcp_read_xid(struct rpc_xprt *xprt, skb_reader_t *desc) 891{ 892 size_t len, used; 893 char *p; 894 895 len = sizeof(xprt->tcp_xid) - xprt->tcp_offset; 896 dprintk("RPC: reading XID (%Zu bytes)\n", len); 897 p = ((char *) &xprt->tcp_xid) + xprt->tcp_offset; 898 used = tcp_copy_data(desc, p, len); 899 xprt->tcp_offset += used; 900 if (used != len) 901 return; 902 xprt->tcp_flags &= ~XPRT_COPY_XID; 903 xprt->tcp_flags |= XPRT_COPY_DATA; 904 xprt->tcp_copied = 4; 905 dprintk("RPC: reading reply for XID %08x\n", 906 ntohl(xprt->tcp_xid)); 907 tcp_check_recm(xprt); 908} 909 910/* 911 * TCP read and complete request 912 */ 913static inline void 914tcp_read_request(struct rpc_xprt *xprt, skb_reader_t *desc) 915{ 916 struct rpc_rqst *req; 917 struct xdr_buf *rcvbuf; 918 size_t len; 919 ssize_t r; 920 921 /* Find and lock the request corresponding to this xid */ 922 spin_lock(&xprt->sock_lock); 923 req = xprt_lookup_rqst(xprt, xprt->tcp_xid); 924 if (!req) { 925 xprt->tcp_flags &= ~XPRT_COPY_DATA; 926 dprintk("RPC: XID %08x request not found!\n", 927 ntohl(xprt->tcp_xid)); 928 spin_unlock(&xprt->sock_lock); 929 return; 930 } 931 932 rcvbuf = &req->rq_private_buf; 933 len = desc->count; 934 if (len > xprt->tcp_reclen - xprt->tcp_offset) { 935 skb_reader_t my_desc; 936 937 len = xprt->tcp_reclen - xprt->tcp_offset; 938 memcpy(&my_desc, desc, sizeof(my_desc)); 939 my_desc.count = len; 940 r = xdr_partial_copy_from_skb(rcvbuf, xprt->tcp_copied, 941 &my_desc, tcp_copy_data); 942 desc->count -= r; 943 desc->offset += r; 944 } else 945 r = xdr_partial_copy_from_skb(rcvbuf, xprt->tcp_copied, 946 desc, tcp_copy_data); 947 948 if (r > 0) { 949 xprt->tcp_copied += r; 950 xprt->tcp_offset += r; 951 } 952 if (r != len) { 953 /* Error when copying to the receive buffer, 954 * usually because we weren't able to allocate 955 * additional buffer pages. All we can do now 956 * is turn off XPRT_COPY_DATA, so the request 957 * will not receive any additional updates, 958 * and time out. 959 * Any remaining data from this record will 960 * be discarded. 961 */ 962 xprt->tcp_flags &= ~XPRT_COPY_DATA; 963 dprintk("RPC: XID %08x truncated request\n", 964 ntohl(xprt->tcp_xid)); 965 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n", 966 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen); 967 goto out; 968 } 969 970 dprintk("RPC: XID %08x read %u bytes\n", 971 ntohl(xprt->tcp_xid), r); 972 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n", 973 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen); 974 975 if (xprt->tcp_copied == req->rq_private_buf.buflen) 976 xprt->tcp_flags &= ~XPRT_COPY_DATA; 977 else if (xprt->tcp_offset == xprt->tcp_reclen) { 978 if (xprt->tcp_flags & XPRT_LAST_FRAG) 979 xprt->tcp_flags &= ~XPRT_COPY_DATA; 980 } 981 982out: 983 if (!(xprt->tcp_flags & XPRT_COPY_DATA)) { 984 dprintk("RPC: %4d received reply complete\n", 985 req->rq_task->tk_pid); 986 xprt_complete_rqst(xprt, req, xprt->tcp_copied); 987 } 988 spin_unlock(&xprt->sock_lock); 989 tcp_check_recm(xprt); 990} 991 992/* 993 * TCP discard extra bytes from a short read 994 */ 995static inline void 996tcp_read_discard(struct rpc_xprt *xprt, skb_reader_t *desc) 997{ 998 size_t len; 999 1000 len = xprt->tcp_reclen - xprt->tcp_offset; 1001 if (len > desc->count) 1002 len = desc->count; 1003 desc->count -= len; 1004 desc->offset += len; 1005 xprt->tcp_offset += len; 1006 dprintk("RPC: discarded %u bytes\n", len); 1007 tcp_check_recm(xprt); 1008} 1009 1010/* 1011 * TCP record receive routine 1012 * We first have to grab the record marker, then the XID, then the data. 1013 */ 1014static int 1015tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, 1016 unsigned int offset, size_t len) 1017{ 1018 struct rpc_xprt *xprt = rd_desc->arg.data; 1019 skb_reader_t desc = { 1020 .skb = skb, 1021 .offset = offset, 1022 .count = len, 1023 .csum = 0 1024 }; 1025 1026 dprintk("RPC: tcp_data_recv\n"); 1027 do { 1028 /* Read in a new fragment marker if necessary */ 1029 /* Can we ever really expect to get completely empty fragments? */ 1030 if (xprt->tcp_flags & XPRT_COPY_RECM) { 1031 tcp_read_fraghdr(xprt, &desc); 1032 continue; 1033 } 1034 /* Read in the xid if necessary */ 1035 if (xprt->tcp_flags & XPRT_COPY_XID) { 1036 tcp_read_xid(xprt, &desc); 1037 continue; 1038 } 1039 /* Read in the request data */ 1040 if (xprt->tcp_flags & XPRT_COPY_DATA) { 1041 tcp_read_request(xprt, &desc); 1042 continue; 1043 } 1044 /* Skip over any trailing bytes on short reads */ 1045 tcp_read_discard(xprt, &desc); 1046 } while (desc.count); 1047 dprintk("RPC: tcp_data_recv done\n"); 1048 return len - desc.count; 1049} 1050 1051static void tcp_data_ready(struct sock *sk, int bytes) 1052{ 1053 struct rpc_xprt *xprt; 1054 read_descriptor_t rd_desc; 1055 1056 read_lock(&sk->sk_callback_lock); 1057 dprintk("RPC: tcp_data_ready...\n"); 1058 if (!(xprt = xprt_from_sock(sk))) { 1059 printk("RPC: tcp_data_ready socket info not found!\n"); 1060 goto out; 1061 } 1062 if (xprt->shutdown) 1063 goto out; 1064 1065 /* We use rd_desc to pass struct xprt to tcp_data_recv */ 1066 rd_desc.arg.data = xprt; 1067 rd_desc.count = 65536; 1068 tcp_read_sock(sk, &rd_desc, tcp_data_recv); 1069out: 1070 read_unlock(&sk->sk_callback_lock); 1071} 1072 1073static void 1074tcp_state_change(struct sock *sk) 1075{ 1076 struct rpc_xprt *xprt; 1077 1078 read_lock(&sk->sk_callback_lock); 1079 if (!(xprt = xprt_from_sock(sk))) 1080 goto out; 1081 dprintk("RPC: tcp_state_change client %p...\n", xprt); 1082 dprintk("RPC: state %x conn %d dead %d zapped %d\n", 1083 sk->sk_state, xprt_connected(xprt), 1084 sock_flag(sk, SOCK_DEAD), 1085 sock_flag(sk, SOCK_ZAPPED)); 1086 1087 switch (sk->sk_state) { 1088 case TCP_ESTABLISHED: 1089 spin_lock_bh(&xprt->sock_lock); 1090 if (!xprt_test_and_set_connected(xprt)) { 1091 /* Reset TCP record info */ 1092 xprt->tcp_offset = 0; 1093 xprt->tcp_reclen = 0; 1094 xprt->tcp_copied = 0; 1095 xprt->tcp_flags = XPRT_COPY_RECM | XPRT_COPY_XID; 1096 rpc_wake_up(&xprt->pending); 1097 } 1098 spin_unlock_bh(&xprt->sock_lock); 1099 break; 1100 case TCP_SYN_SENT: 1101 case TCP_SYN_RECV: 1102 break; 1103 default: 1104 if (xprt_test_and_clear_connected(xprt)) 1105 rpc_wake_up_status(&xprt->pending, -ENOTCONN); 1106 break; 1107 } 1108 out: 1109 read_unlock(&sk->sk_callback_lock); 1110} 1111 1112/* 1113 * Called when more output buffer space is available for this socket. 1114 * We try not to wake our writers until they can make "significant" 1115 * progress, otherwise we'll waste resources thrashing sock_sendmsg 1116 * with a bunch of small requests. 1117 */ 1118static void 1119xprt_write_space(struct sock *sk) 1120{ 1121 struct rpc_xprt *xprt; 1122 struct socket *sock; 1123 1124 read_lock(&sk->sk_callback_lock); 1125 if (!(xprt = xprt_from_sock(sk)) || !(sock = sk->sk_socket)) 1126 goto out; 1127 if (xprt->shutdown) 1128 goto out; 1129 1130 /* Wait until we have enough socket memory */ 1131 if (xprt->stream) { 1132 /* from net/core/stream.c:sk_stream_write_space */ 1133 if (sk_stream_wspace(sk) < sk_stream_min_wspace(sk)) 1134 goto out; 1135 } else { 1136 /* from net/core/sock.c:sock_def_write_space */ 1137 if (!sock_writeable(sk)) 1138 goto out; 1139 } 1140 1141 if (!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)) 1142 goto out; 1143 1144 spin_lock_bh(&xprt->sock_lock); 1145 if (xprt->snd_task) 1146 rpc_wake_up_task(xprt->snd_task); 1147 spin_unlock_bh(&xprt->sock_lock); 1148out: 1149 read_unlock(&sk->sk_callback_lock); 1150} 1151 1152/* 1153 * RPC receive timeout handler. 1154 */ 1155static void 1156xprt_timer(struct rpc_task *task) 1157{ 1158 struct rpc_rqst *req = task->tk_rqstp; 1159 struct rpc_xprt *xprt = req->rq_xprt; 1160 1161 spin_lock(&xprt->sock_lock); 1162 if (req->rq_received) 1163 goto out; 1164 1165 xprt_adjust_cwnd(req->rq_xprt, -ETIMEDOUT); 1166 __xprt_put_cong(xprt, req); 1167 1168 dprintk("RPC: %4d xprt_timer (%s request)\n", 1169 task->tk_pid, req ? "pending" : "backlogged"); 1170 1171 task->tk_status = -ETIMEDOUT; 1172out: 1173 task->tk_timeout = 0; 1174 rpc_wake_up_task(task); 1175 spin_unlock(&xprt->sock_lock); 1176} 1177 1178/* 1179 * Place the actual RPC call. 1180 * We have to copy the iovec because sendmsg fiddles with its contents. 1181 */ 1182int 1183xprt_prepare_transmit(struct rpc_task *task) 1184{ 1185 struct rpc_rqst *req = task->tk_rqstp; 1186 struct rpc_xprt *xprt = req->rq_xprt; 1187 int err = 0; 1188 1189 dprintk("RPC: %4d xprt_prepare_transmit\n", task->tk_pid); 1190 1191 if (xprt->shutdown) 1192 return -EIO; 1193 1194 spin_lock_bh(&xprt->sock_lock); 1195 if (req->rq_received && !req->rq_bytes_sent) { 1196 err = req->rq_received; 1197 goto out_unlock; 1198 } 1199 if (!__xprt_lock_write(xprt, task)) { 1200 err = -EAGAIN; 1201 goto out_unlock; 1202 } 1203 1204 if (!xprt_connected(xprt)) { 1205 err = -ENOTCONN; 1206 goto out_unlock; 1207 } 1208out_unlock: 1209 spin_unlock_bh(&xprt->sock_lock); 1210 return err; 1211} 1212 1213void 1214xprt_transmit(struct rpc_task *task) 1215{ 1216 struct rpc_clnt *clnt = task->tk_client; 1217 struct rpc_rqst *req = task->tk_rqstp; 1218 struct rpc_xprt *xprt = req->rq_xprt; 1219 int status, retry = 0; 1220 1221 1222 dprintk("RPC: %4d xprt_transmit(%u)\n", task->tk_pid, req->rq_slen); 1223 1224 /* set up everything as needed. */ 1225 /* Write the record marker */ 1226 if (xprt->stream) { 1227 u32 *marker = req->rq_svec[0].iov_base; 1228 1229 *marker = htonl(0x80000000|(req->rq_slen-sizeof(*marker))); 1230 } 1231 1232 smp_rmb(); 1233 if (!req->rq_received) { 1234 if (list_empty(&req->rq_list)) { 1235 spin_lock_bh(&xprt->sock_lock); 1236 /* Update the softirq receive buffer */ 1237 memcpy(&req->rq_private_buf, &req->rq_rcv_buf, 1238 sizeof(req->rq_private_buf)); 1239 /* Add request to the receive list */ 1240 list_add_tail(&req->rq_list, &xprt->recv); 1241 spin_unlock_bh(&xprt->sock_lock); 1242 xprt_reset_majortimeo(req); 1243 /* Turn off autodisconnect */ 1244 del_singleshot_timer_sync(&xprt->timer); 1245 } 1246 } else if (!req->rq_bytes_sent) 1247 return; 1248 1249 /* Continue transmitting the packet/record. We must be careful 1250 * to cope with writespace callbacks arriving _after_ we have 1251 * called xprt_sendmsg(). 1252 */ 1253 while (1) { 1254 req->rq_xtime = jiffies; 1255 status = xprt_sendmsg(xprt, req); 1256 1257 if (status < 0) 1258 break; 1259 1260 if (xprt->stream) { 1261 req->rq_bytes_sent += status; 1262 1263 /* If we've sent the entire packet, immediately 1264 * reset the count of bytes sent. */ 1265 if (req->rq_bytes_sent >= req->rq_slen) { 1266 req->rq_bytes_sent = 0; 1267 goto out_receive; 1268 } 1269 } else { 1270 if (status >= req->rq_slen) 1271 goto out_receive; 1272 status = -EAGAIN; 1273 break; 1274 } 1275 1276 dprintk("RPC: %4d xmit incomplete (%d left of %d)\n", 1277 task->tk_pid, req->rq_slen - req->rq_bytes_sent, 1278 req->rq_slen); 1279 1280 status = -EAGAIN; 1281 if (retry++ > 50) 1282 break; 1283 } 1284 1285 /* Note: at this point, task->tk_sleeping has not yet been set, 1286 * hence there is no danger of the waking up task being put on 1287 * schedq, and being picked up by a parallel run of rpciod(). 1288 */ 1289 task->tk_status = status; 1290 1291 switch (status) { 1292 case -EAGAIN: 1293 if (test_bit(SOCK_ASYNC_NOSPACE, &xprt->sock->flags)) { 1294 /* Protect against races with xprt_write_space */ 1295 spin_lock_bh(&xprt->sock_lock); 1296 /* Don't race with disconnect */ 1297 if (!xprt_connected(xprt)) 1298 task->tk_status = -ENOTCONN; 1299 else if (test_bit(SOCK_NOSPACE, &xprt->sock->flags)) { 1300 task->tk_timeout = req->rq_timeout; 1301 rpc_sleep_on(&xprt->pending, task, NULL, NULL); 1302 } 1303 spin_unlock_bh(&xprt->sock_lock); 1304 return; 1305 } 1306 /* Keep holding the socket if it is blocked */ 1307 rpc_delay(task, HZ>>4); 1308 return; 1309 case -ECONNREFUSED: 1310 task->tk_timeout = RPC_REESTABLISH_TIMEOUT; 1311 rpc_sleep_on(&xprt->sending, task, NULL, NULL); 1312 case -ENOTCONN: 1313 return; 1314 default: 1315 if (xprt->stream) 1316 xprt_disconnect(xprt); 1317 } 1318 xprt_release_write(xprt, task); 1319 return; 1320 out_receive: 1321 dprintk("RPC: %4d xmit complete\n", task->tk_pid); 1322 /* Set the task's receive timeout value */ 1323 spin_lock_bh(&xprt->sock_lock); 1324 if (!xprt->nocong) { 1325 int timer = task->tk_msg.rpc_proc->p_timer; 1326 task->tk_timeout = rpc_calc_rto(clnt->cl_rtt, timer); 1327 task->tk_timeout <<= rpc_ntimeo(clnt->cl_rtt, timer) + req->rq_retries; 1328 if (task->tk_timeout > xprt->timeout.to_maxval || task->tk_timeout == 0) 1329 task->tk_timeout = xprt->timeout.to_maxval; 1330 } else 1331 task->tk_timeout = req->rq_timeout; 1332 /* Don't race with disconnect */ 1333 if (!xprt_connected(xprt)) 1334 task->tk_status = -ENOTCONN; 1335 else if (!req->rq_received) 1336 rpc_sleep_on(&xprt->pending, task, NULL, xprt_timer); 1337 __xprt_release_write(xprt, task); 1338 spin_unlock_bh(&xprt->sock_lock); 1339} 1340 1341/* 1342 * Reserve an RPC call slot. 1343 */ 1344static inline void 1345do_xprt_reserve(struct rpc_task *task) 1346{ 1347 struct rpc_xprt *xprt = task->tk_xprt; 1348 1349 task->tk_status = 0; 1350 if (task->tk_rqstp) 1351 return; 1352 if (!list_empty(&xprt->free)) { 1353 struct rpc_rqst *req = list_entry(xprt->free.next, struct rpc_rqst, rq_list); 1354 list_del_init(&req->rq_list); 1355 task->tk_rqstp = req; 1356 xprt_request_init(task, xprt); 1357 return; 1358 } 1359 dprintk("RPC: waiting for request slot\n"); 1360 task->tk_status = -EAGAIN; 1361 task->tk_timeout = 0; 1362 rpc_sleep_on(&xprt->backlog, task, NULL, NULL); 1363} 1364 1365void 1366xprt_reserve(struct rpc_task *task) 1367{ 1368 struct rpc_xprt *xprt = task->tk_xprt; 1369 1370 task->tk_status = -EIO; 1371 if (!xprt->shutdown) { 1372 spin_lock(&xprt->xprt_lock); 1373 do_xprt_reserve(task); 1374 spin_unlock(&xprt->xprt_lock); 1375 } 1376} 1377 1378/* 1379 * Allocate a 'unique' XID 1380 */ 1381static inline u32 xprt_alloc_xid(struct rpc_xprt *xprt) 1382{ 1383 return xprt->xid++; 1384} 1385 1386static inline void xprt_init_xid(struct rpc_xprt *xprt) 1387{ 1388 get_random_bytes(&xprt->xid, sizeof(xprt->xid)); 1389} 1390 1391/* 1392 * Initialize RPC request 1393 */ 1394static void 1395xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt) 1396{ 1397 struct rpc_rqst *req = task->tk_rqstp; 1398 1399 req->rq_timeout = xprt->timeout.to_initval; 1400 req->rq_task = task; 1401 req->rq_xprt = xprt; 1402 req->rq_xid = xprt_alloc_xid(xprt); 1403 dprintk("RPC: %4d reserved req %p xid %08x\n", task->tk_pid, 1404 req, ntohl(req->rq_xid)); 1405} 1406 1407/* 1408 * Release an RPC call slot 1409 */ 1410void 1411xprt_release(struct rpc_task *task) 1412{ 1413 struct rpc_xprt *xprt = task->tk_xprt; 1414 struct rpc_rqst *req; 1415 1416 if (!(req = task->tk_rqstp)) 1417 return; 1418 spin_lock_bh(&xprt->sock_lock); 1419 __xprt_release_write(xprt, task); 1420 __xprt_put_cong(xprt, req); 1421 if (!list_empty(&req->rq_list)) 1422 list_del(&req->rq_list); 1423 xprt->last_used = jiffies; 1424 if (list_empty(&xprt->recv) && !xprt->shutdown) 1425 mod_timer(&xprt->timer, xprt->last_used + XPRT_IDLE_TIMEOUT); 1426 spin_unlock_bh(&xprt->sock_lock); 1427 task->tk_rqstp = NULL; 1428 memset(req, 0, sizeof(*req)); /* mark unused */ 1429 1430 dprintk("RPC: %4d release request %p\n", task->tk_pid, req); 1431 1432 spin_lock(&xprt->xprt_lock); 1433 list_add(&req->rq_list, &xprt->free); 1434 xprt_clear_backlog(xprt); 1435 spin_unlock(&xprt->xprt_lock); 1436} 1437 1438/* 1439 * Set default timeout parameters 1440 */ 1441static void 1442xprt_default_timeout(struct rpc_timeout *to, int proto) 1443{ 1444 if (proto == IPPROTO_UDP) 1445 xprt_set_timeout(to, 5, 5 * HZ); 1446 else 1447 xprt_set_timeout(to, 5, 60 * HZ); 1448} 1449 1450/* 1451 * Set constant timeout 1452 */ 1453void 1454xprt_set_timeout(struct rpc_timeout *to, unsigned int retr, unsigned long incr) 1455{ 1456 to->to_initval = 1457 to->to_increment = incr; 1458 to->to_maxval = incr * retr; 1459 to->to_retries = retr; 1460 to->to_exponential = 0; 1461} 1462 1463unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE; 1464unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE; 1465 1466/* 1467 * Initialize an RPC client 1468 */ 1469static struct rpc_xprt * 1470xprt_setup(int proto, struct sockaddr_in *ap, struct rpc_timeout *to) 1471{ 1472 struct rpc_xprt *xprt; 1473 unsigned int entries; 1474 size_t slot_table_size; 1475 struct rpc_rqst *req; 1476 1477 dprintk("RPC: setting up %s transport...\n", 1478 proto == IPPROTO_UDP? "UDP" : "TCP"); 1479 1480 entries = (proto == IPPROTO_TCP)? 1481 xprt_tcp_slot_table_entries : xprt_udp_slot_table_entries; 1482 1483 if ((xprt = kmalloc(sizeof(struct rpc_xprt), GFP_KERNEL)) == NULL) 1484 return ERR_PTR(-ENOMEM); 1485 memset(xprt, 0, sizeof(*xprt)); /* Nnnngh! */ 1486 xprt->max_reqs = entries; 1487 slot_table_size = entries * sizeof(xprt->slot[0]); 1488 xprt->slot = kmalloc(slot_table_size, GFP_KERNEL); 1489 if (xprt->slot == NULL) { 1490 kfree(xprt); 1491 return ERR_PTR(-ENOMEM); 1492 } 1493 memset(xprt->slot, 0, slot_table_size); 1494 1495 xprt->addr = *ap; 1496 xprt->prot = proto; 1497 xprt->stream = (proto == IPPROTO_TCP)? 1 : 0; 1498 if (xprt->stream) { 1499 xprt->cwnd = RPC_MAXCWND(xprt); 1500 xprt->nocong = 1; 1501 xprt->max_payload = (1U << 31) - 1; 1502 } else { 1503 xprt->cwnd = RPC_INITCWND; 1504 xprt->max_payload = (1U << 16) - (MAX_HEADER << 3); 1505 } 1506 spin_lock_init(&xprt->sock_lock); 1507 spin_lock_init(&xprt->xprt_lock); 1508 init_waitqueue_head(&xprt->cong_wait); 1509 1510 INIT_LIST_HEAD(&xprt->free); 1511 INIT_LIST_HEAD(&xprt->recv); 1512 INIT_WORK(&xprt->sock_connect, xprt_socket_connect, xprt); 1513 INIT_WORK(&xprt->task_cleanup, xprt_socket_autoclose, xprt); 1514 init_timer(&xprt->timer); 1515 xprt->timer.function = xprt_init_autodisconnect; 1516 xprt->timer.data = (unsigned long) xprt; 1517 xprt->last_used = jiffies; 1518 xprt->port = XPRT_MAX_RESVPORT; 1519 1520 /* Set timeout parameters */ 1521 if (to) { 1522 xprt->timeout = *to; 1523 } else 1524 xprt_default_timeout(&xprt->timeout, xprt->prot); 1525 1526 rpc_init_wait_queue(&xprt->pending, "xprt_pending"); 1527 rpc_init_wait_queue(&xprt->sending, "xprt_sending"); 1528 rpc_init_wait_queue(&xprt->resend, "xprt_resend"); 1529 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog"); 1530 1531 /* initialize free list */ 1532 for (req = &xprt->slot[entries-1]; req >= &xprt->slot[0]; req--) 1533 list_add(&req->rq_list, &xprt->free); 1534 1535 xprt_init_xid(xprt); 1536 1537 /* Check whether we want to use a reserved port */ 1538 xprt->resvport = capable(CAP_NET_BIND_SERVICE) ? 1 : 0; 1539 1540 dprintk("RPC: created transport %p with %u slots\n", xprt, 1541 xprt->max_reqs); 1542 1543 return xprt; 1544} 1545 1546/* 1547 * Bind to a reserved port 1548 */ 1549static inline int xprt_bindresvport(struct rpc_xprt *xprt, struct socket *sock) 1550{ 1551 struct sockaddr_in myaddr = { 1552 .sin_family = AF_INET, 1553 }; 1554 int err, port; 1555 1556 /* Were we already bound to a given port? Try to reuse it */ 1557 port = xprt->port; 1558 do { 1559 myaddr.sin_port = htons(port); 1560 err = sock->ops->bind(sock, (struct sockaddr *) &myaddr, 1561 sizeof(myaddr)); 1562 if (err == 0) { 1563 xprt->port = port; 1564 return 0; 1565 } 1566 if (--port == 0) 1567 port = XPRT_MAX_RESVPORT; 1568 } while (err == -EADDRINUSE && port != xprt->port); 1569 1570 printk("RPC: Can't bind to reserved port (%d).\n", -err); 1571 return err; 1572} 1573 1574static void 1575xprt_bind_socket(struct rpc_xprt *xprt, struct socket *sock) 1576{ 1577 struct sock *sk = sock->sk; 1578 1579 if (xprt->inet) 1580 return; 1581 1582 write_lock_bh(&sk->sk_callback_lock); 1583 sk->sk_user_data = xprt; 1584 xprt->old_data_ready = sk->sk_data_ready; 1585 xprt->old_state_change = sk->sk_state_change; 1586 xprt->old_write_space = sk->sk_write_space; 1587 if (xprt->prot == IPPROTO_UDP) { 1588 sk->sk_data_ready = udp_data_ready; 1589 sk->sk_no_check = UDP_CSUM_NORCV; 1590 xprt_set_connected(xprt); 1591 } else { 1592 tcp_sk(sk)->nonagle = 1; /* disable Nagle's algorithm */ 1593 sk->sk_data_ready = tcp_data_ready; 1594 sk->sk_state_change = tcp_state_change; 1595 xprt_clear_connected(xprt); 1596 } 1597 sk->sk_write_space = xprt_write_space; 1598 1599 /* Reset to new socket */ 1600 xprt->sock = sock; 1601 xprt->inet = sk; 1602 write_unlock_bh(&sk->sk_callback_lock); 1603 1604 return; 1605} 1606 1607/* 1608 * Set socket buffer length 1609 */ 1610void 1611xprt_sock_setbufsize(struct rpc_xprt *xprt) 1612{ 1613 struct sock *sk = xprt->inet; 1614 1615 if (xprt->stream) 1616 return; 1617 if (xprt->rcvsize) { 1618 sk->sk_userlocks |= SOCK_RCVBUF_LOCK; 1619 sk->sk_rcvbuf = xprt->rcvsize * xprt->max_reqs * 2; 1620 } 1621 if (xprt->sndsize) { 1622 sk->sk_userlocks |= SOCK_SNDBUF_LOCK; 1623 sk->sk_sndbuf = xprt->sndsize * xprt->max_reqs * 2; 1624 sk->sk_write_space(sk); 1625 } 1626} 1627 1628/* 1629 * Datastream sockets are created here, but xprt_connect will create 1630 * and connect stream sockets. 1631 */ 1632static struct socket * xprt_create_socket(struct rpc_xprt *xprt, int proto, int resvport) 1633{ 1634 struct socket *sock; 1635 int type, err; 1636 1637 dprintk("RPC: xprt_create_socket(%s %d)\n", 1638 (proto == IPPROTO_UDP)? "udp" : "tcp", proto); 1639 1640 type = (proto == IPPROTO_UDP)? SOCK_DGRAM : SOCK_STREAM; 1641 1642 if ((err = sock_create_kern(PF_INET, type, proto, &sock)) < 0) { 1643 printk("RPC: can't create socket (%d).\n", -err); 1644 return NULL; 1645 } 1646 1647 /* If the caller has the capability, bind to a reserved port */ 1648 if (resvport && xprt_bindresvport(xprt, sock) < 0) { 1649 printk("RPC: can't bind to reserved port.\n"); 1650 goto failed; 1651 } 1652 1653 return sock; 1654 1655failed: 1656 sock_release(sock); 1657 return NULL; 1658} 1659 1660/* 1661 * Create an RPC client transport given the protocol and peer address. 1662 */ 1663struct rpc_xprt * 1664xprt_create_proto(int proto, struct sockaddr_in *sap, struct rpc_timeout *to) 1665{ 1666 struct rpc_xprt *xprt; 1667 1668 xprt = xprt_setup(proto, sap, to); 1669 if (IS_ERR(xprt)) 1670 dprintk("RPC: xprt_create_proto failed\n"); 1671 else 1672 dprintk("RPC: xprt_create_proto created xprt %p\n", xprt); 1673 return xprt; 1674} 1675 1676/* 1677 * Prepare for transport shutdown. 1678 */ 1679static void 1680xprt_shutdown(struct rpc_xprt *xprt) 1681{ 1682 xprt->shutdown = 1; 1683 rpc_wake_up(&xprt->sending); 1684 rpc_wake_up(&xprt->resend); 1685 rpc_wake_up(&xprt->pending); 1686 rpc_wake_up(&xprt->backlog); 1687 wake_up(&xprt->cong_wait); 1688 del_timer_sync(&xprt->timer); 1689} 1690 1691/* 1692 * Clear the xprt backlog queue 1693 */ 1694static int 1695xprt_clear_backlog(struct rpc_xprt *xprt) { 1696 rpc_wake_up_next(&xprt->backlog); 1697 wake_up(&xprt->cong_wait); 1698 return 1; 1699} 1700 1701/* 1702 * Destroy an RPC transport, killing off all requests. 1703 */ 1704int 1705xprt_destroy(struct rpc_xprt *xprt) 1706{ 1707 dprintk("RPC: destroying transport %p\n", xprt); 1708 xprt_shutdown(xprt); 1709 xprt_disconnect(xprt); 1710 xprt_close(xprt); 1711 kfree(xprt->slot); 1712 kfree(xprt); 1713 1714 return 0; 1715} 1716