1a246b010SChuck Lever /* 2a246b010SChuck Lever * linux/net/sunrpc/xprtsock.c 3a246b010SChuck Lever * 4a246b010SChuck Lever * Client-side transport implementation for sockets. 5a246b010SChuck Lever * 6a246b010SChuck Lever * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com> 7a246b010SChuck Lever * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com> 8a246b010SChuck Lever * TCP NFS related read + write fixes 9a246b010SChuck Lever * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie> 10a246b010SChuck Lever * 11a246b010SChuck Lever * Rewrite of larges part of the code in order to stabilize TCP stuff. 12a246b010SChuck Lever * Fix behaviour when socket buffer is full. 13a246b010SChuck Lever * (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no> 1455aa4f58SChuck Lever * 1555aa4f58SChuck Lever * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com> 16a246b010SChuck Lever */ 17a246b010SChuck Lever 18a246b010SChuck Lever #include <linux/types.h> 19a246b010SChuck Lever #include <linux/slab.h> 20a246b010SChuck Lever #include <linux/capability.h> 21a246b010SChuck Lever #include <linux/sched.h> 22a246b010SChuck Lever #include <linux/pagemap.h> 23a246b010SChuck Lever #include <linux/errno.h> 24a246b010SChuck Lever #include <linux/socket.h> 25a246b010SChuck Lever #include <linux/in.h> 26a246b010SChuck Lever #include <linux/net.h> 27a246b010SChuck Lever #include <linux/mm.h> 28a246b010SChuck Lever #include <linux/udp.h> 29a246b010SChuck Lever #include <linux/tcp.h> 30a246b010SChuck Lever #include <linux/sunrpc/clnt.h> 3102107148SChuck Lever #include <linux/sunrpc/sched.h> 32a246b010SChuck Lever #include <linux/file.h> 33a246b010SChuck Lever 34a246b010SChuck Lever #include <net/sock.h> 35a246b010SChuck Lever #include <net/checksum.h> 36a246b010SChuck Lever #include <net/udp.h> 37a246b010SChuck Lever #include <net/tcp.h> 38a246b010SChuck Lever 399903cd1cSChuck Lever /* 40c556b754SChuck Lever * xprtsock tunables 41c556b754SChuck Lever */ 42c556b754SChuck Lever unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE; 43c556b754SChuck Lever unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE; 44c556b754SChuck Lever 45c556b754SChuck Lever unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT; 46c556b754SChuck Lever unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT; 47c556b754SChuck Lever 48c556b754SChuck Lever /* 49262965f5SChuck Lever * How many times to try sending a request on a socket before waiting 50262965f5SChuck Lever * for the socket buffer to clear. 51262965f5SChuck Lever */ 52262965f5SChuck Lever #define XS_SENDMSG_RETRY (10U) 53262965f5SChuck Lever 5403bf4b70SChuck Lever /* 5503bf4b70SChuck Lever * Time out for an RPC UDP socket connect. UDP socket connects are 5603bf4b70SChuck Lever * synchronous, but we set a timeout anyway in case of resource 5703bf4b70SChuck Lever * exhaustion on the local host. 5803bf4b70SChuck Lever */ 5903bf4b70SChuck Lever #define XS_UDP_CONN_TO (5U * HZ) 6003bf4b70SChuck Lever 6103bf4b70SChuck Lever /* 6203bf4b70SChuck Lever * Wait duration for an RPC TCP connection to be established. Solaris 6303bf4b70SChuck Lever * NFS over TCP uses 60 seconds, for example, which is in line with how 6403bf4b70SChuck Lever * long a server takes to reboot. 6503bf4b70SChuck Lever */ 6603bf4b70SChuck Lever #define XS_TCP_CONN_TO (60U * HZ) 6703bf4b70SChuck Lever 6803bf4b70SChuck Lever /* 6903bf4b70SChuck Lever * Wait duration for a reply from the RPC portmapper. 7003bf4b70SChuck Lever */ 7103bf4b70SChuck Lever #define XS_BIND_TO (60U * HZ) 7203bf4b70SChuck Lever 7303bf4b70SChuck Lever /* 7403bf4b70SChuck Lever * Delay if a UDP socket connect error occurs. This is most likely some 7503bf4b70SChuck Lever * kind of resource problem on the local host. 7603bf4b70SChuck Lever */ 7703bf4b70SChuck Lever #define XS_UDP_REEST_TO (2U * HZ) 7803bf4b70SChuck Lever 7903bf4b70SChuck Lever /* 8003bf4b70SChuck Lever * The reestablish timeout allows clients to delay for a bit before attempting 8103bf4b70SChuck Lever * to reconnect to a server that just dropped our connection. 8203bf4b70SChuck Lever * 8303bf4b70SChuck Lever * We implement an exponential backoff when trying to reestablish a TCP 8403bf4b70SChuck Lever * transport connection with the server. Some servers like to drop a TCP 8503bf4b70SChuck Lever * connection when they are overworked, so we start with a short timeout and 8603bf4b70SChuck Lever * increase over time if the server is down or not responding. 8703bf4b70SChuck Lever */ 8803bf4b70SChuck Lever #define XS_TCP_INIT_REEST_TO (3U * HZ) 8903bf4b70SChuck Lever #define XS_TCP_MAX_REEST_TO (5U * 60 * HZ) 9003bf4b70SChuck Lever 9103bf4b70SChuck Lever /* 9203bf4b70SChuck Lever * TCP idle timeout; client drops the transport socket if it is idle 9303bf4b70SChuck Lever * for this long. Note that we also timeout UDP sockets to prevent 9403bf4b70SChuck Lever * holding port numbers when there is no RPC traffic. 9503bf4b70SChuck Lever */ 9603bf4b70SChuck Lever #define XS_IDLE_DISC_TO (5U * 60 * HZ) 9703bf4b70SChuck Lever 98a246b010SChuck Lever #ifdef RPC_DEBUG 99a246b010SChuck Lever # undef RPC_DEBUG_DATA 1009903cd1cSChuck Lever # define RPCDBG_FACILITY RPCDBG_TRANS 101a246b010SChuck Lever #endif 102a246b010SChuck Lever 103a246b010SChuck Lever #ifdef RPC_DEBUG_DATA 1049903cd1cSChuck Lever static void xs_pktdump(char *msg, u32 *packet, unsigned int count) 105a246b010SChuck Lever { 106a246b010SChuck Lever u8 *buf = (u8 *) packet; 107a246b010SChuck Lever int j; 108a246b010SChuck Lever 109a246b010SChuck Lever dprintk("RPC: %s\n", msg); 110a246b010SChuck Lever for (j = 0; j < count && j < 128; j += 4) { 111a246b010SChuck Lever if (!(j & 31)) { 112a246b010SChuck Lever if (j) 113a246b010SChuck Lever dprintk("\n"); 114a246b010SChuck Lever dprintk("0x%04x ", j); 115a246b010SChuck Lever } 116a246b010SChuck Lever dprintk("%02x%02x%02x%02x ", 117a246b010SChuck Lever buf[j], buf[j+1], buf[j+2], buf[j+3]); 118a246b010SChuck Lever } 119a246b010SChuck Lever dprintk("\n"); 120a246b010SChuck Lever } 121a246b010SChuck Lever #else 1229903cd1cSChuck Lever static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count) 123a246b010SChuck Lever { 124a246b010SChuck Lever /* NOP */ 125a246b010SChuck Lever } 126a246b010SChuck Lever #endif 127a246b010SChuck Lever 128ffc2e518SChuck Lever struct sock_xprt { 129ffc2e518SChuck Lever struct rpc_xprt xprt; 130ee0ac0c2SChuck Lever 131ee0ac0c2SChuck Lever /* 132ee0ac0c2SChuck Lever * Network layer 133ee0ac0c2SChuck Lever */ 134ee0ac0c2SChuck Lever struct socket * sock; 135ee0ac0c2SChuck Lever struct sock * inet; 13651971139SChuck Lever 13751971139SChuck Lever /* 13851971139SChuck Lever * State of TCP reply receive 13951971139SChuck Lever */ 14051971139SChuck Lever __be32 tcp_fraghdr, 14151971139SChuck Lever tcp_xid; 14251971139SChuck Lever 14351971139SChuck Lever u32 tcp_offset, 14451971139SChuck Lever tcp_reclen; 14551971139SChuck Lever 14651971139SChuck Lever unsigned long tcp_copied, 14751971139SChuck Lever tcp_flags; 148ffc2e518SChuck Lever }; 149ffc2e518SChuck Lever 150e136d092SChuck Lever /* 151e136d092SChuck Lever * TCP receive state flags 152e136d092SChuck Lever */ 153e136d092SChuck Lever #define TCP_RCV_LAST_FRAG (1UL << 0) 154e136d092SChuck Lever #define TCP_RCV_COPY_FRAGHDR (1UL << 1) 155e136d092SChuck Lever #define TCP_RCV_COPY_XID (1UL << 2) 156e136d092SChuck Lever #define TCP_RCV_COPY_DATA (1UL << 3) 157e136d092SChuck Lever 158edb267a6SChuck Lever static void xs_format_peer_addresses(struct rpc_xprt *xprt) 159edb267a6SChuck Lever { 160edb267a6SChuck Lever struct sockaddr_in *addr = (struct sockaddr_in *) &xprt->addr; 161edb267a6SChuck Lever char *buf; 162edb267a6SChuck Lever 163edb267a6SChuck Lever buf = kzalloc(20, GFP_KERNEL); 164edb267a6SChuck Lever if (buf) { 165edb267a6SChuck Lever snprintf(buf, 20, "%u.%u.%u.%u", 166edb267a6SChuck Lever NIPQUAD(addr->sin_addr.s_addr)); 167edb267a6SChuck Lever } 168edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_ADDR] = buf; 169edb267a6SChuck Lever 170edb267a6SChuck Lever buf = kzalloc(8, GFP_KERNEL); 171edb267a6SChuck Lever if (buf) { 172edb267a6SChuck Lever snprintf(buf, 8, "%u", 173edb267a6SChuck Lever ntohs(addr->sin_port)); 174edb267a6SChuck Lever } 175edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_PORT] = buf; 176edb267a6SChuck Lever 177edb267a6SChuck Lever if (xprt->prot == IPPROTO_UDP) 178edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_PROTO] = "udp"; 179edb267a6SChuck Lever else 180edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_PROTO] = "tcp"; 181edb267a6SChuck Lever 182edb267a6SChuck Lever buf = kzalloc(48, GFP_KERNEL); 183edb267a6SChuck Lever if (buf) { 184edb267a6SChuck Lever snprintf(buf, 48, "addr=%u.%u.%u.%u port=%u proto=%s", 185edb267a6SChuck Lever NIPQUAD(addr->sin_addr.s_addr), 186edb267a6SChuck Lever ntohs(addr->sin_port), 187edb267a6SChuck Lever xprt->prot == IPPROTO_UDP ? "udp" : "tcp"); 188edb267a6SChuck Lever } 189edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_ALL] = buf; 190edb267a6SChuck Lever } 191edb267a6SChuck Lever 192edb267a6SChuck Lever static void xs_free_peer_addresses(struct rpc_xprt *xprt) 193edb267a6SChuck Lever { 194edb267a6SChuck Lever kfree(xprt->address_strings[RPC_DISPLAY_ADDR]); 195edb267a6SChuck Lever kfree(xprt->address_strings[RPC_DISPLAY_PORT]); 196edb267a6SChuck Lever kfree(xprt->address_strings[RPC_DISPLAY_ALL]); 197edb267a6SChuck Lever } 198edb267a6SChuck Lever 199b4b5cc85SChuck Lever #define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL) 200b4b5cc85SChuck Lever 20124c5684bSTrond Myklebust static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more) 202b4b5cc85SChuck Lever { 203b4b5cc85SChuck Lever struct msghdr msg = { 204b4b5cc85SChuck Lever .msg_name = addr, 205b4b5cc85SChuck Lever .msg_namelen = addrlen, 20624c5684bSTrond Myklebust .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0), 20724c5684bSTrond Myklebust }; 20824c5684bSTrond Myklebust struct kvec iov = { 20924c5684bSTrond Myklebust .iov_base = vec->iov_base + base, 21024c5684bSTrond Myklebust .iov_len = vec->iov_len - base, 211b4b5cc85SChuck Lever }; 212b4b5cc85SChuck Lever 21324c5684bSTrond Myklebust if (iov.iov_len != 0) 214b4b5cc85SChuck Lever return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len); 215b4b5cc85SChuck Lever return kernel_sendmsg(sock, &msg, NULL, 0, 0); 216b4b5cc85SChuck Lever } 217b4b5cc85SChuck Lever 21824c5684bSTrond Myklebust static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more) 219b4b5cc85SChuck Lever { 22024c5684bSTrond Myklebust struct page **ppage; 22124c5684bSTrond Myklebust unsigned int remainder; 22224c5684bSTrond Myklebust int err, sent = 0; 223b4b5cc85SChuck Lever 22424c5684bSTrond Myklebust remainder = xdr->page_len - base; 22524c5684bSTrond Myklebust base += xdr->page_base; 22624c5684bSTrond Myklebust ppage = xdr->pages + (base >> PAGE_SHIFT); 22724c5684bSTrond Myklebust base &= ~PAGE_MASK; 22824c5684bSTrond Myklebust for(;;) { 22924c5684bSTrond Myklebust unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder); 23024c5684bSTrond Myklebust int flags = XS_SENDMSG_FLAGS; 23124c5684bSTrond Myklebust 23224c5684bSTrond Myklebust remainder -= len; 23324c5684bSTrond Myklebust if (remainder != 0 || more) 23424c5684bSTrond Myklebust flags |= MSG_MORE; 23524c5684bSTrond Myklebust err = sock->ops->sendpage(sock, *ppage, base, len, flags); 23624c5684bSTrond Myklebust if (remainder == 0 || err != len) 23724c5684bSTrond Myklebust break; 23824c5684bSTrond Myklebust sent += err; 23924c5684bSTrond Myklebust ppage++; 24024c5684bSTrond Myklebust base = 0; 24124c5684bSTrond Myklebust } 24224c5684bSTrond Myklebust if (sent == 0) 24324c5684bSTrond Myklebust return err; 24424c5684bSTrond Myklebust if (err > 0) 24524c5684bSTrond Myklebust sent += err; 24624c5684bSTrond Myklebust return sent; 247b4b5cc85SChuck Lever } 248b4b5cc85SChuck Lever 2499903cd1cSChuck Lever /** 2509903cd1cSChuck Lever * xs_sendpages - write pages directly to a socket 2519903cd1cSChuck Lever * @sock: socket to send on 2529903cd1cSChuck Lever * @addr: UDP only -- address of destination 2539903cd1cSChuck Lever * @addrlen: UDP only -- length of destination address 2549903cd1cSChuck Lever * @xdr: buffer containing this request 2559903cd1cSChuck Lever * @base: starting position in the buffer 2569903cd1cSChuck Lever * 257a246b010SChuck Lever */ 25824c5684bSTrond Myklebust static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base) 259a246b010SChuck Lever { 26024c5684bSTrond Myklebust unsigned int remainder = xdr->len - base; 26124c5684bSTrond Myklebust int err, sent = 0; 262a246b010SChuck Lever 263262965f5SChuck Lever if (unlikely(!sock)) 264262965f5SChuck Lever return -ENOTCONN; 265262965f5SChuck Lever 266262965f5SChuck Lever clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags); 26724c5684bSTrond Myklebust if (base != 0) { 26824c5684bSTrond Myklebust addr = NULL; 26924c5684bSTrond Myklebust addrlen = 0; 27024c5684bSTrond Myklebust } 271262965f5SChuck Lever 27224c5684bSTrond Myklebust if (base < xdr->head[0].iov_len || addr != NULL) { 27324c5684bSTrond Myklebust unsigned int len = xdr->head[0].iov_len - base; 27424c5684bSTrond Myklebust remainder -= len; 27524c5684bSTrond Myklebust err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0); 27624c5684bSTrond Myklebust if (remainder == 0 || err != len) 277a246b010SChuck Lever goto out; 27824c5684bSTrond Myklebust sent += err; 279a246b010SChuck Lever base = 0; 280a246b010SChuck Lever } else 28124c5684bSTrond Myklebust base -= xdr->head[0].iov_len; 282a246b010SChuck Lever 28324c5684bSTrond Myklebust if (base < xdr->page_len) { 28424c5684bSTrond Myklebust unsigned int len = xdr->page_len - base; 28524c5684bSTrond Myklebust remainder -= len; 28624c5684bSTrond Myklebust err = xs_send_pagedata(sock, xdr, base, remainder != 0); 28724c5684bSTrond Myklebust if (remainder == 0 || err != len) 288a246b010SChuck Lever goto out; 28924c5684bSTrond Myklebust sent += err; 290a246b010SChuck Lever base = 0; 29124c5684bSTrond Myklebust } else 29224c5684bSTrond Myklebust base -= xdr->page_len; 29324c5684bSTrond Myklebust 29424c5684bSTrond Myklebust if (base >= xdr->tail[0].iov_len) 29524c5684bSTrond Myklebust return sent; 29624c5684bSTrond Myklebust err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0); 297a246b010SChuck Lever out: 29824c5684bSTrond Myklebust if (sent == 0) 29924c5684bSTrond Myklebust return err; 30024c5684bSTrond Myklebust if (err > 0) 30124c5684bSTrond Myklebust sent += err; 30224c5684bSTrond Myklebust return sent; 303a246b010SChuck Lever } 304a246b010SChuck Lever 3059903cd1cSChuck Lever /** 306262965f5SChuck Lever * xs_nospace - place task on wait queue if transmit was incomplete 307262965f5SChuck Lever * @task: task to put to sleep 3089903cd1cSChuck Lever * 309a246b010SChuck Lever */ 310262965f5SChuck Lever static void xs_nospace(struct rpc_task *task) 311a246b010SChuck Lever { 312262965f5SChuck Lever struct rpc_rqst *req = task->tk_rqstp; 313262965f5SChuck Lever struct rpc_xprt *xprt = req->rq_xprt; 314ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 315a246b010SChuck Lever 316262965f5SChuck Lever dprintk("RPC: %4d xmit incomplete (%u left of %u)\n", 317262965f5SChuck Lever task->tk_pid, req->rq_slen - req->rq_bytes_sent, 318262965f5SChuck Lever req->rq_slen); 319a246b010SChuck Lever 320ee0ac0c2SChuck Lever if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) { 321262965f5SChuck Lever /* Protect against races with write_space */ 322262965f5SChuck Lever spin_lock_bh(&xprt->transport_lock); 323a246b010SChuck Lever 324262965f5SChuck Lever /* Don't race with disconnect */ 325262965f5SChuck Lever if (!xprt_connected(xprt)) 326262965f5SChuck Lever task->tk_status = -ENOTCONN; 327ee0ac0c2SChuck Lever else if (test_bit(SOCK_NOSPACE, &transport->sock->flags)) 328262965f5SChuck Lever xprt_wait_for_buffer_space(task); 329a246b010SChuck Lever 330262965f5SChuck Lever spin_unlock_bh(&xprt->transport_lock); 331262965f5SChuck Lever } else 332262965f5SChuck Lever /* Keep holding the socket if it is blocked */ 333262965f5SChuck Lever rpc_delay(task, HZ>>4); 334a246b010SChuck Lever } 335a246b010SChuck Lever 3369903cd1cSChuck Lever /** 337262965f5SChuck Lever * xs_udp_send_request - write an RPC request to a UDP socket 3389903cd1cSChuck Lever * @task: address of RPC task that manages the state of an RPC request 3399903cd1cSChuck Lever * 3409903cd1cSChuck Lever * Return values: 3419903cd1cSChuck Lever * 0: The request has been sent 3429903cd1cSChuck Lever * EAGAIN: The socket was blocked, please call again later to 3439903cd1cSChuck Lever * complete the request 344262965f5SChuck Lever * ENOTCONN: Caller needs to invoke connect logic then call again 3459903cd1cSChuck Lever * other: Some other error occured, the request was not sent 3469903cd1cSChuck Lever */ 347262965f5SChuck Lever static int xs_udp_send_request(struct rpc_task *task) 348a246b010SChuck Lever { 349a246b010SChuck Lever struct rpc_rqst *req = task->tk_rqstp; 350a246b010SChuck Lever struct rpc_xprt *xprt = req->rq_xprt; 351ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 352262965f5SChuck Lever struct xdr_buf *xdr = &req->rq_snd_buf; 353262965f5SChuck Lever int status; 354262965f5SChuck Lever 355262965f5SChuck Lever xs_pktdump("packet data:", 356262965f5SChuck Lever req->rq_svec->iov_base, 357262965f5SChuck Lever req->rq_svec->iov_len); 358262965f5SChuck Lever 359262965f5SChuck Lever req->rq_xtime = jiffies; 360ee0ac0c2SChuck Lever status = xs_sendpages(transport->sock, 361ee0ac0c2SChuck Lever (struct sockaddr *) &xprt->addr, 362ee0ac0c2SChuck Lever xprt->addrlen, xdr, 363ee0ac0c2SChuck Lever req->rq_bytes_sent); 364262965f5SChuck Lever 365262965f5SChuck Lever dprintk("RPC: xs_udp_send_request(%u) = %d\n", 366262965f5SChuck Lever xdr->len - req->rq_bytes_sent, status); 367262965f5SChuck Lever 368262965f5SChuck Lever if (likely(status >= (int) req->rq_slen)) 369262965f5SChuck Lever return 0; 370262965f5SChuck Lever 371262965f5SChuck Lever /* Still some bytes left; set up for a retry later. */ 372262965f5SChuck Lever if (status > 0) 373262965f5SChuck Lever status = -EAGAIN; 374262965f5SChuck Lever 375262965f5SChuck Lever switch (status) { 376262965f5SChuck Lever case -ENETUNREACH: 377262965f5SChuck Lever case -EPIPE: 378262965f5SChuck Lever case -ECONNREFUSED: 379262965f5SChuck Lever /* When the server has died, an ICMP port unreachable message 380262965f5SChuck Lever * prompts ECONNREFUSED. */ 381262965f5SChuck Lever break; 382262965f5SChuck Lever case -EAGAIN: 383262965f5SChuck Lever xs_nospace(task); 384262965f5SChuck Lever break; 385262965f5SChuck Lever default: 386262965f5SChuck Lever dprintk("RPC: sendmsg returned unrecognized error %d\n", 387262965f5SChuck Lever -status); 388262965f5SChuck Lever break; 389262965f5SChuck Lever } 390262965f5SChuck Lever 391262965f5SChuck Lever return status; 392262965f5SChuck Lever } 393262965f5SChuck Lever 394808012fbSChuck Lever static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf) 395808012fbSChuck Lever { 396808012fbSChuck Lever u32 reclen = buf->len - sizeof(rpc_fraghdr); 397808012fbSChuck Lever rpc_fraghdr *base = buf->head[0].iov_base; 398808012fbSChuck Lever *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen); 399808012fbSChuck Lever } 400808012fbSChuck Lever 401262965f5SChuck Lever /** 402262965f5SChuck Lever * xs_tcp_send_request - write an RPC request to a TCP socket 403262965f5SChuck Lever * @task: address of RPC task that manages the state of an RPC request 404262965f5SChuck Lever * 405262965f5SChuck Lever * Return values: 406262965f5SChuck Lever * 0: The request has been sent 407262965f5SChuck Lever * EAGAIN: The socket was blocked, please call again later to 408262965f5SChuck Lever * complete the request 409262965f5SChuck Lever * ENOTCONN: Caller needs to invoke connect logic then call again 410262965f5SChuck Lever * other: Some other error occured, the request was not sent 411262965f5SChuck Lever * 412262965f5SChuck Lever * XXX: In the case of soft timeouts, should we eventually give up 413262965f5SChuck Lever * if sendmsg is not able to make progress? 414262965f5SChuck Lever */ 415262965f5SChuck Lever static int xs_tcp_send_request(struct rpc_task *task) 416262965f5SChuck Lever { 417262965f5SChuck Lever struct rpc_rqst *req = task->tk_rqstp; 418262965f5SChuck Lever struct rpc_xprt *xprt = req->rq_xprt; 419ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 420262965f5SChuck Lever struct xdr_buf *xdr = &req->rq_snd_buf; 421a246b010SChuck Lever int status, retry = 0; 422a246b010SChuck Lever 423808012fbSChuck Lever xs_encode_tcp_record_marker(&req->rq_snd_buf); 424262965f5SChuck Lever 425262965f5SChuck Lever xs_pktdump("packet data:", 426262965f5SChuck Lever req->rq_svec->iov_base, 427262965f5SChuck Lever req->rq_svec->iov_len); 428a246b010SChuck Lever 429a246b010SChuck Lever /* Continue transmitting the packet/record. We must be careful 430a246b010SChuck Lever * to cope with writespace callbacks arriving _after_ we have 431262965f5SChuck Lever * called sendmsg(). */ 432a246b010SChuck Lever while (1) { 433a246b010SChuck Lever req->rq_xtime = jiffies; 434ee0ac0c2SChuck Lever status = xs_sendpages(transport->sock, 435ee0ac0c2SChuck Lever NULL, 0, xdr, req->rq_bytes_sent); 436a246b010SChuck Lever 437262965f5SChuck Lever dprintk("RPC: xs_tcp_send_request(%u) = %d\n", 438262965f5SChuck Lever xdr->len - req->rq_bytes_sent, status); 439262965f5SChuck Lever 440262965f5SChuck Lever if (unlikely(status < 0)) 441a246b010SChuck Lever break; 442a246b010SChuck Lever 443a246b010SChuck Lever /* If we've sent the entire packet, immediately 444a246b010SChuck Lever * reset the count of bytes sent. */ 445262965f5SChuck Lever req->rq_bytes_sent += status; 446ef759a2eSChuck Lever task->tk_bytes_sent += status; 447262965f5SChuck Lever if (likely(req->rq_bytes_sent >= req->rq_slen)) { 448a246b010SChuck Lever req->rq_bytes_sent = 0; 449a246b010SChuck Lever return 0; 450a246b010SChuck Lever } 451262965f5SChuck Lever 452a246b010SChuck Lever status = -EAGAIN; 453262965f5SChuck Lever if (retry++ > XS_SENDMSG_RETRY) 454a246b010SChuck Lever break; 455a246b010SChuck Lever } 456a246b010SChuck Lever 457262965f5SChuck Lever switch (status) { 458262965f5SChuck Lever case -EAGAIN: 459262965f5SChuck Lever xs_nospace(task); 460262965f5SChuck Lever break; 461262965f5SChuck Lever case -ECONNREFUSED: 462262965f5SChuck Lever case -ECONNRESET: 463262965f5SChuck Lever case -ENOTCONN: 464262965f5SChuck Lever case -EPIPE: 465262965f5SChuck Lever status = -ENOTCONN; 466262965f5SChuck Lever break; 467262965f5SChuck Lever default: 468262965f5SChuck Lever dprintk("RPC: sendmsg returned unrecognized error %d\n", 469262965f5SChuck Lever -status); 47043118c29SChuck Lever xprt_disconnect(xprt); 471a246b010SChuck Lever break; 472a246b010SChuck Lever } 473a246b010SChuck Lever 474a246b010SChuck Lever return status; 475a246b010SChuck Lever } 476a246b010SChuck Lever 4779903cd1cSChuck Lever /** 478e0ab53deSTrond Myklebust * xs_tcp_release_xprt - clean up after a tcp transmission 479e0ab53deSTrond Myklebust * @xprt: transport 480e0ab53deSTrond Myklebust * @task: rpc task 481e0ab53deSTrond Myklebust * 482e0ab53deSTrond Myklebust * This cleans up if an error causes us to abort the transmission of a request. 483e0ab53deSTrond Myklebust * In this case, the socket may need to be reset in order to avoid confusing 484e0ab53deSTrond Myklebust * the server. 485e0ab53deSTrond Myklebust */ 486e0ab53deSTrond Myklebust static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 487e0ab53deSTrond Myklebust { 488e0ab53deSTrond Myklebust struct rpc_rqst *req; 489e0ab53deSTrond Myklebust 490e0ab53deSTrond Myklebust if (task != xprt->snd_task) 491e0ab53deSTrond Myklebust return; 492e0ab53deSTrond Myklebust if (task == NULL) 493e0ab53deSTrond Myklebust goto out_release; 494e0ab53deSTrond Myklebust req = task->tk_rqstp; 495e0ab53deSTrond Myklebust if (req->rq_bytes_sent == 0) 496e0ab53deSTrond Myklebust goto out_release; 497e0ab53deSTrond Myklebust if (req->rq_bytes_sent == req->rq_snd_buf.len) 498e0ab53deSTrond Myklebust goto out_release; 499e0ab53deSTrond Myklebust set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state); 500e0ab53deSTrond Myklebust out_release: 501e0ab53deSTrond Myklebust xprt_release_xprt(xprt, task); 502e0ab53deSTrond Myklebust } 503e0ab53deSTrond Myklebust 504e0ab53deSTrond Myklebust /** 5059903cd1cSChuck Lever * xs_close - close a socket 5069903cd1cSChuck Lever * @xprt: transport 5079903cd1cSChuck Lever * 5083167e12cSChuck Lever * This is used when all requests are complete; ie, no DRC state remains 5093167e12cSChuck Lever * on the server we want to save. 510a246b010SChuck Lever */ 5119903cd1cSChuck Lever static void xs_close(struct rpc_xprt *xprt) 512a246b010SChuck Lever { 513ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 514ee0ac0c2SChuck Lever struct socket *sock = transport->sock; 515ee0ac0c2SChuck Lever struct sock *sk = transport->inet; 516a246b010SChuck Lever 517a246b010SChuck Lever if (!sk) 518632e3bdcSTrond Myklebust goto clear_close_wait; 519a246b010SChuck Lever 5209903cd1cSChuck Lever dprintk("RPC: xs_close xprt %p\n", xprt); 5219903cd1cSChuck Lever 522a246b010SChuck Lever write_lock_bh(&sk->sk_callback_lock); 523ee0ac0c2SChuck Lever transport->inet = NULL; 524ee0ac0c2SChuck Lever transport->sock = NULL; 525a246b010SChuck Lever 526a246b010SChuck Lever sk->sk_user_data = NULL; 527a246b010SChuck Lever sk->sk_data_ready = xprt->old_data_ready; 528a246b010SChuck Lever sk->sk_state_change = xprt->old_state_change; 529a246b010SChuck Lever sk->sk_write_space = xprt->old_write_space; 530a246b010SChuck Lever write_unlock_bh(&sk->sk_callback_lock); 531a246b010SChuck Lever 532a246b010SChuck Lever sk->sk_no_check = 0; 533a246b010SChuck Lever 534a246b010SChuck Lever sock_release(sock); 535632e3bdcSTrond Myklebust clear_close_wait: 536632e3bdcSTrond Myklebust smp_mb__before_clear_bit(); 537632e3bdcSTrond Myklebust clear_bit(XPRT_CLOSE_WAIT, &xprt->state); 538632e3bdcSTrond Myklebust smp_mb__after_clear_bit(); 539a246b010SChuck Lever } 540a246b010SChuck Lever 5419903cd1cSChuck Lever /** 5429903cd1cSChuck Lever * xs_destroy - prepare to shutdown a transport 5439903cd1cSChuck Lever * @xprt: doomed transport 5449903cd1cSChuck Lever * 5459903cd1cSChuck Lever */ 5469903cd1cSChuck Lever static void xs_destroy(struct rpc_xprt *xprt) 547a246b010SChuck Lever { 5489903cd1cSChuck Lever dprintk("RPC: xs_destroy xprt %p\n", xprt); 5499903cd1cSChuck Lever 55055aa4f58SChuck Lever cancel_delayed_work(&xprt->connect_worker); 551a246b010SChuck Lever flush_scheduled_work(); 552a246b010SChuck Lever 553a246b010SChuck Lever xprt_disconnect(xprt); 5549903cd1cSChuck Lever xs_close(xprt); 555edb267a6SChuck Lever xs_free_peer_addresses(xprt); 556a246b010SChuck Lever kfree(xprt->slot); 557c8541ecdSChuck Lever kfree(xprt); 558a246b010SChuck Lever } 559a246b010SChuck Lever 5609903cd1cSChuck Lever static inline struct rpc_xprt *xprt_from_sock(struct sock *sk) 5619903cd1cSChuck Lever { 5629903cd1cSChuck Lever return (struct rpc_xprt *) sk->sk_user_data; 5639903cd1cSChuck Lever } 5649903cd1cSChuck Lever 5659903cd1cSChuck Lever /** 5669903cd1cSChuck Lever * xs_udp_data_ready - "data ready" callback for UDP sockets 5679903cd1cSChuck Lever * @sk: socket with data to read 5689903cd1cSChuck Lever * @len: how much data to read 5699903cd1cSChuck Lever * 570a246b010SChuck Lever */ 5719903cd1cSChuck Lever static void xs_udp_data_ready(struct sock *sk, int len) 572a246b010SChuck Lever { 573a246b010SChuck Lever struct rpc_task *task; 574a246b010SChuck Lever struct rpc_xprt *xprt; 575a246b010SChuck Lever struct rpc_rqst *rovr; 576a246b010SChuck Lever struct sk_buff *skb; 577a246b010SChuck Lever int err, repsize, copied; 578d8ed029dSAlexey Dobriyan u32 _xid; 579d8ed029dSAlexey Dobriyan __be32 *xp; 580a246b010SChuck Lever 581a246b010SChuck Lever read_lock(&sk->sk_callback_lock); 5829903cd1cSChuck Lever dprintk("RPC: xs_udp_data_ready...\n"); 5839903cd1cSChuck Lever if (!(xprt = xprt_from_sock(sk))) 584a246b010SChuck Lever goto out; 585a246b010SChuck Lever 586a246b010SChuck Lever if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL) 587a246b010SChuck Lever goto out; 588a246b010SChuck Lever 589a246b010SChuck Lever if (xprt->shutdown) 590a246b010SChuck Lever goto dropit; 591a246b010SChuck Lever 592a246b010SChuck Lever repsize = skb->len - sizeof(struct udphdr); 593a246b010SChuck Lever if (repsize < 4) { 5949903cd1cSChuck Lever dprintk("RPC: impossible RPC reply size %d!\n", repsize); 595a246b010SChuck Lever goto dropit; 596a246b010SChuck Lever } 597a246b010SChuck Lever 598a246b010SChuck Lever /* Copy the XID from the skb... */ 599a246b010SChuck Lever xp = skb_header_pointer(skb, sizeof(struct udphdr), 600a246b010SChuck Lever sizeof(_xid), &_xid); 601a246b010SChuck Lever if (xp == NULL) 602a246b010SChuck Lever goto dropit; 603a246b010SChuck Lever 604a246b010SChuck Lever /* Look up and lock the request corresponding to the given XID */ 6054a0f8c04SChuck Lever spin_lock(&xprt->transport_lock); 606a246b010SChuck Lever rovr = xprt_lookup_rqst(xprt, *xp); 607a246b010SChuck Lever if (!rovr) 608a246b010SChuck Lever goto out_unlock; 609a246b010SChuck Lever task = rovr->rq_task; 610a246b010SChuck Lever 611a246b010SChuck Lever if ((copied = rovr->rq_private_buf.buflen) > repsize) 612a246b010SChuck Lever copied = repsize; 613a246b010SChuck Lever 614a246b010SChuck Lever /* Suck it into the iovec, verify checksum if not done by hw. */ 615a246b010SChuck Lever if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) 616a246b010SChuck Lever goto out_unlock; 617a246b010SChuck Lever 618a246b010SChuck Lever /* Something worked... */ 619a246b010SChuck Lever dst_confirm(skb->dst); 620a246b010SChuck Lever 6211570c1e4SChuck Lever xprt_adjust_cwnd(task, copied); 6221570c1e4SChuck Lever xprt_update_rtt(task); 6231570c1e4SChuck Lever xprt_complete_rqst(task, copied); 624a246b010SChuck Lever 625a246b010SChuck Lever out_unlock: 6264a0f8c04SChuck Lever spin_unlock(&xprt->transport_lock); 627a246b010SChuck Lever dropit: 628a246b010SChuck Lever skb_free_datagram(sk, skb); 629a246b010SChuck Lever out: 630a246b010SChuck Lever read_unlock(&sk->sk_callback_lock); 631a246b010SChuck Lever } 632a246b010SChuck Lever 6339903cd1cSChuck Lever static inline size_t xs_tcp_copy_data(skb_reader_t *desc, void *p, size_t len) 634a246b010SChuck Lever { 635a246b010SChuck Lever if (len > desc->count) 636a246b010SChuck Lever len = desc->count; 637a246b010SChuck Lever if (skb_copy_bits(desc->skb, desc->offset, p, len)) { 638a246b010SChuck Lever dprintk("RPC: failed to copy %zu bytes from skb. %zu bytes remain\n", 639a246b010SChuck Lever len, desc->count); 640a246b010SChuck Lever return 0; 641a246b010SChuck Lever } 642a246b010SChuck Lever desc->offset += len; 643a246b010SChuck Lever desc->count -= len; 644a246b010SChuck Lever dprintk("RPC: copied %zu bytes from skb. %zu bytes remain\n", 645a246b010SChuck Lever len, desc->count); 646a246b010SChuck Lever return len; 647a246b010SChuck Lever } 648a246b010SChuck Lever 6499903cd1cSChuck Lever static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc) 650a246b010SChuck Lever { 65151971139SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 652a246b010SChuck Lever size_t len, used; 653a246b010SChuck Lever char *p; 654a246b010SChuck Lever 65551971139SChuck Lever p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset; 65651971139SChuck Lever len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset; 6579903cd1cSChuck Lever used = xs_tcp_copy_data(desc, p, len); 65851971139SChuck Lever transport->tcp_offset += used; 659a246b010SChuck Lever if (used != len) 660a246b010SChuck Lever return; 661808012fbSChuck Lever 66251971139SChuck Lever transport->tcp_reclen = ntohl(transport->tcp_fraghdr); 66351971139SChuck Lever if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT) 664e136d092SChuck Lever transport->tcp_flags |= TCP_RCV_LAST_FRAG; 665a246b010SChuck Lever else 666e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_LAST_FRAG; 66751971139SChuck Lever transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK; 668808012fbSChuck Lever 669e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR; 67051971139SChuck Lever transport->tcp_offset = 0; 671808012fbSChuck Lever 672a246b010SChuck Lever /* Sanity check of the record length */ 67351971139SChuck Lever if (unlikely(transport->tcp_reclen < 4)) { 6749903cd1cSChuck Lever dprintk("RPC: invalid TCP record fragment length\n"); 675a246b010SChuck Lever xprt_disconnect(xprt); 6769903cd1cSChuck Lever return; 677a246b010SChuck Lever } 678a246b010SChuck Lever dprintk("RPC: reading TCP record fragment of length %d\n", 67951971139SChuck Lever transport->tcp_reclen); 680a246b010SChuck Lever } 681a246b010SChuck Lever 68251971139SChuck Lever static void xs_tcp_check_fraghdr(struct sock_xprt *transport) 683a246b010SChuck Lever { 68451971139SChuck Lever if (transport->tcp_offset == transport->tcp_reclen) { 685e136d092SChuck Lever transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR; 68651971139SChuck Lever transport->tcp_offset = 0; 687e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_LAST_FRAG) { 688e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 689e136d092SChuck Lever transport->tcp_flags |= TCP_RCV_COPY_XID; 69051971139SChuck Lever transport->tcp_copied = 0; 691a246b010SChuck Lever } 692a246b010SChuck Lever } 693a246b010SChuck Lever } 694a246b010SChuck Lever 69551971139SChuck Lever static inline void xs_tcp_read_xid(struct sock_xprt *transport, skb_reader_t *desc) 696a246b010SChuck Lever { 697a246b010SChuck Lever size_t len, used; 698a246b010SChuck Lever char *p; 699a246b010SChuck Lever 70051971139SChuck Lever len = sizeof(transport->tcp_xid) - transport->tcp_offset; 701a246b010SChuck Lever dprintk("RPC: reading XID (%Zu bytes)\n", len); 70251971139SChuck Lever p = ((char *) &transport->tcp_xid) + transport->tcp_offset; 7039903cd1cSChuck Lever used = xs_tcp_copy_data(desc, p, len); 70451971139SChuck Lever transport->tcp_offset += used; 705a246b010SChuck Lever if (used != len) 706a246b010SChuck Lever return; 707e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_XID; 708e136d092SChuck Lever transport->tcp_flags |= TCP_RCV_COPY_DATA; 70951971139SChuck Lever transport->tcp_copied = 4; 710a246b010SChuck Lever dprintk("RPC: reading reply for XID %08x\n", 71151971139SChuck Lever ntohl(transport->tcp_xid)); 71251971139SChuck Lever xs_tcp_check_fraghdr(transport); 713a246b010SChuck Lever } 714a246b010SChuck Lever 7159903cd1cSChuck Lever static inline void xs_tcp_read_request(struct rpc_xprt *xprt, skb_reader_t *desc) 716a246b010SChuck Lever { 71751971139SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 718a246b010SChuck Lever struct rpc_rqst *req; 719a246b010SChuck Lever struct xdr_buf *rcvbuf; 720a246b010SChuck Lever size_t len; 721a246b010SChuck Lever ssize_t r; 722a246b010SChuck Lever 723a246b010SChuck Lever /* Find and lock the request corresponding to this xid */ 7244a0f8c04SChuck Lever spin_lock(&xprt->transport_lock); 72551971139SChuck Lever req = xprt_lookup_rqst(xprt, transport->tcp_xid); 726a246b010SChuck Lever if (!req) { 727e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 728a246b010SChuck Lever dprintk("RPC: XID %08x request not found!\n", 72951971139SChuck Lever ntohl(transport->tcp_xid)); 7304a0f8c04SChuck Lever spin_unlock(&xprt->transport_lock); 731a246b010SChuck Lever return; 732a246b010SChuck Lever } 733a246b010SChuck Lever 734a246b010SChuck Lever rcvbuf = &req->rq_private_buf; 735a246b010SChuck Lever len = desc->count; 73651971139SChuck Lever if (len > transport->tcp_reclen - transport->tcp_offset) { 737a246b010SChuck Lever skb_reader_t my_desc; 738a246b010SChuck Lever 73951971139SChuck Lever len = transport->tcp_reclen - transport->tcp_offset; 740a246b010SChuck Lever memcpy(&my_desc, desc, sizeof(my_desc)); 741a246b010SChuck Lever my_desc.count = len; 74251971139SChuck Lever r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied, 7439903cd1cSChuck Lever &my_desc, xs_tcp_copy_data); 744a246b010SChuck Lever desc->count -= r; 745a246b010SChuck Lever desc->offset += r; 746a246b010SChuck Lever } else 74751971139SChuck Lever r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied, 7489903cd1cSChuck Lever desc, xs_tcp_copy_data); 749a246b010SChuck Lever 750a246b010SChuck Lever if (r > 0) { 75151971139SChuck Lever transport->tcp_copied += r; 75251971139SChuck Lever transport->tcp_offset += r; 753a246b010SChuck Lever } 754a246b010SChuck Lever if (r != len) { 755a246b010SChuck Lever /* Error when copying to the receive buffer, 756a246b010SChuck Lever * usually because we weren't able to allocate 757a246b010SChuck Lever * additional buffer pages. All we can do now 758e136d092SChuck Lever * is turn off TCP_RCV_COPY_DATA, so the request 759a246b010SChuck Lever * will not receive any additional updates, 760a246b010SChuck Lever * and time out. 761a246b010SChuck Lever * Any remaining data from this record will 762a246b010SChuck Lever * be discarded. 763a246b010SChuck Lever */ 764e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 765a246b010SChuck Lever dprintk("RPC: XID %08x truncated request\n", 76651971139SChuck Lever ntohl(transport->tcp_xid)); 767a246b010SChuck Lever dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n", 76851971139SChuck Lever xprt, transport->tcp_copied, transport->tcp_offset, 76951971139SChuck Lever transport->tcp_reclen); 770a246b010SChuck Lever goto out; 771a246b010SChuck Lever } 772a246b010SChuck Lever 773a246b010SChuck Lever dprintk("RPC: XID %08x read %Zd bytes\n", 77451971139SChuck Lever ntohl(transport->tcp_xid), r); 775a246b010SChuck Lever dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n", 77651971139SChuck Lever xprt, transport->tcp_copied, transport->tcp_offset, 77751971139SChuck Lever transport->tcp_reclen); 778a246b010SChuck Lever 77951971139SChuck Lever if (transport->tcp_copied == req->rq_private_buf.buflen) 780e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 78151971139SChuck Lever else if (transport->tcp_offset == transport->tcp_reclen) { 782e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_LAST_FRAG) 783e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 784a246b010SChuck Lever } 785a246b010SChuck Lever 786a246b010SChuck Lever out: 787e136d092SChuck Lever if (!(transport->tcp_flags & TCP_RCV_COPY_DATA)) 78851971139SChuck Lever xprt_complete_rqst(req->rq_task, transport->tcp_copied); 7894a0f8c04SChuck Lever spin_unlock(&xprt->transport_lock); 79051971139SChuck Lever xs_tcp_check_fraghdr(transport); 791a246b010SChuck Lever } 792a246b010SChuck Lever 79351971139SChuck Lever static inline void xs_tcp_read_discard(struct sock_xprt *transport, skb_reader_t *desc) 794a246b010SChuck Lever { 795a246b010SChuck Lever size_t len; 796a246b010SChuck Lever 79751971139SChuck Lever len = transport->tcp_reclen - transport->tcp_offset; 798a246b010SChuck Lever if (len > desc->count) 799a246b010SChuck Lever len = desc->count; 800a246b010SChuck Lever desc->count -= len; 801a246b010SChuck Lever desc->offset += len; 80251971139SChuck Lever transport->tcp_offset += len; 803a246b010SChuck Lever dprintk("RPC: discarded %Zu bytes\n", len); 80451971139SChuck Lever xs_tcp_check_fraghdr(transport); 805a246b010SChuck Lever } 806a246b010SChuck Lever 8079903cd1cSChuck Lever static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len) 808a246b010SChuck Lever { 809a246b010SChuck Lever struct rpc_xprt *xprt = rd_desc->arg.data; 81051971139SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 811a246b010SChuck Lever skb_reader_t desc = { 812a246b010SChuck Lever .skb = skb, 813a246b010SChuck Lever .offset = offset, 814a246b010SChuck Lever .count = len, 815a246b010SChuck Lever }; 816a246b010SChuck Lever 8179903cd1cSChuck Lever dprintk("RPC: xs_tcp_data_recv started\n"); 818a246b010SChuck Lever do { 819a246b010SChuck Lever /* Read in a new fragment marker if necessary */ 820a246b010SChuck Lever /* Can we ever really expect to get completely empty fragments? */ 821e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) { 8229903cd1cSChuck Lever xs_tcp_read_fraghdr(xprt, &desc); 823a246b010SChuck Lever continue; 824a246b010SChuck Lever } 825a246b010SChuck Lever /* Read in the xid if necessary */ 826e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_COPY_XID) { 82751971139SChuck Lever xs_tcp_read_xid(transport, &desc); 828a246b010SChuck Lever continue; 829a246b010SChuck Lever } 830a246b010SChuck Lever /* Read in the request data */ 831e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_COPY_DATA) { 8329903cd1cSChuck Lever xs_tcp_read_request(xprt, &desc); 833a246b010SChuck Lever continue; 834a246b010SChuck Lever } 835a246b010SChuck Lever /* Skip over any trailing bytes on short reads */ 83651971139SChuck Lever xs_tcp_read_discard(transport, &desc); 837a246b010SChuck Lever } while (desc.count); 8389903cd1cSChuck Lever dprintk("RPC: xs_tcp_data_recv done\n"); 839a246b010SChuck Lever return len - desc.count; 840a246b010SChuck Lever } 841a246b010SChuck Lever 8429903cd1cSChuck Lever /** 8439903cd1cSChuck Lever * xs_tcp_data_ready - "data ready" callback for TCP sockets 8449903cd1cSChuck Lever * @sk: socket with data to read 8459903cd1cSChuck Lever * @bytes: how much data to read 8469903cd1cSChuck Lever * 8479903cd1cSChuck Lever */ 8489903cd1cSChuck Lever static void xs_tcp_data_ready(struct sock *sk, int bytes) 849a246b010SChuck Lever { 850a246b010SChuck Lever struct rpc_xprt *xprt; 851a246b010SChuck Lever read_descriptor_t rd_desc; 852a246b010SChuck Lever 853a246b010SChuck Lever read_lock(&sk->sk_callback_lock); 8549903cd1cSChuck Lever dprintk("RPC: xs_tcp_data_ready...\n"); 8559903cd1cSChuck Lever if (!(xprt = xprt_from_sock(sk))) 856a246b010SChuck Lever goto out; 857a246b010SChuck Lever if (xprt->shutdown) 858a246b010SChuck Lever goto out; 859a246b010SChuck Lever 8609903cd1cSChuck Lever /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */ 861a246b010SChuck Lever rd_desc.arg.data = xprt; 862a246b010SChuck Lever rd_desc.count = 65536; 8639903cd1cSChuck Lever tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv); 864a246b010SChuck Lever out: 865a246b010SChuck Lever read_unlock(&sk->sk_callback_lock); 866a246b010SChuck Lever } 867a246b010SChuck Lever 8689903cd1cSChuck Lever /** 8699903cd1cSChuck Lever * xs_tcp_state_change - callback to handle TCP socket state changes 8709903cd1cSChuck Lever * @sk: socket whose state has changed 8719903cd1cSChuck Lever * 8729903cd1cSChuck Lever */ 8739903cd1cSChuck Lever static void xs_tcp_state_change(struct sock *sk) 874a246b010SChuck Lever { 875a246b010SChuck Lever struct rpc_xprt *xprt; 876a246b010SChuck Lever 877a246b010SChuck Lever read_lock(&sk->sk_callback_lock); 878a246b010SChuck Lever if (!(xprt = xprt_from_sock(sk))) 879a246b010SChuck Lever goto out; 8809903cd1cSChuck Lever dprintk("RPC: xs_tcp_state_change client %p...\n", xprt); 881a246b010SChuck Lever dprintk("RPC: state %x conn %d dead %d zapped %d\n", 882a246b010SChuck Lever sk->sk_state, xprt_connected(xprt), 883a246b010SChuck Lever sock_flag(sk, SOCK_DEAD), 884a246b010SChuck Lever sock_flag(sk, SOCK_ZAPPED)); 885a246b010SChuck Lever 886a246b010SChuck Lever switch (sk->sk_state) { 887a246b010SChuck Lever case TCP_ESTABLISHED: 8884a0f8c04SChuck Lever spin_lock_bh(&xprt->transport_lock); 889a246b010SChuck Lever if (!xprt_test_and_set_connected(xprt)) { 89051971139SChuck Lever struct sock_xprt *transport = container_of(xprt, 89151971139SChuck Lever struct sock_xprt, xprt); 89251971139SChuck Lever 893a246b010SChuck Lever /* Reset TCP record info */ 89451971139SChuck Lever transport->tcp_offset = 0; 89551971139SChuck Lever transport->tcp_reclen = 0; 89651971139SChuck Lever transport->tcp_copied = 0; 897e136d092SChuck Lever transport->tcp_flags = 898e136d092SChuck Lever TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID; 89951971139SChuck Lever 90003bf4b70SChuck Lever xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO; 90144fbac22SChuck Lever xprt_wake_pending_tasks(xprt, 0); 902a246b010SChuck Lever } 9034a0f8c04SChuck Lever spin_unlock_bh(&xprt->transport_lock); 904a246b010SChuck Lever break; 905a246b010SChuck Lever case TCP_SYN_SENT: 906a246b010SChuck Lever case TCP_SYN_RECV: 907a246b010SChuck Lever break; 908632e3bdcSTrond Myklebust case TCP_CLOSE_WAIT: 909632e3bdcSTrond Myklebust /* Try to schedule an autoclose RPC calls */ 910632e3bdcSTrond Myklebust set_bit(XPRT_CLOSE_WAIT, &xprt->state); 911632e3bdcSTrond Myklebust if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0) 912632e3bdcSTrond Myklebust schedule_work(&xprt->task_cleanup); 913a246b010SChuck Lever default: 914a246b010SChuck Lever xprt_disconnect(xprt); 915a246b010SChuck Lever } 916a246b010SChuck Lever out: 917a246b010SChuck Lever read_unlock(&sk->sk_callback_lock); 918a246b010SChuck Lever } 919a246b010SChuck Lever 9209903cd1cSChuck Lever /** 921c7b2cae8SChuck Lever * xs_udp_write_space - callback invoked when socket buffer space 922c7b2cae8SChuck Lever * becomes available 9239903cd1cSChuck Lever * @sk: socket whose state has changed 9249903cd1cSChuck Lever * 925a246b010SChuck Lever * Called when more output buffer space is available for this socket. 926a246b010SChuck Lever * We try not to wake our writers until they can make "significant" 927c7b2cae8SChuck Lever * progress, otherwise we'll waste resources thrashing kernel_sendmsg 928a246b010SChuck Lever * with a bunch of small requests. 929a246b010SChuck Lever */ 930c7b2cae8SChuck Lever static void xs_udp_write_space(struct sock *sk) 931a246b010SChuck Lever { 932a246b010SChuck Lever read_lock(&sk->sk_callback_lock); 933c7b2cae8SChuck Lever 934c7b2cae8SChuck Lever /* from net/core/sock.c:sock_def_write_space */ 935c7b2cae8SChuck Lever if (sock_writeable(sk)) { 936c7b2cae8SChuck Lever struct socket *sock; 937c7b2cae8SChuck Lever struct rpc_xprt *xprt; 938c7b2cae8SChuck Lever 939c7b2cae8SChuck Lever if (unlikely(!(sock = sk->sk_socket))) 940a246b010SChuck Lever goto out; 941c7b2cae8SChuck Lever if (unlikely(!(xprt = xprt_from_sock(sk)))) 942c7b2cae8SChuck Lever goto out; 943c7b2cae8SChuck Lever if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags))) 944a246b010SChuck Lever goto out; 945a246b010SChuck Lever 946c7b2cae8SChuck Lever xprt_write_space(xprt); 947a246b010SChuck Lever } 948a246b010SChuck Lever 949c7b2cae8SChuck Lever out: 950c7b2cae8SChuck Lever read_unlock(&sk->sk_callback_lock); 951c7b2cae8SChuck Lever } 952c7b2cae8SChuck Lever 953c7b2cae8SChuck Lever /** 954c7b2cae8SChuck Lever * xs_tcp_write_space - callback invoked when socket buffer space 955c7b2cae8SChuck Lever * becomes available 956c7b2cae8SChuck Lever * @sk: socket whose state has changed 957c7b2cae8SChuck Lever * 958c7b2cae8SChuck Lever * Called when more output buffer space is available for this socket. 959c7b2cae8SChuck Lever * We try not to wake our writers until they can make "significant" 960c7b2cae8SChuck Lever * progress, otherwise we'll waste resources thrashing kernel_sendmsg 961c7b2cae8SChuck Lever * with a bunch of small requests. 962c7b2cae8SChuck Lever */ 963c7b2cae8SChuck Lever static void xs_tcp_write_space(struct sock *sk) 964c7b2cae8SChuck Lever { 965c7b2cae8SChuck Lever read_lock(&sk->sk_callback_lock); 966c7b2cae8SChuck Lever 967c7b2cae8SChuck Lever /* from net/core/stream.c:sk_stream_write_space */ 968c7b2cae8SChuck Lever if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) { 969c7b2cae8SChuck Lever struct socket *sock; 970c7b2cae8SChuck Lever struct rpc_xprt *xprt; 971c7b2cae8SChuck Lever 972c7b2cae8SChuck Lever if (unlikely(!(sock = sk->sk_socket))) 973c7b2cae8SChuck Lever goto out; 974c7b2cae8SChuck Lever if (unlikely(!(xprt = xprt_from_sock(sk)))) 975c7b2cae8SChuck Lever goto out; 976c7b2cae8SChuck Lever if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags))) 977a246b010SChuck Lever goto out; 978a246b010SChuck Lever 979c7b2cae8SChuck Lever xprt_write_space(xprt); 980c7b2cae8SChuck Lever } 981c7b2cae8SChuck Lever 982a246b010SChuck Lever out: 983a246b010SChuck Lever read_unlock(&sk->sk_callback_lock); 984a246b010SChuck Lever } 985a246b010SChuck Lever 986470056c2SChuck Lever static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt) 987a246b010SChuck Lever { 988ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 989ee0ac0c2SChuck Lever struct sock *sk = transport->inet; 990a246b010SChuck Lever 991a246b010SChuck Lever if (xprt->rcvsize) { 992a246b010SChuck Lever sk->sk_userlocks |= SOCK_RCVBUF_LOCK; 993a246b010SChuck Lever sk->sk_rcvbuf = xprt->rcvsize * xprt->max_reqs * 2; 994a246b010SChuck Lever } 995a246b010SChuck Lever if (xprt->sndsize) { 996a246b010SChuck Lever sk->sk_userlocks |= SOCK_SNDBUF_LOCK; 997a246b010SChuck Lever sk->sk_sndbuf = xprt->sndsize * xprt->max_reqs * 2; 998a246b010SChuck Lever sk->sk_write_space(sk); 999a246b010SChuck Lever } 1000a246b010SChuck Lever } 1001a246b010SChuck Lever 100243118c29SChuck Lever /** 1003470056c2SChuck Lever * xs_udp_set_buffer_size - set send and receive limits 100443118c29SChuck Lever * @xprt: generic transport 1005470056c2SChuck Lever * @sndsize: requested size of send buffer, in bytes 1006470056c2SChuck Lever * @rcvsize: requested size of receive buffer, in bytes 100743118c29SChuck Lever * 1008470056c2SChuck Lever * Set socket send and receive buffer size limits. 100943118c29SChuck Lever */ 1010470056c2SChuck Lever static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize) 101143118c29SChuck Lever { 1012470056c2SChuck Lever xprt->sndsize = 0; 1013470056c2SChuck Lever if (sndsize) 1014470056c2SChuck Lever xprt->sndsize = sndsize + 1024; 1015470056c2SChuck Lever xprt->rcvsize = 0; 1016470056c2SChuck Lever if (rcvsize) 1017470056c2SChuck Lever xprt->rcvsize = rcvsize + 1024; 1018470056c2SChuck Lever 1019470056c2SChuck Lever xs_udp_do_set_buffer_size(xprt); 102043118c29SChuck Lever } 102143118c29SChuck Lever 102246c0ee8bSChuck Lever /** 102346c0ee8bSChuck Lever * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport 102446c0ee8bSChuck Lever * @task: task that timed out 102546c0ee8bSChuck Lever * 102646c0ee8bSChuck Lever * Adjust the congestion window after a retransmit timeout has occurred. 102746c0ee8bSChuck Lever */ 102846c0ee8bSChuck Lever static void xs_udp_timer(struct rpc_task *task) 102946c0ee8bSChuck Lever { 103046c0ee8bSChuck Lever xprt_adjust_cwnd(task, -ETIMEDOUT); 103146c0ee8bSChuck Lever } 103246c0ee8bSChuck Lever 1033b85d8806SChuck Lever static unsigned short xs_get_random_port(void) 1034b85d8806SChuck Lever { 1035b85d8806SChuck Lever unsigned short range = xprt_max_resvport - xprt_min_resvport; 1036b85d8806SChuck Lever unsigned short rand = (unsigned short) net_random() % range; 1037b85d8806SChuck Lever return rand + xprt_min_resvport; 1038b85d8806SChuck Lever } 1039b85d8806SChuck Lever 104092200412SChuck Lever /** 1041edb267a6SChuck Lever * xs_print_peer_address - format an IPv4 address for printing 1042edb267a6SChuck Lever * @xprt: generic transport 1043edb267a6SChuck Lever * @format: flags field indicating which parts of the address to render 1044edb267a6SChuck Lever */ 1045edb267a6SChuck Lever static char *xs_print_peer_address(struct rpc_xprt *xprt, enum rpc_display_format_t format) 1046edb267a6SChuck Lever { 1047edb267a6SChuck Lever if (xprt->address_strings[format] != NULL) 1048edb267a6SChuck Lever return xprt->address_strings[format]; 1049edb267a6SChuck Lever else 1050edb267a6SChuck Lever return "unprintable"; 1051edb267a6SChuck Lever } 1052edb267a6SChuck Lever 1053edb267a6SChuck Lever /** 105492200412SChuck Lever * xs_set_port - reset the port number in the remote endpoint address 105592200412SChuck Lever * @xprt: generic transport 105692200412SChuck Lever * @port: new port number 105792200412SChuck Lever * 105892200412SChuck Lever */ 105992200412SChuck Lever static void xs_set_port(struct rpc_xprt *xprt, unsigned short port) 106092200412SChuck Lever { 1061c4efcb1dSChuck Lever struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr; 1062c4efcb1dSChuck Lever 106392200412SChuck Lever dprintk("RPC: setting port for xprt %p to %u\n", xprt, port); 1064c4efcb1dSChuck Lever 1065c4efcb1dSChuck Lever sap->sin_port = htons(port); 106692200412SChuck Lever } 106792200412SChuck Lever 10689903cd1cSChuck Lever static int xs_bindresvport(struct rpc_xprt *xprt, struct socket *sock) 1069a246b010SChuck Lever { 1070a246b010SChuck Lever struct sockaddr_in myaddr = { 1071a246b010SChuck Lever .sin_family = AF_INET, 1072a246b010SChuck Lever }; 1073529b33c6SChuck Lever int err; 1074529b33c6SChuck Lever unsigned short port = xprt->port; 1075a246b010SChuck Lever 1076a246b010SChuck Lever do { 1077a246b010SChuck Lever myaddr.sin_port = htons(port); 1078e6242e92SSridhar Samudrala err = kernel_bind(sock, (struct sockaddr *) &myaddr, 1079a246b010SChuck Lever sizeof(myaddr)); 1080a246b010SChuck Lever if (err == 0) { 1081a246b010SChuck Lever xprt->port = port; 10829903cd1cSChuck Lever dprintk("RPC: xs_bindresvport bound to port %u\n", 10839903cd1cSChuck Lever port); 1084a246b010SChuck Lever return 0; 1085a246b010SChuck Lever } 1086529b33c6SChuck Lever if (port <= xprt_min_resvport) 1087529b33c6SChuck Lever port = xprt_max_resvport; 1088529b33c6SChuck Lever else 1089529b33c6SChuck Lever port--; 1090a246b010SChuck Lever } while (err == -EADDRINUSE && port != xprt->port); 1091a246b010SChuck Lever 10929903cd1cSChuck Lever dprintk("RPC: can't bind to reserved port (%d).\n", -err); 1093a246b010SChuck Lever return err; 1094a246b010SChuck Lever } 1095a246b010SChuck Lever 10969903cd1cSChuck Lever /** 1097b0d93ad5SChuck Lever * xs_udp_connect_worker - set up a UDP socket 10989903cd1cSChuck Lever * @args: RPC transport to connect 10999903cd1cSChuck Lever * 11009903cd1cSChuck Lever * Invoked by a work queue tasklet. 1101a246b010SChuck Lever */ 1102b0d93ad5SChuck Lever static void xs_udp_connect_worker(void *args) 1103a246b010SChuck Lever { 1104a246b010SChuck Lever struct rpc_xprt *xprt = (struct rpc_xprt *) args; 1105ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1106ee0ac0c2SChuck Lever struct socket *sock = transport->sock; 1107b0d93ad5SChuck Lever int err, status = -EIO; 1108a246b010SChuck Lever 1109ec739ef0SChuck Lever if (xprt->shutdown || !xprt_bound(xprt)) 1110a246b010SChuck Lever goto out; 1111a246b010SChuck Lever 1112b0d93ad5SChuck Lever /* Start by resetting any existing state */ 11139903cd1cSChuck Lever xs_close(xprt); 1114b0d93ad5SChuck Lever 1115b0d93ad5SChuck Lever if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) { 1116b0d93ad5SChuck Lever dprintk("RPC: can't create UDP transport socket (%d).\n", -err); 1117a246b010SChuck Lever goto out; 1118a246b010SChuck Lever } 1119a246b010SChuck Lever 1120b0d93ad5SChuck Lever if (xprt->resvport && xs_bindresvport(xprt, sock) < 0) { 1121b0d93ad5SChuck Lever sock_release(sock); 1122b0d93ad5SChuck Lever goto out; 1123b0d93ad5SChuck Lever } 1124b0d93ad5SChuck Lever 1125edb267a6SChuck Lever dprintk("RPC: worker connecting xprt %p to address: %s\n", 1126edb267a6SChuck Lever xprt, xs_print_peer_address(xprt, RPC_DISPLAY_ALL)); 1127edb267a6SChuck Lever 1128ee0ac0c2SChuck Lever if (!transport->inet) { 1129b0d93ad5SChuck Lever struct sock *sk = sock->sk; 1130b0d93ad5SChuck Lever 1131b0d93ad5SChuck Lever write_lock_bh(&sk->sk_callback_lock); 1132b0d93ad5SChuck Lever 1133b0d93ad5SChuck Lever sk->sk_user_data = xprt; 1134b0d93ad5SChuck Lever xprt->old_data_ready = sk->sk_data_ready; 1135b0d93ad5SChuck Lever xprt->old_state_change = sk->sk_state_change; 1136b0d93ad5SChuck Lever xprt->old_write_space = sk->sk_write_space; 1137b0d93ad5SChuck Lever sk->sk_data_ready = xs_udp_data_ready; 1138b0d93ad5SChuck Lever sk->sk_write_space = xs_udp_write_space; 1139b0d93ad5SChuck Lever sk->sk_no_check = UDP_CSUM_NORCV; 1140b079fa7bSTrond Myklebust sk->sk_allocation = GFP_ATOMIC; 1141b0d93ad5SChuck Lever 1142b0d93ad5SChuck Lever xprt_set_connected(xprt); 1143b0d93ad5SChuck Lever 1144b0d93ad5SChuck Lever /* Reset to new socket */ 1145ee0ac0c2SChuck Lever transport->sock = sock; 1146ee0ac0c2SChuck Lever transport->inet = sk; 1147b0d93ad5SChuck Lever 1148b0d93ad5SChuck Lever write_unlock_bh(&sk->sk_callback_lock); 1149b0d93ad5SChuck Lever } 1150470056c2SChuck Lever xs_udp_do_set_buffer_size(xprt); 1151a246b010SChuck Lever status = 0; 1152b0d93ad5SChuck Lever out: 1153b0d93ad5SChuck Lever xprt_wake_pending_tasks(xprt, status); 1154b0d93ad5SChuck Lever xprt_clear_connecting(xprt); 1155b0d93ad5SChuck Lever } 1156b0d93ad5SChuck Lever 11573167e12cSChuck Lever /* 11583167e12cSChuck Lever * We need to preserve the port number so the reply cache on the server can 11593167e12cSChuck Lever * find our cached RPC replies when we get around to reconnecting. 11603167e12cSChuck Lever */ 11613167e12cSChuck Lever static void xs_tcp_reuse_connection(struct rpc_xprt *xprt) 11623167e12cSChuck Lever { 11633167e12cSChuck Lever int result; 1164ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 11653167e12cSChuck Lever struct sockaddr any; 11663167e12cSChuck Lever 11673167e12cSChuck Lever dprintk("RPC: disconnecting xprt %p to reuse port\n", xprt); 11683167e12cSChuck Lever 11693167e12cSChuck Lever /* 11703167e12cSChuck Lever * Disconnect the transport socket by doing a connect operation 11713167e12cSChuck Lever * with AF_UNSPEC. This should return immediately... 11723167e12cSChuck Lever */ 11733167e12cSChuck Lever memset(&any, 0, sizeof(any)); 11743167e12cSChuck Lever any.sa_family = AF_UNSPEC; 1175ee0ac0c2SChuck Lever result = kernel_connect(transport->sock, &any, sizeof(any), 0); 11763167e12cSChuck Lever if (result) 11773167e12cSChuck Lever dprintk("RPC: AF_UNSPEC connect return code %d\n", 11783167e12cSChuck Lever result); 11793167e12cSChuck Lever } 11803167e12cSChuck Lever 1181b0d93ad5SChuck Lever /** 1182b0d93ad5SChuck Lever * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint 1183b0d93ad5SChuck Lever * @args: RPC transport to connect 1184b0d93ad5SChuck Lever * 1185b0d93ad5SChuck Lever * Invoked by a work queue tasklet. 1186b0d93ad5SChuck Lever */ 1187b0d93ad5SChuck Lever static void xs_tcp_connect_worker(void *args) 1188b0d93ad5SChuck Lever { 1189b0d93ad5SChuck Lever struct rpc_xprt *xprt = (struct rpc_xprt *)args; 1190ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1191ee0ac0c2SChuck Lever struct socket *sock = transport->sock; 1192b0d93ad5SChuck Lever int err, status = -EIO; 1193b0d93ad5SChuck Lever 1194ec739ef0SChuck Lever if (xprt->shutdown || !xprt_bound(xprt)) 1195a246b010SChuck Lever goto out; 1196a246b010SChuck Lever 1197ee0ac0c2SChuck Lever if (!sock) { 11983167e12cSChuck Lever /* start from scratch */ 1199b0d93ad5SChuck Lever if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) { 1200b0d93ad5SChuck Lever dprintk("RPC: can't create TCP transport socket (%d).\n", -err); 1201b0d93ad5SChuck Lever goto out; 1202b0d93ad5SChuck Lever } 1203b0d93ad5SChuck Lever 1204b0d93ad5SChuck Lever if (xprt->resvport && xs_bindresvport(xprt, sock) < 0) { 1205b0d93ad5SChuck Lever sock_release(sock); 1206b0d93ad5SChuck Lever goto out; 1207b0d93ad5SChuck Lever } 12083167e12cSChuck Lever } else 12093167e12cSChuck Lever /* "close" the socket, preserving the local port */ 12103167e12cSChuck Lever xs_tcp_reuse_connection(xprt); 1211b0d93ad5SChuck Lever 1212edb267a6SChuck Lever dprintk("RPC: worker connecting xprt %p to address: %s\n", 1213edb267a6SChuck Lever xprt, xs_print_peer_address(xprt, RPC_DISPLAY_ALL)); 1214edb267a6SChuck Lever 1215ee0ac0c2SChuck Lever if (!transport->inet) { 1216b0d93ad5SChuck Lever struct sock *sk = sock->sk; 1217b0d93ad5SChuck Lever 1218b0d93ad5SChuck Lever write_lock_bh(&sk->sk_callback_lock); 1219b0d93ad5SChuck Lever 1220b0d93ad5SChuck Lever sk->sk_user_data = xprt; 1221b0d93ad5SChuck Lever xprt->old_data_ready = sk->sk_data_ready; 1222b0d93ad5SChuck Lever xprt->old_state_change = sk->sk_state_change; 1223b0d93ad5SChuck Lever xprt->old_write_space = sk->sk_write_space; 1224b0d93ad5SChuck Lever sk->sk_data_ready = xs_tcp_data_ready; 1225b0d93ad5SChuck Lever sk->sk_state_change = xs_tcp_state_change; 1226b0d93ad5SChuck Lever sk->sk_write_space = xs_tcp_write_space; 1227b079fa7bSTrond Myklebust sk->sk_allocation = GFP_ATOMIC; 12283167e12cSChuck Lever 12293167e12cSChuck Lever /* socket options */ 12303167e12cSChuck Lever sk->sk_userlocks |= SOCK_BINDPORT_LOCK; 12313167e12cSChuck Lever sock_reset_flag(sk, SOCK_LINGER); 12323167e12cSChuck Lever tcp_sk(sk)->linger2 = 0; 12333167e12cSChuck Lever tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF; 1234b0d93ad5SChuck Lever 1235b0d93ad5SChuck Lever xprt_clear_connected(xprt); 1236b0d93ad5SChuck Lever 1237b0d93ad5SChuck Lever /* Reset to new socket */ 1238ee0ac0c2SChuck Lever transport->sock = sock; 1239ee0ac0c2SChuck Lever transport->inet = sk; 1240b0d93ad5SChuck Lever 1241b0d93ad5SChuck Lever write_unlock_bh(&sk->sk_callback_lock); 1242b0d93ad5SChuck Lever } 1243b0d93ad5SChuck Lever 1244b0d93ad5SChuck Lever /* Tell the socket layer to start connecting... */ 1245262ca07dSChuck Lever xprt->stat.connect_count++; 1246262ca07dSChuck Lever xprt->stat.connect_start = jiffies; 1247e6242e92SSridhar Samudrala status = kernel_connect(sock, (struct sockaddr *) &xprt->addr, 1248c4efcb1dSChuck Lever xprt->addrlen, O_NONBLOCK); 1249a246b010SChuck Lever dprintk("RPC: %p connect status %d connected %d sock state %d\n", 1250a246b010SChuck Lever xprt, -status, xprt_connected(xprt), sock->sk->sk_state); 1251a246b010SChuck Lever if (status < 0) { 1252a246b010SChuck Lever switch (status) { 1253a246b010SChuck Lever case -EINPROGRESS: 1254a246b010SChuck Lever case -EALREADY: 1255a246b010SChuck Lever goto out_clear; 12563167e12cSChuck Lever case -ECONNREFUSED: 12573167e12cSChuck Lever case -ECONNRESET: 12583167e12cSChuck Lever /* retry with existing socket, after a delay */ 12593167e12cSChuck Lever break; 12603167e12cSChuck Lever default: 12613167e12cSChuck Lever /* get rid of existing socket, and retry */ 12623167e12cSChuck Lever xs_close(xprt); 12633167e12cSChuck Lever break; 1264a246b010SChuck Lever } 1265a246b010SChuck Lever } 1266a246b010SChuck Lever out: 126744fbac22SChuck Lever xprt_wake_pending_tasks(xprt, status); 1268a246b010SChuck Lever out_clear: 12692226feb6SChuck Lever xprt_clear_connecting(xprt); 1270a246b010SChuck Lever } 1271a246b010SChuck Lever 12729903cd1cSChuck Lever /** 12739903cd1cSChuck Lever * xs_connect - connect a socket to a remote endpoint 12749903cd1cSChuck Lever * @task: address of RPC task that manages state of connect request 12759903cd1cSChuck Lever * 12769903cd1cSChuck Lever * TCP: If the remote end dropped the connection, delay reconnecting. 127703bf4b70SChuck Lever * 127803bf4b70SChuck Lever * UDP socket connects are synchronous, but we use a work queue anyway 127903bf4b70SChuck Lever * to guarantee that even unprivileged user processes can set up a 128003bf4b70SChuck Lever * socket on a privileged port. 128103bf4b70SChuck Lever * 128203bf4b70SChuck Lever * If a UDP socket connect fails, the delay behavior here prevents 128303bf4b70SChuck Lever * retry floods (hard mounts). 12849903cd1cSChuck Lever */ 12859903cd1cSChuck Lever static void xs_connect(struct rpc_task *task) 1286a246b010SChuck Lever { 1287a246b010SChuck Lever struct rpc_xprt *xprt = task->tk_xprt; 1288ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1289a246b010SChuck Lever 1290b0d93ad5SChuck Lever if (xprt_test_and_set_connecting(xprt)) 1291b0d93ad5SChuck Lever return; 1292b0d93ad5SChuck Lever 1293ee0ac0c2SChuck Lever if (transport->sock != NULL) { 129403bf4b70SChuck Lever dprintk("RPC: xs_connect delayed xprt %p for %lu seconds\n", 129503bf4b70SChuck Lever xprt, xprt->reestablish_timeout / HZ); 129655aa4f58SChuck Lever schedule_delayed_work(&xprt->connect_worker, 129703bf4b70SChuck Lever xprt->reestablish_timeout); 129803bf4b70SChuck Lever xprt->reestablish_timeout <<= 1; 129903bf4b70SChuck Lever if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO) 130003bf4b70SChuck Lever xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO; 13019903cd1cSChuck Lever } else { 13029903cd1cSChuck Lever dprintk("RPC: xs_connect scheduled xprt %p\n", xprt); 130355aa4f58SChuck Lever schedule_work(&xprt->connect_worker); 1304b0d93ad5SChuck Lever 1305a246b010SChuck Lever /* flush_scheduled_work can sleep... */ 1306a246b010SChuck Lever if (!RPC_IS_ASYNC(task)) 1307a246b010SChuck Lever flush_scheduled_work(); 1308a246b010SChuck Lever } 1309a246b010SChuck Lever } 1310a246b010SChuck Lever 1311262ca07dSChuck Lever /** 1312262ca07dSChuck Lever * xs_udp_print_stats - display UDP socket-specifc stats 1313262ca07dSChuck Lever * @xprt: rpc_xprt struct containing statistics 1314262ca07dSChuck Lever * @seq: output file 1315262ca07dSChuck Lever * 1316262ca07dSChuck Lever */ 1317262ca07dSChuck Lever static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 1318262ca07dSChuck Lever { 1319262ca07dSChuck Lever seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n", 1320262ca07dSChuck Lever xprt->port, 1321262ca07dSChuck Lever xprt->stat.bind_count, 1322262ca07dSChuck Lever xprt->stat.sends, 1323262ca07dSChuck Lever xprt->stat.recvs, 1324262ca07dSChuck Lever xprt->stat.bad_xids, 1325262ca07dSChuck Lever xprt->stat.req_u, 1326262ca07dSChuck Lever xprt->stat.bklog_u); 1327262ca07dSChuck Lever } 1328262ca07dSChuck Lever 1329262ca07dSChuck Lever /** 1330262ca07dSChuck Lever * xs_tcp_print_stats - display TCP socket-specifc stats 1331262ca07dSChuck Lever * @xprt: rpc_xprt struct containing statistics 1332262ca07dSChuck Lever * @seq: output file 1333262ca07dSChuck Lever * 1334262ca07dSChuck Lever */ 1335262ca07dSChuck Lever static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 1336262ca07dSChuck Lever { 1337262ca07dSChuck Lever long idle_time = 0; 1338262ca07dSChuck Lever 1339262ca07dSChuck Lever if (xprt_connected(xprt)) 1340262ca07dSChuck Lever idle_time = (long)(jiffies - xprt->last_used) / HZ; 1341262ca07dSChuck Lever 1342262ca07dSChuck Lever seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n", 1343262ca07dSChuck Lever xprt->port, 1344262ca07dSChuck Lever xprt->stat.bind_count, 1345262ca07dSChuck Lever xprt->stat.connect_count, 1346262ca07dSChuck Lever xprt->stat.connect_time, 1347262ca07dSChuck Lever idle_time, 1348262ca07dSChuck Lever xprt->stat.sends, 1349262ca07dSChuck Lever xprt->stat.recvs, 1350262ca07dSChuck Lever xprt->stat.bad_xids, 1351262ca07dSChuck Lever xprt->stat.req_u, 1352262ca07dSChuck Lever xprt->stat.bklog_u); 1353262ca07dSChuck Lever } 1354262ca07dSChuck Lever 1355262965f5SChuck Lever static struct rpc_xprt_ops xs_udp_ops = { 135643118c29SChuck Lever .set_buffer_size = xs_udp_set_buffer_size, 1357edb267a6SChuck Lever .print_addr = xs_print_peer_address, 135812a80469SChuck Lever .reserve_xprt = xprt_reserve_xprt_cong, 135949e9a890SChuck Lever .release_xprt = xprt_release_xprt_cong, 1360bbf7c1ddSChuck Lever .rpcbind = rpc_getport, 136192200412SChuck Lever .set_port = xs_set_port, 13629903cd1cSChuck Lever .connect = xs_connect, 136302107148SChuck Lever .buf_alloc = rpc_malloc, 136402107148SChuck Lever .buf_free = rpc_free, 1365262965f5SChuck Lever .send_request = xs_udp_send_request, 1366fe3aca29SChuck Lever .set_retrans_timeout = xprt_set_retrans_timeout_rtt, 136746c0ee8bSChuck Lever .timer = xs_udp_timer, 1368a58dd398SChuck Lever .release_request = xprt_release_rqst_cong, 1369262965f5SChuck Lever .close = xs_close, 1370262965f5SChuck Lever .destroy = xs_destroy, 1371262ca07dSChuck Lever .print_stats = xs_udp_print_stats, 1372262965f5SChuck Lever }; 1373262965f5SChuck Lever 1374262965f5SChuck Lever static struct rpc_xprt_ops xs_tcp_ops = { 1375edb267a6SChuck Lever .print_addr = xs_print_peer_address, 137612a80469SChuck Lever .reserve_xprt = xprt_reserve_xprt, 1377e0ab53deSTrond Myklebust .release_xprt = xs_tcp_release_xprt, 1378bbf7c1ddSChuck Lever .rpcbind = rpc_getport, 137992200412SChuck Lever .set_port = xs_set_port, 1380262965f5SChuck Lever .connect = xs_connect, 138102107148SChuck Lever .buf_alloc = rpc_malloc, 138202107148SChuck Lever .buf_free = rpc_free, 1383262965f5SChuck Lever .send_request = xs_tcp_send_request, 1384fe3aca29SChuck Lever .set_retrans_timeout = xprt_set_retrans_timeout_def, 13859903cd1cSChuck Lever .close = xs_close, 13869903cd1cSChuck Lever .destroy = xs_destroy, 1387262ca07dSChuck Lever .print_stats = xs_tcp_print_stats, 1388a246b010SChuck Lever }; 1389a246b010SChuck Lever 1390c8541ecdSChuck Lever static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size) 1391c8541ecdSChuck Lever { 1392c8541ecdSChuck Lever struct rpc_xprt *xprt; 1393ffc2e518SChuck Lever struct sock_xprt *new; 1394c8541ecdSChuck Lever 1395c8541ecdSChuck Lever if (addrlen > sizeof(xprt->addr)) { 1396c8541ecdSChuck Lever dprintk("RPC: xs_setup_xprt: address too large\n"); 1397c8541ecdSChuck Lever return ERR_PTR(-EBADF); 1398c8541ecdSChuck Lever } 1399c8541ecdSChuck Lever 1400ffc2e518SChuck Lever new = kzalloc(sizeof(*new), GFP_KERNEL); 1401ffc2e518SChuck Lever if (new == NULL) { 1402c8541ecdSChuck Lever dprintk("RPC: xs_setup_xprt: couldn't allocate rpc_xprt\n"); 1403c8541ecdSChuck Lever return ERR_PTR(-ENOMEM); 1404c8541ecdSChuck Lever } 1405ffc2e518SChuck Lever xprt = &new->xprt; 1406c8541ecdSChuck Lever 1407c8541ecdSChuck Lever xprt->max_reqs = slot_table_size; 1408c8541ecdSChuck Lever xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL); 1409c8541ecdSChuck Lever if (xprt->slot == NULL) { 1410c8541ecdSChuck Lever kfree(xprt); 1411c8541ecdSChuck Lever dprintk("RPC: xs_setup_xprt: couldn't allocate slot table\n"); 1412c8541ecdSChuck Lever return ERR_PTR(-ENOMEM); 1413c8541ecdSChuck Lever } 1414c8541ecdSChuck Lever 1415c8541ecdSChuck Lever memcpy(&xprt->addr, addr, addrlen); 1416c8541ecdSChuck Lever xprt->addrlen = addrlen; 1417c8541ecdSChuck Lever xprt->port = xs_get_random_port(); 1418c8541ecdSChuck Lever 1419c8541ecdSChuck Lever return xprt; 1420c8541ecdSChuck Lever } 1421c8541ecdSChuck Lever 14229903cd1cSChuck Lever /** 14239903cd1cSChuck Lever * xs_setup_udp - Set up transport to use a UDP socket 1424c8541ecdSChuck Lever * @addr: address of remote server 1425c8541ecdSChuck Lever * @addrlen: length of address in bytes 14269903cd1cSChuck Lever * @to: timeout parameters 14279903cd1cSChuck Lever * 14289903cd1cSChuck Lever */ 1429c8541ecdSChuck Lever struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to) 1430a246b010SChuck Lever { 1431c8541ecdSChuck Lever struct rpc_xprt *xprt; 1432a246b010SChuck Lever 1433c8541ecdSChuck Lever xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries); 1434c8541ecdSChuck Lever if (IS_ERR(xprt)) 1435c8541ecdSChuck Lever return xprt; 1436a246b010SChuck Lever 1437c8541ecdSChuck Lever if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0) 1438ec739ef0SChuck Lever xprt_set_bound(xprt); 1439ec739ef0SChuck Lever 1440ec739ef0SChuck Lever xprt->prot = IPPROTO_UDP; 1441808012fbSChuck Lever xprt->tsh_size = 0; 1442a246b010SChuck Lever /* XXX: header size can vary due to auth type, IPv6, etc. */ 1443a246b010SChuck Lever xprt->max_payload = (1U << 16) - (MAX_HEADER << 3); 1444a246b010SChuck Lever 1445b0d93ad5SChuck Lever INIT_WORK(&xprt->connect_worker, xs_udp_connect_worker, xprt); 144603bf4b70SChuck Lever xprt->bind_timeout = XS_BIND_TO; 144703bf4b70SChuck Lever xprt->connect_timeout = XS_UDP_CONN_TO; 144803bf4b70SChuck Lever xprt->reestablish_timeout = XS_UDP_REEST_TO; 144903bf4b70SChuck Lever xprt->idle_timeout = XS_IDLE_DISC_TO; 1450a246b010SChuck Lever 1451262965f5SChuck Lever xprt->ops = &xs_udp_ops; 1452a246b010SChuck Lever 1453a246b010SChuck Lever if (to) 1454a246b010SChuck Lever xprt->timeout = *to; 1455a246b010SChuck Lever else 14569903cd1cSChuck Lever xprt_set_timeout(&xprt->timeout, 5, 5 * HZ); 1457a246b010SChuck Lever 1458edb267a6SChuck Lever xs_format_peer_addresses(xprt); 1459edb267a6SChuck Lever dprintk("RPC: set up transport to address %s\n", 1460edb267a6SChuck Lever xs_print_peer_address(xprt, RPC_DISPLAY_ALL)); 1461edb267a6SChuck Lever 1462c8541ecdSChuck Lever return xprt; 1463a246b010SChuck Lever } 1464a246b010SChuck Lever 14659903cd1cSChuck Lever /** 14669903cd1cSChuck Lever * xs_setup_tcp - Set up transport to use a TCP socket 1467c8541ecdSChuck Lever * @addr: address of remote server 1468c8541ecdSChuck Lever * @addrlen: length of address in bytes 14699903cd1cSChuck Lever * @to: timeout parameters 14709903cd1cSChuck Lever * 14719903cd1cSChuck Lever */ 1472c8541ecdSChuck Lever struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to) 1473a246b010SChuck Lever { 1474c8541ecdSChuck Lever struct rpc_xprt *xprt; 1475a246b010SChuck Lever 1476c8541ecdSChuck Lever xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries); 1477c8541ecdSChuck Lever if (IS_ERR(xprt)) 1478c8541ecdSChuck Lever return xprt; 1479a246b010SChuck Lever 1480c8541ecdSChuck Lever if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0) 1481ec739ef0SChuck Lever xprt_set_bound(xprt); 1482ec739ef0SChuck Lever 1483ec739ef0SChuck Lever xprt->prot = IPPROTO_TCP; 1484808012fbSChuck Lever xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32); 1485808012fbSChuck Lever xprt->max_payload = RPC_MAX_FRAGMENT_SIZE; 1486a246b010SChuck Lever 1487b0d93ad5SChuck Lever INIT_WORK(&xprt->connect_worker, xs_tcp_connect_worker, xprt); 148803bf4b70SChuck Lever xprt->bind_timeout = XS_BIND_TO; 148903bf4b70SChuck Lever xprt->connect_timeout = XS_TCP_CONN_TO; 149003bf4b70SChuck Lever xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO; 149103bf4b70SChuck Lever xprt->idle_timeout = XS_IDLE_DISC_TO; 1492a246b010SChuck Lever 1493262965f5SChuck Lever xprt->ops = &xs_tcp_ops; 1494a246b010SChuck Lever 1495a246b010SChuck Lever if (to) 1496a246b010SChuck Lever xprt->timeout = *to; 1497a246b010SChuck Lever else 14989903cd1cSChuck Lever xprt_set_timeout(&xprt->timeout, 2, 60 * HZ); 1499a246b010SChuck Lever 1500edb267a6SChuck Lever xs_format_peer_addresses(xprt); 1501edb267a6SChuck Lever dprintk("RPC: set up transport to address %s\n", 1502edb267a6SChuck Lever xs_print_peer_address(xprt, RPC_DISPLAY_ALL)); 1503edb267a6SChuck Lever 1504c8541ecdSChuck Lever return xprt; 1505a246b010SChuck Lever } 1506