1 /* 2 * linux/net/sunrpc/xprtsock.c 3 * 4 * Client-side transport implementation for sockets. 5 * 6 * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com> 7 * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com> 8 * TCP NFS related read + write fixes 9 * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie> 10 * 11 * Rewrite of larges part of the code in order to stabilize TCP stuff. 12 * Fix behaviour when socket buffer is full. 13 * (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no> 14 * 15 * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com> 16 */ 17 18 #include <linux/types.h> 19 #include <linux/slab.h> 20 #include <linux/capability.h> 21 #include <linux/pagemap.h> 22 #include <linux/errno.h> 23 #include <linux/socket.h> 24 #include <linux/in.h> 25 #include <linux/net.h> 26 #include <linux/mm.h> 27 #include <linux/udp.h> 28 #include <linux/tcp.h> 29 #include <linux/sunrpc/clnt.h> 30 #include <linux/sunrpc/sched.h> 31 #include <linux/file.h> 32 33 #include <net/sock.h> 34 #include <net/checksum.h> 35 #include <net/udp.h> 36 #include <net/tcp.h> 37 38 /* 39 * xprtsock tunables 40 */ 41 unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE; 42 unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE; 43 44 unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT; 45 unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT; 46 47 /* 48 * We can register our own files under /proc/sys/sunrpc by 49 * calling register_sysctl_table() again. The files in that 50 * directory become the union of all files registered there. 51 * 52 * We simply need to make sure that we don't collide with 53 * someone else's file names! 54 */ 55 56 #ifdef RPC_DEBUG 57 58 static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE; 59 static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE; 60 static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT; 61 static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT; 62 63 static struct ctl_table_header *sunrpc_table_header; 64 65 /* 66 * FIXME: changing the UDP slot table size should also resize the UDP 67 * socket buffers for existing UDP transports 68 */ 69 static ctl_table xs_tunables_table[] = { 70 { 71 .ctl_name = CTL_SLOTTABLE_UDP, 72 .procname = "udp_slot_table_entries", 73 .data = &xprt_udp_slot_table_entries, 74 .maxlen = sizeof(unsigned int), 75 .mode = 0644, 76 .proc_handler = &proc_dointvec_minmax, 77 .strategy = &sysctl_intvec, 78 .extra1 = &min_slot_table_size, 79 .extra2 = &max_slot_table_size 80 }, 81 { 82 .ctl_name = CTL_SLOTTABLE_TCP, 83 .procname = "tcp_slot_table_entries", 84 .data = &xprt_tcp_slot_table_entries, 85 .maxlen = sizeof(unsigned int), 86 .mode = 0644, 87 .proc_handler = &proc_dointvec_minmax, 88 .strategy = &sysctl_intvec, 89 .extra1 = &min_slot_table_size, 90 .extra2 = &max_slot_table_size 91 }, 92 { 93 .ctl_name = CTL_MIN_RESVPORT, 94 .procname = "min_resvport", 95 .data = &xprt_min_resvport, 96 .maxlen = sizeof(unsigned int), 97 .mode = 0644, 98 .proc_handler = &proc_dointvec_minmax, 99 .strategy = &sysctl_intvec, 100 .extra1 = &xprt_min_resvport_limit, 101 .extra2 = &xprt_max_resvport_limit 102 }, 103 { 104 .ctl_name = CTL_MAX_RESVPORT, 105 .procname = "max_resvport", 106 .data = &xprt_max_resvport, 107 .maxlen = sizeof(unsigned int), 108 .mode = 0644, 109 .proc_handler = &proc_dointvec_minmax, 110 .strategy = &sysctl_intvec, 111 .extra1 = &xprt_min_resvport_limit, 112 .extra2 = &xprt_max_resvport_limit 113 }, 114 { 115 .ctl_name = 0, 116 }, 117 }; 118 119 static ctl_table sunrpc_table[] = { 120 { 121 .ctl_name = CTL_SUNRPC, 122 .procname = "sunrpc", 123 .mode = 0555, 124 .child = xs_tunables_table 125 }, 126 { 127 .ctl_name = 0, 128 }, 129 }; 130 131 #endif 132 133 /* 134 * How many times to try sending a request on a socket before waiting 135 * for the socket buffer to clear. 136 */ 137 #define XS_SENDMSG_RETRY (10U) 138 139 /* 140 * Time out for an RPC UDP socket connect. UDP socket connects are 141 * synchronous, but we set a timeout anyway in case of resource 142 * exhaustion on the local host. 143 */ 144 #define XS_UDP_CONN_TO (5U * HZ) 145 146 /* 147 * Wait duration for an RPC TCP connection to be established. Solaris 148 * NFS over TCP uses 60 seconds, for example, which is in line with how 149 * long a server takes to reboot. 150 */ 151 #define XS_TCP_CONN_TO (60U * HZ) 152 153 /* 154 * Wait duration for a reply from the RPC portmapper. 155 */ 156 #define XS_BIND_TO (60U * HZ) 157 158 /* 159 * Delay if a UDP socket connect error occurs. This is most likely some 160 * kind of resource problem on the local host. 161 */ 162 #define XS_UDP_REEST_TO (2U * HZ) 163 164 /* 165 * The reestablish timeout allows clients to delay for a bit before attempting 166 * to reconnect to a server that just dropped our connection. 167 * 168 * We implement an exponential backoff when trying to reestablish a TCP 169 * transport connection with the server. Some servers like to drop a TCP 170 * connection when they are overworked, so we start with a short timeout and 171 * increase over time if the server is down or not responding. 172 */ 173 #define XS_TCP_INIT_REEST_TO (3U * HZ) 174 #define XS_TCP_MAX_REEST_TO (5U * 60 * HZ) 175 176 /* 177 * TCP idle timeout; client drops the transport socket if it is idle 178 * for this long. Note that we also timeout UDP sockets to prevent 179 * holding port numbers when there is no RPC traffic. 180 */ 181 #define XS_IDLE_DISC_TO (5U * 60 * HZ) 182 183 #ifdef RPC_DEBUG 184 # undef RPC_DEBUG_DATA 185 # define RPCDBG_FACILITY RPCDBG_TRANS 186 #endif 187 188 #ifdef RPC_DEBUG_DATA 189 static void xs_pktdump(char *msg, u32 *packet, unsigned int count) 190 { 191 u8 *buf = (u8 *) packet; 192 int j; 193 194 dprintk("RPC: %s\n", msg); 195 for (j = 0; j < count && j < 128; j += 4) { 196 if (!(j & 31)) { 197 if (j) 198 dprintk("\n"); 199 dprintk("0x%04x ", j); 200 } 201 dprintk("%02x%02x%02x%02x ", 202 buf[j], buf[j+1], buf[j+2], buf[j+3]); 203 } 204 dprintk("\n"); 205 } 206 #else 207 static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count) 208 { 209 /* NOP */ 210 } 211 #endif 212 213 struct sock_xprt { 214 struct rpc_xprt xprt; 215 216 /* 217 * Network layer 218 */ 219 struct socket * sock; 220 struct sock * inet; 221 222 /* 223 * State of TCP reply receive 224 */ 225 __be32 tcp_fraghdr, 226 tcp_xid; 227 228 u32 tcp_offset, 229 tcp_reclen; 230 231 unsigned long tcp_copied, 232 tcp_flags; 233 234 /* 235 * Connection of transports 236 */ 237 struct delayed_work connect_worker; 238 unsigned short port; 239 240 /* 241 * UDP socket buffer size parameters 242 */ 243 size_t rcvsize, 244 sndsize; 245 246 /* 247 * Saved socket callback addresses 248 */ 249 void (*old_data_ready)(struct sock *, int); 250 void (*old_state_change)(struct sock *); 251 void (*old_write_space)(struct sock *); 252 }; 253 254 /* 255 * TCP receive state flags 256 */ 257 #define TCP_RCV_LAST_FRAG (1UL << 0) 258 #define TCP_RCV_COPY_FRAGHDR (1UL << 1) 259 #define TCP_RCV_COPY_XID (1UL << 2) 260 #define TCP_RCV_COPY_DATA (1UL << 3) 261 262 static void xs_format_peer_addresses(struct rpc_xprt *xprt) 263 { 264 struct sockaddr_in *addr = (struct sockaddr_in *) &xprt->addr; 265 char *buf; 266 267 buf = kzalloc(20, GFP_KERNEL); 268 if (buf) { 269 snprintf(buf, 20, "%u.%u.%u.%u", 270 NIPQUAD(addr->sin_addr.s_addr)); 271 } 272 xprt->address_strings[RPC_DISPLAY_ADDR] = buf; 273 274 buf = kzalloc(8, GFP_KERNEL); 275 if (buf) { 276 snprintf(buf, 8, "%u", 277 ntohs(addr->sin_port)); 278 } 279 xprt->address_strings[RPC_DISPLAY_PORT] = buf; 280 281 if (xprt->prot == IPPROTO_UDP) 282 xprt->address_strings[RPC_DISPLAY_PROTO] = "udp"; 283 else 284 xprt->address_strings[RPC_DISPLAY_PROTO] = "tcp"; 285 286 buf = kzalloc(48, GFP_KERNEL); 287 if (buf) { 288 snprintf(buf, 48, "addr=%u.%u.%u.%u port=%u proto=%s", 289 NIPQUAD(addr->sin_addr.s_addr), 290 ntohs(addr->sin_port), 291 xprt->prot == IPPROTO_UDP ? "udp" : "tcp"); 292 } 293 xprt->address_strings[RPC_DISPLAY_ALL] = buf; 294 } 295 296 static void xs_free_peer_addresses(struct rpc_xprt *xprt) 297 { 298 kfree(xprt->address_strings[RPC_DISPLAY_ADDR]); 299 kfree(xprt->address_strings[RPC_DISPLAY_PORT]); 300 kfree(xprt->address_strings[RPC_DISPLAY_ALL]); 301 } 302 303 #define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL) 304 305 static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more) 306 { 307 struct msghdr msg = { 308 .msg_name = addr, 309 .msg_namelen = addrlen, 310 .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0), 311 }; 312 struct kvec iov = { 313 .iov_base = vec->iov_base + base, 314 .iov_len = vec->iov_len - base, 315 }; 316 317 if (iov.iov_len != 0) 318 return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len); 319 return kernel_sendmsg(sock, &msg, NULL, 0, 0); 320 } 321 322 static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more) 323 { 324 struct page **ppage; 325 unsigned int remainder; 326 int err, sent = 0; 327 328 remainder = xdr->page_len - base; 329 base += xdr->page_base; 330 ppage = xdr->pages + (base >> PAGE_SHIFT); 331 base &= ~PAGE_MASK; 332 for(;;) { 333 unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder); 334 int flags = XS_SENDMSG_FLAGS; 335 336 remainder -= len; 337 if (remainder != 0 || more) 338 flags |= MSG_MORE; 339 err = sock->ops->sendpage(sock, *ppage, base, len, flags); 340 if (remainder == 0 || err != len) 341 break; 342 sent += err; 343 ppage++; 344 base = 0; 345 } 346 if (sent == 0) 347 return err; 348 if (err > 0) 349 sent += err; 350 return sent; 351 } 352 353 /** 354 * xs_sendpages - write pages directly to a socket 355 * @sock: socket to send on 356 * @addr: UDP only -- address of destination 357 * @addrlen: UDP only -- length of destination address 358 * @xdr: buffer containing this request 359 * @base: starting position in the buffer 360 * 361 */ 362 static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base) 363 { 364 unsigned int remainder = xdr->len - base; 365 int err, sent = 0; 366 367 if (unlikely(!sock)) 368 return -ENOTCONN; 369 370 clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags); 371 if (base != 0) { 372 addr = NULL; 373 addrlen = 0; 374 } 375 376 if (base < xdr->head[0].iov_len || addr != NULL) { 377 unsigned int len = xdr->head[0].iov_len - base; 378 remainder -= len; 379 err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0); 380 if (remainder == 0 || err != len) 381 goto out; 382 sent += err; 383 base = 0; 384 } else 385 base -= xdr->head[0].iov_len; 386 387 if (base < xdr->page_len) { 388 unsigned int len = xdr->page_len - base; 389 remainder -= len; 390 err = xs_send_pagedata(sock, xdr, base, remainder != 0); 391 if (remainder == 0 || err != len) 392 goto out; 393 sent += err; 394 base = 0; 395 } else 396 base -= xdr->page_len; 397 398 if (base >= xdr->tail[0].iov_len) 399 return sent; 400 err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0); 401 out: 402 if (sent == 0) 403 return err; 404 if (err > 0) 405 sent += err; 406 return sent; 407 } 408 409 /** 410 * xs_nospace - place task on wait queue if transmit was incomplete 411 * @task: task to put to sleep 412 * 413 */ 414 static void xs_nospace(struct rpc_task *task) 415 { 416 struct rpc_rqst *req = task->tk_rqstp; 417 struct rpc_xprt *xprt = req->rq_xprt; 418 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 419 420 dprintk("RPC: %5u xmit incomplete (%u left of %u)\n", 421 task->tk_pid, req->rq_slen - req->rq_bytes_sent, 422 req->rq_slen); 423 424 if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) { 425 /* Protect against races with write_space */ 426 spin_lock_bh(&xprt->transport_lock); 427 428 /* Don't race with disconnect */ 429 if (!xprt_connected(xprt)) 430 task->tk_status = -ENOTCONN; 431 else if (test_bit(SOCK_NOSPACE, &transport->sock->flags)) 432 xprt_wait_for_buffer_space(task); 433 434 spin_unlock_bh(&xprt->transport_lock); 435 } else 436 /* Keep holding the socket if it is blocked */ 437 rpc_delay(task, HZ>>4); 438 } 439 440 /** 441 * xs_udp_send_request - write an RPC request to a UDP socket 442 * @task: address of RPC task that manages the state of an RPC request 443 * 444 * Return values: 445 * 0: The request has been sent 446 * EAGAIN: The socket was blocked, please call again later to 447 * complete the request 448 * ENOTCONN: Caller needs to invoke connect logic then call again 449 * other: Some other error occured, the request was not sent 450 */ 451 static int xs_udp_send_request(struct rpc_task *task) 452 { 453 struct rpc_rqst *req = task->tk_rqstp; 454 struct rpc_xprt *xprt = req->rq_xprt; 455 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 456 struct xdr_buf *xdr = &req->rq_snd_buf; 457 int status; 458 459 xs_pktdump("packet data:", 460 req->rq_svec->iov_base, 461 req->rq_svec->iov_len); 462 463 req->rq_xtime = jiffies; 464 status = xs_sendpages(transport->sock, 465 (struct sockaddr *) &xprt->addr, 466 xprt->addrlen, xdr, 467 req->rq_bytes_sent); 468 469 dprintk("RPC: xs_udp_send_request(%u) = %d\n", 470 xdr->len - req->rq_bytes_sent, status); 471 472 if (likely(status >= (int) req->rq_slen)) 473 return 0; 474 475 /* Still some bytes left; set up for a retry later. */ 476 if (status > 0) 477 status = -EAGAIN; 478 479 switch (status) { 480 case -ENETUNREACH: 481 case -EPIPE: 482 case -ECONNREFUSED: 483 /* When the server has died, an ICMP port unreachable message 484 * prompts ECONNREFUSED. */ 485 break; 486 case -EAGAIN: 487 xs_nospace(task); 488 break; 489 default: 490 dprintk("RPC: sendmsg returned unrecognized error %d\n", 491 -status); 492 break; 493 } 494 495 return status; 496 } 497 498 static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf) 499 { 500 u32 reclen = buf->len - sizeof(rpc_fraghdr); 501 rpc_fraghdr *base = buf->head[0].iov_base; 502 *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen); 503 } 504 505 /** 506 * xs_tcp_send_request - write an RPC request to a TCP socket 507 * @task: address of RPC task that manages the state of an RPC request 508 * 509 * Return values: 510 * 0: The request has been sent 511 * EAGAIN: The socket was blocked, please call again later to 512 * complete the request 513 * ENOTCONN: Caller needs to invoke connect logic then call again 514 * other: Some other error occured, the request was not sent 515 * 516 * XXX: In the case of soft timeouts, should we eventually give up 517 * if sendmsg is not able to make progress? 518 */ 519 static int xs_tcp_send_request(struct rpc_task *task) 520 { 521 struct rpc_rqst *req = task->tk_rqstp; 522 struct rpc_xprt *xprt = req->rq_xprt; 523 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 524 struct xdr_buf *xdr = &req->rq_snd_buf; 525 int status, retry = 0; 526 527 xs_encode_tcp_record_marker(&req->rq_snd_buf); 528 529 xs_pktdump("packet data:", 530 req->rq_svec->iov_base, 531 req->rq_svec->iov_len); 532 533 /* Continue transmitting the packet/record. We must be careful 534 * to cope with writespace callbacks arriving _after_ we have 535 * called sendmsg(). */ 536 while (1) { 537 req->rq_xtime = jiffies; 538 status = xs_sendpages(transport->sock, 539 NULL, 0, xdr, req->rq_bytes_sent); 540 541 dprintk("RPC: xs_tcp_send_request(%u) = %d\n", 542 xdr->len - req->rq_bytes_sent, status); 543 544 if (unlikely(status < 0)) 545 break; 546 547 /* If we've sent the entire packet, immediately 548 * reset the count of bytes sent. */ 549 req->rq_bytes_sent += status; 550 task->tk_bytes_sent += status; 551 if (likely(req->rq_bytes_sent >= req->rq_slen)) { 552 req->rq_bytes_sent = 0; 553 return 0; 554 } 555 556 status = -EAGAIN; 557 if (retry++ > XS_SENDMSG_RETRY) 558 break; 559 } 560 561 switch (status) { 562 case -EAGAIN: 563 xs_nospace(task); 564 break; 565 case -ECONNREFUSED: 566 case -ECONNRESET: 567 case -ENOTCONN: 568 case -EPIPE: 569 status = -ENOTCONN; 570 break; 571 default: 572 dprintk("RPC: sendmsg returned unrecognized error %d\n", 573 -status); 574 xprt_disconnect(xprt); 575 break; 576 } 577 578 return status; 579 } 580 581 /** 582 * xs_tcp_release_xprt - clean up after a tcp transmission 583 * @xprt: transport 584 * @task: rpc task 585 * 586 * This cleans up if an error causes us to abort the transmission of a request. 587 * In this case, the socket may need to be reset in order to avoid confusing 588 * the server. 589 */ 590 static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 591 { 592 struct rpc_rqst *req; 593 594 if (task != xprt->snd_task) 595 return; 596 if (task == NULL) 597 goto out_release; 598 req = task->tk_rqstp; 599 if (req->rq_bytes_sent == 0) 600 goto out_release; 601 if (req->rq_bytes_sent == req->rq_snd_buf.len) 602 goto out_release; 603 set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state); 604 out_release: 605 xprt_release_xprt(xprt, task); 606 } 607 608 /** 609 * xs_close - close a socket 610 * @xprt: transport 611 * 612 * This is used when all requests are complete; ie, no DRC state remains 613 * on the server we want to save. 614 */ 615 static void xs_close(struct rpc_xprt *xprt) 616 { 617 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 618 struct socket *sock = transport->sock; 619 struct sock *sk = transport->inet; 620 621 if (!sk) 622 goto clear_close_wait; 623 624 dprintk("RPC: xs_close xprt %p\n", xprt); 625 626 write_lock_bh(&sk->sk_callback_lock); 627 transport->inet = NULL; 628 transport->sock = NULL; 629 630 sk->sk_user_data = NULL; 631 sk->sk_data_ready = transport->old_data_ready; 632 sk->sk_state_change = transport->old_state_change; 633 sk->sk_write_space = transport->old_write_space; 634 write_unlock_bh(&sk->sk_callback_lock); 635 636 sk->sk_no_check = 0; 637 638 sock_release(sock); 639 clear_close_wait: 640 smp_mb__before_clear_bit(); 641 clear_bit(XPRT_CLOSE_WAIT, &xprt->state); 642 smp_mb__after_clear_bit(); 643 } 644 645 /** 646 * xs_destroy - prepare to shutdown a transport 647 * @xprt: doomed transport 648 * 649 */ 650 static void xs_destroy(struct rpc_xprt *xprt) 651 { 652 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 653 654 dprintk("RPC: xs_destroy xprt %p\n", xprt); 655 656 cancel_delayed_work(&transport->connect_worker); 657 flush_scheduled_work(); 658 659 xprt_disconnect(xprt); 660 xs_close(xprt); 661 xs_free_peer_addresses(xprt); 662 kfree(xprt->slot); 663 kfree(xprt); 664 } 665 666 static inline struct rpc_xprt *xprt_from_sock(struct sock *sk) 667 { 668 return (struct rpc_xprt *) sk->sk_user_data; 669 } 670 671 /** 672 * xs_udp_data_ready - "data ready" callback for UDP sockets 673 * @sk: socket with data to read 674 * @len: how much data to read 675 * 676 */ 677 static void xs_udp_data_ready(struct sock *sk, int len) 678 { 679 struct rpc_task *task; 680 struct rpc_xprt *xprt; 681 struct rpc_rqst *rovr; 682 struct sk_buff *skb; 683 int err, repsize, copied; 684 u32 _xid; 685 __be32 *xp; 686 687 read_lock(&sk->sk_callback_lock); 688 dprintk("RPC: xs_udp_data_ready...\n"); 689 if (!(xprt = xprt_from_sock(sk))) 690 goto out; 691 692 if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL) 693 goto out; 694 695 if (xprt->shutdown) 696 goto dropit; 697 698 repsize = skb->len - sizeof(struct udphdr); 699 if (repsize < 4) { 700 dprintk("RPC: impossible RPC reply size %d!\n", repsize); 701 goto dropit; 702 } 703 704 /* Copy the XID from the skb... */ 705 xp = skb_header_pointer(skb, sizeof(struct udphdr), 706 sizeof(_xid), &_xid); 707 if (xp == NULL) 708 goto dropit; 709 710 /* Look up and lock the request corresponding to the given XID */ 711 spin_lock(&xprt->transport_lock); 712 rovr = xprt_lookup_rqst(xprt, *xp); 713 if (!rovr) 714 goto out_unlock; 715 task = rovr->rq_task; 716 717 if ((copied = rovr->rq_private_buf.buflen) > repsize) 718 copied = repsize; 719 720 /* Suck it into the iovec, verify checksum if not done by hw. */ 721 if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) 722 goto out_unlock; 723 724 /* Something worked... */ 725 dst_confirm(skb->dst); 726 727 xprt_adjust_cwnd(task, copied); 728 xprt_update_rtt(task); 729 xprt_complete_rqst(task, copied); 730 731 out_unlock: 732 spin_unlock(&xprt->transport_lock); 733 dropit: 734 skb_free_datagram(sk, skb); 735 out: 736 read_unlock(&sk->sk_callback_lock); 737 } 738 739 static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc) 740 { 741 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 742 size_t len, used; 743 char *p; 744 745 p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset; 746 len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset; 747 used = xdr_skb_read_bits(desc, p, len); 748 transport->tcp_offset += used; 749 if (used != len) 750 return; 751 752 transport->tcp_reclen = ntohl(transport->tcp_fraghdr); 753 if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT) 754 transport->tcp_flags |= TCP_RCV_LAST_FRAG; 755 else 756 transport->tcp_flags &= ~TCP_RCV_LAST_FRAG; 757 transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK; 758 759 transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR; 760 transport->tcp_offset = 0; 761 762 /* Sanity check of the record length */ 763 if (unlikely(transport->tcp_reclen < 4)) { 764 dprintk("RPC: invalid TCP record fragment length\n"); 765 xprt_disconnect(xprt); 766 return; 767 } 768 dprintk("RPC: reading TCP record fragment of length %d\n", 769 transport->tcp_reclen); 770 } 771 772 static void xs_tcp_check_fraghdr(struct sock_xprt *transport) 773 { 774 if (transport->tcp_offset == transport->tcp_reclen) { 775 transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR; 776 transport->tcp_offset = 0; 777 if (transport->tcp_flags & TCP_RCV_LAST_FRAG) { 778 transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 779 transport->tcp_flags |= TCP_RCV_COPY_XID; 780 transport->tcp_copied = 0; 781 } 782 } 783 } 784 785 static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc) 786 { 787 size_t len, used; 788 char *p; 789 790 len = sizeof(transport->tcp_xid) - transport->tcp_offset; 791 dprintk("RPC: reading XID (%Zu bytes)\n", len); 792 p = ((char *) &transport->tcp_xid) + transport->tcp_offset; 793 used = xdr_skb_read_bits(desc, p, len); 794 transport->tcp_offset += used; 795 if (used != len) 796 return; 797 transport->tcp_flags &= ~TCP_RCV_COPY_XID; 798 transport->tcp_flags |= TCP_RCV_COPY_DATA; 799 transport->tcp_copied = 4; 800 dprintk("RPC: reading reply for XID %08x\n", 801 ntohl(transport->tcp_xid)); 802 xs_tcp_check_fraghdr(transport); 803 } 804 805 static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_reader *desc) 806 { 807 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 808 struct rpc_rqst *req; 809 struct xdr_buf *rcvbuf; 810 size_t len; 811 ssize_t r; 812 813 /* Find and lock the request corresponding to this xid */ 814 spin_lock(&xprt->transport_lock); 815 req = xprt_lookup_rqst(xprt, transport->tcp_xid); 816 if (!req) { 817 transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 818 dprintk("RPC: XID %08x request not found!\n", 819 ntohl(transport->tcp_xid)); 820 spin_unlock(&xprt->transport_lock); 821 return; 822 } 823 824 rcvbuf = &req->rq_private_buf; 825 len = desc->count; 826 if (len > transport->tcp_reclen - transport->tcp_offset) { 827 struct xdr_skb_reader my_desc; 828 829 len = transport->tcp_reclen - transport->tcp_offset; 830 memcpy(&my_desc, desc, sizeof(my_desc)); 831 my_desc.count = len; 832 r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied, 833 &my_desc, xdr_skb_read_bits); 834 desc->count -= r; 835 desc->offset += r; 836 } else 837 r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied, 838 desc, xdr_skb_read_bits); 839 840 if (r > 0) { 841 transport->tcp_copied += r; 842 transport->tcp_offset += r; 843 } 844 if (r != len) { 845 /* Error when copying to the receive buffer, 846 * usually because we weren't able to allocate 847 * additional buffer pages. All we can do now 848 * is turn off TCP_RCV_COPY_DATA, so the request 849 * will not receive any additional updates, 850 * and time out. 851 * Any remaining data from this record will 852 * be discarded. 853 */ 854 transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 855 dprintk("RPC: XID %08x truncated request\n", 856 ntohl(transport->tcp_xid)); 857 dprintk("RPC: xprt = %p, tcp_copied = %lu, " 858 "tcp_offset = %u, tcp_reclen = %u\n", 859 xprt, transport->tcp_copied, 860 transport->tcp_offset, transport->tcp_reclen); 861 goto out; 862 } 863 864 dprintk("RPC: XID %08x read %Zd bytes\n", 865 ntohl(transport->tcp_xid), r); 866 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, " 867 "tcp_reclen = %u\n", xprt, transport->tcp_copied, 868 transport->tcp_offset, transport->tcp_reclen); 869 870 if (transport->tcp_copied == req->rq_private_buf.buflen) 871 transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 872 else if (transport->tcp_offset == transport->tcp_reclen) { 873 if (transport->tcp_flags & TCP_RCV_LAST_FRAG) 874 transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 875 } 876 877 out: 878 if (!(transport->tcp_flags & TCP_RCV_COPY_DATA)) 879 xprt_complete_rqst(req->rq_task, transport->tcp_copied); 880 spin_unlock(&xprt->transport_lock); 881 xs_tcp_check_fraghdr(transport); 882 } 883 884 static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc) 885 { 886 size_t len; 887 888 len = transport->tcp_reclen - transport->tcp_offset; 889 if (len > desc->count) 890 len = desc->count; 891 desc->count -= len; 892 desc->offset += len; 893 transport->tcp_offset += len; 894 dprintk("RPC: discarded %Zu bytes\n", len); 895 xs_tcp_check_fraghdr(transport); 896 } 897 898 static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len) 899 { 900 struct rpc_xprt *xprt = rd_desc->arg.data; 901 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 902 struct xdr_skb_reader desc = { 903 .skb = skb, 904 .offset = offset, 905 .count = len, 906 }; 907 908 dprintk("RPC: xs_tcp_data_recv started\n"); 909 do { 910 /* Read in a new fragment marker if necessary */ 911 /* Can we ever really expect to get completely empty fragments? */ 912 if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) { 913 xs_tcp_read_fraghdr(xprt, &desc); 914 continue; 915 } 916 /* Read in the xid if necessary */ 917 if (transport->tcp_flags & TCP_RCV_COPY_XID) { 918 xs_tcp_read_xid(transport, &desc); 919 continue; 920 } 921 /* Read in the request data */ 922 if (transport->tcp_flags & TCP_RCV_COPY_DATA) { 923 xs_tcp_read_request(xprt, &desc); 924 continue; 925 } 926 /* Skip over any trailing bytes on short reads */ 927 xs_tcp_read_discard(transport, &desc); 928 } while (desc.count); 929 dprintk("RPC: xs_tcp_data_recv done\n"); 930 return len - desc.count; 931 } 932 933 /** 934 * xs_tcp_data_ready - "data ready" callback for TCP sockets 935 * @sk: socket with data to read 936 * @bytes: how much data to read 937 * 938 */ 939 static void xs_tcp_data_ready(struct sock *sk, int bytes) 940 { 941 struct rpc_xprt *xprt; 942 read_descriptor_t rd_desc; 943 944 dprintk("RPC: xs_tcp_data_ready...\n"); 945 946 read_lock(&sk->sk_callback_lock); 947 if (!(xprt = xprt_from_sock(sk))) 948 goto out; 949 if (xprt->shutdown) 950 goto out; 951 952 /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */ 953 rd_desc.arg.data = xprt; 954 rd_desc.count = 65536; 955 tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv); 956 out: 957 read_unlock(&sk->sk_callback_lock); 958 } 959 960 /** 961 * xs_tcp_state_change - callback to handle TCP socket state changes 962 * @sk: socket whose state has changed 963 * 964 */ 965 static void xs_tcp_state_change(struct sock *sk) 966 { 967 struct rpc_xprt *xprt; 968 969 read_lock(&sk->sk_callback_lock); 970 if (!(xprt = xprt_from_sock(sk))) 971 goto out; 972 dprintk("RPC: xs_tcp_state_change client %p...\n", xprt); 973 dprintk("RPC: state %x conn %d dead %d zapped %d\n", 974 sk->sk_state, xprt_connected(xprt), 975 sock_flag(sk, SOCK_DEAD), 976 sock_flag(sk, SOCK_ZAPPED)); 977 978 switch (sk->sk_state) { 979 case TCP_ESTABLISHED: 980 spin_lock_bh(&xprt->transport_lock); 981 if (!xprt_test_and_set_connected(xprt)) { 982 struct sock_xprt *transport = container_of(xprt, 983 struct sock_xprt, xprt); 984 985 /* Reset TCP record info */ 986 transport->tcp_offset = 0; 987 transport->tcp_reclen = 0; 988 transport->tcp_copied = 0; 989 transport->tcp_flags = 990 TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID; 991 992 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO; 993 xprt_wake_pending_tasks(xprt, 0); 994 } 995 spin_unlock_bh(&xprt->transport_lock); 996 break; 997 case TCP_SYN_SENT: 998 case TCP_SYN_RECV: 999 break; 1000 case TCP_CLOSE_WAIT: 1001 /* Try to schedule an autoclose RPC calls */ 1002 set_bit(XPRT_CLOSE_WAIT, &xprt->state); 1003 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0) 1004 schedule_work(&xprt->task_cleanup); 1005 default: 1006 xprt_disconnect(xprt); 1007 } 1008 out: 1009 read_unlock(&sk->sk_callback_lock); 1010 } 1011 1012 /** 1013 * xs_udp_write_space - callback invoked when socket buffer space 1014 * becomes available 1015 * @sk: socket whose state has changed 1016 * 1017 * Called when more output buffer space is available for this socket. 1018 * We try not to wake our writers until they can make "significant" 1019 * progress, otherwise we'll waste resources thrashing kernel_sendmsg 1020 * with a bunch of small requests. 1021 */ 1022 static void xs_udp_write_space(struct sock *sk) 1023 { 1024 read_lock(&sk->sk_callback_lock); 1025 1026 /* from net/core/sock.c:sock_def_write_space */ 1027 if (sock_writeable(sk)) { 1028 struct socket *sock; 1029 struct rpc_xprt *xprt; 1030 1031 if (unlikely(!(sock = sk->sk_socket))) 1032 goto out; 1033 if (unlikely(!(xprt = xprt_from_sock(sk)))) 1034 goto out; 1035 if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags))) 1036 goto out; 1037 1038 xprt_write_space(xprt); 1039 } 1040 1041 out: 1042 read_unlock(&sk->sk_callback_lock); 1043 } 1044 1045 /** 1046 * xs_tcp_write_space - callback invoked when socket buffer space 1047 * becomes available 1048 * @sk: socket whose state has changed 1049 * 1050 * Called when more output buffer space is available for this socket. 1051 * We try not to wake our writers until they can make "significant" 1052 * progress, otherwise we'll waste resources thrashing kernel_sendmsg 1053 * with a bunch of small requests. 1054 */ 1055 static void xs_tcp_write_space(struct sock *sk) 1056 { 1057 read_lock(&sk->sk_callback_lock); 1058 1059 /* from net/core/stream.c:sk_stream_write_space */ 1060 if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) { 1061 struct socket *sock; 1062 struct rpc_xprt *xprt; 1063 1064 if (unlikely(!(sock = sk->sk_socket))) 1065 goto out; 1066 if (unlikely(!(xprt = xprt_from_sock(sk)))) 1067 goto out; 1068 if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags))) 1069 goto out; 1070 1071 xprt_write_space(xprt); 1072 } 1073 1074 out: 1075 read_unlock(&sk->sk_callback_lock); 1076 } 1077 1078 static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt) 1079 { 1080 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1081 struct sock *sk = transport->inet; 1082 1083 if (transport->rcvsize) { 1084 sk->sk_userlocks |= SOCK_RCVBUF_LOCK; 1085 sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2; 1086 } 1087 if (transport->sndsize) { 1088 sk->sk_userlocks |= SOCK_SNDBUF_LOCK; 1089 sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2; 1090 sk->sk_write_space(sk); 1091 } 1092 } 1093 1094 /** 1095 * xs_udp_set_buffer_size - set send and receive limits 1096 * @xprt: generic transport 1097 * @sndsize: requested size of send buffer, in bytes 1098 * @rcvsize: requested size of receive buffer, in bytes 1099 * 1100 * Set socket send and receive buffer size limits. 1101 */ 1102 static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize) 1103 { 1104 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1105 1106 transport->sndsize = 0; 1107 if (sndsize) 1108 transport->sndsize = sndsize + 1024; 1109 transport->rcvsize = 0; 1110 if (rcvsize) 1111 transport->rcvsize = rcvsize + 1024; 1112 1113 xs_udp_do_set_buffer_size(xprt); 1114 } 1115 1116 /** 1117 * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport 1118 * @task: task that timed out 1119 * 1120 * Adjust the congestion window after a retransmit timeout has occurred. 1121 */ 1122 static void xs_udp_timer(struct rpc_task *task) 1123 { 1124 xprt_adjust_cwnd(task, -ETIMEDOUT); 1125 } 1126 1127 static unsigned short xs_get_random_port(void) 1128 { 1129 unsigned short range = xprt_max_resvport - xprt_min_resvport; 1130 unsigned short rand = (unsigned short) net_random() % range; 1131 return rand + xprt_min_resvport; 1132 } 1133 1134 /** 1135 * xs_set_port - reset the port number in the remote endpoint address 1136 * @xprt: generic transport 1137 * @port: new port number 1138 * 1139 */ 1140 static void xs_set_port(struct rpc_xprt *xprt, unsigned short port) 1141 { 1142 struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr; 1143 1144 dprintk("RPC: setting port for xprt %p to %u\n", xprt, port); 1145 1146 sap->sin_port = htons(port); 1147 } 1148 1149 static int xs_bindresvport(struct sock_xprt *transport, struct socket *sock) 1150 { 1151 struct sockaddr_in myaddr = { 1152 .sin_family = AF_INET, 1153 }; 1154 int err; 1155 unsigned short port = transport->port; 1156 1157 do { 1158 myaddr.sin_port = htons(port); 1159 err = kernel_bind(sock, (struct sockaddr *) &myaddr, 1160 sizeof(myaddr)); 1161 if (err == 0) { 1162 transport->port = port; 1163 dprintk("RPC: xs_bindresvport bound to port %u\n", 1164 port); 1165 return 0; 1166 } 1167 if (port <= xprt_min_resvport) 1168 port = xprt_max_resvport; 1169 else 1170 port--; 1171 } while (err == -EADDRINUSE && port != transport->port); 1172 1173 dprintk("RPC: can't bind to reserved port (%d).\n", -err); 1174 return err; 1175 } 1176 1177 #ifdef CONFIG_DEBUG_LOCK_ALLOC 1178 static struct lock_class_key xs_key[2]; 1179 static struct lock_class_key xs_slock_key[2]; 1180 1181 static inline void xs_reclassify_socket(struct socket *sock) 1182 { 1183 struct sock *sk = sock->sk; 1184 BUG_ON(sk->sk_lock.owner != NULL); 1185 switch (sk->sk_family) { 1186 case AF_INET: 1187 sock_lock_init_class_and_name(sk, "slock-AF_INET-NFS", 1188 &xs_slock_key[0], "sk_lock-AF_INET-NFS", &xs_key[0]); 1189 break; 1190 1191 case AF_INET6: 1192 sock_lock_init_class_and_name(sk, "slock-AF_INET6-NFS", 1193 &xs_slock_key[1], "sk_lock-AF_INET6-NFS", &xs_key[1]); 1194 break; 1195 1196 default: 1197 BUG(); 1198 } 1199 } 1200 #else 1201 static inline void xs_reclassify_socket(struct socket *sock) 1202 { 1203 } 1204 #endif 1205 1206 /** 1207 * xs_udp_connect_worker - set up a UDP socket 1208 * @work: RPC transport to connect 1209 * 1210 * Invoked by a work queue tasklet. 1211 */ 1212 static void xs_udp_connect_worker(struct work_struct *work) 1213 { 1214 struct sock_xprt *transport = 1215 container_of(work, struct sock_xprt, connect_worker.work); 1216 struct rpc_xprt *xprt = &transport->xprt; 1217 struct socket *sock = transport->sock; 1218 int err, status = -EIO; 1219 1220 if (xprt->shutdown || !xprt_bound(xprt)) 1221 goto out; 1222 1223 /* Start by resetting any existing state */ 1224 xs_close(xprt); 1225 1226 if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) { 1227 dprintk("RPC: can't create UDP transport socket (%d).\n", -err); 1228 goto out; 1229 } 1230 xs_reclassify_socket(sock); 1231 1232 if (xprt->resvport && xs_bindresvport(transport, sock) < 0) { 1233 sock_release(sock); 1234 goto out; 1235 } 1236 1237 dprintk("RPC: worker connecting xprt %p to address: %s\n", 1238 xprt, xprt->address_strings[RPC_DISPLAY_ALL]); 1239 1240 if (!transport->inet) { 1241 struct sock *sk = sock->sk; 1242 1243 write_lock_bh(&sk->sk_callback_lock); 1244 1245 sk->sk_user_data = xprt; 1246 transport->old_data_ready = sk->sk_data_ready; 1247 transport->old_state_change = sk->sk_state_change; 1248 transport->old_write_space = sk->sk_write_space; 1249 sk->sk_data_ready = xs_udp_data_ready; 1250 sk->sk_write_space = xs_udp_write_space; 1251 sk->sk_no_check = UDP_CSUM_NORCV; 1252 sk->sk_allocation = GFP_ATOMIC; 1253 1254 xprt_set_connected(xprt); 1255 1256 /* Reset to new socket */ 1257 transport->sock = sock; 1258 transport->inet = sk; 1259 1260 write_unlock_bh(&sk->sk_callback_lock); 1261 } 1262 xs_udp_do_set_buffer_size(xprt); 1263 status = 0; 1264 out: 1265 xprt_wake_pending_tasks(xprt, status); 1266 xprt_clear_connecting(xprt); 1267 } 1268 1269 /* 1270 * We need to preserve the port number so the reply cache on the server can 1271 * find our cached RPC replies when we get around to reconnecting. 1272 */ 1273 static void xs_tcp_reuse_connection(struct rpc_xprt *xprt) 1274 { 1275 int result; 1276 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1277 struct sockaddr any; 1278 1279 dprintk("RPC: disconnecting xprt %p to reuse port\n", xprt); 1280 1281 /* 1282 * Disconnect the transport socket by doing a connect operation 1283 * with AF_UNSPEC. This should return immediately... 1284 */ 1285 memset(&any, 0, sizeof(any)); 1286 any.sa_family = AF_UNSPEC; 1287 result = kernel_connect(transport->sock, &any, sizeof(any), 0); 1288 if (result) 1289 dprintk("RPC: AF_UNSPEC connect return code %d\n", 1290 result); 1291 } 1292 1293 /** 1294 * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint 1295 * @work: RPC transport to connect 1296 * 1297 * Invoked by a work queue tasklet. 1298 */ 1299 static void xs_tcp_connect_worker(struct work_struct *work) 1300 { 1301 struct sock_xprt *transport = 1302 container_of(work, struct sock_xprt, connect_worker.work); 1303 struct rpc_xprt *xprt = &transport->xprt; 1304 struct socket *sock = transport->sock; 1305 int err, status = -EIO; 1306 1307 if (xprt->shutdown || !xprt_bound(xprt)) 1308 goto out; 1309 1310 if (!sock) { 1311 /* start from scratch */ 1312 if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) { 1313 dprintk("RPC: can't create TCP transport " 1314 "socket (%d).\n", -err); 1315 goto out; 1316 } 1317 xs_reclassify_socket(sock); 1318 1319 if (xprt->resvport && xs_bindresvport(transport, sock) < 0) { 1320 sock_release(sock); 1321 goto out; 1322 } 1323 } else 1324 /* "close" the socket, preserving the local port */ 1325 xs_tcp_reuse_connection(xprt); 1326 1327 dprintk("RPC: worker connecting xprt %p to address: %s\n", 1328 xprt, xprt->address_strings[RPC_DISPLAY_ALL]); 1329 1330 if (!transport->inet) { 1331 struct sock *sk = sock->sk; 1332 1333 write_lock_bh(&sk->sk_callback_lock); 1334 1335 sk->sk_user_data = xprt; 1336 transport->old_data_ready = sk->sk_data_ready; 1337 transport->old_state_change = sk->sk_state_change; 1338 transport->old_write_space = sk->sk_write_space; 1339 sk->sk_data_ready = xs_tcp_data_ready; 1340 sk->sk_state_change = xs_tcp_state_change; 1341 sk->sk_write_space = xs_tcp_write_space; 1342 sk->sk_allocation = GFP_ATOMIC; 1343 1344 /* socket options */ 1345 sk->sk_userlocks |= SOCK_BINDPORT_LOCK; 1346 sock_reset_flag(sk, SOCK_LINGER); 1347 tcp_sk(sk)->linger2 = 0; 1348 tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF; 1349 1350 xprt_clear_connected(xprt); 1351 1352 /* Reset to new socket */ 1353 transport->sock = sock; 1354 transport->inet = sk; 1355 1356 write_unlock_bh(&sk->sk_callback_lock); 1357 } 1358 1359 /* Tell the socket layer to start connecting... */ 1360 xprt->stat.connect_count++; 1361 xprt->stat.connect_start = jiffies; 1362 status = kernel_connect(sock, (struct sockaddr *) &xprt->addr, 1363 xprt->addrlen, O_NONBLOCK); 1364 dprintk("RPC: %p connect status %d connected %d sock state %d\n", 1365 xprt, -status, xprt_connected(xprt), 1366 sock->sk->sk_state); 1367 if (status < 0) { 1368 switch (status) { 1369 case -EINPROGRESS: 1370 case -EALREADY: 1371 goto out_clear; 1372 case -ECONNREFUSED: 1373 case -ECONNRESET: 1374 /* retry with existing socket, after a delay */ 1375 break; 1376 default: 1377 /* get rid of existing socket, and retry */ 1378 xs_close(xprt); 1379 break; 1380 } 1381 } 1382 out: 1383 xprt_wake_pending_tasks(xprt, status); 1384 out_clear: 1385 xprt_clear_connecting(xprt); 1386 } 1387 1388 /** 1389 * xs_connect - connect a socket to a remote endpoint 1390 * @task: address of RPC task that manages state of connect request 1391 * 1392 * TCP: If the remote end dropped the connection, delay reconnecting. 1393 * 1394 * UDP socket connects are synchronous, but we use a work queue anyway 1395 * to guarantee that even unprivileged user processes can set up a 1396 * socket on a privileged port. 1397 * 1398 * If a UDP socket connect fails, the delay behavior here prevents 1399 * retry floods (hard mounts). 1400 */ 1401 static void xs_connect(struct rpc_task *task) 1402 { 1403 struct rpc_xprt *xprt = task->tk_xprt; 1404 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1405 1406 if (xprt_test_and_set_connecting(xprt)) 1407 return; 1408 1409 if (transport->sock != NULL) { 1410 dprintk("RPC: xs_connect delayed xprt %p for %lu " 1411 "seconds\n", 1412 xprt, xprt->reestablish_timeout / HZ); 1413 schedule_delayed_work(&transport->connect_worker, 1414 xprt->reestablish_timeout); 1415 xprt->reestablish_timeout <<= 1; 1416 if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO) 1417 xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO; 1418 } else { 1419 dprintk("RPC: xs_connect scheduled xprt %p\n", xprt); 1420 schedule_delayed_work(&transport->connect_worker, 0); 1421 1422 /* flush_scheduled_work can sleep... */ 1423 if (!RPC_IS_ASYNC(task)) 1424 flush_scheduled_work(); 1425 } 1426 } 1427 1428 /** 1429 * xs_udp_print_stats - display UDP socket-specifc stats 1430 * @xprt: rpc_xprt struct containing statistics 1431 * @seq: output file 1432 * 1433 */ 1434 static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 1435 { 1436 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1437 1438 seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n", 1439 transport->port, 1440 xprt->stat.bind_count, 1441 xprt->stat.sends, 1442 xprt->stat.recvs, 1443 xprt->stat.bad_xids, 1444 xprt->stat.req_u, 1445 xprt->stat.bklog_u); 1446 } 1447 1448 /** 1449 * xs_tcp_print_stats - display TCP socket-specifc stats 1450 * @xprt: rpc_xprt struct containing statistics 1451 * @seq: output file 1452 * 1453 */ 1454 static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 1455 { 1456 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1457 long idle_time = 0; 1458 1459 if (xprt_connected(xprt)) 1460 idle_time = (long)(jiffies - xprt->last_used) / HZ; 1461 1462 seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n", 1463 transport->port, 1464 xprt->stat.bind_count, 1465 xprt->stat.connect_count, 1466 xprt->stat.connect_time, 1467 idle_time, 1468 xprt->stat.sends, 1469 xprt->stat.recvs, 1470 xprt->stat.bad_xids, 1471 xprt->stat.req_u, 1472 xprt->stat.bklog_u); 1473 } 1474 1475 static struct rpc_xprt_ops xs_udp_ops = { 1476 .set_buffer_size = xs_udp_set_buffer_size, 1477 .reserve_xprt = xprt_reserve_xprt_cong, 1478 .release_xprt = xprt_release_xprt_cong, 1479 .rpcbind = rpcb_getport, 1480 .set_port = xs_set_port, 1481 .connect = xs_connect, 1482 .buf_alloc = rpc_malloc, 1483 .buf_free = rpc_free, 1484 .send_request = xs_udp_send_request, 1485 .set_retrans_timeout = xprt_set_retrans_timeout_rtt, 1486 .timer = xs_udp_timer, 1487 .release_request = xprt_release_rqst_cong, 1488 .close = xs_close, 1489 .destroy = xs_destroy, 1490 .print_stats = xs_udp_print_stats, 1491 }; 1492 1493 static struct rpc_xprt_ops xs_tcp_ops = { 1494 .reserve_xprt = xprt_reserve_xprt, 1495 .release_xprt = xs_tcp_release_xprt, 1496 .rpcbind = rpcb_getport, 1497 .set_port = xs_set_port, 1498 .connect = xs_connect, 1499 .buf_alloc = rpc_malloc, 1500 .buf_free = rpc_free, 1501 .send_request = xs_tcp_send_request, 1502 .set_retrans_timeout = xprt_set_retrans_timeout_def, 1503 .close = xs_close, 1504 .destroy = xs_destroy, 1505 .print_stats = xs_tcp_print_stats, 1506 }; 1507 1508 static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size) 1509 { 1510 struct rpc_xprt *xprt; 1511 struct sock_xprt *new; 1512 1513 if (addrlen > sizeof(xprt->addr)) { 1514 dprintk("RPC: xs_setup_xprt: address too large\n"); 1515 return ERR_PTR(-EBADF); 1516 } 1517 1518 new = kzalloc(sizeof(*new), GFP_KERNEL); 1519 if (new == NULL) { 1520 dprintk("RPC: xs_setup_xprt: couldn't allocate " 1521 "rpc_xprt\n"); 1522 return ERR_PTR(-ENOMEM); 1523 } 1524 xprt = &new->xprt; 1525 1526 xprt->max_reqs = slot_table_size; 1527 xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL); 1528 if (xprt->slot == NULL) { 1529 kfree(xprt); 1530 dprintk("RPC: xs_setup_xprt: couldn't allocate slot " 1531 "table\n"); 1532 return ERR_PTR(-ENOMEM); 1533 } 1534 1535 memcpy(&xprt->addr, addr, addrlen); 1536 xprt->addrlen = addrlen; 1537 new->port = xs_get_random_port(); 1538 1539 return xprt; 1540 } 1541 1542 /** 1543 * xs_setup_udp - Set up transport to use a UDP socket 1544 * @addr: address of remote server 1545 * @addrlen: length of address in bytes 1546 * @to: timeout parameters 1547 * 1548 */ 1549 struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to) 1550 { 1551 struct rpc_xprt *xprt; 1552 struct sock_xprt *transport; 1553 1554 xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries); 1555 if (IS_ERR(xprt)) 1556 return xprt; 1557 transport = container_of(xprt, struct sock_xprt, xprt); 1558 1559 if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0) 1560 xprt_set_bound(xprt); 1561 1562 xprt->prot = IPPROTO_UDP; 1563 xprt->tsh_size = 0; 1564 /* XXX: header size can vary due to auth type, IPv6, etc. */ 1565 xprt->max_payload = (1U << 16) - (MAX_HEADER << 3); 1566 1567 INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_connect_worker); 1568 xprt->bind_timeout = XS_BIND_TO; 1569 xprt->connect_timeout = XS_UDP_CONN_TO; 1570 xprt->reestablish_timeout = XS_UDP_REEST_TO; 1571 xprt->idle_timeout = XS_IDLE_DISC_TO; 1572 1573 xprt->ops = &xs_udp_ops; 1574 1575 if (to) 1576 xprt->timeout = *to; 1577 else 1578 xprt_set_timeout(&xprt->timeout, 5, 5 * HZ); 1579 1580 xs_format_peer_addresses(xprt); 1581 dprintk("RPC: set up transport to address %s\n", 1582 xprt->address_strings[RPC_DISPLAY_ALL]); 1583 1584 return xprt; 1585 } 1586 1587 /** 1588 * xs_setup_tcp - Set up transport to use a TCP socket 1589 * @addr: address of remote server 1590 * @addrlen: length of address in bytes 1591 * @to: timeout parameters 1592 * 1593 */ 1594 struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to) 1595 { 1596 struct rpc_xprt *xprt; 1597 struct sock_xprt *transport; 1598 1599 xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries); 1600 if (IS_ERR(xprt)) 1601 return xprt; 1602 transport = container_of(xprt, struct sock_xprt, xprt); 1603 1604 if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0) 1605 xprt_set_bound(xprt); 1606 1607 xprt->prot = IPPROTO_TCP; 1608 xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32); 1609 xprt->max_payload = RPC_MAX_FRAGMENT_SIZE; 1610 1611 INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker); 1612 xprt->bind_timeout = XS_BIND_TO; 1613 xprt->connect_timeout = XS_TCP_CONN_TO; 1614 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO; 1615 xprt->idle_timeout = XS_IDLE_DISC_TO; 1616 1617 xprt->ops = &xs_tcp_ops; 1618 1619 if (to) 1620 xprt->timeout = *to; 1621 else 1622 xprt_set_timeout(&xprt->timeout, 2, 60 * HZ); 1623 1624 xs_format_peer_addresses(xprt); 1625 dprintk("RPC: set up transport to address %s\n", 1626 xprt->address_strings[RPC_DISPLAY_ALL]); 1627 1628 return xprt; 1629 } 1630 1631 /** 1632 * init_socket_xprt - set up xprtsock's sysctls 1633 * 1634 */ 1635 int init_socket_xprt(void) 1636 { 1637 #ifdef RPC_DEBUG 1638 if (!sunrpc_table_header) 1639 sunrpc_table_header = register_sysctl_table(sunrpc_table); 1640 #endif 1641 1642 return 0; 1643 } 1644 1645 /** 1646 * cleanup_socket_xprt - remove xprtsock's sysctls 1647 * 1648 */ 1649 void cleanup_socket_xprt(void) 1650 { 1651 #ifdef RPC_DEBUG 1652 if (sunrpc_table_header) { 1653 unregister_sysctl_table(sunrpc_table_header); 1654 sunrpc_table_header = NULL; 1655 } 1656 #endif 1657 } 1658