*
* Client-side transport implementation for sockets.
*
- * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com>
- * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com>
+ * TCP callback races fixes (C) 1998 Red Hat
+ * TCP send fixes (C) 1998 Red Hat
* TCP NFS related read + write fixes
* (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
*
#endif
-/*
- * How many times to try sending a request on a socket before waiting
- * for the socket buffer to clear.
- */
-#define XS_SENDMSG_RETRY (10U)
-
/*
* Time out for an RPC UDP socket connect. UDP socket connects are
* synchronous, but we set a timeout anyway in case of resource
return (struct sockaddr_in6 *) &xprt->addr;
}
-static void xs_format_ipv4_peer_addresses(struct rpc_xprt *xprt)
+static void xs_format_ipv4_peer_addresses(struct rpc_xprt *xprt,
+ const char *protocol,
+ const char *netid)
{
struct sockaddr_in *addr = xs_addr_in(xprt);
char *buf;
}
xprt->address_strings[RPC_DISPLAY_PORT] = buf;
- buf = kzalloc(8, GFP_KERNEL);
- if (buf) {
- if (xprt->prot == IPPROTO_UDP)
- snprintf(buf, 8, "udp");
- else
- snprintf(buf, 8, "tcp");
- }
- xprt->address_strings[RPC_DISPLAY_PROTO] = buf;
+ xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
buf = kzalloc(48, GFP_KERNEL);
if (buf) {
snprintf(buf, 48, "addr="NIPQUAD_FMT" port=%u proto=%s",
NIPQUAD(addr->sin_addr.s_addr),
ntohs(addr->sin_port),
- xprt->prot == IPPROTO_UDP ? "udp" : "tcp");
+ protocol);
}
xprt->address_strings[RPC_DISPLAY_ALL] = buf;
}
xprt->address_strings[RPC_DISPLAY_UNIVERSAL_ADDR] = buf;
- xprt->address_strings[RPC_DISPLAY_NETID] =
- kstrdup(xprt->prot == IPPROTO_UDP ?
- RPCBIND_NETID_UDP : RPCBIND_NETID_TCP, GFP_KERNEL);
+ xprt->address_strings[RPC_DISPLAY_NETID] = netid;
}
-static void xs_format_ipv6_peer_addresses(struct rpc_xprt *xprt)
+static void xs_format_ipv6_peer_addresses(struct rpc_xprt *xprt,
+ const char *protocol,
+ const char *netid)
{
struct sockaddr_in6 *addr = xs_addr_in6(xprt);
char *buf;
buf = kzalloc(40, GFP_KERNEL);
if (buf) {
- snprintf(buf, 40, NIP6_FMT,
- NIP6(addr->sin6_addr));
+ snprintf(buf, 40, "%p6",&addr->sin6_addr);
}
xprt->address_strings[RPC_DISPLAY_ADDR] = buf;
}
xprt->address_strings[RPC_DISPLAY_PORT] = buf;
- buf = kzalloc(8, GFP_KERNEL);
- if (buf) {
- if (xprt->prot == IPPROTO_UDP)
- snprintf(buf, 8, "udp");
- else
- snprintf(buf, 8, "tcp");
- }
- xprt->address_strings[RPC_DISPLAY_PROTO] = buf;
+ xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
buf = kzalloc(64, GFP_KERNEL);
if (buf) {
- snprintf(buf, 64, "addr="NIP6_FMT" port=%u proto=%s",
- NIP6(addr->sin6_addr),
+ snprintf(buf, 64, "addr=%p6 port=%u proto=%s",
+ &addr->sin6_addr,
ntohs(addr->sin6_port),
- xprt->prot == IPPROTO_UDP ? "udp" : "tcp");
+ protocol);
}
xprt->address_strings[RPC_DISPLAY_ALL] = buf;
buf = kzalloc(36, GFP_KERNEL);
- if (buf) {
- snprintf(buf, 36, NIP6_SEQFMT,
- NIP6(addr->sin6_addr));
- }
+ if (buf)
+ snprintf(buf, 36, "%#p6", &addr->sin6_addr);
+
xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = buf;
buf = kzalloc(8, GFP_KERNEL);
buf = kzalloc(50, GFP_KERNEL);
if (buf) {
- snprintf(buf, 50, NIP6_FMT".%u.%u",
- NIP6(addr->sin6_addr),
- ntohs(addr->sin6_port) >> 8,
- ntohs(addr->sin6_port) & 0xff);
+ snprintf(buf, 50, "%p6.%u.%u",
+ &addr->sin6_addr,
+ ntohs(addr->sin6_port) >> 8,
+ ntohs(addr->sin6_port) & 0xff);
}
xprt->address_strings[RPC_DISPLAY_UNIVERSAL_ADDR] = buf;
- xprt->address_strings[RPC_DISPLAY_NETID] =
- kstrdup(xprt->prot == IPPROTO_UDP ?
- RPCBIND_NETID_UDP6 : RPCBIND_NETID_TCP6, GFP_KERNEL);
+ xprt->address_strings[RPC_DISPLAY_NETID] = netid;
}
static void xs_free_peer_addresses(struct rpc_xprt *xprt)
{
- int i;
+ unsigned int i;
for (i = 0; i < RPC_DISPLAY_MAX; i++)
- kfree(xprt->address_strings[i]);
+ switch (i) {
+ case RPC_DISPLAY_PROTO:
+ case RPC_DISPLAY_NETID:
+ continue;
+ default:
+ kfree(xprt->address_strings[i]);
+ }
}
#define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL)
return sent;
}
+static void xs_nospace_callback(struct rpc_task *task)
+{
+ struct sock_xprt *transport = container_of(task->tk_rqstp->rq_xprt, struct sock_xprt, xprt);
+
+ transport->inet->sk_write_pending--;
+ clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
+}
+
/**
* xs_nospace - place task on wait queue if transmit was incomplete
* @task: task to put to sleep
task->tk_pid, req->rq_slen - req->rq_bytes_sent,
req->rq_slen);
- if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
- /* Protect against races with write_space */
- spin_lock_bh(&xprt->transport_lock);
-
- /* Don't race with disconnect */
- if (!xprt_connected(xprt))
- task->tk_status = -ENOTCONN;
- else if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
- xprt_wait_for_buffer_space(task);
+ /* Protect against races with write_space */
+ spin_lock_bh(&xprt->transport_lock);
+
+ /* Don't race with disconnect */
+ if (xprt_connected(xprt)) {
+ if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
+ /*
+ * Notify TCP that we're limited by the application
+ * window size
+ */
+ set_bit(SOCK_NOSPACE, &transport->sock->flags);
+ transport->inet->sk_write_pending++;
+ /* ...and wait for more buffer space */
+ xprt_wait_for_buffer_space(task, xs_nospace_callback);
+ }
+ } else {
+ clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
+ task->tk_status = -ENOTCONN;
+ }
- spin_unlock_bh(&xprt->transport_lock);
- } else
- /* Keep holding the socket if it is blocked */
- rpc_delay(task, HZ>>4);
+ spin_unlock_bh(&xprt->transport_lock);
}
/**
req->rq_svec->iov_base,
req->rq_svec->iov_len);
- req->rq_xtime = jiffies;
status = xs_sendpages(transport->sock,
xs_addr(xprt),
xprt->addrlen, xdr,
}
switch (status) {
+ case -EAGAIN:
+ xs_nospace(task);
+ break;
case -ENETUNREACH:
case -EPIPE:
case -ECONNREFUSED:
/* When the server has died, an ICMP port unreachable message
* prompts ECONNREFUSED. */
- break;
- case -EAGAIN:
- xs_nospace(task);
+ clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
break;
default:
+ clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
dprintk("RPC: sendmsg returned unrecognized error %d\n",
-status);
- break;
}
return status;
struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
struct xdr_buf *xdr = &req->rq_snd_buf;
int status;
- unsigned int retry = 0;
xs_encode_tcp_record_marker(&req->rq_snd_buf);
* to cope with writespace callbacks arriving _after_ we have
* called sendmsg(). */
while (1) {
- req->rq_xtime = jiffies;
status = xs_sendpages(transport->sock,
NULL, 0, xdr, req->rq_bytes_sent);
return 0;
}
+ if (status != 0)
+ continue;
status = -EAGAIN;
- if (retry++ > XS_SENDMSG_RETRY)
- break;
+ break;
}
switch (status) {
case -ENOTCONN:
case -EPIPE:
status = -ENOTCONN;
+ clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
break;
default:
dprintk("RPC: sendmsg returned unrecognized error %d\n",
-status);
+ clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
xs_tcp_shutdown(xprt);
- break;
}
return status;
{
struct rpc_xprt *xprt;
read_descriptor_t rd_desc;
+ int read;
dprintk("RPC: xs_tcp_data_ready...\n");
/* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
rd_desc.arg.data = xprt;
- rd_desc.count = 65536;
- tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
+ do {
+ rd_desc.count = 65536;
+ read = tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
+ } while (read > 0);
out:
read_unlock(&sk->sk_callback_lock);
}
break;
case TCP_FIN_WAIT1:
/* The client initiated a shutdown of the socket */
+ xprt->connect_cookie++;
xprt->reestablish_timeout = 0;
set_bit(XPRT_CLOSING, &xprt->state);
smp_mb__before_clear_bit();
set_bit(XPRT_CLOSING, &xprt->state);
xprt_force_disconnect(xprt);
case TCP_SYN_SENT:
+ xprt->connect_cookie++;
case TCP_CLOSING:
/*
* If the server closed down the connection, make sure that
if (unlikely(!(sock = sk->sk_socket)))
goto out;
+ clear_bit(SOCK_NOSPACE, &sock->flags);
+
if (unlikely(!(xprt = xprt_from_sock(sk))))
goto out;
- if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
+ if (test_and_clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags) == 0)
goto out;
xprt_write_space(xprt);
if (unlikely(!(sock = sk->sk_socket)))
goto out;
+ clear_bit(SOCK_NOSPACE, &sock->flags);
+
if (unlikely(!(xprt = xprt_from_sock(sk))))
goto out;
- if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
+ if (test_and_clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags) == 0)
goto out;
xprt_write_space(xprt);
nloop++;
} while (err == -EADDRINUSE && nloop != 2);
dprintk("RPC: %s "NIPQUAD_FMT":%u: %s (%d)\n",
- __FUNCTION__, NIPQUAD(myaddr.sin_addr),
+ __func__, NIPQUAD(myaddr.sin_addr),
port, err ? "failed" : "ok", err);
return err;
}
if (port > last)
nloop++;
} while (err == -EADDRINUSE && nloop != 2);
- dprintk("RPC: xs_bind6 "NIP6_FMT":%u: %s (%d)\n",
- NIP6(myaddr.sin6_addr), port, err ? "failed" : "ok", err);
+ dprintk("RPC: xs_bind6 %p6:%u: %s (%d)\n",
+ &myaddr.sin6_addr, port, err ? "failed" : "ok", err);
return err;
}
struct sockaddr *addr = args->dstaddr;
struct rpc_xprt *xprt;
struct sock_xprt *transport;
- const struct rpc_timeout *timeo = &xs_udp_default_timeout;
xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries);
if (IS_ERR(xprt))
xprt->ops = &xs_udp_ops;
- if (args->timeout != NULL)
- timeo = args->timeout;
- memcpy(&xprt->timeout, timeo, sizeof(xprt->timeout));
+ xprt->timeout = &xs_udp_default_timeout;
switch (addr->sa_family) {
case AF_INET:
INIT_DELAYED_WORK(&transport->connect_worker,
xs_udp_connect_worker4);
- xs_format_ipv4_peer_addresses(xprt);
+ xs_format_ipv4_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
break;
case AF_INET6:
if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
INIT_DELAYED_WORK(&transport->connect_worker,
xs_udp_connect_worker6);
- xs_format_ipv6_peer_addresses(xprt);
+ xs_format_ipv6_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
break;
default:
kfree(xprt);
struct sockaddr *addr = args->dstaddr;
struct rpc_xprt *xprt;
struct sock_xprt *transport;
- const struct rpc_timeout *timeo = &xs_tcp_default_timeout;
xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries);
if (IS_ERR(xprt))
xprt->idle_timeout = XS_IDLE_DISC_TO;
xprt->ops = &xs_tcp_ops;
-
- if (args->timeout != NULL)
- timeo = args->timeout;
- memcpy(&xprt->timeout, timeo, sizeof(xprt->timeout));
+ xprt->timeout = &xs_tcp_default_timeout;
switch (addr->sa_family) {
case AF_INET:
xprt_set_bound(xprt);
INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker4);
- xs_format_ipv4_peer_addresses(xprt);
+ xs_format_ipv4_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
break;
case AF_INET6:
if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
xprt_set_bound(xprt);
INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker6);
- xs_format_ipv6_peer_addresses(xprt);
+ xs_format_ipv6_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
break;
default:
kfree(xprt);
Code Review / linux-2.6.git / blobdiff