/*-
 * Copyright (c) 2009-2012 UPLEX - Nils Goroll Systemoptimierung
 * All rights reserved.
 *
 * Authors: Nils Goroll <nils.goroll@uplex.de>
 *	    Geoffrey Simmons <geoff.simmons@uplex.de>
 *
 * 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$
 *
 * VSLP*) Director - Varnish StateLess Persistence Director
 *
 * The purpose of this director is to provide stateless persistence (or
 * persistency if you prefer) for backend requests based upon this simple
 * concept:
 *
 * - Generate a load balancing key, which will be used to select the backend
 *
 *   The key values should be as uniformly distributed as possible.
 *
 *   For all requests which need to hit the same backend server, the same
 *   key must be generated.
 *
 *   For strings, a hash function can be used to generate the key.
 *
 * - Select the preferred backend server using an implementation of consistent
 *   hashing (cf. Karger et al, references below), which ensures that the same
 *   hosts are always chosen for every key (for instance hash of incoming URL)
 *   in the same order (i.e. if the preferred host is down, then alternative
 *   hosts are always chosen in a fixed order).
 *
 *	- Initialize a circular data structure consisting of the hash values of
 *	  "hostname%d" for each host and for a running number from 1 to n (n is
 *	  the number of "replicas").
 *
 *	- For the load balancing key, find the smallest hash value in the circle
 *	  that is larger than the key (searching clockwise and wrapping around
 *	  as necessary).
 *
 *	- If the host thus selected is down, choose alternative hosts by
 *	  continuing to search clockwise in the circle.
 *
 *	On consistent hashing see:
 *	    http://www8.org/w8-papers/2a-webserver/caching/paper2.html
 *	    http://www.audioscrobbler.net/development/ketama/
 *	    svn://svn.audioscrobbler.net/misc/ketama
 *	    http://en.wikipedia.org/wiki/Consistent_hashing
 *
 * This technique allowes to implement persistence without keeping any state,
 * and, in particular, without the need to synchronize state between nodes of a
 * cluster of Varnish servers.
 *
 * Still, all members of a cluster will, if otherwise configured equally,
 * implement the same persistence, in other words, requests sharing a common
 * persistence criterium will be balanced onto the same backend server no matter
 * which Varnish server handles the request.
 *
 * The drawbacks are:
 *
 * - the distribution of requests depends on the number of requests per key and
 *   the uniformity of the distribution of key values. In short, this technique
 *   will generally lead to less good load balancing compared to stateful
 *   techniques.
 *
 * - When a backend server becomes unavailable, every persistence technique has
 *   to reselect a new backend server, but this technique will also switch back
 *   to the preferred server once it becomes healthy again, so the persistence
 *   is generally less stable compared to stateful techniques (which would
 *   continue to use a selected server for as long as possible (or dictated by a
 *   TTL)).
 *
 * *) The name VSLP was chosen to accompany VTLA as another VETR (Varnish
 *    Exception To the Rule) ;-) http://varnish-cache.org/wiki/VTLA
 */

#include "config.h"

#include <stdint.h>
#include <arpa/inet.h>

#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <stdio.h>
#include <math.h>

#include "shmlog.h"
#include "cache.h"
#include "cache_backend.h"
#include "cache_dir_vslp.h"
#include "vrt.h"
#include "vsha256.h"

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

enum vdi_vslp_digest_t {
	DEFAULT = 0,
	CRC32,
	SHA256,
	PLAIN,
	RS
};

struct vdi_vslp_host {
	struct backend			*backend;

	/*
	 * canonical point in the consistent hashing ring which will select this
	 * backend
	 */
	uint32_t			canon_point;
};

/*
 * Contents of the circle used by the VSLP director for consistent hashing.
 *
 * point: point in the circle (hash of host name concatenated with replica
 *	  number)
 *
 * host: index into vdi_vslp->hosts
 */
struct vdi_vslp_hostnode {
	uint32_t		point;
	unsigned int		host;
};

/* Maintains state about hosts chosen by the VSLP director during a session.
 * index - current index in the hash circle (clockwise search)
 * chosen - array of hosts indexed as in vdi_vslp->hosts, chosen[i] is true
 *	    if hosts[i] has been chosen at some time during the session.
 * nchosen - number of hosts chosen during the current session
 */
