/*-
 * Copyright (c) 2006 Verdens Gang AS
 * Copyright (c) 2006-2008 Linpro AS
 * All rights reserved.
 *
 * Author: Poul-Henning Kamp <phk@phk.freebsd.dk>
 *
 * 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 THE 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_hash.c 3732 2009-02-11 07:13:12Z tfheen $
 *
 * This is the central hash-table code, it relies on a chosen hash
 * implementation only for the actual hashing, all the housekeeping
 * happens here.
 *
 * We have two kinds of structures, objecthead and object.  An objecthead
 * corresponds to a given (Host:, URL) tupple, and the objects hung from
 * the objecthead may represent various variations (ie: Vary: header,
 * different TTL etc) instances of that web-entity.
 *
 * Each objecthead has a mutex which locks both its own fields, the
 * list of objects and fields in the objects.
 *
 * The hash implementation must supply a reference count facility on
 * the objecthead, and return with a reference held after a lookup.
 *
 * Lookups in the hash implementation returns with a ref held and each
 * object hung from the objhead holds a ref as well.
 *
 * Objects have refcounts which are locked by the objecthead mutex.
 *
 * New objects are always marked busy, and they can go from busy to
 * not busy only once.
 */

#include "config.h"

#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include <string.h>
#include <limits.h>
#include <sys/types.h>
#include <fcntl.h>

#include "shmlog.h"
#include "cache.h"
#include "stevedore.h"
#include "hash_slinger.h"
#include "vsha256.h"

#include <sys/time.h>
static const struct hash_slinger *hash;
unsigned	save_hash;

double
HSH_Grace(double g)
{
	if (isnan(g))
		return (double)(params->default_grace);
	return (g);
}

/* Precreate an objhead and object for later use */
void
HSH_Prealloc(struct sess *sp)
{
	struct worker *w;
	struct objhead *oh;
	struct object *o;
	struct storage *st;

	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
	CHECK_OBJ_NOTNULL(sp->wrk, WORKER_MAGIC);
	w = sp->wrk;

	if (w->nobjhead == NULL) {
		ALLOC_OBJ(oh, OBJHEAD_MAGIC);
		XXXAN(oh);
		oh->refcnt = 1;
		VTAILQ_INIT(&oh->objects);
		VTAILQ_INIT(&oh->waitinglist);
		Lck_New(&oh->mtx);
		w->nobjhead = oh;
		VSL_stats->n_objecthead++;
	} else
		CHECK_OBJ_NOTNULL(w->nobjhead, OBJHEAD_MAGIC);

	if (w->nobj == NULL) {
		st = STV_alloc(sp, params->obj_workspace);
		XXXAN(st);
		assert(st->space > sizeof *w->nobj);
		o = (void *)st->ptr; /* XXX: align ? */
		st->len = sizeof *o;
		memset(o, 0, sizeof *o);
		o->objstore = st;
		WS_Init(o->ws_o, "obj",
		    st->ptr + st->len, st->space - st->len);
		st->len = st->space;
		WS_Assert(o->ws_o);
		http_Setup(o->http, o->ws_o);
		o->magic = OBJECT_MAGIC;
		o->http->magic = HTTP_MAGIC;
		o->busy_sp = sp;
		o->refcnt = 1;
		o->grace = NAN;
		o->entered = NAN;
		VTAILQ_INIT(&o->store);
		VTAILQ_INIT(&o->esibits);
		w->nobj = o;
		VSL_stats->n_object++;

	} else
		CHECK_OBJ_NOTNULL(w->nobj, OBJECT_MAGIC);
}

void
HSH_Freestore(struct object *o)
{
	struct storage *st, *stn;

	VTAILQ_FOREACH_SAFE(st, &o->store, list, stn) {
		CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
		VTAILQ_REMOVE(&o->store, st, list);
		STV_free(st);
	}
}

void
HSH_Copy(const struct sess *sp, struct objhead *oh)
{
	unsigned u, v;
	char *b;

	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
	CHECK_OBJ_NOTNULL(oh, OBJHEAD_MAGIC);
	if (!save_hash)
		return;

	oh->hash = malloc(sp->lhashptr);
	XXXAN(oh->hash);
	oh->hashlen = sp->lhashptr;
	b = oh->hash;
	for (u = 0; u < sp->ihashptr; u += 2) {
		v = pdiff(sp->hashptr[u], sp->hashptr[u + 1]);
		memcpy(b, sp->hashptr[u], v);
		b += v;
		*b++ = '#';
	}
	*b++ = '\0';
	assert(b <= oh->hash + oh->hashlen);
}

