conntrackd: add `DisableInternalCache' clause

This patch adds the clause `DisableInternalCache' that allows you
to bypass the internal cache. This clause can only be used with
the notrack synchronization mode.

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

11 files changed, 529 insertions, 224 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 8b36642..76f0e73 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -22,6 +22,7 @@ conntrackd_SOURCES = alarm.c main.c run.c hash.c queue.c rbtree.c \
 		    channel.c multichannel.c channel_mcast.c channel_udp.c \
 		    tcp.c channel_tcp.c \
 		    external_cache.c external_inject.c \
+		    internal_cache.c internal_bypass.c \ 
 		    read_config_yy.y read_config_lex.l
 
 # yacc and lex generate dirty code
diff --git a/src/internal_bypass.c b/src/internal_bypass.c
new file mode 100644
index 0000000..4caaf4f
--- /dev/null
+++ b/src/internal_bypass.c
@@ -0,0 +1,165 @@
+/*
+ * (C) 2009 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * 
+ * 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
+ * (at your option) any later version.
+ *
+ * This feature has been sponsored by 6WIND <www.6wind.com>.
+ */
+#include "conntrackd.h"
+#include "sync.h"
+#include "log.h"
+#include "cache.h"
+#include "netlink.h"
+#include "network.h"
+#include "origin.h"
+
+static int _init(void)
+{
+	return 0;
+}
+
+static void _close(void)
+{
+}
+
+static int dump_cb(enum nf_conntrack_msg_type type,
+		   struct nf_conntrack *ct, void *data)
+{
+	char buf[1024];
+	int size, *fd = data;
+
+	size = nfct_snprintf(buf, 1024, ct, NFCT_T_UNKNOWN, NFCT_O_DEFAULT, 0);
+	if (size < 1024) {
+		buf[size] = '\n';
+		size++;
+	}
+	send(*fd, buf, size, 0);
+
+	return NFCT_CB_CONTINUE;
+}
+
+static void dump(int fd, int type)
+{
+	struct nfct_handle *h;
+	u_int32_t family = AF_UNSPEC;
+	int ret;
+
+	h = nfct_open(CONNTRACK, 0);
+	if (h == NULL) {
+		dlog(LOG_ERR, "can't allocate memory for the internal cache");
+		return;
+	}
+	nfct_callback_register(h, NFCT_T_ALL, dump_cb, &fd);
+	ret = nfct_query(h, NFCT_Q_DUMP, &family);
+	if (ret == -1) {
+		dlog(LOG_ERR, "can't dump kernel table");
+	}
+	nfct_close(h);
+}
+
+static void flush(void)
+{
+	nl_flush_conntrack_table(STATE(flush));
+}
+
+struct {
+	uint32_t	new;
+	uint32_t	upd;
+	uint32_t	del;
+} internal_bypass_stats;
+
+static void stats(int fd)
+{
+	char buf[512];
+	int size;
+
+	size = sprintf(buf, "internal bypass:\n"
+			    "connections new:\t\t%12u\n"
+			    "connections updated:\t\t%12u\n"
+			    "connections destroyed:\t\t%12u\n\n",
+			    internal_bypass_stats.new,
+			    internal_bypass_stats.upd,
+			    internal_bypass_stats.del);
+
+	send(fd, buf, size, 0);
+}
+
+/* unused, INTERNAL_F_POPULATE is unset. No cache, nothing to populate. */
+static void populate(struct nf_conntrack *ct)
+{
+}
+
+/* unused, INTERNAL_F_RESYNC is unset. */
+static void 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)
+{
+	return NFCT_CB_CONTINUE;
+}
+
+static void
+event_new_sync(struct nf_conntrack *ct, int origin)
+{
+	struct nethdr *net;
+
+	/* this event has been triggered by me, skip */
+	if (origin != CTD_ORIGIN_NOT_ME)
+		return;
+
+	net = BUILD_NETMSG(ct, NET_T_STATE_NEW);
+	multichannel_send(STATE_SYNC(channel), net);
+	internal_bypass_stats.new++;
+}
+
+static void
+event_update_sync(struct nf_conntrack *ct, int origin)
+{
+	struct nethdr *net;
+
+	/* this event has been triggered by me, skip */
+	if (origin != CTD_ORIGIN_NOT_ME)
+		return;
+
+	net = BUILD_NETMSG(ct, NET_T_STATE_UPD);
+	multichannel_send(STATE_SYNC(channel), net);
+	internal_bypass_stats.upd++;
+}
+
+static int
+event_destroy_sync(struct nf_conntrack *ct, int origin)
+{
+	struct nethdr *net;
+
+	/* this event has been triggered by me, skip */
+	if (origin != CTD_ORIGIN_NOT_ME)
+		return 1;
+
+	net = BUILD_NETMSG(ct, NET_T_STATE_DEL);
+	multichannel_send(STATE_SYNC(channel), net);
+	internal_bypass_stats.del++;
+
+	return 1;
+}
+
+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,
+};
diff --git a/src/internal_cache.c b/src/internal_cache.c
new file mode 100644
index 0000000..daadfd6
--- /dev/null
+++ b/src/internal_cache.c
@@ -0,0 +1,220 @@
+/*
+ * (C) 2009 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * 
+ * 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
+ * (at your option) any later version.
+ */
+#include "conntrackd.h"
+#include "sync.h"
+#include "log.h"
+#include "cache.h"
+#include "netlink.h"
+#include "network.h"
+#include "origin.h"
+
+static inline void sync_send(struct cache_object *obj, int query)
+{
+	STATE_SYNC(sync)->enqueue(obj, query);
+}
+
+static int _init(void)
+{
+	STATE(mode)->internal->data =
+		cache_create("internal", 
+			     STATE_SYNC(sync)->internal_cache_flags,
+			     STATE_SYNC(sync)->internal_cache_extra);
+
+	if (!STATE(mode)->internal->data) {
+		dlog(LOG_ERR, "can't allocate memory for the internal cache");
+		return -1;
+	}
+	return 0;
+}
+
+static void _close(void)
+{
+	cache_destroy(STATE(mode)->internal->data);
+}
+
+static void dump(int fd, int type)
+{
+	cache_dump(STATE(mode)->internal->data, fd, NFCT_O_PLAIN);
+}
+
+static void flush(void)
+{
+	cache_flush(STATE(mode)->internal->data);
+}
+
+static void stats(int fd)
+{
+	cache_stats(STATE(mode)->internal->data, fd);
+}
+
+static void stats_ext(int fd)
+{
+	cache_stats_extended(STATE(mode)->internal->data, fd);
+}
+
+static void populate(struct nf_conntrack *ct)
+{
+	/* This is required by kernels < 2.6.20 */
+	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
+	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS);
+	nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES);
+	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
+	nfct_attr_unset(ct, ATTR_USE);
+
+	cache_update_force(STATE(mode)->internal->data, ct);
+}
+
+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_reval) */
+	if (!STATE(get_retval)) {
+		if (obj->status != C_OBJ_DEAD) {
+			cache_object_set_status(obj, C_OBJ_DEAD);
+			sync_send(obj, NET_T_STATE_DEL);
+			cache_object_put(obj);
+		}
+	}
+
+	return 0;
+}
+
+static void purge(void)
+{
+	cache_iterate(STATE(mode)->internal->data, NULL, purge_step);
+}
+
+static int resync(enum nf_conntrack_msg_type type,
+		  struct nf_conntrack *ct,
+		  void *data)
+{
+	struct cache_object *obj;
+
+	if (ct_filter_conntrack(ct, 1))
+		return NFCT_CB_CONTINUE;
+
+	/* This is required by kernels < 2.6.20 */
+	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
+	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS);
+	nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES);
+	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
+	nfct_attr_unset(ct, ATTR_USE);
+
+	obj = cache_update_force(STATE(mode)->internal->data, ct);
+	if (obj == NULL)
+		return NFCT_CB_CONTINUE;
+
+	switch (obj->status) {
+	case C_OBJ_NEW:
+		sync_send(obj, NET_T_STATE_NEW);
+		break;
+	case C_OBJ_ALIVE:
+		sync_send(obj, NET_T_STATE_UPD);
+		break;
+	}
+	return NFCT_CB_CONTINUE;
+}
+
+static void
+event_new_sync(struct nf_conntrack *ct, int origin)
+{
+	struct cache_object *obj;
+	int id;
+
+	/* this event has been triggered by a direct inject, skip */
+	if (origin == CTD_ORIGIN_INJECT)
+		return;
+
+	/* required by linux kernel <= 2.6.20 */
+	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
+	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS);
+	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);
+	if (obj == NULL) {
+retry:
+		obj = cache_object_new(STATE(mode)->internal->data, ct);
+		if (obj == NULL)
+			return;
+		if (cache_add(STATE(mode)->internal->data, obj, id) == -1) {
+			cache_object_free(obj);
+			return;
+		}
+		/* only synchronize events that have been triggered by other
+		 * processes or the kernel, but don't propagate events that
+		 * have been triggered by conntrackd itself, eg. commits. */
+		if (origin == CTD_ORIGIN_NOT_ME)
+			sync_send(obj, NET_T_STATE_NEW);
+	} else {
+		cache_del(STATE(mode)->internal->data, obj);
+		cache_object_free(obj);
+		goto retry;
+	}
+}
+
+static void
+event_update_sync(struct nf_conntrack *ct, int origin)
+{
+	struct cache_object *obj;
+
+	/* this event has been triggered by a direct inject, skip */
+	if (origin == CTD_ORIGIN_INJECT)
+		return;
+
+	obj = cache_update_force(STATE(mode)->internal->data, ct);
+	if (obj == NULL)
+		return;
+
+	if (origin == CTD_ORIGIN_NOT_ME)
+		sync_send(obj, NET_T_STATE_UPD);
+}
+
+static int
+event_destroy_sync(struct nf_conntrack *ct, int origin)
+{
+	struct cache_object *obj;
+	int id;
+
+	/* this event has been triggered by a direct inject, skip */
+	if (origin == CTD_ORIGIN_INJECT)
+		return 0;
+
+	/* we don't synchronize events for objects that are not in the cache */
+	obj = cache_find(STATE(mode)->internal->data, ct, &id);
+	if (obj == NULL)
+		return 0;
+
+	if (obj->status != C_OBJ_DEAD) {
+		cache_object_set_status(obj, C_OBJ_DEAD);
+		if (origin == CTD_ORIGIN_NOT_ME) {
+			sync_send(obj, NET_T_STATE_DEL);
+		}
+		cache_object_put(obj);
+	}
+	return 1;
+}
+
+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,
+};
diff --git a/src/read_config_lex.l b/src/read_config_lex.l
index b4be6f0..b2d4bdb 100644
--- a/src/read_config_lex.l
+++ b/src/read_config_lex.l
@@ -136,6 +136,7 @@ notrack		[N|n][O|o][T|t][R|r][A|a][C|c][K|k]
 "Type"				{ return T_TYPE; }
 "Priority"			{ return T_PRIO; }
 "NetlinkEventsReliable"		{ return T_NETLINK_EVENTS_RELIABLE; }
