storing client data in tdb and use their given time

Author: avbelov23

etc/tempesta_fw.conf

@@ -1129,3 +1129,39 @@
 # Default:
 #   Health monitor is disabled.
 #
+
+# TAG: client_db
+#
+# Path to a client database file used as a storage for clients info.
+# The same as cache_db.
+#
+# Default:
+#   client_db /opt/tempesta/db/client.tdb;
+#
+
+# TAG: client_tbl_size
+#
+# Size of client drop table.
+#
+# Syntax:
+#   client_tbl_size SIZE
+#
+# Default:
+#   client_tbl_size 16777216;  # 16MB
+#
+
+# TAG: client_lifetime
+#
+# Client life time in seconds. Life time of client accounting data after last
+# client connection was closed. The accounting data is used for Frang limits.
+# Zero value means unlimited life time.
+#
+# Syntax:
+#   client_lifetime NUM;
+#
+# Example:
+#   client_lifetime 3600;
+#
+# Default:
+#   client_lifetime 0;
+#

tempesta_db/core/htrie.c

@@ -806,6 +806,28 @@ tdb_htrie_lookup(TdbHdr *dbh, unsigned long key)
 	return TDB_PTR(dbh, o);
 }
 
+#define TDB_HTRIE_FOREACH_REC(dbh, b_tmp, b, r, body)			\
+	read_lock_bh(&(*b)->lock);					\
+	do {								\
+		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))		\
+				break;					\
+			body;						\
+			r = (TdbRec *)((char *)r + rlen);		\
+		} while ((char *)r + sizeof(*r) - (char *)*b		\
+			 <= TDB_HTRIE_MINDREC);				\
+		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)
 /**
  * Iterate over all records in collision chain with locked buckets.
  * Buckets are inspected according to following rules:
@@ -821,29 +843,11 @@ tdb_htrie_bscan_for_rec(TdbHdr *dbh, TdbBucket **b, unsigned long key)
 	TdbBucket *b_tmp;
 	TdbRec *r;
 
-	read_lock_bh(&(*b)->lock);
-
-	do {
-		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))
-				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);
-
-		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);
+	TDB_HTRIE_FOREACH_REC(dbh, b_tmp, b, r, {
+		if (tdb_live_rec(dbh, r) && r->key == key)
+			/* Unlock the bucket by tdb_rec_put(). */
+			return r;
+	});
 
 	return NULL;
 }
@@ -931,3 +935,76 @@ tdb_htrie_exit(TdbHdr *dbh)
 {
 	free_percpu(dbh->pcpu);
 }
+
+static int
+tdb_htrie_bucket_walk(TdbHdr *dbh, TdbBucket *b, int (*fn)(void *))
+{
+	TdbBucket *b_tmp;
+	TdbRec *r;
+
+	TDB_HTRIE_FOREACH_REC(dbh, b_tmp, &b, r, {
+		if (tdb_live_rec(dbh, r)) {
+			int res = fn(r->data);
+			if (unlikely(res)) {
+				read_unlock_bh(&b->lock);
+				return res;
+			}
+		}
+	});
+
+	return 0;
+}
+
+static int
+tdb_htrie_node_visit(TdbHdr *dbh, TdbHtrieNode *node, int (*fn)(void *))
+{
+	int bits;
+	int res;
+
+	for (bits = 0; bits < TDB_HTRIE_FANOUT; ++bits) {
+		unsigned long o;
+
+		BUG_ON(TDB_HTRIE_RESOLVED(bits));
+
+		o = node->shifts[bits];
+
+		if (likely(!o))
+			continue;
+
+		BUG_ON(TDB_DI2O(o & ~TDB_HTRIE_DBIT) < TDB_HDR_SZ(dbh) + sizeof(TdbExt)
+			   || TDB_DI2O(o & ~TDB_HTRIE_DBIT) > dbh->dbsz);
+
+		if (o & TDB_HTRIE_DBIT) {
+			TdbBucket *b;
+
+			/* We're at a data pointer - resolve it. */
+			o ^= TDB_HTRIE_DBIT;
+			BUG_ON(!o);
+
+			b = (TdbBucket *)TDB_PTR(dbh, TDB_DI2O(o));
+			res = tdb_htrie_bucket_walk(dbh, b, fn);
+			if (unlikely(res))
+				return res;
+		} else {
+			/*
+			 * The recursion depth being hard-limited.
+			 * The function has the deepest nesting 16 and uses only
+			 * about 16 bytes of stack.
+			 */
+			res = tdb_htrie_node_visit(dbh, TDB_PTR(dbh,
+						   TDB_II2O(o)), fn);
+			if (unlikely(res))
+				return res;
+		}
+	}
+
+	return 0;
+}
+
+int
+tdb_htrie_walk(TdbHdr *dbh, int (*fn)(void *))
+{
+	TdbHtrieNode *node = TDB_HTRIE_ROOT(dbh);
+
+	return tdb_htrie_node_visit(dbh, node, fn);
+}

