/* This file is part of Ping903
   Copyright (C) 2020-2023 Sergey Poznyakoff
  
   Ping903 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 3, or (at your option)
   any later version.
  
   Ping903 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 Ping903.  If not, see <http://www.gnu.org/licenses/>.
*/
#include <config.h>
#define _GNU_SOURCE
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/stat.h>

#include <netinet/in_systm.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <netinet/ip_icmp.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <string.h>
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <pthread.h>
#include <limits.h>
#include <poll.h>
#include "ping903.h"
#include "json.h"
#include "defs.h"

/* Time in seconds between two subsequent probes. */
unsigned long probe_interval = 60;
/* Time between two subsequent echo requests within the same probe. */
unsigned long ping_interval = 1;
/* Number of echo requests per probe */
unsigned long ping_count = 10;
/* Number of unanswered echo requests after which the host is declared dead. */
unsigned long ping_tolerance = 3;

/* Initial value for the tmin member of struct hostping */
#define HOSTPING_TMIN_INIT 999999999.0

static double
nabs(double a)
{
	return (a < 0) ? -a : a;
}

static double
nsqrt(double a, double prec)
{
	double x0, x1;

	if (a < 0)
		return 0;
	if (a < prec)
		return 0;
	x1 = a / 2;
	do {
		x0 = x1;
		x1 = (x0 + a / x0) / 2;
	} while (nabs(x1 - x0) > prec);
	return x1;
}

static inline int
timeval_is_null(struct timeval const *tv)
{
	return tv->tv_sec == 0 && tv->tv_usec == 0;
}

static inline double
timeval_to_double(struct timeval const *tv)
{
	return (double)tv->tv_sec + (double)tv->tv_usec / 1000000;
}

enum { NANOSEC_IN_SEC = 1000000000 };

static inline void
timespec_add(struct timespec const *a, struct timespec const *b,
	     struct timespec *res)
{
	res->tv_sec = a->tv_sec + b->tv_sec;
	res->tv_nsec = a->tv_nsec + b->tv_nsec;
	if (res->tv_nsec >= NANOSEC_IN_SEC) {
		++res->tv_sec;
		res->tv_nsec -= NANOSEC_IN_SEC;
	}
}

static inline void
timespec_incr(struct timespec *a, struct timespec const *b)
{
	struct timespec t = *a;
	timespec_add(&t, b, a);
}

static void
hostping_free(HOSTPING *hp)
{
	free(hp->name);
	free(hp->addr);
	pthread_mutex_destroy(&hp->mutex);
	free(hp);
}

static HOSTPING *
hostping_create(char const *name, struct sockaddr *addr, socklen_t addrlen)
{
	HOSTPING *hp;
	size_t hostsize = sizeof(*hp) +
		           (ping_count - 1) * sizeof(hp->nreply[0]);

	hp = malloc(hostsize);

	if (hp) {
		memset(hp, 0, hostsize);
		hp->tmin = HOSTPING_TMIN_INIT;
		hp->name = strdup(name);
		if (!hp)
			goto err;
		hp->addr = malloc(addrlen);
		if (!hp)
			goto err;
		memcpy(hp->addr, addr, addrlen);
		hp->addrlen = addrlen;
		pthread_mutex_init(&hp->mutex, NULL);
	}

	return hp;
err:
	hostping_free(hp);
	return NULL;
}

static inline void
hostping_lock(HOSTPING *host)
{
	pthread_mutex_lock(&host->mutex);
}

static inline void
hostping_unlock(HOSTPING *host)
{
	pthread_mutex_unlock(&host->mutex);
}

typedef struct hostlist {
	HOSTPING *head, *tail;
	size_t count;
} HOSTLIST;

static HOSTLIST *
hostlist_create(void)
{
	HOSTLIST *hl = malloc(sizeof(*hl));
	if (hl) {
		hl->head = hl->tail = NULL;
		hl->count = 0;
	}
	return hl;
}

static void
hostlist_free(HOSTLIST *list)
{
	HOSTPING *hp = list->head;
	while (hp) {
		HOSTPING *next = hp->next;
		hostping_free(hp);
		hp = next;
	}
	free(list);
}

static void
hostlist_add(HOSTLIST *list, HOSTPING *hp)
{
	hp->prev = list->tail;
	if (list->tail)
		list->tail->next = hp;
	else
		list->head = hp;
	list->tail = hp;
	list->count++;
}

static void
hostlist_remove(HOSTLIST *list, HOSTPING *hp)
{
	if (hp->prev)
		hp->prev->next = hp->next;
	else
		list->head = hp->next;
	if (hp->next)
		hp->next->prev = hp->prev;
	else
		list->tail = hp->prev;
	list->count--;
}

static void
hostlist_concat(HOSTLIST *a, HOSTLIST *b)
{
	if (b->head)
		b->head->prev = a->tail;
	if (a->tail)
		a->tail->next = b->head;
	else
		a->head = b->head;
	a->tail = b->head;
	a->count += b->count;

	b->head = b->tail = NULL;
	b->count = 0;
}

static HOSTLIST *hostlist;
static HOSTPING *immutable_hostping_tail;
static RBT_TREE *host_name_tree, *host_addr_tree;
static pthread_rwlock_t hostlist_rwlock = PTHREAD_RWLOCK_INITIALIZER;

typedef enum update_type {
	UPDATE_APPEND,
	UPDATE_REPLACE,
	UPDATE_DELETE
} UPDATE_TYPE;

static pthread_mutex_t update_mutex = PTHREAD_MUTEX_INITIALIZER;

enum {
	CHECK_HOST_NOT_FOUND,
	CHECK_HOST_FOUND,
	CHECK_HOST_IMMUTABLE
};

static int check_host(char const *name, struct sockaddr *addr, socklen_t len);
static int update_add(UPDATE_TYPE t, void *data);

static int
host_name_cmp(HOSTPING *a, HOSTPING *b)
{
	return strcmp(a->name, b->name);
}

