FS-10165: Removed dependancy and autoconf check for libsodium from libks. Moved DHT, bencode, and related tests into libblade. Still requires refactoring in blade from ks_* to blade_*.

libs/libblade/Makefile.am

@@ -16,15 +16,21 @@ libblade_la_SOURCES	+= src/blade_datastore.c
 libblade_la_SOURCES	+= src/blade_identity.c src/blade_module.c src/blade_connection.c
 libblade_la_SOURCES	+= src/blade_session.c src/blade_protocol.c src/blade_space.c src/blade_method.c
 libblade_la_SOURCES	+= src/blade_module_wss.c src/blade_module_chat.c
+libblade_la_SOURCES	+= src/dht/ks_dht.c src/dht/ks_dht_datagram.c src/dht/ks_dht_endpoint.c src/dht/ks_dht_message.c src/dht/ks_dht_transaction.c
+libblade_la_SOURCES	+= src/dht/ks_dht_job.c src/dht/ks_dht_search.c src/dht/ks_dht_publish.c src/dht/ks_dht_distribute.c src/dht/ks_dht_storageitem.c
+libblade_la_SOURCES	+= src/dht/ks_dht_bucket.c src/ks_bencode.c
+
 libblade_la_CFLAGS	= $(AM_CFLAGS) $(AM_CPPFLAGS)
 libblade_la_LDFLAGS     = -version-info 0:1:0 -lncurses -lpthread -lm -lconfig $(AM_LDFLAGS)
 libblade_la_LIBADD      = libunqlite.la
+
 library_includedir	= $(prefix)/include
 library_include_HEADERS = src/include/blade.h src/include/blade_types.h src/include/blade_stack.h
 library_include_HEADERS += src/include/blade_datastore.h
 library_include_HEADERS += src/include/blade_identity.h src/include/blade_module.h src/include/blade_connection.h
 library_include_HEADERS += src/include/blade_session.h src/include/blade_protocol.h src/include/blade_space.h src/include/blade_method.h
 library_include_HEADERS += src/include/unqlite.h test/tap.h
+library_include_HEADERS += src/include/ks_dht.h src/include/ks_bencode.h
 
 tests: libblade.la
 	$(MAKE) -C test tests

libs/libblade/src/dht/ks_dht_datagram.c

