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; 148c8475461SChuck Lever 149c8475461SChuck Lever /* 150c8475461SChuck Lever * Connection of transports 151c8475461SChuck Lever */ 152c8475461SChuck Lever struct work_struct connect_worker; 153c8475461SChuck Lever unsigned short port; 1547c6e066eSChuck Lever 1557c6e066eSChuck Lever /* 1567c6e066eSChuck Lever * UDP socket buffer size parameters 1577c6e066eSChuck Lever */ 1587c6e066eSChuck Lever size_t rcvsize, 1597c6e066eSChuck Lever sndsize; 160314dfd79SChuck Lever 161314dfd79SChuck Lever /* 162314dfd79SChuck Lever * Saved socket callback addresses 163314dfd79SChuck Lever */ 164314dfd79SChuck Lever void (*old_data_ready)(struct sock *, int); 165314dfd79SChuck Lever void (*old_state_change)(struct sock *); 166314dfd79SChuck Lever void (*old_write_space)(struct sock *); 167ffc2e518SChuck Lever }; 168ffc2e518SChuck Lever 169e136d092SChuck Lever /* 170e136d092SChuck Lever * TCP receive state flags 171e136d092SChuck Lever */ 172e136d092SChuck Lever #define TCP_RCV_LAST_FRAG (1UL << 0) 173e136d092SChuck Lever #define TCP_RCV_COPY_FRAGHDR (1UL << 1) 174e136d092SChuck Lever #define TCP_RCV_COPY_XID (1UL << 2) 175e136d092SChuck Lever #define TCP_RCV_COPY_DATA (1UL << 3) 176e136d092SChuck Lever 177edb267a6SChuck Lever static void xs_format_peer_addresses(struct rpc_xprt *xprt) 178edb267a6SChuck Lever { 179edb267a6SChuck Lever struct sockaddr_in *addr = (struct sockaddr_in *) &xprt->addr; 180edb267a6SChuck Lever char *buf; 181edb267a6SChuck Lever 182edb267a6SChuck Lever buf = kzalloc(20, GFP_KERNEL); 183edb267a6SChuck Lever if (buf) { 184edb267a6SChuck Lever snprintf(buf, 20, "%u.%u.%u.%u", 185edb267a6SChuck Lever NIPQUAD(addr->sin_addr.s_addr)); 186edb267a6SChuck Lever } 187edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_ADDR] = buf; 188edb267a6SChuck Lever 189edb267a6SChuck Lever buf = kzalloc(8, GFP_KERNEL); 190edb267a6SChuck Lever if (buf) { 191edb267a6SChuck Lever snprintf(buf, 8, "%u", 192edb267a6SChuck Lever ntohs(addr->sin_port)); 193edb267a6SChuck Lever } 194edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_PORT] = buf; 195edb267a6SChuck Lever 196edb267a6SChuck Lever if (xprt->prot == IPPROTO_UDP) 197edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_PROTO] = "udp"; 198edb267a6SChuck Lever else 199edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_PROTO] = "tcp"; 200edb267a6SChuck Lever 201edb267a6SChuck Lever buf = kzalloc(48, GFP_KERNEL); 202edb267a6SChuck Lever if (buf) { 203edb267a6SChuck Lever snprintf(buf, 48, "addr=%u.%u.%u.%u port=%u proto=%s", 204edb267a6SChuck Lever NIPQUAD(addr->sin_addr.s_addr), 205edb267a6SChuck Lever ntohs(addr->sin_port), 206edb267a6SChuck Lever xprt->prot == IPPROTO_UDP ? "udp" : "tcp"); 207edb267a6SChuck Lever } 208edb267a6SChuck Lever xprt->address_strings[RPC_DISPLAY_ALL] = buf; 209edb267a6SChuck Lever } 210edb267a6SChuck Lever 211edb267a6SChuck Lever static void xs_free_peer_addresses(struct rpc_xprt *xprt) 212edb267a6SChuck Lever { 213edb267a6SChuck Lever kfree(xprt->address_strings[RPC_DISPLAY_ADDR]); 214edb267a6SChuck Lever kfree(xprt->address_strings[RPC_DISPLAY_PORT]); 215edb267a6SChuck Lever kfree(xprt->address_strings[RPC_DISPLAY_ALL]); 216edb267a6SChuck Lever } 217edb267a6SChuck Lever 218b4b5cc85SChuck Lever #define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL) 219b4b5cc85SChuck Lever 22024c5684bSTrond Myklebust static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more) 221b4b5cc85SChuck Lever { 222b4b5cc85SChuck Lever struct msghdr msg = { 223b4b5cc85SChuck Lever .msg_name = addr, 224b4b5cc85SChuck Lever .msg_namelen = addrlen, 22524c5684bSTrond Myklebust .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0), 22624c5684bSTrond Myklebust }; 22724c5684bSTrond Myklebust struct kvec iov = { 22824c5684bSTrond Myklebust .iov_base = vec->iov_base + base, 22924c5684bSTrond Myklebust .iov_len = vec->iov_len - base, 230b4b5cc85SChuck Lever }; 231b4b5cc85SChuck Lever 23224c5684bSTrond Myklebust if (iov.iov_len != 0) 233b4b5cc85SChuck Lever return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len); 234b4b5cc85SChuck Lever return kernel_sendmsg(sock, &msg, NULL, 0, 0); 235b4b5cc85SChuck Lever } 236b4b5cc85SChuck Lever 23724c5684bSTrond Myklebust static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more) 238b4b5cc85SChuck Lever { 23924c5684bSTrond Myklebust struct page **ppage; 24024c5684bSTrond Myklebust unsigned int remainder; 24124c5684bSTrond Myklebust int err, sent = 0; 242b4b5cc85SChuck Lever 24324c5684bSTrond Myklebust remainder = xdr->page_len - base; 24424c5684bSTrond Myklebust base += xdr->page_base; 24524c5684bSTrond Myklebust ppage = xdr->pages + (base >> PAGE_SHIFT); 24624c5684bSTrond Myklebust base &= ~PAGE_MASK; 24724c5684bSTrond Myklebust for(;;) { 24824c5684bSTrond Myklebust unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder); 24924c5684bSTrond Myklebust int flags = XS_SENDMSG_FLAGS; 25024c5684bSTrond Myklebust 25124c5684bSTrond Myklebust remainder -= len; 25224c5684bSTrond Myklebust if (remainder != 0 || more) 25324c5684bSTrond Myklebust flags |= MSG_MORE; 25424c5684bSTrond Myklebust err = sock->ops->sendpage(sock, *ppage, base, len, flags); 25524c5684bSTrond Myklebust if (remainder == 0 || err != len) 25624c5684bSTrond Myklebust break; 25724c5684bSTrond Myklebust sent += err; 25824c5684bSTrond Myklebust ppage++; 25924c5684bSTrond Myklebust base = 0; 26024c5684bSTrond Myklebust } 26124c5684bSTrond Myklebust if (sent == 0) 26224c5684bSTrond Myklebust return err; 26324c5684bSTrond Myklebust if (err > 0) 26424c5684bSTrond Myklebust sent += err; 26524c5684bSTrond Myklebust return sent; 266b4b5cc85SChuck Lever } 267b4b5cc85SChuck Lever 2689903cd1cSChuck Lever /** 2699903cd1cSChuck Lever * xs_sendpages - write pages directly to a socket 2709903cd1cSChuck Lever * @sock: socket to send on 2719903cd1cSChuck Lever * @addr: UDP only -- address of destination 2729903cd1cSChuck Lever * @addrlen: UDP only -- length of destination address 2739903cd1cSChuck Lever * @xdr: buffer containing this request 2749903cd1cSChuck Lever * @base: starting position in the buffer 2759903cd1cSChuck Lever * 276a246b010SChuck Lever */ 27724c5684bSTrond Myklebust static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base) 278a246b010SChuck Lever { 27924c5684bSTrond Myklebust unsigned int remainder = xdr->len - base; 28024c5684bSTrond Myklebust int err, sent = 0; 281a246b010SChuck Lever 282262965f5SChuck Lever if (unlikely(!sock)) 283262965f5SChuck Lever return -ENOTCONN; 284262965f5SChuck Lever 285262965f5SChuck Lever clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags); 28624c5684bSTrond Myklebust if (base != 0) { 28724c5684bSTrond Myklebust addr = NULL; 28824c5684bSTrond Myklebust addrlen = 0; 28924c5684bSTrond Myklebust } 290262965f5SChuck Lever 29124c5684bSTrond Myklebust if (base < xdr->head[0].iov_len || addr != NULL) { 29224c5684bSTrond Myklebust unsigned int len = xdr->head[0].iov_len - base; 29324c5684bSTrond Myklebust remainder -= len; 29424c5684bSTrond Myklebust err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0); 29524c5684bSTrond Myklebust if (remainder == 0 || err != len) 296a246b010SChuck Lever goto out; 29724c5684bSTrond Myklebust sent += err; 298a246b010SChuck Lever base = 0; 299a246b010SChuck Lever } else 30024c5684bSTrond Myklebust base -= xdr->head[0].iov_len; 301a246b010SChuck Lever 30224c5684bSTrond Myklebust if (base < xdr->page_len) { 30324c5684bSTrond Myklebust unsigned int len = xdr->page_len - base; 30424c5684bSTrond Myklebust remainder -= len; 30524c5684bSTrond Myklebust err = xs_send_pagedata(sock, xdr, base, remainder != 0); 30624c5684bSTrond Myklebust if (remainder == 0 || err != len) 307a246b010SChuck Lever goto out; 30824c5684bSTrond Myklebust sent += err; 309a246b010SChuck Lever base = 0; 31024c5684bSTrond Myklebust } else 31124c5684bSTrond Myklebust base -= xdr->page_len; 31224c5684bSTrond Myklebust 31324c5684bSTrond Myklebust if (base >= xdr->tail[0].iov_len) 31424c5684bSTrond Myklebust return sent; 31524c5684bSTrond Myklebust err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0); 316a246b010SChuck Lever out: 31724c5684bSTrond Myklebust if (sent == 0) 31824c5684bSTrond Myklebust return err; 31924c5684bSTrond Myklebust if (err > 0) 32024c5684bSTrond Myklebust sent += err; 32124c5684bSTrond Myklebust return sent; 322a246b010SChuck Lever } 323a246b010SChuck Lever 3249903cd1cSChuck Lever /** 325262965f5SChuck Lever * xs_nospace - place task on wait queue if transmit was incomplete 326262965f5SChuck Lever * @task: task to put to sleep 3279903cd1cSChuck Lever * 328a246b010SChuck Lever */ 329262965f5SChuck Lever static void xs_nospace(struct rpc_task *task) 330a246b010SChuck Lever { 331262965f5SChuck Lever struct rpc_rqst *req = task->tk_rqstp; 332262965f5SChuck Lever struct rpc_xprt *xprt = req->rq_xprt; 333ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 334a246b010SChuck Lever 335262965f5SChuck Lever dprintk("RPC: %4d xmit incomplete (%u left of %u)\n", 336262965f5SChuck Lever task->tk_pid, req->rq_slen - req->rq_bytes_sent, 337262965f5SChuck Lever req->rq_slen); 338a246b010SChuck Lever 339ee0ac0c2SChuck Lever if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) { 340262965f5SChuck Lever /* Protect against races with write_space */ 341262965f5SChuck Lever spin_lock_bh(&xprt->transport_lock); 342a246b010SChuck Lever 343262965f5SChuck Lever /* Don't race with disconnect */ 344262965f5SChuck Lever if (!xprt_connected(xprt)) 345262965f5SChuck Lever task->tk_status = -ENOTCONN; 346ee0ac0c2SChuck Lever else if (test_bit(SOCK_NOSPACE, &transport->sock->flags)) 347262965f5SChuck Lever xprt_wait_for_buffer_space(task); 348a246b010SChuck Lever 349262965f5SChuck Lever spin_unlock_bh(&xprt->transport_lock); 350262965f5SChuck Lever } else 351262965f5SChuck Lever /* Keep holding the socket if it is blocked */ 352262965f5SChuck Lever rpc_delay(task, HZ>>4); 353a246b010SChuck Lever } 354a246b010SChuck Lever 3559903cd1cSChuck Lever /** 356262965f5SChuck Lever * xs_udp_send_request - write an RPC request to a UDP socket 3579903cd1cSChuck Lever * @task: address of RPC task that manages the state of an RPC request 3589903cd1cSChuck Lever * 3599903cd1cSChuck Lever * Return values: 3609903cd1cSChuck Lever * 0: The request has been sent 3619903cd1cSChuck Lever * EAGAIN: The socket was blocked, please call again later to 3629903cd1cSChuck Lever * complete the request 363262965f5SChuck Lever * ENOTCONN: Caller needs to invoke connect logic then call again 3649903cd1cSChuck Lever * other: Some other error occured, the request was not sent 3659903cd1cSChuck Lever */ 366262965f5SChuck Lever static int xs_udp_send_request(struct rpc_task *task) 367a246b010SChuck Lever { 368a246b010SChuck Lever struct rpc_rqst *req = task->tk_rqstp; 369a246b010SChuck Lever struct rpc_xprt *xprt = req->rq_xprt; 370ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 371262965f5SChuck Lever struct xdr_buf *xdr = &req->rq_snd_buf; 372262965f5SChuck Lever int status; 373262965f5SChuck Lever 374262965f5SChuck Lever xs_pktdump("packet data:", 375262965f5SChuck Lever req->rq_svec->iov_base, 376262965f5SChuck Lever req->rq_svec->iov_len); 377262965f5SChuck Lever 378262965f5SChuck Lever req->rq_xtime = jiffies; 379ee0ac0c2SChuck Lever status = xs_sendpages(transport->sock, 380ee0ac0c2SChuck Lever (struct sockaddr *) &xprt->addr, 381ee0ac0c2SChuck Lever xprt->addrlen, xdr, 382ee0ac0c2SChuck Lever req->rq_bytes_sent); 383262965f5SChuck Lever 384262965f5SChuck Lever dprintk("RPC: xs_udp_send_request(%u) = %d\n", 385262965f5SChuck Lever xdr->len - req->rq_bytes_sent, status); 386262965f5SChuck Lever 387262965f5SChuck Lever if (likely(status >= (int) req->rq_slen)) 388262965f5SChuck Lever return 0; 389262965f5SChuck Lever 390262965f5SChuck Lever /* Still some bytes left; set up for a retry later. */ 391262965f5SChuck Lever if (status > 0) 392262965f5SChuck Lever status = -EAGAIN; 393262965f5SChuck Lever 394262965f5SChuck Lever switch (status) { 395262965f5SChuck Lever case -ENETUNREACH: 396262965f5SChuck Lever case -EPIPE: 397262965f5SChuck Lever case -ECONNREFUSED: 398262965f5SChuck Lever /* When the server has died, an ICMP port unreachable message 399262965f5SChuck Lever * prompts ECONNREFUSED. */ 400262965f5SChuck Lever break; 401262965f5SChuck Lever case -EAGAIN: 402262965f5SChuck Lever xs_nospace(task); 403262965f5SChuck Lever break; 404262965f5SChuck Lever default: 405262965f5SChuck Lever dprintk("RPC: sendmsg returned unrecognized error %d\n", 406262965f5SChuck Lever -status); 407262965f5SChuck Lever break; 408262965f5SChuck Lever } 409262965f5SChuck Lever 410262965f5SChuck Lever return status; 411262965f5SChuck Lever } 412262965f5SChuck Lever 413808012fbSChuck Lever static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf) 414808012fbSChuck Lever { 415808012fbSChuck Lever u32 reclen = buf->len - sizeof(rpc_fraghdr); 416808012fbSChuck Lever rpc_fraghdr *base = buf->head[0].iov_base; 417808012fbSChuck Lever *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen); 418808012fbSChuck Lever } 419808012fbSChuck Lever 420262965f5SChuck Lever /** 421262965f5SChuck Lever * xs_tcp_send_request - write an RPC request to a TCP socket 422262965f5SChuck Lever * @task: address of RPC task that manages the state of an RPC request 423262965f5SChuck Lever * 424262965f5SChuck Lever * Return values: 425262965f5SChuck Lever * 0: The request has been sent 426262965f5SChuck Lever * EAGAIN: The socket was blocked, please call again later to 427262965f5SChuck Lever * complete the request 428262965f5SChuck Lever * ENOTCONN: Caller needs to invoke connect logic then call again 429262965f5SChuck Lever * other: Some other error occured, the request was not sent 430262965f5SChuck Lever * 431262965f5SChuck Lever * XXX: In the case of soft timeouts, should we eventually give up 432262965f5SChuck Lever * if sendmsg is not able to make progress? 433262965f5SChuck Lever */ 434262965f5SChuck Lever static int xs_tcp_send_request(struct rpc_task *task) 435262965f5SChuck Lever { 436262965f5SChuck Lever struct rpc_rqst *req = task->tk_rqstp; 437262965f5SChuck Lever struct rpc_xprt *xprt = req->rq_xprt; 438ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 439262965f5SChuck Lever struct xdr_buf *xdr = &req->rq_snd_buf; 440a246b010SChuck Lever int status, retry = 0; 441a246b010SChuck Lever 442808012fbSChuck Lever xs_encode_tcp_record_marker(&req->rq_snd_buf); 443262965f5SChuck Lever 444262965f5SChuck Lever xs_pktdump("packet data:", 445262965f5SChuck Lever req->rq_svec->iov_base, 446262965f5SChuck Lever req->rq_svec->iov_len); 447a246b010SChuck Lever 448a246b010SChuck Lever /* Continue transmitting the packet/record. We must be careful 449a246b010SChuck Lever * to cope with writespace callbacks arriving _after_ we have 450262965f5SChuck Lever * called sendmsg(). */ 451a246b010SChuck Lever while (1) { 452a246b010SChuck Lever req->rq_xtime = jiffies; 453ee0ac0c2SChuck Lever status = xs_sendpages(transport->sock, 454ee0ac0c2SChuck Lever NULL, 0, xdr, req->rq_bytes_sent); 455a246b010SChuck Lever 456262965f5SChuck Lever dprintk("RPC: xs_tcp_send_request(%u) = %d\n", 457262965f5SChuck Lever xdr->len - req->rq_bytes_sent, status); 458262965f5SChuck Lever 459262965f5SChuck Lever if (unlikely(status < 0)) 460a246b010SChuck Lever break; 461a246b010SChuck Lever 462a246b010SChuck Lever /* If we've sent the entire packet, immediately 463a246b010SChuck Lever * reset the count of bytes sent. */ 464262965f5SChuck Lever req->rq_bytes_sent += status; 465ef759a2eSChuck Lever task->tk_bytes_sent += status; 466262965f5SChuck Lever if (likely(req->rq_bytes_sent >= req->rq_slen)) { 467a246b010SChuck Lever req->rq_bytes_sent = 0; 468a246b010SChuck Lever return 0; 469a246b010SChuck Lever } 470262965f5SChuck Lever 471a246b010SChuck Lever status = -EAGAIN; 472262965f5SChuck Lever if (retry++ > XS_SENDMSG_RETRY) 473a246b010SChuck Lever break; 474a246b010SChuck Lever } 475a246b010SChuck Lever 476262965f5SChuck Lever switch (status) { 477262965f5SChuck Lever case -EAGAIN: 478262965f5SChuck Lever xs_nospace(task); 479262965f5SChuck Lever break; 480262965f5SChuck Lever case -ECONNREFUSED: 481262965f5SChuck Lever case -ECONNRESET: 482262965f5SChuck Lever case -ENOTCONN: 483262965f5SChuck Lever case -EPIPE: 484262965f5SChuck Lever status = -ENOTCONN; 485262965f5SChuck Lever break; 486262965f5SChuck Lever default: 487262965f5SChuck Lever dprintk("RPC: sendmsg returned unrecognized error %d\n", 488262965f5SChuck Lever -status); 48943118c29SChuck Lever xprt_disconnect(xprt); 490a246b010SChuck Lever break; 491a246b010SChuck Lever } 492a246b010SChuck Lever 493a246b010SChuck Lever return status; 494a246b010SChuck Lever } 495a246b010SChuck Lever 4969903cd1cSChuck Lever /** 497e0ab53deSTrond Myklebust * xs_tcp_release_xprt - clean up after a tcp transmission 498e0ab53deSTrond Myklebust * @xprt: transport 499e0ab53deSTrond Myklebust * @task: rpc task 500e0ab53deSTrond Myklebust * 501e0ab53deSTrond Myklebust * This cleans up if an error causes us to abort the transmission of a request. 502e0ab53deSTrond Myklebust * In this case, the socket may need to be reset in order to avoid confusing 503e0ab53deSTrond Myklebust * the server. 504e0ab53deSTrond Myklebust */ 505e0ab53deSTrond Myklebust static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 506e0ab53deSTrond Myklebust { 507e0ab53deSTrond Myklebust struct rpc_rqst *req; 508e0ab53deSTrond Myklebust 509e0ab53deSTrond Myklebust if (task != xprt->snd_task) 510e0ab53deSTrond Myklebust return; 511e0ab53deSTrond Myklebust if (task == NULL) 512e0ab53deSTrond Myklebust goto out_release; 513e0ab53deSTrond Myklebust req = task->tk_rqstp; 514e0ab53deSTrond Myklebust if (req->rq_bytes_sent == 0) 515e0ab53deSTrond Myklebust goto out_release; 516e0ab53deSTrond Myklebust if (req->rq_bytes_sent == req->rq_snd_buf.len) 517e0ab53deSTrond Myklebust goto out_release; 518e0ab53deSTrond Myklebust set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state); 519e0ab53deSTrond Myklebust out_release: 520e0ab53deSTrond Myklebust xprt_release_xprt(xprt, task); 521e0ab53deSTrond Myklebust } 522e0ab53deSTrond Myklebust 523e0ab53deSTrond Myklebust /** 5249903cd1cSChuck Lever * xs_close - close a socket 5259903cd1cSChuck Lever * @xprt: transport 5269903cd1cSChuck Lever * 5273167e12cSChuck Lever * This is used when all requests are complete; ie, no DRC state remains 5283167e12cSChuck Lever * on the server we want to save. 529a246b010SChuck Lever */ 5309903cd1cSChuck Lever static void xs_close(struct rpc_xprt *xprt) 531a246b010SChuck Lever { 532ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 533ee0ac0c2SChuck Lever struct socket *sock = transport->sock; 534ee0ac0c2SChuck Lever struct sock *sk = transport->inet; 535a246b010SChuck Lever 536a246b010SChuck Lever if (!sk) 537632e3bdcSTrond Myklebust goto clear_close_wait; 538a246b010SChuck Lever 5399903cd1cSChuck Lever dprintk("RPC: xs_close xprt %p\n", xprt); 5409903cd1cSChuck Lever 541a246b010SChuck Lever write_lock_bh(&sk->sk_callback_lock); 542ee0ac0c2SChuck Lever transport->inet = NULL; 543ee0ac0c2SChuck Lever transport->sock = NULL; 544a246b010SChuck Lever 545a246b010SChuck Lever sk->sk_user_data = NULL; 546314dfd79SChuck Lever sk->sk_data_ready = transport->old_data_ready; 547314dfd79SChuck Lever sk->sk_state_change = transport->old_state_change; 548314dfd79SChuck Lever sk->sk_write_space = transport->old_write_space; 549a246b010SChuck Lever write_unlock_bh(&sk->sk_callback_lock); 550a246b010SChuck Lever 551a246b010SChuck Lever sk->sk_no_check = 0; 552a246b010SChuck Lever 553a246b010SChuck Lever sock_release(sock); 554632e3bdcSTrond Myklebust clear_close_wait: 555632e3bdcSTrond Myklebust smp_mb__before_clear_bit(); 556632e3bdcSTrond Myklebust clear_bit(XPRT_CLOSE_WAIT, &xprt->state); 557632e3bdcSTrond Myklebust smp_mb__after_clear_bit(); 558a246b010SChuck Lever } 559a246b010SChuck Lever 5609903cd1cSChuck Lever /** 5619903cd1cSChuck Lever * xs_destroy - prepare to shutdown a transport 5629903cd1cSChuck Lever * @xprt: doomed transport 5639903cd1cSChuck Lever * 5649903cd1cSChuck Lever */ 5659903cd1cSChuck Lever static void xs_destroy(struct rpc_xprt *xprt) 566a246b010SChuck Lever { 567c8475461SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 568c8475461SChuck Lever 5699903cd1cSChuck Lever dprintk("RPC: xs_destroy xprt %p\n", xprt); 5709903cd1cSChuck Lever 571c8475461SChuck Lever cancel_delayed_work(&transport->connect_worker); 572a246b010SChuck Lever flush_scheduled_work(); 573a246b010SChuck Lever 574a246b010SChuck Lever xprt_disconnect(xprt); 5759903cd1cSChuck Lever xs_close(xprt); 576edb267a6SChuck Lever xs_free_peer_addresses(xprt); 577a246b010SChuck Lever kfree(xprt->slot); 578c8541ecdSChuck Lever kfree(xprt); 579a246b010SChuck Lever } 580a246b010SChuck Lever 5819903cd1cSChuck Lever static inline struct rpc_xprt *xprt_from_sock(struct sock *sk) 5829903cd1cSChuck Lever { 5839903cd1cSChuck Lever return (struct rpc_xprt *) sk->sk_user_data; 5849903cd1cSChuck Lever } 5859903cd1cSChuck Lever 5869903cd1cSChuck Lever /** 5879903cd1cSChuck Lever * xs_udp_data_ready - "data ready" callback for UDP sockets 5889903cd1cSChuck Lever * @sk: socket with data to read 5899903cd1cSChuck Lever * @len: how much data to read 5909903cd1cSChuck Lever * 591a246b010SChuck Lever */ 5929903cd1cSChuck Lever static void xs_udp_data_ready(struct sock *sk, int len) 593a246b010SChuck Lever { 594a246b010SChuck Lever struct rpc_task *task; 595a246b010SChuck Lever struct rpc_xprt *xprt; 596a246b010SChuck Lever struct rpc_rqst *rovr; 597a246b010SChuck Lever struct sk_buff *skb; 598a246b010SChuck Lever int err, repsize, copied; 599d8ed029dSAlexey Dobriyan u32 _xid; 600d8ed029dSAlexey Dobriyan __be32 *xp; 601a246b010SChuck Lever 602a246b010SChuck Lever read_lock(&sk->sk_callback_lock); 6039903cd1cSChuck Lever dprintk("RPC: xs_udp_data_ready...\n"); 6049903cd1cSChuck Lever if (!(xprt = xprt_from_sock(sk))) 605a246b010SChuck Lever goto out; 606a246b010SChuck Lever 607a246b010SChuck Lever if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL) 608a246b010SChuck Lever goto out; 609a246b010SChuck Lever 610a246b010SChuck Lever if (xprt->shutdown) 611a246b010SChuck Lever goto dropit; 612a246b010SChuck Lever 613a246b010SChuck Lever repsize = skb->len - sizeof(struct udphdr); 614a246b010SChuck Lever if (repsize < 4) { 6159903cd1cSChuck Lever dprintk("RPC: impossible RPC reply size %d!\n", repsize); 616a246b010SChuck Lever goto dropit; 617a246b010SChuck Lever } 618a246b010SChuck Lever 619a246b010SChuck Lever /* Copy the XID from the skb... */ 620a246b010SChuck Lever xp = skb_header_pointer(skb, sizeof(struct udphdr), 621a246b010SChuck Lever sizeof(_xid), &_xid); 622a246b010SChuck Lever if (xp == NULL) 623a246b010SChuck Lever goto dropit; 624a246b010SChuck Lever 625a246b010SChuck Lever /* Look up and lock the request corresponding to the given XID */ 6264a0f8c04SChuck Lever spin_lock(&xprt->transport_lock); 627a246b010SChuck Lever rovr = xprt_lookup_rqst(xprt, *xp); 628a246b010SChuck Lever if (!rovr) 629a246b010SChuck Lever goto out_unlock; 630a246b010SChuck Lever task = rovr->rq_task; 631a246b010SChuck Lever 632a246b010SChuck Lever if ((copied = rovr->rq_private_buf.buflen) > repsize) 633a246b010SChuck Lever copied = repsize; 634a246b010SChuck Lever 635a246b010SChuck Lever /* Suck it into the iovec, verify checksum if not done by hw. */ 636a246b010SChuck Lever if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) 637a246b010SChuck Lever goto out_unlock; 638a246b010SChuck Lever 639a246b010SChuck Lever /* Something worked... */ 640a246b010SChuck Lever dst_confirm(skb->dst); 641a246b010SChuck Lever 6421570c1e4SChuck Lever xprt_adjust_cwnd(task, copied); 6431570c1e4SChuck Lever xprt_update_rtt(task); 6441570c1e4SChuck Lever xprt_complete_rqst(task, copied); 645a246b010SChuck Lever 646a246b010SChuck Lever out_unlock: 6474a0f8c04SChuck Lever spin_unlock(&xprt->transport_lock); 648a246b010SChuck Lever dropit: 649a246b010SChuck Lever skb_free_datagram(sk, skb); 650a246b010SChuck Lever out: 651a246b010SChuck Lever read_unlock(&sk->sk_callback_lock); 652a246b010SChuck Lever } 653a246b010SChuck Lever 6549903cd1cSChuck Lever static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc) 655a246b010SChuck Lever { 65651971139SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 657a246b010SChuck Lever size_t len, used; 658a246b010SChuck Lever char *p; 659a246b010SChuck Lever 66051971139SChuck Lever p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset; 66151971139SChuck Lever len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset; 662*9d292316SChuck Lever used = xdr_skb_read_bits(desc, p, len); 66351971139SChuck Lever transport->tcp_offset += used; 664a246b010SChuck Lever if (used != len) 665a246b010SChuck Lever return; 666808012fbSChuck Lever 66751971139SChuck Lever transport->tcp_reclen = ntohl(transport->tcp_fraghdr); 66851971139SChuck Lever if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT) 669e136d092SChuck Lever transport->tcp_flags |= TCP_RCV_LAST_FRAG; 670a246b010SChuck Lever else 671e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_LAST_FRAG; 67251971139SChuck Lever transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK; 673808012fbSChuck Lever 674e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR; 67551971139SChuck Lever transport->tcp_offset = 0; 676808012fbSChuck Lever 677a246b010SChuck Lever /* Sanity check of the record length */ 67851971139SChuck Lever if (unlikely(transport->tcp_reclen < 4)) { 6799903cd1cSChuck Lever dprintk("RPC: invalid TCP record fragment length\n"); 680a246b010SChuck Lever xprt_disconnect(xprt); 6819903cd1cSChuck Lever return; 682a246b010SChuck Lever } 683a246b010SChuck Lever dprintk("RPC: reading TCP record fragment of length %d\n", 68451971139SChuck Lever transport->tcp_reclen); 685a246b010SChuck Lever } 686a246b010SChuck Lever 68751971139SChuck Lever static void xs_tcp_check_fraghdr(struct sock_xprt *transport) 688a246b010SChuck Lever { 68951971139SChuck Lever if (transport->tcp_offset == transport->tcp_reclen) { 690e136d092SChuck Lever transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR; 69151971139SChuck Lever transport->tcp_offset = 0; 692e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_LAST_FRAG) { 693e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 694e136d092SChuck Lever transport->tcp_flags |= TCP_RCV_COPY_XID; 69551971139SChuck Lever transport->tcp_copied = 0; 696a246b010SChuck Lever } 697a246b010SChuck Lever } 698a246b010SChuck Lever } 699a246b010SChuck Lever 70051971139SChuck Lever static inline void xs_tcp_read_xid(struct sock_xprt *transport, skb_reader_t *desc) 701a246b010SChuck Lever { 702a246b010SChuck Lever size_t len, used; 703a246b010SChuck Lever char *p; 704a246b010SChuck Lever 70551971139SChuck Lever len = sizeof(transport->tcp_xid) - transport->tcp_offset; 706a246b010SChuck Lever dprintk("RPC: reading XID (%Zu bytes)\n", len); 70751971139SChuck Lever p = ((char *) &transport->tcp_xid) + transport->tcp_offset; 708*9d292316SChuck Lever used = xdr_skb_read_bits(desc, p, len); 70951971139SChuck Lever transport->tcp_offset += used; 710a246b010SChuck Lever if (used != len) 711a246b010SChuck Lever return; 712e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_XID; 713e136d092SChuck Lever transport->tcp_flags |= TCP_RCV_COPY_DATA; 71451971139SChuck Lever transport->tcp_copied = 4; 715a246b010SChuck Lever dprintk("RPC: reading reply for XID %08x\n", 71651971139SChuck Lever ntohl(transport->tcp_xid)); 71751971139SChuck Lever xs_tcp_check_fraghdr(transport); 718a246b010SChuck Lever } 719a246b010SChuck Lever 7209903cd1cSChuck Lever static inline void xs_tcp_read_request(struct rpc_xprt *xprt, skb_reader_t *desc) 721a246b010SChuck Lever { 72251971139SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 723a246b010SChuck Lever struct rpc_rqst *req; 724a246b010SChuck Lever struct xdr_buf *rcvbuf; 725a246b010SChuck Lever size_t len; 726a246b010SChuck Lever ssize_t r; 727a246b010SChuck Lever 728a246b010SChuck Lever /* Find and lock the request corresponding to this xid */ 7294a0f8c04SChuck Lever spin_lock(&xprt->transport_lock); 73051971139SChuck Lever req = xprt_lookup_rqst(xprt, transport->tcp_xid); 731a246b010SChuck Lever if (!req) { 732e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 733a246b010SChuck Lever dprintk("RPC: XID %08x request not found!\n", 73451971139SChuck Lever ntohl(transport->tcp_xid)); 7354a0f8c04SChuck Lever spin_unlock(&xprt->transport_lock); 736a246b010SChuck Lever return; 737a246b010SChuck Lever } 738a246b010SChuck Lever 739a246b010SChuck Lever rcvbuf = &req->rq_private_buf; 740a246b010SChuck Lever len = desc->count; 74151971139SChuck Lever if (len > transport->tcp_reclen - transport->tcp_offset) { 742a246b010SChuck Lever skb_reader_t my_desc; 743a246b010SChuck Lever 74451971139SChuck Lever len = transport->tcp_reclen - transport->tcp_offset; 745a246b010SChuck Lever memcpy(&my_desc, desc, sizeof(my_desc)); 746a246b010SChuck Lever my_desc.count = len; 74751971139SChuck Lever r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied, 748*9d292316SChuck Lever &my_desc, xdr_skb_read_bits); 749a246b010SChuck Lever desc->count -= r; 750a246b010SChuck Lever desc->offset += r; 751a246b010SChuck Lever } else 75251971139SChuck Lever r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied, 753*9d292316SChuck Lever desc, xdr_skb_read_bits); 754a246b010SChuck Lever 755a246b010SChuck Lever if (r > 0) { 75651971139SChuck Lever transport->tcp_copied += r; 75751971139SChuck Lever transport->tcp_offset += r; 758a246b010SChuck Lever } 759a246b010SChuck Lever if (r != len) { 760a246b010SChuck Lever /* Error when copying to the receive buffer, 761a246b010SChuck Lever * usually because we weren't able to allocate 762a246b010SChuck Lever * additional buffer pages. All we can do now 763e136d092SChuck Lever * is turn off TCP_RCV_COPY_DATA, so the request 764a246b010SChuck Lever * will not receive any additional updates, 765a246b010SChuck Lever * and time out. 766a246b010SChuck Lever * Any remaining data from this record will 767a246b010SChuck Lever * be discarded. 768a246b010SChuck Lever */ 769e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 770a246b010SChuck Lever dprintk("RPC: XID %08x truncated request\n", 77151971139SChuck Lever ntohl(transport->tcp_xid)); 772a246b010SChuck Lever dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n", 77351971139SChuck Lever xprt, transport->tcp_copied, transport->tcp_offset, 77451971139SChuck Lever transport->tcp_reclen); 775a246b010SChuck Lever goto out; 776a246b010SChuck Lever } 777a246b010SChuck Lever 778a246b010SChuck Lever dprintk("RPC: XID %08x read %Zd bytes\n", 77951971139SChuck Lever ntohl(transport->tcp_xid), r); 780a246b010SChuck Lever dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n", 78151971139SChuck Lever xprt, transport->tcp_copied, transport->tcp_offset, 78251971139SChuck Lever transport->tcp_reclen); 783a246b010SChuck Lever 78451971139SChuck Lever if (transport->tcp_copied == req->rq_private_buf.buflen) 785e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 78651971139SChuck Lever else if (transport->tcp_offset == transport->tcp_reclen) { 787e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_LAST_FRAG) 788e136d092SChuck Lever transport->tcp_flags &= ~TCP_RCV_COPY_DATA; 789a246b010SChuck Lever } 790a246b010SChuck Lever 791a246b010SChuck Lever out: 792e136d092SChuck Lever if (!(transport->tcp_flags & TCP_RCV_COPY_DATA)) 79351971139SChuck Lever xprt_complete_rqst(req->rq_task, transport->tcp_copied); 7944a0f8c04SChuck Lever spin_unlock(&xprt->transport_lock); 79551971139SChuck Lever xs_tcp_check_fraghdr(transport); 796a246b010SChuck Lever } 797a246b010SChuck Lever 79851971139SChuck Lever static inline void xs_tcp_read_discard(struct sock_xprt *transport, skb_reader_t *desc) 799a246b010SChuck Lever { 800a246b010SChuck Lever size_t len; 801a246b010SChuck Lever 80251971139SChuck Lever len = transport->tcp_reclen - transport->tcp_offset; 803a246b010SChuck Lever if (len > desc->count) 804a246b010SChuck Lever len = desc->count; 805a246b010SChuck Lever desc->count -= len; 806a246b010SChuck Lever desc->offset += len; 80751971139SChuck Lever transport->tcp_offset += len; 808a246b010SChuck Lever dprintk("RPC: discarded %Zu bytes\n", len); 80951971139SChuck Lever xs_tcp_check_fraghdr(transport); 810a246b010SChuck Lever } 811a246b010SChuck Lever 8129903cd1cSChuck Lever static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len) 813a246b010SChuck Lever { 814a246b010SChuck Lever struct rpc_xprt *xprt = rd_desc->arg.data; 81551971139SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 816a246b010SChuck Lever skb_reader_t desc = { 817a246b010SChuck Lever .skb = skb, 818a246b010SChuck Lever .offset = offset, 819a246b010SChuck Lever .count = len, 820a246b010SChuck Lever }; 821a246b010SChuck Lever 8229903cd1cSChuck Lever dprintk("RPC: xs_tcp_data_recv started\n"); 823a246b010SChuck Lever do { 824a246b010SChuck Lever /* Read in a new fragment marker if necessary */ 825a246b010SChuck Lever /* Can we ever really expect to get completely empty fragments? */ 826e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) { 8279903cd1cSChuck Lever xs_tcp_read_fraghdr(xprt, &desc); 828a246b010SChuck Lever continue; 829a246b010SChuck Lever } 830a246b010SChuck Lever /* Read in the xid if necessary */ 831e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_COPY_XID) { 83251971139SChuck Lever xs_tcp_read_xid(transport, &desc); 833a246b010SChuck Lever continue; 834a246b010SChuck Lever } 835a246b010SChuck Lever /* Read in the request data */ 836e136d092SChuck Lever if (transport->tcp_flags & TCP_RCV_COPY_DATA) { 8379903cd1cSChuck Lever xs_tcp_read_request(xprt, &desc); 838a246b010SChuck Lever continue; 839a246b010SChuck Lever } 840a246b010SChuck Lever /* Skip over any trailing bytes on short reads */ 84151971139SChuck Lever xs_tcp_read_discard(transport, &desc); 842a246b010SChuck Lever } while (desc.count); 8439903cd1cSChuck Lever dprintk("RPC: xs_tcp_data_recv done\n"); 844a246b010SChuck Lever return len - desc.count; 845a246b010SChuck Lever } 846a246b010SChuck Lever 8479903cd1cSChuck Lever /** 8489903cd1cSChuck Lever * xs_tcp_data_ready - "data ready" callback for TCP sockets 8499903cd1cSChuck Lever * @sk: socket with data to read 8509903cd1cSChuck Lever * @bytes: how much data to read 8519903cd1cSChuck Lever * 8529903cd1cSChuck Lever */ 8539903cd1cSChuck Lever static void xs_tcp_data_ready(struct sock *sk, int bytes) 854a246b010SChuck Lever { 855a246b010SChuck Lever struct rpc_xprt *xprt; 856a246b010SChuck Lever read_descriptor_t rd_desc; 857a246b010SChuck Lever 858a246b010SChuck Lever read_lock(&sk->sk_callback_lock); 8599903cd1cSChuck Lever dprintk("RPC: xs_tcp_data_ready...\n"); 8609903cd1cSChuck Lever if (!(xprt = xprt_from_sock(sk))) 861a246b010SChuck Lever goto out; 862a246b010SChuck Lever if (xprt->shutdown) 863a246b010SChuck Lever goto out; 864a246b010SChuck Lever 8659903cd1cSChuck Lever /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */ 866a246b010SChuck Lever rd_desc.arg.data = xprt; 867a246b010SChuck Lever rd_desc.count = 65536; 8689903cd1cSChuck Lever tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv); 869a246b010SChuck Lever out: 870a246b010SChuck Lever read_unlock(&sk->sk_callback_lock); 871a246b010SChuck Lever } 872a246b010SChuck Lever 8739903cd1cSChuck Lever /** 8749903cd1cSChuck Lever * xs_tcp_state_change - callback to handle TCP socket state changes 8759903cd1cSChuck Lever * @sk: socket whose state has changed 8769903cd1cSChuck Lever * 8779903cd1cSChuck Lever */ 8789903cd1cSChuck Lever static void xs_tcp_state_change(struct sock *sk) 879a246b010SChuck Lever { 880a246b010SChuck Lever struct rpc_xprt *xprt; 881a246b010SChuck Lever 882a246b010SChuck Lever read_lock(&sk->sk_callback_lock); 883a246b010SChuck Lever if (!(xprt = xprt_from_sock(sk))) 884a246b010SChuck Lever goto out; 8859903cd1cSChuck Lever dprintk("RPC: xs_tcp_state_change client %p...\n", xprt); 886a246b010SChuck Lever dprintk("RPC: state %x conn %d dead %d zapped %d\n", 887a246b010SChuck Lever sk->sk_state, xprt_connected(xprt), 888a246b010SChuck Lever sock_flag(sk, SOCK_DEAD), 889a246b010SChuck Lever sock_flag(sk, SOCK_ZAPPED)); 890a246b010SChuck Lever 891a246b010SChuck Lever switch (sk->sk_state) { 892a246b010SChuck Lever case TCP_ESTABLISHED: 8934a0f8c04SChuck Lever spin_lock_bh(&xprt->transport_lock); 894a246b010SChuck Lever if (!xprt_test_and_set_connected(xprt)) { 89551971139SChuck Lever struct sock_xprt *transport = container_of(xprt, 89651971139SChuck Lever struct sock_xprt, xprt); 89751971139SChuck Lever 898a246b010SChuck Lever /* Reset TCP record info */ 89951971139SChuck Lever transport->tcp_offset = 0; 90051971139SChuck Lever transport->tcp_reclen = 0; 90151971139SChuck Lever transport->tcp_copied = 0; 902e136d092SChuck Lever transport->tcp_flags = 903e136d092SChuck Lever TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID; 90451971139SChuck Lever 90503bf4b70SChuck Lever xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO; 90644fbac22SChuck Lever xprt_wake_pending_tasks(xprt, 0); 907a246b010SChuck Lever } 9084a0f8c04SChuck Lever spin_unlock_bh(&xprt->transport_lock); 909a246b010SChuck Lever break; 910a246b010SChuck Lever case TCP_SYN_SENT: 911a246b010SChuck Lever case TCP_SYN_RECV: 912a246b010SChuck Lever break; 913632e3bdcSTrond Myklebust case TCP_CLOSE_WAIT: 914632e3bdcSTrond Myklebust /* Try to schedule an autoclose RPC calls */ 915632e3bdcSTrond Myklebust set_bit(XPRT_CLOSE_WAIT, &xprt->state); 916632e3bdcSTrond Myklebust if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0) 917632e3bdcSTrond Myklebust schedule_work(&xprt->task_cleanup); 918a246b010SChuck Lever default: 919a246b010SChuck Lever xprt_disconnect(xprt); 920a246b010SChuck Lever } 921a246b010SChuck Lever out: 922a246b010SChuck Lever read_unlock(&sk->sk_callback_lock); 923a246b010SChuck Lever } 924a246b010SChuck Lever 9259903cd1cSChuck Lever /** 926c7b2cae8SChuck Lever * xs_udp_write_space - callback invoked when socket buffer space 927c7b2cae8SChuck Lever * becomes available 9289903cd1cSChuck Lever * @sk: socket whose state has changed 9299903cd1cSChuck Lever * 930a246b010SChuck Lever * Called when more output buffer space is available for this socket. 931a246b010SChuck Lever * We try not to wake our writers until they can make "significant" 932c7b2cae8SChuck Lever * progress, otherwise we'll waste resources thrashing kernel_sendmsg 933a246b010SChuck Lever * with a bunch of small requests. 934a246b010SChuck Lever */ 935c7b2cae8SChuck Lever static void xs_udp_write_space(struct sock *sk) 936a246b010SChuck Lever { 937a246b010SChuck Lever read_lock(&sk->sk_callback_lock); 938c7b2cae8SChuck Lever 939c7b2cae8SChuck Lever /* from net/core/sock.c:sock_def_write_space */ 940c7b2cae8SChuck Lever if (sock_writeable(sk)) { 941c7b2cae8SChuck Lever struct socket *sock; 942c7b2cae8SChuck Lever struct rpc_xprt *xprt; 943c7b2cae8SChuck Lever 944c7b2cae8SChuck Lever if (unlikely(!(sock = sk->sk_socket))) 945a246b010SChuck Lever goto out; 946c7b2cae8SChuck Lever if (unlikely(!(xprt = xprt_from_sock(sk)))) 947c7b2cae8SChuck Lever goto out; 948c7b2cae8SChuck Lever if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags))) 949a246b010SChuck Lever goto out; 950a246b010SChuck Lever 951c7b2cae8SChuck Lever xprt_write_space(xprt); 952a246b010SChuck Lever } 953a246b010SChuck Lever 954c7b2cae8SChuck Lever out: 955c7b2cae8SChuck Lever read_unlock(&sk->sk_callback_lock); 956c7b2cae8SChuck Lever } 957c7b2cae8SChuck Lever 958c7b2cae8SChuck Lever /** 959c7b2cae8SChuck Lever * xs_tcp_write_space - callback invoked when socket buffer space 960c7b2cae8SChuck Lever * becomes available 961c7b2cae8SChuck Lever * @sk: socket whose state has changed 962c7b2cae8SChuck Lever * 963c7b2cae8SChuck Lever * Called when more output buffer space is available for this socket. 964c7b2cae8SChuck Lever * We try not to wake our writers until they can make "significant" 965c7b2cae8SChuck Lever * progress, otherwise we'll waste resources thrashing kernel_sendmsg 966c7b2cae8SChuck Lever * with a bunch of small requests. 967c7b2cae8SChuck Lever */ 968c7b2cae8SChuck Lever static void xs_tcp_write_space(struct sock *sk) 969c7b2cae8SChuck Lever { 970c7b2cae8SChuck Lever read_lock(&sk->sk_callback_lock); 971c7b2cae8SChuck Lever 972c7b2cae8SChuck Lever /* from net/core/stream.c:sk_stream_write_space */ 973c7b2cae8SChuck Lever if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) { 974c7b2cae8SChuck Lever struct socket *sock; 975c7b2cae8SChuck Lever struct rpc_xprt *xprt; 976c7b2cae8SChuck Lever 977c7b2cae8SChuck Lever if (unlikely(!(sock = sk->sk_socket))) 978c7b2cae8SChuck Lever goto out; 979c7b2cae8SChuck Lever if (unlikely(!(xprt = xprt_from_sock(sk)))) 980c7b2cae8SChuck Lever goto out; 981c7b2cae8SChuck Lever if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags))) 982a246b010SChuck Lever goto out; 983a246b010SChuck Lever 984c7b2cae8SChuck Lever xprt_write_space(xprt); 985c7b2cae8SChuck Lever } 986c7b2cae8SChuck Lever 987a246b010SChuck Lever out: 988a246b010SChuck Lever read_unlock(&sk->sk_callback_lock); 989a246b010SChuck Lever } 990a246b010SChuck Lever 991470056c2SChuck Lever static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt) 992a246b010SChuck Lever { 993ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 994ee0ac0c2SChuck Lever struct sock *sk = transport->inet; 995a246b010SChuck Lever 9967c6e066eSChuck Lever if (transport->rcvsize) { 997a246b010SChuck Lever sk->sk_userlocks |= SOCK_RCVBUF_LOCK; 9987c6e066eSChuck Lever sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2; 999a246b010SChuck Lever } 10007c6e066eSChuck Lever if (transport->sndsize) { 1001a246b010SChuck Lever sk->sk_userlocks |= SOCK_SNDBUF_LOCK; 10027c6e066eSChuck Lever sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2; 1003a246b010SChuck Lever sk->sk_write_space(sk); 1004a246b010SChuck Lever } 1005a246b010SChuck Lever } 1006a246b010SChuck Lever 100743118c29SChuck Lever /** 1008470056c2SChuck Lever * xs_udp_set_buffer_size - set send and receive limits 100943118c29SChuck Lever * @xprt: generic transport 1010470056c2SChuck Lever * @sndsize: requested size of send buffer, in bytes 1011470056c2SChuck Lever * @rcvsize: requested size of receive buffer, in bytes 101243118c29SChuck Lever * 1013470056c2SChuck Lever * Set socket send and receive buffer size limits. 101443118c29SChuck Lever */ 1015470056c2SChuck Lever static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize) 101643118c29SChuck Lever { 10177c6e066eSChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 10187c6e066eSChuck Lever 10197c6e066eSChuck Lever transport->sndsize = 0; 1020470056c2SChuck Lever if (sndsize) 10217c6e066eSChuck Lever transport->sndsize = sndsize + 1024; 10227c6e066eSChuck Lever transport->rcvsize = 0; 1023470056c2SChuck Lever if (rcvsize) 10247c6e066eSChuck Lever transport->rcvsize = rcvsize + 1024; 1025470056c2SChuck Lever 1026470056c2SChuck Lever xs_udp_do_set_buffer_size(xprt); 102743118c29SChuck Lever } 102843118c29SChuck Lever 102946c0ee8bSChuck Lever /** 103046c0ee8bSChuck Lever * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport 103146c0ee8bSChuck Lever * @task: task that timed out 103246c0ee8bSChuck Lever * 103346c0ee8bSChuck Lever * Adjust the congestion window after a retransmit timeout has occurred. 103446c0ee8bSChuck Lever */ 103546c0ee8bSChuck Lever static void xs_udp_timer(struct rpc_task *task) 103646c0ee8bSChuck Lever { 103746c0ee8bSChuck Lever xprt_adjust_cwnd(task, -ETIMEDOUT); 103846c0ee8bSChuck Lever } 103946c0ee8bSChuck Lever 1040b85d8806SChuck Lever static unsigned short xs_get_random_port(void) 1041b85d8806SChuck Lever { 1042b85d8806SChuck Lever unsigned short range = xprt_max_resvport - xprt_min_resvport; 1043b85d8806SChuck Lever unsigned short rand = (unsigned short) net_random() % range; 1044b85d8806SChuck Lever return rand + xprt_min_resvport; 1045b85d8806SChuck Lever } 1046b85d8806SChuck Lever 104792200412SChuck Lever /** 104892200412SChuck Lever * xs_set_port - reset the port number in the remote endpoint address 104992200412SChuck Lever * @xprt: generic transport 105092200412SChuck Lever * @port: new port number 105192200412SChuck Lever * 105292200412SChuck Lever */ 105392200412SChuck Lever static void xs_set_port(struct rpc_xprt *xprt, unsigned short port) 105492200412SChuck Lever { 1055c4efcb1dSChuck Lever struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr; 1056c4efcb1dSChuck Lever 105792200412SChuck Lever dprintk("RPC: setting port for xprt %p to %u\n", xprt, port); 1058c4efcb1dSChuck Lever 1059c4efcb1dSChuck Lever sap->sin_port = htons(port); 106092200412SChuck Lever } 106192200412SChuck Lever 1062c8475461SChuck Lever static int xs_bindresvport(struct sock_xprt *transport, struct socket *sock) 1063a246b010SChuck Lever { 1064a246b010SChuck Lever struct sockaddr_in myaddr = { 1065a246b010SChuck Lever .sin_family = AF_INET, 1066a246b010SChuck Lever }; 1067529b33c6SChuck Lever int err; 1068c8475461SChuck Lever unsigned short port = transport->port; 1069a246b010SChuck Lever 1070a246b010SChuck Lever do { 1071a246b010SChuck Lever myaddr.sin_port = htons(port); 1072e6242e92SSridhar Samudrala err = kernel_bind(sock, (struct sockaddr *) &myaddr, 1073a246b010SChuck Lever sizeof(myaddr)); 1074a246b010SChuck Lever if (err == 0) { 1075c8475461SChuck Lever transport->port = port; 10769903cd1cSChuck Lever dprintk("RPC: xs_bindresvport bound to port %u\n", 10779903cd1cSChuck Lever port); 1078a246b010SChuck Lever return 0; 1079a246b010SChuck Lever } 1080529b33c6SChuck Lever if (port <= xprt_min_resvport) 1081529b33c6SChuck Lever port = xprt_max_resvport; 1082529b33c6SChuck Lever else 1083529b33c6SChuck Lever port--; 1084c8475461SChuck Lever } while (err == -EADDRINUSE && port != transport->port); 1085a246b010SChuck Lever 10869903cd1cSChuck Lever dprintk("RPC: can't bind to reserved port (%d).\n", -err); 1087a246b010SChuck Lever return err; 1088a246b010SChuck Lever } 1089a246b010SChuck Lever 10909903cd1cSChuck Lever /** 1091b0d93ad5SChuck Lever * xs_udp_connect_worker - set up a UDP socket 10929903cd1cSChuck Lever * @args: RPC transport to connect 10939903cd1cSChuck Lever * 10949903cd1cSChuck Lever * Invoked by a work queue tasklet. 1095a246b010SChuck Lever */ 1096b0d93ad5SChuck Lever static void xs_udp_connect_worker(void *args) 1097a246b010SChuck Lever { 1098c8475461SChuck Lever struct sock_xprt *transport = (struct sock_xprt *)args; 1099c8475461SChuck Lever struct rpc_xprt *xprt = &transport->xprt; 1100ee0ac0c2SChuck Lever struct socket *sock = transport->sock; 1101b0d93ad5SChuck Lever int err, status = -EIO; 1102a246b010SChuck Lever 1103ec739ef0SChuck Lever if (xprt->shutdown || !xprt_bound(xprt)) 1104a246b010SChuck Lever goto out; 1105a246b010SChuck Lever 1106b0d93ad5SChuck Lever /* Start by resetting any existing state */ 11079903cd1cSChuck Lever xs_close(xprt); 1108b0d93ad5SChuck Lever 1109b0d93ad5SChuck Lever if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) { 1110b0d93ad5SChuck Lever dprintk("RPC: can't create UDP transport socket (%d).\n", -err); 1111a246b010SChuck Lever goto out; 1112a246b010SChuck Lever } 1113a246b010SChuck Lever 1114c8475461SChuck Lever if (xprt->resvport && xs_bindresvport(transport, sock) < 0) { 1115b0d93ad5SChuck Lever sock_release(sock); 1116b0d93ad5SChuck Lever goto out; 1117b0d93ad5SChuck Lever } 1118b0d93ad5SChuck Lever 1119edb267a6SChuck Lever dprintk("RPC: worker connecting xprt %p to address: %s\n", 11207559c7a2SChuck Lever xprt, xprt->address_strings[RPC_DISPLAY_ALL]); 1121edb267a6SChuck Lever 1122ee0ac0c2SChuck Lever if (!transport->inet) { 1123b0d93ad5SChuck Lever struct sock *sk = sock->sk; 1124b0d93ad5SChuck Lever 1125b0d93ad5SChuck Lever write_lock_bh(&sk->sk_callback_lock); 1126b0d93ad5SChuck Lever 1127b0d93ad5SChuck Lever sk->sk_user_data = xprt; 1128314dfd79SChuck Lever transport->old_data_ready = sk->sk_data_ready; 1129314dfd79SChuck Lever transport->old_state_change = sk->sk_state_change; 1130314dfd79SChuck Lever transport->old_write_space = sk->sk_write_space; 1131b0d93ad5SChuck Lever sk->sk_data_ready = xs_udp_data_ready; 1132b0d93ad5SChuck Lever sk->sk_write_space = xs_udp_write_space; 1133b0d93ad5SChuck Lever sk->sk_no_check = UDP_CSUM_NORCV; 1134b079fa7bSTrond Myklebust sk->sk_allocation = GFP_ATOMIC; 1135b0d93ad5SChuck Lever 1136b0d93ad5SChuck Lever xprt_set_connected(xprt); 1137b0d93ad5SChuck Lever 1138b0d93ad5SChuck Lever /* Reset to new socket */ 1139ee0ac0c2SChuck Lever transport->sock = sock; 1140ee0ac0c2SChuck Lever transport->inet = sk; 1141b0d93ad5SChuck Lever 1142b0d93ad5SChuck Lever write_unlock_bh(&sk->sk_callback_lock); 1143b0d93ad5SChuck Lever } 1144470056c2SChuck Lever xs_udp_do_set_buffer_size(xprt); 1145a246b010SChuck Lever status = 0; 1146b0d93ad5SChuck Lever out: 1147b0d93ad5SChuck Lever xprt_wake_pending_tasks(xprt, status); 1148b0d93ad5SChuck Lever xprt_clear_connecting(xprt); 1149b0d93ad5SChuck Lever } 1150b0d93ad5SChuck Lever 11513167e12cSChuck Lever /* 11523167e12cSChuck Lever * We need to preserve the port number so the reply cache on the server can 11533167e12cSChuck Lever * find our cached RPC replies when we get around to reconnecting. 11543167e12cSChuck Lever */ 11553167e12cSChuck Lever static void xs_tcp_reuse_connection(struct rpc_xprt *xprt) 11563167e12cSChuck Lever { 11573167e12cSChuck Lever int result; 1158ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 11593167e12cSChuck Lever struct sockaddr any; 11603167e12cSChuck Lever 11613167e12cSChuck Lever dprintk("RPC: disconnecting xprt %p to reuse port\n", xprt); 11623167e12cSChuck Lever 11633167e12cSChuck Lever /* 11643167e12cSChuck Lever * Disconnect the transport socket by doing a connect operation 11653167e12cSChuck Lever * with AF_UNSPEC. This should return immediately... 11663167e12cSChuck Lever */ 11673167e12cSChuck Lever memset(&any, 0, sizeof(any)); 11683167e12cSChuck Lever any.sa_family = AF_UNSPEC; 1169ee0ac0c2SChuck Lever result = kernel_connect(transport->sock, &any, sizeof(any), 0); 11703167e12cSChuck Lever if (result) 11713167e12cSChuck Lever dprintk("RPC: AF_UNSPEC connect return code %d\n", 11723167e12cSChuck Lever result); 11733167e12cSChuck Lever } 11743167e12cSChuck Lever 1175b0d93ad5SChuck Lever /** 1176b0d93ad5SChuck Lever * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint 1177b0d93ad5SChuck Lever * @args: RPC transport to connect 1178b0d93ad5SChuck Lever * 1179b0d93ad5SChuck Lever * Invoked by a work queue tasklet. 1180b0d93ad5SChuck Lever */ 1181b0d93ad5SChuck Lever static void xs_tcp_connect_worker(void *args) 1182b0d93ad5SChuck Lever { 1183c8475461SChuck Lever struct sock_xprt *transport = (struct sock_xprt *)args; 1184c8475461SChuck Lever struct rpc_xprt *xprt = &transport->xprt; 1185ee0ac0c2SChuck Lever struct socket *sock = transport->sock; 1186b0d93ad5SChuck Lever int err, status = -EIO; 1187b0d93ad5SChuck Lever 1188ec739ef0SChuck Lever if (xprt->shutdown || !xprt_bound(xprt)) 1189a246b010SChuck Lever goto out; 1190a246b010SChuck Lever 1191ee0ac0c2SChuck Lever if (!sock) { 11923167e12cSChuck Lever /* start from scratch */ 1193b0d93ad5SChuck Lever if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) { 1194b0d93ad5SChuck Lever dprintk("RPC: can't create TCP transport socket (%d).\n", -err); 1195b0d93ad5SChuck Lever goto out; 1196b0d93ad5SChuck Lever } 1197b0d93ad5SChuck Lever 1198c8475461SChuck Lever if (xprt->resvport && xs_bindresvport(transport, sock) < 0) { 1199b0d93ad5SChuck Lever sock_release(sock); 1200b0d93ad5SChuck Lever goto out; 1201b0d93ad5SChuck Lever } 12023167e12cSChuck Lever } else 12033167e12cSChuck Lever /* "close" the socket, preserving the local port */ 12043167e12cSChuck Lever xs_tcp_reuse_connection(xprt); 1205b0d93ad5SChuck Lever 1206edb267a6SChuck Lever dprintk("RPC: worker connecting xprt %p to address: %s\n", 12077559c7a2SChuck Lever xprt, xprt->address_strings[RPC_DISPLAY_ALL]); 1208edb267a6SChuck Lever 1209ee0ac0c2SChuck Lever if (!transport->inet) { 1210b0d93ad5SChuck Lever struct sock *sk = sock->sk; 1211b0d93ad5SChuck Lever 1212b0d93ad5SChuck Lever write_lock_bh(&sk->sk_callback_lock); 1213b0d93ad5SChuck Lever 1214b0d93ad5SChuck Lever sk->sk_user_data = xprt; 1215314dfd79SChuck Lever transport->old_data_ready = sk->sk_data_ready; 1216314dfd79SChuck Lever transport->old_state_change = sk->sk_state_change; 1217314dfd79SChuck Lever transport->old_write_space = sk->sk_write_space; 1218b0d93ad5SChuck Lever sk->sk_data_ready = xs_tcp_data_ready; 1219b0d93ad5SChuck Lever sk->sk_state_change = xs_tcp_state_change; 1220b0d93ad5SChuck Lever sk->sk_write_space = xs_tcp_write_space; 1221b079fa7bSTrond Myklebust sk->sk_allocation = GFP_ATOMIC; 12223167e12cSChuck Lever 12233167e12cSChuck Lever /* socket options */ 12243167e12cSChuck Lever sk->sk_userlocks |= SOCK_BINDPORT_LOCK; 12253167e12cSChuck Lever sock_reset_flag(sk, SOCK_LINGER); 12263167e12cSChuck Lever tcp_sk(sk)->linger2 = 0; 12273167e12cSChuck Lever tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF; 1228b0d93ad5SChuck Lever 1229b0d93ad5SChuck Lever xprt_clear_connected(xprt); 1230b0d93ad5SChuck Lever 1231b0d93ad5SChuck Lever /* Reset to new socket */ 1232ee0ac0c2SChuck Lever transport->sock = sock; 1233ee0ac0c2SChuck Lever transport->inet = sk; 1234b0d93ad5SChuck Lever 1235b0d93ad5SChuck Lever write_unlock_bh(&sk->sk_callback_lock); 1236b0d93ad5SChuck Lever } 1237b0d93ad5SChuck Lever 1238b0d93ad5SChuck Lever /* Tell the socket layer to start connecting... */ 1239262ca07dSChuck Lever xprt->stat.connect_count++; 1240262ca07dSChuck Lever xprt->stat.connect_start = jiffies; 1241e6242e92SSridhar Samudrala status = kernel_connect(sock, (struct sockaddr *) &xprt->addr, 1242c4efcb1dSChuck Lever xprt->addrlen, O_NONBLOCK); 1243a246b010SChuck Lever dprintk("RPC: %p connect status %d connected %d sock state %d\n", 1244a246b010SChuck Lever xprt, -status, xprt_connected(xprt), sock->sk->sk_state); 1245a246b010SChuck Lever if (status < 0) { 1246a246b010SChuck Lever switch (status) { 1247a246b010SChuck Lever case -EINPROGRESS: 1248a246b010SChuck Lever case -EALREADY: 1249a246b010SChuck Lever goto out_clear; 12503167e12cSChuck Lever case -ECONNREFUSED: 12513167e12cSChuck Lever case -ECONNRESET: 12523167e12cSChuck Lever /* retry with existing socket, after a delay */ 12533167e12cSChuck Lever break; 12543167e12cSChuck Lever default: 12553167e12cSChuck Lever /* get rid of existing socket, and retry */ 12563167e12cSChuck Lever xs_close(xprt); 12573167e12cSChuck Lever break; 1258a246b010SChuck Lever } 1259a246b010SChuck Lever } 1260a246b010SChuck Lever out: 126144fbac22SChuck Lever xprt_wake_pending_tasks(xprt, status); 1262a246b010SChuck Lever out_clear: 12632226feb6SChuck Lever xprt_clear_connecting(xprt); 1264a246b010SChuck Lever } 1265a246b010SChuck Lever 12669903cd1cSChuck Lever /** 12679903cd1cSChuck Lever * xs_connect - connect a socket to a remote endpoint 12689903cd1cSChuck Lever * @task: address of RPC task that manages state of connect request 12699903cd1cSChuck Lever * 12709903cd1cSChuck Lever * TCP: If the remote end dropped the connection, delay reconnecting. 127103bf4b70SChuck Lever * 127203bf4b70SChuck Lever * UDP socket connects are synchronous, but we use a work queue anyway 127303bf4b70SChuck Lever * to guarantee that even unprivileged user processes can set up a 127403bf4b70SChuck Lever * socket on a privileged port. 127503bf4b70SChuck Lever * 127603bf4b70SChuck Lever * If a UDP socket connect fails, the delay behavior here prevents 127703bf4b70SChuck Lever * retry floods (hard mounts). 12789903cd1cSChuck Lever */ 12799903cd1cSChuck Lever static void xs_connect(struct rpc_task *task) 1280a246b010SChuck Lever { 1281a246b010SChuck Lever struct rpc_xprt *xprt = task->tk_xprt; 1282ee0ac0c2SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1283a246b010SChuck Lever 1284b0d93ad5SChuck Lever if (xprt_test_and_set_connecting(xprt)) 1285b0d93ad5SChuck Lever return; 1286b0d93ad5SChuck Lever 1287ee0ac0c2SChuck Lever if (transport->sock != NULL) { 128803bf4b70SChuck Lever dprintk("RPC: xs_connect delayed xprt %p for %lu seconds\n", 128903bf4b70SChuck Lever xprt, xprt->reestablish_timeout / HZ); 1290c8475461SChuck Lever schedule_delayed_work(&transport->connect_worker, 129103bf4b70SChuck Lever xprt->reestablish_timeout); 129203bf4b70SChuck Lever xprt->reestablish_timeout <<= 1; 129303bf4b70SChuck Lever if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO) 129403bf4b70SChuck Lever xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO; 12959903cd1cSChuck Lever } else { 12969903cd1cSChuck Lever dprintk("RPC: xs_connect scheduled xprt %p\n", xprt); 1297c8475461SChuck Lever schedule_work(&transport->connect_worker); 1298b0d93ad5SChuck Lever 1299a246b010SChuck Lever /* flush_scheduled_work can sleep... */ 1300a246b010SChuck Lever if (!RPC_IS_ASYNC(task)) 1301a246b010SChuck Lever flush_scheduled_work(); 1302a246b010SChuck Lever } 1303a246b010SChuck Lever } 1304a246b010SChuck Lever 1305262ca07dSChuck Lever /** 1306262ca07dSChuck Lever * xs_udp_print_stats - display UDP socket-specifc stats 1307262ca07dSChuck Lever * @xprt: rpc_xprt struct containing statistics 1308262ca07dSChuck Lever * @seq: output file 1309262ca07dSChuck Lever * 1310262ca07dSChuck Lever */ 1311262ca07dSChuck Lever static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 1312262ca07dSChuck Lever { 1313c8475461SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1314c8475461SChuck Lever 1315262ca07dSChuck Lever seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n", 1316c8475461SChuck Lever transport->port, 1317262ca07dSChuck Lever xprt->stat.bind_count, 1318262ca07dSChuck Lever xprt->stat.sends, 1319262ca07dSChuck Lever xprt->stat.recvs, 1320262ca07dSChuck Lever xprt->stat.bad_xids, 1321262ca07dSChuck Lever xprt->stat.req_u, 1322262ca07dSChuck Lever xprt->stat.bklog_u); 1323262ca07dSChuck Lever } 1324262ca07dSChuck Lever 1325262ca07dSChuck Lever /** 1326262ca07dSChuck Lever * xs_tcp_print_stats - display TCP socket-specifc stats 1327262ca07dSChuck Lever * @xprt: rpc_xprt struct containing statistics 1328262ca07dSChuck Lever * @seq: output file 1329262ca07dSChuck Lever * 1330262ca07dSChuck Lever */ 1331262ca07dSChuck Lever static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 1332262ca07dSChuck Lever { 1333c8475461SChuck Lever struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt); 1334262ca07dSChuck Lever long idle_time = 0; 1335262ca07dSChuck Lever 1336262ca07dSChuck Lever if (xprt_connected(xprt)) 1337262ca07dSChuck Lever idle_time = (long)(jiffies - xprt->last_used) / HZ; 1338262ca07dSChuck Lever 1339262ca07dSChuck Lever seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n", 1340c8475461SChuck Lever transport->port, 1341262ca07dSChuck Lever xprt->stat.bind_count, 1342262ca07dSChuck Lever xprt->stat.connect_count, 1343262ca07dSChuck Lever xprt->stat.connect_time, 1344262ca07dSChuck Lever idle_time, 1345262ca07dSChuck Lever xprt->stat.sends, 1346262ca07dSChuck Lever xprt->stat.recvs, 1347262ca07dSChuck Lever xprt->stat.bad_xids, 1348262ca07dSChuck Lever xprt->stat.req_u, 1349262ca07dSChuck Lever xprt->stat.bklog_u); 1350262ca07dSChuck Lever } 1351262ca07dSChuck Lever 1352262965f5SChuck Lever static struct rpc_xprt_ops xs_udp_ops = { 135343118c29SChuck Lever .set_buffer_size = xs_udp_set_buffer_size, 135412a80469SChuck Lever .reserve_xprt = xprt_reserve_xprt_cong, 135549e9a890SChuck Lever .release_xprt = xprt_release_xprt_cong, 1356bbf7c1ddSChuck Lever .rpcbind = rpc_getport, 135792200412SChuck Lever .set_port = xs_set_port, 13589903cd1cSChuck Lever .connect = xs_connect, 135902107148SChuck Lever .buf_alloc = rpc_malloc, 136002107148SChuck Lever .buf_free = rpc_free, 1361262965f5SChuck Lever .send_request = xs_udp_send_request, 1362fe3aca29SChuck Lever .set_retrans_timeout = xprt_set_retrans_timeout_rtt, 136346c0ee8bSChuck Lever .timer = xs_udp_timer, 1364a58dd398SChuck Lever .release_request = xprt_release_rqst_cong, 1365262965f5SChuck Lever .close = xs_close, 1366262965f5SChuck Lever .destroy = xs_destroy, 1367262ca07dSChuck Lever .print_stats = xs_udp_print_stats, 1368262965f5SChuck Lever }; 1369262965f5SChuck Lever 1370262965f5SChuck Lever static struct rpc_xprt_ops xs_tcp_ops = { 137112a80469SChuck Lever .reserve_xprt = xprt_reserve_xprt, 1372e0ab53deSTrond Myklebust .release_xprt = xs_tcp_release_xprt, 1373bbf7c1ddSChuck Lever .rpcbind = rpc_getport, 137492200412SChuck Lever .set_port = xs_set_port, 1375262965f5SChuck Lever .connect = xs_connect, 137602107148SChuck Lever .buf_alloc = rpc_malloc, 137702107148SChuck Lever .buf_free = rpc_free, 1378262965f5SChuck Lever .send_request = xs_tcp_send_request, 1379fe3aca29SChuck Lever .set_retrans_timeout = xprt_set_retrans_timeout_def, 13809903cd1cSChuck Lever .close = xs_close, 13819903cd1cSChuck Lever .destroy = xs_destroy, 1382262ca07dSChuck Lever .print_stats = xs_tcp_print_stats, 1383a246b010SChuck Lever }; 1384a246b010SChuck Lever 1385c8541ecdSChuck Lever static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size) 1386c8541ecdSChuck Lever { 1387c8541ecdSChuck Lever struct rpc_xprt *xprt; 1388ffc2e518SChuck Lever struct sock_xprt *new; 1389c8541ecdSChuck Lever 1390c8541ecdSChuck Lever if (addrlen > sizeof(xprt->addr)) { 1391c8541ecdSChuck Lever dprintk("RPC: xs_setup_xprt: address too large\n"); 1392c8541ecdSChuck Lever return ERR_PTR(-EBADF); 1393c8541ecdSChuck Lever } 1394c8541ecdSChuck Lever 1395ffc2e518SChuck Lever new = kzalloc(sizeof(*new), GFP_KERNEL); 1396ffc2e518SChuck Lever if (new == NULL) { 1397c8541ecdSChuck Lever dprintk("RPC: xs_setup_xprt: couldn't allocate rpc_xprt\n"); 1398c8541ecdSChuck Lever return ERR_PTR(-ENOMEM); 1399c8541ecdSChuck Lever } 1400ffc2e518SChuck Lever xprt = &new->xprt; 1401c8541ecdSChuck Lever 1402c8541ecdSChuck Lever xprt->max_reqs = slot_table_size; 1403c8541ecdSChuck Lever xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL); 1404c8541ecdSChuck Lever if (xprt->slot == NULL) { 1405c8541ecdSChuck Lever kfree(xprt); 1406c8541ecdSChuck Lever dprintk("RPC: xs_setup_xprt: couldn't allocate slot table\n"); 1407c8541ecdSChuck Lever return ERR_PTR(-ENOMEM); 1408c8541ecdSChuck Lever } 1409c8541ecdSChuck Lever 1410c8541ecdSChuck Lever memcpy(&xprt->addr, addr, addrlen); 1411c8541ecdSChuck Lever xprt->addrlen = addrlen; 1412c8475461SChuck Lever new->port = xs_get_random_port(); 1413c8541ecdSChuck Lever 1414c8541ecdSChuck Lever return xprt; 1415c8541ecdSChuck Lever } 1416c8541ecdSChuck Lever 14179903cd1cSChuck Lever /** 14189903cd1cSChuck Lever * xs_setup_udp - Set up transport to use a UDP socket 1419c8541ecdSChuck Lever * @addr: address of remote server 1420c8541ecdSChuck Lever * @addrlen: length of address in bytes 14219903cd1cSChuck Lever * @to: timeout parameters 14229903cd1cSChuck Lever * 14239903cd1cSChuck Lever */ 1424c8541ecdSChuck Lever struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to) 1425a246b010SChuck Lever { 1426c8541ecdSChuck Lever struct rpc_xprt *xprt; 1427c8475461SChuck Lever struct sock_xprt *transport; 1428a246b010SChuck Lever 1429c8541ecdSChuck Lever xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries); 1430c8541ecdSChuck Lever if (IS_ERR(xprt)) 1431c8541ecdSChuck Lever return xprt; 1432c8475461SChuck Lever transport = container_of(xprt, struct sock_xprt, xprt); 1433a246b010SChuck Lever 1434c8541ecdSChuck Lever if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0) 1435ec739ef0SChuck Lever xprt_set_bound(xprt); 1436ec739ef0SChuck Lever 1437ec739ef0SChuck Lever xprt->prot = IPPROTO_UDP; 1438808012fbSChuck Lever xprt->tsh_size = 0; 1439a246b010SChuck Lever /* XXX: header size can vary due to auth type, IPv6, etc. */ 1440a246b010SChuck Lever xprt->max_payload = (1U << 16) - (MAX_HEADER << 3); 1441a246b010SChuck Lever 1442c8475461SChuck Lever INIT_WORK(&transport->connect_worker, xs_udp_connect_worker, transport); 144303bf4b70SChuck Lever xprt->bind_timeout = XS_BIND_TO; 144403bf4b70SChuck Lever xprt->connect_timeout = XS_UDP_CONN_TO; 144503bf4b70SChuck Lever xprt->reestablish_timeout = XS_UDP_REEST_TO; 144603bf4b70SChuck Lever xprt->idle_timeout = XS_IDLE_DISC_TO; 1447a246b010SChuck Lever 1448262965f5SChuck Lever xprt->ops = &xs_udp_ops; 1449a246b010SChuck Lever 1450a246b010SChuck Lever if (to) 1451a246b010SChuck Lever xprt->timeout = *to; 1452a246b010SChuck Lever else 14539903cd1cSChuck Lever xprt_set_timeout(&xprt->timeout, 5, 5 * HZ); 1454a246b010SChuck Lever 1455edb267a6SChuck Lever xs_format_peer_addresses(xprt); 1456edb267a6SChuck Lever dprintk("RPC: set up transport to address %s\n", 14577559c7a2SChuck Lever xprt->address_strings[RPC_DISPLAY_ALL]); 1458edb267a6SChuck Lever 1459c8541ecdSChuck Lever return xprt; 1460a246b010SChuck Lever } 1461a246b010SChuck Lever 14629903cd1cSChuck Lever /** 14639903cd1cSChuck Lever * xs_setup_tcp - Set up transport to use a TCP socket 1464c8541ecdSChuck Lever * @addr: address of remote server 1465c8541ecdSChuck Lever * @addrlen: length of address in bytes 14669903cd1cSChuck Lever * @to: timeout parameters 14679903cd1cSChuck Lever * 14689903cd1cSChuck Lever */ 1469c8541ecdSChuck Lever struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to) 1470a246b010SChuck Lever { 1471c8541ecdSChuck Lever struct rpc_xprt *xprt; 1472c8475461SChuck Lever struct sock_xprt *transport; 1473a246b010SChuck Lever 1474c8541ecdSChuck Lever xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries); 1475c8541ecdSChuck Lever if (IS_ERR(xprt)) 1476c8541ecdSChuck Lever return xprt; 1477c8475461SChuck Lever transport = container_of(xprt, struct sock_xprt, xprt); 1478a246b010SChuck Lever 1479c8541ecdSChuck Lever if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0) 1480ec739ef0SChuck Lever xprt_set_bound(xprt); 1481ec739ef0SChuck Lever 1482ec739ef0SChuck Lever xprt->prot = IPPROTO_TCP; 1483808012fbSChuck Lever xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32); 1484808012fbSChuck Lever xprt->max_payload = RPC_MAX_FRAGMENT_SIZE; 1485a246b010SChuck Lever 1486c8475461SChuck Lever INIT_WORK(&transport->connect_worker, xs_tcp_connect_worker, transport); 148703bf4b70SChuck Lever xprt->bind_timeout = XS_BIND_TO; 148803bf4b70SChuck Lever xprt->connect_timeout = XS_TCP_CONN_TO; 148903bf4b70SChuck Lever xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO; 149003bf4b70SChuck Lever xprt->idle_timeout = XS_IDLE_DISC_TO; 1491a246b010SChuck Lever 1492262965f5SChuck Lever xprt->ops = &xs_tcp_ops; 1493a246b010SChuck Lever 1494a246b010SChuck Lever if (to) 1495a246b010SChuck Lever xprt->timeout = *to; 1496a246b010SChuck Lever else 14979903cd1cSChuck Lever xprt_set_timeout(&xprt->timeout, 2, 60 * HZ); 1498a246b010SChuck Lever 1499edb267a6SChuck Lever xs_format_peer_addresses(xprt); 1500edb267a6SChuck Lever dprintk("RPC: set up transport to address %s\n", 15017559c7a2SChuck Lever xprt->address_strings[RPC_DISPLAY_ALL]); 1502edb267a6SChuck Lever 1503c8541ecdSChuck Lever return xprt; 1504a246b010SChuck Lever } 1505