//
// 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_EPOLL

#include <errno.h>
#include <fcntl.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h> /* for strerror() */
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <unistd.h>

#include "core/nng_impl.h"
#include "platform/posix/posix_pollq.h"

typedef struct nni_posix_pollq nni_posix_pollq;

#ifndef EFD_CLOEXEC
#define EFD_CLOEXEC 0
#endif
#ifndef EFD_NONBLOCK
#define EFD_NONBLOCK 0
#endif

#define NNI_MAX_EPOLL_EVENTS 64

// flags we always want enabled as long as at least one event is active
#define NNI_EPOLL_FLAGS ((unsigned) EPOLLONESHOT | (unsigned) EPOLLERR)

// Locking strategy:
//
// The pollq mutex protects its own reapq, close state, and the close
// state of the individual pfds.  It also protects the pfd cv, which is
// only signaled when the pfd is closed.  This mutex is only acquired
// when shutting down the pollq, or closing a pfd.  For normal hot-path
// operations we don't need it.
//
// The pfd mutex protects the pfd's own "closing" flag (test and set),
// the callback and arg, and its event mask.  This mutex is used a lot,
// but it should be uncontended excepting possibly when closing.

// nni_posix_pollq is a work structure that manages state for the epoll-based
// pollq implementation
struct nni_posix_pollq {
	nni_mtx  mtx;
	int      epfd;  // epoll handle
	int      evfd;  // event fd (to wake us for other stuff)
	bool     close; // request for worker to exit
	nni_thr  thr;   // worker thread
	nni_list reapq;
};

struct nni_posix_pfd {
	nni_list_node    node;
	nni_posix_pollq *pq;
	int              fd;
	nni_posix_pfd_cb cb;
	void *           arg;
	bool             closed;
	bool             closing;
	bool             reap;
	unsigned         events;
	nni_mtx          mtx;
	nni_cv           cv;
};

// 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 *    pfd;
	nni_posix_pollq *  pq;
	struct epoll_event ev;
	int                rv;

	pq = &nni_posix_global_pollq;

	(void) fcntl(fd, F_SETFD, FD_CLOEXEC);
	(void) fcntl(fd, F_SETFL, O_NONBLOCK);

	if ((pfd = NNI_ALLOC_STRUCT(pfd)) == NULL) {
		return (NNG_ENOMEM);
	}
	nni_mtx_init(&pfd->mtx);
	nni_cv_init(&pfd->cv, &pq->mtx);

	pfd->pq      = pq;
	pfd->fd      = fd;
	pfd->cb      = NULL;
	pfd->arg     = NULL;
	pfd->events  = 0;
	pfd->closing = false;
	pfd->closed  = false;

	NNI_LIST_NODE_INIT(&pfd->node);

	// notifications disabled to begin with
	memset(&ev, 0, sizeof(ev));
	ev.events   = 0;
	ev.data.ptr = pfd;

	if (epoll_ctl(pq->epfd, EPOLL_CTL_ADD, fd, &ev) != 0) {
		rv = nni_plat_errno(errno);
		nni_cv_fini(&pfd->cv);
		nni_mtx_fini(&pfd->mtx);
		NNI_FREE_STRUCT(pfd);
		return (rv);
	}

	*pfdp = pfd;
	return (0);
}

int
nni_posix_pfd_arm(nni_posix_pfd *pfd, unsigned events)
{
	nni_posix_pollq *pq = pfd->pq;

	// NB: We depend on epoll event flags being the same as their POLLIN
	// equivalents.  I.e. POLLIN == EPOLLIN, POLLOUT == EPOLLOUT, and so
	// forth.  This turns out to be true both for Linux and the illumos
	// epoll implementation.

	nni_mtx_lock(&pfd->mtx);
	if (!pfd->closing) {
		struct epoll_event ev;
		pfd->events |= events;
		events = pfd->events;

		memset(&ev, 0, sizeof(ev));
		ev.events   = events | NNI_EPOLL_FLAGS;
		ev.data.ptr = pfd;

		if (epoll_ctl(pq->epfd, EPOLL_CTL_MOD, pfd->fd, &ev) != 0) {
			int rv = nni_plat_errno(errno);
			nni_mtx_unlock(&pfd->mtx);
			return (rv);
		}
	}
	nni_mtx_unlock(&pfd->mtx);
	return (0);
}

