conntrackd: generalize caching infrastructure

This patch generalizes the caching infrastructure to store different
object types. This patch is the first in the series to prepare
support for the synchronization of expectations.

Signed-off-by: Pablo Neira Ayuso <pablo@netfilter.org>

12 files changed, 363 insertions, 288 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 70e496d..a0abeee 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -12,7 +12,7 @@ conntrack_LDADD = ../extensions/libct_proto_tcp.la ../extensions/libct_proto_udp
 conntrackd_SOURCES = alarm.c main.c run.c hash.c queue.c rbtree.c \
 		    local.c log.c mcast.c udp.c netlink.c vector.c \
 		    filter.c fds.c event.c process.c origin.c date.c \
-		    cache.c cache_iterators.c \
+		    cache.c cache-ct.c \
 		    cache_timer.c \
 		    sync-mode.c sync-alarm.c sync-ftfw.c sync-notrack.c \
 		    traffic_stats.c stats-mode.c \
diff --git a/src/cache_iterators.c b/src/cache-ct.c
index 3248c70..2c6fd4e 100644
--- a/src/cache_iterators.c
+++ b/src/cache-ct.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2007 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -22,19 +23,97 @@
 #include "conntrackd.h"
 #include "netlink.h"
 #include "event.h"
+#include "jhash.h"
+#include "network.h"
 
-#include <libnetfilter_conntrack/libnetfilter_conntrack.h>
-#include <sched.h>
 #include <errno.h>
 #include <string.h>
 #include <time.h>
+#include <libnetfilter_conntrack/libnetfilter_conntrack.h>
 
-struct __dump_container {
-	int fd;
-	int type;
-};
+static uint32_t
+cache_hash4_ct(const struct nf_conntrack *ct, const struct hashtable *table)
+{
+	uint32_t a[4] = {
+		[0]	= nfct_get_attr_u32(ct, ATTR_IPV4_SRC),
+		[1]	= nfct_get_attr_u32(ct, ATTR_IPV4_DST),
+		[2]	= nfct_get_attr_u8(ct, ATTR_L3PROTO) << 16 |
+			  nfct_get_attr_u8(ct, ATTR_L4PROTO),
+		[3]	= nfct_get_attr_u16(ct, ATTR_PORT_SRC) << 16 |
+			  nfct_get_attr_u16(ct, ATTR_PORT_DST),
+	};
 
-static int do_dump(void *data1, void *n)
+	/*
+	 * Instead of returning hash % table->hashsize (implying a divide)
+	 * we return the high 32 bits of the (hash * table->hashsize) that will
+	 * give results between [0 and hashsize-1] and same hash distribution,
+	 * but using a multiply, less expensive than a divide. See:
+	 * http://www.mail-archive.com/netdev@vger.kernel.org/msg56623.html
+	 */
+	return ((uint64_t)jhash2(a, 4, 0) * table->hashsize) >> 32;
+}
+
+static uint32_t
+cache_hash6_ct(const struct nf_conntrack *ct, const struct hashtable *table)
+{
+	uint32_t a[10];
+
+	memcpy(&a[0], nfct_get_attr(ct, ATTR_IPV6_SRC), sizeof(uint32_t)*4);
+	memcpy(&a[4], nfct_get_attr(ct, ATTR_IPV6_SRC), sizeof(uint32_t)*4);
+	a[8] = nfct_get_attr_u8(ct, ATTR_ORIG_L3PROTO) << 16 |
+	       nfct_get_attr_u8(ct, ATTR_ORIG_L4PROTO);
+	a[9] = nfct_get_attr_u16(ct, ATTR_ORIG_PORT_SRC) << 16 |
+	       nfct_get_attr_u16(ct, ATTR_ORIG_PORT_DST);
+
+	return ((uint64_t)jhash2(a, 10, 0) * table->hashsize) >> 32;
+}
+
+static uint32_t
+cache_ct_hash(const void *data, const struct hashtable *table)
+{
+	int ret = 0;
+	const struct nf_conntrack *ct = data;
+
+	switch(nfct_get_attr_u8(ct, ATTR_L3PROTO)) {
+		case AF_INET:
+			ret = cache_hash4_ct(ct, table);
+			break;
+		case AF_INET6:
+			ret = cache_hash6_ct(ct, table);
+			break;
+		default:
+			dlog(LOG_ERR, "unknown layer 3 proto in hash");
+			break;
+	}
+	return ret;
+}
+
+static int cache_ct_cmp(const void *data1, const void *data2)
+{
+	const struct cache_object *obj = data1;
+	const struct nf_conntrack *ct = data2;
+
+	return nfct_cmp(obj->ptr, ct, NFCT_CMP_ORIG) &&
+	       nfct_get_attr_u32(obj->ptr, ATTR_ID) ==
+	       nfct_get_attr_u32(ct, ATTR_ID);
+}
+
+static void *cache_ct_alloc(void)
+{
+	return nfct_new();
+}
+
+static void cache_ct_free(void *ptr)
+{
+	nfct_destroy(ptr);
+}
+
+static void cache_ct_copy(void *dst, void *src, unsigned int flags)
+{
+	nfct_copy(dst, src, flags);
+}
+
+static int cache_ct_dump_step(void *data1, void *n)
 {
 	char buf[1024];
 	int size;
@@ -57,13 +136,13 @@ static int do_dump(void *data1, void *n)
 		return 0;
 
 	/* do not show cached timeout, this may confuse users */
-	if (nfct_attr_is_set(obj->ct, ATTR_TIMEOUT))
-		nfct_attr_unset(obj->ct, ATTR_TIMEOUT);
+	if (nfct_attr_is_set(obj->ptr, ATTR_TIMEOUT))
+		nfct_attr_unset(obj->ptr, ATTR_TIMEOUT);
 
 	memset(buf, 0, sizeof(buf));
 	size = nfct_snprintf(buf, 
 			     sizeof(buf), 
-			     obj->ct, 
+			     obj->ptr,
 			     NFCT_T_UNKNOWN, 
 			     container->type,
 			     0);
@@ -91,26 +170,11 @@ static int do_dump(void *data1, void *n)
 	return 0;
 }
 
-void cache_dump(struct cache *c, int fd, int type)
-{
-	struct __dump_container tmp = {
-		.fd	= fd,
-		.type	= type
-	};
-
-	hashtable_iterate(c->h, (void *) &tmp, do_dump);
-}
-
-struct __commit_container {
-	struct nfct_handle 	*h;
-	struct cache 		*c;
-};
-
 static void