+"DisableInternalCache"		{ return T_DISABLE_INTERNAL_CACHE; }
 "DisableExternalCache"		{ return T_DISABLE_EXTERNAL_CACHE; }
 "ErrorQueueLength"		{ return T_ERROR_QUEUE_LENGTH; }
 
diff --git a/src/read_config_yy.y b/src/read_config_yy.y
index 5075cf0..157e945 100644
--- a/src/read_config_yy.y
+++ b/src/read_config_yy.y
@@ -72,7 +72,7 @@ static void __max_dedicated_links_reached(void);
 %token T_FROM T_USERSPACE T_KERNELSPACE T_EVENT_ITER_LIMIT T_DEFAULT
 %token T_NETLINK_OVERRUN_RESYNC T_NICE T_IPV4_DEST_ADDR T_IPV6_DEST_ADDR
 %token T_SCHEDULER T_TYPE T_PRIO T_NETLINK_EVENTS_RELIABLE
-%token T_DISABLE_EXTERNAL_CACHE T_ERROR_QUEUE_LENGTH
+%token T_DISABLE_INTERNAL_CACHE T_DISABLE_EXTERNAL_CACHE T_ERROR_QUEUE_LENGTH
 
 %token <string> T_IP T_PATH_VAL
 %token <val> T_NUMBER
