Hash: Added a spinlocked variant

The spinlocked hash has a main rw spinlock for the data blocks
and then a rw spinlock for each hash chain. Rehashing is asynchronous,
running from an event, and it happens chain-wise, never blocking more
than one chain at a time.

lib/hash.h

@@ -1,8 +1,9 @@
 /*
  *	BIRD Library -- Generic Hash Table
  *
- *	(c) 2013 Ondrej Zajicek <santiago@crfreenet.org>
+ *	(c) 2013	Ondrej Zajicek <santiago@crfreenet.org>
- *	(c) 2013 CZ.NIC z.s.p.o.
+ *	(c) 2024	Maria Matejka <mq@jmq.cz>
+ *	(c) 2013--2024	CZ.NIC z.s.p.o.
  *
  *	Can be freely distributed and used under the terms of the GNU GPL.
  */
@@ -10,13 +11,17 @@
 #ifndef _BIRD_HASH_H_
 #define _BIRD_HASH_H_
 
+/*
+ * Regular hash table
+ */
+
 #define HASH(type)		struct { type **data; uint count; u8 order; }
 #define HASH_TYPE(v)		typeof(** (v).data)
 #define HASH_SIZE(v)		(1U << (v).order)
 
 #define HASH_EQ(v,id,k1,k2...)	(id##_EQ(k1, k2))
 #define HASH_FN(v,id,key...)	((u32) (id##_FN(key)) >> (32 - (v).order))