tempesta_db/core/htrie.h

@@ -140,5 +140,6 @@ 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);
+int tdb_htrie_walk(TdbHdr *dbh, int (*fn)(void *));
 
 #endif /* __HTRIE_H__ */

tempesta_db/core/main.c

@@ -28,13 +28,15 @@
 #include "table.h"
 #include "tdb_if.h"
 
-#define TDB_VERSION	"0.1.16"
+#define TDB_VERSION	"0.1.17"
 
 MODULE_AUTHOR("Tempesta Technologies");
 MODULE_DESCRIPTION("Tempesta DB");
 MODULE_VERSION(TDB_VERSION);
 MODULE_LICENSE("GPL");
 
+static DEFINE_SPINLOCK(get_alloc_lock);
+
 /**
  * Create TDB entry and copy @len contiguous bytes from @data to the entry.
  */
@@ -52,6 +54,7 @@ EXPORT_SYMBOL(tdb_entry_create);
 
 /**
  * Create TDB entry to store @len bytes.
+ * TODO #515 function must holds a lock upon return
  */
 TdbRec *
 tdb_entry_alloc(TDB *db, unsigned long key, size_t *len)
@@ -222,6 +225,72 @@ tdb_get_db(const char *path, int node)
 	return tdb_get(db);
 }
 
+/**
+ * Lookup and get a record if the record is found or create TDB entry to store
+ * @len bytes. If record exist then since we don't copy returned records,
+ * we have to lock the memory location where the record is placed and
+ * the user must call tdb_rec_put() when finish with the record.
+ *
+ * The caller must not call sleeping functions during work with the record.
+ * Typically there is only one large record per bucket, so the bucket lock
+ * is exactly the same as to lock the record. While there could be many
+ * small records in a bucket, so the caller should not perform long jobs
+ * with small records.
+ *
+ * @return pointer to record with acquired bucket lock if the record is
+ * found and create TDB entry without acquired locks otherwise.
+ *
+ * TODO #515 rework the function is lock-free way
+ */
+TdbRec *
+tdb_rec_get_alloc(TDB *db, unsigned long key, size_t *len,
+                  bool (*predicate)(TdbRec *, void *), bool *is_new, void *data)
+{
+	TdbIter iter;
+	TdbRec *r;
+
+	spin_lock(&get_alloc_lock);
+
+	*is_new = false;
+	iter = tdb_rec_get(db, key);
+	while (!TDB_ITER_BAD(iter)) {
+		if ((*predicate)(iter.rec, data)) {
+			spin_unlock(&get_alloc_lock);
+			return iter.rec;
+		}
+		tdb_rec_next(db, &iter);
+	}
+
+	*is_new = true;
+	r = tdb_entry_alloc(db, key, len);
+
+	spin_unlock(&get_alloc_lock);
+
+	return r;
+}
+EXPORT_SYMBOL(tdb_rec_get_alloc);
+
+int
+tdb_entry_walk(TDB *db, int (*fn)(void *))
+{
+	return tdb_htrie_walk(db->hdr, fn);
+}
+EXPORT_SYMBOL(tdb_entry_walk);
+
+void
+tdb_rec_get_lock(void *rec)
+{
+	TdbBucket *b;
+
+	BUG_ON(!rec);
+
+	b = (TdbBucket *)((unsigned long)rec & TDB_HTRIE_DMASK);
+	BUG_ON(!b);
+
+	read_lock_bh(&b->lock);
+}
+EXPORT_SYMBOL(tdb_rec_get_lock);
+
 /**
  * Open database file and @return its descriptor.
  * If the database is already opened, then returns the handler.

tempesta_db/core/tdb.h

@@ -173,6 +173,11 @@ 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);
+TdbRec *tdb_rec_get_alloc(TDB *db, unsigned long key, size_t *len,
+                          bool (*predicate)(TdbRec *, void *), bool *is_new,
+                          void *data);
+int tdb_entry_walk(TDB *db, int (*fn)(void *));
+void tdb_rec_get_lock(void *rec);
 
 /* Open/close database handler. */
 TDB *tdb_open(const char *path, size_t fsize, unsigned int rec_size, int node);