/*
 *  SPL - The SPL Programming Language
 *  Copyright (C) 2004, 2005  Clifford Wolf <clifford@clifford.at>
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 *  webspld.c: A HTTP server (using httpsrv.c) with an empedded SPL engine
 */

#define TCP_PORT 3054

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <time.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <signal.h>
#include <fcntl.h>
#include <limits.h>

#ifdef ENABLE_PTHREAD_SUPPORT
#  include <pthread.h>
#  include <semaphore.h>
#endif

#include "spl.h"
#include "webspl_common.h"
#include "compat.h"

#define error() \
	do { \
		fprintf(stderr, "At %s:%d: %s\n", \
				__FILE__, __LINE__, strerror(errno)); \
		exit(1); \
	} while(0)

struct pool_entry {
	struct spl_vm *vm;
	char *session;
	time_t ping;
#ifdef ENABLE_PTHREAD_SUPPORT
	pthread_mutex_t lck;
#endif
};

static int pool_size = 32;
static struct pool_entry *pool = 0;

static char daemon_basedir[PATH_MAX] = ".";

struct code_cache {
	char *filename;
	struct spl_code *code;
	struct code_cache_check *checks;
	time_t last_use;
};

struct code_cache_check {
	char *filename;
	time_t cached_mtime;
	dev_t cached_dev;
	ino_t cached_ino;
	struct code_cache_check *next;
};

static int code_cache_size = 32;
struct code_cache *code_cache_list = 0;

#ifdef ENABLE_PTHREAD_SUPPORT
static pthread_mutex_t pool_lck = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t code_cache_lock = PTHREAD_MUTEX_INITIALIZER;
static sem_t thread_pool_sem;
static int threads = 1;
#endif

static FILE *profile_file = 0;
static int profile_requests = 0;
static int profile_create_new = 0;
static int profile_resume_mem = 0;
static int profile_resume_disk = 0;
static int profile_compiler = 0;
#ifdef ENABLE_PTHREAD_SUPPORT
static pthread_mutex_t profile_lck = PTHREAD_MUTEX_INITIALIZER;
#endif

#ifdef ENABLE_PTHREAD_SUPPORT
#  define PROFILE_INC(__name) do {			\
		pthread_mutex_lock(&profile_lck);		\
	  	profile_ ## __name ++;			\
		pthread_mutex_unlock(&profile_lck);	\
	} while(0)
#else
#  define PROFILE_INC(__name) do { profile_ ## __name ++; } while(0)
#endif

static volatile int got_sigint = 0;

static int listenfd;

static void sigint_handler(int UNUSED(dummy));
static void sigint_handler(int UNUSED(dummy))
{
	if (!got_sigint) {
		got_sigint = 1;
		printf("\nCaught SIGINT (Ctrl-C) or SIGTERM.\n");
	}
	return;
}

static void help(int ret)
{
	printf("\n");
	printf("Usage: webspld [Options]\n");
	printf("\n");
	printf("	-d		Daemonize (fork to background)\n");
	printf("	-l logfile	Write all output to this logfile\n");
	printf("	-p prof-log	Write profiling data to this logfile\n");
	printf("\n");
	printf("	-P tcp-port	Listen on this TCP port (default=%d)\n", TCP_PORT);
	printf("	-A ip-address	Listen on this IP address (default=INADDR_ANY)\n");
	printf("\n");
	printf("	-S size		Set VM pool size (default=32)\n");
	printf("	-C size		Set code cache size (default=32)\n");
#ifdef ENABLE_PTHREAD_SUPPORT
	printf("	-T threads	Number of threads (default=1)\n");
	printf("\n");
	printf("The number of threads must be less or equal the VM pool size!\n");
#endif

	printf("\n");
	exit(ret);
}

static struct spl_vm *webspld_vm_pool_get(const char *session, int create)
{
#ifdef ENABLE_PTHREAD_SUPPORT
retry_entry_point:;
	pthread_mutex_lock(&pool_lck);
#endif

	int ses_len = strcspn(session, ":");
	char *ses_id = my_strndupa(session, ses_len);

	int oldest_entry = 0;
	time_t oldest_entry_ping = pool[0].ping;

