Fix #122

README.md

@@ -298,7 +298,7 @@ The list of available options:
 Let's see a simple example to understand Tempesta filtering.
 
 Firstly, run Tempesta with enabled [Frang](#Frang) and put some load onto the
-system to make Frang generat a blocking rule:
+system to make Frang generate a blocking rule:
 
         $ dmesg | grep frang
         [tempesta] Warning: frang: connections max num. exceeded for ::ffff:7f00:1: 9 (lim=8)

etc/tempesta_fw.conf

@@ -110,15 +110,19 @@
 
 # TAG: cache
 #
-# Boolean value to enable or disable Web content caching.
-# It can be useful to switch caching off to run Tempesta on the same host as
-# protected HTTP accelerator.
+# Web content caching mode:
+#   0 - no caching, pure proxying mode;
+#   1 - sharding, each NUMA node contains independent shard of whole cache.
+#       This mode has the smallest memory requirements;
+#   2 - replicated, each NUMA node has whole replica of the cache.
+#       It requires more RAM, but delivers the highest performance.
+#       This is default mode.
 #
 # Syntax:
-#   cache on | off
+#   cache [0-2]
 #
 # Default:
-#   cache on;
+#   cache 2;
 
 # TAG: cache_db
 # 

tempesta_db/core/htrie.c

@@ -299,6 +299,9 @@ tdb_alloc_data(TdbHdr *dbh, size_t *len, int bucket_hdr)
 
 		max_data_len = TDB_BLK_SZ - (rptr & ~TDB_BLK_MASK);
 		if (res_len > max_data_len) {
+			TDB_DBG("cannot allocate %lu bytes,"
+				" %lu will be allocated instead\n",
+				res_len, max_data_len);
 			res_len = max_data_len;
 			*len = res_len - hdr_len;
 		}
@@ -795,34 +798,74 @@ tdb_htrie_lookup(TdbHdr *dbh, unsigned long key)
  * deleted/evicted records in it.
  */
 TdbRec *
-tdb_htrie_bscan_for_rec(TdbHdr *dbh, TdbBucket *b, unsigned long key)
+tdb_htrie_bscan_for_rec(TdbHdr *dbh, TdbBucket **b, unsigned long key)
 {
 	TdbBucket *b_tmp;
 	TdbRec *r;
 
-	read_lock_bh(&b->lock);
+	read_lock_bh(&(*b)->lock);
 
 	do {
-		r = TDB_HTRIE_BCKT_1ST_REC(b);
+		r = TDB_HTRIE_BCKT_1ST_REC(*b);
 		do {
 			size_t rlen = sizeof(*r) + TDB_HTRIE_RBODYLEN(dbh, r);
 			rlen = TDB_HTRIE_RALIGN(rlen);
-			if ((char *)r + rlen - (char *)b > TDB_HTRIE_MINDREC
-			    && r != TDB_HTRIE_BCKT_1ST_REC(b))
+			if ((char *)r + rlen - (char *)*b > TDB_HTRIE_MINDREC
+			    && r != TDB_HTRIE_BCKT_1ST_REC(*b))
 				break;
 			if (tdb_live_rec(dbh, r) && r->key == key)
 				/* Unlock the bucket by tdb_rec_put(). */
 				return r;
 			r = (TdbRec *)((char *)r + rlen);
-		} while ((char *)r + sizeof(*r) - (char *)b
+		} while ((char *)r + sizeof(*r) - (char *)*b
 			 <= TDB_HTRIE_MINDREC);
 
-		b_tmp = TDB_HTRIE_BUCKET_NEXT(dbh, b);
+		b_tmp = TDB_HTRIE_BUCKET_NEXT(dbh, *b);
 		if (b_tmp)
 			read_lock_bh(&b_tmp->lock);
-		read_unlock_bh(&b->lock);
-		b = b_tmp;
-	} while (b);
+		read_unlock_bh(&(*b)->lock);
+		*b = b_tmp;
+	} while (*b);
+
+	return NULL;
+}
+
+/**
+ * Called with already locked bucket by tdb_htrie_lookup().
+ * Unlocks the last bucked when all records are read from it.
+ */
+TdbRec *
+tdb_htrie_next_rec(TdbHdr *dbh, TdbRec *r, TdbBucket **b, unsigned long key)
+{
+	TdbBucket *_b = *b;
+
+	do {
+		size_t rlen = TDB_HTRIE_RALIGN(sizeof(*r)
+					       + TDB_HTRIE_RBODYLEN(dbh, r));
+		if ((char *)r + rlen - (char *)_b > TDB_HTRIE_MINDREC)
+			goto next_bckt;
+		r = (TdbRec *)((char *)r + rlen);
+
+		do {
+			rlen = TDB_HTRIE_RALIGN(sizeof(*r)
+						+ TDB_HTRIE_RBODYLEN(dbh, r));
+			if ((char *)r + rlen - (char *)_b > TDB_HTRIE_MINDREC)
+				break;
+			if (tdb_live_rec(dbh, r) && r->key == key)
+				/* Unlock the bucket by tdb_rec_put(). */
+				return r;
+			r = (TdbRec *)((char *)r + rlen);
+		} while ((char *)r + sizeof(*r) - (char *)_b
+			 <= TDB_HTRIE_MINDREC);
+next_bckt:
+		*b = TDB_HTRIE_BUCKET_NEXT(dbh, _b);
+		if (*b) {
+			read_lock_bh(&(*b)->lock);
+			r = TDB_HTRIE_BCKT_1ST_REC(*b);
+		}
+		read_unlock_bh(&_b->lock);
+		_b = *b;
+	} while (_b);
 
 	return NULL;
 }

