* Add apr_pollcb_wakeup(), with similar behavior to

apr_pollset_wakeup(). Along the way, refactor the code for
creating/managing the wakeup pipe to a separate file, wakeup.c

* Add apr_pollcb_method_name(), with similar behavior to
apr_pollset_method_name()

* Add minimal unit tests for apr_pollset_wakeup() and apr_pollcb_wakeup()

apr_pollcb_wakeup() is supported with all the poll methods supported
by pollcb (kqueue, epoll, poll, and event ports).
Submitted by: Neil Conway <nrc@cs.berkeley.edu>


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

8 files changed, 273 insertions, 167 deletions

diff --git a/poll/unix/epoll.c b/poll/unix/epoll.c
index 326dac7b1..751d7c9bd 100644
--- a/poll/unix/epoll.c
+++ b/poll/unix/epoll.c
@@ -147,7 +147,7 @@ static apr_status_t impl_pollset_add(apr_pollset_t *pollset,
                                      const apr_pollfd_t *descriptor)
 {
     struct epoll_event ev = {0};
-    int ret = -1;
+    int ret;
     pfd_elem_t *elem = NULL;
     apr_status_t rv = APR_SUCCESS;
 
@@ -204,7 +204,7 @@ static apr_status_t impl_pollset_remove(apr_pollset_t *pollset,
     struct epoll_event ev = {0}; /* ignored, but must be passed with
                                   * kernel < 2.6.9
                                   */
-    int ret = -1;
+    int ret;
 
     if (descriptor->desc_type == APR_POLL_SOCKET) {
         ret = epoll_ctl(pollset->p->epoll_fd, EPOLL_CTL_DEL,
@@ -245,9 +245,8 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
                                            apr_int32_t *num,
                                            const apr_pollfd_t **descriptors)
 {
-    int ret, i, j;
+    int ret;
     apr_status_t rv = APR_SUCCESS;
-    apr_pollfd_t *fdptr;
 
     if (timeout > 0) {
         timeout /= 1000;
@@ -264,6 +263,9 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
         rv = APR_TIMEUP;
     }
     else {
+        int i, j;
+        apr_pollfd_t *fdptr;
+
         for (i = 0, j = 0; i < ret; i++) {
             if (pollset->flags & APR_POLLSET_NOCOPY) {
                 fdptr = (apr_pollfd_t *)(pollset->p->pollset[i].data.ptr);
@@ -277,7 +279,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_pollset_drain_wakeup_pipe(pollset);
+                drain_wakeup_pipe(pollset->wakeup_pipe);
                 rv = APR_EINTR;
             }
             else {
@@ -319,9 +321,8 @@ static apr_pollset_provider_t impl = {
 
 apr_pollset_provider_t *apr_pollset_provider_epoll = &impl;
 
-static apr_status_t cb_cleanup(void *p_)
+static apr_status_t impl_pollcb_cleanup(apr_pollcb_t *pollcb)
 {
-    apr_pollcb_t *pollcb = (apr_pollcb_t *) p_;
     close(pollcb->fd);
     return APR_SUCCESS;
 }
@@ -358,7 +359,6 @@ static apr_status_t impl_pollcb_create(apr_pollcb_t *pollcb,
     
     pollcb->fd = fd;
     pollcb->pollset.epoll = apr_palloc(p, size * sizeof(struct epoll_event));
-    apr_pool_cleanup_register(p, pollcb, cb_cleanup, apr_pool_cleanup_null);
 
     return APR_SUCCESS;
 }
@@ -370,7 +370,7 @@ static apr_status_t impl_pollcb_add(apr_pollcb_t *pollcb,
     int ret;
     
     ev.events = get_epoll_event(descriptor->reqevents);
-    ev.data.ptr = (void *)descriptor;
+    ev.data.ptr = (void *) descriptor;
 
     if (descriptor->desc_type == APR_POLL_SOCKET) {
         ret = epoll_ctl(pollcb->fd, EPOLL_CTL_ADD,
@@ -395,7 +395,7 @@ static apr_status_t impl_pollcb_remove(apr_pollcb_t *pollcb,
     struct epoll_event ev = {0}; /* ignored, but must be passed with
                                   * kernel < 2.6.9
                                   */
-    int ret = -1;
+    int ret;
     
     if (descriptor->desc_type == APR_POLL_SOCKET) {
         ret = epoll_ctl(pollcb->fd, EPOLL_CTL_DEL,
@@ -437,6 +437,14 @@ static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb,
     else {
         for (i = 0; i < ret; i++) {
             apr_pollfd_t *pollfd = (apr_pollfd_t *)(pollcb->pollset.epoll[i].data.ptr);
+
+            if ((pollcb->flags & APR_POLLSET_WAKEABLE) &&
+                pollfd->desc_type == APR_POLL_FILE &&
+                pollfd->desc.f == pollcb->wakeup_pipe[0]) {
+                drain_wakeup_pipe(pollcb->wakeup_pipe);
+                return APR_EINTR;
+            }
+
             pollfd->rtnevents = get_epoll_revent(pollcb->pollset.epoll[i].events);
 
             rv = func(baton, pollfd);
@@ -454,6 +462,7 @@ static apr_pollcb_provider_t impl_cb = {
     impl_pollcb_add,
     impl_pollcb_remove,
     impl_pollcb_poll,
+    impl_pollcb_cleanup,
     "epoll"
 };
 
diff --git a/poll/unix/kqueue.c b/poll/unix/kqueue.c
index dbe785a15..a4483e0f0 100644
--- a/poll/unix/kqueue.c
+++ b/poll/unix/kqueue.c
@@ -246,7 +246,7 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
                                       apr_int32_t *num,
                                       const apr_pollfd_t **descriptors)
 {
-    int ret, i, j;
+    int ret;
     struct timespec tv, *tvptr;
     apr_status_t rv = APR_SUCCESS;
     apr_pollfd_t fd;
@@ -270,12 +270,14 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
         rv = APR_TIMEUP;
     }
     else {
+        int i, j;
+
         for (i = 0, j = 0; i < ret; i++) {
-            fd = (((pfd_elem_t*)(pollset->p->ke_set[i].udata))->pfd);
+            fd = (((pfd_elem_t *)(pollset->p->ke_set[i].udata))->pfd);
             if ((pollset->flags & APR_POLLSET_WAKEABLE) &&
                 fd.desc_type == APR_POLL_FILE &&
                 fd.desc.f == pollset->wakeup_pipe[0]) {
-                apr_pollset_drain_wakeup_pipe(pollset);
+                drain_wakeup_pipe(pollset->wakeup_pipe);
                 rv = APR_EINTR;
             }
             else {
@@ -294,7 +296,6 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
         }
     }
 
-
     pollset_lock_rings();
 
     /* Shift all PFDs in the Dead Ring to the Free Ring */
@@ -317,9 +318,8 @@ static apr_pollset_provider_t impl = {
 
 apr_pollset_provider_t *apr_pollset_provider_kqueue = &impl;
 
-static apr_status_t cb_cleanup(void *b_)
+static apr_status_t impl_pollcb_cleanup(apr_pollcb_t *pollcb)
 {
-    apr_pollcb_t *pollcb = (apr_pollcb_t *) b_;
     close(pollcb->fd);
     return APR_SUCCESS;
 }
@@ -348,8 +348,7 @@ static apr_status_t impl_pollcb_create(apr_pollcb_t *pollcb,
     }
  
     pollcb->fd = fd;
-    pollcb->pollset.ke = (struct kevent *)apr_pcalloc(p, 2 * size * sizeof(struct kevent));
-    apr_pool_cleanup_register(p, pollcb, cb_cleanup, apr_pool_cleanup_null);
+    pollcb->pollset.ke = (struct kevent *) apr_pcalloc(p, 2 * size * sizeof(struct kevent));
     
     return APR_SUCCESS;
 }
@@ -452,7 +451,14 @@ static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb,
     else {
         for (i = 0; i < ret; i++) {
             apr_pollfd_t *pollfd = (apr_pollfd_t *)(pollcb->pollset.ke[i].udata);
-            
+
+            if ((pollcb->flags & APR_POLLSET_WAKEABLE) &&
+                pollfd->desc_type == APR_POLL_FILE &&
+                pollfd->desc.f == pollcb->wakeup_pipe[0]) {
+                drain_wakeup_pipe(pollcb->wakeup_pipe);
+                return APR_EINTR;
+            }
+
             pollfd->rtnevents = get_kqueue_revent(pollcb->pollset.ke[i].filter,
                                                   pollcb->pollset.ke[i].flags);
             
@@ -472,6 +478,7 @@ static apr_pollcb_provider_t impl_cb = {
     impl_pollcb_add,
     impl_pollcb_remove,
     impl_pollcb_poll,
+    impl_pollcb_cleanup,
     "kqueue"
 };
 
diff --git a/poll/unix/poll.c b/poll/unix/poll.c
index 3727b5b83..615470b74 100644
--- a/poll/unix/poll.c
+++ b/poll/unix/poll.c
@@ -240,7 +240,6 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
 {
     int ret;
     apr_status_t rv = APR_SUCCESS;
-    apr_uint32_t i, j;
 
     if (timeout > 0) {
         timeout /= 1000;
@@ -258,6 +257,8 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
         return APR_TIMEUP;
     }
     else {
+        apr_uint32_t i, j;
+
         for (i = 0, j = 0; i < pollset->nelts; i++) {
             if (pollset->p->pollset[i].revents != 0) {
                 /* Check if the polled descriptor is our
@@ -266,8 +267,8 @@ 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_pollset_drain_wakeup_pipe(pollset);
-                        rv = APR_EINTR;
+                    drain_wakeup_pipe(pollset->wakeup_pipe);
+                    rv = APR_EINTR;
                 }
                 else {
                     pollset->p->result_set[j] = pollset->p->query_set[i];
@@ -305,7 +306,7 @@ static apr_status_t impl_pollcb_create(apr_pollcb_t *pollcb,
                                        apr_uint32_t flags)
 {
 #if APR_HAS_THREADS
-  return APR_ENOTIMPL;
+    return APR_ENOTIMPL;
 #endif
 
     pollcb->fd = -1;
@@ -402,6 +403,14 @@ static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb,
         for (i = 0; i < pollcb->nelts; i++) {
             if (pollcb->pollset.ps[i].revents != 0) {
                 apr_pollfd_t *pollfd = pollcb->copyset[i];
+
+                if ((pollcb->flags & APR_POLLSET_WAKEABLE) &&
+                    pollfd->desc_type == APR_POLL_FILE &&
+                    pollfd->desc.f == pollcb->wakeup_pipe[0]) {
+                    drain_wakeup_pipe(pollcb->wakeup_pipe);
+                    return APR_EINTR;
+                }
+
                 pollfd->rtnevents = get_revent(pollcb->pollset.ps[i].revents);                    
                 rv = func(baton, pollfd);
                 if (rv) {
@@ -418,6 +427,7 @@ static apr_pollcb_provider_t impl_cb = {
     impl_pollcb_add,
     impl_pollcb_remove,
     impl_pollcb_poll,
+    NULL,
     "poll"
 };
 
diff --git a/poll/unix/pollcb.c b/poll/unix/pollcb.c
index 2971b0216..58c0964fa 100644
--- a/poll/unix/pollcb.c
+++ b/poll/unix/pollcb.c
@@ -72,6 +72,20 @@ static apr_pollcb_provider_t *pollcb_provider(apr_pollset_method_e method)
     return provider;
 }
 
+static apr_status_t pollcb_cleanup(void *p)
+{
+    apr_pollcb_t *pollcb = (apr_pollcb_t *) p;
+
+    if (pollcb->provider->cleanup) {
+        (*pollcb->provider->cleanup)(pollcb);
+    }
+    if (pollcb->flags & APR_POLLSET_WAKEABLE) {
+        close_wakeup_pipe(pollcb->wakeup_pipe);
+    }
+
+    return APR_SUCCESS;
+}
+
 APR_DECLARE(apr_status_t) apr_pollcb_create_ex(apr_pollcb_t **ret_pollcb,
                                                apr_uint32_t size,
                                                apr_pool_t *p,
@@ -108,9 +122,15 @@ APR_DECLARE(apr_status_t) apr_pollcb_create_ex(apr_pollcb_t **ret_pollcb,
         }
     }
 
+    if (flags & APR_POLLSET_WAKEABLE) {
+        /* Add room for wakeup descriptor */
+        size++;
+    }
+
     pollcb = apr_palloc(p, sizeof(*pollcb));
     pollcb->nelts = 0;
     pollcb->nalloc = size;
+    pollcb->flags = flags;
     pollcb->pool = p;
     pollcb->provider = provider;
 
@@ -136,6 +156,21 @@ APR_DECLARE(apr_status_t) apr_pollcb_create_ex(apr_pollcb_t **ret_pollcb,
         pollcb->provider = provider;
     }
 
+    if (flags & APR_POLLSET_WAKEABLE) {
+        /* Create wakeup pipe */
+        if ((rv = create_wakeup_pipe(pollcb->pool, &pollcb->wakeup_pfd,
+                                     pollcb->wakeup_pipe)) != APR_SUCCESS) {
+            return rv;
+        }
+
+        if ((rv = apr_pollcb_add(pollcb, &pollcb->wakeup_pfd)) != APR_SUCCESS) {
+            return rv;
+        }
+    }
+    if ((flags & APR_POLLSET_WAKEABLE) || provider->cleanup)
+        apr_pool_cleanup_register(p, pollcb, pollcb_cleanup,
+                                  apr_pool_cleanup_null);
+
     *ret_pollcb = pollcb;
     return APR_SUCCESS;
 }
@@ -169,3 +204,16 @@ APR_DECLARE(apr_status_t) apr_pollcb_poll(apr_pollcb_t *pollcb,
 {
     return (*pollcb->provider->poll)(pollcb, timeout, func, baton);
 }
+
+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]);
+    else
+        return APR_EINIT;
+}
+
+APR_DECLARE(const char *) apr_pollcb_method_name(apr_pollcb_t *pollcb)
+{
+    return pollcb->provider->name;
+}
diff --git a/poll/unix/pollset.c b/poll/unix/pollset.c
index b99cf634d..ec2f55fb4 100644
--- a/poll/unix/pollset.c
+++ b/poll/unix/pollset.c
@@ -30,140 +30,6 @@
 
 static apr_pollset_method_e pollset_default_method = POLLSET_DEFAULT_METHOD;
 
-#if !APR_FILES_AS_SOCKETS
-
-#ifdef WIN32
-
-/* Create a dummy wakeup socket pipe for interrupting the poller
- */
-static apr_status_t create_wakeup_pipe(apr_pollset_t *pollset)
-{
-    apr_status_t rv;
-    apr_pollfd_t fd;
-
-    if ((rv = file_socket_pipe_create(&pollset->wakeup_pipe[0],
-                                      &pollset->wakeup_pipe[1],
-                                      pollset->pool)) != APR_SUCCESS)
-        return rv;
-    fd.reqevents = APR_POLLIN;
-    fd.desc_type = APR_POLL_FILE;
-    fd.desc.f = pollset->wakeup_pipe[0];
-    /* Add the pipe to the pollset
-     */
-    return apr_pollset_add(pollset, &fd);
-}
-
-static apr_status_t close_wakeup_pipe(apr_pollset_t *pollset)
-{
-    apr_status_t rv0 = APR_SUCCESS;
-    apr_status_t rv1 = APR_SUCCESS;
-
-    /* Close both sides of the wakeup pipe */
-    if (pollset->wakeup_pipe[0]) {
-        rv0 = file_socket_pipe_close(pollset->wakeup_pipe[0]);
-        pollset->wakeup_pipe[0] = NULL;
-    }
-    if (pollset->wakeup_pipe[1]) {
-        rv1 = file_socket_pipe_close(pollset->wakeup_pipe[1]);
-        pollset->wakeup_pipe[1] = NULL;
-    }
-    return rv0 ? rv0 : rv1;
-}
-
-#else /* !WIN32 */
-
-static apr_status_t create_wakeup_pipe(apr_pollset_t *pollset)
-{
-    return APR_ENOTIMPL;
-}
-
-static apr_status_t close_wakeup_pipe(apr_pollset_t *pollset)
-{
-    return APR_ENOTIMPL;
-}
-
-#endif /* !WIN32 */
-
-#else  /* APR_FILES_AS_SOCKETS */
-
-/* Create a dummy wakeup pipe for interrupting the poller
- */
-static apr_status_t create_wakeup_pipe(apr_pollset_t *pollset)
-{
-    apr_status_t rv;
-    apr_pollfd_t fd;
-
-    if ((rv = apr_file_pipe_create(&pollset->wakeup_pipe[0],
-                                   &pollset->wakeup_pipe[1],
-                                   pollset->pool)) != APR_SUCCESS)
-        return rv;
-    fd.reqevents = APR_POLLIN;
-    fd.desc_type = APR_POLL_FILE;
-    fd.desc.f = pollset->wakeup_pipe[0];
-
-    {
-        int flags;
-
-        if ((flags = fcntl(pollset->wakeup_pipe[0]->filedes, F_GETFD)) == -1)
-            return errno;
-
-        flags |= FD_CLOEXEC;
-        if (fcntl(pollset->wakeup_pipe[0]->filedes, F_SETFD, flags) == -1)
-            return errno;
-    }
-    {
-        int flags;
-
-        if ((flags = fcntl(pollset->wakeup_pipe[1]->filedes, F_GETFD)) == -1)
-            return errno;
-
-        flags |= FD_CLOEXEC;
-        if (fcntl(pollset->wakeup_pipe[1]->filedes, F_SETFD, flags) == -1)
-            return errno;
-    }
-
-    /* Add the pipe to the pollset
-     */
-    return apr_pollset_add(pollset, &fd);
-}
-
-static apr_status_t close_wakeup_pipe(apr_pollset_t *pollset)
-{
-    apr_status_t rv0;
-    apr_status_t rv1;
-
-    /* Close both sides of the wakeup pipe */
-    if (pollset->wakeup_pipe[0]) {
-        rv0 = apr_file_close(pollset->wakeup_pipe[0]);
-        pollset->wakeup_pipe[0] = NULL;
-    }
-    if (pollset->wakeup_pipe[1]) {
-        rv1 = apr_file_close(pollset->wakeup_pipe[1]);
-        pollset->wakeup_pipe[1] = NULL;
-    }
-    return rv0 ? rv0 : rv1;
-}
-
-#endif /* APR_FILES_AS_SOCKETS */
-
-/* Read and discard what's ever in the wakeup pipe.
- */
-void apr_pollset_drain_wakeup_pipe(apr_pollset_t *pollset)
-{
-    char rb[512];
-    apr_size_t nr = sizeof(rb);
-
-    while (apr_file_read(pollset->wakeup_pipe[0], rb, &nr) == APR_SUCCESS) {
-        /* Although we write just one byte to the other end of the pipe
-         * during wakeup, multiple treads could call the wakeup.
-         * So simply drain out from the input side of the pipe all
-         * the data.
-         */
-        if (nr != sizeof(rb))
-            break;
-    }
-}
-
 static apr_status_t pollset_cleanup(void *p)
 {
     apr_pollset_t *pollset = (apr_pollset_t *) p;
@@ -171,7 +37,7 @@ static apr_status_t pollset_cleanup(void *p)
         (*pollset->provider->cleanup)(pollset);
     }
     if (pollset->flags & APR_POLLSET_WAKEABLE) {
-        close_wakeup_pipe(pollset);
+        close_wakeup_pipe(pollset->wakeup_pipe);
     }
 
     return APR_SUCCESS;
@@ -287,8 +153,15 @@ APR_DECLARE(apr_status_t) apr_pollset_create_ex(apr_pollset_t **ret_pollset,
         pollset->provider = provider;
     }
     if (flags & APR_POLLSET_WAKEABLE) {
+        apr_pollfd_t pfd;
+
         /* Create wakeup pipe */
-        if ((rv = create_wakeup_pipe(pollset)) != APR_SUCCESS) {
+        if ((rv = create_wakeup_pipe(pollset->pool, &pfd,
+                                     pollset->wakeup_pipe)) != APR_SUCCESS) {
+            return rv;
+        }
+
+        if ((rv = apr_pollset_add(pollset, &pfd)) != APR_SUCCESS) {
             return rv;
         }
     }
diff --git a/poll/unix/port.c b/poll/unix/port.c
index 7a31c4683..b11c54c46 100644
--- a/poll/unix/port.c
+++ b/poll/unix/port.c
@@ -405,7 +405,7 @@ static apr_status_t impl_pollset_poll(apr_pollset_t *pollset,
             if ((pollset->flags & APR_POLLSET_WAKEABLE) &&
                 fp.desc_type == APR_POLL_FILE &&
                 fp.desc.f == pollset->wakeup_pipe[0]) {
-                apr_pollset_drain_wakeup_pipe(pollset);
+                drain_wakeup_pipe(pollset->wakeup_pipe);
                 rv = APR_EINTR;
             }
             else {
@@ -458,9 +458,8 @@ static apr_pollset_provider_t impl = {
 
 apr_pollset_provider_t *apr_pollset_provider_port = &impl;
 
-static apr_status_t cb_cleanup(void *p_)
+static apr_status_t impl_pollcb_cleanup(apr_pollcb_t *pollcb)
 {
-    apr_pollcb_t *pollcb = (apr_pollcb_t *) p_;
     close(pollcb->fd);
     return APR_SUCCESS;
 }
@@ -488,7 +487,6 @@ static apr_status_t impl_pollcb_create(apr_pollcb_t *pollcb,
     }
 
     pollcb->pollset.port = apr_palloc(p, size * sizeof(port_event_t));
-    apr_pool_cleanup_register(p, pollcb, cb_cleanup, apr_pool_cleanup_null);
 
     return APR_SUCCESS;
 }
@@ -541,16 +539,25 @@ static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb,
                                      apr_pollcb_cb_t func,
                                      void *baton)
 {
-    apr_pollfd_t *pollfd;
     apr_status_t rv;
-    unsigned int i, nget = 1;
+    unsigned int nget = 1;
 
     rv = call_port_getn(pollcb->fd, pollcb->pollset.port, pollcb->nalloc,
                         &nget, timeout);
 
     if (nget) {
+        unsigned int i;
+
         for (i = 0; i < nget; i++) {
-            pollfd = (apr_pollfd_t *)(pollcb->pollset.port[i].portev_user);
+            apr_pollfd_t *pollfd = (apr_pollfd_t *)(pollcb->pollset.port[i].portev_user);
+
+            if ((pollfd->flags & APR_POLLSET_WAKEABLE) &&
+                pollfd->desc_type == APR_POLL_FILE &&
+                pollfd->desc.f == pollfd->wakeup_pipe[0]) {
+                drain_wakeup_pipe(pollfd->wakeup_pipe);
+                return APR_EINTR;
+            }
+
             pollfd->rtnevents = get_revent(pollcb->pollset.port[i].portev_events);
 
             rv = func(baton, pollfd);
@@ -569,6 +576,7 @@ static apr_pollcb_provider_t impl_cb = {
     impl_pollcb_add,
     impl_pollcb_remove,
     impl_pollcb_poll,
+    impl_pollcb_cleanup,
     "port"
 };
 
diff --git a/poll/unix/select.c b/poll/unix/select.c
index 9288de00c..f4bd482d3 100644
--- a/poll/unix/select.c
+++ b/poll/unix/select.c
@@ -382,7 +382,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_pollset_drain_wakeup_pipe(pollset);
+                drain_wakeup_pipe(pollset->wakeup_pipe);
                 rv = APR_EINTR;
                 continue;
             }
diff --git a/poll/unix/wakeup.c b/poll/unix/wakeup.c
new file mode 100644
index 000000000..f4b20d035
--- /dev/null
+++ b/poll/unix/wakeup.c
@@ -0,0 +1,151 @@
+/* Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "apr.h"
+#include "apr_poll.h"
+#include "apr_time.h"
+#include "apr_portable.h"
+#include "apr_arch_file_io.h"
+#include "apr_arch_networkio.h"
+#include "apr_arch_poll_private.h"
+#include "apr_arch_inherit.h"
+
+#if !APR_FILES_AS_SOCKETS
+
+#ifdef WIN32
+
+apr_status_t create_wakeup_pipe(apr_pool_t *pool, apr_pollfd_t *pfd,
+                                apr_file_t **wakeup_pipe)
+{
+    apr_status_t rv;
+
+    if ((rv = file_socket_pipe_create(&wakeup_pipe[0], &wakeup_pipe[1],
+                                      pool)) != APR_SUCCESS)
+        return rv;
+
+    pfd->reqevents = APR_POLLIN;
+    pfd->desc_type = APR_POLL_FILE;
+    pfd->desc.f = wakeup_pipe[0];
+    return APR_SUCCESS;
+}
+
+apr_status_t close_wakeup_pipe(apr_file_t **wakeup_pipe)
+{
+    apr_status_t rv0 = APR_SUCCESS;
+    apr_status_t rv1 = APR_SUCCESS;
+
+    /* Close both sides of the wakeup pipe */
+    if (wakeup_pipe[0]) {
+        rv0 = file_socket_pipe_close(wakeup_pipe[0]);
+        wakeup_pipe[0] = NULL;
+    }
+    if (wakeup_pipe[1]) {
+        rv1 = file_socket_pipe_close(wakeup_pipe[1]);
+        wakeup_pipe[1] = NULL;
+    }
+    return rv0 ? rv0 : rv1
+}
+
+#else /* !WIN32 */
+
+apr_status_t create_wakeup_pipe(apr_pollfd_t *pfd, apr_file_t **wakeup_pipe)
+{
+    return APR_ENOTIMPL;
+}
+
+apr_status_t close_wakeup_pipe(apr_file_t **wakeup_pipe)
+{
+    return APR_ENOTIMPL;
+}
+
+#endif /* !WIN32 */
+
+#else  /* APR_FILES_AS_SOCKETS */
+
+apr_status_t create_wakeup_pipe(apr_pool_t *pool, apr_pollfd_t *pfd,
+                                apr_file_t **wakeup_pipe)
+{
+    apr_status_t rv;
+
+    if ((rv = apr_file_pipe_create(&wakeup_pipe[0], &wakeup_pipe[1],
+                                   pool)) != APR_SUCCESS)
+        return rv;
+
+    pfd->p = pool;
+    pfd->reqevents = APR_POLLIN;
+    pfd->desc_type = APR_POLL_FILE;
+    pfd->desc.f = wakeup_pipe[0];
+
+    {
+        int flags;
+
+        if ((flags = fcntl(wakeup_pipe[0]->filedes, F_GETFD)) == -1)
+            return errno;
+
+        flags |= FD_CLOEXEC;
+        if (fcntl(wakeup_pipe[0]->filedes, F_SETFD, flags) == -1)
+            return errno;
+    }
+    {
+        int flags;
+
+        if ((flags = fcntl(wakeup_pipe[1]->filedes, F_GETFD)) == -1)
+            return errno;
+
+        flags |= FD_CLOEXEC;
+        if (fcntl(wakeup_pipe[1]->filedes, F_SETFD, flags) == -1)
+            return errno;
+    }
+
+    return APR_SUCCESS;
+}
+
+apr_status_t close_wakeup_pipe(apr_file_t **wakeup_pipe)
+{
+    apr_status_t rv0 = APR_SUCCESS;
+    apr_status_t rv1 = APR_SUCCESS;
+
+    /* Close both sides of the wakeup pipe */
+    if (wakeup_pipe[0]) {
+        rv0 = apr_file_close(wakeup_pipe[0]);
+        wakeup_pipe[0] = NULL;
+    }
+    if (wakeup_pipe[1]) {
+        rv1 = apr_file_close(wakeup_pipe[1]);
+        wakeup_pipe[1] = NULL;
+    }
+    return rv0 ? rv0 : rv1;
+}
+
+#endif /* APR_FILES_AS_SOCKETS */
+
+/* Read and discard whatever is in the wakeup pipe.
+ */
+void drain_wakeup_pipe(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
+         * 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))
+            break;
+    }
+}