static int
host_addr_cmp(HOSTPING *a, HOSTPING *b)
{
	if (a->addrlen != b->addrlen)
		return a->addrlen - b->addrlen;
	return memcmp(a->addr, b->addr, b->addrlen);
}

void
pinger_setup(void)
{
	hostlist = hostlist_create();
	if (!hostlist)
		emalloc_die();
	host_name_tree = rbt_tree_create(host_name_cmp);
	if (!host_name_tree)
		emalloc_die();
	host_addr_tree = rbt_tree_create(host_addr_cmp);
	if (!host_addr_tree)
		emalloc_die();	
}

RBT_LOOKUP_RESULT
pinger_rbt_insert(HOSTPING *hp)
{
	RBT_LOOKUP_RESULT res;
	
	res = rbt_insert(host_name_tree, hp);
	if (res != RBT_LOOKUP_NOENT)
		return res;
	res = rbt_insert(host_addr_tree, hp);
	if (res != RBT_LOOKUP_NOENT)
		rbt_delete(host_name_tree, hp);
	return res;
}

void
pinger_rbt_delete(HOSTPING *hp)
{
	rbt_delete(host_name_tree, hp);
	rbt_delete(host_addr_tree, hp);
}

static inline HOSTPING *
hostping_lookup_by_name(char const *name)
{
	HOSTPING key;
	
	key.name = (char*) name;
	return rbt_lookup(host_name_tree, &key);
}

static inline HOSTPING *
hostping_lookup_by_addr(struct sockaddr *addr, socklen_t addrlen)
{
	HOSTPING key;

	if (!addr)
		return NULL;
	key.addr = addr;
	key.addrlen = addrlen;
	return rbt_lookup(host_addr_tree, &key);
}

RBT_LOOKUP_RESULT
pinger_host_add(char const *name, struct sockaddr *addr, socklen_t addrlen)
{
	RBT_LOOKUP_RESULT res;
	HOSTPING *hp;

	hp = hostping_create(name, addr, addrlen);
	if (!hp)
		return RBT_LOOKUP_FAILURE;
	res = pinger_rbt_insert(hp);
	if (res == RBT_LOOKUP_NOENT)
		hostlist_add(hostlist, hp);
	else
		hostping_free(hp);
	return res;
}

#define FOR_EACH_HOSTPING(hp) for (hp = hostlist->head; hp; hp = hp->next)
#define FOR_EACH_MUTABLE_HOSTPING(hp)			  \
	for (hp = immutable_hostping_tail                 \
		     ? immutable_hostping_tail->next	  \
		     : hostlist->head;			  \
  	     hp;					  \
	     hp = hp->next)

static void
hostping_extract_stat(HOSTPING *host, struct host_stat *st)
{
	memcpy(&st->start_tv, &host->start_tv,
	       sizeof(host->stat_last.start_tv));
	memcpy(&st->stop_tv,
	       timeval_is_null(&host->recv_tv)
	         ? &host->xmit_tv : &host->recv_tv,
	       sizeof(host->stat_last.stop_tv));
	st->xmit_count = host->xmit_count;
	st->recv_count = host->recv_count;
	st->dup_count  = host->dup_count;
	st->tmin       = host->tmin;            
	st->tmax       = host->tmax;
	if (host->recv_count > 0) {
		double total = host->recv_count;
		double avg = host->tsum / total;
		double vari = host->tsumsq / total - avg * avg;
	
		st->avg    = avg;
		st->stddev = nsqrt(vari, 0.0005);
	}
}

static void
hostping_commit(HOSTPING *host)
{	
	if (host->stat_last.status != HOST_STAT_VALID) {
		hostping_extract_stat(host, &host->stat_last);
		host->stat_last.status = HOST_STAT_VALID;
	}
}

/* Reset runtime statistics counters */
static void
hostping_reset(HOSTPING *host)
{
	hostping_lock(host);
	if (host->xmit_count > 0)
		hostping_commit(host);
	host->xmit_count = 0;
	host->recv_count = 0;
	host->dup_count = 0;
	host->tmin = HOSTPING_TMIN_INIT;
	host->tmax = 0;
	host->tsum = 0;
	host->tsumsq = 0;
	memset(host->nreply, 0, ping_count * sizeof(host->nreply[0]));
	switch (host->stat_last.status) {
	case HOST_STAT_INIT:
		break;
	case HOST_STAT_VALID:
		host->stat_last.status = HOST_STAT_PENDING;
		break;
	case HOST_STAT_PENDING:
		host->stat_last.status = HOST_STAT_INVALID;
		break;
	case HOST_STAT_INVALID:
		break;
	}
	hostping_unlock(host);
}

