//
// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
//
// 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.
//

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

#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <unistd.h>

// POSIX AIO using poll().  We use a single poll thread to perform
// I/O operations for the entire system.  This isn't entirely scalable,
// and it might be a good idea to create a few threads and group the
// I/O operations into separate pollers to limit the amount of work each
// thread does, and to scale across multiple cores.  For now we don't
// worry about it.

// nni_posix_pollq is a work structure used by the poller thread, that keeps
// track of all the underlying pipe handles and so forth being used by poll().

// Locking strategy:  We use the pollq lock to guard the lists on the pollq,
// the nfds (which counts the number of items in the pollq), the pollq
// shutdown flags (pq->closing and pq->closed) and the cv on each pfd.  We
// use a lock on the pfd only to protect the the events field (which we treat
// as an atomic bitfield), and the cb and arg pointers.  Note that the pfd
// lock is therefore a leaf lock, which is sometimes acquired while holding
// the pq lock.  Every reasonable effort is made to minimize holding locks.
// (Btw, pfd->fd is not guarded, because it is set at pfd creation and
// persists until the pfd is destroyed.)

typedef struct nni_posix_pollq nni_posix_pollq;

struct nni_posix_pollq {
	nni_mtx  mtx;
	int      nfds;
	int      wakewfd; // write side of waker pipe
	int      wakerfd; // read side of waker pipe
	bool     closing; // request for worker to exit
	bool     closed;
	nni_thr  thr;   // worker thread
	nni_list pollq; // armed nodes
	nni_list reapq;
};

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

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 = &nni_posix_global_pollq;

	// 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.
	// If this code is getting used, you really should be using the
	// kqueue poller, or you need to upgrade your older system.
	int one = 1;
	(void) setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one));
#endif

	if ((pfd = NNI_ALLOC_STRUCT(pfd)) == NULL) {
		return (NNG_ENOMEM);
	}
	NNI_LIST_NODE_INIT(&pfd->node);
	nni_mtx_init(&pfd->mtx);
	nni_cv_init(&pfd->cv, &pq->mtx);
	pfd->fd     = fd;
	pfd->events = 0;
	pfd->cb     = NULL;
	pfd->arg    = NULL;
	pfd->pq     = pq;
	nni_mtx_lock(&pq->mtx);
	if (pq->closing) {
		nni_mtx_unlock(&pq->mtx);
		nni_cv_fini(&pfd->cv);
		nni_mtx_fini(&pfd->mtx);
		NNI_FREE_STRUCT(pfd);
		return (NNG_ECLOSED);
	}
	nni_list_append(&pq->pollq, pfd);
	pq->nfds++;
	nni_mtx_unlock(&pq->mtx);
	*pfdp = pfd;
	return (0);
}

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);
}

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

void
nni_posix_pfd_close(nni_posix_pfd *pfd)
{
	(void) shutdown(pfd->fd, SHUT_RDWR);
}

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

	nni_posix_pfd_close(pfd);

	nni_mtx_lock(&pq->mtx);
	if (nni_list_active(&pq->pollq, pfd)) {
		nni_list_remove(&pq->pollq, pfd);
		pq->nfds--;
	}

	if ((!nni_thr_is_self(&pq->thr)) && (!pq->closed)) {
		nni_list_append(&pq->reapq, pfd);
		nni_plat_pipe_raise(pq->wakewfd);
		while (nni_list_active(&pq->reapq, pfd)) {
			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);
}

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

	nni_mtx_lock(&pfd->mtx);
	pfd->events |= events;
	nni_mtx_unlock(&pfd->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)) {
		nni_plat_pipe_raise(pq->wakewfd);
	}
	return (0);
}

