*
* 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
* exhaustion on the local host.
void (*old_data_ready)(struct sock *, int);
void (*old_state_change)(struct sock *);
void (*old_write_space)(struct sock *);
+ void (*old_error_report)(struct sock *);
};
/*
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;
buf = kzalloc(20, GFP_KERNEL);
if (buf) {
- snprintf(buf, 20, NIPQUAD_FMT,
- NIPQUAD(addr->sin_addr.s_addr));
+ snprintf(buf, 20, "%pI4", &addr->sin_addr.s_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(48, GFP_KERNEL);
if (buf) {
- snprintf(buf, 48, "addr="NIPQUAD_FMT" port=%u proto=%s",
- NIPQUAD(addr->sin_addr.s_addr),
+ snprintf(buf, 48, "addr=%pI4 port=%u proto=%s",
+ &addr->sin_addr.s_addr,
ntohs(addr->sin_port),
- xprt->prot == IPPROTO_UDP ? "udp" : "tcp");
+ protocol);
}
xprt->address_strings[RPC_DISPLAY_ALL] = buf;
buf = kzalloc(30, GFP_KERNEL);
if (buf) {
- snprintf(buf, 30, NIPQUAD_FMT".%u.%u",
- NIPQUAD(addr->sin_addr.s_addr),
+ snprintf(buf, 30, "%pI4.%u.%u",
+ &addr->sin_addr.s_addr,
ntohs(addr->sin_port) >> 8,
ntohs(addr->sin_port) & 0xff);
}
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, "%pI6",&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=%pI6 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, "%pi6", &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, "%pI6.%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)
int err, sent = 0;
if (unlikely(!sock))
- return -ENOTCONN;
+ return -ENOTSOCK;
clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
if (base != 0) {
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;
+ if (!xprt_bound(xprt))
+ return -ENOTCONN;
status = xs_sendpages(transport->sock,
xs_addr(xprt),
xprt->addrlen, xdr,
}
switch (status) {
+ case -ENOTSOCK:
+ status = -ENOTCONN;
+ /* Should we call xs_close() here? */
+ break;
+ 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 -ENOTSOCK:
+ status = -ENOTCONN;
+ /* Should we call xs_close() here? */
+ break;
case -EAGAIN:
xs_nospace(task);
break;
- case -ECONNREFUSED:
case -ECONNRESET:
+ xs_tcp_shutdown(xprt);
+ case -ECONNREFUSED:
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;
xprt_release_xprt(xprt, task);
}
+static void xs_save_old_callbacks(struct sock_xprt *transport, struct sock *sk)
+{
+ transport->old_data_ready = sk->sk_data_ready;
+ transport->old_state_change = sk->sk_state_change;
+ transport->old_write_space = sk->sk_write_space;
+ transport->old_error_report = sk->sk_error_report;
+}
+
+static void xs_restore_old_callbacks(struct sock_xprt *transport, struct sock *sk)
+{
+ sk->sk_data_ready = transport->old_data_ready;
+ sk->sk_state_change = transport->old_state_change;
+ sk->sk_write_space = transport->old_write_space;
+ sk->sk_error_report = transport->old_error_report;
+}
+
/**
* xs_close - close a socket
* @xprt: transport
transport->sock = NULL;
sk->sk_user_data = NULL;
- sk->sk_data_ready = transport->old_data_ready;
- sk->sk_state_change = transport->old_state_change;
- sk->sk_write_space = transport->old_write_space;
+
+ xs_restore_old_callbacks(transport, sk);
write_unlock_bh(&sk->sk_callback_lock);
sk->sk_no_check = 0;
clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
clear_bit(XPRT_CLOSING, &xprt->state);
smp_mb__after_clear_bit();
- xprt_disconnect(xprt);
+ xprt_disconnect_done(xprt);
}
/**
/* Sanity check of the record length */
if (unlikely(transport->tcp_reclen < 4)) {
dprintk("RPC: invalid TCP record fragment length\n");
- xprt_disconnect(xprt);
+ xprt_force_disconnect(xprt);
return;
}
dprintk("RPC: reading TCP record fragment of length %d\n",
{
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);
}
transport->tcp_flags =
TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
- xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
xprt_wake_pending_tasks(xprt, 0);
}
spin_unlock_bh(&xprt->transport_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();
clear_bit(XPRT_CONNECTED, &xprt->state);
/* The server initiated a shutdown of the socket */
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
+ * we back off before reconnecting
+ */
+ if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
+ xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
break;
case TCP_LAST_ACK:
smp_mb__before_clear_bit();
clear_bit(XPRT_CLOSING, &xprt->state);
smp_mb__after_clear_bit();
/* Mark transport as closed and wake up all pending tasks */
- xprt_disconnect(xprt);
+ xprt_disconnect_done(xprt);
}
out:
read_unlock(&sk->sk_callback_lock);
}
/**
+ * xs_tcp_error_report - callback mainly for catching RST events
+ * @sk: socket
+ */
+static void xs_tcp_error_report(struct sock *sk)
+{
+ struct rpc_xprt *xprt;
+
+ read_lock(&sk->sk_callback_lock);
+ if (sk->sk_err != ECONNRESET || sk->sk_state != TCP_ESTABLISHED)
+ goto out;
+ if (!(xprt = xprt_from_sock(sk)))
+ goto out;
+ dprintk("RPC: %s client %p...\n"
+ "RPC: error %d\n",
+ __func__, xprt, sk->sk_err);
+
+ xprt_force_disconnect(xprt);
+out:
+ read_unlock(&sk->sk_callback_lock);
+}
+
+/**
* xs_udp_write_space - callback invoked when socket buffer space
* becomes available
* @sk: socket whose state has changed
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);
if (port > last)
nloop++;
} while (err == -EADDRINUSE && nloop != 2);
- dprintk("RPC: %s "NIPQUAD_FMT":%u: %s (%d)\n",
- __FUNCTION__, NIPQUAD(myaddr.sin_addr),
+ dprintk("RPC: %s %pI4:%u: %s (%d)\n",
+ __func__, &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 %pI6:%u: %s (%d)\n",
+ &myaddr.sin6_addr, port, err ? "failed" : "ok", err);
return err;
}
write_lock_bh(&sk->sk_callback_lock);
+ xs_save_old_callbacks(transport, sk);
+
sk->sk_user_data = xprt;
- transport->old_data_ready = sk->sk_data_ready;
- transport->old_state_change = sk->sk_state_change;
- transport->old_write_space = sk->sk_write_space;
sk->sk_data_ready = xs_udp_data_ready;
sk->sk_write_space = xs_udp_write_space;
sk->sk_no_check = UDP_CSUM_NORCV;
struct socket *sock = transport->sock;
int err, status = -EIO;
- if (xprt->shutdown || !xprt_bound(xprt))
+ if (xprt->shutdown)
goto out;
/* Start by resetting any existing state */
struct socket *sock = transport->sock;
int err, status = -EIO;
- if (xprt->shutdown || !xprt_bound(xprt))
+ if (xprt->shutdown)
goto out;
/* Start by resetting any existing state */
write_lock_bh(&sk->sk_callback_lock);
+ xs_save_old_callbacks(transport, sk);
+
sk->sk_user_data = xprt;
- transport->old_data_ready = sk->sk_data_ready;
- transport->old_state_change = sk->sk_state_change;
- transport->old_write_space = sk->sk_write_space;
sk->sk_data_ready = xs_tcp_data_ready;
sk->sk_state_change = xs_tcp_state_change;
sk->sk_write_space = xs_tcp_write_space;
+ sk->sk_error_report = xs_tcp_error_report;
sk->sk_allocation = GFP_ATOMIC;
/* socket options */
write_unlock_bh(&sk->sk_callback_lock);
}
+ if (!xprt_bound(xprt))
+ return -ENOTCONN;
+
/* Tell the socket layer to start connecting... */
xprt->stat.connect_count++;
xprt->stat.connect_start = jiffies;
struct socket *sock = transport->sock;
int err, status = -EIO;
- if (xprt->shutdown || !xprt_bound(xprt))
+ if (xprt->shutdown)
goto out;
if (!sock) {
struct socket *sock = transport->sock;
int err, status = -EIO;
- if (xprt->shutdown || !xprt_bound(xprt))
+ if (xprt->shutdown)
goto out;
if (!sock) {
return xprt;
}
+static const struct rpc_timeout xs_udp_default_timeout = {
+ .to_initval = 5 * HZ,
+ .to_maxval = 30 * HZ,
+ .to_increment = 5 * HZ,
+ .to_retries = 5,
+};
+
/**
* xs_setup_udp - Set up transport to use a UDP socket
* @args: rpc transport creation arguments
xprt->ops = &xs_udp_ops;
- if (args->timeout)
- xprt->timeout = *args->timeout;
- else
- xprt_set_timeout(&xprt->timeout, 5, 5 * HZ);
+ 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);
return ERR_PTR(-EINVAL);
}
+static const struct rpc_timeout xs_tcp_default_timeout = {
+ .to_initval = 60 * HZ,
+ .to_maxval = 60 * HZ,
+ .to_retries = 2,
+};
+
/**
* xs_setup_tcp - Set up transport to use a TCP socket
* @args: rpc transport creation arguments
xprt->idle_timeout = XS_IDLE_DISC_TO;
xprt->ops = &xs_tcp_ops;
-
- if (args->timeout)
- xprt->timeout = *args->timeout;
- else
- xprt_set_timeout(&xprt->timeout, 2, 60 * HZ);
+ 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);
git://ftp.safe.ca / safe / jmp / linux-2.6 / blobdiff