static int
get_hostping_stat(HOSTPING *host, struct json_value **retval)
{
	struct json_value *obj, *v;
	struct host_stat const *st;
	struct host_stat stbuf;
	static char const *host_stat_status[] = {
		"init",
		"valid",
		"pending",
		"invalid"
	};
	int validity;
	double ts;

	hostping_lock(host);
	if (!(obj = json_new_object()))
		goto err;

	if (!(v = json_new_string(host->name))
	    || json_object_set(obj, "name", v))
		goto err;

	if (host_stat_is_valid(&host->stat_last)) {
		st = &host->stat_last;
		validity = 1;
	} else if (host->stat_last.status == HOST_STAT_INIT
		   && host->xmit_count > ping_tolerance) {
		hostping_extract_stat(host, &stbuf);
		stbuf.status = HOST_STAT_PENDING;
		st = &stbuf;
		validity = 1;
	} else {
		st = NULL;
		validity = 0;
	}

	if (!(v = json_new_bool(validity))
	    || json_object_set(obj, "validity", v))
		goto err;

	if (!(v = json_new_string(host_stat_status[host->stat_last.status]))
	    || json_object_set(obj, "status", v))
		goto err;	
	ts = timeval_to_double(&host->xmit_tv);
	if (!(v = json_new_number(ts))
	    || json_object_set(obj, "xmit-timestamp", v))
			goto err;

	if (st) {
		int is_alive = st->xmit_count - st->recv_count
			        < ping_tolerance;
	
		if (!(v = json_new_string(host_stat_status[st->status]))
		    || json_object_set(obj, "status", v))
			goto err;	

		ts = timeval_to_double(&st->start_tv);
		if (!(v = json_new_number(ts))
		    || json_object_set(obj, "start-timestamp", v))
			goto err;
		ts = timeval_to_double(&st->stop_tv);
		if (!(v = json_new_number(ts))
		    || json_object_set(obj, "stop-timestamp", v))
			goto err;
		if (!(v = json_new_bool(is_alive))
		    || json_object_set(obj, "alive", v))
			goto err;
		if (!(v = json_new_number(st->xmit_count))
		    || json_object_set(obj, "xmit", v))
			goto err;		
		if (!(v = json_new_number(st->recv_count))
		    || json_object_set(obj, "recv", v))
			goto err;
		if (!(v = json_new_number(st->dup_count))
		    || json_object_set(obj, "dup", v))
			goto err;
		if (!(v = json_new_number((double) 100
					  * (st->xmit_count - st->recv_count)
					  / st->xmit_count))
		    || json_object_set(obj, "loss", v))
			goto err;
		
		if (st->recv_count > 0) {
			if (!(v = json_new_number(st->tmin))
			    || json_object_set(obj, "tmin", v))
				goto err;
			if (!(v = json_new_number(st->tmax))
			    || json_object_set(obj, "tmax", v))
				goto err;
			if (!(v = json_new_number(st->avg))
			    || json_object_set(obj, "avg", v))
				goto err;
			if (!(v = json_new_number(st->stddev))
			    || json_object_set(obj, "stddev", v))
				goto err;
		}
	} else {
		if (!(v = json_new_string(host_stat_status[host->stat_last.status]))
		    || json_object_set(obj, "status", v))
		goto err;	
	}

	hostping_unlock(host);

	*retval = obj;
	return GET_OK;
	
  err:
	hostping_unlock(host);
	error("out of memory when formatting statistics for %s", host->name);
	json_value_free(v);
	json_value_free(obj);

	return GET_NOMEM;
}

int
get_hostname_stat(char const *name, struct json_value **retval)
{
	HOSTPING *hp;
	int rc = GET_NOENT;

	*retval = NULL;
	pthread_rwlock_rdlock(&hostlist_rwlock);
	hp = hostping_lookup_by_name(name);
	if (hp)
		rc = get_hostping_stat(hp, retval);
	pthread_rwlock_unlock(&hostlist_rwlock);
	return rc;
}

int
get_all_host_stat(struct json_value **retval)
{
	struct json_value *ar;
	HOSTPING *hp;
	int rc = GET_OK;
	
	if (!(ar = json_new_array()))
		return GET_NOMEM;
	pthread_rwlock_rdlock(&hostlist_rwlock);
	FOR_EACH_HOSTPING(hp) {
		struct json_value *v;
		if (get_hostping_stat(hp, &v)) {
			rc = GET_NOMEM;
			break;
		}
		if (json_array_append(ar, v)) {
			json_value_free(v);
			rc = GET_NOMEM;
			break;
		}
	}
	pthread_rwlock_unlock(&hostlist_rwlock);
	if (rc == GET_OK)
		*retval = ar;
	else
		json_value_free(ar);
	return rc;
}

int
get_all_hosts(struct json_value **retval)
{
	struct json_value *ar;
	HOSTPING *hp;
	int rc = GET_OK;
	
	if (!(ar = json_new_array()))
		return GET_NOMEM;
	pthread_rwlock_rdlock(&hostlist_rwlock);
	FOR_EACH_HOSTPING(hp) {
		struct json_value *jv;
		jv = json_new_string(hp->name);
		if (!jv) {
			rc = GET_NOMEM;
			break;
		}
		if (json_array_append(ar, jv)) {
			json_value_free(jv);
			rc = GET_NOMEM;
			break;
		}
	}
	pthread_rwlock_unlock(&hostlist_rwlock);
	if (rc == GET_OK)
		*retval = ar;
	else
		json_value_free(ar);
	return rc;
}

int 
get_host_matches(struct json_value *ar)
{
	struct addrent {
		char const *name;
		struct addrinfo *res;
	};
	size_t count = json_array_length(ar);
	size_t i;
	struct addrinfo hints;
	int errors = 0;
	
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_INET;
	hints.ai_protocol = IPPROTO_TCP;
	
	pthread_rwlock_rdlock(&hostlist_rwlock);
	for (i = 0; !errors && i < count; i++) {
		char const *name;
		int rc;
		struct addrinfo *res, *rp;
		struct json_value *obj, *jv, *var;
		HOSTPING *hp;
		
		json_array_get(ar, i, &obj);
		json_object_get(obj, "name", &jv);
		name = jv->v.s;

		rc = getaddrinfo(name, NULL, &hints, &res);		
		switch (rc) {
		case 0:
			break;

		case EAI_FAIL:
#ifdef EAI_NODATA
		case EAI_NODATA:
#endif
		case EAI_NONAME:
			if (!(jv = json_new_string("Doesn't resolve"))
			    || json_object_set(obj, "error", jv)) {
				json_value_free(jv);
			}
			errors = 1;
			break;

		case EAI_MEMORY:
			errors = 1;
			break;

		case EAI_SYSTEM:
			error("Internal error resolving %s: %s",
			      name, strerror(errno));
			errors = 1;
			break;

		default:
			error("Internal error resolving %s: %s",
			      name, gai_strerror(rc));
			errors = 1;
			break;
		}
		if (errors)
			break;

		json_object_get(obj, "hosts", &var);

		hp = hostping_lookup_by_name(name);
		if (hp) {
			if (!(jv = json_new_string(hp->name))
			    || json_array_append(var, jv)) {
				json_value_free(jv);
				errors = 1;
			}
		}
		for (rp = res; !errors && rp; rp = rp->ai_next) {
			hp = hostping_lookup_by_addr(rp->ai_addr,
						     rp->ai_addrlen);
			
			if (hp) {
				if (!(jv = json_new_string(hp->name))
				    || json_array_append(var, jv)) {
					json_value_free(jv);
					errors = 1;
				}
			}
		}
		freeaddrinfo(res);
	}
	pthread_rwlock_unlock(&hostlist_rwlock);

	return errors;
}

