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 struct sockaddr_storage addr; 239 unsigned short port; 240 241 /* 242 * UDP socket buffer size parameters 243 */ 244 size_t rcvsize, 245 sndsize; 246 247 /* 248 * Saved socket callback addresses 249 */ 250 void (*old_data_ready)(struct sock *, int); 251 void (*old_state_change)(struct sock *); 252 void (*old_write_space)(struct sock *); 253 }; 254 255 /* 256 * TCP receive state flags 257 */ 258 #define TCP_RCV_LAST_FRAG (1UL << 0) 259 #define TCP_RCV_COPY_FRAGHDR (1UL << 1) 260 #define TCP_RCV_COPY_XID (1UL << 2) 261 #define TCP_RCV_COPY_DATA (1UL << 3) 262 263 static void xs_format_peer_addresses(struct rpc_xprt *xprt) 264 { 265 struct sockaddr_in *addr = (struct sockaddr_in *) &xprt->addr; 266 char *buf; 267 268 buf = kzalloc(20, GFP_KERNEL); 269 if (buf) { 270 snprintf(buf, 20, "%u.%u.%u.%u", 271 NIPQUAD(addr->sin_addr.s_addr)); 272 } 273 xprt->address_strings[RPC_DISPLAY_ADDR] = buf; 274 275 buf = kzalloc(8, GFP_KERNEL); 276 if (buf) { 277 snprintf(buf, 8, "%u", 278 ntohs(addr->sin_port)); 279 } 280 xprt->address_strings[RPC_DISPLAY_PORT] = buf; 281 282 if (xprt->prot == IPPROTO_UDP) 283 xprt->address_strings[RPC_DISPLAY_PROTO] = "udp"; 284 else 285 xprt->address_strings[RPC_DISPLAY_PROTO] = "tcp"; 286 287 buf = kzalloc(48, GFP_KERNEL); 288 if (buf) { 289 snprintf(buf, 48, "addr=%u.%u.%u.%u port=%u proto=%s", 290 NIPQUAD(addr->sin_addr.s_addr), 291 ntohs(addr->sin_port), 292 xprt->prot == IPPROTO_UDP ? "udp" : "tcp"); 293 } 294 xprt->address_strings[RPC_DISPLAY_ALL] = buf; 295 } 296 297 static void xs_free_peer_addresses(struct rpc_xprt *xprt) 298 { 299 kfree(xprt->address_strings[RPC_DISPLAY_ADDR]); 300 kfree(xprt->address_strings[RPC_DISPLAY_PORT]); 301 kfree(xprt->address_strings[RPC_DISPLAY_ALL]); 302 } 303 304 #define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL) 305 306 static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more) 307 { 308 struct msghdr msg = { 309 .msg_name = addr, 310 .msg_namelen = addrlen, 311 .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0), 312 }; 313 struct kvec iov = { 314 .iov_base = vec->iov_base + base, 315 .iov_len = vec->iov_len - base, 316 }; 317 318 if (iov.iov_len != 0) 319 return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len); 320 return kernel_sendmsg(sock, &msg, NULL, 0, 0); 321 } 322 323 static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more) 324 { 325 struct page **ppage; 326 unsigned int remainder; 327 int err, sent = 0; 328 329 remainder = xdr->page_len - base; 330 base += xdr->page_base; 331 ppage = xdr->pages + (base >> PAGE_SHIFT); 332 base &= ~PAGE_MASK; 333 for(;;) { 334 unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder); 335 int flags = XS_SENDMSG_FLAGS; 336 337 remainder -= len; 338 if (remainder != 0 || more) 339 flags |= MSG_MORE; 340 err = sock->ops->sendpage(sock, *ppage, base, len, flags); 341 if (remainder == 0 || err != len) 342 break; 343 sent += err; 344 ppage++; 345 base = 0; 346 } 347 if (sent == 0) 348 return err; 349 if (err > 0) 350 sent += err; 351 return sent; 352 } 353 354 /** 355 * xs_sendpages - write pages directly to a socket 356 * @sock: socket to send on 357 * @addr: UDP only -- address of destination 358 * @addrlen: UDP only -- length of destination address 359 * @xdr: buffer containing this request 360 * @base: starting position in the buffer 361 * 362 */ 363 static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base) 364 { 365 unsigned int remainder = xdr->len - base; 366 int err, sent = 0; 367 368 if (unlikely(!sock)) 369 return -ENOTCONN; 370 371 clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags); 372 if (base != 0) { 373 addr = NULL; 374 addrlen = 0; 375 } 376 377 if (base < xdr->head[0].iov_len || addr != NULL) { 378 unsigned int len = xdr->head[0].iov_len - base; 379 remainder -= len; 380 err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0); 381 if (remainder == 0 || err != len) 382 goto out; 383 sent += err; 384 base = 0; 385 } else 386 base -= xdr->head[0].iov_len; 387 388 if (base < xdr->page_len) { 389 unsigned int len = xdr->page_len - base; 390 remainder -= len; 391 err = xs_send_pagedata(sock, xdr, base, remainder != 0); 392 if (remainder == 0 || err != len) 393 goto out; 394 sent += err; 395 base = 0; 396 } else 397 base -= xdr->page_len; 398 399 if (base >= xdr->tail[0].iov_len) 400 return sent; 401 err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0); 402 out: 403 if (sent == 0) 404 return err; 405 if (err > 0) 406 sent += err; 407 return sent; 408 } 409 410 /** 411 * xs_nospace - place task on wait queue if transmit was incomplete 412 * @task: task to put to sleep 413 * 414 */ 415 static void xs_nospace(struct rpc_task *task) 416 { 417 struct rpc_rqst *req = task->tk_rqstp; 418 struct rpc_xprt *xprt = req->rq_xprt; 419 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 420 421 dprintk("RPC: %5u xmit incomplete (%u left of %u)\n", 422 task->tk_pid, req->rq_slen - req->rq_bytes_sent, 423 req->rq_slen); 424 425 if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) { 426 /* Protect against races with write_space */ 427 spin_lock_bh(&xprt->transport_lock); 428 429 /* Don't race with disconnect */ 430 if (!xprt_connected(xprt)) 431 task->tk_status = -ENOTCONN; 432 else if (test_bit(SOCK_NOSPACE, &transport->sock->flags)) 433 xprt_wait_for_buffer_space(task); 434 435 spin_unlock_bh(&xprt->transport_lock); 436 } else 437 /* Keep holding the socket if it is blocked */ 438 rpc_delay(task, HZ>>4); 439 } 440 441 /** 442 * xs_udp_send_request - write an RPC request to a UDP socket 443 * @task: address of RPC task that manages the state of an RPC request 444 * 445 * Return values: 446 * 0: The request has been sent 447 * EAGAIN: The socket was blocked, please call again later to 448 * complete the request 449 * ENOTCONN: Caller needs to invoke connect logic then call again 450 * other: Some other error occured, the request was not sent 451 */ 452 static int xs_udp_send_request(struct rpc_task *task) 453 { 454 struct rpc_rqst *req = task->tk_rqstp; 455 struct rpc_xprt *xprt = req->rq_xprt; 456 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 457 struct xdr_buf *xdr = &req->rq_snd_buf; 458 int status; 459 460 xs_pktdump("packet data:", 461 req->rq_svec->iov_base, 462 req->rq_svec->iov_len); 463 464 req->rq_xtime = jiffies; 465 status = xs_sendpages(transport->sock, 466 (struct sockaddr *) &xprt->addr, 467 xprt->addrlen, xdr, 468 req->rq_bytes_sent); 469 470 dprintk("RPC: xs_udp_send_request(%u) = %d\n", 471 xdr->len - req->rq_bytes_sent, status); 472 473 if (likely(status >= (int) req->rq_slen)) 474 return 0; 475 476 /* Still some bytes left; set up for a retry later. */ 477 if (status > 0) 478 status = -EAGAIN; 479 480 switch (status) { 481 case -ENETUNREACH: 482 case -EPIPE: 483 case -ECONNREFUSED: 484 /* When the server has died, an ICMP port unreachable message 485 * prompts ECONNREFUSED. */ 486 break; 487 case -EAGAIN: 488 xs_nospace(task); 489 break; 490 default: 491 dprintk("RPC: sendmsg returned unrecognized error %d\n", 492 -status); 493 break; 494 } 495 496 return status; 497 } 498 499 static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf) 500 { 501 u32 reclen = buf->len - sizeof(rpc_fraghdr); 502 rpc_fraghdr *base = buf->head[0].iov_base; 503 *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen); 504 } 505 506 /** 507 * xs_tcp_send_request - write an RPC request to a TCP socket 508 * @task: address of RPC task that manages the state of an RPC request 509 * 510 * Return values: 511 * 0: The request has been sent 512 * EAGAIN: The socket was blocked, please call again later to 513 * complete the request 514 * ENOTCONN: Caller needs to invoke connect logic then call again 515 * other: Some other error occured, the request was not sent 516 * 517 * XXX: In the case of soft timeouts, should we eventually give up 518 * if sendmsg is not able to make progress? 519 */ 520 static int xs_tcp_send_request(struct rpc_task *task) 521 { 522 struct rpc_rqst *req = task->tk_rqstp; 523 struct rpc_xprt *xprt = req->rq_xprt; 524 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 525 struct xdr_buf *xdr = &req->rq_snd_buf; 526 int status, retry = 0; 527 528 xs_encode_tcp_record_marker(&req->rq_snd_buf); 529 530 xs_pktdump("packet data:", 531 req->rq_svec->iov_base, 532 req->rq_svec->iov_len); 533 534 /* Continue transmitting the packet/record. We must be careful 535 * to cope with writespace callbacks arriving _after_ we have 536 * called sendmsg(). */ 537 while (1) { 538 req->rq_xtime = jiffies; 539 status = xs_sendpages(transport->sock, 540 NULL, 0, xdr, req->rq_bytes_sent); 541 542 dprintk("RPC: xs_tcp_send_request(%u) = %d\n", 543 xdr->len - req->rq_bytes_sent, status); 544 545 if (unlikely(status < 0)) 546 break; 547 548 /* If we've sent the entire packet, immediately 549 * reset the count of bytes sent. */ 550 req->rq_bytes_sent += status; 551 task->tk_bytes_sent += status; 552 if (likely(req->rq_bytes_sent >= req->rq_slen)) { 553 req->rq_bytes_sent = 0; 554 return 0; 555 } 556 557 status = -EAGAIN; 558 if (retry++ > XS_SENDMSG_RETRY) 559 break; 560 } 561 562 switch (status) { 563 case -EAGAIN: 564 xs_nospace(task); 565 break; 566 case -ECONNREFUSED: 567 case -ECONNRESET: 568 case -ENOTCONN: 569 case -EPIPE: 570 status = -ENOTCONN; 571 break; 572 default: 573 dprintk("RPC: sendmsg returned unrecognized error %d\n", 574 -status); 575 xprt_disconnect(xprt); 576 break; 577 } 578 579 return status; 580 } 581 582 /** 583 * xs_tcp_release_xprt - clean up after a tcp transmission 584 * @xprt: transport 585 * @task: rpc task 586 * 587 * This cleans up if an error causes us to abort the transmission of a request. 588 * In this case, the socket may need to be reset in order to avoid confusing 589 * the server. 590 */ 591 static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 592 { 593 struct rpc_rqst *req; 594 595 if (task != xprt->snd_task) 596 return; 597 if (task == NULL) 598 goto out_release; 599 req = task->tk_rqstp; 600 if (req->rq_bytes_sent == 0) 601 goto out_release; 602 if (req->rq_bytes_sent == req->rq_snd_buf.len) 603 goto out_release; 604 set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state); 605 out_release: 606 xprt_release_xprt(xprt, task); 607 } 608 609 /** 610 * xs_close - close a socket 611 * @xprt: transport 612 * 613 * This is used when all requests are complete; ie, no DRC state remains 614 * on the server we want to save. 615 */ 616 static void xs_close(struct rpc_xprt *xprt) 617 { 618 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 619 struct socket *sock = transport->sock; 620 struct sock *sk = transport->inet; 621 622 if (!sk) 623 goto clear_close_wait; 624 625 dprintk("RPC: xs_close xprt %p\n", xprt); 626 627 write_lock_bh(&sk->sk_callback_lock); 628 transport->inet = NULL; 629 transport->sock = NULL; 630 631 sk->sk_user_data = NULL; 632 sk->sk_data_ready = transport->old_data_ready; 633 sk->sk_state_change = transport->old_state_change; 634 sk->sk_write_space = transport->old_write_space; 635 write_unlock_bh(&sk->sk_callback_lock); 636 637 sk->sk_no_check = 0; 638 639 sock_release(sock); 640 clear_close_wait: 641 smp_mb__before_clear_bit(); 642 clear_bit(XPRT_CLOSE_WAIT, &xprt->state); 643 smp_mb__after_clear_bit(); 644 } 645 646 /** 647 * xs_destroy - prepare to shutdown a transport 648 * @xprt: doomed transport 649 * 650 */ 651 static void xs_destroy(struct rpc_xprt *xprt) 652 { 653 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 654 655 dprintk("RPC: xs_destroy xprt %p\n", xprt); 656 657 cancel_rearming_delayed_work(&transport->connect_worker); 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 queue_work(rpciod_workqueue, &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_bind(struct sock_xprt *transport, struct socket *sock) 1150 { 1151 struct sockaddr_in myaddr = { 1152 .sin_family = AF_INET, 1153 }; 1154 struct sockaddr_in *sa; 1155 int err; 1156 unsigned short port = transport->port; 1157 1158 if (!transport->xprt.resvport) 1159 port = 0; 1160 sa = (struct sockaddr_in *)&transport->addr; 1161 myaddr.sin_addr = sa->sin_addr; 1162 do { 1163 myaddr.sin_port = htons(port); 1164 err = kernel_bind(sock, (struct sockaddr *) &myaddr, 1165 sizeof(myaddr)); 1166 if (!transport->xprt.resvport) 1167 break; 1168 if (err == 0) { 1169 transport->port = port; 1170 break; 1171 } 1172 if (port <= xprt_min_resvport) 1173 port = xprt_max_resvport; 1174 else 1175 port--; 1176 } while (err == -EADDRINUSE && port != transport->port); 1177 dprintk("RPC: xs_bind "NIPQUAD_FMT":%u: %s (%d)\n", 1178 NIPQUAD(myaddr.sin_addr), port, err ? "failed" : "ok", err); 1179 return err; 1180 } 1181 1182 #ifdef CONFIG_DEBUG_LOCK_ALLOC 1183 static struct lock_class_key xs_key[2]; 1184 static struct lock_class_key xs_slock_key[2]; 1185 1186 static inline void xs_reclassify_socket(struct socket *sock) 1187 { 1188 struct sock *sk = sock->sk; 1189 BUG_ON(sock_owned_by_user(sk)); 1190 switch (sk->sk_family) { 1191 case AF_INET: 1192 sock_lock_init_class_and_name(sk, "slock-AF_INET-NFS", 1193 &xs_slock_key[0], "sk_lock-AF_INET-NFS", &xs_key[0]); 1194 break; 1195 1196 case AF_INET6: 1197 sock_lock_init_class_and_name(sk, "slock-AF_INET6-NFS", 1198 &xs_slock_key[1], "sk_lock-AF_INET6-NFS", &xs_key[1]); 1199 break; 1200 1201 default: 1202 BUG(); 1203 } 1204 } 1205 #else 1206 static inline void xs_reclassify_socket(struct socket *sock) 1207 { 1208 } 1209 #endif 1210 1211 /** 1212 * xs_udp_connect_worker - set up a UDP socket 1213 * @work: RPC transport to connect 1214 * 1215 * Invoked by a work queue tasklet. 1216 */ 1217 static void xs_udp_connect_worker(struct work_struct *work) 1218 { 1219 struct sock_xprt *transport = 1220 container_of(work, struct sock_xprt, connect_worker.work); 1221 struct rpc_xprt *xprt = &transport->xprt; 1222 struct socket *sock = transport->sock; 1223 int err, status = -EIO; 1224 1225 if (xprt->shutdown || !xprt_bound(xprt)) 1226 goto out; 1227 1228 /* Start by resetting any existing state */ 1229 xs_close(xprt); 1230 1231 if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) { 1232 dprintk("RPC: can't create UDP transport socket (%d).\n", -err); 1233 goto out; 1234 } 1235 xs_reclassify_socket(sock); 1236 1237 if (xs_bind(transport, sock)) { 1238 sock_release(sock); 1239 goto out; 1240 } 1241 1242 dprintk("RPC: worker connecting xprt %p to address: %s\n", 1243 xprt, xprt->address_strings[RPC_DISPLAY_ALL]); 1244 1245 if (!transport->inet) { 1246 struct sock *sk = sock->sk; 1247 1248 write_lock_bh(&sk->sk_callback_lock); 1249 1250 sk->sk_user_data = xprt; 1251 transport->old_data_ready = sk->sk_data_ready; 1252 transport->old_state_change = sk->sk_state_change; 1253 transport->old_write_space = sk->sk_write_space; 1254 sk->sk_data_ready = xs_udp_data_ready; 1255 sk->sk_write_space = xs_udp_write_space; 1256 sk->sk_no_check = UDP_CSUM_NORCV; 1257 sk->sk_allocation = GFP_ATOMIC; 1258 1259 xprt_set_connected(xprt); 1260 1261 /* Reset to new socket */ 1262 transport->sock = sock; 1263 transport->inet = sk; 1264 1265 write_unlock_bh(&sk->sk_callback_lock); 1266 } 1267 xs_udp_do_set_buffer_size(xprt); 1268 status = 0; 1269 out: 1270 xprt_wake_pending_tasks(xprt, status); 1271 xprt_clear_connecting(xprt); 1272 } 1273 1274 /* 1275 * We need to preserve the port number so the reply cache on the server can 1276 * find our cached RPC replies when we get around to reconnecting. 1277 */ 1278 static void xs_tcp_reuse_connection(struct rpc_xprt *xprt) 1279 { 1280 int result; 1281 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1282 struct sockaddr any; 1283 1284 dprintk("RPC: disconnecting xprt %p to reuse port\n", xprt); 1285 1286 /* 1287 * Disconnect the transport socket by doing a connect operation 1288 * with AF_UNSPEC. This should return immediately... 1289 */ 1290 memset(&any, 0, sizeof(any)); 1291 any.sa_family = AF_UNSPEC; 1292 result = kernel_connect(transport->sock, &any, sizeof(any), 0); 1293 if (result) 1294 dprintk("RPC: AF_UNSPEC connect return code %d\n", 1295 result); 1296 } 1297 1298 /** 1299 * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint 1300 * @work: RPC transport to connect 1301 * 1302 * Invoked by a work queue tasklet. 1303 */ 1304 static void xs_tcp_connect_worker(struct work_struct *work) 1305 { 1306 struct sock_xprt *transport = 1307 container_of(work, struct sock_xprt, connect_worker.work); 1308 struct rpc_xprt *xprt = &transport->xprt; 1309 struct socket *sock = transport->sock; 1310 int err, status = -EIO; 1311 1312 if (xprt->shutdown || !xprt_bound(xprt)) 1313 goto out; 1314 1315 if (!sock) { 1316 /* start from scratch */ 1317 if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) { 1318 dprintk("RPC: can't create TCP transport " 1319 "socket (%d).\n", -err); 1320 goto out; 1321 } 1322 xs_reclassify_socket(sock); 1323 1324 if (xs_bind(transport, sock)) { 1325 sock_release(sock); 1326 goto out; 1327 } 1328 } else 1329 /* "close" the socket, preserving the local port */ 1330 xs_tcp_reuse_connection(xprt); 1331 1332 dprintk("RPC: worker connecting xprt %p to address: %s\n", 1333 xprt, xprt->address_strings[RPC_DISPLAY_ALL]); 1334 1335 if (!transport->inet) { 1336 struct sock *sk = sock->sk; 1337 1338 write_lock_bh(&sk->sk_callback_lock); 1339 1340 sk->sk_user_data = xprt; 1341 transport->old_data_ready = sk->sk_data_ready; 1342 transport->old_state_change = sk->sk_state_change; 1343 transport->old_write_space = sk->sk_write_space; 1344 sk->sk_data_ready = xs_tcp_data_ready; 1345 sk->sk_state_change = xs_tcp_state_change; 1346 sk->sk_write_space = xs_tcp_write_space; 1347 sk->sk_allocation = GFP_ATOMIC; 1348 1349 /* socket options */ 1350 sk->sk_userlocks |= SOCK_BINDPORT_LOCK; 1351 sock_reset_flag(sk, SOCK_LINGER); 1352 tcp_sk(sk)->linger2 = 0; 1353 tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF; 1354 1355 xprt_clear_connected(xprt); 1356 1357 /* Reset to new socket */ 1358 transport->sock = sock; 1359 transport->inet = sk; 1360 1361 write_unlock_bh(&sk->sk_callback_lock); 1362 } 1363 1364 /* Tell the socket layer to start connecting... */ 1365 xprt->stat.connect_count++; 1366 xprt->stat.connect_start = jiffies; 1367 status = kernel_connect(sock, (struct sockaddr *) &xprt->addr, 1368 xprt->addrlen, O_NONBLOCK); 1369 dprintk("RPC: %p connect status %d connected %d sock state %d\n", 1370 xprt, -status, xprt_connected(xprt), 1371 sock->sk->sk_state); 1372 if (status < 0) { 1373 switch (status) { 1374 case -EINPROGRESS: 1375 case -EALREADY: 1376 goto out_clear; 1377 case -ECONNREFUSED: 1378 case -ECONNRESET: 1379 /* retry with existing socket, after a delay */ 1380 break; 1381 default: 1382 /* get rid of existing socket, and retry */ 1383 xs_close(xprt); 1384 break; 1385 } 1386 } 1387 out: 1388 xprt_wake_pending_tasks(xprt, status); 1389 out_clear: 1390 xprt_clear_connecting(xprt); 1391 } 1392 1393 /** 1394 * xs_connect - connect a socket to a remote endpoint 1395 * @task: address of RPC task that manages state of connect request 1396 * 1397 * TCP: If the remote end dropped the connection, delay reconnecting. 1398 * 1399 * UDP socket connects are synchronous, but we use a work queue anyway 1400 * to guarantee that even unprivileged user processes can set up a 1401 * socket on a privileged port. 1402 * 1403 * If a UDP socket connect fails, the delay behavior here prevents 1404 * retry floods (hard mounts). 1405 */ 1406 static void xs_connect(struct rpc_task *task) 1407 { 1408 struct rpc_xprt *xprt = task->tk_xprt; 1409 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1410 1411 if (xprt_test_and_set_connecting(xprt)) 1412 return; 1413 1414 if (transport->sock != NULL) { 1415 dprintk("RPC: xs_connect delayed xprt %p for %lu " 1416 "seconds\n", 1417 xprt, xprt->reestablish_timeout / HZ); 1418 queue_delayed_work(rpciod_workqueue, 1419 &transport->connect_worker, 1420 xprt->reestablish_timeout); 1421 xprt->reestablish_timeout <<= 1; 1422 if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO) 1423 xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO; 1424 } else { 1425 dprintk("RPC: xs_connect scheduled xprt %p\n", xprt); 1426 queue_delayed_work(rpciod_workqueue, 1427 &transport->connect_worker, 0); 1428 } 1429 } 1430 1431 /** 1432 * xs_udp_print_stats - display UDP socket-specifc stats 1433 * @xprt: rpc_xprt struct containing statistics 1434 * @seq: output file 1435 * 1436 */ 1437 static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 1438 { 1439 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1440 1441 seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n", 1442 transport->port, 1443 xprt->stat.bind_count, 1444 xprt->stat.sends, 1445 xprt->stat.recvs, 1446 xprt->stat.bad_xids, 1447 xprt->stat.req_u, 1448 xprt->stat.bklog_u); 1449 } 1450 1451 /** 1452 * xs_tcp_print_stats - display TCP socket-specifc stats 1453 * @xprt: rpc_xprt struct containing statistics 1454 * @seq: output file 1455 * 1456 */ 1457 static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 1458 { 1459 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1460 long idle_time = 0; 1461 1462 if (xprt_connected(xprt)) 1463 idle_time = (long)(jiffies - xprt->last_used) / HZ; 1464 1465 seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n", 1466 transport->port, 1467 xprt->stat.bind_count, 1468 xprt->stat.connect_count, 1469 xprt->stat.connect_time, 1470 idle_time, 1471 xprt->stat.sends, 1472 xprt->stat.recvs, 1473 xprt->stat.bad_xids, 1474 xprt->stat.req_u, 1475 xprt->stat.bklog_u); 1476 } 1477 1478 static struct rpc_xprt_ops xs_udp_ops = { 1479 .set_buffer_size = xs_udp_set_buffer_size, 1480 .reserve_xprt = xprt_reserve_xprt_cong, 1481 .release_xprt = xprt_release_xprt_cong, 1482 .rpcbind = rpcb_getport_async, 1483 .set_port = xs_set_port, 1484 .connect = xs_connect, 1485 .buf_alloc = rpc_malloc, 1486 .buf_free = rpc_free, 1487 .send_request = xs_udp_send_request, 1488 .set_retrans_timeout = xprt_set_retrans_timeout_rtt, 1489 .timer = xs_udp_timer, 1490 .release_request = xprt_release_rqst_cong, 1491 .close = xs_close, 1492 .destroy = xs_destroy, 1493 .print_stats = xs_udp_print_stats, 1494 }; 1495 1496 static struct rpc_xprt_ops xs_tcp_ops = { 1497 .reserve_xprt = xprt_reserve_xprt, 1498 .release_xprt = xs_tcp_release_xprt, 1499 .rpcbind = rpcb_getport_async, 1500 .set_port = xs_set_port, 1501 .connect = xs_connect, 1502 .buf_alloc = rpc_malloc, 1503 .buf_free = rpc_free, 1504 .send_request = xs_tcp_send_request, 1505 .set_retrans_timeout = xprt_set_retrans_timeout_def, 1506 .close = xs_close, 1507 .destroy = xs_destroy, 1508 .print_stats = xs_tcp_print_stats, 1509 }; 1510 1511 static struct rpc_xprt *xs_setup_xprt(struct rpc_xprtsock_create *args, unsigned int slot_table_size) 1512 { 1513 struct rpc_xprt *xprt; 1514 struct sock_xprt *new; 1515 1516 if (args->addrlen > sizeof(xprt->addr)) { 1517 dprintk("RPC: xs_setup_xprt: address too large\n"); 1518 return ERR_PTR(-EBADF); 1519 } 1520 1521 new = kzalloc(sizeof(*new), GFP_KERNEL); 1522 if (new == NULL) { 1523 dprintk("RPC: xs_setup_xprt: couldn't allocate " 1524 "rpc_xprt\n"); 1525 return ERR_PTR(-ENOMEM); 1526 } 1527 xprt = &new->xprt; 1528 1529 xprt->max_reqs = slot_table_size; 1530 xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL); 1531 if (xprt->slot == NULL) { 1532 kfree(xprt); 1533 dprintk("RPC: xs_setup_xprt: couldn't allocate slot " 1534 "table\n"); 1535 return ERR_PTR(-ENOMEM); 1536 } 1537 1538 memcpy(&xprt->addr, args->dstaddr, args->addrlen); 1539 xprt->addrlen = args->addrlen; 1540 if (args->srcaddr) 1541 memcpy(&new->addr, args->srcaddr, args->addrlen); 1542 new->port = xs_get_random_port(); 1543 1544 return xprt; 1545 } 1546 1547 /** 1548 * xs_setup_udp - Set up transport to use a UDP socket 1549 * @args: rpc transport creation arguments 1550 * 1551 */ 1552 struct rpc_xprt *xs_setup_udp(struct rpc_xprtsock_create *args) 1553 { 1554 struct rpc_xprt *xprt; 1555 struct sock_xprt *transport; 1556 1557 xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries); 1558 if (IS_ERR(xprt)) 1559 return xprt; 1560 transport = container_of(xprt, struct sock_xprt, xprt); 1561 1562 if (ntohs(((struct sockaddr_in *)args->dstaddr)->sin_port) != 0) 1563 xprt_set_bound(xprt); 1564 1565 xprt->prot = IPPROTO_UDP; 1566 xprt->tsh_size = 0; 1567 /* XXX: header size can vary due to auth type, IPv6, etc. */ 1568 xprt->max_payload = (1U << 16) - (MAX_HEADER << 3); 1569 1570 INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_connect_worker); 1571 xprt->bind_timeout = XS_BIND_TO; 1572 xprt->connect_timeout = XS_UDP_CONN_TO; 1573 xprt->reestablish_timeout = XS_UDP_REEST_TO; 1574 xprt->idle_timeout = XS_IDLE_DISC_TO; 1575 1576 xprt->ops = &xs_udp_ops; 1577 1578 if (args->timeout) 1579 xprt->timeout = *args->timeout; 1580 else 1581 xprt_set_timeout(&xprt->timeout, 5, 5 * HZ); 1582 1583 xs_format_peer_addresses(xprt); 1584 dprintk("RPC: set up transport to address %s\n", 1585 xprt->address_strings[RPC_DISPLAY_ALL]); 1586 1587 return xprt; 1588 } 1589 1590 /** 1591 * xs_setup_tcp - Set up transport to use a TCP socket 1592 * @args: rpc transport creation arguments 1593 * 1594 */ 1595 struct rpc_xprt *xs_setup_tcp(struct rpc_xprtsock_create *args) 1596 { 1597 struct rpc_xprt *xprt; 1598 struct sock_xprt *transport; 1599 1600 xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries); 1601 if (IS_ERR(xprt)) 1602 return xprt; 1603 transport = container_of(xprt, struct sock_xprt, xprt); 1604 1605 if (ntohs(((struct sockaddr_in *)args->dstaddr)->sin_port) != 0) 1606 xprt_set_bound(xprt); 1607 1608 xprt->prot = IPPROTO_TCP; 1609 xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32); 1610 xprt->max_payload = RPC_MAX_FRAGMENT_SIZE; 1611 1612 INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker); 1613 xprt->bind_timeout = XS_BIND_TO; 1614 xprt->connect_timeout = XS_TCP_CONN_TO; 1615 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO; 1616 xprt->idle_timeout = XS_IDLE_DISC_TO; 1617 1618 xprt->ops = &xs_tcp_ops; 1619 1620 if (args->timeout) 1621 xprt->timeout = *args->timeout; 1622 else 1623 xprt_set_timeout(&xprt->timeout, 2, 60 * HZ); 1624 1625 xs_format_peer_addresses(xprt); 1626 dprintk("RPC: set up transport to address %s\n", 1627 xprt->address_strings[RPC_DISPLAY_ALL]); 1628 1629 return xprt; 1630 } 1631 1632 /** 1633 * init_socket_xprt - set up xprtsock's sysctls 1634 * 1635 */ 1636 int init_socket_xprt(void) 1637 { 1638 #ifdef RPC_DEBUG 1639 if (!sunrpc_table_header) 1640 sunrpc_table_header = register_sysctl_table(sunrpc_table); 1641 #endif 1642 1643 return 0; 1644 } 1645 1646 /** 1647 * cleanup_socket_xprt - remove xprtsock's sysctls 1648 * 1649 */ 1650 void cleanup_socket_xprt(void) 1651 { 1652 #ifdef RPC_DEBUG 1653 if (sunrpc_table_header) { 1654 unregister_sysctl_table(sunrpc_table_header); 1655 sunrpc_table_header = NULL; 1656 } 1657 #endif 1658 } 1659