struct vslp_host_choice {
	int			index;
	int			*chosen;
	int			nchosen;
};

/*
 * Cache for load balancing decision to allow probing the director from
 * VCL without actually firing of fetch / pass
 *
 * key - the key that was used for this choice
 *
 * vbe - open backend connection
 *
 * host - index into director's host list
 *
 * freshness rules:
 *
 * - key changes => clear all of the state
 *
 *   we have a new loadbalancing decision, so we need to clear the history (for
 *   instance, when hitting a different key upon restart)
 *
 * - !vbe => select the next host in the session
 *
 *   we have restarted with the same key, so we chose the next, alternative host
 *   (assuming that the restart it due to the fact that something went wrong on
 *   the previously chosen host)
 */

struct vslp_strategy_cache {
	uint32_t		key;
	struct vbe_conn		*vbe;
	struct backend		*backend;
};

/* Per session state for VSLP */

struct vdi_vslp_dss {
	unsigned			magic;
#define VSLP_DSS_MAGIC			0x778e7c60

	/* user configuration (via dir.digest and dir.key VCL variables) */
	uint32_t			persistence_key;
	enum vdi_vslp_digest_t		digest;

	/* state caches */
	struct vslp_host_choice 	choice;
	struct vslp_strategy_cache	strategy;
};

struct vdi_vslp {
	unsigned		magic;
#define VDI_VSLP_MAGIC		0xcd0d15f3 /* crc32 */
	struct director		dir;
	struct vdi_vslp_host	*hosts;
	unsigned		nhosts;

	/*
	 * percentage of requests to direct to the next server in the list of
	 * hosts to pre-warm a backup server
	 *
	 * set to 0 to disable
	 */
	double			altsrv_p;
	/* default digest, if not set in per session data */
	enum vdi_vslp_digest_t	digest;

	/*
	 * Time (in seconds) to ramp up a host after it becomes healthy.
	 * Set to 0 for no rampup (immediately use recovered hosts at full capacity)
	 */
	double			rampup;

	/*
	 * Table used for consistent hashing of hosts -- indexes are hashes of
	 * the host name concatenated with a replica number, and values are
	 * indexes into the hosts array. Conceived as a circle -- wrap around
	 * from the end back to 0.
	 */
	struct vdi_vslp_hostnode *hashcircle;
};

static inline char *
VDI_BackendName(const struct vdi_vslp_host h)
{
	struct backend *d = h.backend;
	CHECK_OBJ_NOTNULL(d, BACKEND_MAGIC);
	return(d->vcl_name);
}

static inline double
VDI_Recovered(const struct vdi_vslp_host h)
{
	struct backend *d = h.backend;
	CHECK_OBJ_NOTNULL(d, BACKEND_MAGIC);
	return(d->recovered);
}

/*--------------------------------------------------------------------
 * Maintaining Director Session State
 *--------------------------------------------------------------------
 */

/*
 * prepare the dss for fini or cleanup
 */
static inline void
vdi_vslp_dss_close(struct vdi_vslp_dss *dss, struct sess *sp)
{
	CHECK_OBJ_NOTNULL(dss, VSLP_DSS_MAGIC);
	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);

	if (dss->strategy.vbe && dss->strategy.vbe->fd) {
		/* Safety: Recycle what is there before overwriting */
		if (sp->vbe)
			VBE_RecycleFd(sp);
		sp->vbe = dss->strategy.vbe;
		VBE_RecycleFd(sp);
		AZ(sp->vbe);
	}
}

/*
 * cleanup the dss for re-use with a different key
 */
static inline void
vdi_vslp_dss_clean(struct vdi_vslp_dss *dss, struct sess *sp, struct vdi_vslp *vs)
{
	int			*chosen;
	uint32_t		persistence_key = dss->persistence_key;
	enum vdi_vslp_digest_t	digest = dss->digest;

	CHECK_OBJ_NOTNULL(dss, VSLP_DSS_MAGIC);
	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
	CHECK_OBJ_NOTNULL(vs, VDI_VSLP_MAGIC);

	/*
	 * make sure the dss is from our ws
	 */
	assert(((void *)sp->http->ws->s <= (void *)dss) &&
	    ((void *)dss <= (void *)sp->http->ws->e));

	chosen = NULL;

	vdi_vslp_dss_close(dss, sp);
	if (dss->choice.chosen) {
		assert(((void *)sp->http->ws->s <= (void *)dss->choice.chosen) &&
		    ((void *)dss->choice.chosen <= (void *)sp->http->ws->e));

		memset(dss->choice.chosen, 0, (vs->nhosts * sizeof (int)));
		chosen = dss->choice.chosen;
	}
	memset(dss, 0, sizeof (*dss));

	dss->magic = VSLP_DSS_MAGIC;
	dss->persistence_key = persistence_key;
	dss->digest = digest;
	dss->choice.chosen = chosen;
}

