feat:replace boost library with C++11 std library

This commit is contained in:
Livox-SDK
2020-04-01 20:53:55 +08:00
parent 6d06233944
commit d3258bbd97
1936 changed files with 241862 additions and 60153 deletions
+490
View File
@@ -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 */
+505
View File
@@ -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 */
+460
View File
@@ -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];
#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 */
+224
View File
@@ -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;
}
+247
View File
@@ -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()
{
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);
}
+595
View File
@@ -0,0 +1,595 @@
/* 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 nres = 0;
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; 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[nres] = ep->pfd;
pollset->p->result_set[nres].rtnevents =
get_revent(pollset->p->port_set[i].portev_events);
++nres;
}
/* 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 (nres > 0) { /* any event besides wakeup pipe? */
*num = nres;
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 */
+449
View File
@@ -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;
+151
View File
@@ -0,0 +1,151 @@
/* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "apr.h"
#include "apr_poll.h"
#include "apr_time.h"
#include "apr_portable.h"
#include "apr_arch_file_io.h"
#include "apr_arch_networkio.h"
#include "apr_arch_poll_private.h"
#include "apr_arch_inherit.h"
#if !APR_FILES_AS_SOCKETS
#ifdef WIN32
apr_status_t 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(&wakeup_pipe[0], &wakeup_pipe[1],
pool)) != APR_SUCCESS)
return rv;
pfd->p = pool;
pfd->reqevents = APR_POLLIN;
pfd->desc_type = APR_POLL_FILE;
pfd->desc.f = wakeup_pipe[0];
{
int flags;
if ((flags = fcntl(wakeup_pipe[0]->filedes, F_GETFD)) == -1)
return errno;
flags |= FD_CLOEXEC;
if (fcntl(wakeup_pipe[0]->filedes, F_SETFD, flags) == -1)
return errno;
}
{
int flags;
if ((flags = fcntl(wakeup_pipe[1]->filedes, F_GETFD)) == -1)
return errno;
flags |= FD_CLOEXEC;
if (fcntl(wakeup_pipe[1]->filedes, F_SETFD, flags) == -1)
return errno;
}
return APR_SUCCESS;
}
apr_status_t 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;
}
}
+782
View File
@@ -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 */