@@ -852,9 +852,20 @@ sync_mode_notrack_list:
 
 sync_mode_notrack_line: timeout
 		      | purge
+		      | disable_internal_cache
 		      | disable_external_cache
 		      ;
 
+disable_internal_cache: T_DISABLE_INTERNAL_CACHE T_ON
+{
+	conf.sync.internal_cache_disable = 1;
+};
+
+disable_internal_cache: T_DISABLE_INTERNAL_CACHE T_OFF
+{
+	conf.sync.internal_cache_disable = 0;
+};
+
 disable_external_cache: T_DISABLE_EXTERNAL_CACHE T_ON
 {
 	conf.sync.external_cache_disable = 1;
diff --git a/src/run.c b/src/run.c
index 54ab1a5..803bbcc 100644
--- a/src/run.c
+++ b/src/run.c
@@ -28,6 +28,7 @@
 #include "process.h"
 #include "origin.h"
 #include "date.h"
+#include "internal.h"
 
 #include <errno.h>
 #include <signal.h>
@@ -56,7 +57,9 @@ void killer(int foo)
 	local_server_destroy(&STATE(local));
 	STATE(mode)->kill();
 
-	nfct_close(STATE(dump));	/* cache_wt needs this here */
+	if (STATE(mode)->internal->flags & INTERNAL_F_POPULATE) {
+		nfct_close(STATE(dump));
+	}
 	destroy_fds(STATE(fds)); 
 
 	unlink(CONFIG(lockfile));
@@ -210,9 +213,13 @@ static int local_handler(int fd, void *data)
 		}
 		break;
 	case RESYNC_MASTER:
