caching code.

git-svn-id: file:///svn/unbound/trunk@190 be551aaa-1e26-0410-a405-d3ace91eadb9

daemon/daemon.c

@@ -47,6 +47,7 @@
 #include "daemon/worker.h"
 #include "util/log.h"
 #include "util/config_file.h"
+#include "util/data/msgreply.h"
 #include "services/listen_dnsport.h"
 #include <signal.h>
 
@@ -116,6 +117,13 @@ daemon_init()
 	signal_handling_record();
 	checklock_start();
 	daemon->need_to_exit = 0;
+	daemon->msg_cache = lruhash_create(HASH_DEFAULT_STARTARRAY,
+		HASH_DEFAULT_MAXMEM, msgreply_sizefunc, query_info_compare,
+		query_info_delete, reply_info_delete, NULL);
+	if(!daemon->msg_cache) {
+		free(daemon);
+		return NULL;
+	}
 	alloc_init(&daemon->superalloc, NULL);
 	return daemon;	
 }
@@ -303,6 +311,7 @@ daemon_delete(struct daemon* daemon)
 	if(!daemon)
 		return;
 	listening_ports_free(daemon->ports);
+	lruhash_delete(daemon->msg_cache);
 	alloc_clear(&daemon->superalloc);
 	free(daemon->cwd);
 	free(daemon->pidfile);

daemon/daemon.h

@@ -47,6 +47,7 @@
 struct config_file;
 struct worker;
 struct listen_port;
+struct lruhash;
 
 /**
  * Structure holding worker list.
@@ -71,6 +72,8 @@ struct daemon {
 	int need_to_exit;
 	/** master allocation cache */
 	struct alloc_cache superalloc;
+	/** the message cache, content is struct msgreply_entry* */
+	struct lruhash* msg_cache;
 };
 
 /**

daemon/worker.c

@@ -57,9 +57,7 @@
 #include <signal.h>
 
 /** timeout in seconds for UDP queries to auth servers. TODO: proper rtt */
