LU-10391 lnet: use data_ready callback to trigger accept()

Rather than blocking in lnet_sock_accept(), set up a data_ready
callback, and use that to find out when to call lnet_sock_accept()
again.

This simplifies lnet_sock_accept() (which will be removed in
next patch), and means that we could listen on multiple
sockets, which will be useful for IPv6 support.

The code design in based on that in net/sunrpc/svcsock.c.

Signed-off-by: Mr NeilBrown <neilb@suse.de>
Change-Id: I3015f2f6b6d420af5c8454b6c1a99611b48e7702
Reviewed-on: https://review.whamcloud.com/37302
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: James Simmons <jsimmons@infradead.org>
Reviewed-by: Alexey Lyashkov <alexey.lyashkov@hpe.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lnet/lnet/acceptor.c b/lnet/lnet/acceptor.c
index 3659d0c..c5ecb87 100644 (file)
--- a/lnet/lnet/acceptor.c
+++ b/lnet/lnet/acceptor.c
@@ -45,6 +45,13 @@ static struct {
        struct socket           *pta_sock;
        struct completion       pta_signal;
        struct net              *pta_ns;
+       wait_queue_head_t       pta_waitq;
+       atomic_t                pta_ready;
+#ifdef HAVE_SK_DATA_READY_ONE_ARG
+       void                    (*pta_odata)(struct sock *);
+#else
+       void                    (*pta_odata)(struct sock *, int);
+#endif
 } lnet_acceptor_state = {
        .pta_shutdown = 1
 };
@@ -332,6 +339,24 @@ lnet_accept(struct socket *sock, __u32 magic)
        return rc;
 }
 
+#ifdef HAVE_SK_DATA_READY_ONE_ARG
+static void lnet_acceptor_ready(struct sock *sk)
+#else
+static void lnet_acceptor_ready(struct sock *sk, int len)
+#endif
+{
+       /* Ensure pta_odata has actually been set before calling it */
+       rmb();
+#ifdef HAVE_SK_DATA_READY_ONE_ARG
+       lnet_acceptor_state.pta_odata(sk);
+#else
+       lnet_acceptor_state.pta_odata(sk, 0);
+#endif
+
+       atomic_set(&lnet_acceptor_state.pta_ready, 1);
+       wake_up(&lnet_acceptor_state.pta_waitq);
+}
+
 static int
 lnet_acceptor(void *arg)
 {
@@ -362,6 +387,16 @@ lnet_acceptor(void *arg)
                lnet_acceptor_state.pta_sock = NULL;
        } else {
                LCONSOLE(0, "Accept %s, port %d\n", accept_type, accept_port);
+               init_waitqueue_head(&lnet_acceptor_state.pta_waitq);
+               lnet_acceptor_state.pta_odata =
+                       lnet_acceptor_state.pta_sock->sk->sk_data_ready;
+               /* ensure pta_odata gets set before there is any chance of
+                * lnet_accept_ready() trying to read it.
+                */
+               wmb();
+               lnet_acceptor_state.pta_sock->sk->sk_data_ready =
+                       lnet_acceptor_ready;
+               atomic_set(&lnet_acceptor_state.pta_ready, 1);
        }
 
        /* set init status and unblock parent */
@@ -373,6 +408,12 @@ lnet_acceptor(void *arg)
 
        while (!lnet_acceptor_state.pta_shutdown) {
 
+               wait_event_idle(lnet_acceptor_state.pta_waitq,
+                               lnet_acceptor_state.pta_shutdown ||
+                               atomic_read(&lnet_acceptor_state.pta_ready));
+               if (!atomic_read(&lnet_acceptor_state.pta_ready))
+                       continue;
+               atomic_set(&lnet_acceptor_state.pta_ready, 0);
                rc = lnet_sock_accept(&newsock, lnet_acceptor_state.pta_sock);
                if (rc != 0) {
                        if (rc != -EAGAIN) {
@@ -383,11 +424,8 @@ lnet_acceptor(void *arg)
                        continue;
                }
 
-               /* maybe we're waken up with lnet_sock_abort_accept() */
-               if (lnet_acceptor_state.pta_shutdown) {
-                       sock_release(newsock);
-                       break;
-               }
+               /* make sure we call lnet_sock_accept() again, until it fails */
+               atomic_set(&lnet_acceptor_state.pta_ready, 1);
 
                rc = lnet_sock_getaddr(newsock, true, &peer_ip, &peer_port);
                if (rc != 0) {
@@ -419,6 +457,8 @@ failed:
                sock_release(newsock);
        }
 
+       lnet_acceptor_state.pta_sock->sk->sk_data_ready =
+               lnet_acceptor_state.pta_odata;
        sock_release(lnet_acceptor_state.pta_sock);
        lnet_acceptor_state.pta_sock = NULL;
 
@@ -501,17 +541,11 @@ lnet_acceptor_start(void)
 void
 lnet_acceptor_stop(void)
 {
-       struct sock *sk;
-
        if (lnet_acceptor_state.pta_shutdown) /* not running */
                return;
 
        lnet_acceptor_state.pta_shutdown = 1;
-
-       sk = lnet_acceptor_state.pta_sock->sk;
-
-       /* awake any sleepers using safe method */
-       sk->sk_state_change(sk);
+       wake_up(&lnet_acceptor_state.pta_waitq);
 
        /* block until acceptor signals exit */
        wait_for_completion(&lnet_acceptor_state.pta_signal);

diff --git a/lnet/lnet/lib-socket.c b/lnet/lnet/lib-socket.c
index 642d8b5..99122e8 100644 (file)
--- a/lnet/lnet/lib-socket.c
+++ b/lnet/lnet/lib-socket.c
@@ -335,7 +335,6 @@ lnet_sock_listen(struct socket **sockp,
 int
 lnet_sock_accept(struct socket **newsockp, struct socket *sock)
 {
-       wait_queue_entry_t wait;
        struct socket *newsock;
        int            rc;
 
@@ -354,20 +353,6 @@ lnet_sock_accept(struct socket **newsockp, struct socket *sock)
 #else
        rc = sock->ops->accept(sock, newsock, O_NONBLOCK);
 #endif
-       if (rc == -EAGAIN) {
-               /* Nothing ready, so wait for activity */
-               init_waitqueue_entry(&wait, current);
-               add_wait_queue(sk_sleep(sock->sk), &wait);
-               set_current_state(TASK_INTERRUPTIBLE);
-               schedule();
-               remove_wait_queue(sk_sleep(sock->sk), &wait);
-#ifdef HAVE_KERN_SOCK_ACCEPT_FLAG_ARG
-               rc = sock->ops->accept(sock, newsock, O_NONBLOCK, false);
-#else
-               rc = sock->ops->accept(sock, newsock, O_NONBLOCK);
-#endif
-       }
-
        if (rc != 0)
                goto failed;