[net-next-2.6.git] / net / rds / tcp_listen.c

/*
 * Copyright (c) 2006 Oracle.  All rights reserved.
 *
 * This software is available to you under a choice of one of two
 * licenses.  You may choose to be licensed under the terms of the GNU
 * General Public License (GPL) Version 2, available from the file
 * COPYING in the main directory of this source tree, or the
 * OpenIB.org BSD license below:
 *
 *     Redistribution and use in source and binary forms, with or
 *     without modification, are permitted provided that the following
 *     conditions are met:
 *
 *      - Redistributions of source code must retain the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer.
 *
 *      - Redistributions in binary form must reproduce the above
 *        copyright notice, this list of conditions and the following
 *        disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 *
 */
#include <linux/kernel.h>
#include <linux/in.h>
#include <net/tcp.h>

#include "rds.h"
#include "tcp.h"

/*
 * cheesy, but simple..
 */
static void rds_tcp_accept_worker(struct work_struct *work);
static DECLARE_WORK(rds_tcp_listen_work, rds_tcp_accept_worker);
static struct socket *rds_tcp_listen_sock;

static int rds_tcp_accept_one(struct socket *sock)
{
	struct socket *new_sock = NULL;
	struct rds_connection *conn;
	int ret;
	struct inet_sock *inet;

	ret = sock_create_lite(sock->sk->sk_family, sock->sk->sk_type,
			       sock->sk->sk_protocol, &new_sock);
	if (ret)
		goto out;

	new_sock->type = sock->type;
	new_sock->ops = sock->ops;
	ret = sock->ops->accept(sock, new_sock, O_NONBLOCK);
	if (ret < 0)
		goto out;

	rds_tcp_tune(new_sock);

	inet = inet_sk(new_sock->sk);

	rdsdebug("accepted tcp %u.%u.%u.%u:%u -> %u.%u.%u.%u:%u\n",
		  NIPQUAD(inet->inet_saddr), ntohs(inet->inet_sport),
		  NIPQUAD(inet->inet_daddr), ntohs(inet->inet_dport));

	conn = rds_conn_create(inet->inet_saddr, inet->inet_daddr,
			       &rds_tcp_transport, GFP_KERNEL);
	if (IS_ERR(conn)) {
		ret = PTR_ERR(conn);
		goto out;
	}

	/*
	 * see the comment above rds_queue_delayed_reconnect()
	 */
	if (!rds_conn_transition(conn, RDS_CONN_DOWN, RDS_CONN_CONNECTING)) {
		if (rds_conn_state(conn) == RDS_CONN_UP)
			rds_tcp_stats_inc(s_tcp_listen_closed_stale);
		else
			rds_tcp_stats_inc(s_tcp_connect_raced);
		rds_conn_drop(conn);
		ret = 0;
		goto out;
	}

	rds_tcp_set_callbacks(new_sock, conn);
	rds_connect_complete(conn);
	new_sock = NULL;
	ret = 0;

out:
	if (new_sock)
		sock_release(new_sock);
	return ret;
}

static void rds_tcp_accept_worker(struct work_struct *work)
{
	while (rds_tcp_accept_one(rds_tcp_listen_sock) == 0)
		cond_resched();
}

void rds_tcp_listen_data_ready(struct sock *sk, int bytes)
{
	void (*ready)(struct sock *sk, int bytes);

	rdsdebug("listen data ready sk %p\n", sk);

	read_lock(&sk->sk_callback_lock);
	ready = sk->sk_user_data;
	if (ready == NULL) { /* check for teardown race */
		ready = sk->sk_data_ready;
		goto out;
	}

	/*
	 * ->sk_data_ready is also called for a newly established child socket
	 * before it has been accepted and the accepter has set up their
	 * data_ready.. we only want to queue listen work for our listening
	 * socket
	 */
	if (sk->sk_state == TCP_LISTEN)
		queue_work(rds_wq, &rds_tcp_listen_work);

out:
	read_unlock(&sk->sk_callback_lock);
	ready(sk, bytes);
}