	for (int i=0; i<pool_size; i++)
	{
		if (pool[i].vm && !strcmp(pool[i].session, ses_id)) {
			time_t now = time(0);
			int idle = now - pool[i].ping;
			PROFILE_INC(resume_mem);
			printf("Loading VM from slot #%d (session was idle for %d:%02d:%02d).\n",
					i, idle / (60*60), (idle/60) % 60, idle % 60);
			pool[i].ping = now;
#ifdef ENABLE_PTHREAD_SUPPORT
			if (pthread_mutex_trylock(&pool[i].lck)) {
				printf("Can't lock slot #%d -> retry in 1 second.\n", i);
				pthread_mutex_unlock(&pool_lck);
				my_sleep(1);
				goto retry_entry_point;
			}
			pthread_mutex_unlock(&pool_lck);
#endif
			return pool[i].vm;
		}

		if (oldest_entry_ping > pool[i].ping) {
			oldest_entry_ping = pool[i].ping;
			oldest_entry = i;
		}
	}

	char restore_file_name[1024];
	snprintf(restore_file_name, 1024, "webspl_cache/%s.spld", ses_id);

	FILE *restore_file = fopen(restore_file_name, "r");

	if (!create && !restore_file) {
#ifdef ENABLE_PTHREAD_SUPPORT
		pthread_mutex_unlock(&pool_lck);
#endif
		return 0;
	}

#ifdef ENABLE_PTHREAD_SUPPORT
	if (pthread_mutex_trylock(&pool[oldest_entry].lck)) {
		if (restore_file)
			fclose(restore_file);
		printf("Can't lock slot #%d -> retry in 1 second.\n", oldest_entry);
		pthread_mutex_unlock(&pool_lck);
		my_sleep(1);
		goto retry_entry_point;
	}
#endif

	if (pool[oldest_entry].vm) {
		time_t now = time(0);
		int idle = now - pool[oldest_entry].ping;
		printf("Creating VM in slot #%d (old session was idle for %d:%02d:%02d).\n",
				oldest_entry, idle / (60*60), (idle/60) % 60, idle % 60);

		expire_dumpdir(0, EXPIRE_DUMPDIR_TIMEOUT, EXPIRE_DUMPDIR_INTERVAL);

		char dump_file_name[1024];
		snprintf(dump_file_name, 1024, "webspl_cache/%s.spld",
				pool[oldest_entry].session);

		FILE *dump_file = fopen(dump_file_name, "w");

		if ( dump_file ) {
			printf("Dumping old session to `%s'.\n", dump_file_name);
			if (spl_dump_ascii(pool[oldest_entry].vm, dump_file))
				printf("Dumping `%s' failed!\n", dump_file_name);
			fclose(dump_file);
		} else
			printf("Can't write dumpfile `%s': %s\n",
					dump_file_name, strerror(errno));

		spl_vm_destroy(pool[oldest_entry].vm);
		free(pool[oldest_entry].session);
	} else
		printf("Creating VM in slot #%d (slot was empty).\n", oldest_entry);

	pool[oldest_entry].vm = spl_vm_create();
	pool[oldest_entry].session = strdup(ses_id);
	pool[oldest_entry].ping = time(0);

	my_asprintf(&pool[oldest_entry].vm->path, ".:./spl_modules:%s/spl_modules:%s", daemon_basedir, spl_system_modules_dir());
	pool[oldest_entry].vm->codecache_dir = strdup("./webspl_cache");
	pool[oldest_entry].vm->runloop = spl_simple_runloop;

	spl_builtin_register_all(pool[oldest_entry].vm);
	spl_clib_reg(pool[oldest_entry].vm, "write", spl_mod_cgi_write, 0);

	if (restore_file) {
		spl_restore_ascii(pool[oldest_entry].vm, restore_file);
		fclose(restore_file);
		unlink(restore_file_name);
		PROFILE_INC(resume_disk);
	} else
		PROFILE_INC(create_new);

#ifdef ENABLE_PTHREAD_SUPPORT
	pthread_mutex_unlock(&pool_lck);
#endif
	return pool[oldest_entry].vm;
}

static void webspld_vm_pool_put(struct spl_vm *vm)
{
#ifdef ENABLE_PTHREAD_SUPPORT
	for (int i=0; i<pool_size; i++)
		if (pool[i].vm == vm) {
			struct spl_task *task = spl_task_lookup(vm, "main");
			if (!task || !task->code) {
				printf("Freeing VM in slot #%d (script terminated).\n", i);
				spl_vm_destroy(pool[i].vm);
				free(pool[i].session);
				pool[i].vm = 0;
				pool[i].session = 0;
				pool[i].ping = 0;
			}
			pthread_mutex_unlock(&pool[i].lck);
		}
#endif
}

