Stream reuse

services/outside_network.h

@@ -52,6 +52,7 @@ struct ub_randstate;
 struct pending_tcp;
 struct waiting_tcp;
 struct waiting_udp;
+struct reuse_tcp;
 struct infra_cache;
 struct port_comm;
 struct port_if;
@@ -154,6 +155,21 @@ struct outside_network {
 	size_t num_tcp;
 	/** number of tcp communication points in use. */
 	size_t num_tcp_outgoing;
+	/**
+	 * tree of still-open and waiting tcp connections for reuse.
+	 * can be closed and reopened to get a new tcp connection.
+	 * or reused to the same destination again.  with timeout to close.
+	 * Entries are of type struct reuse_tcp.
+	 * The entries are both active and empty connections.
+	 */
+	rbtree_type tcp_reuse;
+	/** max number of tcp_reuse entries we want to keep open */
+	size_t tcp_reuse_max;
+	/** first and last(oldest) in lru list of reuse connections.
+	 * the oldest can be closed to get a new free pending_tcp if needed
+	 * The list contains empty connections, that wait for timeout or
+	 * a new query that can use the existing connection. */
+	struct reuse_tcp* tcp_reuse_first, *tcp_reuse_last;
 	/** list of tcp comm points that are free for use */
 	struct pending_tcp* tcp_free;
 	/** list of tcp queries waiting for a buffer */
@@ -211,6 +227,62 @@ struct port_comm {
 	struct comm_point* cp;
 };
 
+/**
+ * Reuse TCP connection, still open can be used again.
+ */
+struct reuse_tcp {
+	/** rbtree node with links in tcp_reuse tree. key is NULL when not
+	 * in tree. Both active and empty connections are in the tree.
+	 * key is a pointer to this structure, the members used to compare
+	 * are the sockaddr and and then is-ssl bool, and then ptr value is
+	 * used in case the same address exists several times in the tree
+	 * when there are multiple connections to the same destination to
+	 * make the rbtree items unique. */
+	rbnode_type node;
+	/** the key for the tcp_reuse tree. address of peer, ip4 or ip6,
+	 * and port number of peer */
+	struct sockaddr_storage addr;
+	/** length of addr */
+	socklen_t addrlen;
+	/** lru chain, so that the oldest can be removed to get a new
+	 * connection when all are in (re)use. oldest is last in list.
+	 * The lru only contains empty connections waiting for reuse,
+	 * the ones with active queries are not on the list because they
+	 * do not need to be closed to make space for others.  They already
+	 * service a query so the close for another query does not help
+	 * service a larger number of queries. */
+	struct reuse_tcp* lru_next, *lru_prev;
+	/** true if the reuse_tcp item is on the lru list with empty items */
+	int item_on_lru_list;
+	/** the connection to reuse, the fd is non-1 and is open.
+	 * the addr and port determine where the connection is going,
+	 * and is key to the rbtree.  The SSL ptr determines if it is
+	 * a TLS connection or a plain TCP connection there.  And TLS
+	 * or not is also part of the key to the rbtree.
+	 * There is a timeout and read event on the fd, to close it. */
+	struct pending_tcp* pending;
+	/** rbtree with other queries waiting on the connection, by ID number,
+	 * of type struct waiting_tcp. It is for looking up received
+	 * answers to the structure for callback.  And also to see if ID
+	 * numbers are unused and can be used for a new query.
+	 * The write_wait elements are also in the tree, so that ID numbers
+	 * can be looked up also for them.  They are bool write_wait_queued. */
+	rbtree_type tree_by_id;
+	/** list of queries waiting to be written on the channel,
+	 * if NULL no queries are waiting to be written and the pending->query
+	 * is the query currently serviced.  The first is the next in line.
+	 * They are also in the tree_by_id. Once written, the are removed
+	 * from this list, but stay in the tree. */
+	struct waiting_tcp* write_wait_first, *write_wait_last;
+	/** the outside network it is part of */
+	struct outside_network* outnet;
+};
+
+/** max number of queries on a reuse connection */
+#define MAX_REUSE_TCP_QUERIES 200
+/** timeout for REUSE entries in milliseconds. */
+#define REUSE_TIMEOUT 60000
+
 /**
  * A query that has an answer pending for it.
  */
@@ -255,12 +327,15 @@ struct pending {
 struct pending_tcp {
 	/** next in list of free tcp comm points, or NULL. */
 	struct pending_tcp* next_free;
-	/** the ID for the query; checked in reply */
-	uint16_t id;
 	/** tcp comm point it was sent on (and reply must come back on). */
 	struct comm_point* c;
 	/** the query being serviced, NULL if the pending_tcp is unused. */
 	struct waiting_tcp* query;
+	/** the pre-allocated reuse tcp structure.  if ->pending is nonNULL
+	 * it is in use and the connection is waiting for reuse.
+	 * It is here for memory pre-allocation, and used to make this
+	 * pending_tcp wait for reuse. */
+	struct reuse_tcp reuse;
 };
 
 /**
@@ -269,12 +344,27 @@ struct pending_tcp {
 struct waiting_tcp {
 	/** 
 	 * next in waiting list.
-	 * if pkt==0, this points to the pending_tcp structure.
+	 * if on_tcp_waiting_list==0, this points to the pending_tcp structure.
 	 */
 	struct waiting_tcp* next_waiting;
+	/** if true the item is on the tcp waiting list and next_waiting
+	 * is used for that.  If false, the next_waiting points to the
+	 * pending_tcp */
+	int on_tcp_waiting_list;
+	/** next and prev in query waiting list for stream connection */
+	struct waiting_tcp* write_wait_prev, *write_wait_next;
+	/** true if the waiting_tcp structure is on the write_wait queue */
+	int write_wait_queued;
+	/** entry in reuse.tree_by_id, if key is NULL, not in tree, otherwise,
+	 * this struct is key and sorted by ID (from waiting_tcp.id). */
+	rbnode_type id_node;
+	/** the ID for the query; checked in reply */
+	uint16_t id;
 	/** timeout event; timer keeps running whether the query is
 	 * waiting for a buffer or the tcp reply is pending */
 	struct comm_timer* timer;
+	/** timeout in msec */
+	int timeout;
 	/** the outside network it is part of */
 	struct outside_network* outnet;
 	/** remote address. */
@@ -284,20 +374,23 @@ struct waiting_tcp {
 	/** 
 	 * The query itself, the query packet to send.
 	 * allocated after the waiting_tcp structure.
-	 * set to NULL when the query is serviced and it part of pending_tcp.
-	 * if this is NULL, the next_waiting points to the pending_tcp.
 	 */
 	uint8_t* pkt;
 	/** length of query packet. */
 	size_t pkt_len;
-	/** callback for the timeout, error or reply to the message */
+	/** callback for the timeout, error or reply to the message,
+	 * or NULL if no user is waiting. the entry uses an ID number.
+	 * a query that was written is no longer needed, but the ID number
+	 * and a reply will come back and can be ignored if NULL */
 	comm_point_callback_type* cb;
 	/** callback user argument */
 	void* cb_arg;
 	/** if it uses ssl upstream */
 	int ssl_upstream;
 	/** ref to the tls_auth_name from the serviced_query */
 	char* tls_auth_name;
+	/** the packet was involved in an error, to stop looping errors */
+	int error_count;
 };
 
 /**
@@ -546,6 +639,19 @@ size_t outnet_get_mem(struct outside_network* outnet);
  */
 size_t serviced_get_mem(struct serviced_query* sq);
 
+/** Pick random ID value for a tcp stream, avoids existing IDs. */
+uint16_t reuse_tcp_select_id(struct reuse_tcp* reuse,
+	struct outside_network* outnet);
+
+/** find element in tree by id */
+struct waiting_tcp* reuse_tcp_by_id_find(struct reuse_tcp* reuse, uint16_t id);
+
+/** insert element in tree by id */
+void reuse_tree_by_id_insert(struct reuse_tcp* reuse, struct waiting_tcp* w);
+
+/** delete readwait waiting_tcp elements, deletes the elements in the list */
+void reuse_del_readwait(rbtree_type* tree_by_id);
+
 /** get TCP file descriptor for address, returns -1 on failure,
  * tcp_mss is 0 or maxseg size to set for TCP packets. */
 int outnet_get_tcp_fd(struct sockaddr_storage* addr, socklen_t addrlen, int tcp_mss, int dscp);
@@ -643,4 +749,10 @@ int pending_cmp(const void* key1, const void* key2);
 /** compare function of serviced query rbtree */
 int serviced_cmp(const void* key1, const void* key2);
 
+/** compare function of reuse_tcp rbtree in outside_network struct */
+int reuse_cmp(const void* key1, const void* key2);
+
+/** compare function of reuse_tcp tree_by_id rbtree */
+int reuse_id_cmp(const void* key1, const void* key2);
+
 #endif /* OUTSIDE_NETWORK_H */

testcode/fake_event.c

@@ -1511,6 +1511,18 @@ int serviced_cmp(const void* ATTR_UNUSED(a), const void* ATTR_UNUSED(b))
 	return 0;
 }
 
+int reuse_cmp(const void* ATTR_UNUSED(a), const void* ATTR_UNUSED(b))
+{
+	log_assert(0);
+	return 0;
+}
+
+int reuse_id_cmp(const void* ATTR_UNUSED(a), const void* ATTR_UNUSED(b))
+{
+	log_assert(0);
+	return 0;
+}
+
 /* timers in testbound for autotrust. statistics tested in tdir. */
 struct comm_timer* comm_timer_create(struct comm_base* base, 
 	void (*cb)(void*), void* cb_arg)

testcode/unitmain.c

@@ -839,6 +839,52 @@ static void respip_test(void)
 	respip_conf_actions_test();
 }
 