int __init rds_tcp_listen_init(void)
{
	struct sockaddr_in sin;
	struct socket *sock = NULL;
	int ret;

	ret = sock_create(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
	if (ret < 0)
		goto out;

	sock->sk->sk_reuse = 1;
	rds_tcp_nonagle(sock);

	write_lock_bh(&sock->sk->sk_callback_lock);
	sock->sk->sk_user_data = sock->sk->sk_data_ready;
	sock->sk->sk_data_ready = rds_tcp_listen_data_ready;
	write_unlock_bh(&sock->sk->sk_callback_lock);

	sin.sin_family = PF_INET,
	sin.sin_addr.s_addr = (__force u32)htonl(INADDR_ANY);
	sin.sin_port = (__force u16)htons(RDS_TCP_PORT);

	ret = sock->ops->bind(sock, (struct sockaddr *)&sin, sizeof(sin));
	if (ret < 0)
		goto out;

	ret = sock->ops->listen(sock, 64);
	if (ret < 0)
		goto out;

	rds_tcp_listen_sock = sock;
	sock = NULL;
out:
	if (sock)
		sock_release(sock);
	return ret;
}

void rds_tcp_listen_stop(void)
{
	struct socket *sock = rds_tcp_listen_sock;
	struct sock *sk;

	if (sock == NULL)
		return;

	sk = sock->sk;

	/* serialize with and prevent further callbacks */
	lock_sock(sk);
	write_lock_bh(&sk->sk_callback_lock);
	if (sk->sk_user_data) {
		sk->sk_data_ready = sk->sk_user_data;
		sk->sk_user_data = NULL;
	}
	write_unlock_bh(&sk->sk_callback_lock);
	release_sock(sk);

	/* wait for accepts to stop and close the socket */
	flush_workqueue(rds_wq);
	sock_release(sock);
	rds_tcp_listen_sock = NULL;
}
Commit	Line	Data
70041088 AG	1	/*
	2	* Copyright (c) 2006 Oracle. All rights reserved.
	3	*
	4	* This software is available to you under a choice of one of two
	5	* licenses. You may choose to be licensed under the terms of the GNU
	6	* General Public License (GPL) Version 2, available from the file
	7	* COPYING in the main directory of this source tree, or the
	8	* OpenIB.org BSD license below:
	9	*
	10	* Redistribution and use in source and binary forms, with or
	11	* without modification, are permitted provided that the following
	12	* conditions are met:
	13	*
	14	* - Redistributions of source code must retain the above
	15	* copyright notice, this list of conditions and the following
	16	* disclaimer.
	17	*
	18	* - Redistributions in binary form must reproduce the above
	19	* copyright notice, this list of conditions and the following
	20	* disclaimer in the documentation and/or other materials
	21	* provided with the distribution.
	22	*
	23	* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
	24	* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
	25	* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
	26	* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
	27	* BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
	28	* ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
	29	* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
	30	* SOFTWARE.
	31	*
	32	*/
	33	#include <linux/kernel.h>
	34	#include <linux/in.h>
	35	#include <net/tcp.h>
	36
	37	#include "rds.h"
	38	#include "tcp.h"
	39
	40	/*
	41	* cheesy, but simple..
	42	*/
	43	static void rds_tcp_accept_worker(struct work_struct *work);
	44	static DECLARE_WORK(rds_tcp_listen_work, rds_tcp_accept_worker);
	45	static struct socket *rds_tcp_listen_sock;
	46
	47	static int rds_tcp_accept_one(struct socket *sock)
	48	{
	49	struct socket *new_sock = NULL;
	50	struct rds_connection *conn;
	51	int ret;
	52	struct inet_sock *inet;
	53
	54	ret = sock_create_lite(sock->sk->sk_family, sock->sk->sk_type,
	55	sock->sk->sk_protocol, &new_sock);
	56	if (ret)
	57	goto out;
	58
	59	new_sock->type = sock->type;
	60	new_sock->ops = sock->ops;
	61	ret = sock->ops->accept(sock, new_sock, O_NONBLOCK);
	62	if (ret < 0)
	63	goto out;
	64
65	rds_tcp_tune(new_sock);
66
67	inet = inet_sk(new_sock->sk);
68
69	rdsdebug("accepted tcp %u.%u.%u.%u:%u -> %u.%u.%u.%u:%u\n",
c720c7e8 ED	70	NIPQUAD(inet->inet_saddr), ntohs(inet->inet_sport),
c720c7e8 ED	71	NIPQUAD(inet->inet_daddr), ntohs(inet->inet_dport));
70041088	72
c720c7e8 ED	73	conn = rds_conn_create(inet->inet_saddr, inet->inet_daddr,
c720c7e8 ED	74	&rds_tcp_transport, GFP_KERNEL);
70041088 AG	75	if (IS_ERR(conn)) {
	76	ret = PTR_ERR(conn);
	77	goto out;
	78	}
	79
	80	/*
	81	* see the comment above rds_queue_delayed_reconnect()
	82	*/
	83	if (!rds_conn_transition(conn, RDS_CONN_DOWN, RDS_CONN_CONNECTING)) {
	84	if (rds_conn_state(conn) == RDS_CONN_UP)
	85	rds_tcp_stats_inc(s_tcp_listen_closed_stale);
	86	else
	87	rds_tcp_stats_inc(s_tcp_connect_raced);
	88	rds_conn_drop(conn);
	89	ret = 0;
	90	goto out;
	91	}
	92
	93	rds_tcp_set_callbacks(new_sock, conn);
	94	rds_connect_complete(conn);
	95	new_sock = NULL;
	96	ret = 0;
	97
	98	out:
	99	if (new_sock)
	100	sock_release(new_sock);
	101	return ret;
	102	}
	103
	104	static void rds_tcp_accept_worker(struct work_struct *work)
	105	{
	106	while (rds_tcp_accept_one(rds_tcp_listen_sock) == 0)
	107	cond_resched();
	108	}
	109
	110	void rds_tcp_listen_data_ready(struct sock *sk, int bytes)
	111	{
	112	void (ready)(struct sock sk, int bytes);
	113
	114	rdsdebug("listen data ready sk %p\n", sk);
	115
	116	read_lock(&sk->sk_callback_lock);
	117	ready = sk->sk_user_data;
	118	if (ready == NULL) { /* check for teardown race */
	119	ready = sk->sk_data_ready;
	120	goto out;
	121	}
	122
	123	/*
	124	* ->sk_data_ready is also called for a newly established child socket
	125	* before it has been accepted and the accepter has set up their
	126	* data_ready.. we only want to queue listen work for our listening
	127	* socket
	128	*/
	129	if (sk->sk_state == TCP_LISTEN)
	130	queue_work(rds_wq, &rds_tcp_listen_work);
	131
	132	out:
	133	read_unlock(&sk->sk_callback_lock);
	134	ready(sk, bytes);
	135	}
	136
	137	int __init rds_tcp_listen_init(void)
	138	{
139	struct sockaddr_in sin;
140	struct socket *sock = NULL;
141	int ret;
142
143	ret = sock_create(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
144	if (ret < 0)
145	goto out;
146
147	sock->sk->sk_reuse = 1;
148	rds_tcp_nonagle(sock);
149
150	write_lock_bh(&sock->sk->sk_callback_lock);
151	sock->sk->sk_user_data = sock->sk->sk_data_ready;
152	sock->sk->sk_data_ready = rds_tcp_listen_data_ready;
153	write_unlock_bh(&sock->sk->sk_callback_lock);
154
155	sin.sin_family = PF_INET,
156	sin.sin_addr.s_addr = (__force u32)htonl(INADDR_ANY);
157	sin.sin_port = (__force u16)htons(RDS_TCP_PORT);
158
159	ret = sock->ops->bind(sock, (struct sockaddr *)&sin, sizeof(sin));
160	if (ret < 0)
161	goto out;
162
163	ret = sock->ops->listen(sock, 64);
164	if (ret < 0)
165	goto out;
166
167	rds_tcp_listen_sock = sock;
168	sock = NULL;
169	out:
170	if (sock)
171	sock_release(sock);
172	return ret;
173	}
174
175	void rds_tcp_listen_stop(void)
176	{
177	struct socket *sock = rds_tcp_listen_sock;
178	struct sock *sk;
179
180	if (sock == NULL)
181	return;
182
183	sk = sock->sk;
184
185	/* serialize with and prevent further callbacks */
186	lock_sock(sk);
187	write_lock_bh(&sk->sk_callback_lock);
188	if (sk->sk_user_data) {
189	sk->sk_data_ready = sk->sk_user_data;
190	sk->sk_user_data = NULL;
191	}
192	write_unlock_bh(&sk->sk_callback_lock);
193	release_sock(sk);
194
195	/* wait for accepts to stop and close the socket */
196	flush_workqueue(rds_wq);
197	sock_release(sock);
198	rds_tcp_listen_sock = NULL;
199	}