[urcu.git] / urcu-ht.c


/*
 * TODO: keys are currently assumed <= sizeof(void *). Key target never freed.
 */

#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 <compiler.h>
#include <urcu/jhash.h>
#include <urcu-ht.h>

/* node flags */
#define	NODE_STOLEN	(1 << 0)

struct rcu_ht_node;

struct rcu_ht_node {
	struct rcu_ht_node *next;
	void *key;
	void *data;
	unsigned int flags;
};

struct rcu_ht {
	struct rcu_ht_node **tbl;
	ht_hash_fct hash_fct;
	void (*free_fct)(void *data);	/* fct to free data */
	uint32_t keylen;
	uint32_t hashseed;
	struct ht_size {
		unsigned long add;
		unsigned long lookup;
	} size;
};

struct rcu_ht *ht_new(ht_hash_fct hash_fct, void (*free_fct)(void *data),
		      unsigned long init_size, uint32_t keylen,
		      uint32_t hashseed)
{
	struct rcu_ht *ht;

	ht = calloc(1, sizeof(struct rcu_ht));
	ht->hash_fct = hash_fct;
	ht->free_fct = free_fct;
	ht->size.add = init_size;
	ht->size.lookup = init_size;
	ht->keylen = keylen;
	ht->hashseed = hashseed;
	ht->tbl = calloc(init_size, sizeof(struct rcu_ht_node *));
	return ht;
}

void *ht_lookup(struct rcu_ht *ht, void *key)
{
	unsigned long hash;
	struct rcu_ht_node *node;
	void *ret;

	hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.lookup;

	rcu_read_lock();
	node = rcu_dereference(ht->tbl[hash]);
	for (;;) {
		if (likely(!node)) {
			ret = NULL;
			break;
		}
		if (node->key == key) {
			ret = node->data;
			break;
		}
		node = rcu_dereference(node->next);
	}
	rcu_read_unlock();

	return ret;
}

/*
 * Will re-try until either:
 * - The key is already there (-EEXIST)
 * - We successfully add the key at the head of a table bucket.
 */
int ht_add(struct rcu_ht *ht, void *key, void *data)
{
	struct rcu_ht_node *node, *old_head, *new_head;
	unsigned long hash;
	int ret = 0;

	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
	 * and the nodes traversal (to find duplicates). This ensures
	 * the head pointer has not been reclaimed when cmpxchg is done.
	 * Always adding at the head ensures that we would have to
	 * re-try if a new item has been added concurrently. So we ensure that
	 * we never add duplicates. */
retry:
	rcu_read_lock();

	hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.add;

	old_head = node = rcu_dereference(ht->tbl[hash]);
	for (;;) {
		if (likely(!node)) {
			break;
		}
		if (node->key == key) {
			ret = -EEXIST;
			goto end;
		}
		node = rcu_dereference(node->next);
	}
	new_head->next = old_head;
	if (rcu_cmpxchg_pointer(&ht->tbl[hash], old_head, new_head) != old_head)
		goto restart;
end:
	rcu_read_unlock();

	return ret;

	/* restart loop, release and re-take the read lock to be kind to GP */
restart:
	rcu_read_unlock();
	goto retry;
}

/*
 * 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, *del_node = NULL;
	unsigned long hash;
	void *data;

retry:
	rcu_read_lock();

	hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.lookup;

	prev = &ht->tbl[hash];
	node = rcu_dereference(*prev);
	for (;;) {
		if (likely(!node)) {
			if (del_node) {
				goto end;
			} else {
				goto error;
			}
		}
		if (node->key == key) {
			break;
		}
		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)
		del_node = node;
	goto restart;

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 = del_node->data;
	call_rcu(free, del_node);
	return data;

error:
	data = (void *)(unsigned long)-ENOENT;
	rcu_read_unlock();
	return data;

	/* restart loop, release and re-take the read lock to be kind to GP */
restart:
	rcu_read_unlock();
	goto retry;
}

int ht_delete(struct rcu_ht *ht, void *key)
{
	void *data;

	data = ht_steal(ht, key);
	if (data && data != (void *)(unsigned long)-ENOENT) {
		if (ht->free_fct)
			call_rcu(ht->free_fct, data);
		return 0;
	} else {
		return -ENOENT;
	}
}

