diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 09:11:38 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 09:11:38 +0000 |
commit | bc7e963c37d9c8d1c854ac960df241cfa34e3dc5 (patch) | |
tree | aa35d7414ce9f1326abf6f723f6dfa5b0aa08b1d /poll/unix | |
parent | Initial commit. (diff) | |
download | apr-upstream.tar.xz apr-upstream.zip |
Adding upstream version 1.7.2.upstream/1.7.2upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'poll/unix')
-rw-r--r-- | poll/unix/epoll.c | 490 | ||||
-rw-r--r-- | poll/unix/kqueue.c | 505 | ||||
-rw-r--r-- | poll/unix/poll.c | 460 | ||||
-rw-r--r-- | poll/unix/pollcb.c | 224 | ||||
-rw-r--r-- | poll/unix/pollset.c | 247 | ||||
-rw-r--r-- | poll/unix/port.c | 594 | ||||
-rw-r--r-- | poll/unix/select.c | 449 | ||||
-rw-r--r-- | poll/unix/wakeup.c | 152 | ||||
-rw-r--r-- | poll/unix/z_asio.c | 782 |
9 files changed, 3903 insertions, 0 deletions
diff --git a/poll/unix/epoll.c b/poll/unix/epoll.c new file mode 100644 index 0000000..4ab03f6 --- /dev/null +++ b/poll/unix/epoll.c @@ -0,0 +1,490 @@ +/* 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 defined(HAVE_EPOLL) + +static apr_int16_t get_epoll_event(apr_int16_t event) +{ + apr_int16_t rv = 0; + + if (event & APR_POLLIN) + rv |= EPOLLIN; + if (event & APR_POLLPRI) + rv |= EPOLLPRI; + if (event & APR_POLLOUT) + rv |= EPOLLOUT; + /* APR_POLLNVAL is not handled by epoll. EPOLLERR and EPOLLHUP are return-only */ + + return rv; +} + +static apr_int16_t get_epoll_revent(apr_int16_t event) +{ + apr_int16_t rv = 0; + + if (event & EPOLLIN) + rv |= APR_POLLIN; + if (event & EPOLLPRI) + rv |= APR_POLLPRI; + if (event & EPOLLOUT) + rv |= APR_POLLOUT; + if (event & EPOLLERR) + rv |= APR_POLLERR; + if (event & EPOLLHUP) + rv |= APR_POLLHUP; + /* APR_POLLNVAL is not handled by epoll. */ + + return rv; +} + +struct apr_pollset_private_t +{ + int epoll_fd; + struct epoll_event *pollset; + apr_pollfd_t *result_set; +#if APR_HAS_THREADS + /* A thread mutex to protect operations on the rings */ + apr_thread_mutex_t *ring_lock; +#endif + /* A ring containing all of the pollfd_t that are active */ + APR_RING_HEAD(pfd_query_ring_t, pfd_elem_t) query_ring; + /* A ring of pollfd_t that have been used, and then _remove()'d */ + APR_RING_HEAD(pfd_free_ring_t, pfd_elem_t) free_ring; + /* A ring of pollfd_t where rings that have been _remove()`ed but + might still be inside a _poll() */ + APR_RING_HEAD(pfd_dead_ring_t, pfd_elem_t) dead_ring; +}; + +static apr_status_t impl_pollset_cleanup(apr_pollset_t *pollset) +{ + close(pollset->p->epoll_fd); + return APR_SUCCESS; +} + + +static apr_status_t impl_pollset_create(apr_pollset_t *pollset, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + apr_status_t rv; + int fd; + +#ifdef HAVE_EPOLL_CREATE1 + fd = epoll_create1(EPOLL_CLOEXEC); +#else + fd = epoll_create(size); +#endif + if (fd < 0) { + pollset->p = NULL; + return apr_get_netos_error(); + } + +#ifndef HAVE_EPOLL_CREATE1 + { + int fd_flags; + + if ((fd_flags = fcntl(fd, F_GETFD)) == -1) { + rv = errno; + close(fd); + pollset->p = NULL; + return rv; + } + + fd_flags |= FD_CLOEXEC; + if (fcntl(fd, F_SETFD, fd_flags) == -1) { + rv = errno; + close(fd); + pollset->p = NULL; + return rv; + } + } +#endif + + pollset->p = apr_palloc(p, sizeof(apr_pollset_private_t)); +#if APR_HAS_THREADS + if ((flags & APR_POLLSET_THREADSAFE) && + !(flags & APR_POLLSET_NOCOPY) && + ((rv = apr_thread_mutex_create(&pollset->p->ring_lock, + APR_THREAD_MUTEX_DEFAULT, + p)) != APR_SUCCESS)) { + close(fd); + pollset->p = NULL; + return rv; + } +#else + if (flags & APR_POLLSET_THREADSAFE) { + close(fd); + pollset->p = NULL; + return APR_ENOTIMPL; + } +#endif + pollset->p->epoll_fd = fd; + pollset->p->pollset = apr_palloc(p, size * sizeof(struct epoll_event)); + pollset->p->result_set = apr_palloc(p, size * sizeof(apr_pollfd_t)); + + if (!(flags & APR_POLLSET_NOCOPY)) { + APR_RING_INIT(&pollset->p->query_ring, pfd_elem_t, link); + APR_RING_INIT(&pollset->p->free_ring, pfd_elem_t, link); + APR_RING_INIT(&pollset->p->dead_ring, pfd_elem_t, link); + } + return APR_SUCCESS; +} + +static apr_status_t impl_pollset_add(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + struct epoll_event ev = {0}; + int ret; + pfd_elem_t *elem = NULL; + apr_status_t rv = APR_SUCCESS; + + ev.events = get_epoll_event(descriptor->reqevents); + + if (pollset->flags & APR_POLLSET_NOCOPY) { + ev.data.ptr = (void *)descriptor; + } + else { + pollset_lock_rings(); + + if (!APR_RING_EMPTY(&(pollset->p->free_ring), pfd_elem_t, link)) { + elem = APR_RING_FIRST(&(pollset->p->free_ring)); + APR_RING_REMOVE(elem, link); + } + else { + elem = (pfd_elem_t *) apr_palloc(pollset->pool, sizeof(pfd_elem_t)); + APR_RING_ELEM_INIT(elem, link); + } + elem->pfd = *descriptor; + ev.data.ptr = elem; + } + if (descriptor->desc_type == APR_POLL_SOCKET) { + ret = epoll_ctl(pollset->p->epoll_fd, EPOLL_CTL_ADD, + descriptor->desc.s->socketdes, &ev); + } + else { + ret = epoll_ctl(pollset->p->epoll_fd, EPOLL_CTL_ADD, + descriptor->desc.f->filedes, &ev); + } + + if (0 != ret) { + rv = apr_get_netos_error(); + } + + if (!(pollset->flags & APR_POLLSET_NOCOPY)) { + if (rv != APR_SUCCESS) { + APR_RING_INSERT_TAIL(&(pollset->p->free_ring), elem, pfd_elem_t, link); + } + else { + APR_RING_INSERT_TAIL(&(pollset->p->query_ring), elem, pfd_elem_t, link); + } + pollset_unlock_rings(); + } + + return rv; +} + +static apr_status_t impl_pollset_remove(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + pfd_elem_t *ep; + apr_status_t rv = APR_SUCCESS; + struct epoll_event ev = {0}; /* ignored, but must be passed with + * kernel < 2.6.9 + */ + int ret; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + ret = epoll_ctl(pollset->p->epoll_fd, EPOLL_CTL_DEL, + descriptor->desc.s->socketdes, &ev); + } + else { + ret = epoll_ctl(pollset->p->epoll_fd, EPOLL_CTL_DEL, + descriptor->desc.f->filedes, &ev); + } + if (ret < 0) { + rv = APR_NOTFOUND; + } + + if (!(pollset->flags & APR_POLLSET_NOCOPY)) { + pollset_lock_rings(); + + for (ep = APR_RING_FIRST(&(pollset->p->query_ring)); + ep != APR_RING_SENTINEL(&(pollset->p->query_ring), + pfd_elem_t, link); + ep = APR_RING_NEXT(ep, link)) { + + if (descriptor->desc.s == ep->pfd.desc.s) { + APR_RING_REMOVE(ep, link); + APR_RING_INSERT_TAIL(&(pollset->p->dead_ring), + ep, pfd_elem_t, link); + break; + } + } + + pollset_unlock_rings(); + } + + return rv; +} + +static apr_status_t impl_pollset_poll(apr_pollset_t *pollset, + apr_interval_time_t timeout, + apr_int32_t *num, + const apr_pollfd_t **descriptors) +{ + int ret; + apr_status_t rv = APR_SUCCESS; + + *num = 0; + + if (timeout > 0) { + timeout /= 1000; + } + + ret = epoll_wait(pollset->p->epoll_fd, pollset->p->pollset, pollset->nalloc, + timeout); + if (ret < 0) { + rv = apr_get_netos_error(); + } + else if (ret == 0) { + rv = APR_TIMEUP; + } + else { + int i, j; + const 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); + } + else { + fdptr = &(((pfd_elem_t *) (pollset->p->pollset[i].data.ptr))->pfd); + } + /* Check if the polled descriptor is our + * wakeup pipe. In that case do not put it result set. + */ + 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); + rv = APR_EINTR; + } + else { + pollset->p->result_set[j] = *fdptr; + pollset->p->result_set[j].rtnevents = + get_epoll_revent(pollset->p->pollset[i].events); + j++; + } + } + if (((*num) = j)) { /* any event besides wakeup pipe? */ + rv = APR_SUCCESS; + + if (descriptors) { + *descriptors = pollset->p->result_set; + } + } + } + + if (!(pollset->flags & APR_POLLSET_NOCOPY)) { + pollset_lock_rings(); + + /* Shift all PFDs in the Dead Ring to the Free Ring */ + APR_RING_CONCAT(&(pollset->p->free_ring), &(pollset->p->dead_ring), pfd_elem_t, link); + + pollset_unlock_rings(); + } + + return rv; +} + +static const apr_pollset_provider_t impl = { + impl_pollset_create, + impl_pollset_add, + impl_pollset_remove, + impl_pollset_poll, + impl_pollset_cleanup, + "epoll" +}; + +const apr_pollset_provider_t *const apr_pollset_provider_epoll = &impl; + +static apr_status_t impl_pollcb_cleanup(apr_pollcb_t *pollcb) +{ + close(pollcb->fd); + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_create(apr_pollcb_t *pollcb, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + int fd; + +#ifdef HAVE_EPOLL_CREATE1 + fd = epoll_create1(EPOLL_CLOEXEC); +#else + fd = epoll_create(size); +#endif + + if (fd < 0) { + return apr_get_netos_error(); + } + +#ifndef HAVE_EPOLL_CREATE1 + { + int fd_flags; + apr_status_t rv; + + if ((fd_flags = fcntl(fd, F_GETFD)) == -1) { + rv = errno; + close(fd); + pollcb->fd = -1; + return rv; + } + + fd_flags |= FD_CLOEXEC; + if (fcntl(fd, F_SETFD, fd_flags) == -1) { + rv = errno; + close(fd); + pollcb->fd = -1; + return rv; + } + } +#endif + + pollcb->fd = fd; + pollcb->pollset.epoll = apr_palloc(p, size * sizeof(struct epoll_event)); + + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_add(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + struct epoll_event ev = { 0 }; + int ret; + + ev.events = get_epoll_event(descriptor->reqevents); + ev.data.ptr = (void *) descriptor; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + ret = epoll_ctl(pollcb->fd, EPOLL_CTL_ADD, + descriptor->desc.s->socketdes, &ev); + } + else { + ret = epoll_ctl(pollcb->fd, EPOLL_CTL_ADD, + descriptor->desc.f->filedes, &ev); + } + + if (ret == -1) { + return apr_get_netos_error(); + } + + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_remove(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + apr_status_t rv = APR_SUCCESS; + struct epoll_event ev = {0}; /* ignored, but must be passed with + * kernel < 2.6.9 + */ + int ret; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + ret = epoll_ctl(pollcb->fd, EPOLL_CTL_DEL, + descriptor->desc.s->socketdes, &ev); + } + else { + ret = epoll_ctl(pollcb->fd, EPOLL_CTL_DEL, + descriptor->desc.f->filedes, &ev); + } + + if (ret < 0) { + rv = APR_NOTFOUND; + } + + return rv; +} + + +static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb, + apr_interval_time_t timeout, + apr_pollcb_cb_t func, + void *baton) +{ + int ret, i; + apr_status_t rv = APR_SUCCESS; + + if (timeout > 0) { + timeout /= 1000; + } + + ret = epoll_wait(pollcb->fd, pollcb->pollset.epoll, pollcb->nalloc, + timeout); + if (ret < 0) { + rv = apr_get_netos_error(); + } + else if (ret == 0) { + rv = APR_TIMEUP; + } + 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]) { + apr_poll_drain_wakeup_pipe(pollcb->wakeup_pipe); + return APR_EINTR; + } + + pollfd->rtnevents = get_epoll_revent(pollcb->pollset.epoll[i].events); + + rv = func(baton, pollfd); + if (rv) { + return rv; + } + } + } + + return rv; +} + +static const apr_pollcb_provider_t impl_cb = { + impl_pollcb_create, + impl_pollcb_add, + impl_pollcb_remove, + impl_pollcb_poll, + impl_pollcb_cleanup, + "epoll" +}; + +const apr_pollcb_provider_t *const apr_pollcb_provider_epoll = &impl_cb; + +#endif /* HAVE_EPOLL */ diff --git a/poll/unix/kqueue.c b/poll/unix/kqueue.c new file mode 100644 index 0000000..548464d --- /dev/null +++ b/poll/unix/kqueue.c @@ -0,0 +1,505 @@ +/* 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" + +#ifdef HAVE_KQUEUE + +static apr_int16_t get_kqueue_revent(apr_int16_t event, apr_int16_t flags) +{ + apr_int16_t rv = 0; + + if (event == EVFILT_READ) + rv |= APR_POLLIN; + else if (event == EVFILT_WRITE) + rv |= APR_POLLOUT; + if (flags & EV_EOF) + rv |= APR_POLLHUP; + /* APR_POLLPRI, APR_POLLERR, and APR_POLLNVAL are not handled by this + * implementation. + * TODO: See if EV_ERROR + certain system errors in the returned data field + * should map to APR_POLLNVAL. + */ + return rv; +} + +struct apr_pollset_private_t +{ + int kqueue_fd; + struct kevent kevent; + apr_uint32_t setsize; + struct kevent *ke_set; + apr_pollfd_t *result_set; +#if APR_HAS_THREADS + /* A thread mutex to protect operations on the rings */ + apr_thread_mutex_t *ring_lock; +#endif + /* A ring containing all of the pollfd_t that are active */ + APR_RING_HEAD(pfd_query_ring_t, pfd_elem_t) query_ring; + /* A ring of pollfd_t that have been used, and then _remove'd */ + APR_RING_HEAD(pfd_free_ring_t, pfd_elem_t) free_ring; + /* A ring of pollfd_t where rings that have been _remove'd but + might still be inside a _poll */ + APR_RING_HEAD(pfd_dead_ring_t, pfd_elem_t) dead_ring; +}; + +static apr_status_t impl_pollset_cleanup(apr_pollset_t *pollset) +{ + close(pollset->p->kqueue_fd); + return APR_SUCCESS; +} + +static apr_status_t impl_pollset_create(apr_pollset_t *pollset, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + apr_status_t rv; + pollset->p = apr_palloc(p, sizeof(apr_pollset_private_t)); +#if APR_HAS_THREADS + if (flags & APR_POLLSET_THREADSAFE && + ((rv = apr_thread_mutex_create(&pollset->p->ring_lock, + APR_THREAD_MUTEX_DEFAULT, + p)) != APR_SUCCESS)) { + pollset->p = NULL; + return rv; + } +#else + if (flags & APR_POLLSET_THREADSAFE) { + pollset->p = NULL; + return APR_ENOTIMPL; + } +#endif + + /* POLLIN and POLLOUT are represented in different returned + * events, so we need 2 entries per descriptor in the result set, + * both for what is returned by kevent() and what is returned to + * the caller of apr_pollset_poll() (since it doesn't spend the + * CPU to coalesce separate APR_POLLIN and APR_POLLOUT events + * for the same descriptor) + */ + pollset->p->setsize = 2 * size; + + pollset->p->ke_set = + (struct kevent *) apr_palloc(p, pollset->p->setsize * sizeof(struct kevent)); + + memset(pollset->p->ke_set, 0, pollset->p->setsize * sizeof(struct kevent)); + + pollset->p->kqueue_fd = kqueue(); + + if (pollset->p->kqueue_fd == -1) { + pollset->p = NULL; + return apr_get_netos_error(); + } + + { + int flags; + + if ((flags = fcntl(pollset->p->kqueue_fd, F_GETFD)) == -1) { + rv = errno; + close(pollset->p->kqueue_fd); + pollset->p = NULL; + return rv; + } + + flags |= FD_CLOEXEC; + if (fcntl(pollset->p->kqueue_fd, F_SETFD, flags) == -1) { + rv = errno; + close(pollset->p->kqueue_fd); + pollset->p = NULL; + return rv; + } + } + + pollset->p->result_set = apr_palloc(p, pollset->p->setsize * sizeof(apr_pollfd_t)); + + APR_RING_INIT(&pollset->p->query_ring, pfd_elem_t, link); + APR_RING_INIT(&pollset->p->free_ring, pfd_elem_t, link); + APR_RING_INIT(&pollset->p->dead_ring, pfd_elem_t, link); + + return APR_SUCCESS; +} + +static apr_status_t impl_pollset_add(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + apr_os_sock_t fd; + pfd_elem_t *elem; + apr_status_t rv = APR_SUCCESS; + + pollset_lock_rings(); + + if (!APR_RING_EMPTY(&(pollset->p->free_ring), pfd_elem_t, link)) { + elem = APR_RING_FIRST(&(pollset->p->free_ring)); + APR_RING_REMOVE(elem, link); + } + else { + elem = (pfd_elem_t *) apr_palloc(pollset->pool, sizeof(pfd_elem_t)); + APR_RING_ELEM_INIT(elem, link); + } + elem->pfd = *descriptor; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + if (descriptor->reqevents & APR_POLLIN) { + EV_SET(&pollset->p->kevent, fd, EVFILT_READ, EV_ADD, 0, 0, elem); + + if (kevent(pollset->p->kqueue_fd, &pollset->p->kevent, 1, NULL, 0, + NULL) == -1) { + rv = apr_get_netos_error(); + } + } + + if (descriptor->reqevents & APR_POLLOUT && rv == APR_SUCCESS) { + EV_SET(&pollset->p->kevent, fd, EVFILT_WRITE, EV_ADD, 0, 0, elem); + + if (kevent(pollset->p->kqueue_fd, &pollset->p->kevent, 1, NULL, 0, + NULL) == -1) { + rv = apr_get_netos_error(); + } + } + + if (rv == APR_SUCCESS) { + APR_RING_INSERT_TAIL(&(pollset->p->query_ring), elem, pfd_elem_t, link); + } + else { + APR_RING_INSERT_TAIL(&(pollset->p->free_ring), elem, pfd_elem_t, link); + } + + pollset_unlock_rings(); + + return rv; +} + +static apr_status_t impl_pollset_remove(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + pfd_elem_t *ep; + apr_status_t rv; + apr_os_sock_t fd; + + pollset_lock_rings(); + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + rv = APR_NOTFOUND; /* unless at least one of the specified conditions is */ + if (descriptor->reqevents & APR_POLLIN) { + EV_SET(&pollset->p->kevent, fd, EVFILT_READ, EV_DELETE, 0, 0, NULL); + + if (kevent(pollset->p->kqueue_fd, &pollset->p->kevent, 1, NULL, 0, + NULL) != -1) { + rv = APR_SUCCESS; + } + } + + if (descriptor->reqevents & APR_POLLOUT) { + EV_SET(&pollset->p->kevent, fd, EVFILT_WRITE, EV_DELETE, 0, 0, NULL); + + if (kevent(pollset->p->kqueue_fd, &pollset->p->kevent, 1, NULL, 0, + NULL) != -1) { + rv = APR_SUCCESS; + } + } + + for (ep = APR_RING_FIRST(&(pollset->p->query_ring)); + ep != APR_RING_SENTINEL(&(pollset->p->query_ring), + pfd_elem_t, link); + ep = APR_RING_NEXT(ep, link)) { + + if (descriptor->desc.s == ep->pfd.desc.s) { + APR_RING_REMOVE(ep, link); + APR_RING_INSERT_TAIL(&(pollset->p->dead_ring), + ep, pfd_elem_t, link); + break; + } + } + + pollset_unlock_rings(); + + return rv; +} + +static apr_status_t impl_pollset_poll(apr_pollset_t *pollset, + apr_interval_time_t timeout, + apr_int32_t *num, + const apr_pollfd_t **descriptors) +{ + int ret; + struct timespec tv, *tvptr; + apr_status_t rv = APR_SUCCESS; + + *num = 0; + + if (timeout < 0) { + tvptr = NULL; + } + else { + tv.tv_sec = (long) apr_time_sec(timeout); + tv.tv_nsec = (long) apr_time_usec(timeout) * 1000; + tvptr = &tv; + } + + ret = kevent(pollset->p->kqueue_fd, NULL, 0, pollset->p->ke_set, + pollset->p->setsize, tvptr); + if (ret < 0) { + rv = apr_get_netos_error(); + } + else if (ret == 0) { + rv = APR_TIMEUP; + } + else { + int i, j; + const apr_pollfd_t *fd; + + for (i = 0, j = 0; i < ret; i++) { + 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_poll_drain_wakeup_pipe(pollset->wakeup_pipe); + rv = APR_EINTR; + } + else { + pollset->p->result_set[j] = *fd; + pollset->p->result_set[j].rtnevents = + get_kqueue_revent(pollset->p->ke_set[i].filter, + pollset->p->ke_set[i].flags); + j++; + } + } + if ((*num = j)) { /* any event besides wakeup pipe? */ + rv = APR_SUCCESS; + if (descriptors) { + *descriptors = pollset->p->result_set; + } + } + } + + pollset_lock_rings(); + + /* Shift all PFDs in the Dead Ring to the Free Ring */ + APR_RING_CONCAT(&(pollset->p->free_ring), &(pollset->p->dead_ring), + pfd_elem_t, link); + + pollset_unlock_rings(); + + return rv; +} + +static const apr_pollset_provider_t impl = { + impl_pollset_create, + impl_pollset_add, + impl_pollset_remove, + impl_pollset_poll, + impl_pollset_cleanup, + "kqueue" +}; + +const apr_pollset_provider_t *apr_pollset_provider_kqueue = &impl; + +static apr_status_t impl_pollcb_cleanup(apr_pollcb_t *pollcb) +{ + close(pollcb->fd); + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_create(apr_pollcb_t *pollcb, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + int fd; + + fd = kqueue(); + if (fd < 0) { + return apr_get_netos_error(); + } + + { + int flags; + apr_status_t rv; + + if ((flags = fcntl(fd, F_GETFD)) == -1) { + rv = errno; + close(fd); + pollcb->fd = -1; + return rv; + } + + flags |= FD_CLOEXEC; + if (fcntl(fd, F_SETFD, flags) == -1) { + rv = errno; + close(fd); + pollcb->fd = -1; + return rv; + } + } + + pollcb->fd = fd; + pollcb->pollset.ke = (struct kevent *) apr_pcalloc(p, 2 * size * sizeof(struct kevent)); + + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_add(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + apr_os_sock_t fd; + struct kevent ev; + apr_status_t rv = APR_SUCCESS; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + if (descriptor->reqevents & APR_POLLIN) { + EV_SET(&ev, fd, EVFILT_READ, EV_ADD, 0, 0, descriptor); + + if (kevent(pollcb->fd, &ev, 1, NULL, 0, NULL) == -1) { + rv = apr_get_netos_error(); + } + } + + if (descriptor->reqevents & APR_POLLOUT && rv == APR_SUCCESS) { + EV_SET(&ev, fd, EVFILT_WRITE, EV_ADD, 0, 0, descriptor); + + if (kevent(pollcb->fd, &ev, 1, NULL, 0, NULL) == -1) { + rv = apr_get_netos_error(); + } + } + + return rv; +} + +static apr_status_t impl_pollcb_remove(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + apr_status_t rv; + struct kevent ev; + apr_os_sock_t fd; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + rv = APR_NOTFOUND; /* unless at least one of the specified conditions is */ + if (descriptor->reqevents & APR_POLLIN) { + EV_SET(&ev, fd, EVFILT_READ, EV_DELETE, 0, 0, NULL); + + if (kevent(pollcb->fd, &ev, 1, NULL, 0, NULL) != -1) { + rv = APR_SUCCESS; + } + } + + if (descriptor->reqevents & APR_POLLOUT) { + EV_SET(&ev, fd, EVFILT_WRITE, EV_DELETE, 0, 0, NULL); + + if (kevent(pollcb->fd, &ev, 1, NULL, 0, NULL) != -1) { + rv = APR_SUCCESS; + } + } + + return rv; +} + + +static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb, + apr_interval_time_t timeout, + apr_pollcb_cb_t func, + void *baton) +{ + int ret, i; + struct timespec tv, *tvptr; + apr_status_t rv = APR_SUCCESS; + + if (timeout < 0) { + tvptr = NULL; + } + else { + tv.tv_sec = (long) apr_time_sec(timeout); + tv.tv_nsec = (long) apr_time_usec(timeout) * 1000; + tvptr = &tv; + } + + ret = kevent(pollcb->fd, NULL, 0, pollcb->pollset.ke, 2 * pollcb->nalloc, + tvptr); + + if (ret < 0) { + rv = apr_get_netos_error(); + } + else if (ret == 0) { + rv = APR_TIMEUP; + } + 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]) { + apr_poll_drain_wakeup_pipe(pollcb->wakeup_pipe); + return APR_EINTR; + } + + pollfd->rtnevents = get_kqueue_revent(pollcb->pollset.ke[i].filter, + pollcb->pollset.ke[i].flags); + + rv = func(baton, pollfd); + + if (rv) { + return rv; + } + } + } + + return rv; +} + +static const apr_pollcb_provider_t impl_cb = { + impl_pollcb_create, + impl_pollcb_add, + impl_pollcb_remove, + impl_pollcb_poll, + impl_pollcb_cleanup, + "kqueue" +}; + +const apr_pollcb_provider_t *apr_pollcb_provider_kqueue = &impl_cb; + +#endif /* HAVE_KQUEUE */ diff --git a/poll/unix/poll.c b/poll/unix/poll.c new file mode 100644 index 0000000..5b878f1 --- /dev/null +++ b/poll/unix/poll.c @@ -0,0 +1,460 @@ +/* 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_misc.h" +#include "apr_arch_poll_private.h" + +#if defined(HAVE_POLL) + +#ifdef HAVE_ALLOCA_H +#include <alloca.h> +#endif + +static apr_int16_t get_event(apr_int16_t event) +{ + apr_int16_t rv = 0; + + if (event & APR_POLLIN) + rv |= POLLIN; + if (event & APR_POLLPRI) + rv |= POLLPRI; + if (event & APR_POLLOUT) + rv |= POLLOUT; + /* POLLERR, POLLHUP, and POLLNVAL aren't valid as requested events */ + + return rv; +} + +static apr_int16_t get_revent(apr_int16_t event) +{ + apr_int16_t rv = 0; + + if (event & POLLIN) + rv |= APR_POLLIN; + if (event & POLLPRI) + rv |= APR_POLLPRI; + if (event & POLLOUT) + rv |= APR_POLLOUT; + if (event & POLLERR) + rv |= APR_POLLERR; + if (event & POLLHUP) + rv |= APR_POLLHUP; + if (event & POLLNVAL) + rv |= APR_POLLNVAL; + + return rv; +} + +#ifdef POLL_USES_POLL + +#define SMALL_POLLSET_LIMIT 8 + +APR_DECLARE(apr_status_t) apr_poll(apr_pollfd_t *aprset, apr_int32_t num, + apr_int32_t *nsds, + apr_interval_time_t timeout) +{ + int i, num_to_poll; +#ifdef HAVE_VLA + /* XXX: I trust that this is a segv when insufficient stack exists? */ + struct pollfd pollset[num + 1]; /* +1 since allocating 0 is undefined behaviour */ +#elif defined(HAVE_ALLOCA) + struct pollfd *pollset = alloca(sizeof(struct pollfd) * num); + if (!pollset) + return APR_ENOMEM; +#else + struct pollfd tmp_pollset[SMALL_POLLSET_LIMIT]; + struct pollfd *pollset; + + if (num <= SMALL_POLLSET_LIMIT) { + pollset = tmp_pollset; + } + else { + /* This does require O(n) to copy the descriptors to the internal + * mapping. + */ + pollset = malloc(sizeof(struct pollfd) * num); + /* The other option is adding an apr_pool_abort() fn to invoke + * the pool's out of memory handler + */ + if (!pollset) + return APR_ENOMEM; + } +#endif + for (i = 0; i < num; i++) { + if (aprset[i].desc_type == APR_POLL_SOCKET) { + pollset[i].fd = aprset[i].desc.s->socketdes; + } + else if (aprset[i].desc_type == APR_POLL_FILE) { + pollset[i].fd = aprset[i].desc.f->filedes; + } + else { + break; + } + pollset[i].events = get_event(aprset[i].reqevents); + } + num_to_poll = i; + + if (timeout > 0) { + timeout /= 1000; /* convert microseconds to milliseconds */ + } + + i = poll(pollset, num_to_poll, timeout); + (*nsds) = i; + + if (i > 0) { /* poll() sets revents only if an event was signalled; + * we don't promise to set rtnevents unless an event + * was signalled + */ + for (i = 0; i < num; i++) { + aprset[i].rtnevents = get_revent(pollset[i].revents); + } + } + +#if !defined(HAVE_VLA) && !defined(HAVE_ALLOCA) + if (num > SMALL_POLLSET_LIMIT) { + free(pollset); + } +#endif + + if ((*nsds) < 0) { + return apr_get_netos_error(); + } + if ((*nsds) == 0) { + return APR_TIMEUP; + } + return APR_SUCCESS; +} + + +#endif /* POLL_USES_POLL */ + +struct apr_pollset_private_t +{ + struct pollfd *pollset; + apr_pollfd_t *query_set; + apr_pollfd_t *result_set; +}; + +static apr_status_t impl_pollset_create(apr_pollset_t *pollset, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + if (flags & APR_POLLSET_THREADSAFE) { + return APR_ENOTIMPL; + } +#ifdef WIN32 + if (!APR_HAVE_LATE_DLL_FUNC(WSAPoll)) { + return APR_ENOTIMPL; + } +#endif + pollset->p = apr_palloc(p, sizeof(apr_pollset_private_t)); + pollset->p->pollset = apr_palloc(p, size * sizeof(struct pollfd)); + pollset->p->query_set = apr_palloc(p, size * sizeof(apr_pollfd_t)); + pollset->p->result_set = apr_palloc(p, size * sizeof(apr_pollfd_t)); + + return APR_SUCCESS; +} + +static apr_status_t impl_pollset_add(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + if (pollset->nelts == pollset->nalloc) { + return APR_ENOMEM; + } + + pollset->p->query_set[pollset->nelts] = *descriptor; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + pollset->p->pollset[pollset->nelts].fd = descriptor->desc.s->socketdes; + } + else { +#if APR_FILES_AS_SOCKETS + pollset->p->pollset[pollset->nelts].fd = descriptor->desc.f->filedes; +#else + if ((pollset->flags & APR_POLLSET_WAKEABLE) && + descriptor->desc.f == pollset->wakeup_pipe[0]) + pollset->p->pollset[pollset->nelts].fd = (SOCKET)descriptor->desc.f->filedes; + else + return APR_EBADF; +#endif + } + pollset->p->pollset[pollset->nelts].events = + get_event(descriptor->reqevents); + pollset->nelts++; + + return APR_SUCCESS; +} + +static apr_status_t impl_pollset_remove(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + apr_uint32_t i; + + for (i = 0; i < pollset->nelts; i++) { + if (descriptor->desc.s == pollset->p->query_set[i].desc.s) { + /* Found an instance of the fd: remove this and any other copies */ + apr_uint32_t dst = i; + apr_uint32_t old_nelts = pollset->nelts; + pollset->nelts--; + for (i++; i < old_nelts; i++) { + if (descriptor->desc.s == pollset->p->query_set[i].desc.s) { + pollset->nelts--; + } + else { + pollset->p->pollset[dst] = pollset->p->pollset[i]; + pollset->p->query_set[dst] = pollset->p->query_set[i]; + dst++; + } + } + return APR_SUCCESS; + } + } + + return APR_NOTFOUND; +} + +static apr_status_t impl_pollset_poll(apr_pollset_t *pollset, + apr_interval_time_t timeout, + apr_int32_t *num, + const apr_pollfd_t **descriptors) +{ + int ret; + apr_status_t rv = APR_SUCCESS; + + *num = 0; + +#ifdef WIN32 + /* WSAPoll() requires at least one socket. */ + if (pollset->nelts == 0) { + if (timeout > 0) { + apr_sleep(timeout); + return APR_TIMEUP; + } + return APR_SUCCESS; + } + if (timeout > 0) { + timeout /= 1000; + } + ret = WSAPoll(pollset->p->pollset, pollset->nelts, (int)timeout); +#else + if (timeout > 0) { + timeout /= 1000; + } + ret = poll(pollset->p->pollset, pollset->nelts, timeout); +#endif + if (ret < 0) { + return apr_get_netos_error(); + } + else if (ret == 0) { + 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 + * wakeup pipe. In that case do not put it result set. + */ + 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); + rv = APR_EINTR; + } + else { + pollset->p->result_set[j] = pollset->p->query_set[i]; + pollset->p->result_set[j].rtnevents = + get_revent(pollset->p->pollset[i].revents); + j++; + } + } + } + if ((*num = j)) { /* any event besides wakeup pipe? */ + rv = APR_SUCCESS; + } + } + if (descriptors && (*num)) + *descriptors = pollset->p->result_set; + return rv; +} + +static const apr_pollset_provider_t impl = { + impl_pollset_create, + impl_pollset_add, + impl_pollset_remove, + impl_pollset_poll, + NULL, + "poll" +}; + +const apr_pollset_provider_t *apr_pollset_provider_poll = &impl; + +/* Poll method pollcb. + * This is probably usable only for WIN32 having WSAPoll + */ +static apr_status_t impl_pollcb_create(apr_pollcb_t *pollcb, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ +#if APR_HAS_THREADS + return APR_ENOTIMPL; +#else + pollcb->fd = -1; +#ifdef WIN32 + if (!APR_HAVE_LATE_DLL_FUNC(WSAPoll)) { + return APR_ENOTIMPL; + } +#endif + + pollcb->pollset.ps = apr_palloc(p, size * sizeof(struct pollfd)); + pollcb->copyset = apr_palloc(p, size * sizeof(apr_pollfd_t *)); + + return APR_SUCCESS; +#endif +} + +static apr_status_t impl_pollcb_add(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + if (pollcb->nelts == pollcb->nalloc) { + return APR_ENOMEM; + } + + if (descriptor->desc_type == APR_POLL_SOCKET) { + pollcb->pollset.ps[pollcb->nelts].fd = descriptor->desc.s->socketdes; + } + else { +#if APR_FILES_AS_SOCKETS + pollcb->pollset.ps[pollcb->nelts].fd = descriptor->desc.f->filedes; +#else + return APR_EBADF; +#endif + } + + pollcb->pollset.ps[pollcb->nelts].events = + get_event(descriptor->reqevents); + pollcb->copyset[pollcb->nelts] = descriptor; + pollcb->nelts++; + + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_remove(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + apr_uint32_t i; + + for (i = 0; i < pollcb->nelts; i++) { + if (descriptor->desc.s == pollcb->copyset[i]->desc.s) { + /* Found an instance of the fd: remove this and any other copies */ + apr_uint32_t dst = i; + apr_uint32_t old_nelts = pollcb->nelts; + pollcb->nelts--; + for (i++; i < old_nelts; i++) { + if (descriptor->desc.s == pollcb->copyset[i]->desc.s) { + pollcb->nelts--; + } + else { + pollcb->pollset.ps[dst] = pollcb->pollset.ps[i]; + pollcb->copyset[dst] = pollcb->copyset[i]; + dst++; + } + } + return APR_SUCCESS; + } + } + + return APR_NOTFOUND; +} + +static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb, + apr_interval_time_t timeout, + apr_pollcb_cb_t func, + void *baton) +{ + int ret; + apr_status_t rv = APR_SUCCESS; + apr_uint32_t i; + +#ifdef WIN32 + /* WSAPoll() requires at least one socket. */ + if (pollcb->nelts == 0) { + if (timeout > 0) { + apr_sleep(timeout); + return APR_TIMEUP; + } + return APR_SUCCESS; + } + if (timeout > 0) { + timeout /= 1000; + } + ret = WSAPoll(pollcb->pollset.ps, pollcb->nelts, (int)timeout); +#else + if (timeout > 0) { + timeout /= 1000; + } + ret = poll(pollcb->pollset.ps, pollcb->nelts, timeout); +#endif + if (ret < 0) { + return apr_get_netos_error(); + } + else if (ret == 0) { + return APR_TIMEUP; + } + else { + 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]) { + apr_poll_drain_wakeup_pipe(pollcb->wakeup_pipe); + return APR_EINTR; + } + + pollfd->rtnevents = get_revent(pollcb->pollset.ps[i].revents); + rv = func(baton, pollfd); + if (rv) { + return rv; + } + } + } + } + return rv; +} + +static const apr_pollcb_provider_t impl_cb = { + impl_pollcb_create, + impl_pollcb_add, + impl_pollcb_remove, + impl_pollcb_poll, + NULL, + "poll" +}; + +const apr_pollcb_provider_t *apr_pollcb_provider_poll = &impl_cb; + +#endif /* HAVE_POLL */ diff --git a/poll/unix/pollcb.c b/poll/unix/pollcb.c new file mode 100644 index 0000000..a63ad5c --- /dev/null +++ b/poll/unix/pollcb.c @@ -0,0 +1,224 @@ +/* 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. + */ + +#ifdef WIN32 +/* POSIX defines 1024 for the FD_SETSIZE */ +#define FD_SETSIZE 1024 +#endif + +#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" + +static apr_pollset_method_e pollset_default_method = POLLSET_DEFAULT_METHOD; +#if defined(HAVE_KQUEUE) +extern const apr_pollcb_provider_t *apr_pollcb_provider_kqueue; +#endif +#if defined(HAVE_PORT_CREATE) +extern const apr_pollcb_provider_t *apr_pollcb_provider_port; +#endif +#if defined(HAVE_EPOLL) +extern const apr_pollcb_provider_t *apr_pollcb_provider_epoll; +#endif +#if defined(HAVE_POLL) +extern const apr_pollcb_provider_t *apr_pollcb_provider_poll; +#endif + +static const apr_pollcb_provider_t *pollcb_provider(apr_pollset_method_e method) +{ + const apr_pollcb_provider_t *provider = NULL; + switch (method) { + case APR_POLLSET_KQUEUE: +#if defined(HAVE_KQUEUE) + provider = apr_pollcb_provider_kqueue; +#endif + break; + case APR_POLLSET_PORT: +#if defined(HAVE_PORT_CREATE) + provider = apr_pollcb_provider_port; +#endif + break; + case APR_POLLSET_EPOLL: +#if defined(HAVE_EPOLL) + provider = apr_pollcb_provider_epoll; +#endif + break; + case APR_POLLSET_POLL: +#if defined(HAVE_POLL) + provider = apr_pollcb_provider_poll; +#endif + break; + case APR_POLLSET_SELECT: + case APR_POLLSET_AIO_MSGQ: + case APR_POLLSET_DEFAULT: + break; + } + 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) { + apr_poll_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, + apr_uint32_t flags, + apr_pollset_method_e method) +{ + apr_status_t rv; + apr_pollcb_t *pollcb; + const apr_pollcb_provider_t *provider = NULL; + + *ret_pollcb = NULL; + + #ifdef WIN32 + /* This will work only if ws2_32.dll has WSAPoll funtion. + * We could check the presence of the function here, + * but someone might implement other pollcb method in + * the future. + */ + if (method == APR_POLLSET_DEFAULT) { + method = APR_POLLSET_POLL; + } + #endif + + if (method == APR_POLLSET_DEFAULT) + method = pollset_default_method; + while (provider == NULL) { + provider = pollcb_provider(method); + if (!provider) { + if ((flags & APR_POLLSET_NODEFAULT) == APR_POLLSET_NODEFAULT) + return APR_ENOTIMPL; + if (method == pollset_default_method) + return APR_ENOTIMPL; + method = pollset_default_method; + } + } + + 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; + + rv = (*provider->create)(pollcb, size, p, flags); + if (rv == APR_ENOTIMPL) { + if (method == pollset_default_method) { + return rv; + } + + if ((flags & APR_POLLSET_NODEFAULT) == APR_POLLSET_NODEFAULT) { + return rv; + } + + /* Try with default provider */ + provider = pollcb_provider(pollset_default_method); + if (!provider) { + return APR_ENOTIMPL; + } + rv = (*provider->create)(pollcb, size, p, flags); + if (rv != APR_SUCCESS) { + return rv; + } + pollcb->provider = provider; + } + else if (rv != APR_SUCCESS) { + return rv; + } + + if (flags & APR_POLLSET_WAKEABLE) { + /* Create wakeup pipe */ + if ((rv = apr_poll_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; +} + +APR_DECLARE(apr_status_t) apr_pollcb_create(apr_pollcb_t **pollcb, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + apr_pollset_method_e method = APR_POLLSET_DEFAULT; + return apr_pollcb_create_ex(pollcb, size, p, flags, method); +} + +APR_DECLARE(apr_status_t) apr_pollcb_add(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + return (*pollcb->provider->add)(pollcb, descriptor); +} + +APR_DECLARE(apr_status_t) apr_pollcb_remove(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + return (*pollcb->provider->remove)(pollcb, descriptor); +} + + +APR_DECLARE(apr_status_t) apr_pollcb_poll(apr_pollcb_t *pollcb, + apr_interval_time_t timeout, + apr_pollcb_cb_t func, + void *baton) +{ + 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 new file mode 100644 index 0000000..11b5736 --- /dev/null +++ b/poll/unix/pollset.c @@ -0,0 +1,247 @@ +/* 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. + */ + +#ifdef WIN32 +/* POSIX defines 1024 for the FD_SETSIZE */ +#define FD_SETSIZE 1024 +#endif + +#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" + +static apr_pollset_method_e pollset_default_method = POLLSET_DEFAULT_METHOD; + +static apr_status_t pollset_cleanup(void *p) +{ + apr_pollset_t *pollset = (apr_pollset_t *) p; + if (pollset->provider->cleanup) { + (*pollset->provider->cleanup)(pollset); + } + if (pollset->flags & APR_POLLSET_WAKEABLE) { + apr_poll_close_wakeup_pipe(pollset->wakeup_pipe); + } + + return APR_SUCCESS; +} + +#if defined(HAVE_KQUEUE) +extern const apr_pollset_provider_t *apr_pollset_provider_kqueue; +#endif +#if defined(HAVE_PORT_CREATE) +extern const apr_pollset_provider_t *apr_pollset_provider_port; +#endif +#if defined(HAVE_EPOLL) +extern const apr_pollset_provider_t *apr_pollset_provider_epoll; +#endif +#if defined(HAVE_AIO_MSGQ) +extern const apr_pollset_provider_t *apr_pollset_provider_aio_msgq; +#endif +#if defined(HAVE_POLL) +extern const apr_pollset_provider_t *apr_pollset_provider_poll; +#endif +extern const apr_pollset_provider_t *apr_pollset_provider_select; + +static const apr_pollset_provider_t *pollset_provider(apr_pollset_method_e method) +{ + const apr_pollset_provider_t *provider = NULL; + switch (method) { + case APR_POLLSET_KQUEUE: +#if defined(HAVE_KQUEUE) + provider = apr_pollset_provider_kqueue; +#endif + break; + case APR_POLLSET_PORT: +#if defined(HAVE_PORT_CREATE) + provider = apr_pollset_provider_port; +#endif + break; + case APR_POLLSET_EPOLL: +#if defined(HAVE_EPOLL) + provider = apr_pollset_provider_epoll; +#endif + break; + case APR_POLLSET_AIO_MSGQ: +#if defined(HAVE_AIO_MSGQ) + provider = apr_pollset_provider_aio_msgq; +#endif + break; + case APR_POLLSET_POLL: +#if defined(HAVE_POLL) + provider = apr_pollset_provider_poll; +#endif + break; + case APR_POLLSET_SELECT: + provider = apr_pollset_provider_select; + break; + case APR_POLLSET_DEFAULT: + break; + } + return provider; +} + +APR_DECLARE(apr_status_t) apr_pollset_create_ex(apr_pollset_t **ret_pollset, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags, + apr_pollset_method_e method) +{ + apr_status_t rv; + apr_pollset_t *pollset; + const apr_pollset_provider_t *provider = NULL; + + *ret_pollset = NULL; + + #ifdef WIN32 + /* Favor WSAPoll if supported. + * This will work only if ws2_32.dll has WSAPoll funtion. + * In other cases it will fall back to select() method unless + * the APR_POLLSET_NODEFAULT is added to the flags. + */ + if (method == APR_POLLSET_DEFAULT) { + method = APR_POLLSET_POLL; + } + #endif + + if (method == APR_POLLSET_DEFAULT) + method = pollset_default_method; + while (provider == NULL) { + provider = pollset_provider(method); + if (!provider) { + if ((flags & APR_POLLSET_NODEFAULT) == APR_POLLSET_NODEFAULT) + return APR_ENOTIMPL; + if (method == pollset_default_method) + return APR_ENOTIMPL; + method = pollset_default_method; + } + } + if (flags & APR_POLLSET_WAKEABLE) { + /* Add room for wakeup descriptor */ + size++; + } + + pollset = apr_palloc(p, sizeof(*pollset)); + pollset->nelts = 0; + pollset->nalloc = size; + pollset->pool = p; + pollset->flags = flags; + pollset->provider = provider; + + rv = (*provider->create)(pollset, size, p, flags); + if (rv == APR_ENOTIMPL) { + if (method == pollset_default_method) { + return rv; + } + provider = pollset_provider(pollset_default_method); + if (!provider) { + return APR_ENOTIMPL; + } + rv = (*provider->create)(pollset, size, p, flags); + if (rv != APR_SUCCESS) { + return rv; + } + pollset->provider = provider; + } + else if (rv != APR_SUCCESS) { + return rv; + } + if (flags & APR_POLLSET_WAKEABLE) { + /* Create wakeup pipe */ + if ((rv = apr_poll_create_wakeup_pipe(pollset->pool, &pollset->wakeup_pfd, + pollset->wakeup_pipe)) + != APR_SUCCESS) { + return rv; + } + + if ((rv = apr_pollset_add(pollset, &pollset->wakeup_pfd)) != APR_SUCCESS) { + return rv; + } + } + if ((flags & APR_POLLSET_WAKEABLE) || provider->cleanup) + apr_pool_cleanup_register(p, pollset, pollset_cleanup, + apr_pool_cleanup_null); + + *ret_pollset = pollset; + return APR_SUCCESS; +} + +APR_DECLARE(const char *) apr_pollset_method_name(apr_pollset_t *pollset) +{ + return pollset->provider->name; +} + +APR_DECLARE(const char *) apr_poll_method_defname(void) +{ + const apr_pollset_provider_t *provider = NULL; + + provider = pollset_provider(pollset_default_method); + if (provider) + return provider->name; + else + return "unknown"; +} + +APR_DECLARE(apr_status_t) apr_pollset_create(apr_pollset_t **pollset, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + apr_pollset_method_e method = APR_POLLSET_DEFAULT; + return apr_pollset_create_ex(pollset, size, p, flags, method); +} + +APR_DECLARE(apr_status_t) apr_pollset_destroy(apr_pollset_t * pollset) +{ + if (pollset->flags & APR_POLLSET_WAKEABLE || + pollset->provider->cleanup) + return apr_pool_cleanup_run(pollset->pool, pollset, + pollset_cleanup); + else + return APR_SUCCESS; +} + +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]); + else + return APR_EINIT; +} + +APR_DECLARE(apr_status_t) apr_pollset_add(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + return (*pollset->provider->add)(pollset, descriptor); +} + +APR_DECLARE(apr_status_t) apr_pollset_remove(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + return (*pollset->provider->remove)(pollset, descriptor); +} + +APR_DECLARE(apr_status_t) apr_pollset_poll(apr_pollset_t *pollset, + apr_interval_time_t timeout, + apr_int32_t *num, + const apr_pollfd_t **descriptors) +{ + return (*pollset->provider->poll)(pollset, timeout, num, descriptors); +} diff --git a/poll/unix/port.c b/poll/unix/port.c new file mode 100644 index 0000000..c1e5994 --- /dev/null +++ b/poll/unix/port.c @@ -0,0 +1,594 @@ +/* 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_atomic.h" +#include "apr_arch_file_io.h" +#include "apr_arch_networkio.h" +#include "apr_arch_poll_private.h" +#include "apr_arch_inherit.h" + +#if defined(HAVE_PORT_CREATE) + +static apr_int16_t get_event(apr_int16_t event) +{ + apr_int16_t rv = 0; + + if (event & APR_POLLIN) + rv |= POLLIN; + if (event & APR_POLLPRI) + rv |= POLLPRI; + if (event & APR_POLLOUT) + rv |= POLLOUT; + /* POLLERR, POLLHUP, and POLLNVAL aren't valid as requested events */ + + return rv; +} + +static apr_int16_t get_revent(apr_int16_t event) +{ + apr_int16_t rv = 0; + + if (event & POLLIN) + rv |= APR_POLLIN; + if (event & POLLPRI) + rv |= APR_POLLPRI; + if (event & POLLOUT) + rv |= APR_POLLOUT; + if (event & POLLERR) + rv |= APR_POLLERR; + if (event & POLLHUP) + rv |= APR_POLLHUP; + if (event & POLLNVAL) + rv |= APR_POLLNVAL; + + return rv; +} + + +struct apr_pollset_private_t +{ + int port_fd; + port_event_t *port_set; + apr_pollfd_t *result_set; +#if APR_HAS_THREADS + /* A thread mutex to protect operations on the rings */ + apr_thread_mutex_t *ring_lock; +#endif + /* A ring containing all of the pollfd_t that are active */ + APR_RING_HEAD(pfd_query_ring_t, pfd_elem_t) query_ring; + /* A ring containing the pollfd_t that will be added on the + * next call to apr_pollset_poll(). + */ + APR_RING_HEAD(pfd_add_ring_t, pfd_elem_t) add_ring; + /* A ring of pollfd_t that have been used, and then _remove'd */ + APR_RING_HEAD(pfd_free_ring_t, pfd_elem_t) free_ring; + /* A ring of pollfd_t where rings that have been _remove'd but + might still be inside a _poll */ + APR_RING_HEAD(pfd_dead_ring_t, pfd_elem_t) dead_ring; + /* number of threads in poll */ + volatile apr_uint32_t waiting; +}; + +static apr_status_t call_port_getn(int port, port_event_t list[], + unsigned int max, unsigned int *nget, + apr_interval_time_t timeout) +{ + struct timespec tv, *tvptr; + int ret; + apr_status_t rv = APR_SUCCESS; + + if (timeout < 0) { + tvptr = NULL; + } + else { + tv.tv_sec = (long) apr_time_sec(timeout); + tv.tv_nsec = (long) apr_time_usec(timeout) * 1000; + tvptr = &tv; + } + + list[0].portev_user = (void *)-1; /* so we can double check that an + * event was returned + */ + + ret = port_getn(port, list, max, nget, tvptr); + /* Note: 32-bit port_getn() on Solaris 10 x86 returns large negative + * values instead of 0 when returning immediately. + */ + + if (ret == -1) { + rv = apr_get_netos_error(); + + switch(rv) { + case EINTR: + case ETIME: + if (*nget > 0 && list[0].portev_user != (void *)-1) { + /* This confusing API can return an event at the same time + * that it reports EINTR or ETIME. If that occurs, just + * report the event. With EINTR, nget can be > 0 without + * any event, so check that portev_user was filled in. + * + * (Maybe it will be simplified; see thread + * http://mail.opensolaris.org + * /pipermail/networking-discuss/2009-August/011979.html + * This code will still work afterwards.) + */ + rv = APR_SUCCESS; + break; + } + if (rv == ETIME) { + rv = APR_TIMEUP; + } + /* fall-through */ + default: + *nget = 0; + } + } + else if (*nget == 0) { + rv = APR_TIMEUP; + } + + return rv; +} + +static apr_status_t impl_pollset_cleanup(apr_pollset_t *pollset) +{ + close(pollset->p->port_fd); + return APR_SUCCESS; +} + +static apr_status_t impl_pollset_create(apr_pollset_t *pollset, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + apr_status_t rv = APR_SUCCESS; + pollset->p = apr_palloc(p, sizeof(apr_pollset_private_t)); +#if APR_HAS_THREADS + if (flags & APR_POLLSET_THREADSAFE && + ((rv = apr_thread_mutex_create(&pollset->p->ring_lock, + APR_THREAD_MUTEX_DEFAULT, + p)) != APR_SUCCESS)) { + pollset->p = NULL; + return rv; + } +#else + if (flags & APR_POLLSET_THREADSAFE) { + pollset->p = NULL; + return APR_ENOTIMPL; + } +#endif + pollset->p->waiting = 0; + + pollset->p->port_set = apr_palloc(p, size * sizeof(port_event_t)); + + pollset->p->port_fd = port_create(); + + if (pollset->p->port_fd < 0) { + pollset->p = NULL; + return apr_get_netos_error(); + } + + { + int flags; + + if ((flags = fcntl(pollset->p->port_fd, F_GETFD)) == -1) { + rv = errno; + close(pollset->p->port_fd); + pollset->p = NULL; + return rv; + } + + flags |= FD_CLOEXEC; + if (fcntl(pollset->p->port_fd, F_SETFD, flags) == -1) { + rv = errno; + close(pollset->p->port_fd); + pollset->p = NULL; + return rv; + } + } + + pollset->p->result_set = apr_palloc(p, size * sizeof(apr_pollfd_t)); + + APR_RING_INIT(&pollset->p->query_ring, pfd_elem_t, link); + APR_RING_INIT(&pollset->p->add_ring, pfd_elem_t, link); + APR_RING_INIT(&pollset->p->free_ring, pfd_elem_t, link); + APR_RING_INIT(&pollset->p->dead_ring, pfd_elem_t, link); + + return rv; +} + +static apr_status_t impl_pollset_add(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + apr_os_sock_t fd; + pfd_elem_t *elem; + int res; + apr_status_t rv = APR_SUCCESS; + + pollset_lock_rings(); + + if (!APR_RING_EMPTY(&(pollset->p->free_ring), pfd_elem_t, link)) { + elem = APR_RING_FIRST(&(pollset->p->free_ring)); + APR_RING_REMOVE(elem, link); + } + else { + elem = (pfd_elem_t *) apr_palloc(pollset->pool, sizeof(pfd_elem_t)); + APR_RING_ELEM_INIT(elem, link); + elem->on_query_ring = 0; + } + elem->pfd = *descriptor; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + /* If another thread is polling, notify the kernel immediately; otherwise, + * wait until the next call to apr_pollset_poll(). + */ + if (apr_atomic_read32(&pollset->p->waiting)) { + res = port_associate(pollset->p->port_fd, PORT_SOURCE_FD, fd, + get_event(descriptor->reqevents), (void *)elem); + + if (res < 0) { + rv = apr_get_netos_error(); + APR_RING_INSERT_TAIL(&(pollset->p->free_ring), elem, pfd_elem_t, link); + } + else { + elem->on_query_ring = 1; + APR_RING_INSERT_TAIL(&(pollset->p->query_ring), elem, pfd_elem_t, link); + } + } + else { + APR_RING_INSERT_TAIL(&(pollset->p->add_ring), elem, pfd_elem_t, link); + } + + pollset_unlock_rings(); + + return rv; +} + +static apr_status_t impl_pollset_remove(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + apr_os_sock_t fd; + pfd_elem_t *ep; + apr_status_t rv = APR_SUCCESS; + int res; + int err = 0; + int found; + + pollset_lock_rings(); + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + /* Search the add ring first. This ring is often shorter, + * and it often contains the descriptor being removed. + * (For the common scenario where apr_pollset_poll() + * returns activity for the descriptor and the descriptor + * is then removed from the pollset, it will have just + * been moved to the add ring by apr_pollset_poll().) + * + * If it is on the add ring, it isn't associated with the + * event port yet/anymore. + */ + found = 0; + for (ep = APR_RING_FIRST(&(pollset->p->add_ring)); + ep != APR_RING_SENTINEL(&(pollset->p->add_ring), + pfd_elem_t, link); + ep = APR_RING_NEXT(ep, link)) { + + if (descriptor->desc.s == ep->pfd.desc.s) { + found = 1; + APR_RING_REMOVE(ep, link); + APR_RING_INSERT_TAIL(&(pollset->p->free_ring), + ep, pfd_elem_t, link); + break; + } + } + + if (!found) { + res = port_dissociate(pollset->p->port_fd, PORT_SOURCE_FD, fd); + + if (res < 0) { + /* The expected case for this failure is that another + * thread's call to port_getn() returned this fd and + * disassociated the fd from the event port, and + * impl_pollset_poll() is blocked on the ring lock, + * which this thread holds. + */ + err = errno; + rv = APR_NOTFOUND; + } + + for (ep = APR_RING_FIRST(&(pollset->p->query_ring)); + ep != APR_RING_SENTINEL(&(pollset->p->query_ring), + pfd_elem_t, link); + ep = APR_RING_NEXT(ep, link)) { + + if (descriptor->desc.s == ep->pfd.desc.s) { + APR_RING_REMOVE(ep, link); + ep->on_query_ring = 0; + APR_RING_INSERT_TAIL(&(pollset->p->dead_ring), + ep, pfd_elem_t, link); + if (ENOENT == err) { + rv = APR_SUCCESS; + } + break; + } + } + } + + pollset_unlock_rings(); + + return rv; +} + +static apr_status_t impl_pollset_poll(apr_pollset_t *pollset, + apr_interval_time_t timeout, + apr_int32_t *num, + const apr_pollfd_t **descriptors) +{ + apr_os_sock_t fd; + int ret; + unsigned int nget, i; + apr_int32_t j; + pfd_elem_t *ep; + apr_status_t rv = APR_SUCCESS; + + *num = 0; + nget = 1; + + pollset_lock_rings(); + + apr_atomic_inc32(&pollset->p->waiting); + + while (!APR_RING_EMPTY(&(pollset->p->add_ring), pfd_elem_t, link)) { + ep = APR_RING_FIRST(&(pollset->p->add_ring)); + APR_RING_REMOVE(ep, link); + + if (ep->pfd.desc_type == APR_POLL_SOCKET) { + fd = ep->pfd.desc.s->socketdes; + } + else { + fd = ep->pfd.desc.f->filedes; + } + + ret = port_associate(pollset->p->port_fd, PORT_SOURCE_FD, + fd, get_event(ep->pfd.reqevents), ep); + if (ret < 0) { + rv = apr_get_netos_error(); + APR_RING_INSERT_TAIL(&(pollset->p->free_ring), ep, pfd_elem_t, link); + break; + } + + ep->on_query_ring = 1; + APR_RING_INSERT_TAIL(&(pollset->p->query_ring), ep, pfd_elem_t, link); + } + + pollset_unlock_rings(); + + if (rv != APR_SUCCESS) { + apr_atomic_dec32(&pollset->p->waiting); + return rv; + } + + rv = call_port_getn(pollset->p->port_fd, pollset->p->port_set, + pollset->nalloc, &nget, timeout); + + /* decrease the waiting ASAP to reduce the window for calling + port_associate within apr_pollset_add() */ + apr_atomic_dec32(&pollset->p->waiting); + + pollset_lock_rings(); + + for (i = 0, j = 0; i < nget; i++) { + ep = (pfd_elem_t *)pollset->p->port_set[i].portev_user; + 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); + rv = APR_EINTR; + } + else { + pollset->p->result_set[j] = ep->pfd; + pollset->p->result_set[j].rtnevents = + get_revent(pollset->p->port_set[i].portev_events); + j++; + } + /* If the ring element is still on the query ring, move it + * to the add ring for re-association with the event port + * later. (It may have already been moved to the dead ring + * by a call to pollset_remove on another thread.) + */ + if (ep->on_query_ring) { + APR_RING_REMOVE(ep, link); + ep->on_query_ring = 0; + APR_RING_INSERT_TAIL(&(pollset->p->add_ring), ep, + pfd_elem_t, link); + } + } + if ((*num = j)) { /* any event besides wakeup pipe? */ + rv = APR_SUCCESS; + if (descriptors) { + *descriptors = pollset->p->result_set; + } + } + + /* Shift all PFDs in the Dead Ring to the Free Ring */ + APR_RING_CONCAT(&(pollset->p->free_ring), &(pollset->p->dead_ring), pfd_elem_t, link); + + pollset_unlock_rings(); + + return rv; +} + +static const apr_pollset_provider_t impl = { + impl_pollset_create, + impl_pollset_add, + impl_pollset_remove, + impl_pollset_poll, + impl_pollset_cleanup, + "port" +}; + +const apr_pollset_provider_t *apr_pollset_provider_port = &impl; + +static apr_status_t impl_pollcb_cleanup(apr_pollcb_t *pollcb) +{ + close(pollcb->fd); + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_create(apr_pollcb_t *pollcb, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + pollcb->fd = port_create(); + + if (pollcb->fd < 0) { + return apr_get_netos_error(); + } + + { + int flags; + apr_status_t rv; + + if ((flags = fcntl(pollcb->fd, F_GETFD)) == -1) { + rv = errno; + close(pollcb->fd); + pollcb->fd = -1; + return rv; + } + + flags |= FD_CLOEXEC; + if (fcntl(pollcb->fd, F_SETFD, flags) == -1) { + rv = errno; + close(pollcb->fd); + pollcb->fd = -1; + return rv; + } + } + + pollcb->pollset.port = apr_palloc(p, size * sizeof(port_event_t)); + + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_add(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + int ret, fd; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + ret = port_associate(pollcb->fd, PORT_SOURCE_FD, fd, + get_event(descriptor->reqevents), descriptor); + + if (ret == -1) { + return apr_get_netos_error(); + } + + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_remove(apr_pollcb_t *pollcb, + apr_pollfd_t *descriptor) +{ + int fd, ret; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + ret = port_dissociate(pollcb->fd, PORT_SOURCE_FD, fd); + + if (ret < 0) { + return APR_NOTFOUND; + } + + return APR_SUCCESS; +} + +static apr_status_t impl_pollcb_poll(apr_pollcb_t *pollcb, + apr_interval_time_t timeout, + apr_pollcb_cb_t func, + void *baton) +{ + apr_status_t rv; + 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++) { + apr_pollfd_t *pollfd = (apr_pollfd_t *)(pollcb->pollset.port[i].portev_user); + + 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); + return APR_EINTR; + } + + pollfd->rtnevents = get_revent(pollcb->pollset.port[i].portev_events); + + rv = func(baton, pollfd); + if (rv) { + return rv; + } + rv = apr_pollcb_add(pollcb, pollfd); + } + } + + return rv; +} + +static const apr_pollcb_provider_t impl_cb = { + impl_pollcb_create, + impl_pollcb_add, + impl_pollcb_remove, + impl_pollcb_poll, + impl_pollcb_cleanup, + "port" +}; + +const apr_pollcb_provider_t *apr_pollcb_provider_port = &impl_cb; + +#endif /* HAVE_PORT_CREATE */ diff --git a/poll/unix/select.c b/poll/unix/select.c new file mode 100644 index 0000000..51be3c1 --- /dev/null +++ b/poll/unix/select.c @@ -0,0 +1,449 @@ +/* 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. + */ + +#ifdef WIN32 +/* POSIX defines 1024 for the FD_SETSIZE */ +#define FD_SETSIZE 1024 +#endif + +#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" + +#ifdef POLL_USES_SELECT + +APR_DECLARE(apr_status_t) apr_poll(apr_pollfd_t *aprset, int num, + apr_int32_t *nsds, + apr_interval_time_t timeout) +{ + fd_set readset, writeset, exceptset; + int rv, i; + int maxfd = -1; + struct timeval tv, *tvptr; +#ifdef NETWARE + apr_datatype_e set_type = APR_NO_DESC; +#endif + +#ifdef WIN32 + /* On Win32, select() must be presented with at least one socket to + * poll on, or select() will return WSAEINVAL. So, we'll just + * short-circuit and bail now. + */ + if (num == 0) { + (*nsds) = 0; + if (timeout > 0) { + apr_sleep(timeout); + return APR_TIMEUP; + } + return APR_SUCCESS; + } +#endif + + if (timeout < 0) { + tvptr = NULL; + } + else { + tv.tv_sec = (long) apr_time_sec(timeout); + tv.tv_usec = (long) apr_time_usec(timeout); + tvptr = &tv; + } + + FD_ZERO(&readset); + FD_ZERO(&writeset); + FD_ZERO(&exceptset); + + for (i = 0; i < num; i++) { + apr_os_sock_t fd; + + aprset[i].rtnevents = 0; + + if (aprset[i].desc_type == APR_POLL_SOCKET) { +#ifdef NETWARE + if (HAS_PIPES(set_type)) { + return APR_EBADF; + } + else { + set_type = APR_POLL_SOCKET; + } +#endif + fd = aprset[i].desc.s->socketdes; + } + else if (aprset[i].desc_type == APR_POLL_FILE) { +#if !APR_FILES_AS_SOCKETS + return APR_EBADF; +#else +#ifdef NETWARE + if (aprset[i].desc.f->is_pipe && !HAS_SOCKETS(set_type)) { + set_type = APR_POLL_FILE; + } + else + return APR_EBADF; +#endif /* NETWARE */ + + fd = aprset[i].desc.f->filedes; + +#endif /* APR_FILES_AS_SOCKETS */ + } + else { + break; + } +#if !defined(WIN32) && !defined(NETWARE) /* socket sets handled with array of handles */ + if (fd >= FD_SETSIZE) { + /* XXX invent new error code so application has a clue */ + return APR_EBADF; + } +#endif + if (aprset[i].reqevents & APR_POLLIN) { + FD_SET(fd, &readset); + } + if (aprset[i].reqevents & APR_POLLOUT) { + FD_SET(fd, &writeset); + } + if (aprset[i].reqevents & + (APR_POLLPRI | APR_POLLERR | APR_POLLHUP | APR_POLLNVAL)) { + FD_SET(fd, &exceptset); + } + if ((int) fd > maxfd) { + maxfd = (int) fd; + } + } + +#ifdef NETWARE + if (HAS_PIPES(set_type)) { + rv = pipe_select(maxfd + 1, &readset, &writeset, &exceptset, tvptr); + } + else { +#endif + + rv = select(maxfd + 1, &readset, &writeset, &exceptset, tvptr); + +#ifdef NETWARE + } +#endif + + (*nsds) = rv; + if ((*nsds) == 0) { + return APR_TIMEUP; + } + if ((*nsds) < 0) { + return apr_get_netos_error(); + } + + (*nsds) = 0; + for (i = 0; i < num; i++) { + apr_os_sock_t fd; + + if (aprset[i].desc_type == APR_POLL_SOCKET) { + fd = aprset[i].desc.s->socketdes; + } + else if (aprset[i].desc_type == APR_POLL_FILE) { +#if !APR_FILES_AS_SOCKETS + return APR_EBADF; +#else + fd = aprset[i].desc.f->filedes; +#endif + } + else { + break; + } + if (FD_ISSET(fd, &readset)) { + aprset[i].rtnevents |= APR_POLLIN; + } + if (FD_ISSET(fd, &writeset)) { + aprset[i].rtnevents |= APR_POLLOUT; + } + if (FD_ISSET(fd, &exceptset)) { + aprset[i].rtnevents |= APR_POLLERR; + } + if (aprset[i].rtnevents) { + (*nsds)++; + } + } + + return APR_SUCCESS; +} + +#endif /* POLL_USES_SELECT */ + +struct apr_pollset_private_t +{ + fd_set readset, writeset, exceptset; + int maxfd; + apr_pollfd_t *query_set; + apr_pollfd_t *result_set; + apr_uint32_t flags; +#ifdef NETWARE + int set_type; +#endif +}; + +static apr_status_t impl_pollset_create(apr_pollset_t *pollset, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + if (flags & APR_POLLSET_THREADSAFE) { + pollset->p = NULL; + return APR_ENOTIMPL; + } +#ifdef FD_SETSIZE + if (size > FD_SETSIZE) { + pollset->p = NULL; + return APR_EINVAL; + } +#endif + pollset->p = apr_palloc(p, sizeof(apr_pollset_private_t)); + FD_ZERO(&(pollset->p->readset)); + FD_ZERO(&(pollset->p->writeset)); + FD_ZERO(&(pollset->p->exceptset)); + pollset->p->maxfd = 0; +#ifdef NETWARE + pollset->p->set_type = APR_NO_DESC; +#endif + pollset->p->query_set = apr_palloc(p, size * sizeof(apr_pollfd_t)); + pollset->p->result_set = apr_palloc(p, size * sizeof(apr_pollfd_t)); + + return APR_SUCCESS; +} + +static apr_status_t impl_pollset_add(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + apr_os_sock_t fd; + + if (pollset->nelts == pollset->nalloc) { + return APR_ENOMEM; + } + + pollset->p->query_set[pollset->nelts] = *descriptor; + + if (descriptor->desc_type == APR_POLL_SOCKET) { +#ifdef NETWARE + /* NetWare can't handle mixed descriptor types in select() */ + if (HAS_PIPES(pollset->p->set_type)) { + return APR_EBADF; + } + else { + pollset->p->set_type = APR_POLL_SOCKET; + } +#endif + fd = descriptor->desc.s->socketdes; + } + else { +#if !APR_FILES_AS_SOCKETS + if ((pollset->flags & APR_POLLSET_WAKEABLE) && + descriptor->desc.f == pollset->wakeup_pipe[0]) + fd = (apr_os_sock_t)descriptor->desc.f->filedes; + else + return APR_EBADF; +#else +#ifdef NETWARE + /* NetWare can't handle mixed descriptor types in select() */ + if (descriptor->desc.f->is_pipe && !HAS_SOCKETS(pollset->p->set_type)) { + pollset->p->set_type = APR_POLL_FILE; + fd = descriptor->desc.f->filedes; + } + else { + return APR_EBADF; + } +#else + fd = descriptor->desc.f->filedes; +#endif +#endif + } +#if !defined(WIN32) && !defined(NETWARE) /* socket sets handled with array of handles */ + if (fd >= FD_SETSIZE) { + /* XXX invent new error code so application has a clue */ + return APR_EBADF; + } +#endif + if (descriptor->reqevents & APR_POLLIN) { + FD_SET(fd, &(pollset->p->readset)); + } + if (descriptor->reqevents & APR_POLLOUT) { + FD_SET(fd, &(pollset->p->writeset)); + } + if (descriptor->reqevents & + (APR_POLLPRI | APR_POLLERR | APR_POLLHUP | APR_POLLNVAL)) { + FD_SET(fd, &(pollset->p->exceptset)); + } + if ((int) fd > pollset->p->maxfd) { + pollset->p->maxfd = (int) fd; + } + pollset->nelts++; + return APR_SUCCESS; +} + +static apr_status_t impl_pollset_remove(apr_pollset_t * pollset, + const apr_pollfd_t * descriptor) +{ + apr_uint32_t i; + apr_os_sock_t fd; + + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { +#if !APR_FILES_AS_SOCKETS + return APR_EBADF; +#else + fd = descriptor->desc.f->filedes; +#endif + } + + for (i = 0; i < pollset->nelts; i++) { + if (descriptor->desc.s == pollset->p->query_set[i].desc.s) { + /* Found an instance of the fd: remove this and any other copies */ + apr_uint32_t dst = i; + apr_uint32_t old_nelts = pollset->nelts; + pollset->nelts--; + for (i++; i < old_nelts; i++) { + if (descriptor->desc.s == pollset->p->query_set[i].desc.s) { + pollset->nelts--; + } + else { + pollset->p->query_set[dst] = pollset->p->query_set[i]; + dst++; + } + } + FD_CLR(fd, &(pollset->p->readset)); + FD_CLR(fd, &(pollset->p->writeset)); + FD_CLR(fd, &(pollset->p->exceptset)); + if (((int) fd == pollset->p->maxfd) && (pollset->p->maxfd > 0)) { + pollset->p->maxfd--; + } + return APR_SUCCESS; + } + } + + return APR_NOTFOUND; +} + +static apr_status_t impl_pollset_poll(apr_pollset_t *pollset, + apr_interval_time_t timeout, + apr_int32_t *num, + const apr_pollfd_t **descriptors) +{ + int rs; + apr_uint32_t i, j; + struct timeval tv, *tvptr; + fd_set readset, writeset, exceptset; + apr_status_t rv = APR_SUCCESS; + + *num = 0; + +#ifdef WIN32 + /* On Win32, select() must be presented with at least one socket to + * poll on, or select() will return WSAEINVAL. So, we'll just + * short-circuit and bail now. + */ + if (pollset->nelts == 0) { + if (timeout > 0) { + apr_sleep(timeout); + return APR_TIMEUP; + } + return APR_SUCCESS; + } +#endif + + if (timeout < 0) { + tvptr = NULL; + } + else { + tv.tv_sec = (long) apr_time_sec(timeout); + tv.tv_usec = (long) apr_time_usec(timeout); + tvptr = &tv; + } + + memcpy(&readset, &(pollset->p->readset), sizeof(fd_set)); + memcpy(&writeset, &(pollset->p->writeset), sizeof(fd_set)); + memcpy(&exceptset, &(pollset->p->exceptset), sizeof(fd_set)); + +#ifdef NETWARE + if (HAS_PIPES(pollset->p->set_type)) { + rs = pipe_select(pollset->p->maxfd + 1, &readset, &writeset, &exceptset, + tvptr); + } + else +#endif + rs = select(pollset->p->maxfd + 1, &readset, &writeset, &exceptset, + tvptr); + + if (rs < 0) { + return apr_get_netos_error(); + } + if (rs == 0) { + return APR_TIMEUP; + } + j = 0; + for (i = 0; i < pollset->nelts; i++) { + apr_os_sock_t fd; + if (pollset->p->query_set[i].desc_type == APR_POLL_SOCKET) { + fd = pollset->p->query_set[i].desc.s->socketdes; + } + 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); + rv = APR_EINTR; + continue; + } + else { +#if !APR_FILES_AS_SOCKETS + return APR_EBADF; +#else + fd = pollset->p->query_set[i].desc.f->filedes; +#endif + } + } + if (FD_ISSET(fd, &readset) || FD_ISSET(fd, &writeset) || + FD_ISSET(fd, &exceptset)) { + pollset->p->result_set[j] = pollset->p->query_set[i]; + pollset->p->result_set[j].rtnevents = 0; + if (FD_ISSET(fd, &readset)) { + pollset->p->result_set[j].rtnevents |= APR_POLLIN; + } + if (FD_ISSET(fd, &writeset)) { + pollset->p->result_set[j].rtnevents |= APR_POLLOUT; + } + if (FD_ISSET(fd, &exceptset)) { + pollset->p->result_set[j].rtnevents |= APR_POLLERR; + } + j++; + } + } + if (((*num) = j) != 0) + rv = APR_SUCCESS; + + if (descriptors) + *descriptors = pollset->p->result_set; + return rv; +} + +static const apr_pollset_provider_t impl = { + impl_pollset_create, + impl_pollset_add, + impl_pollset_remove, + impl_pollset_poll, + NULL, + "select" +}; + +const apr_pollset_provider_t *apr_pollset_provider_select = &impl; diff --git a/poll/unix/wakeup.c b/poll/unix/wakeup.c new file mode 100644 index 0000000..6c3dcd6 --- /dev/null +++ b/poll/unix/wakeup.c @@ -0,0 +1,152 @@ +/* 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 apr_poll_create_wakeup_pipe(apr_pool_t *pool, apr_pollfd_t *pfd, + apr_file_t **wakeup_pipe) +{ + apr_status_t rv; + + if ((rv = apr_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 apr_poll_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_socket_pipe_close(wakeup_pipe[0]); + wakeup_pipe[0] = NULL; + } + if (wakeup_pipe[1]) { + rv1 = apr_file_socket_pipe_close(wakeup_pipe[1]); + wakeup_pipe[1] = NULL; + } + return rv0 ? rv0 : rv1; +} + +#else /* !WIN32 */ + +apr_status_t apr_poll_create_wakeup_pipe(apr_pollfd_t *pfd, apr_file_t **wakeup_pipe) +{ + return APR_ENOTIMPL; +} + +apr_status_t apr_poll_close_wakeup_pipe(apr_file_t **wakeup_pipe) +{ + return APR_ENOTIMPL; +} + +#endif /* !WIN32 */ + +#else /* APR_FILES_AS_SOCKETS */ + +apr_status_t apr_poll_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_ex(&wakeup_pipe[0], &wakeup_pipe[1], + APR_WRITE_BLOCK, + 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 apr_poll_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 apr_poll_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; + } +} diff --git a/poll/unix/z_asio.c b/poll/unix/z_asio.c new file mode 100644 index 0000000..48b531c --- /dev/null +++ b/poll/unix/z_asio.c @@ -0,0 +1,782 @@ +/* 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. + * + * + ****************************************************************************** + * + * This implementation is based on a design by John Brooks (IBM Pok) which uses + * the z/OS sockets async i/o facility. When a + * socket is added to the pollset, an async poll is issued for that individual + * socket. It specifies that the kernel should send an IPC message when the + * socket becomes ready. The IPC messages are sent to a single message queue + * that is part of the pollset. apr_pollset_poll waits on the arrival of IPC + * messages or the specified timeout. + * + * Since z/OS does not support async i/o for pipes or files at present, this + * implementation falls back to using ordinary poll() when + * APR_POLLSET_THREADSAFE is unset. + * + * Greg Ames + * April 2012 + */ + +#include "apr.h" +#include "apr_hash.h" +#include "apr_poll.h" +#include "apr_time.h" +#include "apr_portable.h" +#include "apr_arch_inherit.h" +#include "apr_arch_file_io.h" +#include "apr_arch_networkio.h" +#include "apr_arch_poll_private.h" + +#ifdef HAVE_AIO_MSGQ + +#include <sys/msg.h> /* msgget etc */ +#include <time.h> /* timestruct */ +#include <poll.h> /* pollfd */ +#include <limits.h> /* MAX_INT */ + +struct apr_pollset_private_t +{ + int msg_q; /* IPC message queue. The z/OS kernel sends messages + * to this queue when our async polls on individual + * file descriptors complete + */ + apr_pollfd_t *result_set; + apr_uint32_t size; + +#if APR_HAS_THREADS + /* A thread mutex to protect operations on the rings and the hash */ + apr_thread_mutex_t *ring_lock; +#endif + + /* A hash of all active elements used for O(1) _remove operations */ + apr_hash_t *elems; + + APR_RING_HEAD(ready_ring_t, asio_elem_t) ready_ring; + APR_RING_HEAD(prior_ready_ring_t, asio_elem_t) prior_ready_ring; + APR_RING_HEAD(free_ring_t, asio_elem_t) free_ring; + + /* for pipes etc with no asio */ + struct pollfd *pollset; + apr_pollfd_t *query_set; +}; + +typedef enum { + ASIO_INIT = 0, + ASIO_REMOVED, + ASIO_COMPLETE +} asio_state_e; + +typedef struct asio_elem_t asio_elem_t; + +struct asio_msgbuf_t { + long msg_type; /* must be > 0 */ + asio_elem_t *msg_elem; +}; + +struct asio_elem_t +{ + APR_RING_ENTRY(asio_elem_t) link; + apr_pollfd_t pfd; + struct pollfd os_pfd; + struct aiocb a; + asio_state_e state; + struct asio_msgbuf_t msg; +}; + +#define DEBUG 0 + +/* DEBUG settings: 0 - no debug messages at all, + * 1 - should not occur messages, + * 2 - apr_pollset_* entry and exit messages, + * 3 - state changes, memory usage, + * 4 - z/OS, APR, and internal calls, + * 5 - everything else except the timer pop path, + * 6 - everything, including the Event 1 sec timer pop path + * + * each DEBUG level includes all messages produced by lower numbered levels + */ + +#if DEBUG + +#include <assert.h> +#include <unistd.h> /* getpid */ + +#define DBG_BUFF char dbg_msg_buff[256]; + +#define DBG_TEST(lvl) if (lvl <= DEBUG) { + +#define DBG_CORE(msg) sprintf(dbg_msg_buff, "% 8d " __FUNCTION__ \ + " " msg, getpid()), \ + fprintf(stderr, "%s", dbg_msg_buff); +#define DBG_CORE1(msg, var1) sprintf(dbg_msg_buff, "% 8d " __FUNCTION__ \ + " " msg, getpid(), var1), \ + fprintf(stderr, "%s", dbg_msg_buff); +#define DBG_CORE2(msg, var1, var2) sprintf(dbg_msg_buff, "% 8d " __FUNCTION__ \ + " " msg, getpid(), var1, var2), \ + fprintf(stderr, "%s", dbg_msg_buff); +#define DBG_CORE3(msg, var1, var2, var3) \ + sprintf(dbg_msg_buff, "% 8d " __FUNCTION__ \ + " " msg, getpid(), var1, var2, var3), \ + fprintf(stderr, "%s", dbg_msg_buff); +#define DBG_CORE4(msg, var1, var2, var3, var4) \ + sprintf(dbg_msg_buff, "% 8d " __FUNCTION__ \ + " " msg, getpid(), var1, var2, var3, var4),\ + fprintf(stderr, "%s", dbg_msg_buff); + +#define DBG_END } + +#define DBG(lvl, msg) DBG_TEST(lvl) \ + DBG_CORE(msg) \ + DBG_END + +#define DBG1(lvl, msg, var1) DBG_TEST(lvl) \ + DBG_CORE1(msg, var1) \ + DBG_END + +#define DBG2(lvl, msg, var1, var2) DBG_TEST(lvl) \ + DBG_CORE2(msg, var1, var2) \ + DBG_END + +#define DBG3(lvl, msg, var1, var2, var3) \ + DBG_TEST(lvl) \ + DBG_CORE3(msg, var1, var2, var3) \ + DBG_END + +#define DBG4(lvl, msg, var1, var2, var3, var4) \ + DBG_TEST(lvl) \ + DBG_CORE4(msg, var1, var2, var3, var4) \ + DBG_END + +#else /* DEBUG is 0 */ +#define DBG_BUFF +#define DBG(lvl, msg) ((void)0) +#define DBG1(lvl, msg, var1) ((void)0) +#define DBG2(lvl, msg, var1, var2) ((void)0) +#define DBG3(lvl, msg, var1, var2, var3) ((void)0) +#define DBG4(lvl, msg, var1, var2, var3, var4) ((void)0) + +#endif /* DEBUG */ + +static int asyncio(struct aiocb *a) +{ + DBG_BUFF + int rv; + +#ifdef _LP64 +#define AIO BPX4AIO +#else +#define AIO BPX1AIO +#endif + + AIO(sizeof(struct aiocb), a, &rv, &errno, __err2ad()); + DBG2(4, "BPX4AIO aiocb %p rv %d\n", + a, rv); +#ifdef DEBUG + if (rv < 0) { + DBG2(4, "errno %d errnojr %08x\n", + errno, *__err2ad()); + } +#endif + return rv; +} + +static apr_int16_t get_event(apr_int16_t event) +{ + DBG_BUFF + apr_int16_t rv = 0; + DBG(4, "entered\n"); + + if (event & APR_POLLIN) + rv |= POLLIN; + if (event & APR_POLLPRI) + rv |= POLLPRI; + if (event & APR_POLLOUT) + rv |= POLLOUT; + if (event & APR_POLLERR) + rv |= POLLERR; + if (event & APR_POLLHUP) + rv |= POLLHUP; + if (event & APR_POLLNVAL) + rv |= POLLNVAL; + + DBG(4, "exiting\n"); + return rv; +} + +static apr_int16_t get_revent(apr_int16_t event) +{ + DBG_BUFF + apr_int16_t rv = 0; + DBG(4, "entered\n"); + + if (event & POLLIN) + rv |= APR_POLLIN; + if (event & POLLPRI) + rv |= APR_POLLPRI; + if (event & POLLOUT) + rv |= APR_POLLOUT; + if (event & POLLERR) + rv |= APR_POLLERR; + if (event & POLLHUP) + rv |= APR_POLLHUP; + if (event & POLLNVAL) + rv |= APR_POLLNVAL; + + DBG(4, "exiting\n"); + return rv; +} + +static apr_status_t asio_pollset_cleanup(apr_pollset_t *pollset) +{ + DBG_BUFF + int rv; + + DBG(4, "entered\n"); + if (pollset->flags & APR_POLLSET_THREADSAFE) { + rv = msgctl(pollset->p->msg_q, IPC_RMID, NULL); + DBG1(4, "asio_pollset_cleanup: msgctl(IPC_RMID) returned %d\n", rv); + } + + return rv; +} + +static apr_status_t asio_pollset_create(apr_pollset_t *pollset, + apr_uint32_t size, + apr_pool_t *p, + apr_uint32_t flags) +{ + DBG_BUFF + apr_status_t rv; + apr_pollset_private_t *priv; + + DBG1(2, "entered, flags: %x\n", flags); + + priv = pollset->p = apr_pcalloc(p, sizeof(*priv)); + + if (flags & APR_POLLSET_THREADSAFE) { +#if APR_HAS_THREADS + if ((rv = apr_thread_mutex_create(&(priv->ring_lock), + APR_THREAD_MUTEX_DEFAULT, + p)) != APR_SUCCESS) { + DBG1(1, "apr_thread_mutex_create returned %d\n", rv); + pollset->p = NULL; + return rv; + } + rv = msgget(IPC_PRIVATE, S_IWUSR+S_IRUSR); /* user r/w perms */ + if (rv < 0) { +#if DEBUG + perror(__FUNCTION__ " msgget returned < 0 "); +#endif + pollset->p = NULL; + return rv; + } + + DBG2(4, "pollset %p msgget was OK, rv=%d\n", pollset, rv); + priv->msg_q = rv; + priv->elems = apr_hash_make(p); + + APR_RING_INIT(&priv->free_ring, asio_elem_t, link); + APR_RING_INIT(&priv->prior_ready_ring, asio_elem_t, link); + +#else /* APR doesn't have threads but caller wants a threadsafe pollset */ + pollset->p = NULL; + return APR_ENOTIMPL; +#endif + + } else { /* APR_POLLSET_THREADSAFE not set, i.e. no async i/o, + * init fields only needed in old style pollset + */ + + priv->pollset = apr_palloc(p, size * sizeof(struct pollfd)); + priv->query_set = apr_palloc(p, size * sizeof(apr_pollfd_t)); + + if ((!priv->pollset) || (!priv->query_set)) { + pollset->p = NULL; + return APR_ENOMEM; + } + } + + pollset->nelts = 0; + pollset->flags = flags; + pollset->pool = p; + priv->size = size; + priv->result_set = apr_palloc(p, size * sizeof(apr_pollfd_t)); + if (!priv->result_set) { + if (flags & APR_POLLSET_THREADSAFE) { + msgctl(priv->msg_q, IPC_RMID, NULL); + } + pollset->p = NULL; + return APR_ENOMEM; + } + + DBG2(2, "exiting, pollset: %p, type: %s\n", + pollset, + flags & APR_POLLSET_THREADSAFE ? "async" : "POSIX"); + + + return APR_SUCCESS; + +} /* end of asio_pollset_create */ + +static apr_status_t posix_add(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + DBG_BUFF + int fd; + apr_pool_t *p = pollset->pool; + apr_pollset_private_t *priv = pollset->p; + + DBG(4, "entered\n"); + + if (pollset->nelts == priv->size) { + return APR_ENOMEM; + } + + priv->query_set[pollset->nelts] = *descriptor; + if (descriptor->desc_type == APR_POLL_SOCKET) { + fd = descriptor->desc.s->socketdes; + } + else { + fd = descriptor->desc.f->filedes; + } + + priv->pollset[pollset->nelts].fd = fd; + + priv->pollset[pollset->nelts].events = + get_event(descriptor->reqevents); + + pollset->nelts++; + + DBG2(4, "exiting, fd %d added to pollset %p\n", fd, pollset); + + return APR_SUCCESS; +} /* end of posix_add */ + + +static apr_status_t asio_pollset_add(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + DBG_BUFF + asio_elem_t *elem; + apr_status_t rv = APR_SUCCESS; + apr_pollset_private_t *priv = pollset->p; + + pollset_lock_rings(); + DBG(2, "entered\n"); + + if (pollset->flags & APR_POLLSET_THREADSAFE) { + + if (!APR_RING_EMPTY(&(priv->free_ring), asio_elem_t, link)) { + elem = APR_RING_FIRST(&(priv->free_ring)); + APR_RING_REMOVE(elem, link); + DBG1(3, "used recycled memory at %08p\n", elem); + elem->state = ASIO_INIT; + elem->a.aio_cflags = 0; + } + else { + elem = (asio_elem_t *) apr_pcalloc(pollset->pool, sizeof(asio_elem_t)); + DBG1(3, "alloced new memory at %08p\n", elem); + + elem->a.aio_notifytype = AIO_MSGQ; + elem->a.aio_msgev_qid = priv->msg_q; + DBG1(5, "aio_msgev_quid = %d \n", elem->a.aio_msgev_qid); + elem->a.aio_msgev_size = sizeof(asio_elem_t *); + elem->a.aio_msgev_flag = 0; /* wait if queue is full */ + elem->a.aio_msgev_addr = &(elem->msg); + elem->a.aio_buf = &(elem->os_pfd); + elem->a.aio_nbytes = 1; /* number of pfds to poll */ + elem->msg.msg_type = 1; + elem->msg.msg_elem = elem; + } + + /* z/OS only supports async I/O for sockets for now */ + elem->os_pfd.fd = descriptor->desc.s->socketdes; + + APR_RING_ELEM_INIT(elem, link); + elem->a.aio_cmd = AIO_SELPOLL; + elem->a.aio_cflags &= ~AIO_OK2COMPIMD; /* not OK to complete inline*/ + elem->pfd = *descriptor; + elem->os_pfd.events = get_event(descriptor->reqevents); + + if (0 != asyncio(&elem->a)) { + rv = errno; + DBG3(4, "pollset %p asio failed fd %d, errno %p\n", + pollset, elem->os_pfd.fd, rv); +#if DEBUG + perror(__FUNCTION__ " asio failure"); +#endif + } + else { + DBG2(4, "good asio call, adding fd %d to pollset %p\n", + elem->os_pfd.fd, pollset); + + pollset->nelts++; + apr_hash_set(priv->elems, &(elem->os_pfd.fd), sizeof(int), elem); + } + } + else { + /* APR_POLLSET_THREADSAFE isn't set. use POSIX poll in case + * pipes or files are used with this pollset + */ + + rv = posix_add(pollset, descriptor); + } + + DBG1(2, "exiting, rv = %d\n", rv); + + pollset_unlock_rings(); + return rv; +} /* end of asio_pollset_add */ + +static posix_remove(apr_pollset_t *pollset, const apr_pollfd_t *descriptor) +{ + DBG_BUFF + apr_uint32_t i; + apr_pollset_private_t *priv = pollset->p; + + DBG(4, "entered\n"); + for (i = 0; i < pollset->nelts; i++) { + if (descriptor->desc.s == priv->query_set[i].desc.s) { + /* Found an instance of the fd: remove this and any other copies */ + apr_uint32_t dst = i; + apr_uint32_t old_nelts = pollset->nelts; + pollset->nelts--; + for (i++; i < old_nelts; i++) { + if (descriptor->desc.s == priv->query_set[i].desc.s) { + pollset->nelts--; + } + else { + priv->pollset[dst] = priv->pollset[i]; + priv->query_set[dst] = priv->query_set[i]; + dst++; + } + } + DBG(4, "returning OK\n"); + return APR_SUCCESS; + } + } + + DBG(1, "returning APR_NOTFOUND\n"); + return APR_NOTFOUND; + +} /* end of posix_remove */ + +static apr_status_t asio_pollset_remove(apr_pollset_t *pollset, + const apr_pollfd_t *descriptor) +{ + DBG_BUFF + asio_elem_t *elem; + apr_status_t rv = APR_SUCCESS; + apr_pollset_private_t *priv = pollset->p; + /* AIO_CANCEL is synchronous, so autodata works fine. */ + struct aiocb cancel_a = {0}; + + int fd; + + DBG(2, "entered\n"); + + if (!(pollset->flags & APR_POLLSET_THREADSAFE)) { + return posix_remove(pollset, descriptor); + } + + pollset_lock_rings(); + +#if DEBUG + assert(descriptor->desc_type == APR_POLL_SOCKET); +#endif + /* zOS 1.12 doesn't support files for async i/o */ + fd = descriptor->desc.s->socketdes; + + elem = apr_hash_get(priv->elems, &(fd), sizeof(int)); + if (elem == NULL) { + DBG1(1, "couldn't find fd %d\n", fd); + rv = APR_NOTFOUND; + } else { + DBG1(5, "hash found fd %d\n", fd); + /* delete this fd from the hash */ + apr_hash_set(priv->elems, &(fd), sizeof(int), NULL); + + if (elem->state == ASIO_INIT) { + /* asyncio call to cancel */ + cancel_a.aio_cmd = AIO_CANCEL; + cancel_a.aio_buf = &elem->a; /* point to original aiocb */ + + cancel_a.aio_cflags = 0; + cancel_a.aio_cflags2 = 0; + + /* we want the original aiocb to show up on the pollset message queue + * before recycling its memory to eliminate race conditions + */ + + rv = asyncio(&cancel_a); + DBG1(4, "asyncio returned %d\n", rv); + +#if DEBUG + assert(rv == 1); +#endif + } + elem->state = ASIO_REMOVED; + rv = APR_SUCCESS; + } + + DBG1(2, "exiting, rv: %d\n", rv); + + pollset_unlock_rings(); + + return rv; +} /* end of asio_pollset_remove */ + +static posix_poll(apr_pollset_t *pollset, + apr_interval_time_t timeout, + apr_int32_t *num, + const apr_pollfd_t **descriptors) +{ + DBG_BUFF + int rv; + apr_uint32_t i, j; + apr_pollset_private_t *priv = pollset->p; + + DBG(4, "entered\n"); + + if (timeout > 0) { + timeout /= 1000; + } + rv = poll(priv->pollset, pollset->nelts, timeout); + (*num) = rv; + if (rv < 0) { + return apr_get_netos_error(); + } + if (rv == 0) { + return APR_TIMEUP; + } + j = 0; + for (i = 0; i < pollset->nelts; i++) { + if (priv->pollset[i].revents != 0) { + priv->result_set[j] = priv->query_set[i]; + priv->result_set[j].rtnevents = + get_revent(priv->pollset[i].revents); + j++; + } + } + if (descriptors) + *descriptors = priv->result_set; + + DBG(4, "exiting ok\n"); + return APR_SUCCESS; + +} /* end of posix_poll */ + +static process_msg(apr_pollset_t *pollset, struct asio_msgbuf_t *msg) +{ + DBG_BUFF + asio_elem_t *elem = msg->msg_elem; + + switch(elem->state) { + case ASIO_REMOVED: + DBG2(5, "for cancelled elem, recycling memory - elem %08p, fd %d\n", + elem, elem->os_pfd.fd); + APR_RING_INSERT_TAIL(&(pollset->p->free_ring), elem, + asio_elem_t, link); + break; + case ASIO_INIT: + DBG2(4, "adding to ready ring: elem %08p, fd %d\n", + elem, elem->os_pfd.fd); + elem->state = ASIO_COMPLETE; + APR_RING_INSERT_TAIL(&(pollset->p->ready_ring), elem, + asio_elem_t, link); + break; + default: + DBG3(1, "unexpected state: elem %08p, fd %d, state %d\n", + elem, elem->os_pfd.fd, elem->state); +#if DEBUG + assert(0); +#endif + } +} + +static apr_status_t asio_pollset_poll(apr_pollset_t *pollset, + apr_interval_time_t timeout, + apr_int32_t *num, + const apr_pollfd_t **descriptors) +{ + DBG_BUFF + int i, ret; + asio_elem_t *elem, *next_elem; + struct asio_msgbuf_t msg_buff; + struct timespec tv; + apr_status_t rv = APR_SUCCESS; + apr_pollset_private_t *priv = pollset->p; + + DBG(6, "entered\n"); /* chatty - traces every second w/Event */ + + if ((pollset->flags & APR_POLLSET_THREADSAFE) == 0 ) { + return posix_poll(pollset, timeout, num, descriptors); + } + + pollset_lock_rings(); + APR_RING_INIT(&(priv->ready_ring), asio_elem_t, link); + + while (!APR_RING_EMPTY(&(priv->prior_ready_ring), asio_elem_t, link)) { + elem = APR_RING_FIRST(&(priv->prior_ready_ring)); + DBG3(5, "pollset %p elem %p fd %d on prior ready ring\n", + pollset, + elem, + elem->os_pfd.fd); + + APR_RING_REMOVE(elem, link); + + /* + * since USS does not remember what's in our pollset, we have + * to re-add fds which have not been apr_pollset_remove'd + * + * there may have been too many ready fd's to return in the + * result set last time. re-poll inline for both cases + */ + + if (elem->state == ASIO_REMOVED) { + + /* + * async i/o is done since it was found on prior_ready + * the state says the caller is done with it too + * so recycle the elem + */ + + APR_RING_INSERT_TAIL(&(priv->free_ring), elem, + asio_elem_t, link); + continue; /* do not re-add if it has been _removed */ + } + + elem->state = ASIO_INIT; + elem->a.aio_cflags = AIO_OK2COMPIMD; + + if (0 != (ret = asyncio(&elem->a))) { + if (ret == 1) { + DBG(4, "asyncio() completed inline\n"); + /* it's ready now */ + elem->state = ASIO_COMPLETE; + APR_RING_INSERT_TAIL(&(priv->ready_ring), elem, asio_elem_t, + link); + } + else { + DBG2(1, "asyncio() failed, ret: %d, errno: %d\n", + ret, errno); + pollset_unlock_rings(); + return errno; + } + } + DBG1(4, "asyncio() completed rc %d\n", ret); + } + + DBG(6, "after prior ready loop\n"); /* chatty w/timeouts, hence 6 */ + + /* Gather async poll completions that have occurred since the last call */ + while (0 < msgrcv(priv->msg_q, &msg_buff, sizeof(asio_elem_t *), 0, + IPC_NOWAIT)) { + process_msg(pollset, &msg_buff); + } + + /* Suspend if nothing is ready yet. */ + if (APR_RING_EMPTY(&(priv->ready_ring), asio_elem_t, link)) { + + if (timeout >= 0) { + tv.tv_sec = apr_time_sec(timeout); + tv.tv_nsec = apr_time_usec(timeout) * 1000; + } else { + tv.tv_sec = INT_MAX; /* block until something is ready */ + } + + DBG2(6, "nothing on the ready ring " + "- blocking for %d seconds %d ns\n", + tv.tv_sec, tv.tv_nsec); + + pollset_unlock_rings(); /* allow other apr_pollset_* calls while blocked */ + + if (0 >= (ret = __msgrcv_timed(priv->msg_q, &msg_buff, + sizeof(asio_elem_t *), 0, NULL, &tv))) { +#if DEBUG + if (errno == EAGAIN) { + DBG(6, "__msgrcv_timed timed out\n"); /* timeout path, so 6 */ + } + else { + DBG(1, "__msgrcv_timed failed!\n"); + } +#endif + return (errno == EAGAIN) ? APR_TIMEUP : errno; + } + + pollset_lock_rings(); + + process_msg(pollset, &msg_buff); + } + + APR_RING_INIT(&priv->prior_ready_ring, asio_elem_t, link); + + (*num) = 0; + elem = APR_RING_FIRST(&(priv->ready_ring)); + + for (i = 0; + + i < priv->size + && elem != APR_RING_SENTINEL(&(priv->ready_ring), asio_elem_t, link); + i++) { + DBG2(5, "ready ring: elem %08p, fd %d\n", elem, elem->os_pfd.fd); + + priv->result_set[i] = elem->pfd; + priv->result_set[i].rtnevents + = get_revent(elem->os_pfd.revents); + (*num)++; + + elem = APR_RING_NEXT(elem, link); + +#if DEBUG + if (elem == APR_RING_SENTINEL(&(priv->ready_ring), asio_elem_t, link)) { + DBG(5, "end of ready ring reached\n"); + } +#endif + } + + if (descriptors) { + *descriptors = priv->result_set; + } + + /* if the result size is too small, remember which descriptors + * haven't had results reported yet. we will look + * at these descriptors on the next apr_pollset_poll call + */ + + APR_RING_CONCAT(&priv->prior_ready_ring, &(priv->ready_ring), asio_elem_t, link); + + DBG1(2, "exiting, rv = %d\n", rv); + + pollset_unlock_rings(); + + return rv; +} /* end of asio_pollset_poll */ + +static const apr_pollset_provider_t impl = { + asio_pollset_create, + asio_pollset_add, + asio_pollset_remove, + asio_pollset_poll, + asio_pollset_cleanup, + "asio" +}; + +const apr_pollset_provider_t *apr_pollset_provider_aio_msgq = &impl; + +#endif /* HAVE_AIO_MSGQ */ |