void
HSH_Prepare(struct sess *sp, unsigned nhashcount)
{
	char *p;

	SHA256_Init(sp->wrk->sha256ctx);
	if (!save_hash)
		return;

	/* Allocate the pointers we need, align properly. */
	sp->lhashptr = 1;       /* space for NUL */
	sp->ihashptr = 0;
	sp->nhashptr = nhashcount * 2;
	/* WS_Malloc gives aligned allocation */
	p = WS_Malloc(sp->http->ws, sizeof(const char *) * (sp->nhashptr));
	XXXAN(p);
	sp->hashptr = (void*)p;
}

void
HSH_AddString(struct sess *sp, const char *str)
{
	int l;

	if (str == NULL)
		str = "";
	l = strlen(str);

	SHA256_Update(sp->wrk->sha256ctx, str, l);
	SHA256_Update(sp->wrk->sha256ctx, "#", 1);

	if (params->log_hash)
		WSP(sp, SLT_Hash, "%s", str);

	if (!save_hash)
		return;

	/*
	* XXX: handle this by bouncing sp->vcl->nhashcount when it fails
	* XXX: and dispose of this request either by reallocating the
	* XXX: hashptr (if possible) or restarting/error the request
	*/
	xxxassert(sp->ihashptr < sp->nhashptr);

	sp->hashptr[sp->ihashptr] = str;
	sp->hashptr[sp->ihashptr + 1] = str + l;
	sp->ihashptr += 2;
	sp->lhashptr += l + 1;
}