-		STATE(stats).nl_kernel_table_resync++;
-		dlog(LOG_NOTICE, "resync with master table");
-		nl_dump_conntrack_table(STATE(dump));
+		if (STATE(mode)->internal->flags & INTERNAL_F_POPULATE) {
+			STATE(stats).nl_kernel_table_resync++;
+			dlog(LOG_NOTICE, "resync with master table");
+			nl_dump_conntrack_table(STATE(dump));
+		} else {
+			dlog(LOG_NOTICE, "resync is unsupported in this mode");
+		}
 		break;
 	case STATS_RUNTIME:
 		dump_stats_runtime(fd);
@@ -238,8 +245,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)->purge)
-		STATE(mode)->purge();
+	if (STATE(mode)->internal->purge)
+		STATE(mode)->internal->purge();
 
 	nl_send_resync(STATE(resync));
 	add_alarm(&STATE(polling_alarm), CONFIG(poll_kernel_secs), 0);
@@ -264,13 +271,13 @@ static int event_handler(const struct nlmsghdr *nlh,
 
 	switch(type) {
 	case NFCT_T_NEW:
-		STATE(mode)->event_new(ct, origin_type);
+		STATE(mode)->internal->new(ct, origin_type);
 		break;
 	case NFCT_T_UPDATE:
-		STATE(mode)->event_upd(ct, origin_type);
+		STATE(mode)->internal->update(ct, origin_type);
 		break;
 	case NFCT_T_DESTROY:
-		if (STATE(mode)->event_dst(ct, origin_type))
+		if (STATE(mode)->internal->destroy(ct, origin_type))
 			update_traffic_stats(ct);
 		break;
 	default:
@@ -295,7 +302,7 @@ static int dump_handler(enum nf_conntrack_msg_type type,
 
 	switch(type) {
 	case NFCT_T_UPDATE:
-		STATE(mode)->dump(ct);
+		STATE(mode)->internal->populate(ct);
 		break;
 	default:
 		STATE(stats).nl_dump_unknown_type++;
@@ -371,23 +378,26 @@ init(void)
 	}
 	nfct_callback_register(STATE(resync),
 			       NFCT_T_ALL,
-			       STATE(mode)->resync,
+			       STATE(mode)->internal->resync,
 			       NULL);
 	register_fd(nfct_fd(STATE(resync)), STATE(fds));
 	fcntl(nfct_fd(STATE(resync)), F_SETFL, O_NONBLOCK);
 
-	STATE(dump) = nfct_open(CONNTRACK, 0);
-	if (STATE(dump) == NULL) {
-		dlog(LOG_ERR, "can't open netlink handler: %s",
-		     strerror(errno));
-		dlog(LOG_ERR, "no ctnetlink kernel support?");
-		return -1;
-	}
-	nfct_callback_register(STATE(dump), NFCT_T_ALL, dump_handler, NULL);
+	if (STATE(mode)->internal->flags & INTERNAL_F_POPULATE) {
+		STATE(dump) = nfct_open(CONNTRACK, 0);
+		if (STATE(dump) == NULL) {
+			dlog(LOG_ERR, "can't open netlink handler: %s",
+			     strerror(errno));
+			dlog(LOG_ERR, "no ctnetlink kernel support?");
+			return -1;
+		}
+		nfct_callback_register(STATE(dump), NFCT_T_ALL,
+				       dump_handler, NULL);
 
-	if (nl_dump_conntrack_table(STATE(dump)) == -1) {
-		dlog(LOG_ERR, "can't get kernel conntrack table");
-		return -1;
+		if (nl_dump_conntrack_table(STATE(dump)) == -1) {
+			dlog(LOG_ERR, "can't get kernel conntrack table");
+			return -1;
+		}
 	}
 
 	STATE(get) = nfct_open(CONNTRACK, 0);
@@ -499,7 +509,9 @@ static void __run(struct timeval *next_alarm)
 				 *    we resync ourselves.
 				 */
 				nl_resize_socket_buffer(STATE(event));
-				if (CONFIG(nl_overrun_resync) > 0) {
+				if (CONFIG(nl_overrun_resync) > 0 &&
+				    STATE(mode)->internal->flags &
+				    			INTERNAL_F_RESYNC) {
 					add_alarm(&STATE(resync_alarm),
 						  CONFIG(nl_overrun_resync),0);
 				}
@@ -523,8 +535,8 @@ static void __run(struct timeval *next_alarm)
 		}
 		if (FD_ISSET(nfct_fd(STATE(resync)), &readfds)) {
 			nfct_catch(STATE(resync));
-			if (STATE(mode)->purge)
-				STATE(mode)->purge();
+			if (STATE(mode)->internal->purge)
+				STATE(mode)->internal->purge();
 		}
 	} else {
 		/* using polling mode */
diff --git a/src/stats-mode.c b/src/stats-mode.c
index 5cfb638..0403ce2 100644
--- a/src/stats-mode.c
+++ b/src/stats-mode.c
@@ -21,6 +21,7 @@
 #include "cache.h"
 #include "log.h"
 #include "conntrackd.h"
+#include "internal.h"
 
 #include <errno.h>
 #include <string.h>
@@ -87,7 +88,7 @@ static int local_handler_stats(int fd, int type, void *data)
 	return ret;
 }
 
-static void dump_stats(struct nf_conntrack *ct)
+static void populate_stats(struct nf_conntrack *ct)
 {
 	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
 	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS);
@@ -134,11 +135,9 @@ static int purge_step(void *data1, void *data2)
 	return 0;
 }
 
-static int purge_stats(void)
+static void purge_stats(void)
 {
 	cache_iterate(STATE_STATS(cache), NULL, purge_step);
-
-	return 0;
 }
 
 static void
@@ -188,15 +187,20 @@ event_destroy_stats(struct nf_conntrack *ct, int origin)
 	return 0;
 }
 
+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
+};
+
 struct ct_mode stats_mode = {
 	.init 			= init_stats,
 	.run			= NULL,
 	.local			= local_handler_stats,
 	.kill			= kill_stats,
-	.dump			= dump_stats,
-	.resync			= resync_stats,
-	.purge			= purge_stats,
-	.event_new		= event_new_stats,
-	.event_upd		= event_update_stats,
-	.event_dst		= event_destroy_stats
+	.internal		= &internal_cache_stats,
 };