int
pinger_host_delete_by_name(char const *name)
{
	int rc;

	pthread_mutex_lock(&update_mutex);

	if (check_host(name, NULL, 0) == CHECK_HOST_FOUND) {
		char *cp = strdup(name);
		if (cp) {
			if (update_add(UPDATE_DELETE, cp) == 0)
				rc = UPD_OK;
			else {
				rc = UPD_NOMEM;
				free(cp);
			}
		}
	} else {
		rc = UPD_EXISTS;
	}
	pthread_mutex_unlock(&update_mutex);
	return rc;
}

int
pinger_host_add_name(char const *name)
{
	int rc = 0;
	struct addrinfo hints, *res;
	HOSTPING *hp;
	
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_INET;
	hints.ai_protocol = IPPROTO_TCP;
	rc = getaddrinfo(name, NULL, &hints, &res);
	if (rc) {
		return UPD_NORESOLV;
	}

	pthread_mutex_lock(&update_mutex);
	if (check_host(name, res->ai_addr, res->ai_addrlen) == CHECK_HOST_NOT_FOUND) {
		rc = UPD_NOMEM;
		hp = hostping_create(name, res->ai_addr, res->ai_addrlen);
		if (hp) {
			HOSTLIST *hl = hostlist_create();
			if (hl) {
				hostlist_add(hl, hp);
				if (update_add(UPDATE_APPEND, hl) == 0)
					rc = UPD_OK;
				else
					hostlist_free(hl);
			} else
				hostping_free(hp);
		}
	} else {
		rc = UPD_EXISTS;
	}
	pthread_mutex_unlock(&update_mutex);
	freeaddrinfo(res);
	return rc;
}

int
pinger_hostlist_set(struct json_value *obj, char const **err_text,
		    int *err_pos)
{
	int mode = UPDATE_APPEND;
	struct json_value *ar;
	HOSTLIST *tmp = NULL;
	HOSTPING *hp;
	struct addrinfo hints;
	size_t i;
	int ret;
#define RETERR(status,text,pos) do {\
		*err_text = text;   \
		*err_pos = pos;	    \
		ret = status;	    \
		goto err;	    \
} while (0)

	if (!obj)
		RETERR(MHD_HTTP_BAD_REQUEST, "invalid object", 0);
	if (obj->type == json_object) {
		struct json_value *jv;
		if (json_object_get(obj, "mode", &jv)) {
			RETERR(MHD_HTTP_BAD_REQUEST,
			       "\"mode\" attribute missing",
			       0);
		}
		if (strcmp(jv->v.s, "append") == 0)
			mode = UPDATE_APPEND;
		else if (strcmp(jv->v.s, "replace") == 0)
			mode = UPDATE_REPLACE;
		else
			RETERR(MHD_HTTP_BAD_REQUEST, "\"mode\": invalid value",
			       0);
		
		if (json_object_get(obj, "ip-list", &ar))
			RETERR(MHD_HTTP_BAD_REQUEST,
			       "\"ip-list\" attribute missing",
			       0);
	} else if (obj->type == json_array) {
		mode = UPDATE_APPEND;
		ar = obj;
	} else {
		RETERR(MHD_HTTP_BAD_REQUEST, "bad object type", 0);
	}

	tmp = hostlist_create();
	if (!tmp)
		RETERR(MHD_HTTP_INTERNAL_SERVER_ERROR,NULL,0);

	memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_INET;
	hints.ai_protocol = IPPROTO_TCP;

	for (i = 0; i < ar->v.a->oc; i++) {
		struct addrinfo *res;
		int rc;
		struct json_value *jv;

		if (json_array_get(ar, i, &jv) || jv->type != json_string)
			RETERR(MHD_HTTP_BAD_REQUEST,
			       "bad object type", i + 1);
		
		rc = getaddrinfo(jv->v.s, NULL, &hints, &res);
		if (rc)
			RETERR(MHD_HTTP_BAD_REQUEST,
			       "host name does not resolve", i + 1);
		switch (check_host(jv->v.s, res->ai_addr, res->ai_addrlen)) {
		case CHECK_HOST_NOT_FOUND:
			/* Safe to add */
			break;
		case CHECK_HOST_FOUND:
			if (mode == UPDATE_REPLACE) {
				/* It's OK in replace mode */
				break;
			}
			/* fall through */
		case CHECK_HOST_IMMUTABLE:
			freeaddrinfo(res);
			RETERR(MHD_HTTP_BAD_REQUEST,
			       "host name or IP address already exists",
			       i + 1);
		}
		hp = hostping_create(jv->v.s, res->ai_addr, res->ai_addrlen);
		freeaddrinfo(res);
		if (!hp)
			RETERR(MHD_HTTP_INTERNAL_SERVER_ERROR, NULL, i + 1);
		hostlist_add(tmp, hp);
	}

	if (update_add(mode, tmp)) {
		RETERR(MHD_HTTP_INTERNAL_SERVER_ERROR, NULL, 0);
	} else {
		*err_text = NULL;
		*err_pos = 0;
		return MHD_HTTP_OK;
	}
err:
	hostlist_free(tmp);
	return ret;
}

typedef struct update_entry {
	enum update_type type;
	struct update_entry *next;
	union {
		void *ptr;
		HOSTLIST *hlist;
		char *name;
	} v;
} UPDATE_ENTRY;