static void *
vdi_vslp_dss_ini(struct director_sess *ds, struct sess *sp)
{
	struct vdi_vslp_dss	*dss;
	struct vdi_vslp		*vs;

	(void) ds;
	CAST_OBJ_NOTNULL(vs, sp->director->priv, VDI_VSLP_MAGIC);

	dss = WS_Calloc(sp->http->ws, 1, sizeof(*dss));
	AN(dss);

	/* see also vdi_vslp_dss_clean */
	dss->choice.chosen = WS_Calloc(sp->http->ws, vs->nhosts, sizeof(int));
	AN(dss->choice.chosen);

	dss->magic = VSLP_DSS_MAGIC;

	return (dss);
}

/*
 * cleanup of the dss when a different director needs the
 * director session state
 */

static void
vdi_vslp_dss_fini(void *dss_priv, struct sess *sp)
{
	struct vdi_vslp_dss	*dss;

	CAST_OBJ_NOTNULL(dss, dss_priv, VSLP_DSS_MAGIC);

	vdi_vslp_dss_close(dss, sp);
}

/*--------------------------------------------------------------------
 * Consistent hashing
 *--------------------------------------------------------------------
 */

typedef int (*compar)( const void*, const void* );

static int
hostnode_compare(struct vdi_vslp_hostnode *a, struct vdi_vslp_hostnode *b)
{
    return (a->point == b->point) ? 0 : ((a->point > b->point) ? 1 : -1);
}

// Number of replicas of each host to hash into the consistent hash circle
// TODO: Set as a parameter in VCL, with a reasonable default
#define REPLICAS 150

// Debug flag for the director (unused in documented diag_bitmap flags)
#define DIAG_DIRECTOR 0x10000000

/*
 * Choose a host using consistent hashing
 * vs - pointer to VSLP director
 * key - hash key (only used for inital selection)
 * dss- the current director session
 * returns an index into vs->hosts for the next chosen host in this session,
 *	   or -1 if all available hosts have been exhausted
 */
static inline int
vdi_vslp_choose(struct sess *sp, struct vdi_vslp *vs, uint32_t key,
		struct vdi_vslp_dss *dss)
{
    const int n = vs->nhosts * REPLICAS;
    int choice, i;
    struct vdi_vslp_hostnode *circle = vs->hashcircle;

    /*
     * Fail if all of the available hosts have already been chosen
     */
    if (dss->choice.nchosen == vs->nhosts)
	return -1;

    /*
     * If no host has been chosen so far in this session, then choose the first
     * host on the circle whose key is greater than or equal to the given key,
     * searching clockwise.
     */
    if (dss->choice.nchosen == 0) {
	if (key < circle[0].point) {
	    dss->choice.index = 0;
	    dss->choice.chosen[circle[0].host] = 1;
	    dss->choice.nchosen = 1;
	    DSL(DIAG_DIRECTOR, SLT_Debug, sp->id,
		"chose preferred %s (point = %u) for key %u",
		VDI_BackendName(vs->hosts[circle[0].host]), circle[0].point,
		key);
	    return circle[0].host;
	}

	int low = 0, high = n;
	choice = -1;
	do {
	    i = (high + low) >> 1;
	    if (circle[i].point == key)
		choice = i;
	    else if (i == n - 1)
		choice = n - 1;
	    else if (circle[i].point < key && circle[i+1].point >= key)
		choice = i + 1;
	    else if (circle[i].point > key)
                if (i == 0)
                    choice = 0;
		else
                    high = i;
	    else
		low = i;
	} while (choice == -1);

	dss->choice.index = choice;
	dss->choice.chosen[circle[choice].host] = 1;
	dss->choice.nchosen = 1;

	DSL(DIAG_DIRECTOR, SLT_Debug, sp->id,
	    "chose preferred %s (point = %u) for key %u",
	    VDI_BackendName(vs->hosts[circle[choice].host]),
	    circle[choice].point, key);
	return circle[choice].host;
    }

