From: Marek Vavruša <marek.vavrusa@nic.cz>
Date: Fri, 29 Aug 2014 19:15:38 +0000 (+0200)
Subject: cache: implemented transactions
X-Git-Tag: v1.0.0-beta1~419
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=98076cd8b14aa3302d166da1bf95cf256a592dda;p=thirdparty%2Fknot-resolver.git

cache: implemented transactions
---

diff --git a/lib/cache.c b/lib/cache.c
index 9b7a89ea2..6785978ae 100644
--- a/lib/cache.c
+++ b/lib/cache.c
@@ -17,6 +17,14 @@ struct kr_cache
 	mm_ctx_t *pool;
 };
 
+struct kr_txn
+{
+	MDB_dbi dbi;
+	MDB_txn *txn;
+	MDB_txn *parent;
+	mm_ctx_t *mm;
+};
+
 /*                       MDB access                                           */
 
 static int dbase_open(struct kr_cache *cache, const char *handle)
@@ -52,8 +60,7 @@ static int dbase_open(struct kr_cache *cache, const char *handle)
 		return ret;
 	}
 
-	DEBUG_MSG("opened '%s'\n", handle);
-
+	DEBUG_MSG("OPEN '%s'\n", handle);
 	return 0;
 }
 
@@ -61,45 +68,26 @@ static void dbase_close(struct kr_cache *cache)
 {
 	mdb_close(cache->env, cache->dbi);
 	mdb_env_close(cache->env);
+	DEBUG_MSG("CLOSE\n");
 }
 
 /*                       data access                                          */
 
-static MDB_cursor *cursor_acquire(struct kr_cache *cache, unsigned flags)
+static MDB_cursor *cursor_acquire(struct kr_txn *txn)
 {
-	MDB_txn *txn = NULL;
 	MDB_cursor *cursor = NULL;
-	int ret = mdb_txn_begin(cache->env, NULL, flags, &txn);
-	if (ret != 0) {
-		return NULL;
-	}
 
-	ret = mdb_cursor_open(txn, cache->dbi, &cursor);
+	int ret = mdb_cursor_open(txn->txn, txn->dbi, &cursor);
 	if (ret != 0) {
-		mdb_txn_abort(txn);
 		return NULL;
 	}
 
-	DEBUG_MSG("cursor acquire\n");
-
 	return cursor;
 }
 
-static int cursor_release(MDB_cursor *cursor, bool commit)
+static void cursor_release(MDB_cursor *cursor)
 {
-	MDB_txn *txn = mdb_cursor_txn(cursor);
 	mdb_cursor_close(cursor);
-
-	int ret = 0;
-	if (commit) {
-		DEBUG_MSG("cursor release / commit\n");
-		ret = mdb_txn_commit(txn);
-	} else {
-		DEBUG_MSG("cursor release\n");
-		mdb_txn_abort(txn);
-	}
-
-	return ret;
 }
 
 /*                       data serialization                                   */
@@ -113,6 +101,23 @@ static MDB_val pack_key(const knot_dname_t *name)
 	return key;
 }
 
