Follow up to r1895106: Use less expensive atomics for wakeup.

If pipe writers (wakeup) put a single byte until consume it's consumed by the
reader (drain), an atomic cas for the writers (still) and an atomic (re)set
for the reader is enough (not need to cas on the reader side).

This requires that the reader never blocks on read though (e.g. spurious return
from poll), so (re)make the read side on the pipe non-blocking (finally).

Since synchronous non-blocking read is not a thing for Windows' Readfile(), add
a ->socket flag this arch's to apr_file_t (like the existing ->pipe one) which
file_socket_pipe_create() will set to make apr_file_read/write() handle
non-blocking (nor overlapped) socket pipes with WSARecv/Send().



git-svn-id: https://svn.apache.org/repos/asf/apr/apr/trunk@1895175 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 31 insertions, 39 deletions

diff --git a/poll/os2/pollset.c b/poll/os2/pollset.c
index 2ec848105..c407f5d98 100644
--- a/poll/os2/pollset.c
+++ b/poll/os2/pollset.c
@@ -67,7 +67,6 @@ APR_DECLARE(apr_status_t) apr_pollset_create(apr_pollset_t **pollset,
 
         if (rc == APR_SUCCESS) {
             apr_sockaddr_t *listen_address;
-            apr_socket_timeout_set((*pollset)->wake_listen, 0);
             apr_sockaddr_info_get(&listen_address, "", APR_UNIX, 0, 0, p);
             rc = apr_socket_bind((*pollset)->wake_listen, listen_address);
 
@@ -80,6 +79,7 @@ APR_DECLARE(apr_status_t) apr_pollset_create(apr_pollset_t **pollset,
                 wake_poll_fd.client_data = NULL;
                 apr_pollset_add(*pollset, &wake_poll_fd);
                 apr_socket_addr_get(&(*pollset)->wake_address, APR_LOCAL, (*pollset)->wake_listen);
+                apr_socket_timeout_set((*pollset)->wake_listen, 0);
 
                 rc = apr_socket_create(&(*pollset)->wake_sender, APR_UNIX, SOCK_DGRAM, 0, p);
             }
@@ -263,17 +263,14 @@ APR_DECLARE(apr_status_t) apr_pollset_poll(apr_pollset_t *pollset,
 
         if (rtnevents) {
             if (i == 0 && pollset->wake_listen != NULL) {
+                char ch;
+                apr_size_t len = 1;
                 struct apr_sockaddr_t from_addr;
-                char buffer[16];
-                apr_size_t buflen;
-                for (;;) {
-                    buflen = sizeof(buffer);
-                    rv = apr_socket_recvfrom(&from_addr, pollset->wake_listen,
-                                             MSG_DONTWAIT, buffer, &buflen);
-                    if (rv != APR_SUCCESS) {
-                        break;
-                    }
-                    /* Woken up, drain the pipe still. */
+                rv = apr_socket_recvfrom(&from_addr, pollset->wake_listen,
+                                          MSG_DONTWAIT, &ch, &len);
+                if (rv == APR_SUCCESS) {
+                    /* Woken up, senders can fill the pipe again */
+                    apr_atomic_set32(&pollset->wakeup_set, 0);
                     rc = APR_EINTR;
                 }
             }
@@ -298,12 +295,15 @@ APR_DECLARE(apr_status_t) apr_pollset_poll(apr_pollset_t *pollset,
 
 APR_DECLARE(apr_status_t) apr_pollset_wakeup(apr_pollset_t *pollset)
 {
-    if (pollset->wake_sender) {
+    if (!pollset->wake_sender)
+        return APR_EINIT;
+
+    if (apr_atomic_cas32(&pollset->wakeup_set, 1, 0) == 0) {
         apr_size_t len = 1;
         return apr_socket_sendto(pollset->wake_sender, pollset->wake_address, 0, "", &len);
     }
 
-    return APR_EINIT;
+    return APR_SUCCESS;
 }
 
 
diff --git a/poll/unix/pollcb.c b/poll/unix/pollcb.c
index 1ab8cd3d2..103ab9fa6 100644
--- a/poll/unix/pollcb.c
+++ b/poll/unix/pollcb.c
@@ -214,14 +214,13 @@ APR_DECLARE(apr_status_t) apr_pollcb_poll(apr_pollcb_t *pollcb,
 
 APR_DECLARE(apr_status_t) apr_pollcb_wakeup(apr_pollcb_t *pollcb)
 {
-    if (pollcb->flags & APR_POLLSET_WAKEABLE) {
-        if (apr_atomic_cas32(&pollcb->wakeup_set, 1, 0) == 0)
-            return apr_file_putc(1, pollcb->wakeup_pipe[1]);
-        else
-           return APR_SUCCESS;
-    }
-    else
+    if (!(pollcb->flags & APR_POLLSET_WAKEABLE))
         return APR_EINIT;
+
+    if (apr_atomic_cas32(&pollcb->wakeup_set, 1, 0) == 0)
+        return apr_file_putc(1, pollcb->wakeup_pipe[1]);
+
+    return APR_SUCCESS;
 }
 
 APR_DECLARE(const char *) apr_pollcb_method_name(apr_pollcb_t *pollcb)
diff --git a/poll/unix/pollset.c b/poll/unix/pollset.c
index 0c4508251..630bd4a57 100644
--- a/poll/unix/pollset.c
+++ b/poll/unix/pollset.c
@@ -218,14 +218,13 @@ APR_DECLARE(apr_status_t) apr_pollset_destroy(apr_pollset_t * pollset)
 
 APR_DECLARE(apr_status_t) apr_pollset_wakeup(apr_pollset_t *pollset)
 {
-    if (pollset->flags & APR_POLLSET_WAKEABLE) {
-        if (apr_atomic_cas32(&pollset->wakeup_set, 1, 0) == 0)
-            return apr_file_putc(1, pollset->wakeup_pipe[1]);
-        else
-           return APR_SUCCESS;
-    }
-    else
+    if (!(pollset->flags & APR_POLLSET_WAKEABLE))
         return APR_EINIT;
+
+    if (apr_atomic_cas32(&pollset->wakeup_set, 1, 0) == 0)
+        return apr_file_putc(1, pollset->wakeup_pipe[1]);
+
+    return APR_SUCCESS;
 }
 
 APR_DECLARE(apr_status_t) apr_pollset_add(apr_pollset_t *pollset,
diff --git a/poll/unix/wakeup.c b/poll/unix/wakeup.c
index 118272368..9561a57dc 100644
--- a/poll/unix/wakeup.c
+++ b/poll/unix/wakeup.c
@@ -81,8 +81,9 @@ apr_status_t apr_poll_create_wakeup_pipe(apr_pool_t *pool, apr_pollfd_t *pfd,
 {
     apr_status_t rv;
 
-    if ((rv = apr_file_pipe_create(&wakeup_pipe[0], &wakeup_pipe[1],
-                                   pool)) != APR_SUCCESS)
+    /* Read end of the pipe is non-blocking */
+    if ((rv = apr_file_pipe_create_ex(&wakeup_pipe[0], &wakeup_pipe[1],
+                                      APR_WRITE_BLOCK, pool)))
         return rv;
 
     pfd->p = pool;
@@ -137,16 +138,9 @@ apr_status_t apr_poll_close_wakeup_pipe(apr_file_t **wakeup_pipe)
  */
 void apr_poll_drain_wakeup_pipe(volatile apr_uint32_t *wakeup_set, apr_file_t **wakeup_pipe)
 {
+    char ch;
 
-    while (apr_atomic_cas32(wakeup_set, 0, 1) > 0) {
-        char ch;
-        /* though we write just one byte to the other end of the pipe
-         * during wakeup, multiple threads could call the wakeup.
-         * So simply drain out from the input side of the pipe all
-         * the data.
-         */
-        if (apr_file_getc(&ch, wakeup_pipe[0]) != APR_SUCCESS)
-            break;
-    }
+    (void)apr_file_getc(&ch, wakeup_pipe[0]);
+    apr_atomic_set32(wakeup_set, 0);
 }