/* Delete all old elements. Allow concurrent writer accesses. */
int ht_delete_all(struct rcu_ht *ht)
{
	unsigned long i;
	struct rcu_ht_node **prev, *node, *inext;
	int cnt = 0;

	for (i = 0; i < ht->size.lookup; i++) {
		rcu_read_lock();
		prev = &ht->tbl[i];
		/*
		 * Cut the head. After that, we own the first element.
		 */
		node = rcu_xchg_pointer(prev, NULL);
		if (!node) {
			rcu_read_unlock();
			continue;
		}
		/*
		 * We manage a list shared with concurrent writers and readers.
		 * Note that a concurrent add may or may not be deleted by us,
		 * depending if it arrives before or after the head is cut.
		 * "node" points to our first node. Remove first elements
		 * iteratively.
		 */
		for (;;) {
			inext = NULL;
			prev = &node->next;
			if (prev)
				inext = rcu_xchg_pointer(prev, NULL);
			/*
			 * "node" is the first element of the list we have cut.
			 * We therefore own it, no concurrent writer may delete
			 * it. There can only be concurrent lookups. Concurrent
			 * add can only be done on a bucket head, but we've cut
			 * it already. inext is also owned by us, because we
			 * have exchanged it for "NULL". It will therefore be
			 * safe to use it after a G.P.
			 */
			rcu_read_unlock();
			if (node->data)
				call_rcu(ht->free_fct, node->data);
			call_rcu(free, node);
			cnt++;
			if (likely(!inext))
				break;
			rcu_read_lock();
			node = inext;
		}
	}
	return cnt;
}

/*
 * Should only be called when no more concurrent readers nor writers can
 * possibly access the table.
 */
int ht_destroy(struct rcu_ht *ht)
{
	int ret;

	ret = ht_delete_all(ht);
	free(ht->tbl);
	free(ht);
	return ret;
}

/*
 * Expects keys <= than pointer size to be encoded in the pointer itself.
 */
uint32_t ht_jhash(void *key, uint32_t length, uint32_t initval)
{
	uint32_t ret;
	void *vkey;

	if (length <= sizeof(void *))
		vkey = &key;
	else
		vkey = key;
	ret = jhash(vkey, length, initval);
	return ret;
}
Commit	Line	Data
2ed95849	1
5e28c532 MD	2	/*
	3	* TODO: keys are currently assumed <= sizeof(void *). Key target never freed.
	4	*/
	5
2ed95849 MD	6	#define _LGPL_SOURCE
2ed95849 MD	7	#include <stdlib.h>
e0ba718a MD	8	#include <errno.h>
	9	#include <assert.h>
	10	#include <stdio.h>
	11
2ed95849	12	#include <urcu.h>
e0ba718a	13	#include <urcu-defer.h>
2ed95849 MD	14	#include <arch.h>
2ed95849 MD	15	#include <arch_atomic.h>
2ed95849	16	#include <compiler.h>
674f7a69	17	#include <urcu/jhash.h>
e0ba718a	18	#include <urcu-ht.h>
2ed95849	19
3df2df75 MD	20	/* node flags */
	21	#define NODE_STOLEN (1 << 0)
	22