+static int del_entry(MDB_cursor *cur)
+{
+	/* Remember duplicate data count. */
+	size_t rr_count = 0;
+	mdb_cursor_count(cur, &rr_count);
+
+	/* Remove key if last entry. */
+	int ret = 0;
+	if (rr_count == 1) {
+		ret = mdb_cursor_del(cur, MDB_NODUPDATA);
+	} else {
+		ret = mdb_cursor_del(cur, 0);
+	}
+
+	return ret;
+}
+
 static int pack_entry(MDB_cursor *cur, const knot_dname_t *name, uint16_t type,
                       const knot_rdata_t *rdata, uint32_t expire)
 {
@@ -127,7 +132,7 @@ static int pack_entry(MDB_cursor *cur, const knot_dname_t *name, uint16_t type,
 	MDB_val key = pack_key(name);
 	MDB_val data = { datalen, buf };
 
-	return mdb_cursor_put(cur, &key, &data, 0);
+	return mdb_cursor_put(cur, &key, &data, MDB_APPENDDUP);
 }
 
 static int pack_list(MDB_cursor *cur, const knot_rrset_t *rr)
@@ -146,7 +151,7 @@ static int pack_list(MDB_cursor *cur, const knot_rrset_t *rr)
 
 #ifndef NDEBUG
 	char *owner = knot_dname_to_str(rr->owner);
-	DEBUG_MSG("packed RR '%s' TYPE %u, %u RDATA RC=%s\n", owner, rr->type, rr->rrs.rr_count, mdb_strerror(ret));
+	DEBUG_MSG("STORE RR '%s' TYPE %u, %u RDATA => %s\n", owner, rr->type, rr->rrs.rr_count, mdb_strerror(ret));
 	free(owner);
 #endif
 
@@ -163,8 +168,7 @@ static int unpack_entry(MDB_cursor *cur, knot_rrset_t *rr, MDB_val *data, uint32
 
 	/* Check if TTL expired (with negative grace period). */
 	if (knot_rdata_ttl(rd) <= now + KR_TTL_GRACE) {
-		mdb_cursor_del(cur, 0);
-		return 0;
+		return del_entry(cur);
 	}
 
 	return knot_rdataset_add(&rr->rrs, rd, mm);
@@ -172,22 +176,27 @@ static int unpack_entry(MDB_cursor *cur, knot_rrset_t *rr, MDB_val *data, uint32
 
 static int unpack_list(MDB_cursor *cur, knot_rrset_t *rr, mm_ctx_t *mm)
 {
-	int ret = 0;
 	uint32_t now = time(NULL);
 
 	MDB_val key = pack_key(rr->owner);
-	MDB_val data;
+	MDB_val data = { 0, NULL };
+
+	/* Fetch first entry. */
+	int ret = mdb_cursor_get(cur, &key, &data, MDB_SET_KEY);
 
-	while ((ret = mdb_cursor_get(cur, &key, &data, MDB_NEXT_DUP)) == 0) {
+	/* Unpack, and find chained duplicates. */
+	while (ret == 0) {
 		ret = unpack_entry(cur, rr, &data, now, mm);
 		if (ret != 0) {
 			return ret;
 		}
+
+		ret = mdb_cursor_get(cur, &key, &data, MDB_NEXT_DUP);
 	}
 
 #ifndef NDEBUG
 	char *owner = knot_dname_to_str(rr->owner);
-	DEBUG_MSG("unpacked RR '%s' TYPE %u, %u RDATA\n", owner, rr->type, rr->rrs.rr_count);
+	DEBUG_MSG("LOAD RR '%s' TYPE %u => %u RECORDS\n", owner, rr->type, rr->rrs.rr_count);
 	free(owner);
 #endif
 
@@ -224,44 +233,95 @@ void kr_cache_close(struct kr_cache *cache)
 	mm_free(cache->pool, cache);
 }
 
-int kr_cache_query(struct kr_cache *cache, knot_rrset_t *rr, mm_ctx_t *mm)
+struct kr_txn *kr_cache_txn_begin(struct kr_cache *cache, struct kr_txn *parent, mm_ctx_t *mm)
+{
+	struct kr_txn *txn = mm_alloc(mm, sizeof(struct kr_txn));
+	if (txn == NULL) {
+		return NULL;
+	}
+	memset(txn, 0, sizeof(struct kr_txn));
+
+	txn->dbi = cache->dbi;
+	txn->mm  = mm;
+	if (parent) {
+		txn->parent = parent->txn;
+	}
+
+	int ret = mdb_txn_begin(cache->env, txn->parent, 0, &txn->txn);
+	if (ret != 0) {
+		mm_free(mm, txn);
+		return NULL;
+	}
+
+#ifndef NDEBUG
+	MDB_stat stat;
+	mdb_stat(txn->txn, txn->dbi, &stat);
+	DEBUG_MSG("TX_BEGIN, %zu entries\n", stat.ms_entries);
+#endif
+
+	return txn;
+}
+
+int kr_cache_txn_commit(struct kr_txn *txn)
 {
-	MDB_cursor *cursor = cursor_acquire(cache, MDB_RDONLY);
+
+#ifndef NDEBUG
+	MDB_stat stat;
+	mdb_stat(txn->txn, txn->dbi, &stat);
+	DEBUG_MSG("TX_COMMIT, %zu entries\n", stat.ms_entries);
+#endif
+
+	int ret = mdb_txn_commit(txn->txn);
+	mm_free(txn->mm, txn);
+	return ret;
+}
+
+void kr_cache_txn_abort(struct kr_txn *txn)
+{
+
+#ifndef NDEBUG
+	MDB_stat stat;
+	mdb_stat(txn->txn, txn->dbi, &stat);
+	DEBUG_MSG("TX_ABORT, %zu entries\n", stat.ms_entries);
+#endif
+
+	mdb_txn_abort(txn->txn);
+	mm_free(txn->mm, txn);
+}
+
+int kr_cache_query(struct kr_txn *txn, knot_rrset_t *rr)
+{
+	MDB_cursor *cursor = cursor_acquire(txn);
 	if (cursor == NULL) {
 		return -1;
 	}
 
-	int ret = unpack_list(cursor, rr, mm);
+	int ret = unpack_list(cursor, rr, txn->mm);
 	if (knot_rrset_empty(rr)) {
 		ret = -1;
 	}
 
-	cursor_release(cursor, false);
+	cursor_release(cursor);
 	return ret;
 }
 
-int kr_cache_insert(struct kr_cache *cache, const knot_rrset_t *rr, unsigned flags)
+int kr_cache_insert(struct kr_txn *txn, const knot_rrset_t *rr, unsigned flags)
 {
-	MDB_cursor *cursor = cursor_acquire(cache, 0);
+	MDB_cursor *cursor = cursor_acquire(txn);
 	if (cursor == NULL) {
 	                return -1;
 	}
 
 	/* TODO: cache eviction if full */
-
 	int ret = pack_list(cursor, rr);
-	if (ret != 0) {
-		cursor_release(cursor, false);
-		return ret;
-	}
 
-	cursor_release(cursor, true);
+	cursor_release(cursor);
 	return ret;
 }
 
-int kr_cache_remove(struct kr_cache *cache, const knot_rrset_t *rr)
+int kr_cache_remove(struct kr_txn *txn, const knot_rrset_t *rr)
 {
-	MDB_cursor *cursor = cursor_acquire(cache, 0);
+	MDB_cursor *cursor = cursor_acquire(txn);
 	if (cursor == NULL) {
 	                return -1;
 	}
@@ -276,12 +336,12 @@ int kr_cache_remove(struct kr_cache *cache, const knot_rrset_t *rr)
 			mdb_cursor_del(cursor, 0);
 #ifndef NDEBUG
 			char *owner = knot_dname_to_str(rr->owner);
-			DEBUG_MSG("removed RR '%s' TYPE %u (%p)\n", owner, rr->type, PACKED_RDATA(data.mv_data));
+			DEBUG_MSG("DEL RR '%s' TYPE %u (%p)\n", owner, rr->type, PACKED_RDATA(data.mv_data));
 			free(owner);
 #endif
 		}
 	}
 
-	cursor_release(cursor, true);
+	cursor_release(cursor);
 	return ret;
 }
diff --git a/lib/cache.h b/lib/cache.h
index 92dd77f6c..74c1a6126 100644
--- a/lib/cache.h
+++ b/lib/cache.h
@@ -18,10 +18,15 @@ limitations under the License.
 #include <libknot/rrset.h>
 
 struct kr_cache;
+struct kr_txn;
 
 struct kr_cache *kr_cache_open(const char *handle, unsigned flags, mm_ctx_t *mm);
 void kr_cache_close(struct kr_cache *cache);
 
-int kr_cache_query(struct kr_cache *cache, knot_rrset_t *rr, mm_ctx_t *mm);
-int kr_cache_insert(struct kr_cache *cache, const knot_rrset_t *rr, unsigned flags);
-int kr_cache_remove(struct kr_cache *cache, const knot_rrset_t *rr);
\ No newline at end of file
+struct kr_txn *kr_cache_txn_begin(struct kr_cache *cache, struct kr_txn *parent, mm_ctx_t *mm);
+int kr_cache_txn_commit(struct kr_txn *txn);
+void kr_cache_txn_abort(struct kr_txn *txn);
+
+int kr_cache_query(struct kr_txn *txn, knot_rrset_t *rr);
+int kr_cache_insert(struct kr_txn *txn, const knot_rrset_t *rr, unsigned flags);
+int kr_cache_remove(struct kr_txn *txn, const knot_rrset_t *rr);
\ No newline at end of file
diff --git a/lib/context.c b/lib/context.c
index 0ee782db2..53960bf5f 100644
--- a/lib/context.c
+++ b/lib/context.c
@@ -42,6 +42,8 @@ int kr_result_init(struct kr_context *ctx, struct kr_result *result)
 {
 	memset(result, 0, sizeof(struct kr_result));
 
+	/* Initialize answer packet. */
+
 	knot_pkt_t *ans = knot_pkt_new(NULL, KNOT_WIRE_MAX_PKTSIZE, ctx->pool);
 	if (ans == NULL) {
 		return -1;
@@ -49,6 +51,7 @@ int kr_result_init(struct kr_context *ctx, struct kr_result *result)
 
 	struct kr_query *qry = kr_rplan_next(&ctx->rplan);
 	if (qry == NULL) {
+		knot_pkt_free(&ans);
 		return -1;
 	}
 
@@ -58,12 +61,20 @@ int kr_result_init(struct kr_context *ctx, struct kr_result *result)
 
 	result->ans = ans;
 
+	/* Start cache transaction. */
+	result->txn = kr_cache_txn_begin(ctx->cache, NULL, ctx->pool);
+	if (result->txn == NULL) {
+		knot_pkt_free(&ans);
+		return -1;
+	}
+
 	return 0;
 }
 
 int kr_result_deinit(struct kr_result *result)
 {
 	knot_pkt_free(&result->ans);
+	kr_cache_txn_commit(result->txn);
 
 	return 0;
 }
diff --git a/lib/context.h b/lib/context.h
index c9e671145..5a1387559 100644
--- a/lib/context.h
+++ b/lib/context.h
@@ -30,6 +30,7 @@ struct kr_result {
 	struct timeval t_start, t_end;
 	unsigned total_rtt;
 	unsigned nr_queries;
+	struct kr_txn *txn;
 };
 
 /*! \brief Name resolution context. */
diff --git a/lib/layer/iterate.c b/lib/layer/iterate.c
index 8a752c91f..cae054f33 100644
--- a/lib/layer/iterate.c
+++ b/lib/layer/iterate.c
@@ -209,7 +209,7 @@ static int resolve_auth(knot_pkt_t *pkt, struct kr_layer_param *param)
 		}
 
 		/* Update cache. */
-		kr_cache_insert(resolve->cache, &an->rr[i], 0);
+		kr_cache_insert(result->txn, &an->rr[i], 0);
 
 		/* Check canonical name. */
 		follow_cname_chain(&cname, &an->rr[i], param);
@@ -275,7 +275,7 @@ static int prepare_query(knot_layer_t *ctx, knot_pkt_t *pkt)
 	/* TODO: hacked cache */
 	knot_rrset_t cached_reply;
 	knot_rrset_init(&cached_reply, next->sname, next->stype, next->sclass);
-	if (kr_cache_query(resolve->cache, &cached_reply, resolve->pool) == 0) {
+	if (kr_cache_query(result->txn, &cached_reply) == 0) {
 		/* Solve this from cache. */
 		update_result(next, result, &cached_reply);
 		knot_rdataset_clear(&cached_reply.rrs, resolve->pool);