struct object *
HSH_Lookup(struct sess *sp)
{
	struct worker *w;
	struct objhead *oh;
	struct object *o, *busy_o, *grace_o, *sup_o;

	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
	CHECK_OBJ_NOTNULL(sp->wrk, WORKER_MAGIC);
	CHECK_OBJ_NOTNULL(sp->http, HTTP_MAGIC);
	AN(hash);
	w = sp->wrk;

	HSH_Prealloc(sp);
	SHA256_Final(sp->wrk->nobjhead->digest, sp->wrk->sha256ctx);
	
	if (sp->objhead != NULL) {
		CHECK_OBJ_NOTNULL(sp->objhead, OBJHEAD_MAGIC);
		oh = sp->objhead;
		sp->objhead = NULL;
		CHECK_OBJ_NOTNULL(oh, OBJHEAD_MAGIC);
		Lck_Lock(&oh->mtx);
	} else {
		AN(w->nobjhead);
		oh = hash->lookup(sp, w->nobjhead);
		CHECK_OBJ_NOTNULL(oh, OBJHEAD_MAGIC);
		if (oh == w->nobjhead)
			w->nobjhead = NULL;
		Lck_Lock(&oh->mtx);
	}
	AZ(sp->objhead);

	busy_o = NULL;
	grace_o = NULL;
	VTAILQ_FOREACH(o, &oh->objects, list) {
		if (o->busy_sp) {
			assert(oh->nbusy > 0);
			busy_o = o;
			continue;
		}
		if (!o->cacheable)
			continue;
		if (o->ttl == 0)
			continue;
		if (BAN_CheckObject(o, sp)) 
			continue;
		if (o->vary != NULL && !VRY_Match(sp, o->vary))
			continue;

		/* If still valid, use it */
		if (o->ttl >= sp->t_req)
			break;

		/* Remember any matching objects inside their grace period */
		if (o->ttl + HSH_Grace(o->grace) >= sp->t_req)
			grace_o = o;
	}

	/*
	 * If we have found an object we like _and_ a grace object along the
	 * way, we know that we wont't need the grace object any more
	 */
	sup_o = NULL;
	if (o != NULL && grace_o != NULL && o != grace_o) {
		sup_o = grace_o;
		grace_o == NULL;

		/* 
		 * as we have the oh locked already, avoid taking it another
		 * time in exp_timer->HSH_Deref
		 */
		VTAILQ_REMOVE(&oh->objects, sup_o, list);
		sup_o->objhead = NULL;
		/* do the rest on way out outside the oh lock */
	}

	/*
	 * If we have a object in grace
	 * - and req.grace is also satisfied
	 * - and we're currently fetching the object
	 *    or this is a restart
	 * return the grace object
	 */

	if (o == NULL && grace_o != NULL &&
	    (sp->restarts || (grace_o->child != NULL)) &&
#if 0
	    grace_o->child != NULL &&
#endif
	    grace_o->ttl + HSH_Grace(sp->grace) >= sp->t_req)
		o = grace_o;

	if (o != NULL) {
		/* We found an object we like */
		o->refcnt++;
		if (o->hits < INT_MAX)
			o->hits++;

		Lck_Unlock(&oh->mtx);
		(void)hash->deref(oh);
		goto out;
	}

	if (busy_o != NULL) {
		/* There are one or more busy objects, wait for them */
		if (sp->esis == 0) {
			/* 
			 * restarts have waited already, so they go to the front
			 * of the waitinglist
			 */
			if (sp->restarts)
				VTAILQ_INSERT_HEAD(&oh->waitinglist, sp, list);
			else
				VTAILQ_INSERT_TAIL(&oh->waitinglist, sp, list);
		}
		if (params->diag_bitmap & 0x20)
			WSP(sp, SLT_Debug,
			"@%f %p on waiting list %p <%s>", TIM_mono(), sp, oh, oh->hash);
		sp->objhead = oh;
		Lck_Unlock(&oh->mtx);

		o = NULL;
		goto out;
	}

	/* Insert (precreated) object in objecthead */
	o = w->nobj;
	w->nobj = NULL;
	o->objhead = oh;
	/* XXX: Should this not be ..._HEAD now ? */
	VTAILQ_INSERT_TAIL(&oh->objects, o, list);
	oh->nbusy++;
	/* NB: do not deref objhead the new object inherits our reference */
	if (grace_o != NULL) {
		grace_o->child = o;
		o->parent = grace_o;
		grace_o->refcnt++;
	}
	Lck_Unlock(&oh->mtx);
	/*
	 * XXX: This may be too early, relative to pass objects.
	 * XXX: possibly move to when we commit to have it in the cache.
	 */
	BAN_NewObj(o);
	/* return with looking after the sup_o we removed outside oh->mtx */

  out:
	if (sup_o) {
		/* this finalizes sup_o->objhead = NULL, but outside the oh lock */
		(void) hash->deref(oh);

		/*
		 * exp_timer will pick it up and call HSH_Deref, but because we have
		 * unset sup_o->objhead above, we won't need the oh lock again
		 */
		VSL_stats->n_superseded++;		
		VSL_stats->n_expired--;
		sup_o->grace = 0;
		EXP_Rearm(sup_o);
	}

	return (o);
}

static void
hsh_rush_n(struct objhead *oh, unsigned n)
{
	struct sess *sp;

	while (n--) {
		sp = VTAILQ_FIRST(&oh->waitinglist);
		if (sp == NULL)
			return;
		VTAILQ_REMOVE(&oh->waitinglist, sp, list);
		DSL(0x20, SLT_Debug, 0, "@%f %p off waiting list %p <%s>", TIM_mono(), sp, oh, oh->hash);
		
		sp->restarts++;
		WRK_QueueSession(sp);
	}
}

static inline void
hsh_rush(struct objhead *oh)
{
	Lck_AssertHeld(&oh->mtx);
	hsh_rush_n(oh, params->rush_exponent);
}

void
HSH_Drop(struct sess *sp)
{
	struct object *o;

	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
	o = sp->obj;
	CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
	assert(o->busy_sp);
	assert(o->refcnt > 0);
	o->ttl = 0;
	o->cacheable = 0;
	HSH_Unbusy(sp);
	/* no need to rush because HSH_Unbusy will rush */
	HSH_Deref(&sp->obj, DONT_RUSH);
}