static UPDATE_ENTRY *update_head, *update_tail;

static void save_mutable_ip_list(void);

static int
update_add(UPDATE_TYPE t, void *data)
{
	UPDATE_ENTRY *uent = malloc(sizeof(*uent));
	
	if (!uent)
		return -1;
	
	uent->type = t;
	uent->v.ptr = data;
	uent->next = NULL;

	if (uent->type == UPDATE_REPLACE) {
		while (update_head) {
			UPDATE_ENTRY *next = update_head->next;
			switch (update_head->type) {
			case UPDATE_APPEND:
			case UPDATE_REPLACE:
				hostlist_free(update_head->v.hlist);
				break;
			case UPDATE_DELETE:
				free(update_head->v.name);
			}
			free(update_head);
			update_head = next;
		}
		update_tail = NULL;
	}
	
	if (update_tail)
		update_tail->next = uent;
	else
		update_head = uent;
	update_tail = uent;
	return 0;
}

static void
hostlist_copy_stat(HOSTLIST *hl)
{
	HOSTPING *hp;
	for (hp = hl->head; hp; hp = hp->next) {
		HOSTPING *cur = hostping_lookup_by_name(hp->name);
		if (cur)
			hp->stat_last = cur->stat_last;
	}
}

static void
hostlist_index(HOSTLIST *hl)
{
	HOSTPING *hp;
	for (hp = hl->head; hp; hp = hp->next) {
		pinger_rbt_insert(hp); //FIXME: Handle RBT_LOOKUP_FAILURE (out of memory)
	}
}	

static void
hostlist_unindex(HOSTLIST *hl)
{
	HOSTPING *hp;
	for (hp = hl->head; hp; hp = hp->next) {
		pinger_rbt_delete(hp);
	}
}

void
p903_update_commit(void)
{
	HOSTPING *hp;
	int upd;
	
	pthread_mutex_lock(&update_mutex);
	pthread_rwlock_wrlock(&hostlist_rwlock);
	upd = update_head != NULL;
	while (update_head) {
		UPDATE_ENTRY *next = update_head->next;
		switch (update_head->type) {
		case UPDATE_APPEND:
			hostlist_index(update_head->v.hlist);
			hostlist_concat(hostlist, update_head->v.hlist);
			hostlist_free(update_head->v.hlist);
			break;
			
		case UPDATE_REPLACE:
			hostlist_copy_stat(update_head->v.hlist);
			if (immutable_hostping_tail) {
				HOSTLIST *tmp = update_head->v.hlist;

				hp = immutable_hostping_tail->next;
				if (tmp->head)
					tmp->head->prev = immutable_hostping_tail;
				immutable_hostping_tail->next = tmp->head;
				hostlist->tail = tmp->tail;
				hostlist->count += tmp->count;
				while (hp) {
					HOSTPING *next = hp->next;
					pinger_rbt_delete(hp);
					hostping_free(hp);
					hp = next;
					hostlist->count--;
				}
				hostlist_index(tmp);
				free(tmp);
			} else {
				hostlist_unindex(hostlist);
				hostlist_free(hostlist);
				hostlist = update_head->v.hlist;
				hostlist_index(hostlist);
			}
			break;
			
		case UPDATE_DELETE:
			hp = hostping_lookup_by_name(update_head->v.name);
			if (hp) {
				hostlist_remove(hostlist, hp);
				pinger_rbt_delete(hp);
				hostping_free(hp);
			}
			free(update_head->v.name);
		}

		free(update_head);
		update_head = next;
	}
	update_tail = NULL;
	if (upd)
		save_mutable_ip_list();
	pthread_rwlock_unlock(&hostlist_rwlock);
	pthread_mutex_unlock(&update_mutex);
}	

/*
 * Check if the given host is already in the hostlist.  Look for
 * matching name or, if specified, address.
 * Return value:
 *   CHECK_HOST_NOT_FOUND  - Host not found.
 *   CHECK_HOST_FOUND      - Host found.
 *   CHECK_HOST_IMMUTABLE  - Host found and is immutable.
 */
static int
check_host(char const *name, struct sockaddr *addr, socklen_t len)
{
	HOSTPING *hp;
	UPDATE_ENTRY *uent;
	int found;

	hp = hostping_lookup_by_name(name);
	if (!hp)
		hp = hostping_lookup_by_addr(addr, len);
	if (hp) {
		if (hp->immutable)
			return CHECK_HOST_IMMUTABLE;
		found = 1;
	} else
		found = 0;
	for (uent = update_head; uent; uent = uent->next) {
		switch (uent->type) {
		case UPDATE_APPEND:
		case UPDATE_REPLACE:
			for (hp = uent->v.hlist->head; hp; hp = hp->next) {
				if (strcmp(hp->name, name) == 0) {
					found++;
					break;
				}
			}
			break;
		case UPDATE_DELETE:
			if (strcmp(uent->v.name, name) == 0)
				found--;
		}
	}
	return found > 0 ? CHECK_HOST_FOUND : CHECK_HOST_NOT_FOUND;
}

static int ping_fd;

static pthread_mutex_t sendq_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t sendq_cond = PTHREAD_COND_INITIALIZER;
static int send_p;

static unsigned long probe_num; /* Ordinal number of the current probe. */
/* Totals for the current probe: */
static unsigned xmit_total;     /* Number of requests transmitted. */
static unsigned recv_total;     /* Number of replies received. */

#define ICMP_HEADER_LEN (offsetof(struct icmp, icmp_data))
#define PING_DATALEN (64 - ICMP_HEADER_LEN)

size_t data_length = PING_DATALEN;
static unsigned char *data_buffer;

/* Ping identifier (for the icmp_id member of struct icmp) */
static int ping_ident;

/* Constants for sequence number database management */

/* Max. time in seconds after which a sequence database entry can be reused. */
#define MAX_PING_TIMEOUT 10