static void webspld_vm_pool_cleanup()
{
#ifdef ENABLE_PTHREAD_SUPPORT
	/* locking everything */
	pthread_mutex_lock(&pool_lck);
#endif
	for (int i=0; i<pool_size; i++)
	{
#ifdef ENABLE_PTHREAD_SUPPORT
		pthread_mutex_lock(&pool[i].lck);
#endif
		if (!pool[i].vm) continue;

		printf("Dumping session `%s'.\n", pool[i].session);

		char dump_file_name[1024];
		snprintf(dump_file_name, 1024, "webspl_cache/%s.spld",
				pool[i].session);

		FILE *dump_file = fopen(dump_file_name, "w");

		if ( dump_file ) {
			if (spl_dump_ascii(pool[i].vm, dump_file))
				printf("Dumping `%s' failed!\n", dump_file_name);
			fclose(dump_file);
		} else
			printf("Can't write dumpfile `%s': %s\n",
					dump_file_name, strerror(errno));

		spl_vm_destroy(pool[i].vm);
		free(pool[i].session);

		pool[i].vm = 0;
		pool[i].session = 0;
		pool[i].ping = 0;
	}
}

static void free_code_cache(struct code_cache *c)
{
	struct code_cache_check *chk = c->checks;
	while (chk) {
		struct code_cache_check *next_chk = chk->next;
		free(chk->filename);
		free(chk);
		chk = next_chk;
	}

	if (c->filename) free(c->filename);
	if (c->code) spl_code_put(c->code);
	memset(c, 0, sizeof(struct code_cache));
}

static struct code_cache *wrap_spl_malloc_file_c;

static void *wrap_spl_malloc_file(const char *filename, int *size)
{
	struct code_cache *c = wrap_spl_malloc_file_c;
	struct code_cache_check *chk = c->checks;

	while (chk) {
		if (!strcmp(chk->filename, filename))
			goto do_not_add_twice;
		chk = chk->next;
	}

	struct stat sb;
	int rc = lstat(filename, &sb);
	if (rc) return 0;

	chk = malloc(sizeof(struct code_cache_check));
	chk->filename = strdup(filename);
	chk->cached_mtime = sb.st_mtime;
	chk->cached_dev = sb.st_dev;
	chk->cached_ino = sb.st_ino;
	chk->next = c->checks;
	c->checks = chk;

do_not_add_twice:
	return spl_malloc_file(filename, size);
}

static struct spl_code *cached_filename_to_codepage(struct spl_vm *vm, const char *file)
{
#ifdef ENABLE_PTHREAD_SUPPORT
	pthread_mutex_lock(&code_cache_lock);
#endif

	struct code_cache *c = 0;
	struct code_cache *oldest_c = code_cache_list;
	struct spl_code *code = 0;

	for (int i=0; i<code_cache_size; i++)
	{
		if (oldest_c->last_use > code_cache_list[i].last_use)
			oldest_c = code_cache_list + i;

		if (!code_cache_list[i].last_use)
			continue;

		if (strcmp(code_cache_list[i].filename, file))
			continue;

		struct code_cache_check *chk = code_cache_list[i].checks;

		while (chk) {
			struct stat sb;
			int rc = lstat(chk->filename, &sb);

			if (rc || chk->cached_mtime != sb.st_mtime || chk->cached_dev != sb.st_dev || chk->cached_ino != sb.st_ino)
				goto flush_this_entry;

			chk = chk->next;
		}

		c = code_cache_list + i;
		break;

flush_this_entry:
		free_code_cache(code_cache_list + i);

		if (oldest_c->last_use > code_cache_list[i].last_use)
			oldest_c = code_cache_list + i;
		break;
	}

