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 }