Fix drain wakeup pipe issue when multiple threads call apr_pollset_wakeup/apr_pollcb_wakeup for the same pollset filling up drain pipe. Use atomics so that wakeup call is noop if some other thread allready done this

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

8 files changed, 32 insertions, 22 deletions

diff --git a/poll/unix/epoll.c b/poll/unix/epoll.c
index 4ab03f67c..c00a1094d 100644
--- a/poll/unix/epoll.c
+++ b/poll/unix/epoll.c
@@ -289,7 +289,7 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
             if ((pollset->flags & APR_POLLSET_WAKEABLE) &&
                 fdptr->desc_type == APR_POLL_FILE &&
                 fdptr->desc.f == pollset->wakeup_pipe[0]) {
-                apr_poll_drain_wakeup_pipe(pollset->wakeup_pipe);
+                apr_poll_drain_wakeup_pipe(&pollset->wakeup_set, pollset->wakeup_pipe);
                 rv = APR_EINTR;
             }
             else {
@@ -460,7 +460,7 @@ static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb,
             if ((pollcb->flags & APR_POLLSET_WAKEABLE) &&
                 pollfd->desc_type == APR_POLL_FILE &&
                 pollfd->desc.f == pollcb->wakeup_pipe[0]) {
-                apr_poll_drain_wakeup_pipe(pollcb->wakeup_pipe);
+                apr_poll_drain_wakeup_pipe(&pollcb->wakeup_set, pollcb->wakeup_pipe);
                 return APR_EINTR;
             }
 
diff --git a/poll/unix/kqueue.c b/poll/unix/kqueue.c
index 548464db1..e8a1ef95b 100644
--- a/poll/unix/kqueue.c
+++ b/poll/unix/kqueue.c
@@ -286,7 +286,7 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
             if ((pollset->flags & APR_POLLSET_WAKEABLE) &&
                 fd->desc_type == APR_POLL_FILE &&
                 fd->desc.f == pollset->wakeup_pipe[0]) {
-                apr_poll_drain_wakeup_pipe(pollset->wakeup_pipe);
+                apr_poll_drain_wakeup_pipe(&pollset->wakeup_set, pollset->wakeup_pipe);
                 rv = APR_EINTR;
             }
             else {
@@ -473,7 +473,7 @@ static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb,
             if ((pollcb->flags & APR_POLLSET_WAKEABLE) &&
                 pollfd->desc_type == APR_POLL_FILE &&
                 pollfd->desc.f == pollcb->wakeup_pipe[0]) {
-                apr_poll_drain_wakeup_pipe(pollcb->wakeup_pipe);
+                apr_poll_drain_wakeup_pipe(&pollcb->wakeup_set, pollcb->wakeup_pipe);
                 return APR_EINTR;
             }
 
diff --git a/poll/unix/poll.c b/poll/unix/poll.c
index e96a53cff..07ed79674 100644
--- a/poll/unix/poll.c
+++ b/poll/unix/poll.c
@@ -275,7 +275,7 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
                 if ((pollset->flags & APR_POLLSET_WAKEABLE) &&
                     pollset->p->query_set[i].desc_type == APR_POLL_FILE &&
                     pollset->p->query_set[i].desc.f == pollset->wakeup_pipe[0]) {
-                    apr_poll_drain_wakeup_pipe(pollset->wakeup_pipe);
+                    apr_poll_drain_wakeup_pipe(&pollset->wakeup_set, pollset->wakeup_pipe);
                     rv = APR_EINTR;
                 }
                 else {
@@ -422,7 +422,7 @@ static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb,
                 if ((pollcb->flags & APR_POLLSET_WAKEABLE) &&
                     pollfd->desc_type == APR_POLL_FILE &&
                     pollfd->desc.f == pollcb->wakeup_pipe[0]) {
-                    apr_poll_drain_wakeup_pipe(pollcb->wakeup_pipe);
+                    apr_poll_drain_wakeup_pipe(&pollcb->wakeup_set, pollcb->wakeup_pipe);
                     return APR_EINTR;
                 }
 
diff --git a/poll/unix/pollcb.c b/poll/unix/pollcb.c
index a63ad5c9c..1ab8cd3d2 100644
--- a/poll/unix/pollcb.c
+++ b/poll/unix/pollcb.c
@@ -23,6 +23,7 @@
 #include "apr_poll.h"
 #include "apr_time.h"
 #include "apr_portable.h"
+#include "apr_atomic.h"
 #include "apr_arch_file_io.h"
 #include "apr_arch_networkio.h"
 #include "apr_arch_poll_private.h"
@@ -134,6 +135,7 @@ APR_DECLARE(apr_status_t) apr_pollcb_create_ex(apr_pollcb_t **ret_pollcb,
     pollcb->flags = flags;
     pollcb->pool = p;
     pollcb->provider = provider;
+    pollcb->wakeup_set = 0;
 
     rv = (*provider->create)(pollcb, size, p, flags);
     if (rv == APR_ENOTIMPL) {
@@ -212,8 +214,12 @@ 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)
-        return apr_file_putc(1, pollcb->wakeup_pipe[1]);
+    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
         return APR_EINIT;
 }
diff --git a/poll/unix/pollset.c b/poll/unix/pollset.c
index 13c939371..0c4508251 100644
--- a/poll/unix/pollset.c
+++ b/poll/unix/pollset.c
@@ -23,6 +23,7 @@
 #include "apr_poll.h"
 #include "apr_time.h"
 #include "apr_portable.h"
+#include "apr_atomic.h"
 #include "apr_arch_file_io.h"
 #include "apr_arch_networkio.h"
 #include "apr_arch_poll_private.h"
@@ -140,6 +141,7 @@ APR_DECLARE(apr_status_t) apr_pollset_create_ex(apr_pollset_t **ret_pollset,
     pollset->pool = p;
     pollset->flags = flags;
     pollset->provider = provider;
+    pollset->wakeup_set = 0;
 
     rv = (*provider->create)(pollset, size, p, flags);
     if (rv == APR_ENOTIMPL) {
@@ -216,8 +218,12 @@ 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)
-        return apr_file_putc(1, pollset->wakeup_pipe[1]);
+    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
         return APR_EINIT;
 }
diff --git a/poll/unix/port.c b/poll/unix/port.c
index c1e599412..638a8cba7 100644
--- a/poll/unix/port.c
+++ b/poll/unix/port.c
@@ -411,7 +411,7 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
         if ((pollset->flags & APR_POLLSET_WAKEABLE) &&
             ep->pfd.desc_type == APR_POLL_FILE &&
             ep->pfd.desc.f == pollset->wakeup_pipe[0]) {
-            apr_poll_drain_wakeup_pipe(pollset->wakeup_pipe);
+            apr_poll_drain_wakeup_pipe(&pollset->wakeup_set, pollset->wakeup_pipe);
             rv = APR_EINTR;
         }
         else {
@@ -563,7 +563,7 @@ static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb,
             if ((pollcb->flags & APR_POLLSET_WAKEABLE) &&
                 pollfd->desc_type == APR_POLL_FILE &&
                 pollfd->desc.f == pollcb->wakeup_pipe[0]) {
-                apr_poll_drain_wakeup_pipe(pollcb->wakeup_pipe);
+                apr_poll_drain_wakeup_pipe(&pollcb->wakeup_set, pollcb->wakeup_pipe);
                 return APR_EINTR;
             }
 
diff --git a/poll/unix/select.c b/poll/unix/select.c
index 51be3c1cd..f2c1a1328 100644
--- a/poll/unix/select.c
+++ b/poll/unix/select.c
@@ -401,7 +401,7 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
         else {
             if ((pollset->flags & APR_POLLSET_WAKEABLE) &&
                 pollset->p->query_set[i].desc.f == pollset->wakeup_pipe[0]) {
-                apr_poll_drain_wakeup_pipe(pollset->wakeup_pipe);
+                apr_poll_drain_wakeup_pipe(&pollset->wakeup_set, pollset->wakeup_pipe);
                 rv = APR_EINTR;
                 continue;
             }
diff --git a/poll/unix/wakeup.c b/poll/unix/wakeup.c
index 01b84f9a9..ccc67d133 100644
--- a/poll/unix/wakeup.c
+++ b/poll/unix/wakeup.c
@@ -18,6 +18,7 @@
 #include "apr_poll.h"
 #include "apr_time.h"
 #include "apr_portable.h"
+#include "apr_atomic.h"
 #include "apr_arch_file_io.h"
 #include "apr_arch_networkio.h"
 #include "apr_arch_poll_private.h"
@@ -36,9 +37,6 @@ apr_status_t apr_poll_create_wakeup_pipe(apr_pool_t *pool, apr_pollfd_t *pfd,
                                       pool)) != APR_SUCCESS)
         return rv;
 
-    /* Read end of the pipe is non-blocking */
-    apr_file_pipe_timeout_set(wakeup_pipe[0], 0);
-
     pfd->reqevents = APR_POLLIN;
     pfd->desc_type = APR_POLL_FILE;
     pfd->desc.f = wakeup_pipe[0];
@@ -139,18 +137,18 @@ apr_status_t apr_poll_close_wakeup_pipe(apr_file_t **wakeup_pipe)
 
 /* Read and discard whatever is in the wakeup pipe.
  */
-void apr_poll_drain_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 rb[512];
-    apr_size_t nr = sizeof(rb);
 
-    while (apr_file_read(wakeup_pipe[0], rb, &nr) == APR_SUCCESS) {
-        /* Although we write just one byte to the other end of the pipe
+    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 (nr != sizeof(rb))
+        if (apr_file_getc(&ch, wakeup_pipe[0]) != APR_SUCCESS)
             break;
     }
 }
+