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

[fs/lustre-release.git] / lnet / lnet / acceptor.c

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);