Merge git://git.kernel.org/pub/scm/linux/kernel/git/ebiederm/sysctl-2.6
[linux-2.6.git] / net / rds / tcp_listen.c
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",
70                   NIPQUAD(inet->saddr), ntohs(inet->sport),
71                   NIPQUAD(inet->daddr), ntohs(inet->dport));
72
73         conn = rds_conn_create(inet->saddr, inet->daddr, &rds_tcp_transport,
74                                GFP_KERNEL);
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 }