int
nni_posix_pfd_fd(nni_posix_pfd *pfd)
{
	return (pfd->fd);
}

void
nni_posix_pfd_set_cb(nni_posix_pfd *pfd, nni_posix_pfd_cb cb, void *arg)
{
	nni_mtx_lock(&pfd->mtx);
	pfd->cb  = cb;
	pfd->arg = arg;
	nni_mtx_unlock(&pfd->mtx);
}

void
nni_posix_pfd_close(nni_posix_pfd *pfd)
{
	nni_mtx_lock(&pfd->mtx);
	if (!pfd->closing) {
		nni_posix_pollq *  pq = pfd->pq;
		struct epoll_event ev; // Not actually used.
		pfd->closing = true;

		(void) shutdown(pfd->fd, SHUT_RDWR);
		(void) epoll_ctl(pq->epfd, EPOLL_CTL_DEL, pfd->fd, &ev);
	}
	nni_mtx_unlock(&pfd->mtx);
}

void
nni_posix_pfd_fini(nni_posix_pfd *pfd)
{
	nni_posix_pollq *pq = pfd->pq;

	nni_posix_pfd_close(pfd);

	// We have to synchronize with the pollq thread (unless we are
	// on that thread!)
	NNI_ASSERT(!nni_thr_is_self(&pq->thr));

	uint64_t one = 1;

	nni_mtx_lock(&pq->mtx);
	nni_list_append(&pq->reapq, pfd);

	// Wake the remote side.  For now we assume this always
	// succeeds.  The only failure modes here occur when we
	// have already excessively signaled this (2^64 times
	// with no read!!), or when the evfd is closed, or some
	// kernel bug occurs.  Those errors would manifest as
	// a hang waiting for the poller to reap the pfd in fini,
	// if it were possible for them to occur.  (Barring other
	// bugs, it isn't.)
	if (write(pq->evfd, &one, sizeof(one)) != sizeof(one)) {
		nni_panic("BUG! write to epoll fd incorrect!");
	}

	while (!pfd->closed) {
		nni_cv_wait(&pfd->cv);
	}
	nni_mtx_unlock(&pq->mtx);

	// We're exclusive now.

	(void) close(pfd->fd);
	nni_cv_fini(&pfd->cv);
	nni_mtx_fini(&pfd->mtx);
	NNI_FREE_STRUCT(pfd);
}

static void
nni_posix_pollq_reap(nni_posix_pollq *pq)
{
	nni_posix_pfd *pfd;
	nni_mtx_lock(&pq->mtx);
	while ((pfd = nni_list_first(&pq->reapq)) != NULL) {
		nni_list_remove(&pq->reapq, pfd);

		// Let fini know we're done with it, and it's safe to
		// remove.
		pfd->closed = true;
		nni_cv_wake(&pfd->cv);
	}
	nni_mtx_unlock(&pq->mtx);
}