    /*
     * If a host has already been chosen during this session, choose the next
     * alternate -- the next host on the circle not already chosen (searching
     * clockwise).
     *
     * note on the indexes used here:
     *
     * dss->choice.chosen[] has (vdi_vslp *)vs->nhosts     entries
     * circle[]		    has n = vs->nhosts * REPLICAS  entries
     */
    choice = dss->choice.index;
    do {
	if (choice == n-1)
	    choice = 0;
	else choice++;
    } while (dss->choice.chosen[circle[choice].host]);

    dss->choice.index = choice;
    dss->choice.chosen[circle[choice].host] = 1;
    dss->choice.nchosen++;

    DSL(DIAG_DIRECTOR, SLT_Debug, sp->id,
	"chose alternate %d %s (point = %u) for key %u",
	dss->choice.nchosen, VDI_BackendName(vs->hosts[circle[choice].host]),
	circle[choice].point, key);
    return circle[choice].host;
}

/*--------------------------------------------------------------------
 * Rampup
 *--------------------------------------------------------------------
 */

/*
 * For a chosen host within the rampup interval, decide whether to use
 * it or an alternative. The probabilty of using it increases linearly
 * with time since recovery.
 * sp - current session
 * vs - VSLP director
 * host - the chosen host, already determined to be within rampup
 * ds - the director session
 * t_since_recovery - time since host recovered (from TIM_mono())
 * returns an index into vs->hosts (either the preferred host or an alternate)
 */
static inline int
vdi_vslp_rampup(struct sess *sp, struct vdi_vslp *vs, int host,
		struct vdi_vslp_dss *dss, double t_since_recovery) {
	double	r;
	int alternate;

	VSL(SLT_Director, sp->id, "host %s in rampup",
	    VDI_BackendName(vs->hosts[host]));

	r = random() / 2147483648.0;	/* 2^31 */
	assert(r >= 0.0 && r < 1.0);

	if (r * vs->rampup <= t_since_recovery)
	    return host;

	/*
	 * Pick another host
	 * Choose the next host on the circle whose last recovery time is
	 * earlier than the preferred host's recovery time.
	 */
	while ((alternate = vdi_vslp_choose(sp, vs, 0, dss)) != -1)
		if (vs->hosts[alternate].backend->healthy
		    && (VDI_Recovered(vs->hosts[alternate]) == -1
			|| VDI_Recovered(vs->hosts[alternate])
			   < VDI_Recovered(vs->hosts[host])))
		break;

	if (alternate == -1) {
	    VSL(SLT_Director, sp->id, "no alternate host for rampup found");
	    return host;
	}

	return alternate;
}

/*--------------------------------------------------------------------
 * Actual backend selection
 *--------------------------------------------------------------------
 */

static struct vdi_vslp_dss *
vdi_vslp_try_healthy(struct sess *sp, struct vdi_vslp *vs, unsigned int i,
    struct vdi_vslp_dss *dss, const char *type_str)
{
	struct backend		*d;
	
	d = vs->hosts[i].backend;
	if (d->healthy) {
		VSL(SLT_Director, sp->id, "name %s vcl_name %s "
		    "%s %s healthy",
		    vs->dir.name,
		    vs->dir.vcl_name,
		    type_str,
		    VDI_BackendName(vs->hosts[i]));

		dss->strategy.backend = d;
		return (dss);
	}
	return (NULL);
}

static struct vdi_vslp_dss *
vdi_vslp_try_connect(struct sess *sp, struct vdi_vslp *vs, unsigned int i,
    struct vdi_vslp_dss *dss, const char *type_str)
{
	struct vbe_conn		*vbe;

	if (vdi_vslp_try_healthy(sp, vs, i, dss, type_str) &&
	    (vbe = VBE_GetVbe(sp, dss->strategy.backend))) {
		VSL(SLT_Director, sp->id, "name %s vcl_name %s "
		    "%s %s connected",
		    vs->dir.name,
		    vs->dir.vcl_name,
		    type_str,
		    VDI_BackendName(vs->hosts[i]));

		/* cache vbe */
		dss->strategy.vbe = vbe;
		return (dss);
	}
	return (NULL);
}