2ed95849 MD	23	struct rcu_ht_node;
	24
	25	struct rcu_ht_node {
	26	struct rcu_ht_node *next;
	27	void *key;
	28	void *data;
3df2df75	29	unsigned int flags;
2ed95849 MD	30	};
	31
	32	struct rcu_ht {
674f7a69	33	struct rcu_ht_node **tbl;
2ed95849 MD	34	ht_hash_fct hash_fct;
2ed95849 MD	35	void (free_fct)(void data); /* fct to free data */
5e28c532 MD	36	uint32_t keylen;
5e28c532 MD	37	uint32_t hashseed;
674f7a69 MD	38	struct ht_size {
	39	unsigned long add;
	40	unsigned long lookup;
	41	} size;
2ed95849 MD	42	};
2ed95849 MD	43
674f7a69	44	struct rcu_ht ht_new(ht_hash_fct hash_fct, void (free_fct)(void *data),
5e28c532 MD	45	unsigned long init_size, uint32_t keylen,
5e28c532 MD	46	uint32_t hashseed)
2ed95849 MD	47	{
	48	struct rcu_ht *ht;
	49
	50	ht = calloc(1, sizeof(struct rcu_ht));
	51	ht->hash_fct = hash_fct;
	52	ht->free_fct = free_fct;
674f7a69 MD	53	ht->size.add = init_size;
674f7a69 MD	54	ht->size.lookup = init_size;
5e28c532 MD	55	ht->keylen = keylen;
5e28c532 MD	56	ht->hashseed = hashseed;
674f7a69	57	ht->tbl = calloc(init_size, sizeof(struct rcu_ht_node *));
ab7d5fc6	58	return ht;
2ed95849 MD	59	}
2ed95849 MD	60
2ed95849 MD	61	void ht_lookup(struct rcu_ht ht, void *key)
	62	{
	63	unsigned long hash;
	64	struct rcu_ht_node *node;
	65	void *ret;
	66
5e28c532	67	hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.lookup;
2ed95849 MD	68
2ed95849 MD	69	rcu_read_lock();
5e28c532	70	node = rcu_dereference(ht->tbl[hash]);
2ed95849 MD	71	for (;;) {
	72	if (likely(!node)) {
	73	ret = NULL;
	74	break;
	75	}
	76	if (node->key == key) {
	77	ret = node->data;
	78	break;
	79	}
	80	node = rcu_dereference(node->next);
	81	}
	82	rcu_read_unlock();
ab7d5fc6 MD	83
ab7d5fc6 MD	84	return ret;
2ed95849 MD	85	}
	86
	87	/*
	88	* Will re-try until either:
	89	* - The key is already there (-EEXIST)
	90	* - We successfully add the key at the head of a table bucket.
	91	*/
	92	int ht_add(struct rcu_ht ht, void key, void *data)
	93	{
	94	struct rcu_ht_node node, old_head, *new_head;
	95	unsigned long hash;
	96	int ret = 0;
	97
	98	new_head = calloc(1, sizeof(struct rcu_ht_node));
	99	new_head->key = key;
	100	new_head->data = data;
3df2df75	101	new_head->flags = 0;
2ed95849 MD	102	/* here comes the fun and tricky part.
	103	* Add at the beginning with a cmpxchg.
	104	* Hold a read lock between the moment the first element is read
	105	* and the nodes traversal (to find duplicates). This ensures
	106	* the head pointer has not been reclaimed when cmpxchg is done.
	107	* Always adding at the head ensures that we would have to
	108	* re-try if a new item has been added concurrently. So we ensure that
	109	* we never add duplicates. */
	110	retry:
	111	rcu_read_lock();
	112
5e28c532 MD	113	hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.add;
	114
	115	old_head = node = rcu_dereference(ht->tbl[hash]);
2ed95849 MD	116	for (;;) {
	117	if (likely(!node)) {
	118	break;
	119	}
	120	if (node->key == key) {
	121	ret = -EEXIST;
	122	goto end;
	123	}
	124	node = rcu_dereference(node->next);
	125	}
5e28c532 MD	126	new_head->next = old_head;
5e28c532 MD	127	if (rcu_cmpxchg_pointer(&ht->tbl[hash], old_head, new_head) != old_head)
2ed95849 MD	128	goto restart;
	129	end:
	130	rcu_read_unlock();
	131
	132	return ret;
	133
	134	/* restart loop, release and re-take the read lock to be kind to GP */
	135	restart:
	136	rcu_read_unlock();
	137	goto retry;
	138	}
	139
	140	/*
	141	* Restart until we successfully remove the entry, or no entry is left
	142	* ((void *)(unsigned long)-ENOENT).
e0ba718a MD	143	* Deal with concurrent stealers by doing an extra verification pass to check
	144	* that no element in the list are still pointing to the element stolen.
	145	* This could happen if two concurrent steal for consecutive objects are
	146	* executed. A pointer to an object being stolen could be saved by the
	147	* concurrent stealer for the previous object.
	148	* Also, given that in this precise scenario, another stealer can also want to
	149	* delete the doubly-referenced object; use a "stolen" flag to let only one
	150	* stealer delete the object.
2ed95849	151	*/
ab7d5fc6	152	void ht_steal(struct rcu_ht ht, void *key)
2ed95849	153	{
2e6070d0	154	struct rcu_ht_node *prev, node, *del_node = NULL;
2ed95849	155	unsigned long hash;
ab7d5fc6	156	void *data;
2ed95849	157
2ed95849 MD	158	retry:
	159	rcu_read_lock();
	160
5e28c532 MD	161	hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.lookup;
	162
	163	prev = &ht->tbl[hash];
2ed95849 MD	164	node = rcu_dereference(*prev);
	165	for (;;) {
	166	if (likely(!node)) {
2e6070d0 MD	167	if (del_node) {
	168	goto end;
	169	} else {
2e6070d0 MD	170	goto error;
2e6070d0 MD	171	}
2ed95849 MD	172	}
	173	if (node->key == key) {
	174	break;
	175	}
	176	prev = &node->next;
	177	node = rcu_dereference(*prev);
	178	}
e0ba718a	179
85f6ba5f MD	180	if (!del_node) {
	181	/*
	182	* Another concurrent thread stole it ? If so, let it deal with
3df2df75 MD	183	* this. Assume NODE_STOLEN is the only flag. If this changes,
3df2df75 MD	184	* read flags before cmpxchg.
85f6ba5f	185	*/
3df2df75	186	if (cmpxchg(&node->flags, 0, NODE_STOLEN) != 0)
85f6ba5f MD	187	goto error;
85f6ba5f MD	188	}
e0ba718a	189
2ed95849	190	/* Found it ! pointer to object is in "prev" */
7ce4eb38	191	if (rcu_cmpxchg_pointer(prev, node, node->next) == node)
2e6070d0 MD	192	del_node = node;
2e6070d0 MD	193	goto restart;
5e28c532	194
2e6070d0 MD	195	end:
	196	/*
	197	* From that point, we own node. Note that there can still be concurrent
	198	* RCU readers using it. We can free it outside of read lock after a GP.
	199	*/
2ed95849 MD	200	rcu_read_unlock();
2ed95849 MD	201
567d7550 MD	202	data = del_node->data;
567d7550 MD	203	call_rcu(free, del_node);
5e28c532	204	return data;
ab7d5fc6	205
5e28c532	206	error:
e0ba718a	207	data = (void *)(unsigned long)-ENOENT;
5e28c532	208	rcu_read_unlock();
ab7d5fc6	209	return data;
2ed95849 MD	210
	211	/* restart loop, release and re-take the read lock to be kind to GP */
	212	restart:
	213	rcu_read_unlock();
	214	goto retry;
	215	}
	216
	217	int ht_delete(struct rcu_ht ht, void key)
	218	{
ab7d5fc6	219	void *data;
2ed95849	220
ab7d5fc6	221	data = ht_steal(ht, key);
5e28c532 MD	222	if (data && data != (void *)(unsigned long)-ENOENT) {
5e28c532 MD	223	if (ht->free_fct)
ab7d5fc6	224	call_rcu(ht->free_fct, data);
2ed95849 MD	225	return 0;
	226	} else {
	227	return -ENOENT;
	228	}
	229	}