enum {
	MAX_SEQNO = USHRT_MAX,    /* Max. value for the sequence number */
	MOD_SEQNO = MAX_SEQNO + 1 /* Modulus for computing next sequence
				     number. */
};

/* Sequence number index entry. */
struct seqidx {
	HOSTPING *host;           /* Associated host. */
	struct timeval tv;        /* Time the echo was sent. */
	unsigned long probe_num;  /* Number of the probe within which the
				     echo was sent. */
	int ping_num;             /* Number of echo request within the probe. */
};

static struct seqidx *seqidx;     /* Sequence number database. */
static unsigned short next_seqno; /* Next sequence number. */

/* Protect simultaneous access to seqidx. */
static pthread_mutex_t seqno_mutex = PTHREAD_MUTEX_INITIALIZER;

/* Allocate and return the sequence number for the given host and
 * transmission time.
 * Return negative value if the number cannot be allocated.
 */
static int
seqno_alloc(HOSTPING *host, struct timeval *tv)
{
	int n = next_seqno;
	do {
		if (tv->tv_sec - seqidx[n].tv.tv_sec > MAX_PING_TIMEOUT) {
			memcpy(&seqidx[n].tv, tv, sizeof(*tv));
			seqidx[n].host = host;
			seqidx[n].probe_num = probe_num;
			seqidx[n].ping_num = host->xmit_count;
			next_seqno = (n + 1) % MOD_SEQNO;
			return n;
		}
		n = (n + 1) % MOD_SEQNO;
	} while (n != next_seqno);
	error("no free sequence numbers");
	return -1;
}

/* Check the validity of the echo reply SEQ.  Return 0 if the reply is
 * valid, -1 otherwise.
 * This function is called when both the seqidx array and the HOSTPING
 * structure associated with SEQ are locked.
 */
static int
check_reply(int seq)
{
	int n = seqidx[seq].ping_num;
	HOSTPING *host = seqidx[seq].host;
		
	if (seqidx[seq].probe_num != probe_num) {
		/* Case 1.
		 * A latecomer reply, which arrived after its probe round
		 * was committed (see hostping_commit).
		 */
		if (verbose > 1)
			info("%s: reply for discarded echo request #%d, "
			     "seqno %d; probe_num=%lu, current=%lu",
			     host->name, n, seq, seqidx[seq].probe_num,
			     probe_num);
	} else if (host->xmit_count == 0) {
		/* Case 2.
		 * A reply came while no echo requests were transmitted yet.
		 */
		info("%s: stray reply #%d, seqno %d; probe_num=%lu",
		     host->name, n, seq, probe_num);
	} else if (n >= 0 && n < ping_count) {
		if (n > host->xmit_count) {
			/* Case 3.
			 * Similar to 2, except that some echoes were sent.
			 */
			error("%s: phantom reply #%d, seqno %d; xmit_count=%lu",
			      host->name, n, seq, host->xmit_count);
		} else if (++host->nreply[n] > 1) {
			/* Case 4.
			 * Duplicate reply.
			 */
			host->dup_count++;
			info("%s: duplicate reply for echo #%d, seqno %d",
			     host->name, n, seq);
		} else if (host->recv_count == host->xmit_count) {
			/* Case 5.
			 * Similar to 2 and 3.
			 * Each echo request was replied to, and yet another
			 * reply arrived, which is not a duplicate.
			 */
			error("%s: unexpected reply #%d, seqno %d; "
			      "xmit_count=recv_count=%lu",
			      host->name, n, seq, host->xmit_count);
		}
		/* Case 6.
		 * This is a valid reply.
		 */
		return 0;
	} else {
		/* Case 7.
		 * A reply with impossible echo request number.
		 * This one should not happen indeed.
		 */
		error("%s: reply for unregistered echo #%d, seqno %d",
		      host->name, n, seq);
	}
	return -1;
}	

/* Given an echo sequence number, return the locked HOSTPING structure
 * associated with this echo request, or NULL if the reply is
 * invalid.
 */
static HOSTPING *
hostping_from_seqno(int seq)
{
	HOSTPING *host;

	pthread_mutex_lock(&seqno_mutex);
	host = seqidx[seq].host;
	if (host) {
		hostping_lock(host);
		if (check_reply(seq)) {
			hostping_unlock(host);
			host = NULL;
		}
	} else
		fatal("no host found for sequence number %d", seq);
	pthread_mutex_unlock(&seqno_mutex);
	return host;
}

void
p903_init(void)
{
        struct protoent *proto;
	int i;
	FILE *fp;
	HOSTPING *hp;

	/* Mark all hosts configured so far as immutable and mark the
	   end of the immutable hostlist segment. */
	for (hp = hostlist->head; hp; hp = hp->next)
		hp->immutable = 1;
	immutable_hostping_tail = hostlist->tail;

	/* Read in mutable IP list. */
	fp = fopen(MUTABLE_IP_LIST_FILE, "r");
	if (fp) {
		int rc = file_read_ip_list(fp, MUTABLE_IP_LIST_FILE);
		if (rc == CF_RET_FAIL) 
			exit(1);
		fclose(fp);
	} else if (errno != ENOENT) {
		fatal("can't open %s: %s", MUTABLE_IP_LIST_FILE,
		      strerror(errno));
		exit(1);
	}
	
	if (hostlist->count == 0) {
		info("no IP addresses configured, starting anyway");
	}

	/* Open ICMP socket */
	proto = getprotobyname("icmp");
	if (!proto) {
		fatal("no entry for icmp in the system protocol database");
		exit(1);
	}
        ping_fd = socket(AF_INET, SOCK_RAW, proto->p_proto);
        if (ping_fd < 0) {
                fatal("can't create ICMP socket: %s", strerror(errno));
		exit(1);
        }

	/* Initialize ICMP identifier. */
	ping_ident = getpid() & 0xffff;

	/* Initialize payload buffer */
	data_buffer = emalloc(data_length);
	for (i = 0; i < data_length; i++)
		data_buffer[i] = i;

	/* Initialize sequence number database. */
	seqidx = calloc(MAX_SEQNO + 1, sizeof(seqidx[0]));
	if (!seqidx)
		emalloc_die();	
}

