Fix reorder possible reordering of destroy messages under message omission. This patch introduces the TimeoutDestroy clause to determine how long a conntrack remains in the internal cache once it has been destroy from the kernel table.

10 files changed, 105 insertions, 49 deletions

diff --git a/include/cache.h b/include/cache.h
index f5afbe5..442a563 100644
--- a/include/cache.h
+++ b/include/cache.h
@@ -82,6 +82,7 @@ struct us_conntrack *cache_add(struct cache *c, struct nf_conntrack *ct);
 struct us_conntrack *cache_update(struct cache *c, struct nf_conntrack *ct);
 struct us_conntrack *cache_update_force(struct cache *c, struct nf_conntrack *ct);
 int cache_del(struct cache *c, struct nf_conntrack *ct);
+struct us_conntrack *cache_del_timeout(struct cache *c, struct nf_conntrack *ct, int timeout);
 int cache_test(struct cache *c, struct nf_conntrack *ct);
 void cache_stats(const struct cache *c, int fd);
 struct us_conntrack *cache_get_conntrack(struct cache *, void *);
diff --git a/include/conntrackd.h b/include/conntrackd.h
index 57ac7e4..b266289 100644
--- a/include/conntrackd.h
+++ b/include/conntrackd.h
@@ -76,6 +76,7 @@ struct ct_conf {
 	int refresh;
 	int cache_timeout;		/* cache entries timeout */
 	int commit_timeout;		/* committed entries timeout */
+	int del_timeout;
 	unsigned int netlink_buffer_size;
 	unsigned int netlink_buffer_size_max_grown;
 	unsigned char ignore_protocol[IPPROTO_MAX];
diff --git a/include/us-conntrack.h b/include/us-conntrack.h
index 3d71e22..9eafa3b 100644
--- a/include/us-conntrack.h
+++ b/include/us-conntrack.h
@@ -1,12 +1,13 @@
 #ifndef _US_CONNTRACK_H_
 #define _US_CONNTRACK_H_
 
+#include "alarm.h"
 #include <libnetfilter_conntrack/libnetfilter_conntrack.h>
 
-/* be careful, do not modify the layout */
 struct us_conntrack {
 	struct 	nf_conntrack *ct;
-	struct  cache *cache;          /* add new attributes here */
+	struct  cache *cache; 
+	struct	alarm_block alarm;
 	char 	data[0];
 };
 
diff --git a/src/alarm.c b/src/alarm.c
index 91ee2ca..fe938a0 100644
--- a/src/alarm.c
+++ b/src/alarm.c
@@ -123,7 +123,7 @@ do_alarm_run(struct timeval *next_run)
 {
 	struct list_head alarm_run_queue;
 	struct rb_node *node;
-	struct alarm_block *this;
+	struct alarm_block *this, *tmp;
 	struct timeval tv;
 
 	gettimeofday(&tv, NULL);
@@ -138,7 +138,8 @@ do_alarm_run(struct timeval *next_run)
 		list_add(&this->list, &alarm_run_queue);
 	}
 
-	list_for_each_entry(this, &alarm_run_queue, list) {
+	/* must be safe as entries can vanish from the callback */
+	list_for_each_entry_safe(this, tmp, &alarm_run_queue, list) {
 		rb_erase(&this->node, &alarm_root);
 		RB_CLEAR_NODE(&this->node);
 		this->function(this, this->data);
diff --git a/src/cache.c b/src/cache.c
index 73d539a..eac9a78 100644
--- a/src/cache.c
+++ b/src/cache.c
@@ -237,6 +237,8 @@ void cache_destroy(struct cache *c)
 	free(c);
 }
 
+static void __del_timeout(struct alarm_block *a, void *data);
+
 static struct us_conntrack *__add(struct cache *c, struct nf_conntrack *ct)
 {
 	unsigned i;
@@ -258,6 +260,8 @@ static struct us_conntrack *__add(struct cache *c, struct nf_conntrack *ct)
 	if (u) {
 		char *data = u->data;
 
+		init_alarm(&u->alarm, u, __del_timeout);
+
         	for (i = 0; i < c->num_features; i++) {
 			c->features[i]->add(u, data);
 			data += c->features[i]->size;
@@ -324,8 +328,7 @@ static struct us_conntrack *__update(struct cache *c, struct nf_conntrack *ct)
 	return NULL;
 }
 
-static struct us_conntrack *
-__cache_update(struct cache *c, struct nf_conntrack *ct)
+struct us_conntrack *cache_update(struct cache *c, struct nf_conntrack *ct)
 {
 	struct us_conntrack *u;
 
@@ -339,15 +342,6 @@ __cache_update(struct cache *c, struct nf_conntrack *ct)
 	return NULL;
 }
 
-struct us_conntrack *cache_update(struct cache *c, struct nf_conntrack *ct)
-{
-	struct us_conntrack *u;
-
-	u = __cache_update(c, ct);
-
-	return u;
-}
-
 struct us_conntrack *cache_update_force(struct cache *c,
 					struct nf_conntrack *ct)
 {
@@ -379,6 +373,24 @@ int cache_test(struct cache *c, struct nf_conntrack *ct)
 	return ret != NULL;
 }
 
+static void __del2(struct cache *c, struct us_conntrack *u)
+{
+	unsigned i;
+	char *data = u->data;
+	struct nf_conntrack *p = u->ct;
+
+	for (i = 0; i < c->num_features; i++) {
+		c->features[i]->destroy(u, data);
+		data += c->features[i]->size;
+	}
+
+	if (c->extra && c->extra->destroy)
+		c->extra->destroy(u, ((char *) u) + c->extra_offset);
+
+	hashtable_del(c->h, u);
+	free(p);
+}
+
 static int __del(struct cache *c, struct nf_conntrack *ct)
 {
 	size_t size = c->h->datasize;
@@ -389,20 +401,8 @@ static int __del(struct cache *c, struct nf_conntrack *ct)
 
 	u = (struct us_conntrack *) hashtable_test(c->h, u);
 	if (u) {
-		unsigned i;
-		char *data = u->data;
-		struct nf_conntrack *p = u->ct;
-
-		for (i = 0; i < c->num_features; i++) {
-			c->features[i]->destroy(u, data);
-			data += c->features[i]->size;
-		}
-
-		if (c->extra && c->extra->destroy)
-			c->extra->destroy(u, ((char *) u) + c->extra_offset);
-
-		hashtable_del(c->h, u);
-		free(p);
+		del_alarm(&u->alarm);
+		__del2(c, u);
 		return 1;
 	}
 	return 0;
@@ -419,6 +419,37 @@ int cache_del(struct cache *c, struct nf_conntrack *ct)
 	return 0;
 }
 
+static void __del_timeout(struct alarm_block *a, void *data)
+{
+	struct us_conntrack *u = (struct us_conntrack *) data;
+	struct cache *c = u->cache;
+
+	__del2(u->cache, u);
+	c->del_ok++;
+}
+
+struct us_conntrack *
+cache_del_timeout(struct cache *c, struct nf_conntrack *ct, int timeout)
+{
+	size_t size = c->h->datasize;
+	char buf[size];
+	struct us_conntrack *u = (struct us_conntrack *) buf;
+
+	if (timeout <= 0)
+		cache_del(c, ct);
+
+	u->ct = ct;
+
+	u = (struct us_conntrack *) hashtable_test(c->h, u);
+	if (u) {
+		if (!alarm_pending(&u->alarm)) {
+			add_alarm(&u->alarm, timeout, 0);
+			return u;
+		}
+	}
+	return NULL;
+}
+
 struct us_conntrack *cache_get_conntrack(struct cache *c, void *data)
 {
 	return (struct us_conntrack *)((char*)data - c->extra_offset);
diff --git a/src/cache_lifetime.c b/src/cache_lifetime.c
index ad3416a..cf84d20 100644
--- a/src/cache_lifetime.c
+++ b/src/cache_lifetime.c
@@ -53,7 +53,13 @@ static int lifetime_dump(struct us_conntrack *u,
 
 	gettimeofday(&tv, NULL);
 
-	return sprintf(buf, " [active since %lds]", tv.tv_sec - *lifetime);
+	if (alarm_pending(&u->alarm))
+		return sprintf(buf, " [active since %lds] [expires in %lds]",
+			       tv.tv_sec - *lifetime, 
+			       u->alarm.tv.tv_sec - tv.tv_sec);
+	else
+		return sprintf(buf, " [active since %lds]", 
+			       tv.tv_sec - *lifetime);
 }
 
 struct cache_feature lifetime_feature = {
diff --git a/src/read_config_lex.l b/src/read_config_lex.l
index fe2090b..1350afc 100644
--- a/src/read_config_lex.l
+++ b/src/read_config_lex.l
@@ -103,6 +103,7 @@ ftfw		[F|f][T|t][F|f][W|w]
 "CLOSE"				{ return T_CLOSE; }
 "LISTEN"			{ return T_LISTEN; }
 "LogFileBufferSize"		{ return T_STAT_BUFFER_SIZE; }
+"DestroyTimeout"		{ return T_DESTROY_TIMEOUT; }
 
 {is_on}			{ return T_ON; }
 {is_off}		{ return T_OFF; }
diff --git a/src/read_config_yy.y b/src/read_config_yy.y
index 86fee9b..0bc5e3c 100644
--- a/src/read_config_yy.y
+++ b/src/read_config_yy.y
@@ -52,7 +52,7 @@ struct ct_conf conf;
 %token T_REPLICATE T_FOR T_IFACE 
 %token T_ESTABLISHED T_SYN_SENT T_SYN_RECV T_FIN_WAIT 
 %token T_CLOSE_WAIT T_LAST_ACK T_TIME_WAIT T_CLOSE T_LISTEN
-%token T_SYSLOG T_WRITE_THROUGH T_STAT_BUFFER_SIZE
+%token T_SYSLOG T_WRITE_THROUGH T_STAT_BUFFER_SIZE T_DESTROY_TIMEOUT
 
 
 %token <string> T_IP T_PATH_VAL
@@ -429,6 +429,7 @@ sync_line: refreshtime
 	 | listen_to
 	 | state_replication
 	 | cache_writethrough
+	 | destroy_timeout
 	 ;
 
 sync_mode_alarm: T_SYNC_MODE T_ALARM '{' sync_mode_alarm_list '}'
@@ -469,6 +470,11 @@ window_size: T_WINDOWSIZE T_NUMBER
 	conf.window_size = $2;
 };
 
+destroy_timeout: T_DESTROY_TIMEOUT T_NUMBER
+{
+	conf.del_timeout = $2;
+};
+
 relax_transitions: T_RELAX_TRANSITIONS
 {
 	fprintf(stderr, "Notice: RelaxTransitions clause is obsolete. "
@@ -746,5 +752,9 @@ init_config(char *filename)
 	if (CONFIG(window_size) == 0)
 		CONFIG(window_size) = 20;
 
+	/* double of 120 seconds which is common timeout of a final state */
+	if (conf.flags & CTD_SYNC_FTFW && CONFIG(del_timeout) == 0)
+		CONFIG(del_timeout) = 240;
+
 	return 0;
 }
diff --git a/src/sync-ftfw.c b/src/sync-ftfw.c
index 0b98513..77f8fd4 100644
--- a/src/sync-ftfw.c
+++ b/src/sync-ftfw.c
@@ -390,6 +390,7 @@ static void ftfw_send(struct nethdr *net, struct us_conntrack *u)
 	switch(ntohs(pld->query)) {
 	case NFCT_Q_CREATE:
 	case NFCT_Q_UPDATE:
+	case NFCT_Q_DESTROY:
 		cn = (struct cache_ftfw *) 
 			cache_get_extra(STATE_SYNC(internal), u);
 
@@ -402,9 +403,6 @@ static void ftfw_send(struct nethdr *net, struct us_conntrack *u)
 		list_add_tail(&cn->rs_list, &rs_list);
 		rs_list_len++;
 		break;
-	case NFCT_Q_DESTROY:
-		queue_add(rs_queue, net, net->len);
-		break;
 	}
 }
 
@@ -429,10 +427,10 @@ static int tx_queue_xmit(void *data1, const void *data2)
 	return 0;
 }
 
-static int tx_list_xmit(struct list_head *i, struct us_conntrack *u)
+static int tx_list_xmit(struct list_head *i, struct us_conntrack *u, int type)
 {
 	int ret;
-	struct nethdr *net = BUILD_NETMSG(u->ct, NFCT_Q_UPDATE);
+	struct nethdr *net = BUILD_NETMSG(u->ct, type);
 	size_t len = prepare_send_netmsg(STATE_SYNC(mcast_client), net);
 
 	dp("tx_list sq: %u fl:%u len:%u\n",
@@ -460,7 +458,10 @@ static void ftfw_run(void)
 		struct us_conntrack *u;
 
 		u = cache_get_conntrack(STATE_SYNC(internal), cn);
-		tx_list_xmit(&cn->tx_list, u);
+		if (alarm_pending(&u->alarm))
+			tx_list_xmit(&cn->tx_list, u, NFCT_Q_DESTROY);
+		else
+			tx_list_xmit(&cn->tx_list, u, NFCT_Q_UPDATE);
 	}
 
 	/* reset alive alarm */
diff --git a/src/sync-mode.c b/src/sync-mode.c
index cbb4769..a952a5b 100644
--- a/src/sync-mode.c
+++ b/src/sync-mode.c
@@ -344,17 +344,15 @@ static void dump_sync(struct nf_conntrack *ct)
 		debug_ct(ct, "resync");
 }
 
-static void mcast_send_sync(struct us_conntrack *u,
-			    struct nf_conntrack *ct,
-			    int query)
+static void mcast_send_sync(struct us_conntrack *u, int query)
 {
 	size_t len;
 	struct nethdr *net;
 
-	if (!state_helper_verdict(query, ct))
+	if (!state_helper_verdict(query, u->ct))
 		return;
 
-	net = BUILD_NETMSG(ct, query);
+	net = BUILD_NETMSG(u->ct, query);
 	len = prepare_send_netmsg(STATE_SYNC(mcast_client), net);
 	mcast_buffered_send_netmsg(STATE_SYNC(mcast_client), net, len);
 	if (STATE_SYNC(sync)->send)
@@ -370,8 +368,10 @@ static int purge_step(void *data1, void *data2)
 	ret = nfct_query(h, NFCT_Q_GET, u->ct);
 	if (ret == -1 && errno == ENOENT) {
 		debug_ct(u->ct, "overrun purge resync");
-		mcast_send_sync(u, u->ct, NFCT_Q_DESTROY);
-		cache_del(STATE_SYNC(internal), u->ct);
+		if (cache_del_timeout(STATE_SYNC(internal),
+				      u->ct, 
+				      CONFIG(del_timeout)))
+			mcast_send_sync(u, NFCT_Q_DESTROY);
 	}
 
 	return 0;
@@ -406,7 +406,7 @@ static int overrun_sync(enum nf_conntrack_msg_type type,
 	if (!cache_test(STATE_SYNC(internal), ct)) {
 		if ((u = cache_update_force(STATE_SYNC(internal), ct))) {
 			debug_ct(u->ct, "overrun resync");
-			mcast_send_sync(u, u->ct, NFCT_Q_UPDATE);
+			mcast_send_sync(u, NFCT_Q_UPDATE);
 		}
 	}
 
@@ -427,7 +427,7 @@ static void event_new_sync(struct nf_conntrack *ct)
 	nfct_attr_unset(ct, ATTR_REPL_COUNTER_PACKETS);
 retry:
 	if ((u = cache_add(STATE_SYNC(internal), ct))) {
-		mcast_send_sync(u, ct, NFCT_Q_CREATE);
+		mcast_send_sync(u, NFCT_Q_CREATE);
 		debug_ct(u->ct, "internal new");
 	} else {
 		if (errno == EEXIST) {
@@ -453,16 +453,19 @@ static void event_update_sync(struct nf_conntrack *ct)
 		return;
 	}
 	debug_ct(u->ct, "internal update");
-	mcast_send_sync(u, ct, NFCT_Q_UPDATE);
+	mcast_send_sync(u, NFCT_Q_UPDATE);
 }
 
 static int event_destroy_sync(struct nf_conntrack *ct)
 {
+	struct us_conntrack *u;
+
 	if (!CONFIG(cache_write_through))
 		nfct_attr_unset(ct, ATTR_TIMEOUT);
 
-	if (cache_del(STATE_SYNC(internal), ct)) {
-		mcast_send_sync(NULL, ct, NFCT_Q_DESTROY);
+	u = cache_del_timeout(STATE_SYNC(internal), ct, CONFIG(del_timeout));
+	if (u != NULL) {
+		mcast_send_sync(u, NFCT_Q_DESTROY);
 		debug_ct(ct, "internal destroy");
 		return 1;
 	} else {