/* -
 * Copyright (c) 2006 Verdens Gang AS
 * Copyright (c) 2006 Linpro AS
 * Copyright (c) 2007 OmniTI Computer Consulting, Inc.
 * Copyright (c) 2007 Theo Schlossnagle
 * Copyright (c) 2010-2011 UPLEX, Nils Goroll
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * $Id: cache_acceptor_ports.c 3580 2009-02-05 08:45:25Z tfheen $
 *
 * note on use of vca_del():
 *
 * the original man page for port_associate(3C) states:
 *
 *    When an event for a PORT_SOURCE_FD object is retrieved,
 *    the object no longer has an association with the port.
 *
 * This can be read along the lines of sparing the
 * port_dissociate after port_getn(), but in fact,
 * port_dissociate should be used
 *
 * Ref: http://opensolaris.org/jive/thread.jspa?threadID=129476&tstart=0
 */

#include "config.h"
#if defined(HAVE_PORT_CREATE)

#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <rctl.h>
#include <strings.h>

#include <port.h>
#include <sys/time.h>

#include "shmlog.h"
#include "cache.h"
#include "cache_acceptor.h"

/* max number of events to process in one go */

#define	MAX_EVENTS	256

static pthread_t vca_ports_thread;
int solaris_dport = -1;

static VTAILQ_HEAD(, sess) sesshead = VTAILQ_HEAD_INITIALIZER(sesshead);

/* cache_httpconn.c: HTC_Rx codes */

#define	HTC_RX_OVERFLOW		-2 /* recv workspace overflowed */
#define	HTC_RX_ERROR		-1 /* read() in HTC_Rx returned <= 0 */
#define	HTC_RX_NEEDMORE		 0 /* header not complete yet */
#define	HTX_RX_COMPLETE		 1 /* ready */

static inline void
vca_add(int fd, void *data)
{
	int r;
	while ((r = port_associate(solaris_dport, PORT_SOURCE_FD, fd,
		    POLLIN, data) == -1) &&
	    errno == EAGAIN) {
		VSL(SLT_Debug, 0, "EAGAIN in vca_add. "
		    "probably exceeded process.max-port-events. Sleeping");
		TIM_sleep(params->accept_fd_holdoff * 0.001);
	}
	AZ(r);
}

static inline void
vca_del(int fd)
{
	port_dissociate(solaris_dport, PORT_SOURCE_FD, fd);
}

static inline void
vca_port_close(struct sess *sp, const char *reason)
{
	vca_del(sp->fd);
	VTAILQ_REMOVE(&sesshead, sp, list);
	vca_close_session(sp, reason);
	SES_Delete(sp);
}

static inline void
vca_port_ev(port_event_t *ev) {
	struct sess *sp;
	int i, events;

	CAST_OBJ_NOTNULL(sp, ev->portev_user, SESS_MAGIC);
	assert(sp->fd >= 0);

	/* received an fd to watch from vca_ports_pass */
	if (ev->portev_source == PORT_SOURCE_USER) {
		AZ(sp->obj);
		VTAILQ_INSERT_TAIL(&sesshead, sp, list);
		vca_add(sp->fd, sp);

		return;
	}

	assert(ev->portev_source == PORT_SOURCE_FD);
	events = ev->portev_events;

	/*
	 * Note: POLLHUP can mean we have received a FIN, but we might still get
	 * data from the connection, so we need to read anyway
	 */
	if (events & (POLLIN|POLLHUP|POLLPRI)) {
		i = HTC_Rx(sp->htc);

		if (i == HTC_RX_NEEDMORE) {
			if (events & POLLHUP) {
				/* can't expect any more data */
				vca_port_close(sp, "more needed but hung up");
				return;
			}

			/* incomplete header, wait for more data */
			vca_add(sp->fd, sp);
			return;
		}

		if ((i == HTC_RX_ERROR) && (events & POLLPRI)) {
			/* zero bytes OK for PRI, wait for more data */
			vca_add(sp->fd, sp);
			return;
		}

		/*
		 * vca_handover will handle all other HTC_RX cases.
		 * we stop tracking this fd/sp and are done
		 *
		 * see note at top regarding vca_del()
		 */

		vca_del(sp->fd);
		VTAILQ_REMOVE(&sesshead, sp, list);

		vca_handover(sp, i);
		return;
	}

	if (events & POLLERR) {
		vca_port_close(sp, "POLLERR");
		return;
	}

	/* XXX should we ever get POLLNVAL ? */
	if (events & POLLNVAL) {
		vca_port_close(sp, "POLLNVAL");
		return;
	}

	/* catch all, specific assertions for the likely culprits */
	AZ(events & POLLRDNORM);
	AZ(events & POLLRDBAND);
	AZ(events & POLLOUT);
	AZ(events & POLLWRBAND);

	INCOMPL();
}

