//
|
// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
|
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
|
// Copyright 2018 Liam Staskawicz <liam@stask.net>
|
//
|
// This software is supplied under the terms of the MIT License, a
|
// copy of which should be located in the distribution where this
|
// file was obtained (LICENSE.txt). A copy of the license may also be
|
// found online at https://opensource.org/licenses/MIT.
|
//
|
|
#ifdef NNG_HAVE_KQUEUE
|
|
#include <errno.h>
|
#include <fcntl.h>
|
#include <stdbool.h>
|
#include <stdio.h>
|
#include <string.h> /* for strerror() */
|
#include <sys/event.h>
|
#include <sys/socket.h>
|
#include <unistd.h>
|
|
#include "core/nng_impl.h"
|
#include "platform/posix/posix_pollq.h"
|
|
typedef struct nni_posix_pollq nni_posix_pollq;
|
|
// nni_posix_pollq is a work structure that manages state for the kqueue-based
|
// pollq implementation
|
struct nni_posix_pollq {
|
nni_mtx mtx;
|
int wake_wfd; // write side of wake pipe
|
int wake_rfd; // read side of wake pipe
|
bool closed; // request for worker to exit
|
int kq; // kqueue handle
|
nni_thr thr; // worker thread
|
nni_list reapq; // items to reap
|
};
|
|
struct nni_posix_pfd {
|
nni_list_node node; // linkage into the reap list
|
nni_posix_pollq *pq; // associated pollq
|
int fd; // file descriptor to poll
|
void * arg; // user data
|
nni_posix_pfd_cb cb; // user callback on event
|
bool closed;
|
unsigned events;
|
nni_cv cv; // signaled when poller has unregistered
|
nni_mtx mtx;
|
};
|
|
#define NNI_MAX_KQUEUE_EVENTS 64
|
|
// single global instance for now
|
static nni_posix_pollq nni_posix_global_pollq;
|
|
int
|
nni_posix_pfd_init(nni_posix_pfd **pfdp, int fd)
|
{
|
nni_posix_pfd * pf;
|
nni_posix_pollq *pq;
|
struct kevent ev[2];
|
unsigned flags = EV_ADD | EV_DISABLE | EV_CLEAR;
|
|
// Set this is as soon as possible (narrow the close-exec race as
|
// much as we can; better options are system calls that suppress
|
// this behavior from descriptor creation.)
|
(void) fcntl(fd, F_SETFD, FD_CLOEXEC);
|
(void) fcntl(fd, F_SETFL, O_NONBLOCK);
|
#ifdef SO_NOSIGPIPE
|
// Darwin lacks MSG_NOSIGNAL, but has a socket option.
|
int one = 1;
|
(void) setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one));
|
#endif
|
|
pq = &nni_posix_global_pollq;
|
|
if ((pf = NNI_ALLOC_STRUCT(pf)) == NULL) {
|
return (NNG_ENOMEM);
|
}
|
|
nni_mtx_init(&pf->mtx);
|
nni_cv_init(&pf->cv, &pq->mtx);
|
|
pf->pq = pq;
|
pf->fd = fd;
|
pf->cb = NULL;
|
pf->arg = NULL;
|
pf->events = 0;
|
pf->closed = false;
|
|
NNI_LIST_NODE_INIT(&pf->node);
|
*pfdp = pf;
|
// Create entries in the kevent queue, without enabling them.
|
EV_SET(&ev[0], (uintptr_t) fd, EVFILT_READ, flags, 0, 0, pf);
|
EV_SET(&ev[1], (uintptr_t) fd, EVFILT_WRITE, flags, 0, 0, pf);
|
|
// We update the kqueue list, without polling for events.
|
if (kevent(pq->kq, ev, 2, NULL, 0, NULL) != 0) {
|
int rv;
|
rv = nni_plat_errno(errno);
|
nni_cv_fini(&pf->cv);
|
nni_mtx_fini(&pf->mtx);
|
NNI_FREE_STRUCT(pf);
|
return (rv);
|
}
|
|
return (0);
|
}
|
|
void
|
nni_posix_pfd_close(nni_posix_pfd *pf)
|
{
|
nni_posix_pollq *pq = pf->pq;
|
|
nni_mtx_lock(&pq->mtx);
|
if (!pf->closed) {
|
struct kevent ev[2];
|
pf->closed = true;
|
EV_SET(&ev[0], pf->fd, EVFILT_READ, EV_DELETE, 0, 0, pf);
|
EV_SET(&ev[1], pf->fd, EVFILT_WRITE, EV_DELETE, 0, 0, pf);
|
(void) shutdown(pf->fd, SHUT_RDWR);
|
// This should never fail -- no allocations, just deletion.
|
(void) kevent(pq->kq, ev, 2, NULL, 0, NULL);
|
}
|
nni_mtx_unlock(&pq->mtx);
|
}
|
|
void
|
nni_posix_pfd_fini(nni_posix_pfd *pf)
|
{
|
nni_posix_pollq *pq;
|
|
pq = pf->pq;
|
|
nni_posix_pfd_close(pf);
|
|
// All consumers take care to move finalization to the reap thread,
|
// unless they are synchronous on user threads.
|
NNI_ASSERT(!nni_thr_is_self(&pq->thr));
|
|
nni_mtx_lock(&pq->mtx);
|
|
// If we're running on the callback, then don't bother to kick
|
// the pollq again. This is necessary because we cannot modify
|
// the poller while it is polling.
|
if ((!nni_thr_is_self(&pq->thr)) && (!pq->closed)) {
|
nni_list_append(&pq->reapq, pf);
|
nni_plat_pipe_raise(pq->wake_wfd);
|
while (nni_list_active(&pq->reapq, pf)) {
|
nni_cv_wait(&pf->cv);
|
}
|
}
|
nni_mtx_unlock(&pq->mtx);
|
|
(void) close(pf->fd);
|
nni_cv_fini(&pf->cv);
|
nni_mtx_fini(&pf->mtx);
|
NNI_FREE_STRUCT(pf);
|
}
|
|
int
|
nni_posix_pfd_fd(nni_posix_pfd *pf)
|
{
|
return (pf->fd);
|
}
|
|
void
|
nni_posix_pfd_set_cb(nni_posix_pfd *pf, nni_posix_pfd_cb cb, void *arg)
|
{
|
NNI_ASSERT(cb != NULL); // must not be null when established.
|
nni_mtx_lock(&pf->mtx);
|
pf->cb = cb;
|
pf->arg = arg;
|
nni_mtx_unlock(&pf->mtx);
|
}
|
|
int
|
nni_posix_pfd_arm(nni_posix_pfd *pf, unsigned events)
|
{
|
struct kevent ev[2];
|
int nev = 0;
|
unsigned flags = EV_ENABLE | EV_DISPATCH | EV_CLEAR;
|
nni_posix_pollq *pq = pf->pq;
|
|
nni_mtx_lock(&pf->mtx);
|
if (pf->closed) {
|
events = 0;
|
} else {
|
pf->events |= events;
|
events = pf->events;
|
}
|
nni_mtx_unlock(&pf->mtx);
|
|
if (events == 0) {
|
// No events, and kqueue is oneshot, so nothing to do.
|
return (0);
|
}
|
|
if (events & POLLIN) {
|
EV_SET(&ev[nev++], pf->fd, EVFILT_READ, flags, 0, 0, pf);
|
}
|
if (events & POLLOUT) {
|
EV_SET(&ev[nev++], pf->fd, EVFILT_WRITE, flags, 0, 0, pf);
|
}
|
while (kevent(pq->kq, ev, nev, NULL, 0, NULL) != 0) {
|
if (errno == EINTR) {
|
continue;
|
}
|
return (nni_plat_errno(errno));
|
}
|
return (0);
|
}
|
|
static void
|
nni_posix_pollq_reap(nni_posix_pollq *pq)
|
{
|
nni_posix_pfd *pf;
|
nni_mtx_lock(&pq->mtx);
|
while ((pf = nni_list_first(&pq->reapq)) != NULL) {
|
nni_list_remove(&pq->reapq, pf);
|
nni_cv_wake(&pf->cv);
|
}
|
nni_mtx_unlock(&pq->mtx);
|
}
|
|
static void
|
nni_posix_poll_thr(void *arg)
|
{
|
nni_posix_pollq *pq = arg;
|
|
nni_thr_set_name(NULL, "nng:poll:kqueue");
|
|
for (;;) {
|
int n;
|
struct kevent evs[NNI_MAX_KQUEUE_EVENTS];
|
nni_posix_pfd * pf;
|
nni_posix_pfd_cb cb;
|
void * cbarg;
|
unsigned revents;
|
|
nni_mtx_lock(&pq->mtx);
|
if (pq->closed) {
|
nni_mtx_unlock(&pq->mtx);
|
nni_posix_pollq_reap(pq);
|
return;
|
}
|
nni_mtx_unlock(&pq->mtx);
|
n = kevent(pq->kq, NULL, 0, evs, NNI_MAX_KQUEUE_EVENTS, NULL);
|
|
for (int i = 0; i < n; i++) {
|
struct kevent *ev = &evs[i];
|
|
switch (ev->filter) {
|
case EVFILT_READ:
|
revents = POLLIN;
|
break;
|
case EVFILT_WRITE:
|
revents = POLLOUT;
|
break;
|
}
|
if (ev->udata == NULL) {
|
nni_plat_pipe_clear(pq->wake_rfd);
|
nni_posix_pollq_reap(pq);
|
continue;
|
}
|
pf = (void *) ev->udata;
|
if (ev->flags & EV_ERROR) {
|
revents |= POLLHUP;
|
}
|
|
nni_mtx_lock(&pf->mtx);
|
cb = pf->cb;
|
cbarg = pf->arg;
|
pf->events &= ~(revents);
|
nni_mtx_unlock(&pf->mtx);
|
|
if (cb != NULL) {
|
cb(pf, revents, cbarg);
|
}
|
}
|
}
|
}
|
|
static void
|
nni_posix_pollq_destroy(nni_posix_pollq *pq)
|
{
|
nni_mtx_lock(&pq->mtx);
|
pq->closed = true;
|
nni_mtx_unlock(&pq->mtx);
|
nni_plat_pipe_raise(pq->wake_wfd);
|
|
nni_thr_fini(&pq->thr);
|
nni_plat_pipe_close(pq->wake_wfd, pq->wake_rfd);
|
|
if (pq->kq >= 0) {
|
close(pq->kq);
|
pq->kq = -1;
|
}
|
nni_mtx_fini(&pq->mtx);
|
}
|
|
static int
|
nni_posix_pollq_create(nni_posix_pollq *pq)
|
{
|
int rv;
|
struct kevent ev;
|
|
if ((pq->kq = kqueue()) < 0) {
|
return (nni_plat_errno(errno));
|
}
|
|
nni_mtx_init(&pq->mtx);
|
NNI_LIST_INIT(&pq->reapq, nni_posix_pfd, node);
|
|
if (((rv = nni_thr_init(&pq->thr, nni_posix_poll_thr, pq)) != 0) ||
|
((rv = nni_plat_pipe_open(&pq->wake_wfd, &pq->wake_rfd)) != 0)) {
|
nni_posix_pollq_destroy(pq);
|
return (rv);
|
}
|
|
// Register the wake pipe. We use this to synchronize closing
|
// file descriptors.
|
EV_SET(&ev, (uintptr_t) pq->wake_rfd, EVFILT_READ, EV_ADD | EV_CLEAR,
|
0, 0, NULL);
|
|
if ((rv = kevent(pq->kq, &ev, 1, NULL, 0, NULL)) != 0) {
|
rv = nni_plat_errno(rv);
|
nni_posix_pollq_destroy(pq);
|
return (rv);
|
}
|
|
nni_thr_run(&pq->thr);
|
return (0);
|
}
|
|
int
|
nni_posix_pollq_sysinit(void)
|
{
|
return (nni_posix_pollq_create(&nni_posix_global_pollq));
|
}
|
|
void
|
nni_posix_pollq_sysfini(void)
|
{
|
nni_posix_pollq_destroy(&nni_posix_global_pollq);
|
}
|
|
#endif // NNG_HAVE_KQUEUE
|