#define TRY_HEALTHY (1)
#define TRY_CONNECT (2)

#define VDI_VSLP_TRY(how, sp,vs,i,dss,s)				\
	if ((ret = ((how == TRY_CONNECT)				\
			? vdi_vslp_try_connect(sp, vs, i, dss, s)	\
	                : vdi_vslp_try_healthy(sp, vs, i, dss, s))))	\
		return (ret)

/*
 * vdi_vslp_strategy
 *
 * Go through all the whole process of backend selection, including making the
 * actual backend connection.
 *
 * Cache the selection decision and return our director session state
 *
 * Caching is important for VRT access to loadbalancer decision: We need to make
 * the actual decision before actually going into pass/fetch
 *
 * Note: We *could* cache in sp->vbe, but a little redundancy should help to
 *	 keep this clear
 *
 */

static struct vdi_vslp_dss *
vdi_vslp_strategy(struct sess *sp, int how)
{
	int			host;

	uint32_t		key;

	struct vdi_vslp		*vs;
	struct vdi_vslp_dss	*dss, *ret;

	double r;

	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
	CHECK_OBJ_NOTNULL(sp->director, DIRECTOR_MAGIC);
	CAST_OBJ_NOTNULL(vs, sp->director->priv, VDI_VSLP_MAGIC);

	dss = VDI_get_dss(sp);
	CHECK_OBJ_NOTNULL(dss, VSLP_DSS_MAGIC);

	if ((key = dss->persistence_key)) {
		key--;
	} else {
		/* No valid key. Fall back to hash of URL */
		key = crc32_l(sp->http->hd[HTTP_HDR_URL].b,
		    strlen(sp->http->hd[HTTP_HDR_URL].b));
		dss->persistence_key = key;
	}

	/*
	 * If the key has changed, start from scratch, otherwise re-use cached
	 * decision/connection
	 */

	if (dss->strategy.vbe)
		AN(dss->strategy.backend);

	if (dss->strategy.backend) {
		if (dss->strategy.key != key) {
			vdi_vslp_dss_clean(dss, sp, vs);
		} else if (how == TRY_HEALTHY) {
			/* dss->strategy.key == key */
			return (dss);
		} else {
			struct vbe_conn		*vbe;

			/* how == TRY_CONNECT */
			if (dss->strategy.vbe)
				return (dss);

			/* try to connect to the cached backend */
			if ((vbe = VBE_GetVbe(sp, dss->strategy.backend))) {
				VSL(SLT_Director, sp->id, "name %s vcl_name %s "
				    "%s %s connected_cached",
				    vs->dir.name,
				    vs->dir.vcl_name,
				    "-",
				    dss->strategy.backend->vcl_name);

				dss->strategy.vbe = vbe;
				return (dss);
			}

			dss->strategy.vbe = NULL;
			dss->strategy.backend = NULL;
		}
	}

	AZ(dss->strategy.vbe);
	AZ(dss->strategy.backend);
	dss->strategy.key = key;

	/*
	 * When getting here the second time (when dss->choice.nchosen is set),
	 * we must be handling a restart (for which we assume that it was
	 * triggered by a malfunctioning backend, which we don't want to use
	 * again), so we skip all the fancy logic and use the first backend we
	 * find
	 *
	 * note that when the cache key was changed, we loose all history
	 * (vdi_vslp_dss_clean above), so we might still be restarting when we
	 * think we aren't.
	 */

	if (dss->choice.nchosen)
		goto vslp_last;

	/*
	 * 1st step: choose preferred backend by key from consistent hashing
	 *	     ring
	 */

	host = vdi_vslp_choose(sp, vs, key, dss);

	/*
	 * Even if the primary backend is healthy, we might still use another
	 * one for two reasons:
	 *
	 * - the primary backend has just become healthy and still is in the
	 *   rampup period or
	 *
	 * - we have an alternative server probability set to warm up that
	 *   server. The alternative server is always the next server in the
	 *   ring, so it does not change when the health state changes.
	 *
	 */

	if (vs->hosts[host].backend->healthy) {
		double t_since_recovery = TIM_mono() -
		    VDI_Recovered(vs->hosts[host]);

		DSL(DIAG_DIRECTOR, SLT_Debug, sp->id,
		    "%s: t_since_recovery = %f, rampup = %f",
		    VDI_BackendName(vs->hosts[host]),
		    t_since_recovery,
		    vs->rampup);

		if ((vs->rampup > 0.0) && (t_since_recovery < vs->rampup)) {
			unsigned int rampup_chosen =
			    vdi_vslp_rampup(sp, vs, host, dss, t_since_recovery);
			VDI_VSLP_TRY(how, sp, vs, rampup_chosen, dss,
			    (rampup_chosen == host) ?
			    "preferred rampup" :
			    "alternative rampup");
		} else if (vs->altsrv_p > 0.0) {
			/* from cache_dir_random.c */
			r = random() / 2147483648.0;	/* 2^31 */
			assert(r >= 0.0 && r < 1.0);
			if (r < vs->altsrv_p) {
			  int alternate = vdi_vslp_choose(sp, vs, key, dss);
			  VDI_VSLP_TRY(how, sp, vs, alternate, dss, "altsrv_p");
			}
		}

		/*
		 * we also get here when the rampup / alternative servers were
		 * sick
		 */
		VDI_VSLP_TRY(how, sp, vs, host, dss, "preferred");
	}

	/*
	 * If the preferred backend is unavailable, we try backends in the order
	 * dictated by the consistent hashing ring. The order is inidividual per
	 * key and should be random (depending on the quality of the hash used
	 * for initialization of the hosts in the ring)
	 */

  vslp_last:
	while ((host = vdi_vslp_choose(sp, vs, key, dss)) != -1)
		if (vs->hosts[host].backend->healthy)
		    VDI_VSLP_TRY(how, sp, vs, host, dss, "last");

	VSL(SLT_Director, sp->id, "name %s vcl_name %s"
	    " No host found for key %u",
	    vs->dir.name,
	    vs->dir.vcl_name,
	    key);
	return (NULL);
}