static void *
vca_main(void *arg)
{
	struct sess *sp;

	/*
	 * timeouts:
	 *
	 * min_ts : Minimum timeout for port_getn
	 * min_t  : ^ equivalent in floating point representation
	 *
	 * max_ts : Maximum timeout for port_getn
	 * max_t  : ^ equivalent in floating point representation
	 *
	 * with (nevents == 1), we should always choose the correct port_getn
	 * timeout to check session timeouts, so max is just a safety measure
	 * (if this implementation is correct, it could be set to an "infinte"
	 *  value)
	 *
	 * with (nevents > 1), min and max define the acceptable range for
	 * - additional latency of keep-alive connections and
	 * - additional tolerance for handling session timeouts
	 *
	 */

	/*				       ms      us      ns */
	static struct timespec	min_ts = {0L, 100L * 1000L * 1000L };
	static double		min_t  = 0.1;			/* 100 ms */
	static struct timespec	max_ts = {1L, 0L};		/* 1 s */
	static double		max_t  = 1.0;			/* 1 s */

	struct timespec ts;
	struct timespec *timeout;

	(void) arg;

	solaris_dport = port_create();
	assert(solaris_dport >= 0);

	timeout = &max_ts;

	while (1) {
		port_event_t ev[MAX_EVENTS];
		int nevents, ei, ret;
		double now, deadline;

		/*
		 * XXX Do we want to scale this up dynamically to increase
		 * efficiency in high throughput situations? - would need to
		 * start with one to keep latency low at any rate
		 *
		 * Note: when increasing nevents, we must lower min_ts and
		 *	 max_ts
		 */
		nevents = 1;

		/*
		 * see disucssion in
		 * - https://issues.apache.org/bugzilla/show_bug.cgi?id=47645
		 * - http://mail.opensolaris.org/pipermail/networking-discuss/2009-August/011979.html
		 *
		 * comment from apr/poll/unix/port.c :
		 *
		 * This confusing API can return an event at the same time
		 * that it reports EINTR or ETIME.
		 *
		 */

		ret = port_getn(solaris_dport, ev, MAX_EVENTS, &nevents, timeout);

		if (ret < 0)
			assert((errno == EINTR) || (errno == ETIME));

		for (ei = 0; (ei < nevents); ei++) {
			vca_port_ev(ev + ei);
		}

		/* check for timeouts */
		now = TIM_real();
		deadline = now - params->sess_timeout;

		/*
		 * This loop assumes that the oldest sessions are always at the
		 * beginning of the list (which is the case if we guarantee to
		 * enqueue at the tail only
		 *
		 */

		for (;;) {
			sp = VTAILQ_FIRST(&sesshead);
			if (sp == NULL)
				break;
			if (sp->t_open > deadline) {
				break;
			}
			VTAILQ_REMOVE(&sesshead, sp, list);
			if (sp->fd != -1) {
				vca_del(sp->fd);
			}
			vca_close_session(sp, "timeout");
			SES_Delete(sp);
		}

		/*
		 * Calculate the timeout for the next get_portn
		 */

		if (sp) {
			double tmo = (sp->t_open + params->sess_timeout) - now;

			/* we should have removed all expired sps */
			assert(tmo > 0.0);

			if (tmo < min_t) {
				timeout = &min_ts;
			} else if (tmo > max_t) {
				timeout = &max_ts;
			} else {
				TIM_t2ts(tmo, &ts);
				timeout = &ts;
			}
		} else {
			timeout = &max_ts;
		}
	}
}