-__do_commit_step(struct __commit_container *tmp, struct cache_object *obj)
+cache_ct_commit_step(struct __commit_container *tmp, struct cache_object *obj)
 {
 	int ret, retry = 1, timeout;
-	struct nf_conntrack *ct = obj->ct;
+	struct nf_conntrack *ct = obj->ptr;
 
 	if (CONFIG(commit_timeout)) {
 		timeout = CONFIG(commit_timeout);
@@ -153,29 +217,29 @@ retry:
 	}
 }
 
-static int do_commit_related(void *data, void *n)
+static int cache_ct_commit_related(void *data, void *n)
 {
 	struct cache_object *obj = n;
 
-	if (ct_is_related(obj->ct))
-		__do_commit_step(data, obj);
+	if (ct_is_related(obj->ptr))
+		cache_ct_commit_step(data, obj);
 
 	/* keep iterating even if we have found errors */
 	return 0;
 }
 
-static int do_commit_master(void *data, void *n)
+static int cache_ct_commit_master(void *data, void *n)
 {
 	struct cache_object *obj = n;
 
-	if (ct_is_related(obj->ct))
+	if (ct_is_related(obj->ptr))
 		return 0;
 
-	__do_commit_step(data, obj);
+	cache_ct_commit_step(data, obj);
 	return 0;
 }
 
-int cache_commit(struct cache *c, struct nfct_handle *h, int clientfd)
+static int cache_ct_commit(struct cache *c, struct nfct_handle *h, int clientfd)
 {
 	unsigned int commit_ok, commit_fail;
 	struct __commit_container tmp = {
@@ -200,7 +264,7 @@ int cache_commit(struct cache *c, struct nfct_handle *h, int clientfd)
 			hashtable_iterate_limit(c->h, &tmp,
 						STATE_SYNC(commit).current,
 						CONFIG(general).commit_steps,
-						do_commit_master);
+						cache_ct_commit_master);
 		if (STATE_SYNC(commit).current < CONFIG(hashsize)) {
 			STATE_SYNC(commit).state = COMMIT_STATE_MASTER;
 			/* give it another step as soon as possible */
@@ -214,7 +278,7 @@ int cache_commit(struct cache *c, struct nfct_handle *h, int clientfd)
 			hashtable_iterate_limit(c->h, &tmp,
 						STATE_SYNC(commit).current,
 						CONFIG(general).commit_steps,
-						do_commit_related);
+						cache_ct_commit_related);
 		if (STATE_SYNC(commit).current < CONFIG(hashsize)) {
 			STATE_SYNC(commit).state = COMMIT_STATE_RELATED;
 			/* give it another step as soon as possible */
@@ -251,18 +315,44 @@ int cache_commit(struct cache *c, struct nfct_handle *h, int clientfd)
 	return 1;
 }
 
-static int do_flush(void *data, void *n)
+static struct nethdr *
+cache_ct_build_msg(const struct cache_object *obj, int type)
 {
-	struct cache *c = data;
-	struct cache_object *obj = n;
-
-	cache_del(c, obj);
-	cache_object_free(obj);
-	return 0;
+	return BUILD_NETMSG_FROM_CT(obj->ptr, type);
 }
 
-void cache_flush(struct cache *c)
-{
-	hashtable_iterate(c->h, c, do_flush);
-	c->stats.flush++;
-}
+/* template to cache conntracks coming from the kernel. */
+struct cache_ops cache_sync_internal_ct_ops = {
+	.hash		= cache_ct_hash,
+	.cmp		= cache_ct_cmp,
+	.alloc		= cache_ct_alloc,
+	.free		= cache_ct_free,
+	.copy		= cache_ct_copy,
+	.dump_step	= cache_ct_dump_step,
+	.commit		= NULL,
+	.build_msg	= cache_ct_build_msg,
+};
+
+/* template to cache conntracks coming from the network. */
+struct cache_ops cache_sync_external_ct_ops = {
+	.hash		= cache_ct_hash,
+	.cmp		= cache_ct_cmp,
+	.alloc		= cache_ct_alloc,
+	.free		= cache_ct_free,
+	.copy		= cache_ct_copy,
+	.dump_step	= cache_ct_dump_step,
+	.commit		= cache_ct_commit,
+	.build_msg	= NULL,
+};
+
+/* template to cache conntracks for the statistics mode. */
+struct cache_ops cache_stats_ct_ops = {
+	.hash		= cache_ct_hash,
+	.cmp		= cache_ct_cmp,
+	.alloc		= cache_ct_alloc,
+	.free		= cache_ct_free,
+	.copy		= cache_ct_copy,
+	.dump_step	= cache_ct_dump_step,
+	.commit		= NULL,
+	.build_msg	= NULL,
+};
diff --git a/src/cache.c b/src/cache.c
index f411121..efdab0e 100644
--- a/src/cache.c
+++ b/src/cache.c
@@ -1,5 +1,6 @@
 /*
- * (C) 2006-2009 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
  * 
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
@@ -28,80 +29,14 @@
 #include <string.h>
 #include <time.h>
 
-static uint32_t
-__hash4(const struct nf_conntrack *ct, const struct hashtable *table)
-{
-	uint32_t a[4] = {
-		[0]	= nfct_get_attr_u32(ct, ATTR_IPV4_SRC),
-		[1]	= nfct_get_attr_u32(ct, ATTR_IPV4_DST),
-		[2]	= nfct_get_attr_u8(ct, ATTR_L3PROTO) << 16 |
-			  nfct_get_attr_u8(ct, ATTR_L4PROTO),
-		[3]	= nfct_get_attr_u16(ct, ATTR_PORT_SRC) << 16 |
-			  nfct_get_attr_u16(ct, ATTR_PORT_DST),
-	};
-
-	/*
-	 * Instead of returning hash % table->hashsize (implying a divide)
-	 * we return the high 32 bits of the (hash * table->hashsize) that will
-	 * give results between [0 and hashsize-1] and same hash distribution,
-	 * but using a multiply, less expensive than a divide. See:
-	 * http://www.mail-archive.com/netdev@vger.kernel.org/msg56623.html
-	 */
-	return ((uint64_t)jhash2(a, 4, 0) * table->hashsize) >> 32;
-}
-
-static uint32_t
-__hash6(const struct nf_conntrack *ct, const struct hashtable *table)
-{
-	uint32_t a[10];
-
-	memcpy(&a[0], nfct_get_attr(ct, ATTR_IPV6_SRC), sizeof(uint32_t)*4);
-	memcpy(&a[4], nfct_get_attr(ct, ATTR_IPV6_SRC), sizeof(uint32_t)*4);
-	a[8] = nfct_get_attr_u8(ct, ATTR_ORIG_L3PROTO) << 16 |
-	       nfct_get_attr_u8(ct, ATTR_ORIG_L4PROTO);
-	a[9] = nfct_get_attr_u16(ct, ATTR_ORIG_PORT_SRC) << 16 |
-	       nfct_get_attr_u16(ct, ATTR_ORIG_PORT_DST);
-
-	return ((uint64_t)jhash2(a, 10, 0) * table->hashsize) >> 32;
-}
-
-static uint32_t hash(const void *data, const struct hashtable *table)
-{
-	int ret = 0;
-	const struct nf_conntrack *ct = data;
-
-	switch(nfct_get_attr_u8(ct, ATTR_L3PROTO)) {
-		case AF_INET:
-			ret = __hash4(ct, table);
-			break;
-		case AF_INET6:
-			ret = __hash6(ct, table);
-			break;
-		default:
-			dlog(LOG_ERR, "unknown layer 3 proto in hash");
-			break;
-	}
-
-	return ret;
-}
-
-static int compare(const void *data1, const void *data2)
-{
-	const struct cache_object *obj = data1;
-	const struct nf_conntrack *ct = data2;
-
-	return nfct_cmp(obj->ct, ct, NFCT_CMP_ORIG) &&
-	       nfct_get_attr_u32(obj->ct, ATTR_ID) ==
-	       nfct_get_attr_u32(ct, ATTR_ID);
-}
-
 struct cache_feature *cache_feature[CACHE_MAX_FEATURE] = {
 	[TIMER_FEATURE]		= &timer_feature,
 };
 