tempesta_db/core/htrie.h

@@ -10,8 +10,8 @@
  * or (at your option) any later version.
  *
  * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
- * FOR A PARTICULAR PURPOSE.
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
  * See the GNU General Public License for more details.
  *
  * You should have received a copy of the GNU General Public License along with
@@ -134,7 +134,9 @@ TdbVRec *tdb_htrie_extend_rec(TdbHdr *dbh, TdbVRec *rec, size_t size);
 TdbRec *tdb_htrie_insert(TdbHdr *dbh, unsigned long key, void *data,
 			 size_t *len);
 TdbBucket *tdb_htrie_lookup(TdbHdr *dbh, unsigned long key);
-TdbRec *tdb_htrie_bscan_for_rec(TdbHdr *dbh, TdbBucket *b, unsigned long key);
+TdbRec *tdb_htrie_bscan_for_rec(TdbHdr *dbh, TdbBucket **b, unsigned long key);
+TdbRec *tdb_htrie_next_rec(TdbHdr *dbh, TdbRec *r, TdbBucket **b,
+			   unsigned long key);
 TdbHdr *tdb_htrie_init(void *p, size_t db_size, unsigned int rec_len);
 void tdb_htrie_exit(TdbHdr *dbh);
 

tempesta_db/core/if.c