	if (!c)
	{
		c = oldest_c;
		free_code_cache(c);

		wrap_spl_malloc_file_c = c;

		PROFILE_INC(compiler);
		printf("Compiling `%s' (cache bay %d).\n", file, (int)(c - code_cache_list));

		char *spl_source = wrap_spl_malloc_file(file, 0);
		if (!spl_source) {
			spl_report(SPL_REPORT_HOST, vm, "Can't read script file!\n");
#ifdef ENABLE_PTHREAD_SUPPORT
			pthread_mutex_unlock(&code_cache_lock);
#endif
			return 0;
		}

		struct spl_asm *as = spl_asm_create();
		as->vm = vm;

		if ( spl_compiler(as, spl_source, file, wrap_spl_malloc_file, 1) ) {
			spl_asm_destroy(as);
			free(spl_source);
#ifdef ENABLE_PTHREAD_SUPPORT
			pthread_mutex_unlock(&code_cache_lock);
#endif
			return 0;
		}

		spl_asm_add(as, SPL_OP_HALT, "1");
		spl_optimizer(as);

		code = spl_asm_dump(as);
		code->id = strdup(file);
		spl_asm_destroy(as);
		free(spl_source);

		c->last_use = time(0);
		c->filename = strdup(file);
		c->code = spl_code_get(code);
	}
	else
	{
		printf("Using cached code page (cache bay %d).\n", (int)(c - code_cache_list));
		c->last_use = time(0);
		code = spl_code_get(c->code);
	}

#ifdef ENABLE_PTHREAD_SUPPORT
	pthread_mutex_unlock(&code_cache_lock);
#endif
	return code;
}

struct handle_conn_args {
	int fd;
	struct sockaddr_in addr;
};

static void *handle_conn(void *args_p)
{
	struct handle_conn_args *args = args_p;

	printf("\nConnection from %s:%u.\n",
		inet_ntoa(args->addr.sin_addr), ntohs(args->addr.sin_port));
 
	PROFILE_INC(requests);
	handle_http_request(args->fd,
			webspld_vm_pool_get, webspld_vm_pool_put,
			code_cache_list ? cached_filename_to_codepage : 0);

	if (close(args->fd) < 0) error();
	printf("Connection closed.\n");

#ifdef ENABLE_PTHREAD_SUPPORT
	sem_post(&thread_pool_sem);
	free(args);
#endif
	return 0;
}

static void listen_loop()
{
	struct sockaddr_in addr;
	socklen_t addrlen;
	time_t last_profile_out = time(0);
	int rc, fd;

	printf("Listening...\n");

#ifdef ENABLE_PTHREAD_SUPPORT
	sem_init(&thread_pool_sem, 0, threads);
#endif

	while (!got_sigint)
	{
		struct timeval select_timeout;
		select_timeout.tv_sec = 3;
		select_timeout.tv_usec = 0;

		fd_set listen_fds;
		FD_ZERO(&listen_fds);
		FD_SET(listenfd, &listen_fds);

		rc = select(listenfd+1, &listen_fds, NULL, NULL, &select_timeout);
		if (rc < 0 && (errno == EINTR || errno == EAGAIN)) continue;
		if (rc == 0) continue;
		if (rc < 0) error();

		addrlen = sizeof(addr);
		fd = accept(listenfd, (struct sockaddr *) &addr, &addrlen);
		if (fd < 0 && (errno == EINTR || errno == EAGAIN)) continue;
		if (fd < 0) error();

#ifdef ENABLE_PTHREAD_SUPPORT
		struct handle_conn_args *args = malloc(sizeof(struct handle_conn_args));
		args->addr = addr;
		args->fd = fd;

		while (sem_wait(&thread_pool_sem) == -1) { /* loop */ }

		pthread_t dummy_thread_id;
		if (pthread_create(&dummy_thread_id, 0, handle_conn, args) != 0) error();
		pthread_detach(dummy_thread_id);
#else
		struct handle_conn_args args;
		args.addr = addr;
		args.fd = fd;

		handle_conn(&args);
#endif
		if (profile_file) {
			time_t now = time(0);
			if (now > last_profile_out+60) {
#ifdef ENABLE_PTHREAD_SUPPORT
				pthread_mutex_lock(&profile_lck);
#endif
				fprintf(profile_file, "%15Ld %15Ld %7d %7d %7d %7d %7d\n",
					(long long)last_profile_out,
					(long long)now, profile_requests,
					profile_create_new, profile_resume_mem,
					profile_resume_disk, profile_compiler);
				fflush(profile_file);
				last_profile_out = now;
				profile_requests = 0;
				profile_create_new = 0;
				profile_resume_mem = 0;
				profile_resume_disk = 0;
				profile_compiler = 0;
#ifdef ENABLE_PTHREAD_SUPPORT
				pthread_mutex_unlock(&profile_lck);
#endif
			}
		}
	}

#ifdef ENABLE_PTHREAD_SUPPORT
	printf("Waiting for worker threads to shut down..\n");
	for (int i=0; i<threads; i++)
		while (sem_wait(&thread_pool_sem) == -1) { /* loop */ }
#endif
}