static void
vca_ports_pass(struct sess *sp)
{
	int r;
	while ((r = port_send(solaris_dport, 0, sp)) == -1 &&
	    errno == EAGAIN) {
		VSL(SLT_Debug, 0, "EAGAIN in vca_ports_pass. "
		    "probably exceeded process.max-port-events. Sleeping");
		TIM_sleep(params->accept_fd_holdoff * 0.001);
	}
	AZ(r);
}

/*--------------------------------------------------------------------*/

#ifdef RCTL_TRACE
void
print_priv(rctlblk_t *rblk) {
	char		*priv_str;
	rctl_priv_t	priv;

	if (! rblk)
		return;

	priv = rctlblk_get_privilege(rblk);
	switch (priv) {
	case RCPRIV_BASIC:
		priv_str = "basic";
		break;
	case RCPRIV_PRIVILEGED:
		priv_str = "priv";
		break;
	case RCPRIV_SYSTEM:
		priv_str = "sys";
		break;
	default:
		INCOMPL();
	}

	fprintf(stderr, "debug rblk priv %s val %llu\n",
	    priv_str,
	    rctlblk_get_value(rblk));
}
#else
#define	print_priv(rblk)	((void)0)
#endif /* RCTL_TRACE */

#define	MAXRBLKS	5

/*
 * get priv, basic, sys privilege. Assume that only one of each
 * exists
 */

static int
vca_ports_get_rctls(const char *name, rctlblk_t *blks[], int nblks,
    rctlblk_t **basic, rctlblk_t **priv, rctlblk_t **sys)
{
	rctlblk_t *rblk, *orblk;
	int i;

	*basic = *priv = *sys = NULL;
	i = 0;

	if (nblks < 1) {
		errno = ENOSPC;
		return (-1);
	}

	rblk = blks[i++];
	if (getrctl(name, NULL, rblk, RCTL_FIRST) == -1) {
		return (-1);
	}

	do {
		switch (rctlblk_get_privilege(rblk)) {
		case RCPRIV_BASIC:
			AZ(*basic);
			*basic = rblk;
			print_priv(*basic);
			break;
		case RCPRIV_PRIVILEGED:
			AZ(*priv);
			*priv = rblk;
			print_priv(*priv);
			break;
		case RCPRIV_SYSTEM:
			AZ(*sys);
			*sys = rblk;
			print_priv(*sys);
			break;
		default:
			INCOMPL();
		}

		orblk = rblk;
		rblk = blks[i++];

	} while ((i <= nblks) &&
	(getrctl(name, orblk, rblk, RCTL_NEXT) == 0));

	if ((errno != ESRCH) && (errno != ENOENT)) {
		return (-1);
	}

	return (0);
}

#define	MAXRBLKS	5

/*
 * In a typical configuration, most file descriptors in Varnish are needed for
 * client connections, and potentially we need to look after all of them in the
 * acceptor (extreme case when the maximum number of connections is reached with
 * connections either not having sent a request or waiting for the next request
 * in HTTP/1.1 keepalive). So we should be able to create a port event for every
 * file descriptor.
 *
 * Additions filedescriptors are needed for backend and control connections, but
 * also we generate events for passing sessions in vca_ports_pass. Thus, setting
 * process.max-port-events eqal to process.max-file-descriptor should be a good
 * approximation.
 *
 * here we check the resource limits and try to adjust them, such that
 * process.max-port-events is closest to process.max-file-descriptor
 */