-
+#define HASH_FNO(id,key...)	id##_FN(key)
 
 #define HASH_INIT(v,pool,init_order)					\
   ({									\
@@ -202,19 +207,260 @@
 #define HASH_WALK_FILTER_END } while (0)
 
 
-#define HASH_WALK_ITER(v, id, n, iter)					\
+/*
+ * Atomic hash table with data-local spinlocks
+ */
+
+#define SPINHASH(type)							\
+struct {								\
+  _Atomic uint count;							\
+  rw_spinlock lock;							\
+  uint cur_order, new_order;						\
+  struct { type *data; rw_spinlock lock; } *cur, *new;			\
+  pool *pool;								\
+  event rehash;								\
+  event_list *target;							\
+}
+
+#define SPINHASH_INIT(v,id,_pool,_target)				\
+  ({									\
+    atomic_store_explicit(&(v).count, 0, memory_order_relaxed);		\
+    (v).cur_order = id##_ORDER;						\
+    (v).new_order = 0;							\
+    (v).cur = mb_allocz(_pool, (1U << id##_ORDER) * sizeof *(v).cur);	\
+    (v).new = NULL;							\
+    (v).pool = _pool;							\
+    (v).rehash = (event) { .hook = id##_REHASH, .data = &(v), };	\
+    (v).target = _target;						\
+  })
+
+#define SPINHASH_FREE(v)						\
+  ({									\
+    ev_postpone(&(v).rehash);						\
+    mb_free((v).cur);							\
+    ASSERT_DIE((v).new == NULL);					\
+    (v).cur = NULL;							\
+    (v).cur_order = 0;							\
+    (v).pool = NULL;							\
+    (v).target = NULL;							\
+  })
+
+#define SPINHASH_BEGIN_CHAIN(v,id,rw,n,key...)				\
   do {									\
-    uint _hash_walk_iter_put = 0;					\
+    typeof (&v) _v = &(v);						\
-    uint _shift = 32 - (v).order;					\
+    rws_read_lock(&_v->lock);						\
-    for ( ; !_hash_walk_iter_put; (iter) += (1U << _shift)) {		\
+    u32 _hh = id##_FN(key);						\
-      _hash_walk_iter_put = ((iter) + (1U << _shift) == 0);		\
+    SPINHASH_BEGIN_CHAIN_INDEX(v,_hh,rw,n);				\
-      for (HASH_TYPE(v) *n = (v).data[(iter) >> _shift]; n; n = id##_NEXT((n)))\
-	if (HASH_FN(v, id, id##_KEY(n)) >= ((iter) >> _shift))		\
 
-#define HASH_WALK_ITER_PUT	(_hash_walk_iter_put = 1)
+#define SPINHASH_BEGIN_CHAIN_INDEX(v,h,rw,n)				\
+    u32 _ch = (h) >> (32 - (v).cur_order);				\
+    rw_spinlock *_lock = &(v).cur[_ch].lock;				\
+    rws_##rw##_lock(_lock);						\
+    typeof (&(v).cur[_ch].data) n = &(v).cur[_ch].data;			\
+    if (*n == SPINHASH_REHASH_SENTINEL) {				\
+      rws_##rw##_unlock(_lock);						\
+      u32 _nh = (h) >> (32 - (v).new_order);				\
+      _lock = &(v).new[_nh].lock;					\
+      rws_##rw##_lock(_lock);						\
+      n = &(v).new[_nh].data;						\
+      ASSERT_DIE(*n != SPINHASH_REHASH_SENTINEL);			\
+    };
 
-#define HASH_WALK_ITER_END } } while (0)
+#define SPINHASH_END_CHAIN_INDEX(rw)					\
+    rws_##rw##_unlock(_lock);						\
 
+#define SPINHASH_END_CHAIN(rw)						\
+    SPINHASH_END_CHAIN_INDEX(rw);					\
+    rws_read_unlock(&_v->lock);						\
+  } while (0)
+
+#define SPINHASH_FIND(v,id,key...)					\
+  ({									\
+    typeof ((v).cur[0].data) _n;					\
+    SPINHASH_BEGIN_CHAIN(v,id,read,_c,key);				\
+    while ((*_c) && !HASH_EQ(v,id,id##_KEY((*_c)), key))		\
+      _c = &id##_NEXT((*_c));						\
+    _n = *_c;								\
+    SPINHASH_END_CHAIN(read);						\
+    _n;									\
+  })
+
+#define SPINHASH_INSERT(v,id,n)						\
+  do {									\
+    rws_read_lock(&(v).lock);						\
+    uint _h = HASH_FNO(id, id##_KEY(n));				\
+    uint _ch = _h >> (32 - (v).cur_order);				\
+    rws_write_lock(&(v).cur[_ch].lock);					\
+    if ((v).cur[_ch].data == SPINHASH_REHASH_SENTINEL) {		\
+      uint _nh = _h >> (32 - (v).new_order);				\
+      rws_write_lock(&(v).new[_nh].lock);				\
+      ASSERT_DIE((v).new[_nh].data != SPINHASH_REHASH_SENTINEL);	\
+      id##_NEXT(n) = (v).new[_nh].data;					\
+      (v).new[_nh].data = n;						\
+      rws_write_unlock(&(v).new[_nh].lock);				\
+    } else {								\
+      id##_NEXT(n) = (v).cur[_ch].data;					\
+      (v).cur[_ch].data = n;						\
+    }									\
+    rws_write_unlock(&(v).cur[_ch].lock);				\
+    uint count = atomic_fetch_add_explicit(&(v).count, 1, memory_order_relaxed);\
+    SPINHASH_REQUEST_REHASH(v,id,count);				\
+    rws_read_unlock(&(v).lock);						\
+  } while (0)								\
+
+#define SPINHASH_REMOVE(v,id,n)						\
+  do {									\
+    typeof(n) _n = (n);							\
+    SPINHASH_BEGIN_CHAIN(v,id,write,_c,id##_KEY(_n))			\
+      for (; *_c; _c = &id##_NEXT((*_c)))				\
+	if (_n == *_c) {						\
+	  SPINHASH_DO_REMOVE(v,id,_c);					\
+	  break;							\
+	}								\
+    SPINHASH_END_CHAIN(write);						\
+    uint count = atomic_load_explicit(&(v).count, memory_order_relaxed);\
+    SPINHASH_REQUEST_REHASH(v,id,count);				\
+  } while (0)
+
+#define SPINHASH_DO_REMOVE(v,id,c)					\
+  atomic_fetch_sub_explicit(&(v).count, 1, memory_order_relaxed);	\
+  *c = id##_NEXT((*c));							\
+
+#define SPINHASH_WALK(v,id,n)						\
+  SPINHASH_WALK_CHAINS(v,id,read,nn)					\
+  for (typeof (*nn) n = *nn; n; n = id##_NEXT(n)) {			\
+
+#define SPINHASH_WALK_END						\
+  }									\
+  SPINHASH_WALK_CHAINS_END(read)					\
+
+#define SPINHASH_WALK_FILTER(v,id,rw,nn)				\
+  SPINHASH_WALK_CHAINS(v,id,rw,nn)					\
+  for (; nn && *nn; nn = nn ? &id##_NEXT((*nn)) : NULL)
+
+#define SPINHASH_WALK_FILTER_END(rw) SPINHASH_WALK_CHAINS_END(rw)
+
+#define SPINHASH_WALK_CHAINS(v,id,rw,nn)				\
+  do {									\
+    typeof (&v) _v = &(v);						\
+    rws_read_lock(&_v->lock);						\
+    for (uint _h = 0; !(_h >> _v->cur_order); _h++) {			\
+      SPINHASH_BEGIN_CHAIN_INDEX(v,_h,rw,nn);				\
+
+#define SPINHASH_WALK_CHAINS_END(rw)					\
+      SPINHASH_END_CHAIN_INDEX(rw);					\
+    }									\
+    rws_read_unlock(&_v->lock);						\
+  } while (0)
+
+#define SPINHASH_CHECK_REHASH(v,id,count) SPINHASH_CHECK_REHASH_(v,id,count,id##_PARAMS)
+
+#define SPINHASH_CHECK_REHASH_(v,id,count,args)				\
+  ({									\
+    uint order = (v).new_order ?: (v).cur_order;			\
+    uint size = 1U << order;						\
+    ((count > size REHASH_HI_MARK(args)) && (order < REHASH_HI_BOUND(args))) ? \
+    REHASH_HI_STEP(args) :						\
+    ((count < size REHASH_LO_MARK(args)) && (order > REHASH_LO_BOUND(args))) ? \
+    -REHASH_LO_STEP(args) :						\
+    0;									\
+  })
+
+#define SPINHASH_REQUEST_REHASH(v,id,count)				\
+  if (SPINHASH_CHECK_REHASH(v,id,count) && (v).target)			\
+      ev_send((v).target, &(v).rehash);
+
+#define SPINHASH_DEFINE_REHASH_FN(id,type)				\
+static void id##_REHASH(void *_v) {					\
+  SPINHASH(type) *v = _v;						\
+  SPINHASH_REHASH_FN_BODY(v,id,type);					\
+}
+
+#define SPINHASH_REHASH_FN_BODY(v,id,type)				\
+  int step;								\
+  SPINHASH_REHASH_PREPARE(v,id,type,step);				\
+  if (step) {								\
+    if (step > 0) SPINHASH_REHASH_UP(id,type,step);			\
+    if (step < 0) SPINHASH_REHASH_DOWN(id,type,-step);			\
+    SPINHASH_REHASH_FINISH(v,id);					\
+  }									\
+
+#define SPINHASH_REHASH_PREPARE(v,id,type,step)				\
+  rws_write_lock(&(v)->lock);						\
+  ASSERT_DIE((v)->new_order == 0);					\
+  uint _cb = atomic_load_explicit(&(v)->count, memory_order_relaxed);	\
+  step = SPINHASH_CHECK_REHASH((*(v)),id,_cb);				\
+  if (step) {								\
+    (v)->new_order = (v)->cur_order + step;				\
+    uint nsz = 1U << (v)->new_order;					\
+    (v)->new = mb_alloc((v)->pool, nsz * sizeof *(v)->new);		\
+    for (uint i=0; i<nsz; i++) {					\
+      (v)->new[i].data = SPINHASH_REHASH_SENTINEL;			\
+      (v)->new[i].lock = (rw_spinlock) {};				\
+    }									\
+  }									\
+  rws_write_unlock(&(v)->lock);						\
+
+#define SPINHASH_REHASH_FINISH(v,id)					\
+  ASSERT_DIE(step);							\
+  rws_write_lock(&(v)->lock);						\
+  (v)->cur = (v)->new; (v)->cur_order = (v)->new_order;			\
+  (v)->new = NULL; (v)->new_order = 0;					\
+  uint _ce = atomic_load_explicit(&(v)->count, memory_order_relaxed);	\
+  SPINHASH_REQUEST_REHASH(*(v),id,_ce)					\
+  rws_write_unlock(&(v)->lock);						\
+  mb_free((v)->new);							\
+
+#define SPINHASH_REHASH_UP(v,id,type,step)				\
+  for (uint i=0; !(i >> (v)->cur_order); i++) {				\
+    rws_write_lock(&(v)->cur[i].lock);					\
+    for (uint p=0; !(p >> step); p++) {					\
+      uint ppos = (i << step) | p;					\
+      rws_write_lock(&(v)->new[ppos].lock);				\
+      ASSERT_DIE((v)->new[ppos].data == SPINHASH_REHASH_SENTINEL);	\
+      (v)->new[ppos].data = NULL;					\
+    }									\
+    for (type *n; n = (v)->cur[i].data; ) {				\
+      (v)->cur[i].data = id##_NEXT(n);					\
+      uint _h = HASH_FNO(id, id##_KEY(n));				\
+      ASSERT_DIE((_h >> (32 - (v)->cur_order)) == i);			\
+      uint _nh = _h >> (32 - (v)->new_order);				\
+      id##_NEXT(n) = (v)->new[_nh].data;				\
+      (v)->new[_nh].data = n;						\
+    }									\
+    (v)->cur[i].data = SPINHASH_REHASH_SENTINEL;			\
+    for (uint p=0; !(p >> step); p++)					\
+      rws_write_unlock(&(v)->new[((i+1) << step) - p - 1].lock);	\
+    rws_write_unlock(&(v)->cur[i].lock);				\
+  }									\
+
+#define SPINHASH_REHASH_DOWN(v,id,type,step)				\
+  for (uint i=0; !(i >> (v)->cur_order); i++) {				\
+    uint p = i >> step;							\
+    rws_write_lock(&(v)->cur[i].lock);					\
+    rws_write_lock(&(v)->new[p].lock);					\
+    if (i == (p << step)) {						\
+      ASSERT_DIE((v)->new[p].data == SPINHASH_REHASH_SENTINEL);		\
+      (v)->new[p].data = NULL;						\
+    } else								\
+      ASSERT_DIE((v)->new[p].data != SPINHASH_REHASH_SENTINEL);		\
+    for (type *n; n = (v)->cur[i].data; ) {				\
+      (v)->cur[i].data = id##_NEXT(n);					\
+      id##_NEXT(n) = (v)->new[p].data;					\
+      (v)->new[p].data = n;						\
+    }									\
+    (v)->cur[i].data = SPINHASH_REHASH_SENTINEL;			\
+    rws_write_unlock(&(v)->new[p].lock);				\
+    rws_write_unlock(&(v)->cur[i].lock);				\
+  }
+
+
+#define SPINHASH_REHASH_SENTINEL  ((void *) 1)
+
+
+/*
+ * Memory hashing functions
+ */
 
 static inline void
 mem_hash_init(u64 *h)

lib/hash_test.c

@@ -11,6 +11,9 @@
 #include "test/birdtest.h"
 
 #include "lib/hash.h"
+#include "lib/event.h"
+
+#include <pthread.h>
 
 struct test_node {
   struct test_node *next;	/* Hash chain */
@@ -285,43 +288,119 @@ t_walk_filter(void)
   return 1;
 }
 
-static int
+
-t_walk_iter(void)
+/*
+ * Spinlocked hashes
+ */
+
+struct st_node {
+  struct st_node *next;	/* Hash chain */
+  u32 key;
+};
+
+#define ST_KEY(n)		n->key
+#define ST_NEXT(n)		n->next
+#define ST_EQ(n1,n2)		n1 == n2
+#define ST_FN(n)		(n) ^ u32_hash((n))
+#define ST_ORDER		4
+
+#define ST_PARAMS		*1, *8, 3, 2, 3, 9
+
+#define ST_MAX			16384
+#define ST_READERS		1
+
+static uint const st_skip[] = { 3, 7, 13, 17, 23, 37 };
+
+typedef SPINHASH(struct st_node) shtest;
+
+static _Atomic uint st_end = 0;
+static _Atomic uint st_skip_pos = 0;
+
+static void *
+st_rehash_thread(void *_v)
 {
-  init_hash();
+  shtest *v = _v;
-  fill_hash();
+  int step;
 
-  u32 hit = 0;
+  the_bird_lock();
-
+  while (!atomic_load_explicit(&st_end, memory_order_relaxed))
-  u32 prev_hash = ~0;
-  for (uint cnt = 0; cnt < MAX_NUM; )
   {
-    u32 last_hash = ~0;
+    birdloop_yield();
-//    printf("PUT!\n");
+    SPINHASH_REHASH_PREPARE(v, ST, struct st_node, step);
-    HASH_WALK_ITER(hash, TEST, n, hit)
-    {
-      cnt++;
-      u32 cur_hash = HASH_FN(hash, TEST, n->key);
-      /*
-      printf("C%08x L%08x P%08x K%08x H%08x N%p S%d I%ld\n",
-	  cur_hash, last_hash, prev_hash, n->key, hit, n, _shift, n - &nodes[0]);
-	  */
 
-      if (last_hash == ~0U)
+    if (!step)		continue;
-      {
+    if (step < 0)	SPINHASH_REHASH_DOWN(v, ST, struct st_node, -step);
-	if (prev_hash != ~0U)
+    if (step > 0)	SPINHASH_REHASH_UP  (v, ST, struct st_node,  step);
-	  bt_assert(prev_hash < cur_hash);
-	last_hash = prev_hash = cur_hash;
-      }
-      else
-	bt_assert(last_hash == cur_hash);
 
-      if (cnt < MAX_NUM)
+    SPINHASH_REHASH_FINISH(v, ST);
-	HASH_WALK_ITER_PUT;
+  }
-    }
+  the_bird_unlock();
-    HASH_WALK_ITER_END;
+
+  return NULL;
+}
+
+static void *
+st_find_thread(void *_v)
+{
+  shtest *v = _v;
+
+  uint skip = st_skip[atomic_fetch_add_explicit(&st_skip_pos, 1, memory_order_acq_rel)];
+
+  for (u64 i = 0; !atomic_load_explicit(&st_end, memory_order_relaxed); i += skip)
+  {
+    struct st_node *n = SPINHASH_FIND(*v, ST, i % ST_MAX);
+    ASSERT_DIE(!n || (n->key == i % ST_MAX));
   }
 
+  return NULL;
+}
+
+static void *
+st_update_thread(void *_v)
+{
+  shtest *v = _v;
+
+  struct st_node block[ST_MAX];
+  for (uint i = 0; i < ST_MAX; i++)
+    block[i] = (struct st_node) { .key = i, };
+
+  for (uint r = 0; r < 32; r++)
+  {
+    for (uint i = 0; i < ST_MAX; i++)
+      SPINHASH_INSERT(*v, ST, (&block[i]));
+
+    for (uint i = 0; i < ST_MAX; i++)
+      SPINHASH_REMOVE(*v, ST, (&block[i]));
+  }
+
+  atomic_store_explicit(&st_end, 1, memory_order_release);
+
+  return NULL;
+}
+
+int
+t_spinhash_basic(void)
+{
+  pthread_t reader[6], updater, rehasher;
+
+  shtest v = {};
+  void *ST_REHASH = NULL;
+  SPINHASH_INIT(v, ST, rp_new(&root_pool, the_bird_domain.the_bird, "Test pool"), NULL);
+  the_bird_unlock();
+
+  for (int i=0; i<ST_READERS; i++)
+    pthread_create(&reader[i], NULL, st_find_thread, &v);
+
+  pthread_create(&rehasher, NULL, st_rehash_thread, &v);
+  pthread_create(&updater, NULL, st_update_thread, &v);
+
+  pthread_join(updater, NULL);
+  pthread_join(rehasher, NULL);
+
+  for (int i=0; i<ST_READERS; i++)
+    pthread_join(reader[i], NULL);
+
+  the_bird_lock();
   return 1;
 }
 
@@ -339,7 +418,8 @@ main(int argc, char *argv[])
   bt_test_suite(t_walk_delsafe_remove, 	"HASH_WALK_DELSAFE and HASH_REMOVE");
   bt_test_suite(t_walk_delsafe_remove2,	"HASH_WALK_DELSAFE and HASH_REMOVE2. HASH_REMOVE2 is HASH_REMOVE and smart auto-resize function");
   bt_test_suite(t_walk_filter,		"HASH_WALK_FILTER");
-  bt_test_suite(t_walk_iter,		"HASH_WALK_ITER");
+
+  bt_test_suite(t_spinhash_basic,	"SPINHASH insert, remove, find and rehash");
 
   return bt_exit_value();
 }