static unsigned short
icmp_cksum(unsigned char * addr, int len)
{
	register int sum = 0;
	unsigned short answer = 0;
	unsigned short *wp;

	for (wp = (unsigned short *) addr; len > 1; wp++, len -= 2)
		sum += *wp;

	/* Take in an odd byte if present */
	if (len == 1) {
		*(unsigned char *) & answer = *(unsigned char *) wp;
		sum += answer;
	}

	sum = (sum >> 16) + (sum & 0xffff);	/* add high 16 to low 16 */
	sum += (sum >> 16);		/* add carry */
	answer = ~sum;		/* truncate to 16 bits */
	return answer;
}

static int
icmp_generic_encode(unsigned char * buffer, size_t bufsize, int type,
		    int ident, int seqno)
{
	struct icmp *icmp;
	
	if (bufsize < ICMP_MINLEN)
		return -1;
	icmp = (struct icmp *) buffer;
	icmp->icmp_type = type;
	icmp->icmp_code = 0;
	icmp->icmp_cksum = 0;
	icmp->icmp_seq = htons(seqno);
	icmp->icmp_id = htons(ident);
	
	icmp->icmp_cksum = icmp_cksum(buffer, bufsize);
	return 0;
}

static int
icmp_generic_decode(unsigned char * buffer, size_t bufsize,
		    struct ip **ipp, struct icmp ** icmpp)
{
	size_t hlen;
	unsigned short cksum;
	struct ip *ip;
	struct icmp *icmp;
	
	/* IP header */
	ip = (struct ip *) buffer;
	hlen = ip->ip_hl << 2;
	if (bufsize < hlen + ICMP_MINLEN)
		return -1;

	/* ICMP header */
	icmp = (struct icmp *) (buffer + hlen);

	/* Prepare return values */
	*ipp = ip;
	*icmpp = icmp;

	/* Recompute checksum */
	cksum = icmp->icmp_cksum;
	icmp->icmp_cksum = 0;
	icmp->icmp_cksum = icmp_cksum((unsigned char *) icmp, bufsize - hlen);
	icmp->icmp_seq = ntohs(icmp->icmp_seq);
	icmp->icmp_id = ntohs(icmp->icmp_id);

	if (icmp->icmp_cksum != cksum)
		return 1;
	
	return 0;
}

static void
send_echo(HOSTPING *host, unsigned char *ping_buffer)
{
	struct icmp *icmp = (struct icmp *) ping_buffer;
	size_t buflen;
	ssize_t n;
	int seqno;

	pthread_mutex_lock(&seqno_mutex);
	hostping_lock(host);
	gettimeofday(&host->xmit_tv, NULL);
	seqno = seqno_alloc(host, &host->xmit_tv);
	pthread_mutex_unlock(&seqno_mutex);
	
	if (seqno >= 0) {
		memcpy(icmp->icmp_data, &host->xmit_tv, sizeof(host->xmit_tv));
		if (data_buffer)
			memcpy(icmp->icmp_data + sizeof(host->xmit_tv),
			       data_buffer,
			       data_length - sizeof(host->xmit_tv));
		buflen = ICMP_HEADER_LEN + data_length;

		if (verbose > 3)
			info("sending %zu bytes to %s, icmp_seq=%d",
			     buflen, host->name,
			     seqno);
		icmp_generic_encode(ping_buffer, buflen,
				    ICMP_ECHO, ping_ident, seqno);

		n = sendto(ping_fd, (char *) ping_buffer, buflen, 0,
			   host->addr, host->addrlen);
		if (n < 0) {
			error("%s: sendto: %s", host->name, strerror(errno));
		} else {
			if (host->xmit_count == 0)
				host->start_tv = host->xmit_tv;
			host->xmit_count++;
			xmit_total++;
			if (n != buflen)
				error("ping: wrote %s %zu chars, ret=%ld\n",
				      host->name, buflen, n);
		}
	} else {
		error("can't allocate sequence number for %s", host->name);
	}
	hostping_unlock(host);		
}

static void hostping_commit(HOSTPING *host);

void *
p903_sender(void *p)
{
	unsigned char *ping_buffer;
	struct pollfd pfd;
	int n;
	unsigned send_count = 0;
	double d;
	struct timespec delay;
	
	pfd.fd = ping_fd;
	pfd.events = POLLOUT;
	
	ping_buffer = emalloc(sizeof(struct icmp) + data_length);

	d = (double) ping_interval / hostlist->count;
	delay.tv_sec = (int) d;
	delay.tv_nsec = (d - delay.tv_sec) * NANOSEC_IN_SEC;

	pthread_mutex_lock(&sendq_mutex);
	while (1) {
		struct timespec ts;
		HOSTPING *hp;
		
		if (!send_p) {
			while (!send_p)
				pthread_cond_wait(&sendq_cond, &sendq_mutex);
			send_count = 0;
		}
		clock_gettime(CLOCK_MONOTONIC, &ts);
		FOR_EACH_HOSTPING(hp) {
			struct timespec nts;
			clock_gettime(CLOCK_MONOTONIC, &nts);
			timespec_incr(&nts, &delay);
			n = poll(&pfd, 1, -1);
			if (n == 1) {
				send_echo(hp, ping_buffer);
			} else if (n == -1) {
				fatal("poll: %s", strerror(errno));
				exit(1);
			}
			if (hp->next)
				clock_nanosleep(CLOCK_MONOTONIC,
						TIMER_ABSTIME, &nts,
						NULL);
			
		}
		send_count++;
		if (send_count == ping_count)
			send_p = 0;
		else {
			ts.tv_sec += ping_interval;
			clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &ts,
					NULL);
		}
	}
	pthread_mutex_unlock(&sendq_mutex);
	return NULL;
}