/*--------------------------------------------------------------------
 * Standard Director Interface
 * (dss interface is located on top)
 *--------------------------------------------------------------------
 */

static struct vbe_conn *
vdi_vslp_getfd(struct sess *sp)
{
	struct vdi_vslp_dss	*dss = vdi_vslp_strategy(sp, TRY_CONNECT);
	struct vbe_conn		*vbe;

	if (dss == NULL)
		return (NULL);

	vbe = dss->strategy.vbe;
	/* remove the cached vbe because now we're actually going to use it */
	dss->strategy.vbe = NULL;

	return (vbe);
}

const static struct backend *
vdi_vslp_choice(struct sess *sp)
{
	struct vdi_vslp_dss     *dss = vdi_vslp_strategy(sp, TRY_HEALTHY);

	if (dss == NULL)
		return (NULL);

	return (dss->strategy.backend);
}


static unsigned
vdi_vslp_healthy(const struct sess *sp)
{
	struct vdi_vslp *vs;
	int i;

	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
	CHECK_OBJ_NOTNULL(sp->director, DIRECTOR_MAGIC);
	CAST_OBJ_NOTNULL(vs, sp->director->priv, VDI_VSLP_MAGIC);

	for (i = 0; i < vs->nhosts; i++) {
		if (vs->hosts[i].backend->healthy)
			return 1;
	}
	return 0;
}

/*lint -e{818} not const-able */
static void
vdi_vslp_fini(struct director *d)
{
	int i;
	struct vdi_vslp *vs;
	struct vdi_vslp_host *vh;

	CHECK_OBJ_NOTNULL(d, DIRECTOR_MAGIC);
	CAST_OBJ_NOTNULL(vs, d->priv, VDI_VSLP_MAGIC);

	free(vs->hashcircle);

	vh = vs->hosts;
	for (i = 0; i < vs->nhosts; i++, vh++)
		VBE_DropRef(vh->backend);
	free(vs->hosts);
	free(vs->dir.vcl_name);
	vs->dir.magic = 0;
	FREE_OBJ(vs);
}

/*
 * hash with the chosen digest and return key
 *
 * 0 means undefined, so this function will never return 0 unless
 * for an error
 */