void
HSH_Unbusy(const struct sess *sp)
{
	struct object *o;
	struct objhead *oh;
	struct object *parent;

	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
	o = sp->obj;
	CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
	assert(o->busy_sp);
	assert(o->refcnt > 0);
	if (o->ws_o->overflow)
		VSL_stats->n_objoverflow++;
	if (params->diag_bitmap & 0x40)
		WSP(sp, SLT_Debug,
		    "Object %u workspace free %u", o->xid, WS_Free(o->ws_o));

	oh = o->objhead;
	if (oh != NULL) {
		CHECK_OBJ(oh, OBJHEAD_MAGIC);
		Lck_Lock(&oh->mtx);
		oh->nbusy--;
	}
	o->busy_sp = NULL;
	if (oh != NULL) {
		if (params->diag_bitmap & 0x40)
			WSP(sp, SLT_Debug,
			"@%f HSH_Unbusy Calling hsh_rush on oh %p o %p o->xid %u", 
			TIM_mono(), oh, o, o->xid);
		hsh_rush(oh);
	}

	/*
	 * here, parent is a former grace object: We are returning from getting
	 * a new copy of an object we had an old copy of in the cache. In
	 * HSH_Lookup, we added o->parent as a reference to the grace object to
	 * keep it available while we get the fresh copy, so we need to drop the
	 * reference now
	 */

	parent = o->parent;
	o->parent = NULL;
	if (parent != NULL)
		parent->child = NULL;
	if (oh != NULL)
		Lck_Unlock(&oh->mtx);
	if (parent != NULL) {
		if (params->diag_bitmap & 0x40)
			WSP(sp, SLT_Debug,
			"@%f HSH_Unbusy Calling HSH_Deref on (o %p xid %u)->parent = (%p xid %u)", 
			TIM_mono(),
			o, o->xid, 
			parent, parent->xid);
		/*
		 * No need to rush here if we've alrady rushed above
		 */
		HSH_Deref(&parent, (oh == NULL) ? DO_RUSH : DONT_RUSH);
	}
}

void
HSH_Ref(struct object *o)
{
	struct objhead *oh;

	CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
	oh = o->objhead;
	CHECK_OBJ_NOTNULL(oh, OBJHEAD_MAGIC);
	Lck_Lock(&oh->mtx);
	assert(o->refcnt > 0);
	o->refcnt++;
	Lck_Unlock(&oh->mtx);
}

void
HSH_Deref(struct object **oo, int do_rush)
{
	struct object *o;
	struct objhead *oh;
	unsigned r;

	AN(oo);
	o = *oo;
	*oo = NULL;
	CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
	oh = o->objhead;
	if (oh != NULL) {
		CHECK_OBJ(oh, OBJHEAD_MAGIC);

		/* drop ref on object */
		Lck_Lock(&oh->mtx);
	}
	assert(o->refcnt > 0);
	r = --o->refcnt;

	/*
	 * we should only rush for this object's head, if
	 * - this was a real dereference from a delivery (do_rush is set)
	 * - we are going to keep the object (r > 0)
	 * - we don't have busy objects on the oh (if we do, we'll rush on HSH_Unbusy)
	 *
	 * Note: expiry methods should call with DONT_RUSH, so
	 *       the test for r is syntactic sugar
	 */

	if ((oh != NULL) &&
	    (do_rush == DO_RUSH) && r &&
	    (oh->nbusy == 0)) {
		DSL(0x40, SLT_Debug, 0, "@%f HSH_Deref Calling hsh_rush on oh %p o %p o->xid %u",
		    TIM_mono(),
		    oh, o, o->xid);
		hsh_rush(oh);
	}
	if (oh != NULL) {
		if (!r)
			VTAILQ_REMOVE(&oh->objects, o, list);
		Lck_Unlock(&oh->mtx);
	}

	/* If still referenced, done */
	if (r != 0)
		return;

	BAN_DestroyObj(o);
	DSL(0x40, SLT_Debug, 0, "Object %u workspace min free %u",
	    o->xid, WS_Free(o->ws_o));

	if (o->vary != NULL)
		free(o->vary);

	ESI_Destroy(o);
	HSH_Freestore(o);
	STV_free(o->objstore);
	VSL_stats->n_object--;

	if (oh == NULL)
		return;

	/* Drop our ref on the objhead */
	if (hash->deref(oh))
		return;

	/* rush anyone still on the waitinglist before destroying the oh */
	hsh_rush_n(oh, UINT_MAX);

	assert(VTAILQ_EMPTY(&oh->objects));
	assert(oh->nbusy == 0);
	Lck_Delete(&oh->mtx);
	VSL_stats->n_objecthead--;
	free(oh->hash);
	FREE_OBJ(oh);
}

void
HSH_Init(void)
{

	save_hash = params->save_hash;
	hash = heritage.hash;
	if (hash->start != NULL)
		hash->start();
}