static void
log_echo(struct sockaddr *addr,	socklen_t addrlen,
	 struct icmp *icmp, struct ip *ip,
	 size_t datalen, double rtt)
{
	char hbuf[NI_MAXHOST];

	datalen -= ip->ip_hl << 2;
	if (getnameinfo(addr, addrlen, hbuf, sizeof(hbuf),
			NULL, 0, NI_NUMERICHOST)) {
		strcpy(hbuf, "UNKNOWN");
	}
	
	info("%zu bytes from %s: icmp_seq=%u ttl=%d time=%.3f ms", datalen,
	     hbuf, icmp->icmp_seq, ip->ip_ttl, rtt);
}

void *
p903_receiver(void *p)
{
	size_t buflen = sizeof(struct icmp) + data_length;
	unsigned char *ping_buffer;

	ping_buffer = emalloc(buflen);
	
	while (1) {
		struct sockaddr_storage addr;
		socklen_t addrlen = sizeof(addr);
		ssize_t n;
		int rc;
		struct icmp *icmp;
		struct ip *ip;
		HOSTPING *host;

		n = recvfrom(ping_fd, ping_buffer, buflen, 0,
			     (struct sockaddr *) &addr, &addrlen);
		if (n < 0) {
			error("recvfrom: %s", strerror(errno));
			continue;
		}

		rc = icmp_generic_decode(ping_buffer, n, &ip, &icmp);
		if (rc < 0) {
			char hbuf[NI_MAXHOST];
			if (getnameinfo((struct sockaddr *) &addr,
					addrlen,
					hbuf, sizeof(hbuf),
					NULL, 0, NI_NUMERICHOST)) {
				error("packet too short (%zu bytes)", n);
			} else {
				error("packet too short (%zu bytes), from %s",
				      n, hbuf);
			}
			continue;
		}

		if (icmp->icmp_type != ICMP_ECHOREPLY
		    || icmp->icmp_id != ping_ident)
			continue;

		recv_total++;

		if (rc) {
			char hbuf[NI_MAXHOST];
			if (getnameinfo((struct sockaddr *) &addr,
					addrlen,
					hbuf, sizeof(hbuf),
					NULL, 0, NI_NUMERICHOST)) {
				error("checksum mismatch");
			} else {
				error("checksum mismatch from %s", hbuf);
			}
			continue;
		}

		host = hostping_from_seqno(icmp->icmp_seq);
		if (host) {
			struct timeval tv_orig, tv_diff, *tp;
			double rtt;

			gettimeofday(&host->recv_tv, NULL);
			
			tp = (struct timeval *) icmp->icmp_data;

			/* Avoid unaligned data: */
			memcpy(&tv_orig, tp, sizeof (tv_orig));
			timersub(&host->recv_tv, &tv_orig, &tv_diff);

			rtt = timeval_to_double(&tv_diff) * 1000.0;
			host->tsum += rtt;
			host->tsumsq += rtt * rtt;
			if (rtt < host->tmin)
				host->tmin = rtt;
			if (rtt > host->tmax)
				host->tmax = rtt;
			
			host->recv_count++;

			if (verbose > 2)
				log_echo((struct sockaddr *)&addr, addrlen,
					 icmp, ip, n, rtt);
			if (host->recv_count == ping_count)
				hostping_commit(host);
			hostping_unlock(host);
		}
	}
}

static int
create_dirs(char const *filename)
{
	char *namebuf;
	char *p;
	char *endp;
	struct stat st;
	size_t len;
	
	p = strrchr(filename, '/');
	if (!p)
		return -1;
	len = p - filename;
	namebuf = malloc(len + 1);
	if (!namebuf)
		return -1;
	memcpy(namebuf, filename, len);
	namebuf[len] = 0;
	endp = namebuf + len;
	
	while ((p = strrchr(namebuf, '/')) != NULL && p > namebuf) {
		*p = 0;
		if (stat(namebuf, &st)) {
			if (errno == ENOENT)
				continue;
			error("can't stat %s: %s", namebuf, strerror(errno));
			goto err;
		}
		if (S_ISDIR(st.st_mode)) {
			break;
		} else {
			error("file name component %s is not a directory",
			      namebuf);
			goto err;
		}
	}

	while (p < endp) {
		*p = '/';
		if (mkdir(namebuf, 0755)) {
			error("can't create directory %s: %s",
			      namebuf, strerror(errno));
			goto err;
		}
		p = p + strlen(p);
	}
	return 0;
err:
	free(namebuf);
	return -1;
}
	
static void
save_mutable_ip_list(void)
{
	FILE *fp;
	HOSTPING *hp;
	
	info("saving mutable IP address list");
	fp = fopen(MUTABLE_IP_LIST_FILE, "w");
	if (!fp) {
		if (errno == ENOENT) {
			create_dirs(MUTABLE_IP_LIST_FILE);
			fp = fopen(MUTABLE_IP_LIST_FILE, "w");
		}
		if (!fp) {
			fatal("can't open %s for writing: %s",
			      MUTABLE_IP_LIST_FILE,
			      strerror(errno));
			return;
		}
	}
	FOR_EACH_MUTABLE_HOSTPING(hp) {
		fprintf(fp, "%s\n", hp->name);
	}
	fclose(fp);
}

void *
p903_scheduler(void *p)
{
	while (1) {
		HOSTPING *hp;

		pthread_mutex_lock(&sendq_mutex);
		/* Reset all statistics */
		FOR_EACH_HOSTPING(hp) {
			hostping_reset(hp);
		}
		/* Commit updates */
		p903_update_commit();

		probe_num++;
		send_p = 1;
		pthread_cond_broadcast(&sendq_cond);
		pthread_mutex_unlock(&sendq_mutex);
		
		sleep(probe_interval);
		if (verbose)
			info("total sent=%u, received=%u",
			     xmit_total, recv_total);
		xmit_total = recv_total = 0;
	}
}