@@ -0,0 +1,60 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+#include "sodium.h"
+
+KS_DECLARE(ks_status_t) ks_dht_datagram_create(ks_dht_datagram_t **datagram,
+											   ks_pool_t *pool,
+											   ks_dht_t *dht,
+											   ks_dht_endpoint_t *endpoint,
+											   const ks_sockaddr_t *raddr)
+{
+	ks_dht_datagram_t *dg;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(datagram);
+	ks_assert(pool);
+	ks_assert(dht);
+	ks_assert(endpoint);
+	ks_assert(raddr);
+	ks_assert(raddr->family == AF_INET || raddr->family == AF_INET6);
+
+	*datagram = dg = ks_pool_alloc(pool, sizeof(ks_dht_datagram_t));
+	ks_assert(dg);
+
+	dg->pool = pool;
+	dg->dht = dht;
+	dg->endpoint = endpoint;
+	dg->raddr = *raddr;
+
+	memcpy(dg->buffer, dht->recv_buffer, dht->recv_buffer_length);
+	dg->buffer_length = dht->recv_buffer_length;
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		ks_dht_datagram_destroy(datagram);
+	}
+	return ret;
+}
+
+KS_DECLARE(void) ks_dht_datagram_destroy(ks_dht_datagram_t **datagram)
+{
+	ks_dht_datagram_t *dg;
+
+	ks_assert(datagram);
+	ks_assert(*datagram);
+
+	dg = *datagram;
+
+	ks_pool_free(dg->pool, datagram);
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/dht/ks_dht_distribute.c

@@ -0,0 +1,65 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+#include "sodium.h"
+
+KS_DECLARE(ks_status_t) ks_dht_distribute_create(ks_dht_distribute_t **distribute,
+												 ks_pool_t *pool,
+												 ks_dht_storageitem_callback_t callback,
+												 void *data,
+												 int64_t cas,
+												 ks_dht_storageitem_t *item)
+{
+	ks_dht_distribute_t *d;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(distribute);
+	ks_assert(pool);
+	ks_assert(cas >= 0);
+	ks_assert(item);
+
+	*distribute = d = ks_pool_alloc(pool, sizeof(ks_dht_distribute_t));
+	ks_assert(d);
+
+	d->pool = pool;
+
+	d->callback = callback;
+	d->data = data;
+	ks_mutex_create(&d->mutex, KS_MUTEX_FLAG_DEFAULT, d->pool);
+	ks_assert(d->mutex);
+	d->cas = cas;
+	d->item = item;
+	
+	ks_dht_storageitem_reference(d->item);
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		if (d) ks_dht_distribute_destroy(distribute);
+	}
+	return ret;
+}
+
+KS_DECLARE(void) ks_dht_distribute_destroy(ks_dht_distribute_t **distribute)
+{
+	ks_dht_distribute_t *d;
+
+	ks_assert(distribute);
+	ks_assert(*distribute);
+
+	d = *distribute;
+
+	if (d->mutex) ks_mutex_destroy(&d->mutex);
+	ks_dht_storageitem_dereference(d->item);
+	
+	ks_pool_free(d->pool, distribute);
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/dht/ks_dht_endpoint.c

@@ -0,0 +1,61 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+#include "sodium.h"
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_endpoint_create(ks_dht_endpoint_t **endpoint,
+											   ks_pool_t *pool,
+											   const ks_sockaddr_t *addr,
+											   ks_socket_t sock)
+{
+	ks_dht_endpoint_t *ep;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(endpoint);
+	ks_assert(pool);
+	ks_assert(addr);
+	ks_assert(addr->family == AF_INET || addr->family == AF_INET6);
+
+	*endpoint = ep = ks_pool_alloc(pool, sizeof(ks_dht_endpoint_t));
+	ks_assert(ep);
+
+	ep->pool = pool;
+	ep->addr = *addr;
+	ep->sock = sock;
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		if (ep) ks_dht_endpoint_destroy(endpoint);
+	}
+	return ret;
+}
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_endpoint_destroy(ks_dht_endpoint_t **endpoint)
+{
+	ks_dht_endpoint_t *ep;
+
+	ks_assert(endpoint);
+	ks_assert(*endpoint);
+
+	ep = *endpoint;
+
+	if (ep->sock != KS_SOCK_INVALID) ks_socket_close(&ep->sock);
+
+	ks_pool_free(ep->pool, endpoint);
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/dht/ks_dht_job.c

@@ -0,0 +1,136 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+
+KS_DECLARE(ks_status_t) ks_dht_job_create(ks_dht_job_t **job,
+										  ks_pool_t *pool,
+										  const ks_sockaddr_t *raddr,
+										  int32_t attempts,
+										  void *data)
+{
+	ks_dht_job_t *j;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(job);
+	ks_assert(pool);
+	//ks_assert(dht);
+	ks_assert(attempts > 0 && attempts <= 10);
+
+	*job = j = ks_pool_alloc(pool, sizeof(ks_dht_job_t));
+	ks_assert(j);
+
+	j->pool = pool;
+	j->state = KS_DHT_JOB_STATE_QUERYING;
+	if (raddr) j->raddr = *raddr;
+	j->attempts = attempts;
+	j->data = data;
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		if (j) ks_dht_job_destroy(job);
+	}
+	return ret;
+}
+
+KS_DECLARE(void) ks_dht_job_build_ping(ks_dht_job_t *job, ks_dht_job_callback_t query_callback, ks_dht_job_callback_t finish_callback)
+{
+	ks_assert(job);
+	ks_assert(query_callback);
+
+	job->query_callback = query_callback;
+	job->finish_callback = finish_callback;
+}
+
+KS_DECLARE(void) ks_dht_job_build_findnode(ks_dht_job_t *job,
+										   ks_dht_job_callback_t query_callback,
+										   ks_dht_job_callback_t finish_callback,
+										   ks_dht_nodeid_t *target)
+{
+	ks_assert(job);
+	ks_assert(query_callback);
+	ks_assert(target);
+
+	job->query_callback = query_callback;
+	job->finish_callback = finish_callback;
+	job->query_target = *target;
+}
+
+KS_DECLARE(void) ks_dht_job_build_get(ks_dht_job_t *job,
+									  ks_dht_job_callback_t query_callback,
+									  ks_dht_job_callback_t finish_callback,
+									  ks_dht_nodeid_t *target,
+									  const uint8_t *salt,
+									  ks_size_t salt_length)
+{
+	ks_assert(job);
+	ks_assert(query_callback);
+	ks_assert(target);
+
+	job->query_callback = query_callback;
+	job->finish_callback = finish_callback;
+	job->query_target = *target;
+	if (salt && salt_length > 0) job->query_salt = ben_blob(salt, salt_length);
+}
+
+KS_DECLARE(void) ks_dht_job_build_put(ks_dht_job_t *job,
+									  ks_dht_job_callback_t query_callback,
+									  ks_dht_job_callback_t finish_callback,
+									  ks_dht_token_t *token,
+									  int64_t cas,
+									  ks_dht_storageitem_t *item)
+{
+	ks_assert(job);
+	ks_assert(query_callback);
+	ks_assert(token);
+	ks_assert(item);
+
+	job->query_callback = query_callback;
+	job->finish_callback = finish_callback;
+	job->query_token = *token;
+	job->query_cas = cas;
+	job->query_storageitem = item;
+	ks_dht_storageitem_reference(job->query_storageitem);
+}
+
+KS_DECLARE(void) ks_dht_job_build_search(ks_dht_job_t *job,
+										 ks_dht_job_callback_t query_callback,
+										 ks_dht_job_callback_t finish_callback)
+{
+	ks_assert(job);
+	ks_assert(query_callback);
+
+	job->query_callback = query_callback;
+	job->finish_callback = finish_callback;
+}
+
+KS_DECLARE(void) ks_dht_job_destroy(ks_dht_job_t **job)
+{
+	ks_dht_job_t *j;
+
+	ks_assert(job);
+	ks_assert(*job);
+
+	j = *job;
+
+	if (j->query_salt) ben_free(j->query_salt);
+	if (j->response_id) ks_dhtrt_release_node(j->response_id);
+	for (int32_t i = 0; i < j->response_nodes_count; ++i) ks_dhtrt_release_node(j->response_nodes[i]);
+	for (int32_t i = 0; i < j->response_nodes6_count; ++i) ks_dhtrt_release_node(j->response_nodes6[i]);
+
+	if (j->query_storageitem) ks_dht_storageitem_dereference(j->query_storageitem);
+	if (j->response_storageitem) ks_dht_storageitem_dereference(j->response_storageitem);
+
+	if (j->error_description) ben_free(j->error_description);
+
+	ks_pool_free(j->pool, job);
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/dht/ks_dht_message.c

@@ -0,0 +1,122 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+
+KS_DECLARE(ks_status_t) ks_dht_message_create(ks_dht_message_t **message,
+											  ks_pool_t *pool,
+											  ks_dht_endpoint_t *endpoint,
+											  const ks_sockaddr_t *raddr,
+											  ks_bool_t alloc_data)
+{
+	ks_dht_message_t *m;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(message);
+	ks_assert(pool);
+
+	*message = m = ks_pool_alloc(pool, sizeof(ks_dht_message_t));
+	ks_assert(m);
+
+	m->pool = pool;
+	m->endpoint = endpoint;
+	m->raddr = *raddr;
+	if (alloc_data) {
+		m->data = ben_dict();
+		ks_assert(m->data);
+	}
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		ks_dht_message_destroy(message);
+	}
+	return ret;
+}
+
+KS_DECLARE(void) ks_dht_message_destroy(ks_dht_message_t **message)
+{
+	ks_dht_message_t *m;
+
+	ks_assert(message);
+	ks_assert(*message);
+
+	m = *message;
+
+    if (m->data) {
+		ben_free(m->data);
+		m->data = NULL;
+	}
+
+	ks_pool_free(m->pool, message);
+}
+
+
+KS_DECLARE(ks_status_t) ks_dht_message_parse(ks_dht_message_t *message, const uint8_t *buffer, ks_size_t buffer_length)
+{
+	struct bencode *t;
+	struct bencode *y;
+	const char *tv;
+	const char *yv;
+	ks_size_t tv_len;
+	ks_size_t yv_len;
+
+	ks_assert(message);
+	ks_assert(message->pool);
+	ks_assert(buffer);
+	ks_assert(!message->data);
+
+    message->data = ben_decode((const void *)buffer, buffer_length);
+	if (!message->data) {
+		ks_log(KS_LOG_DEBUG, "Message cannot be decoded\n");
+		return KS_STATUS_FAIL;
+	}
+
+	ks_log(KS_LOG_DEBUG, "Message decoded\n");
+	ks_log(KS_LOG_DEBUG, "%s\n", ben_print(message->data));
+	
+    t = ben_dict_get_by_str(message->data, "t");
+	if (!t) {
+		ks_log(KS_LOG_DEBUG, "Message missing required key 't'\n");
+		return KS_STATUS_FAIL;
+	}
+
+	tv = ben_str_val(t);
+	tv_len = ben_str_len(t);
+	if (tv_len > KS_DHT_MESSAGE_TRANSACTIONID_MAX_SIZE) {
+		ks_log(KS_LOG_DEBUG, "Message 't' value has an unexpectedly large size of %d\n", tv_len);
+		return KS_STATUS_FAIL;
+	}
+
+	memcpy(message->transactionid, tv, tv_len);
+	message->transactionid_length = tv_len;
+	// @todo hex output of transactionid
+	//ks_log(KS_LOG_DEBUG, "Message transaction id is %d\n", message->transactionid);
+
+    y = ben_dict_get_by_str(message->data, "y");
+	if (!y) {
+		ks_log(KS_LOG_DEBUG, "Message missing required key 'y'\n");
+		return KS_STATUS_FAIL;
+	}
+
+	yv = ben_str_val(y);
+	yv_len = ben_str_len(y);
+	if (yv_len >= KS_DHT_MESSAGE_TYPE_MAX_SIZE) {
+		ks_log(KS_LOG_DEBUG, "Message 'y' value has an unexpectedly large size of %d\n", yv_len);
+		return KS_STATUS_FAIL;
+	}
+
+	memcpy(message->type, yv, yv_len);
+	message->type[yv_len] = '\0';
+	//ks_log(KS_LOG_DEBUG, "Message type is '%s'\n", message->type);
+
+	return KS_STATUS_SUCCESS;
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/dht/ks_dht_publish.c

@@ -0,0 +1,62 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+#include "sodium.h"
+
+KS_DECLARE(ks_status_t) ks_dht_publish_create(ks_dht_publish_t **publish,
+											  ks_pool_t *pool,
+											  ks_dht_job_callback_t callback,
+											  void *data,
+											  int64_t cas,
+											  ks_dht_storageitem_t *item)
+{
+	ks_dht_publish_t *p;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(publish);
+	ks_assert(pool);
+	ks_assert(cas >= 0);
+	ks_assert(item);
+
+	*publish = p = ks_pool_alloc(pool, sizeof(ks_dht_publish_t));
+	ks_assert(p);
+
+	p->pool = pool;
+
+	p->callback = callback;
+	p->data = data;
+	p->cas = cas;
+	p->item = item;
+	
+	ks_dht_storageitem_reference(p->item);
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		if (p) ks_dht_publish_destroy(publish);
+	}
+	return ret;
+}
+
+KS_DECLARE(void) ks_dht_publish_destroy(ks_dht_publish_t **publish)
+{
+	ks_dht_publish_t *p;
+
+	ks_assert(publish);
+	ks_assert(*publish);
+
+	p = *publish;
+
+	ks_dht_storageitem_dereference(p->item);
+	
+	ks_pool_free(p->pool, publish);
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/dht/ks_dht_search.c

@@ -0,0 +1,71 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+#include "sodium.h"
+
+KS_DECLARE(ks_status_t) ks_dht_search_create(ks_dht_search_t **search,
+											 ks_pool_t *pool,
+											 ks_dhtrt_routetable_t *table,
+											 const ks_dht_nodeid_t *target,
+											 ks_dht_job_callback_t callback,
+											 void *data)
+{
+	ks_dht_search_t *s;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(search);
+	ks_assert(pool);
+	ks_assert(table);
+	ks_assert(target);
+
+	*search = s = ks_pool_alloc(pool, sizeof(ks_dht_search_t));
+	ks_assert(s);
+
+	s->pool = pool;
+
+	ks_mutex_create(&s->mutex, KS_MUTEX_FLAG_DEFAULT, s->pool);
+	ks_assert(s->mutex);
+
+	s->table = table;
+	memcpy(s->target.id, target->id, KS_DHT_NODEID_SIZE);
+
+	s->callback = callback;
+	s->data = data;
+
+	ks_hash_create(&s->searched, KS_HASH_MODE_ARBITRARY, KS_HASH_FLAG_NOLOCK | KS_HASH_FLAG_DUP_CHECK, s->pool);
+	ks_assert(s->searched);
+	ks_hash_set_keysize(s->searched, KS_DHT_NODEID_SIZE);
+
+	s->searching = 0;
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		if (s) ks_dht_search_destroy(search);
+	}
+	return ret;
+}
+
+KS_DECLARE(void) ks_dht_search_destroy(ks_dht_search_t **search)
+{
+	ks_dht_search_t *s;
+
+	ks_assert(search);
+	ks_assert(*search);
+
+	s = *search;
+
+	if (s->searched) ks_hash_destroy(&s->searched);
+	if (s->mutex) ks_mutex_destroy(&s->mutex);
+
+	ks_pool_free(s->pool, search);
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/dht/ks_dht_storageitem.c

@@ -0,0 +1,214 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+#include "sodium.h"
+
+KS_DECLARE(ks_status_t) ks_dht_storageitem_create_immutable_internal(ks_dht_storageitem_t **item,
+																	 ks_pool_t *pool,
+																	 ks_dht_nodeid_t *target,
+																	 struct bencode *v,
+																	 ks_bool_t clone_v)
+{
+	ks_dht_storageitem_t *si;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(item);
+	ks_assert(pool);
+	ks_assert(v);
+	ks_assert(SHA_DIGEST_LENGTH == KS_DHT_NODEID_SIZE);
+
+	*item = si = ks_pool_alloc(pool, sizeof(ks_dht_storageitem_t));
+	ks_assert(si);
+
+	si->pool = pool;
+	si->id = *target;
+	si->mutable = KS_FALSE;
+	si->expiration = ks_time_now() + ((ks_time_t)KS_DHT_STORAGEITEM_EXPIRATION * KS_USEC_PER_SEC);
+	si->keepalive = ks_time_now() + ((ks_time_t)KS_DHT_STORAGEITEM_KEEPALIVE * KS_USEC_PER_SEC);
+	si->v = clone_v ? ben_clone(v) : v;
+	ks_assert(si->v);
+
+	si->refc = 1;
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		if (si) ks_dht_storageitem_destroy(item);
+	}
+	return ret;
+}
+
+KS_DECLARE(ks_status_t) ks_dht_storageitem_create_immutable(ks_dht_storageitem_t **item,
+															ks_pool_t *pool,
+															ks_dht_nodeid_t *target,
+															const uint8_t *value,
+															ks_size_t value_length)
+{
+	struct bencode *v = NULL;
+
+	ks_assert(item);
+	ks_assert(pool);
+	ks_assert(value);
+	ks_assert(value_length > 0);
+	ks_assert(SHA_DIGEST_LENGTH == KS_DHT_NODEID_SIZE);
+
+	v = ben_blob(value, value_length);
+	ks_assert(v);
+	
+	return ks_dht_storageitem_create_immutable_internal(item, pool, target, v, KS_FALSE);
+}
+
+KS_DECLARE(ks_status_t) ks_dht_storageitem_create_mutable_internal(ks_dht_storageitem_t **item,
+																   ks_pool_t *pool,
+																   ks_dht_nodeid_t *target,
+																   struct bencode *v,
+																   ks_bool_t clone_v,
+																   ks_dht_storageitem_pkey_t *pk,
+																   struct bencode *salt,
+																   ks_bool_t clone_salt,
+																   int64_t sequence,
+																   ks_dht_storageitem_signature_t *signature)
+{
+	ks_dht_storageitem_t *si;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(item);
+	ks_assert(pool);
+	ks_assert(v);
+	ks_assert(SHA_DIGEST_LENGTH == KS_DHT_NODEID_SIZE);
+	ks_assert(pk);
+	ks_assert(signature);
+
+	*item = si = ks_pool_alloc(pool, sizeof(ks_dht_storageitem_t));
+	ks_assert(si);
+
+	si->pool = pool;
+	si->id = *target;
+	si->mutable = KS_TRUE;
+	si->expiration = ks_time_now() + ((ks_time_t)KS_DHT_STORAGEITEM_EXPIRATION * KS_USEC_PER_SEC);
+	si->keepalive = ks_time_now() + ((ks_time_t)KS_DHT_STORAGEITEM_KEEPALIVE * KS_USEC_PER_SEC);
+	si->v = clone_v ? ben_clone(v) : v;
+	ks_assert(si->v);
+
+	si->refc = 1;
+
+	ks_mutex_create(&si->mutex, KS_MUTEX_FLAG_DEFAULT, si->pool);
+	ks_assert(si->mutex);
+	
+	si->pk = *pk;
+	if (salt) {
+		si->salt = clone_salt ? ben_clone(salt) : salt;
+		ks_assert(si->salt);
+	}
+	si->seq = sequence;
+	si->sig = *signature;
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		if (si) ks_dht_storageitem_destroy(item);
+	}
+	return ret;
+}
+
+KS_DECLARE(ks_status_t) ks_dht_storageitem_create_mutable(ks_dht_storageitem_t **item,
+														  ks_pool_t *pool,
+														  ks_dht_nodeid_t *target,
+														  const uint8_t *value,
+														  ks_size_t value_length,
+														  ks_dht_storageitem_pkey_t *pk,
+														  const uint8_t *salt,
+														  ks_size_t salt_length,
+														  int64_t sequence,
+														  ks_dht_storageitem_signature_t *signature)
+{
+	struct bencode *v = NULL;
+	struct bencode *s = NULL;
+
+	ks_assert(item);
+	ks_assert(pool);
+	ks_assert(value);
+	ks_assert(value_length > 0);
+	ks_assert(SHA_DIGEST_LENGTH == KS_DHT_NODEID_SIZE);
+	ks_assert(pk);
+	ks_assert(signature);
+
+	v = ben_blob(value, value_length);
+	if (salt && salt_length > 0) s = ben_blob(salt, salt_length);
+	return ks_dht_storageitem_create_mutable_internal(item, pool, target, v, KS_FALSE, pk, s, KS_FALSE, sequence, signature);
+}
+
+KS_DECLARE(void) ks_dht_storageitem_update_mutable(ks_dht_storageitem_t *item, struct bencode *v, int64_t sequence, ks_dht_storageitem_signature_t *signature)
+{
+	ks_assert(item);
+	ks_assert(v);
+	ks_assert(sequence);
+	ks_assert(signature);
+
+	ks_mutex_lock(item->mutex);
+	ben_free(item->v);
+	item->v = ben_clone(v);
+	item->seq = sequence;
+	item->sig = *signature;
+	ks_mutex_unlock(item->mutex);
+}
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_storageitem_destroy(ks_dht_storageitem_t **item)
+{
+	ks_dht_storageitem_t *si;
+	
+	ks_assert(item);
+	ks_assert(*item);
+
+	si = *item;
+
+	if (si->v) {
+		ben_free(si->v);
+		si->v = NULL;
+	}
+	if (si->mutex) ks_mutex_destroy(&si->mutex);
+	if (si->salt) {
+		ben_free(si->salt);
+		si->salt = NULL;
+	}
+
+	ks_pool_free(si->pool, item);
+}
+
+KS_DECLARE(void) ks_dht_storageitem_reference(ks_dht_storageitem_t *item)
+{
+	ks_assert(item);
+
+	ks_mutex_lock(item->mutex);
+	item->refc++;
+	ks_mutex_unlock(item->mutex);
+}
+
+KS_DECLARE(void) ks_dht_storageitem_dereference(ks_dht_storageitem_t *item)
+{
+	ks_assert(item);
+
+	ks_mutex_lock(item->mutex);
+	item->refc--;
+	ks_mutex_unlock(item->mutex);
+
+	ks_assert(item->refc >= 0);
+}
+
+KS_DECLARE(void) ks_dht_storageitem_callback(ks_dht_storageitem_t *item, ks_dht_storageitem_callback_t callback)
+{
+	ks_assert(item);
+
+	item->callback = callback;
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/dht/ks_dht_transaction.c

@@ -0,0 +1,54 @@
+#include "ks_dht.h"
+#include "ks_dht-int.h"
+
+KS_DECLARE(ks_status_t) ks_dht_transaction_create(ks_dht_transaction_t **transaction,
+												  ks_pool_t *pool,
+												  ks_dht_job_t *job,
+												  uint32_t transactionid,
+												  ks_dht_job_callback_t callback)
+{
+	ks_dht_transaction_t *t;
+	ks_status_t ret = KS_STATUS_SUCCESS;
+
+	ks_assert(transaction);
+	ks_assert(pool);
+	ks_assert(job);
+
+	*transaction = t = ks_pool_alloc(pool, sizeof(ks_dht_transaction_t));
+	ks_assert(t);
+
+	t->pool = pool;
+	t->job = job;
+	t->transactionid = transactionid;
+	t->callback = callback;
+	t->expiration = ks_time_now() + ((ks_time_t)KS_DHT_TRANSACTION_EXPIRATION * KS_USEC_PER_SEC);
+
+	// done:
+	if (ret != KS_STATUS_SUCCESS) {
+		if (t) ks_dht_transaction_destroy(transaction);
+	}
+	return ret;
+}
+
+KS_DECLARE(void) ks_dht_transaction_destroy(ks_dht_transaction_t **transaction)
+{
+	ks_dht_transaction_t *t;
+
+	ks_assert(transaction);
+	ks_assert(*transaction);
+
+	t = *transaction;
+
+	ks_pool_free(t->pool, transaction);
+}
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/include/blade.h

@@ -48,6 +48,8 @@
 #include "blade_datastore.h"
 #include "blade_space.h"
 #include "blade_method.h"
+#include "ks_dht.h"
+#include "ks_bencode.h"
 
 KS_BEGIN_EXTERN_C
 

libs/libblade/src/include/ks_bencode.h

@@ -0,0 +1,730 @@
+/*
+ * libbencodetools
+ *
+ * Written by Heikki Orsila <heikki.orsila@iki.fi> and
+ * Janne Kulmala <janne.t.kulmala@tut.fi> in 2011.
+ */
+
+#ifndef TYPEVALIDATOR_BENCODE_H
+#define TYPEVALIDATOR_BENCODE_H
+
+#include <stdlib.h>
+#include <stdio.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/* Used to verify format strings in compile time */
+#ifdef __GNUC__
+#define BEN_CHECK_FORMAT(...)  __attribute__ ((format( __VA_ARGS__ )))
+#else
+#define BEN_CHECK_FORMAT(args)
+#endif
+
+enum {
+	BENCODE_BOOL = 1,
+	BENCODE_DICT,
+	BENCODE_INT,
+	BENCODE_LIST,
+	BENCODE_STR,
+	BENCODE_USER,
+};
+
+enum {
+	BEN_OK = 0, /* No errors. Set to zero. Non-zero implies an error. */
+	BEN_INVALID,      /* Invalid data was given to decoder */
+	BEN_INSUFFICIENT, /* Insufficient amount of data for decoding */
+	BEN_NO_MEMORY,    /* Memory allocation failed */
+	BEN_MISMATCH,     /* A given structure did not match unpack format */
+};
+
+struct bencode {
+	char type;
+};
+
+struct bencode_bool {
+	char type;
+	char b;
+};
+
+struct bencode_dict_node {
+	long long hash;
+	struct bencode *key;
+	struct bencode *value;
+	size_t next;
+};
+
+struct bencode_dict {
+	char type;
+	char shared; /* non-zero means that the internal data is shared with
+			other instances and should not be freed */
+	size_t n;
+	size_t alloc;
+	size_t *buckets;
+	struct bencode_dict_node *nodes;
+};
+
+struct bencode_int {
+	char type;
+	long long ll;
+};
+
+struct bencode_list {
+	char type;
+	char shared; /* non-zero means that the internal data is shared with
+			other instances and should not be freed */
+	size_t n;
+	size_t alloc;
+	struct bencode **values;
+};
+
+struct bencode_str {
+	char type;
+	size_t len;
+	char *s;
+};
+
+struct ben_decode_ctx;
+struct ben_encode_ctx;
+
+struct bencode_type {
+	size_t size;
+	struct bencode *(*decode) (struct ben_decode_ctx *ctx);
+	int (*encode) (struct ben_encode_ctx *ctx, const struct bencode *b);
+	size_t (*get_size) (const struct bencode *b);
+	void (*free) (struct bencode *b);
+	int (*cmp) (const struct bencode *a, const struct bencode *b);
+};
+
+struct bencode_user {
+	char type;
+	struct bencode_type *info;
+};
+
+struct bencode_error {
+	int error;  /* 0 if no errors */
+	int line;   /* Error line: 0 is the first line */
+	size_t off; /* Error offset in bytes from the start */
+};
+
+/* Allocate an instance of a user-defined type */
+void *ben_alloc_user(struct bencode_type *type);
+
+/*
+ * Try to set capacity of a list or a dict to 'n' objects.
+ * The function does nothing if 'n' is less than or equal to the number of
+ * objects in 'b'. That is, nothing happens if n <= ben_{dict|list}_len(b).
+ *
+ * This function is used only for advice. The implementation need not obey it.
+ *
+ * The function returns 0 if the new capacity is used, otherwise -1.
+ *
+ * Note: This can be used to make construction of lists and dicts
+ * more efficient when the number of inserted items is known in advance.
+ */
+int ben_allocate(struct bencode *b, size_t n);
+
+/*
+ * Returns an identical but a separate copy of structure b. Returns NULL if
+ * there is no memory to make a copy. The copy is made recursively.
+ */
+struct bencode *ben_clone(const struct bencode *b);
+
+/*
+ * Returns a weak reference copy of structure b. Only a minimum amount of
+ * data is copied because the returned structure references to the same
+ * internal data as the original structure. As a result, the original
+ * structure must remain valid until the copy is destroyed.
+ *
+ * This function is used for optimization for special cases.
+ */
+struct bencode *ben_shared_clone(const struct bencode *b);
+
+/*
+ * ben_cmp() is similar to strcmp(). It compares integers, strings and lists
+ * similar to Python. User-defined types can be also compared.
+ * Note: an integer is always less than a string.
+ *
+ * ben_cmp(a, b) returns a negative value if "a < b", 0 if "a == b",
+ * or a positive value if "a > b".
+ *
+ * Algorithm for comparing dictionaries is:
+ * If 'a' and 'b' have different number of keys or keys have different values,
+ * a non-zero value is returned. Otherwise, they have the exact same keys
+ * and comparison is done in ben_cmp() order of keys. The value for each key
+ * is compared, and the first inequal value (ben_cmp() != 0) defines the
+ * return value of the comparison.
+ *
+ * Note: recursive dictionaries in depth have the same issues.
+ */
+int ben_cmp(const struct bencode *a, const struct bencode *b);
+
+/* Same as ben_cmp(), but the second argument is a C string */
+int ben_cmp_with_str(const struct bencode *a, const char *s);
+
+/*
+ * Comparison function suitable for qsort(). Uses ben_cmp(), so this can be
+ * used to order both integer and string arrays.
+ */
+int ben_cmp_qsort(const void *a, const void *b);
+
+/*
+ * Decode 'data' with 'len' bytes of data. Returns NULL on error.
+ * The encoded data must be exactly 'len' bytes (not less), otherwise NULL
+ * is returned. ben_decode2() function supports partial decoding ('len' is
+ * larger than actual decoded message) and gives more accurate error reports.
+ */
+struct bencode *ben_decode(const void *data, size_t len);
+
+/*
+ * Same as ben_decode(), but allows one to set start offset for decoding with
+ * 'off' and reports errors more accurately.
+ *
+ * '*off' must point to decoding start offset inside 'data'.
+ * If decoding is successful, '*off' is updated to point to the next byte
+ * after the decoded message.
+ *
+ * If 'error != NULL', it is updated according to the success and error of
+ * the decoding. BEN_OK is success, BEN_INVALID means invalid data.
+ * BEN_INSUFFICIENT means data is invalid but could be valid if more data
+ * was given for decoding. BEN_NO_MEMORY means decoding ran out of memory.
+ */
+struct bencode *ben_decode2(const void *data, size_t len, size_t *off, int *error);
+
+/*
+ * Same as ben_decode2(), but allows one to define user types.
+ */
+struct bencode *ben_decode3(const void *data, size_t len, size_t *off, int *error, struct bencode_type *types[128]);
+
+/*
+ * Same as ben_decode(), but decodes data encoded with ben_print(). This is
+ * whitespace tolerant, so intended Python syntax can also be read.
+ * The decoder skips comments that begin with a '#' character.
+ * The comment starts from '#' character and ends at the end of the same line.
+ *
+ * For example, this can be used to read in config files written as a Python
+ * dictionary.
+ *
+ * ben_decode_printed2() fills information about the error in
+ * struct bencode_error.
+ * error->error is 0 on success, otherwise it is an error code
+ * (see ben_decode2()).
+ * error->line is the line number where error occured.
+ * error->off is the byte offset of error (approximation).
+ */
+struct bencode *ben_decode_printed(const void *data, size_t len);
+struct bencode *ben_decode_printed2(const void *data, size_t len, size_t *off, struct bencode_error *error);
+
+/* Get the serialization size of bencode structure 'b' */
+size_t ben_encoded_size(const struct bencode *b);
+
+/* encode 'b'. Return encoded data with a pointer, and length in '*len' */
+void *ben_encode(size_t *len, const struct bencode *b);
+
+/*
+ * encode 'b' into 'data' buffer with at most 'maxlen' bytes.
+ * Returns the size of encoded data.
+ */
+size_t ben_encode2(char *data, size_t maxlen, const struct bencode *b);
+
+/*
+ * You must use ben_free() for all allocated bencode structures after use.
+ * If b == NULL, ben_free does nothing.
+ *
+ * ben_free() frees all the objects contained within the bencoded structure.
+ * It recursively iterates over lists and dictionaries and frees objects.
+ */
+void ben_free(struct bencode *b);
+
+long long ben_str_hash(const struct bencode *b);
+long long ben_int_hash(const struct bencode *b);
+long long ben_hash(const struct bencode *b);
+
+/* Create a string from binary data with len bytes */
+struct bencode *ben_blob(const void *data, size_t len);
+
+/* Create a boolean from integer */
+struct bencode *ben_bool(int b);
+
+/* Create an empty dictionary */
+struct bencode *ben_dict(void);
+
+/*
+ * Try to locate 'key' in dictionary. Returns the associated value, if found.
+ * Returns NULL if the key does not exist.
+ */
+struct bencode *ben_dict_get(const struct bencode *d, const struct bencode *key);
+
+struct bencode *ben_dict_get_by_str(const struct bencode *d, const char *key);
+struct bencode *ben_dict_get_by_int(const struct bencode *d, long long key);
+
+struct bencode_keyvalue {
+	struct bencode *key;
+	struct bencode *value;
+};
+
+/*
+ * Returns an array of key-value pairs in key order as defined by ben_cmp().
+ * Array elements are struct bencode_keyvalue members. Returns NULL if
+ * the array can not be allocated or the bencode object is not a dictionary.
+ * The returned array must be freed by using free(). The length of the
+ * array can be determined with ben_dict_len(d).
+ *
+ * Warning: key and value pointers in the array are pointers to exact same
+ * objects in the dictionary. Therefore, the dictionary and its key-values
+ * must exist while the same keys and values are accessed from the array.
+ */
+struct bencode_keyvalue *ben_dict_ordered_items(const struct bencode *d);
+
+/*
+ * Try to locate 'key' in dictionary. Returns the associated value, if found.
+ * The value must be later freed with ben_free(). Returns NULL if the key
+ * does not exist.
+ */
+struct bencode *ben_dict_pop(struct bencode *d, const struct bencode *key);
+
+struct bencode *ben_dict_pop_by_str(struct bencode *d, const char *key);
+struct bencode *ben_dict_pop_by_int(struct bencode *d, long long key);
+
+/*
+ * Set 'key' in dictionary to be 'value'. An old value exists for the key
+ * is freed if it exists. 'key' and 'value' are owned by the dictionary
+ * after a successful call (one may not call ben_free() for 'key' or
+ * 'value'). One may free 'key' and 'value' if the call is unsuccessful.
+ *
+ * Returns 0 on success, -1 on failure (no memory).
+ */
+int ben_dict_set(struct bencode *d, struct bencode *key, struct bencode *value);
+
+/* Same as ben_dict_set(), but the key is a C string */
+int ben_dict_set_by_str(struct bencode *d, const char *key, struct bencode *value);
+
+/* Same as ben_dict_set(), but the key and value are C strings */
+int ben_dict_set_str_by_str(struct bencode *d, const char *key, const char *value);
+
+struct bencode *ben_int(long long ll);
+
+/* Create an empty list */
+struct bencode *ben_list(void);
+
+/*
+ * Append 'b' to 'list'. Returns 0 on success, -1 on failure (no memory).
+ * One may not call ben_free(b) after a successful call, because the list owns
+ * the object 'b'.
+ */
+int ben_list_append(struct bencode *list, struct bencode *b);
+
+int ben_list_append_str(struct bencode *list, const char *s);
+int ben_list_append_int(struct bencode *list, long long ll);
+
+/* Remove and return value at position 'pos' in list */
+struct bencode *ben_list_pop(struct bencode *list, size_t pos);
+
+/*
+ * Returns a Python formatted C string representation of 'b' on success,
+ * NULL on failure. The returned string should be freed with free().
+ *
+ * Note: The string is terminated with '\0'. All instances of '\0' bytes in
+ * the bencoded data are escaped so that there is only one '\0' byte
+ * in the generated string at the end.
+ */
+char *ben_print(const struct bencode *b);
+
+/* Create a string from C string (note bencode string may contain '\0'. */
+struct bencode *ben_str(const char *s);
+
+/* Return a human readable explanation of error returned with ben_decode2() */
+const char *ben_strerror(int error);
+
+/*
+ * Unpack a Bencoded structure similar to scanf(). Takes a format string and
+ * a list of pointers as variable arguments. The given b structure is checked
+ * against the format and values are unpacked using the given specifiers.
+ * A specifier begins with a percent (%) that follows a string of specifier
+ * characters documented below.
+ * The syntax is similar to Python format for recursive data structures, and
+ * consists of tokens {, }, [, ] with any number of spaces between them.
+ * The keys of a dictionary are given as literal strings or integers and
+ * matched against the keys of the Bencoded structure.
+ *
+ * Unpack modifiers:
+ *     l    The integer is of type long or unsigned long, and the type of the
+ *          argument is expected to be long * or unsigned long *.
+ *     ll   The integer is a long long or an unsigned long long, and the
+ *          argument is long long * or unsigned long long *.
+ *     L    Same as ll.
+ *     q    Same as ll.
+ *
+ * Unpack specifiers:
+ *     %ps  The Bencode value must be a string and a pointer to a string
+ *          (char **) is expected to be given as arguments. Note, returns a
+ *          reference to the internal string buffer. The returned memory should
+ *          not be freed and it has the same life time as the Bencode string.
+ *
+ *     %pb  Takes any structure and writes a pointer given as an argument.
+ *          The argument is expected to be "struct bencode **". Note, returns a
+ *          reference to the value inside the structure passed to ben_unpack().
+ *          The returned memory should not be freed and it has the same life
+ *          time as the original structure.
+ *
+ *     %d   The bencode value is expected to be a (signed) integer. The
+ *          preceeding conversion modifiers define the type of the given
+ *          pointer.
+
+ *     %u   The bencode value is expected to be an unsigned integer. The
+ *          preceeding conversion modifiers define the type of the given
+ *          pointer.
+ */
+int ben_unpack(const struct bencode *b, const char *fmt, ...)
+	BEN_CHECK_FORMAT(scanf, 2, 3);
+
+int ben_unpack2(const struct bencode *b, size_t *off, struct bencode_error *error, const char *fmt, ...)
+	BEN_CHECK_FORMAT(scanf, 4, 5);
+
+/*
+ * Pack a Bencoded structure similar to printf(). Takes a format string and
+ * a list of values as variable arguments.
+ * Works similarly to ben_decode_printed(), but allows the string to values
+ * specifiers which are replaced with values given as arguments.
+ * A specifier begins with a percent (%) that follows a string of specifier
+ * characters documented below.
+ *
+ * Value modifiers:
+ *     l    The integer is of type long or unsigned long.
+ *     ll   The integer is a long long or an unsigned long long.
+ *     L    Same as ll.
+ *     q    Same as ll.
+ *
+ * Value specifiers:
+ *     %s   A string pointer (char *) expected to be given as argument. A new
+ *          Bencode string is constructed from the given string.
+ *
+ *     %pb  A Bencode structure (struct bencode *) is expected to be given as
+ *          argument. Note, takes ownership of the structure, even when an
+ *          error is returned.
+ *
+ *     %d   Constructs a new integer from the given (signed) integer. The
+ *          preceeding conversion modifiers define the type of the value.
+ *
+ *     %u   Constructs a new integer from the given unsigned integer. The
+ *          preceeding conversion modifiers define the type of the value.
+ */
+struct bencode *ben_pack(const char *fmt, ...)
+	BEN_CHECK_FORMAT(printf, 1, 2);
+
+/* ben_is_bool() returns 1 iff b is a boolean, 0 otherwise */
+static inline int ben_is_bool(const struct bencode *b)
+{
+	return b->type == BENCODE_BOOL;
+}
+static inline int ben_is_dict(const struct bencode *b)
+{
+	return b->type == BENCODE_DICT;
+}
+static inline int ben_is_int(const struct bencode *b)
+{
+	return b->type == BENCODE_INT;
+}
+static inline int ben_is_list(const struct bencode *b)
+{
+	return b->type == BENCODE_LIST;
+}
+static inline int ben_is_str(const struct bencode *b)
+{
+	return b->type == BENCODE_STR;
+}
+static inline int ben_is_user(const struct bencode *b)
+{
+	return b->type == BENCODE_USER;
+}
+
+/*
+ * ben_bool_const_cast(b) returns "(const struct bencode_bool *) b" if the
+ * underlying object is a boolean, NULL otherwise.
+ */
+static inline const struct bencode_bool *ben_bool_const_cast(const struct bencode *b)
+{
+	return b->type == BENCODE_BOOL ? ((const struct bencode_bool *) b) : NULL;
+}
+
+/*
+ * ben_bool_cast(b) returns "(struct bencode_bool *) b" if the
+ * underlying object is a boolean, NULL otherwise.
+ */
+static inline struct bencode_bool *ben_bool_cast(struct bencode *b)
+{
+	return b->type == BENCODE_BOOL ? ((struct bencode_bool *) b) : NULL;
+}
+
+static inline const struct bencode_dict *ben_dict_const_cast(const struct bencode *b)
+{
+	return b->type == BENCODE_DICT ? ((const struct bencode_dict *) b) : NULL;
+}
+static inline struct bencode_dict *ben_dict_cast(struct bencode *b)
+{
+	return b->type == BENCODE_DICT ? ((struct bencode_dict *) b) : NULL;
+}
+
+static inline const struct bencode_int *ben_int_const_cast(const struct bencode *i)
+{
+	return i->type == BENCODE_INT ? ((const struct bencode_int *) i) : NULL;
+}
+static inline struct bencode_int *ben_int_cast(struct bencode *i)
+{
+	return i->type == BENCODE_INT ? ((struct bencode_int *) i) : NULL;
+}
+
+static inline const struct bencode_list *ben_list_const_cast(const struct bencode *list)
+{
+	return list->type == BENCODE_LIST ? ((const struct bencode_list *) list) : NULL;
+}
+static inline struct bencode_list *ben_list_cast(struct bencode *list)
+{
+	return list->type == BENCODE_LIST ? ((struct bencode_list *) list) : NULL;
+}
+
+static inline const struct bencode_str *ben_str_const_cast(const struct bencode *str)
+{
+	return str->type == BENCODE_STR ? ((const struct bencode_str *) str) : NULL;
+}
+static inline struct bencode_str *ben_str_cast(struct bencode *str)
+{
+	return str->type == BENCODE_STR ? ((struct bencode_str *) str) : NULL;
+}
+
+static inline const struct bencode_user *ben_user_const_cast(const struct bencode *user)
+{
+	return user->type == BENCODE_USER ? ((const struct bencode_user *) user) : NULL;
+}
+static inline struct bencode_user *ben_user_cast(struct bencode *user)
+{
+	return user->type == BENCODE_USER ? ((struct bencode_user *) user) : NULL;
+}
+
+static inline int ben_is_user_type(const struct bencode *b, struct bencode_type *type)
+{
+	return b->type == BENCODE_USER ? ((const struct bencode_user *) b)->info == type : 0;
+}
+
+static inline const void *ben_user_type_const_cast(const struct bencode *b, struct bencode_type *type)
+{
+	return (b->type == BENCODE_USER && ((const struct bencode_user *) b)->info == type) ? b : NULL;
+}
+static inline void *ben_user_type_cast(struct bencode *b, struct bencode_type *type)
+{
+	return (b->type == BENCODE_USER && ((const struct bencode_user *) b)->info == type) ? b : NULL;
+}
+
+/* Return the number of keys in a dictionary 'b' */
+static inline size_t ben_dict_len(const struct bencode *b)
+{
+	return ben_dict_const_cast(b)->n;
+}
+
+/* Return the number of items in a list 'b' */
+static inline size_t ben_list_len(const struct bencode *b)
+{
+	return ben_list_const_cast(b)->n;
+}
+
+/* ben_list_get(list, i) returns object at position i in list */
+static inline struct bencode *ben_list_get(const struct bencode *list, size_t i)
+{
+	const struct bencode_list *l = ben_list_const_cast(list);
+	if (i >= l->n) {
+		fprintf(stderr, "bencode: List index out of bounds\n");
+		abort();
+	}
+	return l->values[i];
+}
+
+/*
+ * ben_list_set(list, i, b) sets object b to list at position i.
+ * The old value at position i is freed.
+ * The program aborts if position i is out of bounds.
+ */
+void ben_list_set(struct bencode *list, size_t i, struct bencode *b);
+
+/* Return the number of bytes in a string 'b' */
+static inline size_t ben_str_len(const struct bencode *b)
+{
+	return ben_str_const_cast(b)->len;
+}
+
+/* Return boolean value (0 or 1) of 'b' */
+static inline int ben_bool_val(const struct bencode *b)
+{
+	return ben_bool_const_cast(b)->b ? 1 : 0;
+}
+
+/* Return integer value of 'b' */
+static inline long long ben_int_val(const struct bencode *b)
+{
+	return ben_int_const_cast(b)->ll;
+}
+
+/*
+ * Note: the string is always zero terminated. Also, the string may
+ * contain more than one zero.
+ * bencode strings are not compatible with C strings.
+ */
+static inline const char *ben_str_val(const struct bencode *b)
+{
+	return ben_str_const_cast(b)->s;
+}
+
+/*
+ * ben_list_for_each() is an iterator macro for bencoded lists.
+ *
+ * Note, it is not allowed to change the list while iterating except by
+ * using ben_list_pop_current().
+ *
+ * pos is a size_t.
+ *
+ * Example:
+ *
+ * size_t pos;
+ * struct bencode *list = xxx;
+ * struct bencode *value;
+ * ben_list_for_each(value, pos, list) {
+ *         inspect(value);
+ * }
+ */
+#define ben_list_for_each(value, pos, l) \
+	for ((pos) = (size_t) 0;		    \
+	     (pos) < (ben_list_const_cast(l))->n && \
+	     ((value) = ((const struct bencode_list *) (l))->values[(pos)]) != NULL ; \
+	     (pos)++)
+
+/*
+ * ben_list_pop_current() returns and removes the current item at 'pos'
+ * while iterating the list with ben_list_for_each().
+ * It can be used more than once per walk, but only once per item.
+ * Example below:
+ *
+ * Filter out all items from list whose string value does not begin with "foo".
+ *
+ * ben_list_for_each(value, pos, list) {
+ *         if (strncmp(ben_str_val(value), "foo", 3) != 0)
+ *                 ben_free(ben_list_pop_current(&pos, list));
+ * }
+ */
+static inline struct bencode *ben_list_pop_current(struct bencode *list,
+						   size_t *pos)
+{
+	struct bencode *value = ben_list_pop(list, *pos);
+	(*pos)--;
+	return value;
+}
+
+/*
+ * ben_dict_for_each() is an iterator macro for bencoded dictionaries.
+ *
+ * Note, it is not allowed to change the dictionary while iterating except
+ * by using ben_dict_pop_current().
+ *
+ * struct bencode *dict = ben_dict();
+ * size_t pos;
+ * struct bencode *key;
+ * struct bencode *value;
+ * ben_dict_set_str_by_str(dict, "foo", "bar");
+ *
+ * ben_dict_for_each(key, value, pos, dict) {
+ *     use(key, value);
+ * }
+ *
+ * pos is a size_t.
+ */
+#define ben_dict_for_each(bkey, bvalue, pos, d) \
+	for ((pos) = 0; \
+	     (pos) < (ben_dict_const_cast(d))->n && \
+	     ((bkey) = ((const struct bencode_dict *) (d))->nodes[(pos)].key) != NULL && \
+	     ((bvalue) = ((const struct bencode_dict *) (d))->nodes[(pos)].value) != NULL; \
+	     (pos)++)
+
+/*
+ * ben_dict_pop_current() deletes the current item at 'pos' while iterating
+ * the dictionary with ben_dict_for_each(). It can be used more than once
+ * per walk, but only once per item. Example below:
+ *
+ * Filter out all items from dictionary whose key does not begin with "foo".
+ *
+ * ben_dict_for_each(key, value, pos, dict) {
+ *     if (strncmp(ben_str_val(key), "foo", 3) != 0)
+ *         ben_free(ben_dict_pop_current(dict, &pos));
+ * }
+ */
+struct bencode *ben_dict_pop_current(struct bencode *dict, size_t *pos);
+
+/* Report an error while decoding. Returns NULL. */
+void *ben_insufficient_ptr(struct ben_decode_ctx *ctx);
+void *ben_invalid_ptr(struct ben_decode_ctx *ctx);
+void *ben_oom_ptr(struct ben_decode_ctx *ctx);
+
+/*
+ * Decode from the current position of 'ctx'.
+ *
+ * This function is used to implement decoders for user-defined types.
+ */
+struct bencode *ben_ctx_decode(struct ben_decode_ctx *ctx);
+
+/*
+ * Test whether the input of 'ctx' has at least n bytes left.
+ * Returns 0 when there is enough bytes left and -1 when there isn't.
+ *
+ * This function is used to implement decoders for user-defined types.
+ */
+int ben_need_bytes(const struct ben_decode_ctx *ctx, size_t n);
+
+/*
+ * Returns the character in current position of 'ctx'.
+ *
+ * This function is used to implement decoders for user-defined types.
+ */
+char ben_current_char(const struct ben_decode_ctx *ctx);
+
+/*
+ * Get the next n bytes from input.
+ * Returns pointer to the data or NULL when there aren't enough bytes left.
+ *
+ * This function is used to implement decoders for user-defined types.
+ */
+const char *ben_current_buf(const struct ben_decode_ctx *ctx, size_t n);
+
+/*
+ * Increments current position by n.
+ *
+ * This function is used to implement decoders for user-defined types.
+ */
+void ben_skip(struct ben_decode_ctx *ctx, size_t n);
+
+/*
+ * Encode to the output of 'ctx'. The size of the encoded data can be obtained
+ * with ben_encoded_size().
+ *
+ * This function is used to implement encoders for user-defined types.
+ */
+int ben_ctx_encode(struct ben_encode_ctx *ctx, const struct bencode *b);
+
+/*
+ * Append one character to output of 'ctx'. The amount of bytes written to the
+ * output must be the same as returned by get_size().
+ *
+ * This function is used to implement encoders for user-defined types.
+ */
+int ben_put_char(struct ben_encode_ctx *ctx, char c);
+
+/*
+ * Append data to output of 'ctx'. The amount of bytes written to the output
+ * must be the same as returned by get_size().
+ *
+ * This function is used to implement encoders for user-defined types.
+ */
+int ben_put_buffer(struct ben_encode_ctx *ctx, const void *buf, size_t len);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

libs/libblade/src/include/ks_dht-int.h

@@ -0,0 +1,447 @@
+#ifndef KS_DHT_INT_H
+#define KS_DHT_INT_H
+
+#include "ks.h"
+
+KS_BEGIN_EXTERN_C
+
+/**
+ * Determines the appropriate endpoint to reach a remote address.
+ * If an endpoint is provided, nothing more needs to be done.
+ * If no endpoint is provided, first it will check for an active endpoint it can route though.
+ * If no active endpoint is available and autorouting is enabled it will attempt to bind a usable endpoint.
+ * @param dht pointer to the dht instance
+ * @param raddr pointer to the remote address
+ * @param endpoint dereferenced in/out pointer to the endpoint, if populated then returns immediately
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, ...
+ * @see ks_ip_route
+ * @see ks_hash_read_unlock
+ * @see ks_addr_set
+ * @see ks_dht_bind
+ */
+KS_DECLARE(ks_status_t) ks_dht_autoroute_check(ks_dht_t *dht, const ks_sockaddr_t *raddr, ks_dht_endpoint_t **endpoint);
+
+/**
+ * Called internally to receive datagrams from endpoints.
+ * Handles datagrams by dispatching each through a threadpool.
+ * @param dht pointer to the dht instance
+ * @param timeout time in ms to wait for an incoming datagram on any endpoint
+ */
+KS_DECLARE(void) ks_dht_pulse_endpoints(ks_dht_t *dht, int32_t timeout);
+
+/**
+ * Called internally to expire or reannounce storage item data.
+ * Handles reannouncing and purging of expiring storage items.
+ * @param dht pointer to the dht instance
+ */
+KS_DECLARE(void) ks_dht_pulse_storageitems(ks_dht_t *dht);
+
+/**
+ * Called internally to process job state machine.
+ * Handles completing and purging of finished jobs.
+ * @param dht pointer to the dht instance
+ */
+KS_DECLARE(void) ks_dht_pulse_jobs(ks_dht_t *dht);
+
+/**
+ * Called internally to send queued messages.
+ * Handles throttling of message sending to ensure system buffers are not overloaded and messages are not dropped.
+ * @param dht pointer to the dht instance
+ */
+KS_DECLARE(void) ks_dht_pulse_send(ks_dht_t *dht);
+
+/**
+ * Called internally to expire transactions.
+ * Handles purging of expired and finished transactions.
+ * @param dht pointer to the dht instance
+ */
+KS_DECLARE(void) ks_dht_pulse_transactions(ks_dht_t *dht);
+
+/**
+ * Called internally to expire and cycle tokens.
+ * Handles cycling new secret entropy for token generation.
+ * @param dht pointer to the dht instance
+ */
+KS_DECLARE(void) ks_dht_pulse_tokens(ks_dht_t *dht);
+
+/**
+ * Converts a ks_dht_nodeid_t into it's hex string representation.
+ * @param id pointer to the nodeid
+ * @param buffer pointer to the buffer able to contain at least (KS_DHT_NODEID_SIZE * 2) + 1 characters
+ * @return The pointer to the front of the populated string buffer
+ */
+KS_DECLARE(char *) ks_dht_hex(const uint8_t *data, char *buffer, ks_size_t len);
+
+/**
+ * Compacts address information as per the DHT specifications.
+ * @param address pointer to the address being compacted from
+ * @param buffer pointer to the buffer containing compacted data
+ * @param buffer_length pointer to the buffer length consumed
+ * @param buffer_size max size of the buffer
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_NO_MEM
+ */
+KS_DECLARE(ks_status_t) ks_dht_utility_compact_addressinfo(const ks_sockaddr_t *address,
+														   uint8_t *buffer,
+														   ks_size_t *buffer_length,
+														   ks_size_t buffer_size);
+
+/**
+ * Expands address information as per the DHT specifications.
+ * @param buffer pointer to the buffer containing compacted data
+ * @param buffer_length pointer to the buffer length consumed
+ * @param buffer_size max size of the buffer
+ * @param address pointer to the address being expanded into
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_NO_MEM, ...
+ * @see ks_addr_set_raw
+ */
+KS_DECLARE(ks_status_t) ks_dht_utility_expand_addressinfo(const uint8_t *buffer,
+														  ks_size_t *buffer_length,
+														  ks_size_t buffer_size,
+														  ks_sockaddr_t *address);
+
+/**
+ * Compacts node information as per the DHT specifications.
+ * Compacts address information after the nodeid.
+ * @param nodeid pointer to the nodeid being compacted from
+ * @param address pointer to the address being compacted from
+ * @param buffer pointer to the buffer containing compacted data
+ * @param buffer_length pointer to the buffer length consumed
+ * @param buffer_size max size of the buffer
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_NO_MEM, ...
+ * @see ks_dht_utility_compact_addressinfo
+ */
+KS_DECLARE(ks_status_t) ks_dht_utility_compact_nodeinfo(const ks_dht_nodeid_t *nodeid,
+														const ks_sockaddr_t *address,
+														uint8_t *buffer,
+														ks_size_t *buffer_length,
+														ks_size_t buffer_size);
+
+/**
+ * Expands address information as per the DHT specifications.
+ * Expands compacted address information after the nodeid.
+ * @param buffer pointer to the buffer containing compacted data
+ * @param buffer_length pointer to the buffer length consumed
+ * @param buffer_size max size of the buffer
+ * @param address pointer to the address being expanded into
+ * @param nodeid pointer to the nodeid being expanded into
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_NO_MEM, ...
+ * @see ks_dht_utility_expand_addressinfo
+ */
+KS_DECLARE(ks_status_t) ks_dht_utility_expand_nodeinfo(const uint8_t *buffer,
+													   ks_size_t *buffer_length,
+													   ks_size_t buffer_size,
+													   ks_dht_nodeid_t *nodeid,
+													   ks_sockaddr_t *address);
+
+/**
+ * Extracts a ks_dht_nodeid_t from a bencode dictionary given a string key.
+ * @param args pointer to the bencode dictionary
+ * @param key string key in the bencode dictionary to extract the value from
+ * @param nodeid dereferenced out pointer to the nodeid
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_ARG_INVALID
+ */
+KS_DECLARE(ks_status_t) ks_dht_utility_extract_nodeid(struct bencode *args, const char *key, ks_dht_nodeid_t **nodeid);
+
+/**
+ * Extracts a ks_dht_token_t from a bencode dictionary given a string key.
+ * @param args pointer to the bencode dictionary
+ * @param key string key in the bencode dictionary to extract the value from
+ * @param nodeid dereferenced out pointer to the token
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_ARG_INVALID
+ */
+KS_DECLARE(ks_status_t) ks_dht_utility_extract_token(struct bencode *args, const char *key, ks_dht_token_t **token);
+
+/**
+ * Generates an opaque write token based on a shifting secret value, the remote address and target nodeid of interest.
+ * This token ensures that future operations can be verified to the remote peer and target id requested.
+ * @param secret rotating secret portion of the token hash
+ * @param raddr pointer to the remote address used for the ip and port in the token hash
+ * @param target pointer to the nodeid of the target used for the token hash
+ * @param token pointer to the output token being generated
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_FAIL
+ */
+KS_DECLARE(ks_status_t) ks_dht_token_generate(uint32_t secret, const ks_sockaddr_t *raddr, ks_dht_nodeid_t *target, ks_dht_token_t *token);
+
+/**
+ * Verify an opaque write token matches the provided remote address and target nodeid.
+ * Handles checking against the last two secret values for the token hash.
+ * @param dht pointer to the dht instance
+ * @param raddr pointer to the remote address used for the ip and port in the token hash
+ * @param target pointer to the nodeid of the target used for the token hash
+ * @param token pointer to the input token being compared
+ * @return Either KS_TRUE if verification passes, otherwise KS_FALSE
+ */
+KS_DECLARE(ks_bool_t) ks_dht_token_verify(ks_dht_t *dht, const ks_sockaddr_t *raddr, ks_dht_nodeid_t *target, ks_dht_token_t *token);
+
+/**
+ * Encodes a message for transmission as a UDP datagram and sends it.
+ * Uses the internally tracked local endpoint and remote address to route the UDP datagram.
+ * @param dht pointer to the dht instance
+ * @param message pointer to the message being sent
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, ...
+ * @see ks_socket_sendto
+ */
+KS_DECLARE(ks_status_t) ks_dht_send(ks_dht_t *dht, ks_dht_message_t *message);
+
+/**
+ * Sets up the common parts of a query message.
+ * Determines the local endpoint aware of autorouting, assigns the remote address, generates a transaction, and queues a callback.
+ * @param dht pointer to the dht instance
+ * @param job pointer to the job
+ * @param query string value of the query type, for example "ping"
+ * @param callback callback to be called when response to transaction is received
+ * @param transaction dereferenced out pointer to the allocated transaction, may be NULL to ignore output
+ * @param message dereferenced out pointer to the allocated message
+ * @param args dereferenced out pointer to the allocated bencode args, may be NULL to ignore output
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_FAIL, ...
+ * @see ks_dht_autoroute_check
+ * @see ks_dht_transaction_alloc
+ * @see ks_dht_transaction_init
+ * @see ks_dht_message_alloc
+ * @see ks_dht_message_init
+ * @see ks_dht_message_query
+ * @see ks_hash_insert
+ */
+KS_DECLARE(ks_status_t) ks_dht_query_setup(ks_dht_t *dht,
+										   ks_dht_job_t *job,
+										   const char *query,
+										   ks_dht_job_callback_t callback,
+										   ks_dht_transaction_t **transaction,
+										   ks_dht_message_t **message,
+										   struct bencode **args);
+
+/**
+ * Sets up the common parts of a response message.
+ * Determines the local endpoint aware of autorouting, assigns the remote address, and assigns the transaction.
+ * @param dht pointer to the dht instance
+ * @param ep pointer to the endpoint, may be NULL to find an endpoint or autoroute one
+ * @param raddr pointer to the remote address
+ * @param transactionid pointer to the buffer containing the transactionid, may be of variable size depending on the querying node
+ * @param transactionid_length length of the transactionid buffer
+ * @param message dereferenced out pointer to the allocated message
+ * @param args dereferenced out pointer to the allocated bencode args, may be NULL to ignore output
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, ...
+ * @see ks_dht_autoroute_check
+ * @see ks_dht_message_alloc
+ * @see ks_dht_message_init
+ * @see ks_dht_message_response
+ */
+KS_DECLARE(ks_status_t) ks_dht_response_setup(ks_dht_t *dht,
+											  ks_dht_endpoint_t *ep,
+											  const ks_sockaddr_t *raddr,
+											  uint8_t *transactionid,
+											  ks_size_t transactionid_length,
+											  ks_dht_message_t **message,
+											  struct bencode **args);
+												
+						
+KS_DECLARE(ks_status_t) ks_dht_error(ks_dht_t *dht,
+									 ks_dht_endpoint_t *ep,
+									 const ks_sockaddr_t *raddr,
+									 uint8_t *transactionid,
+									 ks_size_t transactionid_length,
+									 long long errorcode,
+									 const char *errorstr);
+
+KS_DECLARE(void) ks_dht_pulse_jobs(ks_dht_t *dht);
+KS_DECLARE(ks_status_t) ks_dht_query_ping(ks_dht_t *dht, ks_dht_job_t *job);
+KS_DECLARE(ks_status_t) ks_dht_query_findnode(ks_dht_t *dht, ks_dht_job_t *job);
+KS_DECLARE(ks_status_t) ks_dht_query_get(ks_dht_t *dht, ks_dht_job_t *job);
+KS_DECLARE(ks_status_t) ks_dht_query_put(ks_dht_t *dht, ks_dht_job_t *job);
+
+KS_DECLARE(void *)ks_dht_process(ks_thread_t *thread, void *data);
+
+KS_DECLARE(ks_status_t) ks_dht_process_query(ks_dht_t *dht, ks_dht_message_t *message);
+KS_DECLARE(ks_status_t) ks_dht_process_response(ks_dht_t *dht, ks_dht_message_t *message);
+KS_DECLARE(ks_status_t) ks_dht_process_error(ks_dht_t *dht, ks_dht_message_t *message);
+
+KS_DECLARE(ks_status_t) ks_dht_process_query_ping(ks_dht_t *dht, ks_dht_message_t *message);
+KS_DECLARE(ks_status_t) ks_dht_process_response_ping(ks_dht_t *dht, ks_dht_job_t *job);
+
+KS_DECLARE(ks_status_t) ks_dht_process_query_findnode(ks_dht_t *dht, ks_dht_message_t *message);
+KS_DECLARE(ks_status_t) ks_dht_process_response_findnode(ks_dht_t *dht, ks_dht_job_t *job);
+
+KS_DECLARE(ks_status_t) ks_dht_process_query_get(ks_dht_t *dht, ks_dht_message_t *message);
+KS_DECLARE(ks_status_t) ks_dht_process_response_get(ks_dht_t *dht, ks_dht_job_t *job);
+
+KS_DECLARE(ks_status_t) ks_dht_process_query_put(ks_dht_t *dht, ks_dht_message_t *message);
+KS_DECLARE(ks_status_t) ks_dht_process_response_put(ks_dht_t *dht, ks_dht_job_t *job);
+
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_datagram_create(ks_dht_datagram_t **datagram,
+											   ks_pool_t *pool,
+											   ks_dht_t *dht,
+											   ks_dht_endpoint_t *endpoint,
+											   const ks_sockaddr_t *raddr);
+KS_DECLARE(void) ks_dht_datagram_destroy(ks_dht_datagram_t **datagram);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_job_create(ks_dht_job_t **job,
+										  ks_pool_t *pool,
+										  const ks_sockaddr_t *raddr,
+										  int32_t attempts,
+										  void *data);
+KS_DECLARE(void) ks_dht_job_build_ping(ks_dht_job_t *job, ks_dht_job_callback_t query_callback, ks_dht_job_callback_t finish_callback);
+KS_DECLARE(void) ks_dht_job_build_findnode(ks_dht_job_t *job,
+										   ks_dht_job_callback_t query_callback,
+										   ks_dht_job_callback_t finish_callback,
+										   ks_dht_nodeid_t *target);
+KS_DECLARE(void) ks_dht_job_build_get(ks_dht_job_t *job,
+									  ks_dht_job_callback_t query_callback,
+									  ks_dht_job_callback_t finish_callback,
+									  ks_dht_nodeid_t *target,
+									  const uint8_t *salt,
+									  ks_size_t salt_length);
+KS_DECLARE(void) ks_dht_job_build_put(ks_dht_job_t *job,
+									  ks_dht_job_callback_t query_callback,
+									  ks_dht_job_callback_t finish_callback,
+									  ks_dht_token_t *token,
+									  int64_t cas,
+									  ks_dht_storageitem_t *item);
+KS_DECLARE(void) ks_dht_job_build_search(ks_dht_job_t *job,
+										 ks_dht_job_callback_t query_callback,
+										 ks_dht_job_callback_t finish_callback);
+KS_DECLARE(void) ks_dht_job_build_search_findnode(ks_dht_job_t *job,
+									ks_dht_nodeid_t *target,
+									uint32_t family,
+									ks_dht_job_callback_t query_callback,
+									ks_dht_job_callback_t finish_callback);
+KS_DECLARE(void) ks_dht_job_destroy(ks_dht_job_t **job);
+
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_endpoint_create(ks_dht_endpoint_t **endpoint,
+											   ks_pool_t *pool,
+											   const ks_sockaddr_t *addr,
+											   ks_socket_t sock);
+KS_DECLARE(void) ks_dht_endpoint_destroy(ks_dht_endpoint_t **endpoint);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_message_create(ks_dht_message_t **message,
+											  ks_pool_t *pool,
+											  ks_dht_endpoint_t *endpoint,
+											  const ks_sockaddr_t *raddr,
+											  ks_bool_t alloc_data);
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_message_destroy(ks_dht_message_t **message);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_message_parse(ks_dht_message_t *message, const uint8_t *buffer, ks_size_t buffer_length);
+
+																								
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_search_create(ks_dht_search_t **search,
+											 ks_pool_t *pool,
+											 ks_dhtrt_routetable_t *table,
+											 const ks_dht_nodeid_t *target,
+											 ks_dht_job_callback_t callback,
+											 void *data);
+KS_DECLARE(void) ks_dht_search_destroy(ks_dht_search_t **search);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_publish_create(ks_dht_publish_t **publish,
+											  ks_pool_t *pool,
+											  ks_dht_job_callback_t callback,
+											  void *data,
+											  int64_t cas,
+											  ks_dht_storageitem_t *item);
+KS_DECLARE(void) ks_dht_publish_destroy(ks_dht_publish_t **publish);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_distribute_create(ks_dht_distribute_t **distribute,
+												 ks_pool_t *pool,
+												 ks_dht_storageitem_callback_t callback,
+												 void *data,
+												 int64_t cas,
+												 ks_dht_storageitem_t *item);
+KS_DECLARE(void) ks_dht_distribute_destroy(ks_dht_distribute_t **distribute);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_storageitem_target_immutable_internal(struct bencode *value, ks_dht_nodeid_t *target);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_storageitem_target_mutable_internal(ks_dht_storageitem_pkey_t *pk, struct bencode *salt, ks_dht_nodeid_t *target);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_storageitem_create_immutable_internal(ks_dht_storageitem_t **item,
+																	 ks_pool_t *pool,
+																	 ks_dht_nodeid_t *target,
+																	 struct bencode *v,
+																	 ks_bool_t clone_v);
+KS_DECLARE(ks_status_t) ks_dht_storageitem_create_immutable(ks_dht_storageitem_t **item,
+															ks_pool_t *pool,
+															ks_dht_nodeid_t *target,
+															const uint8_t *value,
+															ks_size_t value_length);
+KS_DECLARE(ks_status_t) ks_dht_storageitem_create_mutable_internal(ks_dht_storageitem_t **item,
+																   ks_pool_t *pool,
+																   ks_dht_nodeid_t *target,
+																   struct bencode *v,
+																   ks_bool_t clone_v,
+																   ks_dht_storageitem_pkey_t *pk,
+																   struct bencode *salt,
+																   ks_bool_t clone_salt,
+																   int64_t sequence,
+																   ks_dht_storageitem_signature_t *signature);
+KS_DECLARE(ks_status_t) ks_dht_storageitem_create_mutable(ks_dht_storageitem_t **item,
+														  ks_pool_t *pool,
+														  ks_dht_nodeid_t *target,
+														  const uint8_t *value,
+														  ks_size_t value_length,
+														  ks_dht_storageitem_pkey_t *pk,
+														  const uint8_t *salt,
+														  ks_size_t salt_length,
+														  int64_t sequence,
+														  ks_dht_storageitem_signature_t *signature);
+KS_DECLARE(void) ks_dht_storageitem_update_mutable(ks_dht_storageitem_t *item, struct bencode *v, int64_t sequence, ks_dht_storageitem_signature_t *signature);
+KS_DECLARE(void) ks_dht_storageitem_destroy(ks_dht_storageitem_t **item);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_transaction_create(ks_dht_transaction_t **transaction,
+												  ks_pool_t *pool,
+												  ks_dht_job_t *job,
+												  uint32_t transactionid,
+												  ks_dht_job_callback_t callback);
+KS_DECLARE(void) ks_dht_transaction_destroy(ks_dht_transaction_t **transaction);
+
+KS_END_EXTERN_C
+
+#endif /* KS_DHT_INT_H */
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/src/include/ks_dht.h

@@ -0,0 +1,596 @@
+#ifndef KS_DHT_H
+#define KS_DHT_H
+
+#include "ks.h"
+#include "ks_bencode.h"
+#include "sodium.h"
+
+KS_BEGIN_EXTERN_C
+
+
+#define KS_DHT_DEFAULT_PORT 5309
+
+#define KS_DHT_TPOOL_MIN 2
+#define KS_DHT_TPOOL_MAX 8
+#define KS_DHT_TPOOL_STACK (1024 * 256)
+#define KS_DHT_TPOOL_IDLE 10
+
+#define KS_DHT_DATAGRAM_BUFFER_SIZE 1000
+
+//#define KS_DHT_RECV_BUFFER_SIZE 0xFFFF
+
+#define KS_DHT_NODEID_SIZE 20
+
+#define KS_DHT_RESPONSE_NODES_MAX_SIZE 8
+
+#define KS_DHT_MESSAGE_TRANSACTIONID_MAX_SIZE 20
+#define KS_DHT_MESSAGE_TYPE_MAX_SIZE 20
+#define KS_DHT_MESSAGE_QUERY_MAX_SIZE 20
+#define KS_DHT_MESSAGE_ERROR_MAX_SIZE 256
+
+#define KS_DHT_TRANSACTION_EXPIRATION 10
+#define KS_DHT_TRANSACTIONS_PULSE 1
+
+#define KS_DHT_SEARCH_RESULTS_MAX_SIZE 8 // @todo replace with KS_DHTRT_BUCKET_SIZE
+
+#define KS_DHT_STORAGEITEM_PKEY_SIZE crypto_sign_PUBLICKEYBYTES
+#define KS_DHT_STORAGEITEM_SKEY_SIZE crypto_sign_SECRETKEYBYTES
+#define KS_DHT_STORAGEITEM_SALT_MAX_SIZE 64
+#define KS_DHT_STORAGEITEM_SIGNATURE_SIZE crypto_sign_BYTES
+#define KS_DHT_STORAGEITEM_EXPIRATION 7200
+#define KS_DHT_STORAGEITEM_KEEPALIVE 300
+#define KS_DHT_STORAGEITEMS_PULSE 10
+
+#define KS_DHT_TOKEN_SIZE SHA_DIGEST_LENGTH
+#define KS_DHT_TOKEN_EXPIRATION 300
+#define KS_DHT_TOKENS_PULSE 1
+
+#define  KS_DHTRT_MAXQUERYSIZE 20
+
+typedef struct ks_dht_s ks_dht_t;
+typedef struct ks_dht_datagram_s ks_dht_datagram_t;
+typedef struct ks_dht_job_s ks_dht_job_t;
+typedef struct ks_dht_nodeid_s ks_dht_nodeid_t;
+typedef struct ks_dht_token_s ks_dht_token_t;
+typedef struct ks_dht_storageitem_pkey_s ks_dht_storageitem_pkey_t;
+typedef struct ks_dht_storageitem_skey_s ks_dht_storageitem_skey_t;
+typedef struct ks_dht_storageitem_signature_s ks_dht_storageitem_signature_t;
+typedef struct ks_dht_message_s ks_dht_message_t;
+typedef struct ks_dht_endpoint_s ks_dht_endpoint_t;
+typedef struct ks_dht_transaction_s ks_dht_transaction_t;
+typedef struct ks_dht_search_s ks_dht_search_t;
+typedef struct ks_dht_publish_s ks_dht_publish_t;
+typedef struct ks_dht_distribute_s ks_dht_distribute_t;
+typedef struct ks_dht_node_s ks_dht_node_t;
+typedef struct ks_dhtrt_routetable_s ks_dhtrt_routetable_t;
+typedef struct ks_dhtrt_querynodes_s ks_dhtrt_querynodes_t;
+typedef struct ks_dht_storageitem_s ks_dht_storageitem_t;
+
+
+typedef ks_status_t (*ks_dht_job_callback_t)(ks_dht_t *dht, ks_dht_job_t *job);
+typedef ks_status_t (*ks_dht_message_callback_t)(ks_dht_t *dht, ks_dht_message_t *message);
+//typedef ks_status_t (*ks_dht_search_callback_t)(ks_dht_t *dht, ks_dht_search_t *search);
+typedef ks_status_t (*ks_dht_storageitem_callback_t)(ks_dht_t *dht, ks_dht_storageitem_t *item);
+
+
+struct ks_dht_datagram_s {
+	ks_pool_t *pool;
+	ks_dht_t *dht;
+	ks_dht_endpoint_t *endpoint;
+	ks_sockaddr_t raddr;
+	uint8_t buffer[KS_DHT_DATAGRAM_BUFFER_SIZE];
+	ks_size_t buffer_length;
+};
+
+/**
+ * Note: This must remain a structure for casting from raw data
+ */
+struct ks_dht_nodeid_s {
+	uint8_t id[KS_DHT_NODEID_SIZE];
+};
+
+enum ks_afflags_t { ifv4=AF_INET, ifv6=AF_INET6, ifboth=AF_INET+AF_INET6};
+enum ks_dht_nodetype_t { KS_DHT_REMOTE=0x01, 
+						 KS_DHT_LOCAL=0x02, 
+						 KS_DHT_BOTH=KS_DHT_REMOTE+KS_DHT_LOCAL };
+
+enum ks_create_node_flags_t { 
+						 KS_DHTRT_CREATE_DEFAULT=0,
+						 KS_DHTRT_CREATE_PING,
+						 KS_DHTRT_CREATE_TOUCH 
+};		
+				
+struct ks_dht_node_s {
+    ks_dht_nodeid_t  nodeid;
+    ks_sockaddr_t    addr;
+    enum ks_dht_nodetype_t type;               /* local or remote */
+    ks_dhtrt_routetable_t* table;
+    ks_rwl_t        *reflock;          
+};
+
+struct ks_dht_token_s {
+	uint8_t token[KS_DHT_TOKEN_SIZE];
+};
+
+enum ks_dht_job_state_t {
+	KS_DHT_JOB_STATE_QUERYING,
+	KS_DHT_JOB_STATE_RESPONDING,
+	KS_DHT_JOB_STATE_EXPIRING,
+	KS_DHT_JOB_STATE_COMPLETING,
+};
+
+enum ks_dht_job_result_t {
+	KS_DHT_JOB_RESULT_SUCCESS = 0,
+	KS_DHT_JOB_RESULT_EXPIRED,
+	KS_DHT_JOB_RESULT_ERROR,
+	KS_DHT_JOB_RESULT_FAILURE,
+};
+
+struct ks_dht_job_s {
+	ks_pool_t *pool;
+	ks_dht_t *dht;
+	ks_dht_job_t *next;
+
+	enum ks_dht_job_state_t state;
+	enum ks_dht_job_result_t result;
+
+	ks_sockaddr_t raddr; // will obtain local endpoint node id when creating message using raddr
+	int32_t attempts;
+
+	//enum ks_dht_job_type_t type;
+	ks_dht_job_callback_t query_callback;
+	ks_dht_job_callback_t finish_callback;
+
+	void *data;
+	ks_dht_message_t *response;
+
+	// job specific query parameters
+	ks_dht_nodeid_t query_target;
+	struct bencode *query_salt;
+	int64_t query_cas;
+	ks_dht_token_t query_token;
+	ks_dht_storageitem_t *query_storageitem;
+
+	// error response parameters
+	int64_t error_code;
+	struct bencode *error_description;
+	
+	// job specific response parameters
+	ks_dht_node_t *response_id;
+	ks_dht_node_t *response_nodes[KS_DHT_RESPONSE_NODES_MAX_SIZE];
+	ks_size_t response_nodes_count;
+	ks_dht_node_t *response_nodes6[KS_DHT_RESPONSE_NODES_MAX_SIZE];
+	ks_size_t response_nodes6_count;
+	
+	ks_dht_token_t response_token;
+	int64_t response_seq;
+	ks_bool_t response_hasitem;
+	ks_dht_storageitem_t *response_storageitem;
+};
+
+struct ks_dhtrt_routetable_s {
+    void*       internal;                       
+    ks_pool_t*  pool;                           
+    ks_logger_t logger;
+};
+
+struct ks_dhtrt_querynodes_s {
+    ks_dht_nodeid_t nodeid;                   /* in: id to query                   */
+    enum ks_afflags_t family;                 /* in: AF_INET or AF_INET6 or both   */
+    enum ks_dht_nodetype_t type;              /* remote, local, or  both           */
+    uint8_t        max;                       /* in: maximum to return             */
+    uint8_t        count;                     /* out: number returned              */
+    ks_dht_node_t* nodes[ KS_DHTRT_MAXQUERYSIZE ]; /* out: array of peers (ks_dht_node_t* nodes[incount]) */
+};
+
+struct ks_dht_storageitem_pkey_s {
+	uint8_t key[KS_DHT_STORAGEITEM_PKEY_SIZE];
+};
+
+struct ks_dht_storageitem_skey_s {
+	uint8_t key[KS_DHT_STORAGEITEM_SKEY_SIZE];
+};
+
+struct ks_dht_storageitem_signature_s {
+	uint8_t sig[KS_DHT_STORAGEITEM_SIGNATURE_SIZE];
+};
+
+struct ks_dht_message_s {
+	ks_pool_t *pool;
+	ks_dht_endpoint_t *endpoint;
+	ks_sockaddr_t raddr;
+	struct bencode *data;
+	uint8_t transactionid[KS_DHT_MESSAGE_TRANSACTIONID_MAX_SIZE];
+	ks_size_t transactionid_length;
+	ks_dht_transaction_t *transaction;
+	char type[KS_DHT_MESSAGE_TYPE_MAX_SIZE];
+	struct bencode *args;
+	ks_dht_nodeid_t args_id;
+};
+
+struct ks_dht_endpoint_s {
+	ks_pool_t *pool;
+	ks_sockaddr_t addr;
+	ks_socket_t sock;
+};
+
+struct ks_dht_transaction_s {
+	ks_pool_t *pool;
+	ks_dht_job_t *job;
+	uint32_t transactionid;
+	//ks_dht_nodeid_t target; // @todo look at moving this into job now
+	ks_dht_job_callback_t callback;
+	ks_time_t expiration;
+	ks_bool_t finished;
+};
+
+struct ks_dht_search_s {
+	ks_pool_t *pool;
+	ks_dhtrt_routetable_t *table;
+	ks_dht_nodeid_t target;
+	ks_dht_job_callback_t callback;
+	void *data;
+	ks_mutex_t *mutex;
+	ks_hash_t *searched;
+	int32_t searching;
+	ks_dht_node_t *results[KS_DHT_SEARCH_RESULTS_MAX_SIZE];
+	ks_dht_nodeid_t distances[KS_DHT_SEARCH_RESULTS_MAX_SIZE];
+	ks_size_t results_length;
+};
+
+struct ks_dht_publish_s {
+	ks_pool_t *pool;
+	ks_dht_job_callback_t callback;
+	void *data;
+	int64_t cas;
+	ks_dht_storageitem_t *item;
+};
+
+struct ks_dht_distribute_s {
+	ks_pool_t *pool;
+	ks_dht_storageitem_callback_t callback;
+	void *data;
+	ks_mutex_t *mutex;
+	int32_t publishing;
+	int64_t cas;
+	ks_dht_storageitem_t *item;
+};
+
+struct ks_dht_storageitem_s {
+	ks_pool_t *pool;
+	ks_dht_nodeid_t id;
+	ks_time_t expiration;
+	ks_time_t keepalive;
+	struct bencode *v;
+
+	ks_mutex_t *mutex;
+	volatile int32_t refc;
+	ks_dht_storageitem_callback_t callback;
+
+	ks_bool_t mutable;
+	ks_dht_storageitem_pkey_t pk;
+	ks_dht_storageitem_skey_t sk;
+	struct bencode *salt;
+	int64_t seq;
+	ks_dht_storageitem_signature_t sig;
+};
+
+struct ks_dht_s {
+	ks_pool_t *pool;
+	ks_bool_t pool_alloc;
+
+	ks_thread_pool_t *tpool;
+	ks_bool_t tpool_alloc;
+
+	ks_dht_nodeid_t nodeid;
+	// @todo make sure this node is unlocked, and never gets destroyed, should also never use local nodes in search results as they can be internal
+	// network addresses, not what others have contacted through
+	ks_dht_node_t *node;
+	
+	ks_bool_t autoroute;
+	ks_port_t autoroute_port;
+	
+	ks_hash_t *registry_type;
+	ks_hash_t *registry_query;
+	ks_hash_t *registry_error;
+
+	ks_dht_endpoint_t **endpoints;
+	int32_t endpoints_length;
+	int32_t endpoints_size;
+	ks_hash_t *endpoints_hash;
+	struct pollfd *endpoints_poll;
+
+	ks_q_t *send_q;
+	ks_dht_message_t *send_q_unsent;
+	uint8_t recv_buffer[KS_DHT_DATAGRAM_BUFFER_SIZE + 1]; // Add 1, if we receive it then overflow error
+	ks_size_t recv_buffer_length;
+
+	ks_mutex_t *jobs_mutex;
+	ks_dht_job_t *jobs_first;
+	ks_dht_job_t *jobs_last;
+
+	ks_time_t transactions_pulse;
+	ks_mutex_t *transactionid_mutex;
+	volatile uint32_t transactionid_next;
+	ks_hash_t *transactions_hash;
+
+	ks_dhtrt_routetable_t *rt_ipv4;
+	ks_dhtrt_routetable_t *rt_ipv6;
+
+	ks_time_t tokens_pulse;
+	volatile uint32_t token_secret_current;
+	volatile uint32_t token_secret_previous;
+	ks_time_t token_secret_expiration;
+
+	ks_time_t storageitems_pulse;
+	ks_hash_t *storageitems_hash;
+};
+
+/**
+ * Constructor function for ks_dht_t.
+ * Will allocate and initialize internal state including registration of message handlers.
+ * @param dht dereferenced out pointer to the allocated dht instance
+ * @param pool pointer to the memory pool used by the dht instance, may be NULL to create a new memory pool internally
+ * @param tpool pointer to a thread pool used by the dht instance, may be NULL to create a new thread pool internally
+ * @param nodeid pointer to the nodeid for this dht instance
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_NO_MEM
+ */
+KS_DECLARE(ks_status_t) ks_dht_create(ks_dht_t **dht, ks_pool_t *pool, ks_thread_pool_t *tpool, ks_dht_nodeid_t *nodeid);
+						
+/**
+ * Destructor function for ks_dht_t.
+ * Will deinitialize and deallocate internal state.
+ * @param dht dereferenced in/out pointer to the dht instance, NULL upon return
+ */
+KS_DECLARE(void) ks_dht_destroy(ks_dht_t **dht);
+
+/**
+ * Enable or disable (default) autorouting support.
+ * When enabled, autorouting will allow sending to remote addresses on interfaces which are not yet bound.
+ * The address will be bound with the provided autoroute port when this occurs.
+ * @param dht pointer to the dht instance
+ * @param autoroute enable or disable autorouting
+ * @param port when enabling autorouting this port will be used to bind new addresses, may be 0 to use the default DHT port
+ */
+KS_DECLARE(void) ks_dht_autoroute(ks_dht_t *dht, ks_bool_t autoroute, ks_port_t port);
+
+/**
+ * Register a callback for a specific message type.
+ * Will overwrite any duplicate handlers.
+ * @param dht pointer to the dht instance
+ * @param value string of the type text under the 'y' key of a message
+ * @param callback the callback to be called when a message matches
+ */
+KS_DECLARE(void) ks_dht_register_type(ks_dht_t *dht, const char *value, ks_dht_message_callback_t callback);
+
+/**
+ * Register a callback for a specific message query.
+ * Will overwrite any duplicate handlers.
+ * @param dht pointer to the dht instance
+ * @param value string of the type text under the 'q' key of a message
+ * @param callback the callback to be called when a message matches
+ */
+KS_DECLARE(void) ks_dht_register_query(ks_dht_t *dht, const char *value, ks_dht_message_callback_t callback);
+
+/**
+ * Register a callback for a specific message error.
+ * Will overwrite any duplicate handlers.
+ * @param dht pointer to the dht instance
+ * @param value string of the errorcode under the first item of the 'e' key of a message
+ * @param callback the callback to be called when a message matches
+ */
+KS_DECLARE(void) ks_dht_register_error(ks_dht_t *dht, const char *value, ks_dht_message_callback_t callback);
+
+/**
+ * Bind a local address and port for receiving UDP datagrams.
+ * @param dht pointer to the dht instance
+ * @param addr pointer to the local address information
+ * @param endpoint dereferenced out pointer to the allocated endpoint, may be NULL to ignore endpoint output
+ * @return The ks_status_t result: KS_STATUS_SUCCESS, KS_STATUS_FAIL, ...
+ * @see ks_socket_option
+ * @see ks_addr_bind
+ * @see ks_dht_endpoint_alloc
+ * @see ks_dht_endpoint_init
+ * @see ks_hash_insert
+ * @see ks_dhtrt_initroute
+ * @see ks_dhtrt_create_node
+ */
+KS_DECLARE(ks_status_t) ks_dht_bind(ks_dht_t *dht, const ks_sockaddr_t *addr, ks_dht_endpoint_t **endpoint);
+
+/**
+ * Pulse the internals of dht.
+ * Handles receiving UDP datagrams, dispatching processing, handles expirations, throttled message sending, route table pulsing, etc.
+ * @param dht pointer to the dht instance
+ * @param timeout timeout value used when polling sockets for new UDP datagrams
+ */
+KS_DECLARE(void) ks_dht_pulse(ks_dht_t *dht, int32_t timeout);
+
+
+KS_DECLARE(char *) ks_dht_hex(const uint8_t *data, char *buffer, ks_size_t len);
+KS_DECLARE(uint8_t *) ks_dht_dehex(uint8_t *data, const char *buffer, ks_size_t len);
+						
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_storageitem_target_immutable(const uint8_t *value, ks_size_t value_length, ks_dht_nodeid_t *target);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_storageitem_target_mutable(ks_dht_storageitem_pkey_t *pk, const uint8_t *salt, ks_size_t salt_length, ks_dht_nodeid_t *target);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_storageitem_signature_generate(ks_dht_storageitem_signature_t *sig,
+															  ks_dht_storageitem_skey_t *sk,
+															  const uint8_t *salt,
+															  ks_size_t salt_length,
+															  int64_t sequence,
+															  const uint8_t *value,
+															  ks_size_t value_length);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_storageitem_reference(ks_dht_storageitem_t *item);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_storageitem_dereference(ks_dht_storageitem_t *item);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_storageitem_callback(ks_dht_storageitem_t *item, ks_dht_storageitem_callback_t callback);
+						
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_storageitems_read_lock(ks_dht_t *dht);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_storageitems_read_unlock(ks_dht_t *dht);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_storageitems_write_lock(ks_dht_t *dht);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_storageitems_write_unlock(ks_dht_t *dht);
+
+/**
+ *
+ */
+KS_DECLARE(ks_dht_storageitem_t *) ks_dht_storageitems_find(ks_dht_t *dht, ks_dht_nodeid_t *target);
+
+/**
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dht_storageitems_insert(ks_dht_t *dht, ks_dht_storageitem_t *item);
+						
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_ping(ks_dht_t *dht, const ks_sockaddr_t *raddr, ks_dht_job_callback_t callback, void *data);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_findnode(ks_dht_t *dht,
+								 const ks_sockaddr_t *raddr,
+								 ks_dht_job_callback_t callback,
+								 void *data,
+								 ks_dht_nodeid_t *target);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_get(ks_dht_t *dht,
+							const ks_sockaddr_t *raddr,
+							ks_dht_job_callback_t callback,
+							void *data,
+							ks_dht_nodeid_t *target,
+							const uint8_t *salt,
+							ks_size_t salt_length);
+
+/**
+ *
+ */
+KS_DECLARE(void) ks_dht_put(ks_dht_t *dht,
+							const ks_sockaddr_t *raddr,
+							ks_dht_job_callback_t callback,
+							void *data,
+							ks_dht_token_t *token,
+							int64_t cas,
+							ks_dht_storageitem_t *item);
+						
+/**
+ * Create a network search of the closest nodes to a target.
+ * @param dht pointer to the dht instance
+ * @param family either AF_INET or AF_INET6 for the appropriate network to search
+ * @param target pointer to the nodeid for the target to be searched
+ * @param callback an optional callback to add to the search when it is finished
+ * @param search dereferenced out pointer to the allocated search, may be NULL to ignore search output
+ * @see ks_dht_search_create
+ * @see ks_hash_insert
+ * @see ks_dht_findnode
+ */
+KS_DECLARE(void) ks_dht_search(ks_dht_t *dht,
+							   ks_dht_job_callback_t callback,
+							   void *data,
+							   ks_dhtrt_routetable_t *table,
+							   ks_dht_nodeid_t *target);
+
+KS_DECLARE(void) ks_dht_publish(ks_dht_t *dht,
+								const ks_sockaddr_t *raddr,
+								ks_dht_job_callback_t callback,
+								void *data,
+								int64_t cas,
+								ks_dht_storageitem_t *item);
+
+KS_DECLARE(void) ks_dht_distribute(ks_dht_t *dht,
+								   ks_dht_storageitem_callback_t callback,
+								   void *data,
+								   ks_dhtrt_routetable_t *table,
+								   int64_t cas,
+								   ks_dht_storageitem_t *item);
+
+/**
+ * route table methods
+ *
+ */
+KS_DECLARE(ks_status_t) ks_dhtrt_initroute(ks_dhtrt_routetable_t **tableP, 
+											ks_dht_t *dht, 
+											ks_pool_t *pool); 
+KS_DECLARE(void) ks_dhtrt_deinitroute(ks_dhtrt_routetable_t **table);
+
+KS_DECLARE(ks_status_t)        ks_dhtrt_create_node(ks_dhtrt_routetable_t* table,
+													ks_dht_nodeid_t nodeid,
+													enum ks_dht_nodetype_t type,
+													char* ip, unsigned short port,
+                                                    enum ks_create_node_flags_t flags,
+													ks_dht_node_t** node);
+
+KS_DECLARE(ks_status_t)        ks_dhtrt_delete_node(ks_dhtrt_routetable_t* table, ks_dht_node_t* node);
+
+KS_DECLARE(ks_status_t)        ks_dhtrt_touch_node(ks_dhtrt_routetable_t* table,  ks_dht_nodeid_t nodeid);
+KS_DECLARE(ks_status_t)        ks_dhtrt_expire_node(ks_dhtrt_routetable_t* table,  ks_dht_nodeid_t nodeid);
+
+KS_DECLARE(uint8_t)            ks_dhtrt_findclosest_nodes(ks_dhtrt_routetable_t* table, ks_dhtrt_querynodes_t* query);
+KS_DECLARE(ks_dht_node_t*)     ks_dhtrt_find_node(ks_dhtrt_routetable_t* table, ks_dht_nodeid_t id);
+
+KS_DECLARE(ks_status_t)        ks_dhtrt_sharelock_node(ks_dht_node_t* node);
+KS_DECLARE(ks_status_t)        ks_dhtrt_release_node(ks_dht_node_t* node);
+KS_DECLARE(ks_status_t)        ks_dhtrt_release_querynodes(ks_dhtrt_querynodes_t* query);
+
+KS_DECLARE(void)               ks_dhtrt_process_table(ks_dhtrt_routetable_t* table);
+
+KS_DECLARE(uint32_t)           ks_dhtrt_serialize(ks_dhtrt_routetable_t* table, void** ptr);
+KS_DECLARE(ks_status_t)        ks_dhtrt_deserialize(ks_dhtrt_routetable_t* table, void* ptr);
+																							 
+/* debugging aids */
+KS_DECLARE(void)               ks_dhtrt_dump(ks_dhtrt_routetable_t* table, int level);
+
+																																				
+KS_END_EXTERN_C
+
+#endif /* KS_DHT_H */
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
+ */

libs/libblade/test/Makefile.am

@@ -18,6 +18,21 @@ blades_SOURCES = blades.c tap.c
 blades_CFLAGS = $(AM_CFLAGS)
 blades_LDADD = $(TEST_LDADD)
 
+check_PROGRAMS += testdht2
+testdht2_SOURCES = testdht2.c tap.c
+testdht2_CFLAGS = $(AM_CFLAGS)
+testdht2_LDADD = $(TEST_LDADD)
+
+check_PROGRAMS += testbuckets
+testbuckets_SOURCES = testbuckets.c tap.c
+testbuckets_CFLAGS = $(AM_CFLAGS)
+testbuckets_LDADD = $(TEST_LDADD)
+
+check_PROGRAMS += nodeidgen
+nodeidgen_SOURCES = nodeidgen.c tap.c
+nodeidgen_CFLAGS = $(AM_CFLAGS)
+nodeidgen_LDADD = $(TEST_LDADD)
+
 
 TESTS=$(check_PROGRAMS)
 

libs/libblade/test/nodeidgen.c

@@ -0,0 +1,16 @@
+#include <ks.h>
+#include <ks_dht.h>
+#include <ks_dht-int.h>
+#include <tap.h>
+
+int main() {
+	uint8_t nodeid[KS_DHT_NODEID_SIZE];
+	char buf[KS_DHT_NODEID_SIZE * 2 + 1];
+	
+	randombytes_buf(nodeid, KS_DHT_NODEID_SIZE);
+
+	ks_dht_hex(nodeid, buf, KS_DHT_NODEID_SIZE);
+
+	printf(buf);
+	return 0;
+}

libs/libblade/test/testdht2.c

@@ -0,0 +1,385 @@
+#include <ks.h>
+#include <ks_dht.h>
+#include <ks_dht-int.h>
+#include <tap.h>
+
+ks_dht_storageitem_skey_t sk;
+ks_dht_storageitem_pkey_t pk;
+
+ks_status_t dht2_updated_callback(ks_dht_t *dht, ks_dht_storageitem_t *item)
+{
+	diag("dht2_updated_callback\n");
+	return KS_STATUS_SUCCESS;
+}
+
+ks_status_t dht2_distribute_callback(ks_dht_t *dht, ks_dht_storageitem_t *item)
+{
+	diag("dht2_distribute_callback\n");
+	return KS_STATUS_SUCCESS;
+}
+
+ks_status_t dht2_put_callback(ks_dht_t *dht, ks_dht_job_t *job)
+{
+	diag("dht2_put_callback\n");
+	return KS_STATUS_SUCCESS;
+}
+
+ks_status_t dht2_get_token_callback(ks_dht_t *dht, ks_dht_job_t *job)
+{
+	char buf[KS_DHT_TOKEN_SIZE * 2 + 1];
+	const char *v = "Hello World!";
+	size_t v_len = strlen(v);
+	ks_dht_storageitem_signature_t sig;
+	ks_dht_storageitem_t *mutable = NULL;
+	
+	diag("dht2_get_token_callback %s\n", ks_dht_hex(job->response_token.token, buf, KS_DHT_TOKEN_SIZE));
+
+	ks_dht_storageitem_signature_generate(&sig, &sk, NULL, 0, 1, (uint8_t *)v, v_len);
+	// @todo check if exists
+	ks_dht_storageitem_create_mutable(&mutable, dht->pool, &job->query_target, (uint8_t *)v, v_len, &pk, NULL, 0, 1, &sig);
+	mutable->sk = sk;
+	ks_dht_storageitems_insert(dht, mutable);
+	
+	ks_dht_put(dht, &job->raddr, dht2_put_callback, NULL, &job->response_token, 0, mutable);
+	return KS_STATUS_SUCCESS;
+}
+
+ks_status_t dht2_search_callback(ks_dht_t *dht, ks_dht_job_t *job)
+{
+	ks_dht_search_t *search = (ks_dht_search_t *)job->data;
+	diag("dht2_search_callback %d\n", search->results_length);
+	return KS_STATUS_SUCCESS;
+}
+
+int main() {
+	//ks_size_t buflen;
+  ks_status_t err;
+  int mask = 0;
+  ks_dht_nodeid_t nodeid;
+  ks_dht_t *dht1 = NULL;
+  ks_dht_t *dht2 = NULL;
+  ks_dht_t *dht3 = NULL;
+  ks_dht_endpoint_t *ep1;
+  ks_dht_endpoint_t *ep2;
+  ks_dht_endpoint_t *ep3;
+  ks_bool_t have_v4, have_v6;
+  char v4[48] = {0}, v6[48] = {0};
+  ks_sockaddr_t addr;
+  ks_sockaddr_t raddr1;
+  //ks_sockaddr_t raddr2;
+  //ks_sockaddr_t raddr3;
+  ks_dht_nodeid_t target;
+  //ks_dht_storageitem_t *immutable = NULL;
+  ks_dht_storageitem_t *mutable1 = NULL;
+  ks_dht_storageitem_t *mutable2 = NULL;
+  const char *v = "Hello World!";
+  size_t v_len = strlen(v);
+  //ks_dht_storageitem_skey_t sk; //= { { 0xe0, 0x6d, 0x31, 0x83, 0xd1, 0x41, 0x59, 0x22, 0x84, 0x33, 0xed, 0x59, 0x92, 0x21, 0xb8, 0x0b,
+  //0xd0, 0xa5, 0xce, 0x83, 0x52, 0xe4, 0xbd, 0xf0, 0x26, 0x2f, 0x76, 0x78, 0x6e, 0xf1, 0xc7, 0x4d,
+  //0xb7, 0xe7, 0xa9, 0xfe, 0xa2, 0xc0, 0xeb, 0x26, 0x9d, 0x61, 0xe3, 0xb3, 0x8e, 0x45, 0x0a, 0x22,
+  //0xe7, 0x54, 0x94, 0x1a, 0xc7, 0x84, 0x79, 0xd6, 0xc5, 0x4e, 0x1f, 0xaf, 0x60, 0x37, 0x88, 0x1d } };
+  //ks_dht_storageitem_pkey_t pk; //= { { 0x77, 0xff, 0x84, 0x90, 0x5a, 0x91, 0x93, 0x63, 0x67, 0xc0, 0x13, 0x60, 0x80, 0x31, 0x04, 0xf9,
+  //0x24, 0x32, 0xfc, 0xd9, 0x04, 0xa4, 0x35, 0x11, 0x87, 0x6d, 0xf5, 0xcd, 0xf3, 0xe7, 0xe5, 0x48 } };
+  //uint8_t sk1[KS_DHT_STORAGEITEM_SKEY_SIZE];
+  //uint8_t pk1[KS_DHT_STORAGEITEM_PKEY_SIZE];
+  ks_dht_storageitem_signature_t sig;
+  //char sk_buf[KS_DHT_STORAGEITEM_SKEY_SIZE * 2 + 1];
+  //char pk_buf[KS_DHT_STORAGEITEM_PKEY_SIZE * 2 + 1];
+  //const char *test1vector = "3:seqi1e1:v12:Hello World!";
+  //const char *test1vector = "4:salt6:foobar3:seqi1e1:v12:Hello World!";
+  //size_t test1vector_len = strlen(test1vector);
+  //uint8_t test1vector_sig[KS_DHT_STORAGEITEM_SIGNATURE_SIZE];
+  //char test1vector_buf[KS_DHT_STORAGEITEM_SIGNATURE_SIZE * 2 + 1];
+
+  err = ks_init();
+  ok(!err);
+
+  ks_global_set_default_logger(KS_LOG_LEVEL_INFO);
+
+  err = ks_find_local_ip(v4, sizeof(v4), &mask, AF_INET, NULL);
+  ok(err == KS_STATUS_SUCCESS);
+  have_v4 = !zstr_buf(v4);
+  
+  //err = ks_find_local_ip(v6, sizeof(v6), NULL, AF_INET6, NULL);
+  //ok(err == KS_STATUS_SUCCESS);
+  have_v6 = KS_FALSE;//!zstr_buf(v6);
+
+  ok(have_v4 || have_v6);
+
+  if (have_v4) {
+	  diag("Binding to %s on ipv4\n", v4);
+  }
+  if (have_v6) {
+	  diag("Binding to %s on ipv6\n", v6);
+  }
+
+  ks_dht_dehex(nodeid.id, "0000000000000000000000000000000000000001", KS_DHT_NODEID_SIZE);
+  err = ks_dht_create(&dht1, NULL, NULL, &nodeid);
+  ok(err == KS_STATUS_SUCCESS);
+  
+  ks_dht_dehex(nodeid.id, "0000000000000000000000000000000000000002", KS_DHT_NODEID_SIZE);
+  err = ks_dht_create(&dht2, NULL, NULL, &nodeid);
+  ok(err == KS_STATUS_SUCCESS);
+
+  ks_dht_dehex(nodeid.id, "0000000000000000000000000000000000000003", KS_DHT_NODEID_SIZE);
+  err = ks_dht_create(&dht3, NULL, NULL, &nodeid);
+  ok(err == KS_STATUS_SUCCESS);
+  
+  if (have_v4) {
+    err = ks_addr_set(&addr, v4, KS_DHT_DEFAULT_PORT, AF_INET);
+	ok(err == KS_STATUS_SUCCESS);
+	
+    err = ks_dht_bind(dht1, &addr, &ep1);
+    ok(err == KS_STATUS_SUCCESS);
+
+	raddr1 = addr;
+	
+	err = ks_addr_set(&addr, v4, KS_DHT_DEFAULT_PORT + 1, AF_INET);
+	ok(err == KS_STATUS_SUCCESS);
+	
+	err = ks_dht_bind(dht2, &addr, &ep2);
+	ok(err == KS_STATUS_SUCCESS);
+
+	//raddr2 = addr;
+
+	err = ks_addr_set(&addr, v4, KS_DHT_DEFAULT_PORT + 2, AF_INET);
+	ok(err == KS_STATUS_SUCCESS);
+	
+	err = ks_dht_bind(dht3, &addr, &ep3);
+	ok(err == KS_STATUS_SUCCESS);
+
+	//raddr3 = addr;
+  }
+
+  if (have_v6) {
+	err = ks_addr_set(&addr, v6, KS_DHT_DEFAULT_PORT, AF_INET6);
+	ok(err == KS_STATUS_SUCCESS);
+	  
+    err = ks_dht_bind(dht1, &addr, NULL);
+    ok(err == KS_STATUS_SUCCESS);
+
+	err = ks_addr_set(&addr, v6, KS_DHT_DEFAULT_PORT + 1, AF_INET6);
+	ok(err == KS_STATUS_SUCCESS);
+
+	err = ks_dht_bind(dht2, &addr, NULL);
+	ok(err == KS_STATUS_SUCCESS);
+
+	err = ks_addr_set(&addr, v6, KS_DHT_DEFAULT_PORT + 2, AF_INET6);
+	ok(err == KS_STATUS_SUCCESS);
+
+	err = ks_dht_bind(dht3, &addr, NULL);
+	ok(err == KS_STATUS_SUCCESS);
+  }
+
+  diag("Ping test\n");
+  
+  ks_dht_ping(dht2, &raddr1, NULL, NULL); // (QUERYING)
+
+  ks_dht_pulse(dht2, 100); // Send queued ping from dht2 to dht1 (RESPONDING)
+
+  ks_dht_pulse(dht1, 100); // Receive and process ping query from dht2, queue and send ping response
+
+  ok(ks_dhtrt_find_node(dht1->rt_ipv4, dht2->nodeid) == NULL); // The node should be dubious, and thus not be returned as good yet
+
+  ks_dht_pulse(dht2, 100); // Receive and process ping response from dht1 (PROCESSING then COMPLETING)
+
+  ok(ks_dhtrt_find_node(dht2->rt_ipv4, dht1->nodeid) != NULL); // The node should be good, and thus be returned as good
+
+  ks_dht_pulse(dht2, 100); // Call finish callback and purge the job (COMPLETING)
+
+  diag("Pulsing for route table pings\n"); // Wait for route table pinging to catch up
+  for (int i = 0; i < 10; ++i) {
+	  ks_dht_pulse(dht1, 100);
+	  ks_dht_pulse(dht2, 100);
+	  ks_dht_pulse(dht3, 100);
+  }
+  ok(ks_dhtrt_find_node(dht1->rt_ipv4, dht2->nodeid) != NULL); // The node should be good by now, and thus be returned as good
+
+  
+  ks_dht_ping(dht3, &raddr1, NULL, NULL); // (QUERYING)
+
+  ks_dht_pulse(dht3, 100); // Send queued ping from dht3 to dht1 (RESPONDING)
+  
+  ks_dht_pulse(dht1, 100); // Receive and process ping query from dht3, queue and send ping response
+
+  ok(ks_dhtrt_find_node(dht1->rt_ipv4, dht3->nodeid) == NULL); // The node should be dubious, and thus not be returned as good yet
+
+  ks_dht_pulse(dht3, 100); // Receive and process ping response from dht1 (PROCESSING then COMPLETING)
+
+  ok(ks_dhtrt_find_node(dht3->rt_ipv4, dht1->nodeid) != NULL); // The node should be good, and thus be returned as good
+
+  ks_dht_pulse(dht3, 100); // Call finish callback and purge the job (COMPLETING)
+
+  diag("Pulsing for route table pings\n"); // Wait for route table pinging to catch up
+  for (int i = 0; i < 10; ++i) {
+	  ks_dht_pulse(dht1, 100);
+	  ks_dht_pulse(dht2, 100);
+	  ks_dht_pulse(dht3, 100);
+  }
+  ok(ks_dhtrt_find_node(dht1->rt_ipv4, dht2->nodeid) != NULL); // The node should be good by now, and thus be returned as good
+
+  // Test bootstrap find_node from dht3 to dht1 to find dht2 nodeid
+
+  /*
+  diag("Find_Node test\n");
+
+  ks_dht_findnode(dht3, NULL, &raddr1, NULL, NULL, &dht2->nodeid);
+
+  ks_dht_pulse(dht3, 100); // Send queued findnode from dht3 to dht1
+
+  ks_dht_pulse(dht1, 100); // Receive and process findnode query from dht3, queue and send findnode response
+
+  ok(ks_dhtrt_find_node(dht1->rt_ipv4, dht3->nodeid) == NULL); // The node should be dubious, and thus not be returned as good yet
+
+  ks_dht_pulse(dht3, 100); // Receive and process findnode response from dht1
+  
+  ks_dht_pulse(dht3, 100); // Call finish callback and purge the job (COMPLETING)
+
+  ok(ks_dhtrt_find_node(dht3->rt_ipv4, dht2->nodeid) == NULL); // The node should be dubious, and thus not be returned as good yet
+  
+  diag("Pulsing for route table pings\n"); // Wait for route table pinging to catch up
+  for (int i = 0; i < 10; ++i) {
+	  ks_dht_pulse(dht1, 100);
+	  ks_dht_pulse(dht2, 100);
+	  ks_dht_pulse(dht3, 100);
+  }
+  ok(ks_dhtrt_find_node(dht3->rt_ipv4, dht2->nodeid) != NULL); // The node should be good by now, and thus be returned as good
+  */
+
+  diag("Search test\n");
+  
+  ks_dht_search(dht3, dht2_search_callback, NULL, dht3->rt_ipv4, &dht2->nodeid);
+  diag("Pulsing for route table pings\n"); // Wait for route table pinging to catch up
+  for (int i = 0; i < 20; ++i) {
+	  ks_dht_pulse(dht1, 100);
+	  ks_dht_pulse(dht2, 100);
+	  ks_dht_pulse(dht3, 100);
+  }
+  
+  //diag("Get test\n");
+  
+
+  /*
+  ks_dht_storageitem_target_immutable((uint8_t *)v, v_len, &target);
+  ks_dht_storageitem_create_immutable(&immutable, dht1->pool, &target, (uint8_t *)v, v_len);
+  ks_dht_storageitems_insert(dht1, immutable);
+  */
+  
+  /*
+  crypto_sign_keypair(pk.key, sk.key);
+
+  ks_dht_storageitem_signature_generate(&sig, &sk, NULL, 0, 1, (uint8_t *)v, v_len);
+  ks_dht_storageitem_target_mutable(&pk, NULL, 0, &target);
+  ks_dht_storageitem_create_mutable(&mutable, dht1->pool, &target, (uint8_t *)v, v_len, &pk, NULL, 0, 1, &sig);
+  mutable->sk = sk;
+  ks_dht_storageitems_insert(dht1, mutable);
+
+  ks_dht_get(dht2, &raddr1, dht2_get_callback, NULL, &target, NULL, 0);
+ 
+  ks_dht_pulse(dht2, 100); // send get query
+
+  ks_dht_pulse(dht1, 100); // receive get query and send get response
+
+  ks_dht_pulse(dht2, 100); // receive get response
+
+  ok(ks_dht_storageitems_find(dht2, &target) != NULL); // item should be verified and stored
+
+  ks_dht_pulse(dht2, 100); // Call finish callback and purge the job (COMPLETING)
+  */
+
+  /*
+  diag("Put test\n");
+
+  crypto_sign_keypair(pk.key, sk.key);
+
+  ks_dht_storageitem_target_mutable(&pk, NULL, 0, &target);
+
+  ks_dht_get(dht2, &raddr1, dht2_get_token_callback, NULL, &target, NULL, 0); // create job
+  
+  for (int i = 0; i < 20; ++i) {
+	  ks_dht_pulse(dht1, 100);
+	  ks_dht_pulse(dht2, 100);
+	  ks_dht_pulse(dht3, 100);
+  }
+  */
+
+  /*
+  diag("Publish test\n");
+  
+  crypto_sign_keypair(pk.key, sk.key);
+
+  ks_dht_storageitem_target_mutable(&pk, NULL, 0, &target);
+  
+  ks_dht_storageitem_signature_generate(&sig, &sk, NULL, 0, 1, (uint8_t *)v, v_len);
+  
+  ks_dht_storageitem_create_mutable(&mutable, dht2->pool, &target, (uint8_t *)v, v_len, &pk, NULL, 0, 1, &sig);
+  mutable->sk = sk;
+  ks_dht_storageitems_insert(dht2, mutable);
+  
+  ks_dht_publish(dht2, &raddr1, dht2_put_callback, NULL, 0, mutable); // create job
+  
+  for (int i = 0; i < 20; ++i) {
+	  ks_dht_pulse(dht1, 100);
+	  ks_dht_pulse(dht2, 100);
+	  ks_dht_pulse(dht3, 100);
+  }
+  */
+
+  
+  diag("Distribute test\n");
+  
+  crypto_sign_keypair(pk.key, sk.key);
+
+  ks_dht_storageitem_target_mutable(&pk, NULL, 0, &target);
+  
+  ks_dht_storageitem_signature_generate(&sig, &sk, NULL, 0, 1, (uint8_t *)v, v_len);
+  
+  ks_dht_storageitem_create_mutable(&mutable2, dht2->pool, &target, (uint8_t *)v, v_len, &pk, NULL, 0, 1, &sig);
+  mutable2->sk = sk;
+  ks_dht_storageitems_insert(dht2, mutable2);
+  
+  ks_dht_distribute(dht2, dht2_distribute_callback, NULL, dht2->rt_ipv4, 0, mutable2); // create job
+  
+  for (int i = 0; i < 30; ++i) {
+	  ks_dht_pulse(dht1, 100);
+	  ks_dht_pulse(dht2, 100);
+	  ks_dht_pulse(dht3, 100);
+  }
+  ks_dht_storageitem_dereference(mutable2);
+  ok(mutable2->refc == 0);
+
+  mutable1 = ks_dht_storageitems_find(dht1, &target);
+  ok(mutable1 != NULL);
+  
+  ks_dht_storageitem_callback(mutable1, dht2_updated_callback);
+  ks_dht_storageitem_callback(mutable2, dht2_updated_callback);
+
+  ks_dht_storageitem_signature_generate(&sig, &sk, NULL, 0, 2, (uint8_t *)v, v_len);
+  mutable1->seq = 2;
+  mutable1->sig = sig;
+  
+  //ks_dht_storageitem_signature_generate(&sig, &sk, NULL, 0, 2, (uint8_t *)v, v_len);
+  //mutable2->seq = 2;
+  //mutable2->sig = sig;
+
+  ks_dht_distribute(dht2, dht2_distribute_callback, NULL, dht2->rt_ipv4, 0, mutable2);
+  for (int i = 0; i < 30; ++i) {
+	  ks_dht_pulse(dht1, 100);
+	  ks_dht_pulse(dht2, 100);
+	  ks_dht_pulse(dht3, 100);
+  }
+  ks_dht_storageitem_dereference(mutable1);
+  
+  /* Cleanup and shutdown */
+  diag("Cleanup\n");
+
+  ks_dht_destroy(&dht3);
+
+  ks_dht_destroy(&dht2);
+
+  ks_dht_destroy(&dht1);
+
+  ks_shutdown();
+  
+  done_testing();
+}