ab7d5fc6	230
674f7a69	231	/* Delete all old elements. Allow concurrent writer accesses. */
5e28c532	232	int ht_delete_all(struct rcu_ht *ht)
674f7a69 MD	233	{
	234	unsigned long i;
	235	struct rcu_ht_node *prev, node, *inext;
5e28c532	236	int cnt = 0;
674f7a69 MD	237
	238	for (i = 0; i < ht->size.lookup; i++) {
	239	rcu_read_lock();
5e28c532 MD	240	prev = &ht->tbl[i];
	241	/*
	242	* Cut the head. After that, we own the first element.
	243	*/
	244	node = rcu_xchg_pointer(prev, NULL);
	245	if (!node) {
674f7a69 MD	246	rcu_read_unlock();
	247	continue;
	248	}
	249	/*
	250	* We manage a list shared with concurrent writers and readers.
	251	* Note that a concurrent add may or may not be deleted by us,
	252	* depending if it arrives before or after the head is cut.
	253	* "node" points to our first node. Remove first elements
	254	* iteratively.
	255	*/
	256	for (;;) {
5e28c532	257	inext = NULL;
674f7a69	258	prev = &node->next;
5e28c532 MD	259	if (prev)
5e28c532 MD	260	inext = rcu_xchg_pointer(prev, NULL);
674f7a69 MD	261	/*
	262	* "node" is the first element of the list we have cut.
	263	* We therefore own it, no concurrent writer may delete
	264	* it. There can only be concurrent lookups. Concurrent
	265	* add can only be done on a bucket head, but we've cut
	266	* it already. inext is also owned by us, because we
	267	* have exchanged it for "NULL". It will therefore be
	268	* safe to use it after a G.P.
	269	*/
	270	rcu_read_unlock();
	271	if (node->data)
	272	call_rcu(ht->free_fct, node->data);
	273	call_rcu(free, node);
5e28c532	274	cnt++;
674f7a69 MD	275	if (likely(!inext))
	276	break;
	277	rcu_read_lock();
	278	node = inext;
	279	}
	280	}
5e28c532	281	return cnt;
674f7a69 MD	282	}
	283
	284	/*
	285	* Should only be called when no more concurrent readers nor writers can
	286	* possibly access the table.
	287	*/
5e28c532	288	int ht_destroy(struct rcu_ht *ht)
674f7a69	289	{
5e28c532 MD	290	int ret;
	291
	292	ret = ht_delete_all(ht);
674f7a69 MD	293	free(ht->tbl);
674f7a69 MD	294	free(ht);
5e28c532	295	return ret;
674f7a69 MD	296	}
674f7a69 MD	297
5e28c532 MD	298	/*
	299	* Expects keys <= than pointer size to be encoded in the pointer itself.
	300	*/
674f7a69	301	uint32_t ht_jhash(void *key, uint32_t length, uint32_t initval)
ab7d5fc6	302	{
5e28c532 MD	303	uint32_t ret;
	304	void *vkey;
	305
0eb566f4	306	if (length <= sizeof(void *))
5e28c532 MD	307	vkey = &key;
	308	else
	309	vkey = key;
	310	ret = jhash(vkey, length, initval);
	311	return ret;
ab7d5fc6	312	}