X-Git-Url: https://git.liburcu.org/?p=urcu.git;a=blobdiff_plain;f=rculfhash.c;h=1391b6bb630102c18138ffe953e745e806808d62;hp=6cc606d9e0dcbea64a7af89327c3874c636caa4c;hb=7fb6f7638628c96521849489b11eaceb473f280b;hpb=bce63dfd0a2306452c9e39f5df01789e77f3f44a

diff --git a/rculfhash.c b/rculfhash.c
index 6cc606d..1391b6b 100644
--- a/rculfhash.c
+++ b/rculfhash.c
@@ -221,6 +221,13 @@ struct cds_lfht {
 	cds_lfht_compare_fct compare_fct;
 	unsigned long hash_seed;
 	int flags;
+	/*
+	 * We need to put the work threads offline (QSBR) when taking this
+	 * mutex, because we use synchronize_rcu within this mutex critical
+	 * section, which waits on read-side critical sections, and could
+	 * therefore cause grace-period deadlock if we hold off RCU G.P.
+	 * completion.
+	 */
 	pthread_mutex_t resize_mutex;	/* resize mutex: add/del mutex */
 	unsigned int in_progress_resize, in_progress_destroy;
 	void (*cds_lfht_call_rcu)(struct rcu_head *head,
@@ -228,6 +235,8 @@ struct cds_lfht {
 	void (*cds_lfht_synchronize_rcu)(void);
 	void (*cds_lfht_rcu_read_lock)(void);
 	void (*cds_lfht_rcu_read_unlock)(void);
+	void (*cds_lfht_rcu_thread_offline)(void);
+	void (*cds_lfht_rcu_thread_online)(void);
 	unsigned long count;		/* global approximate item count */
 	struct ht_items_count *percpu_count;	/* per-cpu item count */
 };
@@ -743,10 +752,8 @@ struct cds_lfht_node *_cds_lfht_add(struct cds_lfht *ht,
 		iter = rcu_dereference(iter_prev->p.next);
 		assert(iter_prev->p.reverse_hash <= node->p.reverse_hash);
 		for (;;) {
-			/* TODO: check if removed */
 			if (unlikely(!clear_flag(iter)))
 				goto insert;
-			/* TODO: check if removed */
 			if (likely(clear_flag(iter)->p.reverse_hash > node->p.reverse_hash))
 				goto insert;
 			next = rcu_dereference(clear_flag(iter)->p.next);
@@ -874,11 +881,17 @@ void init_table_hash(struct cds_lfht *ht, unsigned long i,
 	}
 }
 
+/*
+ * Holding RCU read lock to protect _cds_lfht_add against memory
+ * reclaim that could be performed by other call_rcu worker threads (ABA
+ * problem).
+ */
 static
 void init_table_link(struct cds_lfht *ht, unsigned long i, unsigned long len)
 {
 	unsigned long j;
 
+	ht->cds_lfht_rcu_thread_online();
 	ht->cds_lfht_rcu_read_lock();
 	for (j = 0; j < len; j++) {
 		struct cds_lfht_node *new_node =
@@ -892,13 +905,9 @@ void init_table_link(struct cds_lfht *ht, unsigned long i, unsigned long len)
 			break;
 	}
 	ht->cds_lfht_rcu_read_unlock();
+	ht->cds_lfht_rcu_thread_offline();
 }
 
-/*
- * Holding RCU read lock to protect _cds_lfht_add against memory
- * reclaim that could be performed by other call_rcu worker threads (ABA
- * problem).
- */
 static
 void init_table(struct cds_lfht *ht,
 		unsigned long first_order, unsigned long len_order)
@@ -913,6 +922,11 @@ void init_table(struct cds_lfht *ht,
 
 		len = !i ? 1 : 1UL << (i - 1);
 		dbg_printf("init order %lu len: %lu\n", i, len);
+
+		/* Stop expand if the resize target changes under us */
+		if (CMM_LOAD_SHARED(ht->t.resize_target) < (!i ? 1 : (1UL << i)))
+			break;
+
 		ht->t.tbl[i] = calloc(1, sizeof(struct rcu_level)
 				+ (len * sizeof(struct _cds_lfht_node)));
 
@@ -937,11 +951,32 @@ void init_table(struct cds_lfht *ht,
 	}
 }
 
+/*
+ * Holding RCU read lock to protect _cds_lfht_remove against memory
+ * reclaim that could be performed by other call_rcu worker threads (ABA
+ * problem).
+ * For a single level, we logically remove and garbage collect each node.
+ *
+ * As a design choice, we perform logical removal and garbage collection on a
+ * node-per-node basis to simplify this algorithm. We also assume keeping good
+ * cache locality of the operation would overweight possible performance gain
+ * that could be achieved by batching garbage collection for multiple levels.
+ * However, this would have to be justified by benchmarks.
+ *
+ * Concurrent removal and add operations are helping us perform garbage
+ * collection of logically removed nodes. We guarantee that all logically
+ * removed nodes have been garbage-collected (unlinked) before call_rcu is
+ * invoked to free a hole level of dummy nodes (after a grace period).
+ *
+ * Logical removal and garbage collection can therefore be done in batch or on a
+ * node-per-node basis, as long as the guarantee above holds.
+ */
 static
 void remove_table(struct cds_lfht *ht, unsigned long i, unsigned long len)
 {
 	unsigned long j;
 
+	ht->cds_lfht_rcu_thread_online();
 	ht->cds_lfht_rcu_read_lock();
 	for (j = 0; j < len; j++) {
 		struct cds_lfht_node *fini_node =
@@ -957,13 +992,9 @@ void remove_table(struct cds_lfht *ht, unsigned long i, unsigned long len)
 			break;
 	}
 	ht->cds_lfht_rcu_read_unlock();
+	ht->cds_lfht_rcu_thread_offline();
 }
 
-/*
- * Holding RCU read lock to protect _cds_lfht_remove against memory
- * reclaim that could be performed by other call_rcu worker threads (ABA
- * problem).
- */
 static
 void fini_table(struct cds_lfht *ht,
 		unsigned long first_order, unsigned long len_order)
@@ -974,13 +1005,27 @@ void fini_table(struct cds_lfht *ht,
 		   first_order, first_order + len_order);
 	end_order = first_order + len_order;
 	assert(first_order > 0);
-	assert(ht->t.size == (1UL << (first_order - 1)));
 	for (i = end_order - 1; i >= first_order; i--) {
 		unsigned long len;
 
 		len = !i ? 1 : 1UL << (i - 1);
 		dbg_printf("fini order %lu len: %lu\n", i, len);
 
+		/* Stop shrink if the resize target changes under us */
+		if (CMM_LOAD_SHARED(ht->t.resize_target) > (1UL << (i - 1)))
+			break;
+
+		cmm_smp_wmb();	/* populate data before RCU size */
+		CMM_STORE_SHARED(ht->t.size, 1UL << (i - 1));
+
+		/*
+		 * We need to wait for all add operations to reach Q.S. (and
+		 * thus use the new table for lookups) before we can start
+		 * releasing the old dummy nodes. Otherwise their lookup will
+		 * return a logically removed node as insert position.
+		 */
+		ht->cds_lfht_synchronize_rcu();
+
 		/*
 		 * Set "removed" flag in dummy nodes about to be removed.
 		 * Unlink all now-logically-removed dummy node pointers.
@@ -1006,7 +1051,9 @@ struct cds_lfht *cds_lfht_new(cds_lfht_hash_fct hash_fct,
 					void (*func)(struct rcu_head *head)),
 			void (*cds_lfht_synchronize_rcu)(void),
 			void (*cds_lfht_rcu_read_lock)(void),
-			void (*cds_lfht_rcu_read_unlock)(void))
+			void (*cds_lfht_rcu_read_unlock)(void),
+			void (*cds_lfht_rcu_thread_offline)(void),
+			void (*cds_lfht_rcu_thread_online)(void))
 {
 	struct cds_lfht *ht;
 	unsigned long order;
@@ -1022,14 +1069,19 @@ struct cds_lfht *cds_lfht_new(cds_lfht_hash_fct hash_fct,
 	ht->cds_lfht_synchronize_rcu = cds_lfht_synchronize_rcu;
 	ht->cds_lfht_rcu_read_lock = cds_lfht_rcu_read_lock;
 	ht->cds_lfht_rcu_read_unlock = cds_lfht_rcu_read_unlock;
+	ht->cds_lfht_rcu_thread_offline = cds_lfht_rcu_thread_offline;
+	ht->cds_lfht_rcu_thread_online = cds_lfht_rcu_thread_online;
 	ht->percpu_count = alloc_per_cpu_items_count();
 	/* this mutex should not nest in read-side C.S. */
 	pthread_mutex_init(&ht->resize_mutex, NULL);
 	order = get_count_order_ulong(max(init_size, MIN_TABLE_SIZE)) + 1;
 	ht->flags = flags;
+	ht->cds_lfht_rcu_thread_offline();
 	pthread_mutex_lock(&ht->resize_mutex);
+	ht->t.resize_target = 1UL << (order - 1);
 	init_table(ht, 0, order);
 	pthread_mutex_unlock(&ht->resize_mutex);
+	ht->cds_lfht_rcu_thread_online();
 	return ht;
 }
 
@@ -1255,17 +1307,6 @@ void _do_cds_lfht_shrink(struct cds_lfht *ht,
 	       old_size, old_order, new_size, new_order);
 	assert(new_size < old_size);
 
-	cmm_smp_wmb();	/* populate data before RCU size */
-	CMM_STORE_SHARED(ht->t.size, new_size);
-
-	/*
-	 * We need to wait for all add operations to reach Q.S. (and
-	 * thus use the new table for lookups) before we can start
-	 * releasing the old dummy nodes. Otherwise their lookup will
-	 * return a logically removed node as insert position.
-	 */
-	ht->cds_lfht_synchronize_rcu();
-
 	/* Remove and unlink all dummy nodes to remove. */
 	fini_table(ht, new_order, old_order - new_order);
 }
@@ -1291,7 +1332,7 @@ void _do_cds_lfht_resize(struct cds_lfht *ht)
 		ht->t.resize_initiated = 0;
 		/* write resize_initiated before read resize_target */
 		cmm_smp_mb();
-	} while (new_size != CMM_LOAD_SHARED(ht->t.resize_target));
+	} while (ht->t.size != CMM_LOAD_SHARED(ht->t.resize_target));
 }
 
 static
@@ -1314,9 +1355,11 @@ void cds_lfht_resize(struct cds_lfht *ht, unsigned long new_size)
 {
 	resize_target_update_count(ht, new_size);
 	CMM_STORE_SHARED(ht->t.resize_initiated, 1);
+	ht->cds_lfht_rcu_thread_offline();
 	pthread_mutex_lock(&ht->resize_mutex);
 	_do_cds_lfht_resize(ht);
 	pthread_mutex_unlock(&ht->resize_mutex);
+	ht->cds_lfht_rcu_thread_online();
 }
 
 static
@@ -1326,9 +1369,11 @@ void do_resize_cb(struct rcu_head *head)
 		caa_container_of(head, struct rcu_resize_work, head);
 	struct cds_lfht *ht = work->ht;
 
+	ht->cds_lfht_rcu_thread_offline();
 	pthread_mutex_lock(&ht->resize_mutex);
 	_do_cds_lfht_resize(ht);
 	pthread_mutex_unlock(&ht->resize_mutex);
+	ht->cds_lfht_rcu_thread_online();
 	poison_free(work);
 	cmm_smp_mb();	/* finish resize before decrement */
 	uatomic_dec(&ht->in_progress_resize);