+#include "services/outside_network.h"
+/** add number of new IDs to the reuse tree, randomly chosen */
+static void tcpid_addmore(struct reuse_tcp* reuse,
+	struct outside_network* outnet, unsigned int addnum)
+{
+	unsigned int i;
+	struct waiting_tcp* w;
+	for(i=0; i<addnum; i++) {
+		uint16_t id = reuse_tcp_select_id(reuse, outnet);
+		unit_assert(!reuse_tcp_by_id_find(reuse, id));
+		w = calloc(1, sizeof(*w));
+		unit_assert(w);
+		w->id = id;
+		w->outnet = outnet;
+		w->next_waiting = (void*)reuse->pending;
+		reuse_tree_by_id_insert(reuse, w);
+	}
+}
+
+/** fill up the reuse ID tree and test assertions */
+static void tcpid_fillup(struct reuse_tcp* reuse,
+	struct outside_network* outnet)
+{
+	int t, numtest=3;
+	for(t=0; t<numtest; t++) {
+		rbtree_init(&reuse->tree_by_id, reuse_id_cmp);
+		tcpid_addmore(reuse, outnet, 65535);
+		reuse_del_readwait(&reuse->tree_by_id);
+	}
+}
+
+/** test TCP ID selection */
+static void tcpid_test(void)
+{
+	struct pending_tcp pend;
+	struct outside_network outnet;
+	unit_show_func("services/outside_network.c", "reuse_tcp_select_id");
+	memset(&pend, 0, sizeof(pend));
+	pend.reuse.pending = &pend;
+	memset(&outnet, 0, sizeof(outnet));
+	outnet.rnd = ub_initstate(NULL);
+	rbtree_init(&pend.reuse.tree_by_id, reuse_id_cmp);
+	tcpid_fillup(&pend.reuse, &outnet);
+	ub_randfree(outnet.rnd);
+}
+
 void unit_show_func(const char* file, const char* func)
 {
 	printf("test %s:%s\n", file, func);
@@ -907,6 +953,7 @@ main(int argc, char* argv[])
 	infra_test();
 	ldns_test();
 	msgparse_test();
+	tcpid_test();
 #ifdef CLIENT_SUBNET
 	ecs_test();
 #endif /* CLIENT_SUBNET */

testdata/tcp_reuse.tdir/tcp_reuse.conf

@@ -0,0 +1,17 @@
+server:
+	verbosity: 5
+	# num-threads: 1
+	interface: 127.0.0.1
+	port: @PORT@
+	use-syslog: no
+	directory: .
+	pidfile: "unbound.pid"
+	chroot: ""
+	username: ""
+	do-not-query-localhost: no
+
+	tcp-upstream: yes
+
+forward-zone:
+	name: "."
+	forward-addr: "127.0.0.1@@TOPORT@"

testdata/tcp_reuse.tdir/tcp_reuse.conf2

@@ -0,0 +1,39 @@
+# this is the upstream server that has pipelining and responds to queries.
+server:
+	verbosity: 1
+	# num-threads: 1
+	interface: 127.0.0.1
+	port: @PORT@
+	use-syslog: no
+	directory: .
+	pidfile: "unbound2.pid"
+	chroot: ""
+	username: ""
+	do-not-query-localhost: no
+	tcp-idle-timeout: 10000
+
+	log-queries: yes
+	log-replies: yes
+	log-identity: "upstream"
+
+	local-zone: "." refuse
+	local-zone: "example.com" static
+	local-data: "www.example.com  A 10.20.30.40"
+	local-data: "www1.example.com  A 10.20.30.41"
+	local-data: "www2.example.com  A 10.20.30.42"
+	local-data: "www3.example.com  A 10.20.30.43"
+	local-data: "www4.example.com  A 10.20.30.44"
+	local-data: "www5.example.com  A 10.20.30.45"
+	local-data: "www6.example.com  A 10.20.30.46"
+	local-data: "www7.example.com  A 10.20.30.47"
+
+	local-zone: "drop.net" deny
+	local-zone: "refuse.net" refuse
+
+	local-zone: "more.net" redirect
+	local-data: "more.net A 10.20.30.40"
+
+# if queries escape, send them to localhost
+forward-zone:
+	name: "."
+	forward-addr: "127.0.0.1@@TOPORT@"

testdata/tcp_reuse.tdir/tcp_reuse.dsc

@@ -0,0 +1,16 @@
+BaseName: tcp_reuse
+Version: 1.0
+Description: Test tcp stream reuse.
+CreationDate: Wed Jun 03 09:37:00 CET 2020
+Maintainer: Wouter Wijngaards
+Category: 
+Component:
+CmdDepends: 
+Depends: 
+Help:
+Pre: tcp_reuse.pre
+Post: tcp_reuse.post
+Test: tcp_reuse.test
+AuxFiles: 
+Passed:
+Failure:

testdata/tcp_reuse.tdir/tcp_reuse.post

@@ -0,0 +1,19 @@
+# #-- tcp_reuse.post --#
+# source the master var file when it's there
+[ -f ../.tpkg.var.master ] && source ../.tpkg.var.master
+# source the test var file when it's there
+[ -f .tpkg.var.test ] && source .tpkg.var.test
+#
+# do your teardown here
+. ../common.sh
+kill_pid `cat unbound2.pid`
+if test -f unbound2.log; then
+	echo ">>> upstream log"
+	cat unbound2.log
+fi
+#kill_pid $UNBOUND_PID
+kill_pid `cat unbound.pid`
+if test -f unbound.log; then
+	echo ">>> unbound log"
+	cat unbound.log
+fi