diff --git a/src/sync-alarm.c b/src/sync-alarm.c
index 4757026..0fc7943 100644
--- a/src/sync-alarm.c
+++ b/src/sync-alarm.c
@@ -109,7 +109,8 @@ 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_SYNC(internal), obj);
+	struct cache_alarm *ca =
+		cache_get_extra(STATE(mode)->internal->data, obj);
 	if (queue_add(STATE_SYNC(tx_queue), &ca->qnode))
 		cache_object_get(obj);
 }
@@ -134,7 +135,7 @@ 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_SYNC(internal), ca);
+		obj = cache_data_get_object(STATE(mode)->internal->data, ca);
 		type = object_status_to_network_type(obj->status);
 		net = BUILD_NETMSG(obj->ct, type);
 		multichannel_send(STATE_SYNC(channel), net);
diff --git a/src/sync-ftfw.c b/src/sync-ftfw.c
index 0d31e17..86edeab 100644
--- a/src/sync-ftfw.c
+++ b/src/sync-ftfw.c
@@ -166,7 +166,8 @@ static void ftfw_kill(void)
 static int do_cache_to_tx(void *data1, void *data2)
 {
 	struct cache_object *obj = data2;
-	struct cache_ftfw *cn = cache_get_extra(STATE_SYNC(internal), obj);
+	struct cache_ftfw *cn =
+		cache_get_extra(STATE(mode)->internal->data, obj);
 
 	if (queue_in(rs_queue, &cn->qnode)) {
 		queue_del(&cn->qnode);
@@ -224,7 +225,8 @@ static int ftfw_local(int fd, int type, void *data)
 		break;
 	case SEND_BULK:
 		dlog(LOG_NOTICE, "sending bulk update");
-		cache_iterate(STATE_SYNC(internal), NULL, do_cache_to_tx);
+		cache_iterate(STATE(mode)->internal->data,
+			      NULL, do_cache_to_tx);
 		break;
 	case STATS_RSQUEUE:
 		ftfw_local_queue(fd);
@@ -303,7 +305,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_SYNC(internal), cn);
+			obj = cache_data_get_object(STATE(mode)->internal->data, cn);
 			cache_object_put(obj);
 			return 0;
 		}