static inline uint32_t
vdi_vslp_hash(const enum vdi_vslp_digest_t digest, const char *s)
{
	uint32_t	res;

	switch (digest) {
	case DEFAULT:
	case CRC32:
		res = crc32_l(s, strlen(s));
		break;
	case SHA256:
	{
		struct SHA256Context sha256;
		union {
			unsigned char digest[32];
			uint32_t uint32_digest[8];
		} sha256_digest;

		SHA256_Init(&sha256);
		SHA256_Update(&sha256, s, strlen(s));
		SHA256_Final(sha256_digest.digest, &sha256);

		/*
		 * use low 32 bits only
		 * XXX: Are these the best bits to pick?
		 */
		res = sha256_digest.uint32_digest[7];
		break;
	}
	case PLAIN:
	{
		res = 0;
		while (*s)
			res += *s++;
		break;
	}
	case RS:
	{
		/* hash function from Robert Sedgwicks 'Algorithms in C' book */
		const uint32_t b    = 378551;
		uint32_t a	    = 63689;

		res = 0;

		while (*s) {
			res = res * a + (*s++);
			a *= b;
		}
		break;
	}
	default:
		INCOMPL();
		return (0);
	}

	res++;
	/* may wrap, so check for 0 */
	return (res ? res : 1);
}

void
VRT_init_dir_vslp(struct cli *cli, struct director **bp,
    const struct vrt_dir_vslp *t)
{
	struct vdi_vslp *vs;
	struct vdi_vslp_host *vh;
	const struct vrt_dir_vslp_entry *te;
	int i, j;

	ASSERT_CLI();
	(void)cli;

	ALLOC_OBJ(vs, VDI_VSLP_MAGIC);
	XXXAN(vs);
	vs->hosts = calloc(sizeof *vh, t->nmember);
	XXXAN(vs->hosts);

	vs->dir.magic = DIRECTOR_MAGIC;
	vs->dir.priv = vs;
	vs->dir.name = "vslp";
	REPLACE(vs->dir.vcl_name, t->name);
	vs->dir.getfd = vdi_vslp_getfd;
	vs->dir.fini = vdi_vslp_fini;
	vs->dir.healthy = vdi_vslp_healthy;
	vs->dir.backend_name = NULL;
	vs->dir.recovered = NULL;
	vs->dir.dss_ini = vdi_vslp_dss_ini;
	vs->dir.dss_fini = vdi_vslp_dss_fini;
	vs->dir.choice = vdi_vslp_choice;

	vs->nhosts = t->nmember;

	vs->altsrv_p = t->altsrv_p;
	vs->rampup = t->rampup;

	vs->hashcircle = calloc(vs->nhosts * REPLICAS,
	    sizeof (struct vdi_vslp_hostnode));
	XXXAN(vs->hashcircle);

	te = t->members;
	for (i = 0; i < t->nmember; i++, te++) {
		vs->hosts[i].backend = VBE_AddBackend(cli, te->host);
		for (j = 0; j < REPLICAS; j++) {
			int len = strlen(VDI_BackendName(vs->hosts[i]))
			    + (j == 0 ? 0 : log10(j)) + 2;
			char s[len];

			sprintf(s, "%s%d", VDI_BackendName(vs->hosts[i]), j);
			vs->hashcircle[i * REPLICAS + j].point =
			    vdi_vslp_hash(SHA256, s);
			vs->hashcircle[i * REPLICAS + j].host = i;
		}
		vs->hosts[i].canon_point = vs->hashcircle[i * REPLICAS].point;
	}
	qsort( (void *) vs->hashcircle, vs->nhosts * REPLICAS,
	    sizeof (struct vdi_vslp_hostnode), (compar) hostnode_compare);

	if (params->diag_bitmap & DIAG_DIRECTOR)
		for (i = 0; i < vs->nhosts; i++)
			for (j = 0; j < REPLICAS; j++)
				DSL(DIAG_DIRECTOR, SLT_Debug, 0,
				"VSLP hashcircle[%d] = {point = %u, host = %d}",
				i * REPLICAS + j,
				vs->hashcircle[i * REPLICAS + j].point,
				vs->hashcircle[i * REPLICAS + j].host);

	*bp = &vs->dir;
}

/* WO */