-#define UDP_QUERY_TIMEOUT 5
+#define UDP_QUERY_TIMEOUT 4
-/** the size of ID and flags, opcode, rcode in dns packet */
-#define ID_AND_FLAGS 4
 
 void 
 worker_send_cmd(struct worker* worker, ldns_buffer* buffer,
@@ -79,14 +77,32 @@ worker_send_cmd(struct worker* worker, ldns_buffer* buffer,
 static void 
 replyerror(int r, struct worker* worker)
 {
-	LDNS_QR_SET(ldns_buffer_begin(worker->query_reply.c->buffer));
+	ldns_buffer* buf = worker->query_reply.c->buffer;
-	LDNS_RCODE_SET(ldns_buffer_begin(worker->query_reply.c->buffer), r);
+	uint16_t flags;
+	verbose(VERB_DETAIL, "reply with error");
+	
+	ldns_buffer_clear(buf);
+	ldns_buffer_write_u16(buf, worker->query_id);
+	flags = (uint16_t)(0x8000 | r); /* QR and retcode*/
+	flags |= (worker->query_flags & 0x0100); /* copy RD bit */
+	ldns_buffer_write_u16(buf, flags);
+	flags = 1;
+	ldns_buffer_write_u16(buf, flags);
+	flags = 0;
+	ldns_buffer_write(buf, &flags, sizeof(uint16_t));
+	ldns_buffer_write(buf, &flags, sizeof(uint16_t));
+	ldns_buffer_write(buf, &flags, sizeof(uint16_t));
+	ldns_buffer_write(buf, worker->qinfo.qname, worker->qinfo.qnamesize);
+	ldns_buffer_write_u16(buf, worker->qinfo.qtype);
+	ldns_buffer_write_u16(buf, worker->qinfo.qclass);
+	ldns_buffer_flip(buf);
 	comm_point_send_reply(&worker->query_reply);
 	if(worker->num_requests == 1)  {
 		/* no longer at max, start accepting again. */
 		listen_resume(worker->front);
 	}
 	worker->num_requests --;
+	query_info_clear(&worker->qinfo);
 }
 
 /** process incoming replies from the network */
@@ -95,6 +111,8 @@ worker_handle_reply(struct comm_point* c, void* arg, int error,
 	struct comm_reply* ATTR_UNUSED(reply_info))
 {
 	struct worker* worker = (struct worker*)arg;
+	struct reply_info* rep;
+	struct msgreply_entry* e;
 	verbose(VERB_DETAIL, "reply to query with stored ID %d", 
 		worker->query_id);
 	LDNS_ID_SET(ldns_buffer_begin(worker->query_reply.c->buffer),
@@ -104,19 +122,40 @@ worker_handle_reply(struct comm_point* c, void* arg, int error,
 		return 0;
 	}
 	/* woohoo a reply! */
-	ldns_buffer_clear(worker->query_reply.c->buffer);
+	rep = (struct reply_info*)malloc(sizeof(struct reply_info));
-	ldns_buffer_skip(worker->query_reply.c->buffer, ID_AND_FLAGS);
+	if(!rep) {
-	ldns_buffer_write(worker->query_reply.c->buffer, 
+		log_err("out of memory");
-		ldns_buffer_at(c->buffer, ID_AND_FLAGS), 
+		replyerror(LDNS_RCODE_SERVFAIL, worker);
-		ldns_buffer_limit(c->buffer) - ID_AND_FLAGS);
+		return 0;
-	LDNS_QR_SET(ldns_buffer_begin(worker->query_reply.c->buffer));
+	}
-	ldns_buffer_flip(worker->query_reply.c->buffer);
+	rep->replysize = ldns_buffer_limit(c->buffer) - 2; /* minus ID */
+	log_info("got reply of size %d", rep->replysize);
+	rep->reply = (uint8_t*)malloc(rep->replysize);
+	if(!rep->reply) {
+		free(rep);
+		log_err("out of memory");
+		replyerror(LDNS_RCODE_SERVFAIL, worker);
+		return 0;
+	}
+	memmove(rep->reply, ldns_buffer_at(c->buffer, 2), rep->replysize);
+	ldns_buffer_write_u16_at(worker->query_reply.c->buffer, 0, 
+		worker->query_id);
+	reply_info_answer(rep, worker->query_flags, worker->query_reply.c->
+		buffer);
 	comm_point_send_reply(&worker->query_reply);
 	if(worker->num_requests == 1)  {
 		/* no longer at max, start accepting again. */
 		listen_resume(worker->front);
 	}
 	worker->num_requests --;
+	/* store or update reply in the cache */
+	if(!(e = query_info_entrysetup(&worker->qinfo, rep, 
+		worker->query_hash))) {
+		log_err("out of memory");
+		return 0;
+	}
+	lruhash_insert(worker->daemon->msg_cache, worker->query_hash, 
+		&e->entry, rep);
 	return 0;
 }
 
@@ -127,6 +166,8 @@ worker_process_query(struct worker* worker)
 	/* query the forwarding address */
 	worker->query_id = LDNS_ID_WIRE(ldns_buffer_begin(
 		worker->query_reply.c->buffer));
+	worker->query_flags = ldns_buffer_read_u16_at(worker->
+		query_reply.c->buffer, 2);
 	verbose(VERB_DETAIL, "process_query ID %d", worker->query_id);
 	pending_udp_query(worker->back, worker->query_reply.c->buffer, 
 		&worker->fwd_addr, worker->fwd_addrlen, UDP_QUERY_TIMEOUT,
@@ -212,6 +253,8 @@ worker_handle_request(struct comm_point* c, void* arg, int error,
 {
 	struct worker* worker = (struct worker*)arg;
 	int ret;
+	hashvalue_t h;
+	struct lruhash_entry* e;
 	verbose(VERB_DETAIL, "worker handle request");
 	if(error != NETEVENT_NOERROR) {
 		log_err("called with err=%d", error);
@@ -234,7 +277,27 @@ worker_handle_request(struct comm_point* c, void* arg, int error,
 		comm_point_drop_reply(repinfo);
 		return 0;
 	}
+	/* see if query is in the cache */
+	if(!query_info_parse(&worker->qinfo, c->buffer)) {
+		LDNS_QR_SET(ldns_buffer_begin(c->buffer));
+		LDNS_RCODE_SET(ldns_buffer_begin(c->buffer), 
+			LDNS_RCODE_FORMERR);
+		return 1;
+	}
+	h = query_info_hash(&worker->qinfo);
+	if((e=lruhash_lookup(worker->daemon->msg_cache, h, &worker->qinfo, 
+		0))) {
+		/* answer from cache */
+		query_info_clear(&worker->qinfo);
+		/* id is still in the buffer, no need to touch it */
+		reply_info_answer((struct reply_info*)e->data, 
+			ldns_buffer_read_u16_at(c->buffer, 2), c->buffer);
+		return 1;
+	}
+	ldns_buffer_rewind(c->buffer);
+
 	/* answer it */
+	worker->query_hash = h;
 	worker->num_requests ++;
 	if(worker->num_requests >= 1)  {
 		/* the max request number has been reached, stop accepting */

daemon/worker.h

@@ -47,6 +47,7 @@
 #include "util/netevent.h"
 #include "util/locks.h"
 #include "util/alloc.h"
+#include "util/data/msgreply.h"
 struct listen_dnsport;
 struct outside_network;
 struct config_file;
@@ -96,6 +97,12 @@ struct worker {
 	struct comm_reply query_reply;
 	/** id of query */
 	uint16_t query_id;
+	/** flags uint16 from query */
+	uint16_t query_flags;
+	/** the query_info structure from the query */
+	struct query_info qinfo;
+	/** hash value of the query qinfo */
+	hashvalue_t query_hash;
 
 	/** address to forward to */
 	struct sockaddr_storage fwd_addr;

doc/Changelog

@@ -4,6 +4,8 @@
 	  in files it has not yet read in.
 	- threaded hash table test. 
 	- unit test runs lock-verify afterwards and checks result.
+	- need writelock to update data on hash_insert.
+	- message cache code, msgreply code.
 
 21 March 2007: Wouter
 	- unit test of hash table, fixup locking problem in table_grow().

doc/TODO

@@ -3,3 +3,6 @@ o use real entropy to make random (ID, port) numbers more random.
 o in production mode, do not free memory on exit. In debug mode, test leaks.
 o profile memory allocation, and if performance issues, use special memory
   allocator. For example, with caches per thread.
+o do not malloc query_info qname of query for answers from cache. Use ptr
+  to incoming commpt buffer.
+o sendmsg, writev.

testcode/lock_verify.c

@@ -132,7 +132,7 @@ read_header(FILE* in)
 	if(fread(&t, sizeof(t), 1, in) != 1 ||	
 		fread(&thrno, sizeof(thrno), 1, in) != 1 ||
 		fread(&p, sizeof(p), 1, in) != 1) {
-		fatal_exit("fread: %s", strerror(errno));
+		fatal_exit("fread failed");
 	}
 	/* check these values are sorta OK */
 	if(!have_values) {
@@ -191,7 +191,7 @@ static void read_create(rbtree_t* all, FILE* in)
 	   fread(&o->id.instance, sizeof(int), 1, in) != 1 ||	
 	   !readup_str(&o->create_file, in) ||
 	   fread(&o->create_line, sizeof(int), 1, in) != 1)
-		fatal_exit("fread: %s", strerror(errno));
+		fatal_exit("fread failed");
 	o->smaller = rbtree_create(order_lock_cmp);
 	o->node.key = &o->id;
 	if(!rbtree_insert(all, &o->node)) {
@@ -238,7 +238,7 @@ static void read_lock(rbtree_t* all, FILE* in, int val)
 	   fread(&now_id.instance, sizeof(int), 1, in) != 1 ||	
 	   !readup_str(&ref->file, in) ||
 	   fread(&ref->line, sizeof(int), 1, in) != 1)
-		fatal_exit("fread: %s", strerror(errno));
+		fatal_exit("fread failed");
 	if(verb) printf("read lock %u %u %u %u %s %d\n", 
 		(unsigned)prev_id.thr, (unsigned)prev_id.instance,
 		(unsigned)now_id.thr, (unsigned)now_id.instance,

util/data/msgreply.c

@@ -41,6 +41,7 @@
 
 #include "config.h"
 #include "util/data/msgreply.h"
+#include "util/storage/lookup3.h"
 #include "util/log.h"
 
 /** determine length of a dname in buffer, no compression pointers allowed. */
@@ -78,8 +79,15 @@ int query_info_parse(struct query_info* m, ldns_buffer* query)
 	log_assert(ldns_buffer_position(query) == 0);
 	m->has_cd = (int)LDNS_CD_WIRE(q);
 	ldns_buffer_skip(query, LDNS_HEADER_SIZE);
-	m->qname = ldns_buffer_current(query);
+	q = ldns_buffer_current(query);
-	m->qnamesize = query_dname_len(query);
+	if((m->qnamesize = query_dname_len(query)) == 0)
+		return 0; /* parse error */
+	if(!(m->qname = (uint8_t*)malloc(m->qnamesize))) {
+		log_err("query_info_parse: out of memory");
+		return 0; /* out of memory */
+	}
+	memmove(m->qname, q, m->qnamesize);
+
 	if(ldns_buffer_remaining(query) < 4)
 		return 0; /* need qtype, qclass */
 	m->qtype = ldns_buffer_read_u16(query);
@@ -111,12 +119,75 @@ int query_info_compare(void* m1, void* m2)
 
 void query_info_clear(struct query_info* m)
 {
+	free(m->qname);
 	m->qname = NULL;
 }
 
-void msgreply_clear(struct msgreply* m)
+void reply_info_clear(struct reply_info* m)
 {
-	if(m->qname_malloced)
-		free(m->q.qname);
 	free(m->reply);
+	m->reply = NULL;
+}
+
+size_t msgreply_sizefunc(void* k, void* d)
+{
+	struct query_info* q = (struct query_info*)k;
+	struct reply_info* r = (struct reply_info*)d;
+	return sizeof(struct msgreply_entry) + sizeof(struct reply_info)
+		+ r->replysize + q->qnamesize;
+}
+
+void query_info_delete(void *k, void* ATTR_UNUSED(arg))
+{
+	struct query_info* q = (struct query_info*)k;
+	query_info_clear(q);
+	free(q);
+}
+
+void reply_info_delete(void* d, void* ATTR_UNUSED(arg))
+{
+	struct reply_info* r = (struct reply_info*)d;
+	reply_info_clear(r);
+	free(r);
+}
+
+hashvalue_t query_info_hash(struct query_info *q)
+{
+	hashvalue_t h = 0xab;
+	h = hashlittle(&q->qtype, sizeof(q->qtype), h);
+	h = hashlittle(&q->qclass, sizeof(q->qclass), h);
+	h = hashlittle(&q->has_cd, sizeof(q->has_cd), h);
+	h = hashlittle(q->qname, q->qnamesize, h);
+	return h;
+}
+
+void reply_info_answer(struct reply_info* rep, uint16_t qflags, 
+	ldns_buffer* buffer)
+{
+	uint16_t flags;
+	ldns_buffer_clear(buffer);
+	ldns_buffer_skip(buffer, 2); /* ID */
+	flags = ldns_read_uint16(rep->reply+2);
+	flags |= (qflags & 0x0100); /* copy RD bit */
+	log_info("flags %x", flags);
+	ldns_buffer_write_u16(buffer, flags);
+	ldns_buffer_write(buffer, rep->reply+2, rep->replysize-2);
+	ldns_buffer_flip(buffer);
+}
+
+struct msgreply_entry* query_info_entrysetup(struct query_info* q,
+	struct reply_info* r, hashvalue_t h)
+{
+	struct msgreply_entry* e = (struct msgreply_entry*)malloc( 
+		sizeof(struct msgreply_entry));
+	if(!e) return NULL;
+	memcpy(&e->key, q, sizeof(*q));
+	e->entry.hash = h;
+	e->entry.key = e;
+	e->entry.data = r;
+	lock_rw_init(&e->entry.lock);
+	lock_protect(&e->entry.lock, e, sizeof(*e));
+	lock_protect(&e->entry.lock, e->key.qname, e->key.qnamesize);
+	q->qname = NULL;
+	return e;
 }

util/data/msgreply.h

@@ -41,6 +41,7 @@
 
 #ifndef UTIL_DATA_MSGREPLY_H
 #define UTIL_DATA_MSGREPLY_H
+#include "util/storage/lruhash.h"
 
 /**
  * Structure to store query information that makes answers to queries
@@ -64,17 +65,23 @@ struct query_info {
  * To use it, copy over the flags from reply and modify using flags from
  * the query (RD,CD if not AA). prepend ID. 
  */
-struct msgreply {
+struct reply_info {
-	/** id of this entry. */
-	struct query_info q;
-	/** if q.qname is allocated by malloc or not */
-	int qname_malloced;
 	/** the reply packet, skips ID, starts with flags/opcode/rcode word */
 	uint8_t* reply;
 	/** the reply size */
 	size_t replysize;
 };
 
+/**
+ * Structure to keep hash table entry for message replies.
+ */
+struct msgreply_entry {
+	/** the hash table key */
+	struct query_info key;
+	/** the hash table entry, data is struct reply_info* */
+	struct lruhash_entry entry;
+};
+
 /** 
  * Parse wire query into a queryinfo structure, return 0 on parse error. 
  * initialises the (prealloced) queryinfo structure as well. sets reply to 0.
@@ -103,7 +110,40 @@ void query_info_clear(struct query_info* m);
   * allowed, returns 0 on failure. */
 size_t query_dname_len(ldns_buffer* query);
 
-/** clear out msgreply structure */
+/** clear out reply info structure */
-void msgreply_clear(struct msgreply* m);
+void reply_info_clear(struct reply_info* m);
+
+/** calculate size of struct query_info + reply_info */
+size_t msgreply_sizefunc(void* k, void* d);
+
+/** delete query_info key structure */
+void query_info_delete(void *q, void* arg);
+
+/** delete reply_info data structure */
+void reply_info_delete(void* d, void* arg);
+
+/** calculate hash value of query_info */
+hashvalue_t query_info_hash(struct query_info *q);
+
+/** 
+ * Generate answer from reply_info.
+ * @param rep: reply to fill in.
+ * @param qflags: flags word from the query.
+ * @param buf: buffer to put reply into. Note that the query ID must 
+ *	be put in the buffer by caller.
+ *	The buffer must be large enough.
+ */
+void reply_info_answer(struct reply_info* rep, uint16_t qflags, 
+	ldns_buffer* buf);
+
+/**
+ * Setup query info entry
+ * @param q: query info to copy. Emptied as if clear is called.
+ * @param r: reply to init data.
+ * @param h: hash value.
+ * @return: newly allocated message reply cache item.
+ */
+struct msgreply_entry* query_info_entrysetup(struct query_info* q,
+	struct reply_info* r, hashvalue_t h);
 
 #endif /* UTIL_DATA_MSGREPLY_H */

util/storage/lruhash.c

@@ -93,6 +93,7 @@ void
 bin_delete(struct lruhash* table, struct lruhash_bin* bin)
 {
 	struct lruhash_entry* p, *np;
+	void *d;
 	if(!bin)
 		return;
 	lock_quick_destroy(&bin->lock);
@@ -100,8 +101,9 @@ bin_delete(struct lruhash* table, struct lruhash_bin* bin)
 	bin->overflow_list = NULL;
 	while(p) {
 		np = p->overflow_next;
+		d = p->data;
 		(*table->delkeyfunc)(p->key, table->cb_arg);
-		(*table->deldatafunc)(p->data, table->cb_arg);
+		(*table->deldatafunc)(d, table->cb_arg);
 		p = np;
 	}
 }
@@ -307,13 +309,15 @@ lruhash_insert(struct lruhash* table, hashvalue_t hash,
 		table->num++;
 		table->space_used += need_size;
 	} else {
-		/* if so: update data */
+		/* if so: update data - needs a writelock */
 		table->space_used += need_size -
 			(*table->sizefunc)(found->key, found->data);
 		(*table->delkeyfunc)(entry->key, table->cb_arg);
+		lru_touch(table, found);
+		lock_rw_wrlock(&found->lock);
 		(*table->deldatafunc)(found->data, table->cb_arg);
 		found->data = data;
-		lru_touch(table, found);
+		lock_rw_unlock(&found->lock);
 	}
 	lock_quick_unlock(&bin->lock);
 	if(table->space_used > table->space_max)