static void
nni_posix_poll_thr(void *arg)
{
	nni_posix_pollq *pq     = arg;
	int              nalloc = 0;
	struct pollfd *  fds    = NULL;
	nni_posix_pfd ** pfds   = NULL;

	for (;;) {
		int            nfds;
		unsigned       events;
		nni_posix_pfd *pfd;

		nni_mtx_lock(&pq->mtx);
		while (nalloc < (pq->nfds + 1)) {
			int n = pq->nfds + 128;

			// Drop the lock while we sleep or allocate.  This
			// allows additional items to be added or removed (!)
			// while we wait.
			nni_mtx_unlock(&pq->mtx);

			// Toss the old ones first; avoids *doubling* memory
			// consumption during alloc.
			NNI_FREE_STRUCTS(fds, nalloc);
			NNI_FREE_STRUCTS(pfds, nalloc);
			nalloc = 0;

			if ((pfds = NNI_ALLOC_STRUCTS(pfds, n)) == NULL) {
				nni_msleep(10); // sleep for a bit, try later
			} else if ((fds = NNI_ALLOC_STRUCTS(fds, n)) == NULL) {
				NNI_FREE_STRUCTS(pfds, n);
				nni_msleep(10);
			} else {
				nalloc = n;
			}
			nni_mtx_lock(&pq->mtx);
		}

		// The waker pipe is set up so that we will be woken
		// when it is written (this allows us to be signaled).
		fds[0].fd      = pq->wakerfd;
		fds[0].events  = POLLIN;
		fds[0].revents = 0;
		pfds[0]        = NULL;
		nfds           = 1;

		// Also lets reap anything that was in the reaplist!
		while ((pfd = nni_list_first(&pq->reapq)) != NULL) {
			nni_list_remove(&pq->reapq, pfd);
			nni_cv_wake(&pfd->cv);
		}

		// If we're closing down, bail now.  This is done *after* we
		// have ensured that the reapq is empty.  Anything still in
		// the pollq is not going to receive further callbacks.
		if (pq->closing) {
			pq->closed = true;
			nni_mtx_unlock(&pq->mtx);
			break;
		}

		// Set up the poll list.
		NNI_LIST_FOREACH (&pq->pollq, pfd) {

			nni_mtx_lock(&pfd->mtx);
			events = pfd->events;
			nni_mtx_unlock(&pfd->mtx);

			if (events != 0) {
				fds[nfds].fd      = pfd->fd;
				fds[nfds].events  = events;
				fds[nfds].revents = 0;
				pfds[nfds]        = pfd;
				nfds++;
			}
		}
		nni_mtx_unlock(&pq->mtx);

		// We could get the result from poll, and avoid iterating
		// over the entire set of pollfds, but since on average we
		// will be walking half the list, doubling the work we do
		// (the condition with a potential pipeline stall) seems like
		// adding complexity with no real benefit.  It also makes the
		// worst case even worse.
		(void) poll(fds, nfds, -1);

		// If the waker pipe was signaled, read from it.
		if (fds[0].revents & POLLIN) {
			NNI_ASSERT(fds[0].fd == pq->wakerfd);
			nni_plat_pipe_clear(pq->wakerfd);
		}

		for (int i = 1; i < nfds; i++) {
			if ((events = fds[i].revents) != 0) {
				nni_posix_pfd_cb cb;
				void *           arg;

				pfd = pfds[i];

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

				if (cb) {
					cb(pfd, events, arg);
				}
			}
		}
	}

	NNI_FREE_STRUCTS(fds, nalloc);
	NNI_FREE_STRUCTS(pfds, nalloc);
}

static void
nni_posix_pollq_destroy(nni_posix_pollq *pq)
{
	nni_mtx_lock(&pq->mtx);
	pq->closing = true;
	nni_mtx_unlock(&pq->mtx);

	nni_plat_pipe_raise(pq->wakewfd);

	nni_thr_fini(&pq->thr);
	nni_plat_pipe_close(pq->wakewfd, pq->wakerfd);
	nni_mtx_fini(&pq->mtx);
}

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

	NNI_LIST_INIT(&pq->pollq, nni_posix_pfd, node);
	NNI_LIST_INIT(&pq->reapq, nni_posix_pfd, node);
	pq->closing = false;
	pq->closed  = false;

	if ((rv = nni_plat_pipe_open(&pq->wakewfd, &pq->wakerfd)) != 0) {
		return (rv);
	}
	if ((rv = nni_thr_init(&pq->thr, nni_posix_poll_thr, pq)) != 0) {
		nni_plat_pipe_close(pq->wakewfd, pq->wakerfd);
		return (rv);
	}
	nni_thr_set_name(&pq->thr, "nng:poll:poll");
	nni_mtx_init(&pq->mtx);
	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);
}