static void
nni_posix_poll_thr(void *arg)
{
	nni_posix_pollq *  pq = arg;
	struct epoll_event events[NNI_MAX_EPOLL_EVENTS];

	for (;;) {
		int  n;
		bool reap = false;

		n = epoll_wait(pq->epfd, events, NNI_MAX_EPOLL_EVENTS, -1);
		if ((n < 0) && (errno == EBADF)) {
			// Epoll fd closed, bail.
			return;
		}

		// dispatch events
		for (int i = 0; i < n; ++i) {
			const struct epoll_event *ev;

			ev = &events[i];
			// If the waker pipe was signaled, read from it.
			if ((ev->data.ptr == NULL) &&
			    (ev->events & (unsigned) POLLIN)) {
				uint64_t clear;
				if (read(pq->evfd, &clear, sizeof(clear)) !=
				    sizeof(clear)) {
					nni_panic("read from evfd incorrect!");
				}
				reap = true;
			} else {
				nni_posix_pfd *  pfd = ev->data.ptr;
				nni_posix_pfd_cb cb;
				void *           cbarg;
				unsigned         mask;

				mask = ev->events &
				    ((unsigned) EPOLLIN | (unsigned) EPOLLOUT |
				        (unsigned) EPOLLERR);

				nni_mtx_lock(&pfd->mtx);
				pfd->events &= ~mask;
				cb    = pfd->cb;
				cbarg = pfd->arg;
				nni_mtx_unlock(&pfd->mtx);

				// Execute the callback with lock released
				if (cb != NULL) {
					cb(pfd, mask, cbarg);
				}
			}
		}

		if (reap) {
			nni_posix_pollq_reap(pq);
			nni_mtx_lock(&pq->mtx);
			if (pq->close) {
				nni_mtx_unlock(&pq->mtx);
				return;
			}
			nni_mtx_unlock(&pq->mtx);
		}
	}
}

static void
nni_posix_pollq_destroy(nni_posix_pollq *pq)
{
	uint64_t one = 1;

	nni_mtx_lock(&pq->mtx);
	pq->close = true;

	if (write(pq->evfd, &one, sizeof(one)) != sizeof(one)) {
		// This should never occur, and if it does it could
		// lead to a hang.
		nni_panic("BUG! unable to write to evfd!");
	}
	nni_mtx_unlock(&pq->mtx);

	nni_thr_fini(&pq->thr);

	close(pq->evfd);
	close(pq->epfd);

	nni_mtx_fini(&pq->mtx);
}

static int
nni_posix_pollq_add_eventfd(nni_posix_pollq *pq)
{
	// add event fd so we can wake ourself on exit
	struct epoll_event ev;
	int                fd;

	memset(&ev, 0, sizeof(ev));

	if ((fd = eventfd(0, EFD_NONBLOCK)) < 0) {
		return (nni_plat_errno(errno));
	}
	(void) fcntl(fd, F_SETFD, FD_CLOEXEC);
	(void) fcntl(fd, F_SETFL, O_NONBLOCK);

	// This is *NOT* one shot.  We want to wake EVERY single time.
	ev.events   = EPOLLIN;
	ev.data.ptr = 0;

	if (epoll_ctl(pq->epfd, EPOLL_CTL_ADD, fd, &ev) != 0) {
		(void) close(fd);
		return (nni_plat_errno(errno));
	}
	pq->evfd = fd;
	return (0);
}

static int
nni_posix_pollq_create(nni_posix_pollq *pq)
{
	int rv;

#if NNG_HAVE_EPOLL_CREATE1
	if ((pq->epfd = epoll_create1(EPOLL_CLOEXEC)) < 0) {
		return (nni_plat_errno(errno));
	}
#else
	// Old Linux.  Size is a "hint" about number of descriptors.
	// Hopefully not a hard limit, and not used in modern Linux.
	if ((pq->epfd = epoll_create(16)) < 0) {
		return (nni_plat_errno(errno));
	}
	(void) fcntl(pq->epfd, F_SETFD, FD_CLOEXEC);
#endif

	pq->close = false;

	NNI_LIST_INIT(&pq->reapq, nni_posix_pfd, node);
	nni_mtx_init(&pq->mtx);

	if ((rv = nni_posix_pollq_add_eventfd(pq)) != 0) {
		(void) close(pq->epfd);
		nni_mtx_fini(&pq->mtx);
		return (rv);
	}
	if ((rv = nni_thr_init(&pq->thr, nni_posix_poll_thr, pq)) != 0) {
		(void) close(pq->epfd);
		(void) close(pq->evfd);
		nni_mtx_fini(&pq->mtx);
		return (rv);
	}
	nni_thr_set_name(&pq->thr, "nng:poll:epoll");
	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_EPOLL