@@ -314,7 +316,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_SYNC(internal), cn);
+		obj = cache_data_get_object(STATE(mode)->internal->data, cn);
 		cache_object_put(obj);
 		break;
 	}
@@ -347,7 +349,7 @@ static int digest_msg(const struct nethdr *net)
 
 	} else if (IS_RESYNC(net)) {
 		dp("RESYNC ALL\n");
-		cache_iterate(STATE_SYNC(internal), NULL, do_cache_to_tx);
+		cache_iterate(STATE(mode)->internal->data, NULL, do_cache_to_tx);
 		return MSG_CTL;
 
 	} else if (IS_ALIVE(net))
@@ -464,7 +466,7 @@ static void rs_queue_purge_full(void)
 		struct cache_object *obj;
 
 		cn = (struct cache_ftfw *)n;
-		obj = cache_data_get_object(STATE_SYNC(internal), cn);
+		obj = cache_data_get_object(STATE(mode)->internal->data, cn);
 		cache_object_put(obj);
 		break;
 	}
@@ -512,7 +514,7 @@ 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_SYNC(internal), cn);
+		obj = cache_data_get_object(STATE(mode)->internal->data, cn);
 		type = object_status_to_network_type(obj->status);
 		net = BUILD_NETMSG(obj->ct, type);
 		nethdr_set_hello(net);