static void
vca_ports_check_rctls(void)
{
	rctlblk_t *newblk;
	rctlblk_t *blks[MAXRBLKS];
	rctlblk_t *basic, *priv, *sys, *rblk;
	rctl_qty_t nfd;
	rctl_qty_t nev, limit;
	int i;
	size_t size = rctlblk_size();

	const char *name_ev = "process.max-port-events";
	const char *name_fd = "process.max-file-descriptor";

	newblk = calloc(MAXRBLKS+1, size);
	for (i = 0; i < MAXRBLKS+1; i++)
		blks[i] = (rctlblk_t *)((uint8_t *)newblk + ((i + 1) * size));

	AN(newblk);
	AN(blks);

	rblk = blks[0];
	if (getrctl(name_fd, NULL, rblk, RCTL_FIRST) == -1) {
		(void) fprintf(stderr, "warn: failed to get %s rctl: %s\n",
		    name_fd, strerror(errno));
		goto out;
	}
	nfd = rctlblk_get_enforced_value(rblk);

	if (getrctl(name_ev, NULL, rblk, RCTL_FIRST) == -1) {
		(void) fprintf(stderr, "warn: failed to get %s rctl: %s\n",
		    name_ev, strerror(errno));
		goto out;
	}
	nev = rctlblk_get_enforced_value(rblk);

	/* CASE1: shortcut to check if all is nice */
	if (nev == nfd)
		goto out;

	if (vca_ports_get_rctls(name_ev, blks, MAXRBLKS,
		&basic, &priv, &sys) != 0) {
		(void) fprintf(stderr, "warn: failed to get all %s rctls: %s\n",
		    name_ev, strerror(errno));
		goto out;
	}

	/* check if we need to lower the limit */
	if (nev > nfd) {
		if (basic) {
			/* CASE2: lower basic limit replace */
			memcpy(newblk, basic, size);
			rctlblk_set_value(newblk, nfd);
			if (setrctl(name_ev, basic, newblk, RCTL_REPLACE) == -1) {
				(void) fprintf(stderr, "warn: failed lower %s "
				    "rctl to %llu (replace): %s\n",
				    name_ev, nfd, strerror(errno));
			}
		} else {
			/* CASE3: lower basic limit insert */
			rctlblk_set_value(newblk, nfd);
			rctlblk_set_privilege(newblk, RCPRIV_BASIC);
			rctlblk_set_local_action(newblk, RCTL_LOCAL_DENY, 0);
			if (setrctl(name_ev, NULL, newblk, RCTL_INSERT) == -1) {
				(void) fprintf(stderr, "warn: failed lower %s "
				    "rctl to %llu (insert): %s\n",
				    name_ev, nfd, strerror(errno));
			}
		}
		goto out;
	}

	/* try to raise the limit, if possible */

	print_priv(sys);
	print_priv(priv);
	print_priv(basic);

	if (sys) {
		limit = nev = rctlblk_get_enforced_value(sys);
		if (nev < nfd) {
			fprintf(stderr, "warn: system limit for %s=%llu "
			    "lower than effective limit for %s=%llu",
			    name_ev, nev,
			    name_fd, nfd);
			goto out;
		}
	}
	if (priv) {
		if (rctlblk_get_value(priv) < nfd) {
			/* CASE4: raise priv limit replace */
			nev = (nfd < limit) ? nfd : limit;

			memcpy(newblk, priv, size);
			rctlblk_set_value(newblk, nev);
			if (setrctl(name_ev, priv, newblk, RCTL_REPLACE) == -1) {
				limit = rctlblk_get_value(priv);
				(void) fprintf(stderr, "warn: failed raise priviliged"
				    " %s rctl limit from %llu to %llu (replace): %s\n",
				    name_ev, limit, nfd, strerror(errno));
			} else {
				limit = nev;
			}
		}
	}

	if (basic) {
		if (rctlblk_get_value(basic) < nfd) {
			/* CASE5: raise basic limit replace */
			nev = (nfd < limit) ? nfd : limit;

			memcpy(newblk, basic, size);
			rctlblk_set_value(newblk, nev);
			if (setrctl(name_ev, basic, newblk, RCTL_REPLACE) == -1) {
				(void) fprintf(stderr, "warn: failed raise basic"
				    " %s rctl limit to %llu (replace): %s\n",
				    name_ev, nfd, strerror(errno));
			}
		}
	}

  out:
	free(newblk);
}

static void
vca_ports_init(void)
{
	vca_ports_check_rctls();
	AZ(pthread_create(&vca_ports_thread, NULL, vca_main, NULL));
}

struct acceptor acceptor_ports = {
	.name =		"ports",
	.init =		vca_ports_init,
	.pass =		vca_ports_pass
};

#endif /* defined(HAVE_PORT_CREATE) */