-struct cache *cache_create(const char *name, 
+struct cache *cache_create(const char *name, enum cache_type type,
 			   unsigned int features, 
-			   struct cache_extra *extra)
+			   struct cache_extra *extra,
+			   struct cache_ops *ops)
 {
 	size_t size = sizeof(struct cache_object);
 	int i, j = 0;
@@ -110,12 +45,16 @@ struct cache *cache_create(const char *name,
 	unsigned int feature_offset[CACHE_MAX_FEATURE] = {};
 	unsigned int feature_type[CACHE_MAX_FEATURE] = {};
 
+	if (type == CACHE_T_NONE || type >= CACHE_T_MAX)
+		return NULL;
+
 	c = malloc(sizeof(struct cache));
 	if (!c)
 		return NULL;
 	memset(c, 0, sizeof(struct cache));
 
 	strcpy(c->name, name);
+	c->type = type;
 
 	for (i = 0; i < CACHE_MAX_FEATURE; i++) {
 		if ((1 << i) & features) {
@@ -150,11 +89,19 @@ struct cache *cache_create(const char *name,
 	}
 	memcpy(c->feature_offset, feature_offset, sizeof(unsigned int) * j);
 
+	if (!ops || !ops->hash || !ops->cmp ||
+	    !ops->alloc || !ops->copy || !ops->free) {
+		free(c->feature_offset);
+		free(c->features);
+		free(c);
+		return NULL;
+	}
+	c->ops = ops;
+
 	c->h = hashtable_create(CONFIG(hashsize),
 				CONFIG(limit),
-				hash,
-				compare);
-
+				c->ops->hash,
+				c->ops->cmp);
 	if (!c->h) {
 		free(c->features);
 		free(c->feature_offset);
@@ -175,7 +122,7 @@ void cache_destroy(struct cache *c)
 	free(c);
 }
 
-struct cache_object *cache_object_new(struct cache *c, struct nf_conntrack *ct)
+struct cache_object *cache_object_new(struct cache *c, void *ptr)
 {
 	struct cache_object *obj;
 
@@ -187,13 +134,14 @@ struct cache_object *cache_object_new(struct cache *c, struct nf_conntrack *ct)
 	}
 	obj->cache = c;
 
-	if ((obj->ct = nfct_new()) == NULL) {
+	obj->ptr = c->ops->alloc();
+	if (obj->ptr == NULL) {
 		free(obj);
 		errno = ENOMEM;
 		c->stats.add_fail_enomem++;
 		return NULL;
 	}
-	nfct_copy(obj->ct, ct, NFCT_CP_OVERRIDE);
+	c->ops->copy(obj->ptr, ptr, NFCT_CP_OVERRIDE);
 	obj->status = C_OBJ_NONE;
 	c->stats.objects++;
 
@@ -203,7 +151,8 @@ struct cache_object *cache_object_new(struct cache *c, struct nf_conntrack *ct)
 void cache_object_free(struct cache_object *obj)
 {
 	obj->cache->stats.objects--;
-	nfct_destroy(obj->ct);
+	obj->cache->ops->free(obj->ptr);
+
 	free(obj);
 }
 
@@ -271,13 +220,12 @@ int cache_add(struct cache *c, struct cache_object *obj, int id)
 	return 0;
 }
 
-void cache_update(struct cache *c, struct cache_object *obj, int id,
-		  struct nf_conntrack *ct)
+void cache_update(struct cache *c, struct cache_object *obj, int id, void *ptr)
 {
 	char *data = obj->data;
 	unsigned int i;
 
-	nfct_copy(obj->ct, ct, NFCT_CP_META);
+	c->ops->copy(obj->ptr, ptr, NFCT_CP_META);
 
 	for (i = 0; i < c->num_features; i++) {
 		c->features[i]->update(obj, data);
@@ -322,23 +270,22 @@ void cache_del(struct cache *c, struct cache_object *obj)
 	__del(c, obj);
 }
 
-struct cache_object *
-cache_update_force(struct cache *c, struct nf_conntrack *ct)
+struct cache_object *cache_update_force(struct cache *c, void *ptr)
 {
 	struct cache_object *obj;
 	int id;
 
-	obj = cache_find(c, ct, &id);
+	obj = cache_find(c, ptr, &id);
 	if (obj) {
 		if (obj->status != C_OBJ_DEAD) {
-			cache_update(c, obj, id, ct);
+			cache_update(c, obj, id, ptr);
 			return obj;
 		} else {
 			cache_del(c, obj);
 			cache_object_free(obj);
 		}
 	}
-	obj = cache_object_new(c, ct);
+	obj = cache_object_new(c, ptr);
 	if (obj == NULL)
 		return NULL;
 
@@ -350,11 +297,10 @@ cache_update_force(struct cache *c, struct nf_conntrack *ct)
 	return obj;
 }
 
-struct cache_object *
-cache_find(struct cache *c, struct nf_conntrack *ct, int *id)
+struct cache_object *cache_find(struct cache *c, void *ptr, int *id)
 {
-	*id = hashtable_hash(c->h, ct);
-	return ((struct cache_object *) hashtable_find(c->h, ct, *id));
+	*id = hashtable_hash(c->h, ptr);
+	return ((struct cache_object *) hashtable_find(c->h, ptr, *id));
 }
 
 struct cache_object *cache_data_get_object(struct cache *c, void *data)
@@ -432,3 +378,33 @@ void cache_iterate_limit(struct cache *c, void *data,
 {
 	hashtable_iterate_limit(c->h, data, from, steps, iterate);
 }
+
+void cache_dump(struct cache *c, int fd, int type)
+{
+	struct __dump_container tmp = {
+		.fd	= fd,
+		.type	= type
+	};
+	hashtable_iterate(c->h, (void *) &tmp, c->ops->dump_step);
+}
+
+int cache_commit(struct cache *c, struct nfct_handle *h, int clientfd)
+{
+	return c->ops->commit(c, h, clientfd);
+}
+
+static int do_flush(void *data, void *n)
+{
+	struct cache *c = data;
+	struct cache_object *obj = n;
+
+	cache_del(c, obj);
+	cache_object_free(obj);
+	return 0;
+}
+
+void cache_flush(struct cache *c)
+{
+	hashtable_iterate(c->h, c, do_flush);
+	c->stats.flush++;
+}
diff --git a/src/external_cache.c b/src/external_cache.c
index 59c706a..073f309 100644
--- a/src/external_cache.c
+++ b/src/external_cache.c
@@ -28,9 +28,9 @@ static struct cache *external;
 
 static int external_cache_init(void)
 {
-	external = cache_create("external",
+	external = cache_create("external", CACHE_T_CT,
 				STATE_SYNC(sync)->external_cache_flags,
-				NULL);
+				NULL, &cache_sync_external_ct_ops);
 	if (external == NULL) {
 		dlog(LOG_ERR, "can't allocate memory for the external cache");
 		return -1;
diff --git a/src/internal_bypass.c b/src/internal_bypass.c
index 1e1478f..8ecec34 100644
--- a/src/internal_bypass.c
+++ b/src/internal_bypass.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2009 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -16,17 +17,18 @@
 #include "network.h"
 #include "origin.h"
 
-static int _init(void)
+static int internal_bypass_init(void)
 {
 	return 0;
 }
 
-static void _close(void)
+static void internal_bypass_close(void)
 {
 }
 
-static int dump_cb(enum nf_conntrack_msg_type type,
-		   struct nf_conntrack *ct, void *data)
+static int
+internal_bypass_ct_dump_cb(enum nf_conntrack_msg_type type,
+			   struct nf_conntrack *ct, void *data)
 {
 	char buf[1024];
 	int size, *fd = data;
@@ -44,7 +46,7 @@ static int dump_cb(enum nf_conntrack_msg_type type,
 	return NFCT_CB_CONTINUE;
 }
 
-static void dump(int fd, int type)
+static void internal_bypass_ct_dump(int fd, int type)
 {
 	struct nfct_handle *h;
 	u_int32_t family = AF_UNSPEC;
@@ -55,7 +57,7 @@ static void dump(int fd, int type)
 		dlog(LOG_ERR, "can't allocate memory for the internal cache");
 		return;
 	}
-	nfct_callback_register(h, NFCT_T_ALL, dump_cb, &fd);
+	nfct_callback_register(h, NFCT_T_ALL, internal_bypass_ct_dump_cb, &fd);
 	ret = nfct_query(h, NFCT_Q_DUMP, &family);
 	if (ret == -1) {
 		dlog(LOG_ERR, "can't dump kernel table");
@@ -63,7 +65,7 @@ static void dump(int fd, int type)
 	nfct_close(h);
 }
 
-static void flush(void)
+static void internal_bypass_ct_flush(void)
 {
 	nl_flush_conntrack_table(STATE(flush));
 }
@@ -74,7 +76,7 @@ struct {
 	uint32_t	del;
 } internal_bypass_stats;
 
-static void stats(int fd)
+static void internal_bypass_ct_stats(int fd)
 {
 	char buf[512];
 	int size;
@@ -91,25 +93,24 @@ static void stats(int fd)
 }
 
 /* unused, INTERNAL_F_POPULATE is unset. No cache, nothing to populate. */
-static void populate(struct nf_conntrack *ct)
+static void internal_bypass_ct_populate(struct nf_conntrack *ct)
 {
 }
 
 /* unused, INTERNAL_F_RESYNC is unset. */
-static void purge(void)
+static void internal_bypass_ct_purge(void)
 {
 }
 
 /* unused, INTERNAL_F_RESYNC is unset. Nothing to resync, we have no cache. */
-static int resync(enum nf_conntrack_msg_type type,
-		  struct nf_conntrack *ct,
-		  void *data)
+static int
+internal_bypass_ct_resync(enum nf_conntrack_msg_type type,
+			  struct nf_conntrack *ct, void *data)
 {
 	return NFCT_CB_CONTINUE;
 }
 
-static void
-event_new_sync(struct nf_conntrack *ct, int origin)
+static void internal_bypass_ct_event_new(struct nf_conntrack *ct, int origin)
 {
 	struct nethdr *net;
 
@@ -122,8 +123,7 @@ event_new_sync(struct nf_conntrack *ct, int origin)
 	internal_bypass_stats.new++;
 }
 
-static void
-event_update_sync(struct nf_conntrack *ct, int origin)
+static void internal_bypass_ct_event_upd(struct nf_conntrack *ct, int origin)
 {
 	struct nethdr *net;
 
@@ -136,8 +136,7 @@ event_update_sync(struct nf_conntrack *ct, int origin)
 	internal_bypass_stats.upd++;
 }
 
-static int
-event_destroy_sync(struct nf_conntrack *ct, int origin)
+static int internal_bypass_ct_event_del(struct nf_conntrack *ct, int origin)
 {
 	struct nethdr *net;
 
@@ -153,16 +152,18 @@ event_destroy_sync(struct nf_conntrack *ct, int origin)
 }
 
 struct internal_handler internal_bypass = {
-	.init			= _init,
-	.close			= _close,
-	.dump			= dump,
-	.flush			= flush,
-	.stats			= stats,
-	.stats_ext		= stats,
-	.populate		= populate,
-	.purge			= purge,
-	.resync			= resync,
-	.new			= event_new_sync,
-	.update			= event_update_sync,
-	.destroy		= event_destroy_sync,
+	.init			= internal_bypass_init,
+	.close			= internal_bypass_close,
+	.ct = {
+		.dump			= internal_bypass_ct_dump,
+		.flush			= internal_bypass_ct_flush,
+		.stats			= internal_bypass_ct_stats,
+		.stats_ext		= internal_bypass_ct_stats,
+		.populate		= internal_bypass_ct_populate,
+		.purge			= internal_bypass_ct_purge,
+		.resync			= internal_bypass_ct_resync,
+		.new			= internal_bypass_ct_event_new,
+		.upd			= internal_bypass_ct_event_upd,
+		.del			= internal_bypass_ct_event_del,
+	},
 };
diff --git a/src/internal_cache.c b/src/internal_cache.c
index e50e1db..7a698e6 100644
--- a/src/internal_cache.c
+++ b/src/internal_cache.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2009 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -19,46 +20,47 @@ static inline void sync_send(struct cache_object *obj, int query)
 	STATE_SYNC(sync)->enqueue(obj, query);
 }
 
-static int _init(void)
+static int internal_cache_init(void)
 {
-	STATE(mode)->internal->data =
-		cache_create("internal", 
+	STATE(mode)->internal->ct.data =
+		cache_create("internal", CACHE_T_CT,
 			     STATE_SYNC(sync)->internal_cache_flags,
-			     STATE_SYNC(sync)->internal_cache_extra);
+			     STATE_SYNC(sync)->internal_cache_extra,
+			     &cache_sync_internal_ct_ops);
 
-	if (!STATE(mode)->internal->data) {
+	if (!STATE(mode)->internal->ct.data) {
 		dlog(LOG_ERR, "can't allocate memory for the internal cache");
 		return -1;
 	}
 	return 0;
 }
 
-static void _close(void)
+static void internal_cache_close(void)
 {
-	cache_destroy(STATE(mode)->internal->data);
+	cache_destroy(STATE(mode)->internal->ct.data);
 }
 
-static void dump(int fd, int type)
+static void internal_cache_ct_dump(int fd, int type)
 {
-	cache_dump(STATE(mode)->internal->data, fd, type);
+	cache_dump(STATE(mode)->internal->ct.data, fd, type);
 }
 
-static void flush(void)
+static void internal_cache_ct_flush(void)
 {
-	cache_flush(STATE(mode)->internal->data);
+	cache_flush(STATE(mode)->internal->ct.data);
 }
 
-static void stats(int fd)
+static void internal_cache_ct_stats(int fd)
 {
-	cache_stats(STATE(mode)->internal->data, fd);
+	cache_stats(STATE(mode)->internal->ct.data, fd);
 }
 
-static void stats_ext(int fd)
+static void internal_cache_ct_stats_ext(int fd)
 {
-	cache_stats_extended(STATE(mode)->internal->data, fd);
+	cache_stats_extended(STATE(mode)->internal->ct.data, fd);
 }
 
-static void populate(struct nf_conntrack *ct)
+static void internal_cache_ct_populate(struct nf_conntrack *ct)
 {
 	/* This is required by kernels < 2.6.20 */
 	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
@@ -67,15 +69,15 @@ static void populate(struct nf_conntrack *ct)
 	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
 	nfct_attr_unset(ct, ATTR_USE);
 
-	cache_update_force(STATE(mode)->internal->data, ct);
+	cache_update_force(STATE(mode)->internal->ct.data, ct);
 }
 
-static int purge_step(void *data1, void *data2)
+static int internal_cache_ct_purge_step(void *data1, void *data2)
 {
 	struct cache_object *obj = data2;
 
 	STATE(get_retval) = 0;
-	nl_get_conntrack(STATE(get), obj->ct);	/* modifies STATE(get_reval) */
+	nl_get_conntrack(STATE(get), obj->ptr);	/* modifies STATE(get_reval) */
 	if (!STATE(get_retval)) {
 		if (obj->status != C_OBJ_DEAD) {
 			cache_object_set_status(obj, C_OBJ_DEAD);
@@ -87,14 +89,15 @@ static int purge_step(void *data1, void *data2)
 	return 0;
 }
 
-static void purge(void)
+static void internal_cache_ct_purge(void)
 {
-	cache_iterate(STATE(mode)->internal->data, NULL, purge_step);
+	cache_iterate(STATE(mode)->internal->ct.data, NULL,
+			internal_cache_ct_purge_step);
 }
 
-static int resync(enum nf_conntrack_msg_type type,
-		  struct nf_conntrack *ct,
-		  void *data)
+static int
+internal_cache_ct_resync(enum nf_conntrack_msg_type type,
+			 struct nf_conntrack *ct, void *data)
 {
 	struct cache_object *obj;
 
@@ -108,7 +111,7 @@ static int resync(enum nf_conntrack_msg_type type,
 	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
 	nfct_attr_unset(ct, ATTR_USE);
 
-	obj = cache_update_force(STATE(mode)->internal->data, ct);
+	obj = cache_update_force(STATE(mode)->internal->ct.data, ct);
 	if (obj == NULL)
 		return NFCT_CB_CONTINUE;
 
@@ -123,8 +126,7 @@ static int resync(enum nf_conntrack_msg_type type,
 	return NFCT_CB_CONTINUE;
 }
 
-static void
-event_new_sync(struct nf_conntrack *ct, int origin)
+static void internal_cache_ct_event_new(struct nf_conntrack *ct, int origin)
 {
 	struct cache_object *obj;
 	int id;
@@ -139,13 +141,13 @@ event_new_sync(struct nf_conntrack *ct, int origin)
 	nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES);
 	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
 
-	obj = cache_find(STATE(mode)->internal->data, ct, &id);
+	obj = cache_find(STATE(mode)->internal->ct.data, ct, &id);
 	if (obj == NULL) {
 retry:
-		obj = cache_object_new(STATE(mode)->internal->data, ct);
+		obj = cache_object_new(STATE(mode)->internal->ct.data, ct);
 		if (obj == NULL)
 			return;
-		if (cache_add(STATE(mode)->internal->data, obj, id) == -1) {
+		if (cache_add(STATE(mode)->internal->ct.data, obj, id) == -1) {
 			cache_object_free(obj);
 			return;
 		}
@@ -155,14 +157,13 @@ retry:
 		if (origin == CTD_ORIGIN_NOT_ME)
 			sync_send(obj, NET_T_STATE_NEW);
 	} else {
-		cache_del(STATE(mode)->internal->data, obj);
+		cache_del(STATE(mode)->internal->ct.data, obj);
 		cache_object_free(obj);
 		goto retry;
 	}
 }
 
-static void
-event_update_sync(struct nf_conntrack *ct, int origin)
+static void internal_cache_ct_event_upd(struct nf_conntrack *ct, int origin)
 {
 	struct cache_object *obj;
 
@@ -170,7 +171,7 @@ event_update_sync(struct nf_conntrack *ct, int origin)
 	if (origin == CTD_ORIGIN_INJECT)
 		return;
 
-	obj = cache_update_force(STATE(mode)->internal->data, ct);
+	obj = cache_update_force(STATE(mode)->internal->ct.data, ct);
 	if (obj == NULL)
 		return;
 
@@ -178,8 +179,7 @@ event_update_sync(struct nf_conntrack *ct, int origin)
 		sync_send(obj, NET_T_STATE_UPD);
 }
 
-static int
-event_destroy_sync(struct nf_conntrack *ct, int origin)
+static int internal_cache_ct_event_del(struct nf_conntrack *ct, int origin)
 {
 	struct cache_object *obj;
 	int id;
@@ -189,7 +189,7 @@ event_destroy_sync(struct nf_conntrack *ct, int origin)
 		return 0;
 
 	/* we don't synchronize events for objects that are not in the cache */
-	obj = cache_find(STATE(mode)->internal->data, ct, &id);
+	obj = cache_find(STATE(mode)->internal->ct.data, ct, &id);
 	if (obj == NULL)
 		return 0;
 
@@ -205,16 +205,18 @@ event_destroy_sync(struct nf_conntrack *ct, int origin)
 
 struct internal_handler internal_cache = {
 	.flags			= INTERNAL_F_POPULATE | INTERNAL_F_RESYNC,
-	.init			= _init,
-	.close			= _close,
-	.dump			= dump,
-	.flush			= flush,
-	.stats			= stats,
-	.stats_ext		= stats_ext,
-	.populate		= populate,
-	.purge			= purge,
-	.resync			= resync,
-	.new			= event_new_sync,
-	.update			= event_update_sync,
-	.destroy		= event_destroy_sync,
+	.init			= internal_cache_init,
+	.close			= internal_cache_close,
+	.ct = {
+		.dump			= internal_cache_ct_dump,
+		.flush			= internal_cache_ct_flush,
+		.stats			= internal_cache_ct_stats,
+		.stats_ext		= internal_cache_ct_stats_ext,
+		.populate		= internal_cache_ct_populate,
+		.purge			= internal_cache_ct_purge,
+		.resync			= internal_cache_ct_resync,
+		.new			= internal_cache_ct_event_new,
+		.upd			= internal_cache_ct_event_upd,
+		.del			= internal_cache_ct_event_del,
+	},
 };
diff --git a/src/run.c b/src/run.c
index 265a949..f8d3fad 100644
--- a/src/run.c
+++ b/src/run.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2009 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -241,8 +242,8 @@ static void do_overrun_resync_alarm(struct alarm_block *a, void *data)
 
 static void do_polling_alarm(struct alarm_block *a, void *data)
 {
-	if (STATE(mode)->internal->purge)
-		STATE(mode)->internal->purge();
+	if (STATE(mode)->internal->ct.purge)
+		STATE(mode)->internal->ct.purge();
 
 	nl_send_resync(STATE(resync));
 	add_alarm(&STATE(polling_alarm), CONFIG(poll_kernel_secs), 0);
@@ -267,13 +268,13 @@ static int event_handler(const struct nlmsghdr *nlh,
 
 	switch(type) {
 	case NFCT_T_NEW:
-		STATE(mode)->internal->new(ct, origin_type);
+		STATE(mode)->internal->ct.new(ct, origin_type);
 		break;
 	case NFCT_T_UPDATE:
-		STATE(mode)->internal->update(ct, origin_type);
+		STATE(mode)->internal->ct.upd(ct, origin_type);
 		break;
 	case NFCT_T_DESTROY:
-		if (STATE(mode)->internal->destroy(ct, origin_type))
+		if (STATE(mode)->internal->ct.del(ct, origin_type))
 			update_traffic_stats(ct);
 		break;
 	default:
@@ -298,7 +299,7 @@ static int dump_handler(enum nf_conntrack_msg_type type,
 
 	switch(type) {
 	case NFCT_T_UPDATE:
-		STATE(mode)->internal->populate(ct);
+		STATE(mode)->internal->ct.populate(ct);
 		break;
 	default:
 		STATE(stats).nl_dump_unknown_type++;
@@ -363,7 +364,7 @@ init(void)
 	}
 	nfct_callback_register(STATE(resync),
 			       NFCT_T_ALL,
-			       STATE(mode)->internal->resync,
+			       STATE(mode)->internal->ct.resync,
 			       NULL);
 	register_fd(nfct_fd(STATE(resync)), STATE(fds));
 	fcntl(nfct_fd(STATE(resync)), F_SETFL, O_NONBLOCK);
@@ -537,8 +538,8 @@ static void run_events(struct timeval *next_alarm)
 	/* we previously requested a resync due to buffer overrun. */
 	if (FD_ISSET(nfct_fd(STATE(resync)), &readfds)) {
 		nfct_catch(STATE(resync));
-		if (STATE(mode)->internal->purge)
-			STATE(mode)->internal->purge();
+		if (STATE(mode)->internal->ct.purge)
+			STATE(mode)->internal->ct.purge();
 	}
 
 	if (STATE(mode)->run)
diff --git a/src/stats-mode.c b/src/stats-mode.c
index 0403ce2..c7a81e3 100644
--- a/src/stats-mode.c
+++ b/src/stats-mode.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2007 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -37,7 +38,9 @@ static int init_stats(void)
 	}
 	memset(state.stats, 0, sizeof(struct ct_stats_state));
 
-	STATE_STATS(cache) = cache_create("stats", NO_FEATURES, NULL);
+	STATE_STATS(cache) = cache_create("stats", CACHE_T_CT,
+					  NO_FEATURES, NULL,
+					  &cache_stats_ct_ops);
 	if (!STATE_STATS(cache)) {
 		dlog(LOG_ERR, "can't allocate memory for the "
 			      "external cache");
@@ -88,7 +91,7 @@ static int local_handler_stats(int fd, int type, void *data)
 	return ret;
 }
 
-static void populate_stats(struct nf_conntrack *ct)
+static void stats_populate(struct nf_conntrack *ct)
 {
 	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
 	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS);
@@ -100,7 +103,7 @@ static void populate_stats(struct nf_conntrack *ct)
 	cache_update_force(STATE_STATS(cache), ct);
 }
 
-static int resync_stats(enum nf_conntrack_msg_type type,
+static int stats_resync(enum nf_conntrack_msg_type type,
 			struct nf_conntrack *ct,
 			void *data)
 {
@@ -125,23 +128,22 @@ static int purge_step(void *data1, void *data2)
 	struct cache_object *obj = data2;
 
 	STATE(get_retval) = 0;
-	nl_get_conntrack(STATE(get), obj->ct); /* modifies STATE(get_retval) */
+	nl_get_conntrack(STATE(get), obj->ptr); /* modifies STATE(get_retval) */
 	if (!STATE(get_retval)) {
 		cache_del(STATE_STATS(cache), obj);
-		dlog_ct(STATE(stats_log), obj->ct, NFCT_O_PLAIN);
+		dlog_ct(STATE(stats_log), obj->ptr, NFCT_O_PLAIN);
 		cache_object_free(obj);
 	}
 
 	return 0;
 }
 
-static void purge_stats(void)
+static void stats_purge(void)
 {
 	cache_iterate(STATE_STATS(cache), NULL, purge_step);
 }
 
-static void
-event_new_stats(struct nf_conntrack *ct, int origin)
+static void stats_event_new(struct nf_conntrack *ct, int origin)
 {
 	int id;
 	struct cache_object *obj;
@@ -162,15 +164,13 @@ event_new_stats(struct nf_conntrack *ct, int origin)
 	return;
 }
 
-static void
-event_update_stats(struct nf_conntrack *ct, int origin)
+static void stats_event_upd(struct nf_conntrack *ct, int origin)
 {
 	nfct_attr_unset(ct, ATTR_TIMEOUT);
 	cache_update_force(STATE_STATS(cache), ct);
 }
 
-static int
-event_destroy_stats(struct nf_conntrack *ct, int origin)
+static int stats_event_del(struct nf_conntrack *ct, int origin)
 {
 	int id;
 	struct cache_object *obj;
@@ -189,12 +189,14 @@ event_destroy_stats(struct nf_conntrack *ct, int origin)
 
 static struct internal_handler internal_cache_stats = {
 	.flags			= INTERNAL_F_POPULATE | INTERNAL_F_RESYNC,
-	.populate		= populate_stats,
-	.resync			= resync_stats,
-	.purge			= purge_stats,
-	.new			= event_new_stats,
-	.update			= event_update_stats,
-	.destroy		= event_destroy_stats
+	.ct = {
+		.populate		= stats_populate,
+		.resync			= stats_resync,
+		.purge			= stats_purge,
+		.new			= stats_event_new,
+		.upd			= stats_event_upd,
+		.del			= stats_event_del,
+	},
 };
 
 struct ct_mode stats_mode = {
diff --git a/src/sync-alarm.c b/src/sync-alarm.c
index b555dd5..8d6b34d 100644
--- a/src/sync-alarm.c
+++ b/src/sync-alarm.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2007 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -110,7 +111,7 @@ static int alarm_recv(const struct nethdr *net)
 static void alarm_enqueue(struct cache_object *obj, int query)
 {
 	struct cache_alarm *ca =
-		cache_get_extra(STATE(mode)->internal->data, obj);
+		cache_get_extra(STATE(mode)->internal->ct.data, obj);
 	if (queue_add(STATE_SYNC(tx_queue), &ca->qnode) > 0)
 		cache_object_get(obj);
 }
@@ -135,9 +136,9 @@ static int tx_queue_xmit(struct queue_node *n, const void *data)
 		int type;
 
 		ca = (struct cache_alarm *)n;
-		obj = cache_data_get_object(STATE(mode)->internal->data, ca);
+		obj = cache_data_get_object(STATE(mode)->internal->ct.data, ca);
 		type = object_status_to_network_type(obj->status);
-		net = BUILD_NETMSG(obj->ct, type);
+		net = obj->cache->ops->build_msg(obj, type);
 		multichannel_send(STATE_SYNC(channel), net);
 		cache_object_put(obj);
 		break;
diff --git a/src/sync-ftfw.c b/src/sync-ftfw.c
index 581b5ca..55eda0b 100644
--- a/src/sync-ftfw.c
+++ b/src/sync-ftfw.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2008 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -169,7 +170,7 @@ static int do_cache_to_tx(void *data1, void *data2)
 {
 	struct cache_object *obj = data2;
 	struct cache_ftfw *cn =
-		cache_get_extra(STATE(mode)->internal->data, obj);
+		cache_get_extra(STATE(mode)->internal->ct.data, obj);
 
 	if (queue_in(rs_queue, &cn->qnode)) {
 		queue_del(&cn->qnode);
@@ -227,7 +228,7 @@ static int ftfw_local(int fd, int type, void *data)
 		break;
 	case SEND_BULK:
 		dlog(LOG_NOTICE, "sending bulk update");
-		cache_iterate(STATE(mode)->internal->data,
+		cache_iterate(STATE(mode)->internal->ct.data,
 			      NULL, do_cache_to_tx);
 		break;
 	case STATS_RSQUEUE:
@@ -307,7 +308,7 @@ static int rs_queue_empty(struct queue_node *n, const void *data)
 		cn = (struct cache_ftfw *) n;
 		if (h == NULL) {
 			queue_del(n);
-			obj = cache_data_get_object(STATE(mode)->internal->data, cn);
+			obj = cache_data_get_object(STATE(mode)->internal->ct.data, cn);
 			cache_object_put(obj);
 			return 0;
 		}
@@ -318,7 +319,7 @@ static int rs_queue_empty(struct queue_node *n, const void *data)
 
 		dp("queue: deleting from queue (seq=%u)\n", cn->seq);
 		queue_del(n);
-		obj = cache_data_get_object(STATE(mode)->internal->data, cn);
+		obj = cache_data_get_object(STATE(mode)->internal->ct.data, cn);
 		cache_object_put(obj);
 		break;
 	}
@@ -351,7 +352,7 @@ static int digest_msg(const struct nethdr *net)
 
 	} else if (IS_RESYNC(net)) {
 		dp("RESYNC ALL\n");
-		cache_iterate(STATE(mode)->internal->data, NULL, do_cache_to_tx);
+		cache_iterate(STATE(mode)->internal->ct.data, NULL, do_cache_to_tx);
 		return MSG_CTL;
 
 	} else if (IS_ALIVE(net))
@@ -468,7 +469,7 @@ static void rs_queue_purge_full(void)
 		struct cache_object *obj;
 
 		cn = (struct cache_ftfw *)n;
-		obj = cache_data_get_object(STATE(mode)->internal->data, cn);
+		obj = cache_data_get_object(STATE(mode)->internal->ct.data, cn);
 		cache_object_put(obj);
 		break;
 	}
@@ -516,9 +517,9 @@ static int tx_queue_xmit(struct queue_node *n, const void *data)
 		struct nethdr *net;
 
 		cn = (struct cache_ftfw *)n;
-		obj = cache_data_get_object(STATE(mode)->internal->data, cn);
+		obj = cache_data_get_object(STATE(mode)->internal->ct.data, cn);
 		type = object_status_to_network_type(obj->status);
-		net = BUILD_NETMSG(obj->ct, type);
+		net = obj->cache->ops->build_msg(obj, type);
 		nethdr_set_hello(net);
 
 		dp("tx_list sq: %u fl:%u len:%u\n",
@@ -551,7 +552,7 @@ static void ftfw_xmit(void)
 static void ftfw_enqueue(struct cache_object *obj, int type)
 {
 	struct cache_ftfw *cn =
-		cache_get_extra(STATE(mode)->internal->data, obj);
+		cache_get_extra(STATE(mode)->internal->ct.data, obj);
 	if (queue_in(rs_queue, &cn->qnode)) {
 		queue_del(&cn->qnode);
 		queue_add(STATE_SYNC(tx_queue), &cn->qnode);
diff --git a/src/sync-mode.c b/src/sync-mode.c
index 5351110..34d9706 100644
--- a/src/sync-mode.c
+++ b/src/sync-mode.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2007 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -251,7 +252,7 @@ static void do_reset_cache_alarm(struct alarm_block *a, void *data)
 		exit(EXIT_SUCCESS);
 	}
 	/* this is not required if events don't get lost */
-	STATE(mode)->internal->flush();
+	STATE(mode)->internal->ct.flush();
 }
 
 static int init_sync(void)
@@ -471,7 +472,7 @@ static int local_handler_sync(int fd, int type, void *data)
 	switch(type) {
 	case DUMP_INTERNAL:
 		if (fork_process_new(CTD_PROC_ANY, 0, NULL, NULL) == 0) {
-			STATE(mode)->internal->dump(fd, NFCT_O_PLAIN);
+			STATE(mode)->internal->ct.dump(fd, NFCT_O_PLAIN);
 			exit(EXIT_SUCCESS);
 		}
 		break;
@@ -483,7 +484,7 @@ static int local_handler_sync(int fd, int type, void *data)
 		break;
 	case DUMP_INT_XML:
 		if (fork_process_new(CTD_PROC_ANY, 0, NULL, NULL) == 0) {
-			STATE(mode)->internal->dump(fd, NFCT_O_XML);
+			STATE(mode)->internal->ct.dump(fd, NFCT_O_XML);
 			exit(EXIT_SUCCESS);
 		}
 		break;
@@ -512,14 +513,14 @@ static int local_handler_sync(int fd, int type, void *data)
 		/* inmediate flush, remove pending flush scheduled if any */
 		del_alarm(&STATE_SYNC(reset_cache_alarm));
 		dlog(LOG_NOTICE, "flushing caches");
-		STATE(mode)->internal->flush();
+		STATE(mode)->internal->ct.flush();
 		STATE_SYNC(external)->flush();
 		break;
 	case FLUSH_INT_CACHE:
 		/* inmediate flush, remove pending flush scheduled if any */
 		del_alarm(&STATE_SYNC(reset_cache_alarm));
 		dlog(LOG_NOTICE, "flushing internal cache");
-		STATE(mode)->internal->flush();
+		STATE(mode)->internal->ct.flush();
 		break;
 	case FLUSH_EXT_CACHE:
 		dlog(LOG_NOTICE, "flushing external cache");
@@ -529,7 +530,7 @@ static int local_handler_sync(int fd, int type, void *data)
 		killer(0);
 		break;
 	case STATS:
-		STATE(mode)->internal->stats(fd);
+		STATE(mode)->internal->ct.stats(fd);
 		STATE_SYNC(external)->stats(fd);
 		dump_traffic_stats(fd);
 		multichannel_stats(STATE_SYNC(channel), fd);
@@ -540,7 +541,7 @@ static int local_handler_sync(int fd, int type, void *data)
 		multichannel_stats(STATE_SYNC(channel), fd);
 		break;
 	case STATS_CACHE:
-		STATE(mode)->internal->stats_ext(fd);
+		STATE(mode)->internal->ct.stats_ext(fd);
 		STATE_SYNC(external)->stats_ext(fd);
 		break;
 	case STATS_LINK:
diff --git a/src/sync-notrack.c b/src/sync-notrack.c
index 06af58b..e25cfd8 100644
--- a/src/sync-notrack.c
+++ b/src/sync-notrack.c
@@ -76,7 +76,7 @@ static int do_cache_to_tx(void *data1, void *data2)
 {
 	struct cache_object *obj = data2;
 	struct cache_notrack *cn =
-		cache_get_extra(STATE(mode)->internal->data, obj);
+		cache_get_extra(STATE(mode)->internal->ct.data, obj);
 	if (queue_add(STATE_SYNC(tx_queue), &cn->qnode) > 0)
 		cache_object_get(obj);
 	return 0;
@@ -127,7 +127,7 @@ static int notrack_local(int fd, int type, void *data)
 		if (CONFIG(sync).internal_cache_disable) {
 			kernel_resync();
 		} else {
-			cache_iterate(STATE(mode)->internal->data,
+			cache_iterate(STATE(mode)->internal->ct.data,
 				      NULL, do_cache_to_tx);
 		}
 		break;
@@ -148,7 +148,7 @@ static int digest_msg(const struct nethdr *net)
 		if (CONFIG(sync).internal_cache_disable) {
 			kernel_resync();
 		} else {
-			cache_iterate(STATE(mode)->internal->data,
+			cache_iterate(STATE(mode)->internal->ct.data,
 				      NULL, do_cache_to_tx);
 		}
 		return MSG_CTL;
@@ -197,9 +197,9 @@ static int tx_queue_xmit(struct queue_node *n, const void *data2)
 		struct nethdr *net;
 
 		cn = (struct cache_ftfw *)n;
-		obj = cache_data_get_object(STATE(mode)->internal->data, cn);
+		obj = cache_data_get_object(STATE(mode)->internal->ct.data, cn);
 		type = object_status_to_network_type(obj->status);;
-		net = BUILD_NETMSG(obj->ct, type);
+		net = obj->cache->ops->build_msg(obj, type);
 
 		multichannel_send(STATE_SYNC(channel), net);
 		queue_del(n);
@@ -219,7 +219,7 @@ static void notrack_xmit(void)
 static void notrack_enqueue(struct cache_object *obj, int query)
 {
 	struct cache_notrack *cn =
-		cache_get_extra(STATE(mode)->internal->data, obj);
+		cache_get_extra(STATE(mode)->internal->ct.data, obj);
 	if (queue_add(STATE_SYNC(tx_queue), &cn->qnode) > 0)
 		cache_object_get(obj);
 }