@@ -546,7 +548,8 @@ static void ftfw_xmit(void)
 
 static void ftfw_enqueue(struct cache_object *obj, int type)
 {
-	struct cache_ftfw *cn = cache_get_extra(STATE_SYNC(internal), obj);
+	struct cache_ftfw *cn =
+		cache_get_extra(STATE(mode)->internal->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 63fae68..ecc2f0d 100644
--- a/src/sync-mode.c
+++ b/src/sync-mode.c
@@ -28,6 +28,7 @@
 #include "queue.h"
 #include "process.h"
 #include "origin.h"
+#include "internal.h"
 #include "external.h"
 
 #include <errno.h>
@@ -246,7 +247,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 */
-	cache_flush(STATE_SYNC(internal));
+	STATE(mode)->internal->flush();
 }
 
 static int init_sync(void)
@@ -276,15 +277,15 @@ static int init_sync(void)
 	if (STATE_SYNC(sync)->init)
 		STATE_SYNC(sync)->init();
 
-	STATE_SYNC(internal) =
-		cache_create("internal", 
-			     STATE_SYNC(sync)->internal_cache_flags,
-			     STATE_SYNC(sync)->internal_cache_extra);
+	if (CONFIG(sync).internal_cache_disable == 0) {
+		STATE(mode)->internal = &internal_cache;
+	} else {
+		STATE(mode)->internal = &internal_bypass;
+		dlog(LOG_NOTICE, "disabling internal cache");
 
-	if (!STATE_SYNC(internal)) {
-		dlog(LOG_ERR, "can't allocate memory for the internal cache");
-		return -1;
 	}
+	if (STATE(mode)->internal->init() == -1)
+		return -1;
 
 	if (CONFIG(sync).external_cache_disable == 0) {
 		STATE_SYNC(external) = &external_cache;
@@ -389,7 +390,7 @@ static void run_sync(fd_set *readfds)
 
 static void kill_sync(void)
 {
-	cache_destroy(STATE_SYNC(internal));
+	STATE(mode)->internal->close();
 	STATE_SYNC(external)->close();
 
 	multichannel_close(STATE_SYNC(channel));
@@ -466,7 +467,7 @@ static int local_handler_sync(int fd, int type, void *data)
 	case DUMP_INTERNAL:
 		ret = fork_process_new(CTD_PROC_ANY, 0, NULL, NULL);
 		if (ret == 0) {
-			cache_dump(STATE_SYNC(internal), fd, NFCT_O_PLAIN);
+			STATE(mode)->internal->dump(fd, NFCT_O_PLAIN);
 			exit(EXIT_SUCCESS);
 		}
 		break;
@@ -480,7 +481,7 @@ static int local_handler_sync(int fd, int type, void *data)
 	case DUMP_INT_XML:
 		ret = fork_process_new(CTD_PROC_ANY, 0, NULL, NULL);
 		if (ret == 0) {
-			cache_dump(STATE_SYNC(internal), fd, NFCT_O_XML);
+			STATE(mode)->internal->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");
-		cache_flush(STATE_SYNC(internal));
+		STATE(mode)->internal->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");
-		cache_flush(STATE_SYNC(internal));
+		STATE(mode)->internal->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:
-		cache_stats(STATE_SYNC(internal), fd);
+		STATE(mode)->internal->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:
-		cache_stats_extended(STATE_SYNC(internal), fd);
+		STATE(mode)->internal->stats_ext(fd);
 		STATE_SYNC(external)->stats_ext(fd);
 		break;
 	case STATS_LINK:
@@ -559,167 +560,10 @@ static int local_handler_sync(int fd, int type, void *data)
 	return ret;
 }
 
-static void sync_send(struct cache_object *obj, int query)
-{
-	STATE_SYNC(sync)->enqueue(obj, query);
-}
-
-static void dump_sync(struct nf_conntrack *ct)
-{
-	/* This is required by kernels < 2.6.20 */
-	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
-	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS);
-	nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES);
-	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
-	nfct_attr_unset(ct, ATTR_USE);
-
-	cache_update_force(STATE_SYNC(internal), ct);
-}
-
-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_reval) */
-	if (!STATE(get_retval)) {
-		if (obj->status != C_OBJ_DEAD) {
-			cache_object_set_status(obj, C_OBJ_DEAD);
-			sync_send(obj, NET_T_STATE_DEL);
-			cache_object_put(obj);
-		}
-	}
-
-	return 0;
-}
-
-static int purge_sync(void)
-{
-	cache_iterate(STATE_SYNC(internal), NULL, purge_step);
-
-	return 0;
-}
-
-static int resync_sync(enum nf_conntrack_msg_type type,
-		       struct nf_conntrack *ct,
-		       void *data)
-{
-	struct cache_object *obj;
-
-	if (ct_filter_conntrack(ct, 1))
-		return NFCT_CB_CONTINUE;
-
-	/* This is required by kernels < 2.6.20 */
-	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
-	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS);
-	nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES);
-	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
-	nfct_attr_unset(ct, ATTR_USE);
-
-	obj = cache_update_force(STATE_SYNC(internal), ct);
-	if (obj == NULL)
-		return NFCT_CB_CONTINUE;
-
-	switch (obj->status) {
-	case C_OBJ_NEW:
-		sync_send(obj, NET_T_STATE_NEW);
-		break;
-	case C_OBJ_ALIVE:
-		sync_send(obj, NET_T_STATE_UPD);
-		break;
-	}
-	return NFCT_CB_CONTINUE;
-}
-
-static void
-event_new_sync(struct nf_conntrack *ct, int origin)
-{
-	struct cache_object *obj;
-	int id;
-
-	/* this event has been triggered by a direct inject, skip */
-	if (origin == CTD_ORIGIN_INJECT)
-		return;
-
-	/* required by linux kernel <= 2.6.20 */
-	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_BYTES);
-	nfct_attr_unset(ct, ATTR_ORIG_COUNTER_PACKETS);
-	nfct_attr_unset(ct, ATTR_REPL_COUNTER_BYTES);
-	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
-
-	obj = cache_find(STATE_SYNC(internal), ct, &id);
-	if (obj == NULL) {
-retry:
-		obj = cache_object_new(STATE_SYNC(internal), ct);
-		if (obj == NULL)
-			return;
-		if (cache_add(STATE_SYNC(internal), obj, id) == -1) {
-			cache_object_free(obj);
-			return;
-		}
-		/* only synchronize events that have been triggered by other
-		 * processes or the kernel, but don't propagate events that
-		 * have been triggered by conntrackd itself, eg. commits. */
-		if (origin == CTD_ORIGIN_NOT_ME)
-			sync_send(obj, NET_T_STATE_NEW);
-	} else {
-		cache_del(STATE_SYNC(internal), obj);
-		cache_object_free(obj);
-		goto retry;
-	}
-}
-
-static void
-event_update_sync(struct nf_conntrack *ct, int origin)
-{
-	struct cache_object *obj;
-
-	/* this event has been triggered by a direct inject, skip */
-	if (origin == CTD_ORIGIN_INJECT)
-		return;
-
-	obj = cache_update_force(STATE_SYNC(internal), ct);
-	if (obj == NULL)
-		return;
-
-	if (origin == CTD_ORIGIN_NOT_ME)
-		sync_send(obj, NET_T_STATE_UPD);
-}
-
-static int
-event_destroy_sync(struct nf_conntrack *ct, int origin)
-{
-	struct cache_object *obj;
-	int id;
-
-	/* this event has been triggered by a direct inject, skip */
-	if (origin == CTD_ORIGIN_INJECT)
-		return 0;
-
-	/* we don't synchronize events for objects that are not in the cache */
-	obj = cache_find(STATE_SYNC(internal), ct, &id);
-	if (obj == NULL)
-		return 0;
-
-	if (obj->status != C_OBJ_DEAD) {
-		cache_object_set_status(obj, C_OBJ_DEAD);
-		if (origin == CTD_ORIGIN_NOT_ME) {
-			sync_send(obj, NET_T_STATE_DEL);
-		}
-		cache_object_put(obj);
-	}
-	return 1;
-}
-
 struct ct_mode sync_mode = {
 	.init 			= init_sync,
 	.run			= run_sync,
 	.local			= local_handler_sync,
 	.kill			= kill_sync,
-	.dump			= dump_sync,
-	.resync			= resync_sync,
-	.purge			= purge_sync,
-	.event_new		= event_new_sync,
-	.event_upd		= event_update_sync,
-	.event_dst		= event_destroy_sync
+	/* the internal handler is set in run-time. */
 };
