LU-10391 lnet: discard lnet_sock_accept()

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

diff --git a/lnet/lnet/acceptor.c b/lnet/lnet/acceptor.c
index 232018b..beb499b 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
 };
@@ -181,18 +188,12 @@ lnet_connect(struct socket **sockp, lnet_nid_t peer_nid,
                cr.acr_version = LNET_PROTO_ACCEPTOR_VERSION;
                cr.acr_nid     = peer_nid;
 
-               if (the_lnet.ln_testprotocompat != 0) {
+               if (the_lnet.ln_testprotocompat) {
                        /* single-shot proto check */
-                       lnet_net_lock(LNET_LOCK_EX);
-                       if ((the_lnet.ln_testprotocompat & 4) != 0) {
+                       if (test_and_clear_bit(2, &the_lnet.ln_testprotocompat))
                                cr.acr_version++;
-                               the_lnet.ln_testprotocompat &= ~4;
-                       }
-                       if ((the_lnet.ln_testprotocompat & 8) != 0) {
+                       if (test_and_clear_bit(3, &the_lnet.ln_testprotocompat))
                                cr.acr_magic = LNET_PROTO_MAGIC;
-                               the_lnet.ln_testprotocompat &= ~8;
-                       }
-                       lnet_net_unlock(LNET_LOCK_EX);
                }
 
                rc = lnet_sock_write(sock, &cr, sizeof(cr),
@@ -246,13 +247,12 @@ lnet_accept(struct socket *sock, __u32 magic)
                                               accept_timeout);
 
                        if (rc != 0)
-                               CERROR("Error sending magic+version in response"
-                                      "to LNET magic from %pI4h: %d\n",
+                               CERROR("Error sending magic+version in response to LNET magic from %pI4h: %d\n",
                                       &peer_ip, rc);
                        return -EPROTO;
                }
 
-               if (magic == le32_to_cpu(LNET_PROTO_TCP_MAGIC))
+               if (lnet_accept_magic(magic, LNET_PROTO_TCP_MAGIC))
                        str = "'old' socknal/tcpnal";
                else
                        str = "unrecognised";
@@ -292,8 +292,7 @@ lnet_accept(struct socket *sock, __u32 magic)
                                       accept_timeout);
 
                if (rc != 0)
-                       CERROR("Error sending magic+version in response"
-                              "to version %d from %pI4h: %d\n",
+                       CERROR("Error sending magic+version in response to version %d from %pI4h: %d\n",
                               peer_version, &peer_ip, rc);
                return -EPROTO;
        }
@@ -340,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)
 {
@@ -370,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 */
@@ -381,21 +408,25 @@ lnet_acceptor(void *arg)
 
        while (!lnet_acceptor_state.pta_shutdown) {
 
-               rc = lnet_sock_accept(&newsock, lnet_acceptor_state.pta_sock);
+               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 = kernel_accept(lnet_acceptor_state.pta_sock, &newsock,
+                                  SOCK_NONBLOCK);
                if (rc != 0) {
                        if (rc != -EAGAIN) {
                                CWARN("Accept error %d: pausing...\n", rc);
-                               set_current_state(TASK_UNINTERRUPTIBLE);
-                               schedule_timeout(cfs_time_seconds(1));
+                               schedule_timeout_uninterruptible(
+                                       cfs_time_seconds(1));
                        }
                        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) {
@@ -427,6 +458,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;
 
@@ -509,17 +542,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);