X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=urcu-ht.c;h=7213c39ce1414d788149dc864f9114bcb814acb0;hb=3df2df75972d2ccaa1a1fa01bad06b0eda7f76f2;hp=0bbdb5d4e8284d28e8de006cb0ea5476eb9dc7f4;hpb=0eb566f42f9694c397afd96a469e1d89af8a4ff5;p=urcu.git

diff --git a/urcu-ht.c b/urcu-ht.c
index 0bbdb5d..7213c39 100644
--- a/urcu-ht.c
+++ b/urcu-ht.c
@@ -5,16 +5,20 @@
 
 #define _LGPL_SOURCE
 #include <stdlib.h>
+#include <errno.h>
+#include <assert.h>
+#include <stdio.h>
+
 #include <urcu.h>
+#include <urcu-defer.h>
 #include <arch.h>
 #include <arch_atomic.h>
-#include <assert.h>
 #include <compiler.h>
-#include <urcu-defer.h>
-#include <errno.h>
-#include <urcu-ht.h>
 #include <urcu/jhash.h>
-#include <stdio.h>
+#include <urcu-ht.h>
+
+/* node flags */
+#define	NODE_STOLEN	(1 << 0)
 
 struct rcu_ht_node;
 
@@ -22,6 +26,7 @@ struct rcu_ht_node {
 	struct rcu_ht_node *next;
 	void *key;
 	void *data;
+	unsigned int flags;
 };
 
 struct rcu_ht {
@@ -93,6 +98,7 @@ int ht_add(struct rcu_ht *ht, void *key, void *data)
 	new_head = calloc(1, sizeof(struct rcu_ht_node));
 	new_head->key = key;
 	new_head->data = data;
+	new_head->flags = 0;
 	/* here comes the fun and tricky part.
 	 * Add at the beginning with a cmpxchg.
 	 * Hold a read lock between the moment the first element is read
@@ -134,10 +140,18 @@ restart:
 /*
  * Restart until we successfully remove the entry, or no entry is left
  * ((void *)(unsigned long)-ENOENT).
+ * Deal with concurrent stealers by doing an extra verification pass to check
+ * that no element in the list are still pointing to the element stolen.
+ * This could happen if two concurrent steal for consecutive objects are
+ * executed. A pointer to an object being stolen could be saved by the
+ * concurrent stealer for the previous object.
+ * Also, given that in this precise scenario, another stealer can also want to
+ * delete the doubly-referenced object; use a "stolen" flag to let only one
+ * stealer delete the object.
  */
 void *ht_steal(struct rcu_ht *ht, void *key)
 {
-	struct rcu_ht_node **prev, *node;
+	struct rcu_ht_node **prev, *node, *del_node = NULL;
 	unsigned long hash;
 	void *data;
 
@@ -150,8 +164,11 @@ retry:
 	node = rcu_dereference(*prev);
 	for (;;) {
 		if (likely(!node)) {
-			data = (void *)(unsigned long)-ENOENT;
-			goto error;
+			if (del_node) {
+				goto end;
+			} else {
+				goto error;
+			}
 		}
 		if (node->key == key) {
 			break;
@@ -159,18 +176,35 @@ retry:
 		prev = &node->next;
 		node = rcu_dereference(*prev);
 	}
+
+	if (!del_node) {
+		/*
+		 * Another concurrent thread stole it ? If so, let it deal with
+		 * this. Assume NODE_STOLEN is the only flag. If this changes,
+		 * read flags before cmpxchg.
+		 */
+		if (cmpxchg(&node->flags, 0, NODE_STOLEN) != 0)
+			goto error;
+	}
+
 	/* Found it ! pointer to object is in "prev" */
-	if (rcu_cmpxchg_pointer(prev, node, node->next) != node)
-		goto restart;
+	if (rcu_cmpxchg_pointer(prev, node, node->next) == node)
+		del_node = node;
+	goto restart;
 
-	/* From that point, we own node. We can free it outside of read lock */
+end:
+	/*
+	 * From that point, we own node. Note that there can still be concurrent
+	 * RCU readers using it. We can free it outside of read lock after a GP.
+	 */
 	rcu_read_unlock();
 
-	data = node->data;
-	call_rcu(free, node);
+	data = del_node->data;
+	call_rcu(free, del_node);
 	return data;
 
 error:
+	data = (void *)(unsigned long)-ENOENT;
 	rcu_read_unlock();
 	return data;