diff --git a/src/sync-notrack.c b/src/sync-notrack.c
index d9f273e..c4ad941 100644
--- a/src/sync-notrack.c
+++ b/src/sync-notrack.c
@@ -74,12 +74,44 @@ static void tx_queue_add_ctlmsg(uint32_t flags, uint32_t from, uint32_t to)
 static int do_cache_to_tx(void *data1, void *data2)
 {
 	struct cache_object *obj = data2;
-	struct cache_notrack *cn = cache_get_extra(STATE_SYNC(internal), obj);
+	struct cache_notrack *cn =
+		cache_get_extra(STATE(mode)->internal->data, obj);
 	if (queue_add(STATE_SYNC(tx_queue), &cn->qnode))
 		cache_object_get(obj);
 	return 0;
 }
 
+static int kernel_resync_cb(enum nf_conntrack_msg_type type,
+			    struct nf_conntrack *ct, void *data)
+{
+	struct nethdr *net;
+
+	net = BUILD_NETMSG(ct, NET_T_STATE_NEW);
+	multichannel_send(STATE_SYNC(channel), net);
+
+	return NFCT_CB_CONTINUE;
+}
+
+/* Only used if the internal cache is disabled. */
+static void kernel_resync(void)
+{
+	struct nfct_handle *h;
+	u_int32_t family = AF_UNSPEC;
+	int ret;
+
+	h = nfct_open(CONNTRACK, 0);
+	if (h == NULL) {
+		dlog(LOG_ERR, "can't allocate memory for the internal cache");
+		return;
+	}
+	nfct_callback_register(h, NFCT_T_ALL, kernel_resync_cb, NULL);
+	ret = nfct_query(h, NFCT_Q_DUMP, &family);
+	if (ret == -1) {
+		dlog(LOG_ERR, "can't dump kernel table");
+	}
+	nfct_close(h);
+}
+
 static int notrack_local(int fd, int type, void *data)
 {
 	int ret = LOCAL_RET_OK;
@@ -91,7 +123,12 @@ static int notrack_local(int fd, int type, void *data)
 		break;
 	case SEND_BULK:
 		dlog(LOG_NOTICE, "sending bulk update");
-		cache_iterate(STATE_SYNC(internal), NULL, do_cache_to_tx);
+		if (CONFIG(sync).internal_cache_disable) {
+			kernel_resync();
+		} else {
+			cache_iterate(STATE(mode)->internal->data,
+				      NULL, do_cache_to_tx);
+		}
 		break;
 	default:
 		ret = 0;
@@ -107,7 +144,12 @@ static int digest_msg(const struct nethdr *net)
 		return MSG_DATA;
 
 	if (IS_RESYNC(net)) {
-		cache_iterate(STATE_SYNC(internal), NULL, do_cache_to_tx);
+		if (CONFIG(sync).internal_cache_disable) {
+			kernel_resync();
+		} else {
+			cache_iterate(STATE(mode)->internal->data,
+				      NULL, do_cache_to_tx);
+		}
 		return MSG_CTL;
 	}
 
@@ -154,7 +196,7 @@ 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_SYNC(internal), cn);
+		obj = cache_data_get_object(STATE(mode)->internal->data, cn);
 		type = object_status_to_network_type(obj->status);;
 		net = BUILD_NETMSG(obj->ct, type);
 
@@ -175,7 +217,8 @@ static void notrack_xmit(void)
 
 static void notrack_enqueue(struct cache_object *obj, int query)
 {
-	struct cache_notrack *cn = cache_get_extra(STATE_SYNC(internal), obj);
+	struct cache_notrack *cn =
+		cache_get_extra(STATE(mode)->internal->data, obj);
 	if (queue_add(STATE_SYNC(tx_queue), &cn->qnode))
 		cache_object_get(obj);
 }