void
VRT_l_dir_key_ip(struct sess *sp, struct sockaddr *sa)
{
	/* partly taken from VRT_IP_string */
	struct vdi_vslp_dss *dss;
	const struct sockaddr_in *si4;
	const struct sockaddr_in6 *si6;

	dss = VDI_get_dss(sp);
	if (! dss)
		return;

	switch (sa->sa_family) {
	case AF_INET:
		si4 = (const void *)sa;
		dss->persistence_key = ntohl(si4->sin_addr.s_addr);
		break;
	case AF_INET6:
		si6 = (const void *)sa;
		/*
		 * use low 32 bits to also include ipv4 mapping
		 *
		 */
		dss->persistence_key =
		    si6->sin6_addr.s6_addr[15] +
		(si6->sin6_addr.s6_addr[14] << 8) +
		(si6->sin6_addr.s6_addr[13] << 16) +
		(si6->sin6_addr.s6_addr[12] << 24);
		break;
	default:
		INCOMPL();
	}
}

/*
 * XXX: vargs declaration only to match tcl script
 *
 * will need to either support vargs or find a way to change
 * tcp script output
 */

void
VRT_l_dir_key_string(struct sess *sp, const char *s, ...)
{
	struct vdi_vslp_dss *dss;
	struct vdi_vslp *vs;
	enum vdi_vslp_digest_t digest;

	dss = VDI_get_dss(sp);
	if (! dss)
		return;

	if (s == NULL) {
		dss->persistence_key = 0;
		return;
	}

	CHECK_OBJ_NOTNULL(sp->director, DIRECTOR_MAGIC);

	if ((dss->digest == DEFAULT) &&
	    (sp->director->priv != NULL) &&
	    (((struct vdi_vslp *)sp->director->priv)->magic == VDI_VSLP_MAGIC)) {
		CAST_OBJ_NOTNULL(vs, sp->director->priv, VDI_VSLP_MAGIC);
		digest = vs->digest;
	} else {
		digest = dss->digest;
	}

	dss->persistence_key = vdi_vslp_hash(digest, s);
}

void
VRT_l_dir_key_int(struct sess *sp, const uint32_t key)
{
	struct vdi_vslp_dss *dss;

	dss = VDI_get_dss(sp);
	if (! dss)
		return;

	dss->persistence_key = key;
}

/* RO */

uint32_t
VRT_r_dir_key(struct sess *sp)
{
	struct vdi_vslp_dss *dss;

	dss = VDI_get_dss(sp);
	if (! dss)
		return (0);

	return(dss->persistence_key);
}

/* RW */

/* default, crc32, sha256, plain */

void
VRT_l_dir_digest(struct sess *sp, const char *digest_str)
{
	struct vdi_vslp_dss *dss;
	enum vdi_vslp_digest_t digest = DEFAULT;
	const char *s = digest_str;

	/* XXX: error reporting? */
	if ((s == NULL) || (*s == '\0'))
		return;

	dss = VDI_get_dss(sp);
	if (! dss)
		return;

	switch (*digest_str) {
	case 'd':
		if ((*++s == 'e') &&
		(*++s == 'f') &&
		(*++s == 'a') &&
		(*++s == 'u') &&
		(*++s == 'l') &&
		(*++s == 't') &&
		(*++s == '\0')) {
			digest = DEFAULT;
		}
		break;
	case 'c':
		if ((*++s == 'r') &&
		(*++s == 'c') &&
		(*++s == '3') &&
		(*++s == '2') &&
		(*++s == '\0')) {
			digest = CRC32;
		}
		break;
	case 's':
		if ((*++s == 'h') &&
		(*++s == 'a') &&
		(*++s == '2') &&
		(*++s == '5') &&
		(*++s == '6') &&
		(*++s == '\0')) {
			digest = SHA256;
		}
		break;
	case 'p':
		if ((*++s == 'l') &&
		(*++s == 'a') &&
		(*++s == 'i') &&
		(*++s == 'n') &&
		(*++s == '\0')) {
			digest = PLAIN;
		}
		break;
	case 'r':
		if ((*++s == 's') &&
		(*++s == '\0')) {
			digest = RS;
		}
		break;
	default:
		/* XXX: report error how ? */
		return;
	}
	dss->digest = digest;
}

const char *
VRT_r_dir_digest(struct sess *sp)
{
	struct vdi_vslp_dss *dss;
	dss = VDI_get_dss(sp);

	if (! dss)
		return ("(INVALIDDIRECTOR)");

	switch (dss->digest) {
	case DEFAULT:
		return ("default");
	case CRC32:
		return ("crc32");
	case SHA256:
		return ("sha256");
	case PLAIN:
		return ("plain");
	case RS:
		return ("rs");
	}
	return ("(INVALID)");
}
