Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt; --- lib/loop_poll.c | 50 ++++++++++++++++++++++++---------------- lib/loop_poll_kqueue.c | 60 ++++++++++++++++++++++++++++++++++-------------- 2 files changed, 73 insertions(+), 37 deletions(-) diff --git a/lib/loop_poll.c b/lib/loop_poll.c index 1f6b960..30f8666 100644 --- a/lib/loop_poll.c +++ b/lib/loop_poll.c @@ -36,6 +36,16 @@ /* logs, std(in|out|err), pipe */ #define POLL_FDS_USED_MISC 50 +#ifdef HAVE_EPOLL +#define USE_EPOLL 1 +#else + #ifdef HAVE_KQUEUE + #define USE_KQUEUE 1 + #else + #define USE_POLL 1 + #endif /* HAVE_KQUEUE */ +#endif /* HAVE_EPOLL */ + static int32_t _qb_signal_add_to_jobs_(struct qb_loop *l, struct qb_poll_entry *pe); @@ -188,15 +198,15 @@ qb_loop_poll_create(struct qb_loop *l) s->low_fds_event_fn = NULL; s->not_enough_fds = 0; -#ifdef HAVE_EPOLL +#ifdef USE_EPOLL (void)qb_epoll_init(s); -#else -#ifdef HAVE_KQUEUE +#endif +#ifdef USE_KQUEUE (void)qb_kqueue_init(s); -#else +#endif +#ifdef USE_POLL (void)qb_poll_init(s); -#endif /* HAVE_KQUEUE */ -#endif /* HAVE_EPOLL */ +#endif /* USE_POLL */ return (struct qb_loop_source *)s; } @@ -229,10 +239,6 @@ _get_empty_array_position_(struct qb_poll_source *s) uint32_t install_pos; int32_t res = 0; struct qb_poll_entry *pe; -#ifndef HAVE_EPOLL - struct pollfd *ufds; - int32_t new_size = 0; -#endif /* HAVE_EPOLL */ for (found = 0, install_pos = 0; install_pos < s->poll_entry_count; install_pos++) { @@ -245,6 +251,15 @@ _get_empty_array_position_(struct qb_poll_source *s) } if (found == 0) { +#ifdef USE_POLL + struct pollfd *ufds; + int32_t new_size = (s->poll_entry_count + 1) * sizeof(struct pollfd); + ufds = realloc(s->ufds, new_size); + if (ufds == NULL) { + return -ENOMEM; + } + s->ufds = ufds; +#endif /* USE_POLL */ /* * Grow pollfd list */ @@ -252,16 +267,6 @@ _get_empty_array_position_(struct qb_poll_source *s) if (res != 0) { return res; } -#ifndef HAVE_EPOLL -#ifndef HAVE_KQUEUE - new_size = (s->poll_entry_count + 1) * sizeof(struct pollfd); - ufds = realloc(s->ufds, new_size); - if (ufds == NULL) { - return -ENOMEM; - } - s->ufds = ufds; -#endif -#endif /* HAVE_EPOLL */ s->poll_entry_count += 1; install_pos = s->poll_entry_count - 1; @@ -336,6 +341,11 @@ qb_loop_poll_add(struct qb_loop * lp, size = ((struct qb_poll_source *)l->fd_source)->poll_entry_count; res = _poll_add_(l, p, fd, events, data, &pe); + if (res != 0) { + qb_util_perror(LOG_ERR, + "couldn't add poll entryfor FD %d", fd); + return res; + } new_size = ((struct qb_poll_source *)l->fd_source)->poll_entry_count; pe->poll_dispatch_fn = dispatch_fn; diff --git a/lib/loop_poll_kqueue.c b/lib/loop_poll_kqueue.c index b33825b..8a59496 100644 --- a/lib/loop_poll_kqueue.c +++ b/lib/loop_poll_kqueue.c @@ -53,12 +53,10 @@ _add(struct qb_poll_source *s, struct qb_poll_entry *pe, int32_t fd, int32_t eve { int32_t res = 0; struct kevent ke; - int kents = _poll_to_filter_(events); + short filters = _poll_to_filter_(events); - /* fill out the kevent struct */ - EV_SET(&ke, pe->check, kents, EV_ADD, 0, NULL, pe); + EV_SET(&ke, fd, filters, EV_ADD, 0, NULL, pe); - /* set the event */ res = kevent(s->epollfd, &ke, 1, NULL, 0, NULL); if (res == -1) { res = -errno; @@ -68,23 +66,34 @@ _add(struct qb_poll_source *s, struct qb_poll_entry *pe, int32_t fd, int32_t eve return res; } - static int32_t _mod(struct qb_poll_source *s, struct qb_poll_entry *pe, int32_t fd, int32_t events) { + int32_t res = 0; + struct kevent ke[2]; + short new_filters = _poll_to_filter_(events); + short old_filters = _poll_to_filter_(pe->ufd.events); + + EV_SET(&ke[0], fd, old_filters, EV_DELETE, 0, NULL, pe); + EV_SET(&ke[1], fd, new_filters, EV_ADD, 0, NULL, pe); + + res = kevent(s->epollfd, ke, 2, NULL, 0, NULL); + if (res == -1) { + res = -errno; + qb_util_perror(LOG_ERR, "kevent(mod)"); + } + return res; } static int32_t -_del(struct qb_poll_source *s, struct qb_poll_entry *pe, int32_t fd, int32_t arr_index) +_del(struct qb_poll_source *s, struct qb_poll_entry *pe, int32_t fd, int32_t events) { int32_t res = 0; struct kevent ke; - int kents = 0; //_poll_to_filter_(events); + short filters = _poll_to_filter_(events); - /* fill out the kevent struct */ - EV_SET(&ke, pe->check, kents, EV_DELETE, 0, NULL, pe); + EV_SET(&ke, fd, filters, EV_DELETE, 0, NULL, pe); - /* set the event */ res = kevent(s->epollfd, &ke, 1, NULL, 0, NULL); if (res == -1) { res = -errno; @@ -97,31 +106,47 @@ static int32_t _poll_and_add_to_jobs_(struct qb_loop_source *src, int32_t ms_timeout) { int32_t i; - int32_t res; int32_t event_count; int32_t new_jobs = 0; - int32_t revents; + int32_t revents = 0; struct qb_poll_entry *pe = NULL; struct qb_poll_source *s = (struct qb_poll_source *)src; struct kevent events[MAX_EVENTS]; struct timespec timeout = { 0, 0 }; + struct timespec *timeout_pt = &timeout; - qb_timespec_add_ms(&timeout, ms_timeout); - + if (ms_timeout > 0) { + qb_timespec_add_ms(&timeout, ms_timeout); + } else if (ms_timeout < 0) { + timeout_pt = NULL; + } qb_poll_fds_usage_check_(s); retry_poll: - event_count = kevent(s->epollfd, NULL, 0, events, MAX_EVENTS, NULL); + event_count = kevent(s->epollfd, NULL, 0, events, MAX_EVENTS, timeout_pt); if (errno == EINTR && event_count == -1) { goto retry_poll; } else if (event_count == -1) { + qb_util_perror(LOG_ERR, "kevent(poll)"); return -errno; } for (i = 0; i < event_count; i++) { + revents = 0; + if (events[i].flags) { + qb_util_log(LOG_INFO, + "got flags %d on fd %d.", events[i].flags, pe->ufd.fd); + } if (events[i].flags & EV_ERROR) { - revents = POLLHUP; + qb_util_log(LOG_WARNING, + "got EV_ERROR on fd %d.", pe->ufd.fd); + revents |= POLLERR; + } + if (events[i].flags & EV_EOF) { + qb_util_log(LOG_INFO, + "got EV_EOF on fd %d.", pe->ufd.fd); + revents |= POLLHUP; } if (events[i].filter == EVFILT_READ) { revents |= POLLIN; @@ -130,7 +155,7 @@ retry_poll: revents |= POLLOUT; } pe = events[i].udata; - if (pe->check != events[i].ident) { + if (pe->ufd.fd != events[i].ident) { qb_util_log(LOG_WARNING, "can't find poll entry for new event."); continue; @@ -164,6 +189,7 @@ qb_kqueue_init(struct qb_poll_source *s) s->epollfd = kqueue(); if (s->epollfd < 0) { + qb_util_perror(LOG_ERR, "kqueue()"); return -errno; } s->driver.fini = _fini; -- 1.7.9.3

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt; --- lib/ipc_us.c | 14 ++++++++------ lib/ipcs.c | 6 ++---- tests/check_ipc.c | 1 + 3 files changed, 11 insertions(+), 10 deletions(-) diff --git a/lib/ipc_us.c b/lib/ipc_us.c index df4f735..08bedd7 100644 --- a/lib/ipc_us.c +++ b/lib/ipc_us.c @@ -191,14 +191,11 @@ retry_recv: if (result == -1) { return -errno; } -#if defined(QB_SOLARIS) || defined(QB_BSD) || defined(QB_DARWIN) - /* On many OS poll never return POLLHUP or POLLERR. - * EOF is detected when recvmsg return 0. - */ if (result == 0) { - return -errno; //ENOTCONN + qb_util_log(LOG_DEBUG, + "recv(fd %d) got 0 bytes assuming ENOTCONN", s); + return -ENOTCONN; } -#endif processed += result; if (processed != len) { @@ -225,6 +222,7 @@ qb_ipc_us_recv_ready(struct qb_ipc_one_way * one_way, int32_t ms_timeout) } else if (poll_events == -1) { return -errno; } else if (poll_events == 1 && (ufds.revents & (POLLERR | POLLHUP))) { + qb_util_log(LOG_DEBUG, "poll(fd %d) got POLLHUP", one_way->u.us.sock); return -ENOTCONN; } return 0; @@ -251,6 +249,8 @@ retry_recv: } } if (result == 0) { + qb_util_log(LOG_DEBUG, "recv(fd %d) got 0 bytes assuming ENOTCONN", + one_way->u.us.sock); return -ENOTCONN; } if (result == -1) { @@ -289,6 +289,8 @@ retry_recv: } } if (result == 0) { + qb_util_log(LOG_DEBUG, "recv(fd %d) got 0 bytes assuming ENOTCONN", + one_way->u.us.sock); return -ENOTCONN; } if (result == -1) { diff --git a/lib/ipcs.c b/lib/ipcs.c index efc31d1..33a887b 100644 --- a/lib/ipcs.c +++ b/lib/ipcs.c @@ -598,14 +598,12 @@ _process_request_(struct qb_ipcs_connection *c, int32_t ms_timeout) } res = size; goto cleanup; - } - c->stats.requests++; - - if (hdr->id == QB_IPC_MSG_DISCONNECT) { + } else if (size == 0 || hdr->id == QB_IPC_MSG_DISCONNECT) { qb_util_log(LOG_DEBUG, "client requesting a disconnect"); qb_ipcs_disconnect(c); res = -ESHUTDOWN; } else { + c->stats.requests++; res = c->service->serv_fns.msg_process(c, hdr, hdr->size); /* 0 == good, negative == backoff */ if (res < 0) { diff --git a/tests/check_ipc.c b/tests/check_ipc.c index 8e0d3a3..191d005 100644 --- a/tests/check_ipc.c +++ b/tests/check_ipc.c @@ -697,6 +697,7 @@ main(void) qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD, QB_LOG_FILTER_FILE, "*", LOG_TRACE); qb_log_ctl(QB_LOG_STDERR, QB_LOG_CONF_ENABLED, QB_TRUE); + qb_log_format_set(QB_LOG_STDERR, "[%p] %f:%l %b"); srunner_run_all(sr, CK_VERBOSE); number_failed = srunner_ntests_failed(sr); -- 1.7.9.3

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt; --- lib/ipcc.c | 51 ++++++++++++++++++++++++++++++++------------------- 1 file changed, 32 insertions(+), 19 deletions(-) diff --git a/lib/ipcc.c b/lib/ipcc.c index 1e397da..6322994 100644 --- a/lib/ipcc.c +++ b/lib/ipcc.c @@ -100,6 +100,26 @@ disconnect_and_cleanup: return NULL; } +static void +_check_connection_state(struct qb_ipcc_connection * c, int32_t res) +{ + if (res >= 0) return; + + if (res != -EAGAIN && + res != -ETIMEDOUT && + res != -EINTR && +#ifdef EWOULDBLOCK + res != -EWOULDBLOCK && +#endif + res != -EINVAL) { + errno = -res; + qb_util_perror(LOG_DEBUG, + "interpreting result %d as a disconnect", + res); + c->is_connected = QB_FALSE; + } +} + static struct qb_ipc_one_way * _event_sock_one_way_get(struct qb_ipcc_connection * c) { @@ -152,13 +172,13 @@ qb_ipcc_send(struct qb_ipcc_connection * c, const void *msg_ptr, size_t msg_len) res2 = qb_ipc_us_send(&c->setup, msg_ptr, 1); } while (res2 == -EAGAIN); if (res2 == -EPIPE) { - c->is_connected = QB_FALSE; - return -ENOTCONN; + res2 = -ENOTCONN; } if (res2 != 1) { res = res2; } } + _check_connection_state(c, res); return res; } @@ -207,13 +227,13 @@ qb_ipcc_sendv(struct qb_ipcc_connection * c, const struct iovec * iov, res2 = qb_ipc_us_send(&c->setup, &res, 1); } while (res2 == -EAGAIN); if (res2 == -EPIPE) { - c->is_connected = QB_FALSE; - return -ENOTCONN; + res2 = -ENOTCONN; } if (res2 != 1) { res = res2; } } + _check_connection_state(c, res); return res; } @@ -235,14 +255,10 @@ qb_ipcc_recv(struct qb_ipcc_connection * c, void *msg_ptr, res2 = qb_ipc_us_recv_ready(ow, 0); if (res2 < 0) { - if (res2 == -ENOTCONN) { - c->is_connected = QB_FALSE; - } - return res2; - } else { - return res; + res = res2; } } + _check_connection_state(c, res); return res; } @@ -335,22 +351,19 @@ qb_ipcc_event_recv(struct qb_ipcc_connection * c, void *msg_pt, if (ow) { res = qb_ipc_us_recv_ready(ow, ms_timeout); if (res < 0) { - if (res == -ENOTCONN) { - c->is_connected = QB_FALSE; - } + _check_connection_state(c, res); return res; } } size = c->funcs.recv(&c->event, msg_pt, msg_len, ms_timeout); if (size < 0) { + _check_connection_state(c, size); return size; } if (c->needs_sock_for_poll) { res = qb_ipc_us_recv(&c->setup, &one_byte, 1, -1); if (res < 0) { - if (res == -ENOTCONN) { - c->is_connected = QB_FALSE; - } + _check_connection_state(c, res); return res; } } @@ -361,6 +374,7 @@ void qb_ipcc_disconnect(struct qb_ipcc_connection *c) { struct qb_ipc_one_way *ow = NULL; + int32_t res = 0; qb_util_log(LOG_DEBUG, "%s()", __func__); @@ -370,9 +384,8 @@ qb_ipcc_disconnect(struct qb_ipcc_connection *c) ow = _event_sock_one_way_get(c); if (ow) { - if (qb_ipc_us_recv_ready(ow, 0) == -ENOTCONN) { - c->is_connected = QB_FALSE; - } + res = qb_ipc_us_recv_ready(ow, 0); + _check_connection_state(c, res); qb_ipcc_us_sock_close(ow->u.us.sock); } -- 1.7.9.3