int main(int argc, char **argv)
{
	int tcp_port = TCP_PORT;
	char *bind_addr = 0;
	int daemonize = 0;
	char *logfile = 0;

	for (int opt; (opt = getopt(argc, argv, "P:A:S:C:T:dl:p:")) != -1; ) {
		switch ( opt )
		{
			case 'P':
				tcp_port = atoi(optarg);
				if (tcp_port <= 0) tcp_port = TCP_PORT;
				break;
			case 'A':
				bind_addr = optarg;
				break;
			case 'S':
				pool_size = atoi(optarg);
				if (pool_size <= 0) pool_size = 1;
				break;
			case 'C':
				code_cache_size = atoi(optarg);
				break;
#ifdef ENABLE_PTHREAD_SUPPORT
			case 'T':
				threads = atoi(optarg);
				if (threads <= 0) threads = 1;
				break;
#endif
			case 'd':
				daemonize = 1;
				break;
			case 'l':
				logfile = optarg;
				break;
			case 'p':
				profile_file = fopen(optarg, "a");
				break;
			default:
				help(1);
		}
	}

#ifdef ENABLE_PTHREAD_SUPPORT
	if (threads > pool_size)
		help(1);
#endif

	if (daemonize) {
		close(0); close(1); close(2);
		open("/dev/null", O_RDONLY);
		open("/dev/null", O_WRONLY);
		open("/dev/null", O_WRONLY);
		if (fork()) return 0;
	}

	if (logfile) {
		close(1); close(2);
		open(logfile, O_CREAT|O_WRONLY|O_APPEND, 0666);
		open(logfile, O_CREAT|O_WRONLY|O_APPEND, 0666);
	}

	int rc, on = 1;
	struct sockaddr_in addr;
	struct linger sl = { 1, 5 };

	getcwd(daemon_basedir, PATH_MAX);

	printf("Loading SPL CGI module.\n");
	SPL_REGISTER_BUILTIN_MODULE(cgi);
	spl_report = spl_mod_cgi_reportfunc;

	printf("Allocating VM pool (%d slots).\n", pool_size);
	pool = calloc(pool_size, sizeof(struct pool_entry));

	if (code_cache_size > 0) {
		printf("Allocating code cache (%d bays).\n", code_cache_size);
		code_cache_list = calloc(code_cache_size, sizeof(struct code_cache));
	}

	create_dumpdir(0);
	expire_dumpdir(0, EXPIRE_DUMPDIR_TIMEOUT, EXPIRE_DUMPDIR_INTERVAL);

#ifdef ENABLE_PTHREAD_SUPPORT
	for (int i=0; i<pool_size; i++)
		pthread_mutex_init(&pool[i].lck, 0);
#endif

	listenfd = socket(AF_INET, SOCK_STREAM, 0);
	if (listenfd < 0) error();

	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	if (bind_addr) {
		if (inet_aton(bind_addr, &addr.sin_addr) == 0) help(1);
	} else
		addr.sin_addr.s_addr = htonl(INADDR_ANY);
	addr.sin_port = htons(tcp_port);

	rc = setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &on, (socklen_t) sizeof(on));
	if (rc < 0) error();

	rc = setsockopt(listenfd, SOL_SOCKET, SO_LINGER, &sl, (socklen_t) sizeof(sl));
	if (rc < 0) error();

	rc = bind(listenfd, (struct sockaddr *) &addr, sizeof(addr));
	if (rc < 0) error();

	printf("Bound to port %d on %s.\n\n",
			tcp_port, inet_ntoa(addr.sin_addr));

	rc = listen(listenfd, 32);
	if (rc < 0) error();

	fcntl(listenfd, F_SETFL, O_NONBLOCK);

	signal(SIGINT, sigint_handler);
	signal(SIGTERM, sigint_handler);
	signal(SIGPIPE, SIG_IGN);

	if (profile_file) {
		fprintf(profile_file, "%15s %15s %7s %7s %7s %7s %7s\n",
				"begin", "end", "http", "vm-new",
				"vm-mem", "vm-disk", "compile");
		fflush(profile_file);
	}

	listen_loop();

	printf("Running cleanup procedures..\n");
	webspld_vm_pool_cleanup();
	SPL_FREE_BUILTIN_MODULES();
	free(pool);

	if (code_cache_size) {
		for (int i=0; i<code_cache_size; i++)
			free_code_cache(code_cache_list + i);
		free(code_cache_list);
	}

	printf("Bye, bye.\n");
	close(listenfd);

	return 0;
}