@@ -176,6 +176,7 @@ tdb_if_insert(struct sk_buff *skb, struct netlink_callback *cb)
 static int
 tdb_if_select(struct sk_buff *skb, struct netlink_callback *cb)
 {
+	TdbIter iter;
 	unsigned long key;
 	TdbMsg *resp_m, *m = cb->data;
 	TdbRec *res;
@@ -204,7 +205,8 @@ tdb_if_select(struct sk_buff *skb, struct netlink_callback *cb)
 	 * 2. use many netlink frames to send probably large data set.
 	 */
 	key = tdb_hash_calc(m->recs[0].data, m->recs[0].klen);
-	res = tdb_rec_get(db, key);
+	iter = tdb_rec_get(db, key);
+	res = iter.rec;
 	if (res) {
 		resp_m->rec_n = 1;
 		if (TDB_HTRIE_VARLENRECS(db->hdr)) {

tempesta_db/core/main.c

@@ -12,8 +12,8 @@
  * or (at your option) any later version.
  *
  * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
- * FOR A PARTICULAR PURPOSE.
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
  * See the GNU General Public License for more details.
  *
  * You should have received a copy of the GNU General Public License along with
@@ -28,7 +28,7 @@
 #include "table.h"
 #include "tdb_if.h"
 
-#define TDB_VERSION	"0.1.14"
+#define TDB_VERSION	"0.1.15"
 
 MODULE_AUTHOR("Tempesta Technologies");
 MODULE_DESCRIPTION("Tempesta DB");
@@ -60,6 +60,25 @@ tdb_entry_add(TDB *db, TdbVRec *r, size_t size)
 }
 EXPORT_SYMBOL(tdb_entry_add);
 
+/**
+ * Check available room in @trec and allocate new record if it's not enough.
+ * Chop tail of @trec if we allocated more space, but can't use the tail
+ * w/o data fragmentation.
+ */
+void *
+tdb_entry_get_room(TDB *db, TdbVRec **r, char *curr_ptr, size_t tail_len,
+		   size_t tot_size)
+{
+	if (likely((*r)->data + (*r)->len - curr_ptr >= tail_len))
+		return curr_ptr;
+
+	(*r)->len -= curr_ptr - (*r)->data;
+
+	*r = tdb_htrie_extend_rec(db->hdr, *r, tot_size);
+	return *r ? (*r)->data : NULL;
+}
+EXPORT_SYMBOL(tdb_entry_get_room);
+
 /**
  * Lookup and get a record.
  * Since we don't copy returned records, we have to lock the memory location
@@ -75,22 +94,46 @@ EXPORT_SYMBOL(tdb_entry_add);
  * @return pointer to record with acquired bucket lock if the record is
  * found and NULL without acquired locks otherwise.
  */
-void *
+TdbIter
 tdb_rec_get(TDB *db, unsigned long key)
 {
-	TdbBucket *b = tdb_htrie_lookup(db->hdr, key);
-	if (!b)
-		return NULL;
+	TdbIter iter = { NULL };
+
+	iter.bckt = tdb_htrie_lookup(db->hdr, key);
+	if (!iter.bckt)
+		goto out;
 
-	return tdb_htrie_bscan_for_rec(db->hdr, b, key);
+	iter.rec = tdb_htrie_bscan_for_rec(db->hdr, (TdbBucket **)&iter.bckt,
+					   key);
+out:
+	return iter;
 }
 EXPORT_SYMBOL(tdb_rec_get);
 
+/**
+ * Get next record from full key collision chain.
+ */
+void
+tdb_rec_next(TDB *db, TdbIter *iter)
+{
+	BUG_ON(!iter->bckt);
+
+	iter->rec = tdb_htrie_next_rec(db->hdr, iter->rec,
+				       (TdbBucket **)&iter->bckt,
+				       iter->rec->key);
+}
+EXPORT_SYMBOL(tdb_rec_next);
+
 void
 tdb_rec_put(void *rec)
 {
-	TdbBucket *b = (TdbBucket *)TDB_HTRIE_DALIGN((unsigned long)rec);
+	TdbBucket *b;
+
+	BUG_ON(!rec);
+
+	b = (TdbBucket *)TDB_HTRIE_DALIGN((unsigned long)rec);
 	BUG_ON(!b);
+
 	read_unlock_bh(&b->lock);
 }
 EXPORT_SYMBOL(tdb_rec_put);
@@ -121,14 +164,14 @@ tdb_info(char *buf, size_t len)
  * characters in long) followed by TDB_SUFFIX.
  */
 static TDB *
-tdb_get_db(const char *path)
+tdb_get_db(const char *path, int node)
 {
-	int len;
+	int full_len, len;
 	char *slash;
 	TDB *db;
 
-	len = strlen(path);
-	if (strncmp(path + len - sizeof(TDB_SUFFIX) + 1,
+	full_len = strlen(path);
+	if (strncmp(path + full_len - sizeof(TDB_SUFFIX) + 1,
 		    TDB_SUFFIX, sizeof(TDB_SUFFIX) - 1))
 	{
 		TDB_ERR("Bad table suffix for %s\n", path);
@@ -139,8 +182,8 @@ tdb_get_db(const char *path)
 		TDB_ERR("Please specify absolute path to %s\n", path);
 		return NULL;
 	}
-	len = len - (slash - path) - sizeof(TDB_SUFFIX);
-	if (len > TDB_TBLNAME_LEN) {
+	len = full_len - (slash - path) - sizeof(TDB_SUFFIX);
+	if (len >= TDB_TBLNAME_LEN) {
 		TDB_ERR("Too long table name %s\n", path);
 		return NULL;
 	}
@@ -154,8 +197,10 @@ tdb_get_db(const char *path)
 		TDB_ERR("Cannot allocate new db handler\n");
 		return NULL;
 	}
-	strncpy(db->path, path, TDB_PATH_LEN - 1);
-	strncpy(db->tbl_name, slash + 1, len);
+	snprintf(db->path, TDB_PATH_LEN, "%.*s%X.tdb",
+		 (int)(full_len - sizeof(TDB_SUFFIX)), path, node);
+	snprintf(db->tbl_name, TDB_TBLNAME_LEN, "%.*s%X.tdb",
+		 (int)(len - sizeof(TDB_SUFFIX)), slash + 1, node);
 
 	return tdb_get(db);
 }
@@ -176,7 +221,7 @@ tdb_open(const char *path, size_t fsize, unsigned int rec_size, int node)
 		return NULL;
 	}
 
-	db = tdb_get_db(path);
+	db = tdb_get_db(path, node);
 	if (!db)
 		return NULL;
 
@@ -223,6 +268,9 @@ __do_close_table(TDB *db)
 void
 tdb_close(TDB *db)
 {
+	if (!db)
+		return;
+
 	if (!atomic_dec_and_test(&db->count))
 		return;
 

tempesta_db/core/tdb.h

@@ -108,6 +108,16 @@ typedef struct {
 /* Common interface for database records of all kinds. */
 typedef TdbFRec TdbRec;
 
+/**
+ * Iterator for TDB full key collision chains.
+ */
+typedef struct {
+	TdbRec	*rec;
+	void	*bckt;
+} TdbIter;
+
+#define TDB_ITER_BAD(i)		(!(i).rec)
+
 /**
  * We use very small index nodes size of only one cache line.
  * So overall memory footprint of the index is mininal by a cost of more LLC
@@ -123,8 +133,10 @@ typedef TdbFRec TdbRec;
  */
 #define TDB_HTRIE_MINDREC	(L1_CACHE_BYTES * 2)
 
-/* Convert internal offsets to system pointer. */
+/* Convert internal offset to system pointer. */
 #define TDB_PTR(h, o)		(void *)((char *)(h) + (o))
+/* Convert system pointer to internal offset. */
+#define TDB_OFF(h, p)		(long)((char *)(p) - (char *)(h))
 /* Get index and data block indexes by byte offset and vise versa. */
 #define TDB_O2DI(o)		((o) / TDB_HTRIE_MINDREC)
 #define TDB_O2II(o)		((o) / TDB_HTRIE_NODE_SZ)
@@ -151,7 +163,10 @@ typedef TdbFRec TdbRec;
  */
 TdbRec *tdb_entry_create(TDB *db, unsigned long key, void *data, size_t *len);
 TdbVRec *tdb_entry_add(TDB *db, TdbVRec *r, size_t size);
-void *tdb_rec_get(TDB *db, unsigned long key);
+void *tdb_entry_get_room(TDB *db, TdbVRec **r, char *curr_ptr, size_t tail_len,
+			 size_t tot_size);
+TdbIter tdb_rec_get(TDB *db, unsigned long key);
+void tdb_rec_next(TDB *db, TdbIter *iter);
 void tdb_rec_put(void *rec);
 int tdb_info(char *buf, size_t len);
 
@@ -175,4 +190,10 @@ tdb_put(TDB *db)
 		kfree(db);
 }
 
+static inline TdbVRec *
+tdb_next_rec_chunk(TDB *db, TdbVRec *r)
+{
+	return TDB_PTR(db->hdr, TDB_DI2O(r->chunk_next));
+}
+
 #endif /* __TDB_H__ */