//
// Copyright 2019 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.
//

#ifdef NNG_HAVE_PORT_CREATE

#include <errno.h>
#include <fcntl.h>
#include <port.h>
#include <sched.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h> /* for strerror() */
#include <unistd.h>

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

#define NNI_MAX_PORTEV 64
typedef struct nni_posix_pollq nni_posix_pollq;

// nni_posix_pollq is a work structure that manages state for the port-event
// based pollq implementation.  We only really need to keep track of the
// single thread, and the associated port itself.
struct nni_posix_pollq {
	int     port; // port id (from port_create)
	nni_thr thr;  // worker thread
};

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

// 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_pollq *pq;
	nni_posix_pfd *  pfd;

	pq = &nni_posix_global_pollq;

	if ((pfd = NNI_ALLOC_STRUCT(pfd)) == NULL) {
		return (NNG_ENOMEM);
	}
	(void) fcntl(fd, F_SETFD, FD_CLOEXEC);
	(void) fcntl(fd, F_SETFL, O_NONBLOCK);

	nni_mtx_init(&pfd->mtx);
	nni_cv_init(&pfd->cv, &pfd->mtx);
	pfd->closed  = false;
	pfd->closing = false;
	pfd->fd      = fd;
	pfd->pq      = pq;
	pfd->cb      = NULL;
	pfd->data    = NULL;
	*pfdp        = pfd;
	return (0);
}

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

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

	nni_mtx_lock(&pfd->mtx);
	if (!pfd->closing) {
		pfd->closing = true;
		(void) shutdown(pfd->fd, SHUT_RDWR);
		port_dissociate(pq->port, PORT_SOURCE_FD, pfd->fd);
	}
	nni_mtx_unlock(&pfd->mtx);

	// Send the wake event to the poller to synchronize with it.
	// Note that port_send should only really fail if out of memory
	// or we run into a resource limit.
}

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

	nni_posix_pfd_close(pfd);

	NNI_ASSERT(!nni_thr_is_self(&pq->thr));

	while (port_send(pq->port, 1, pfd) != 0) {
		if ((errno == EBADF) || (errno == EBADFD)) {
			pfd->closed = true;
			break;
		}
		sched_yield(); // try again later...
	}

	nni_mtx_lock(&pfd->mtx);
	while (!pfd->closed) {
		nni_cv_wait(&pfd->cv);
	}
	nni_mtx_unlock(&pfd->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);
	if (!pfd->closing) {
		pfd->events |= events;
		if (port_associate(pq->port, PORT_SOURCE_FD, pfd->fd,
		        (int) pfd->events, pfd) != 0) {
			int rv = nni_plat_errno(errno);
			nni_mtx_unlock(&pfd->mtx);
			return (rv);
		}
	}
	nni_mtx_unlock(&pfd->mtx);
	return (0);
}

static void
nni_posix_poll_thr(void *arg)
{
	for (;;) {
		nni_posix_pollq *pq = arg;
		port_event_t     ev[NNI_MAX_PORTEV];
		nni_posix_pfd *  pfd;
		unsigned         events;
		nni_posix_pfd_cb cb;
		void *           arg;
		unsigned         n;

		n = 1; // wake us even on just one event
		if (port_getn(pq->port, ev, NNI_MAX_PORTEV, &n, NULL) != 0) {
			if (errno == EINTR) {
				continue;
			}
			return;
		}

		// We run through the returned ports twice.  First we
		// get the callbacks.  Then we do the reaps.  This way
		// we ensure that we only reap *after* callbacks have run.
		for (unsigned i = 0; i < n; i++) {
			if (ev[i].portev_source != PORT_SOURCE_FD) {
				continue;
			}
			pfd    = ev[i].portev_user;
			events = ev[i].portev_events;

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

			if (cb != NULL) {
				cb(pfd, events, arg);
			}
		}
		for (unsigned i = 0; i < n; i++) {
			if (ev[i].portev_source != PORT_SOURCE_USER) {
				continue;
			}

			// User event telling us to stop doing things.
			// We signal back to use this as a coordination
			// event between the pollq and the thread
			// handler. NOTE: It is absolutely critical
			// that there is only a single thread per
			// pollq.  Otherwise we cannot be sure that we
			// are blocked completely,
			pfd = ev[i].portev_user;
			nni_mtx_lock(&pfd->mtx);
			pfd->closed = true;
			nni_cv_wake(&pfd->cv);
			nni_mtx_unlock(&pfd->mtx);
		}
	}
}

static void
nni_posix_pollq_destroy(nni_posix_pollq *pq)
{
	(void) close(pq->port);
	nni_thr_fini(&pq->thr);
}

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

	if ((pq->port = port_create()) < 0) {
		return (nni_plat_errno(errno));
	}

	if ((rv = nni_thr_init(&pq->thr, nni_posix_poll_thr, pq)) != 0) {
		nni_posix_pollq_destroy(pq);
		return (rv);
	}
	nni_thr_set_name(&pq->thr, "nng:poll:port");

	nni_thr_run(&pq->thr);
	return (0);
}

void
nni_posix_pfd_set_cb(nni_posix_pfd *pfd, nni_posix_pfd_cb cb, void *arg)
{
	NNI_ASSERT(cb != NULL); // must not be null when established.

	nni_mtx_lock(&pfd->mtx);
	pfd->cb   = cb;
	pfd->data = arg;
	nni_mtx_unlock(&pfd->mtx);
}

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_PORT_CREATE