Skip to content

Commit 980a953

Browse files
jservjserv
committed
Add a concurrent map
1 parent 388d590 commit 980a953

16 files changed

+1282
-1
lines changed

‎.clang-format

+3
Original file line numberDiff line numberDiff line change
@@ -26,3 +26,6 @@ ForEachMacros:
2626
- rb_list_foreach_safe
2727
- EV_FOREACH
2828
- LIST_FOREACH
29+
- LIST_FOREACH_POP
30+
- MAP_FOREACH
31+
- MAP_FOREACH_WITH_HASH

‎README.md

+2-1
Original file line numberDiff line numberDiff line change
@@ -31,12 +31,13 @@ purpose of these programs is to be illustrative and educational.
3131
* [Synchronization](https://en.wikipedia.org/wiki/Synchronization_(computer_science))
3232
- [mcslock](mcslock/): An MCS lock implementation.
3333
- [seqlock](seqlock/): A seqlock (sequence lock) implementation.
34-
- [hp\_list](hp_list)/: A concurrent linked list utilizing Hazard Pointers.
34+
- [hp\_list](hp_list/)/: A concurrent linked list utilizing Hazard Pointers.
3535
- [rcu-list](rcu-list/): A concurrent linked list utilizing the simplified RCU algorithm.
3636
- [qsbr](qsbr/): An implementation of Quiescent state based reclamation (QSBR).
3737
- [list-move](list-move/): Evaluation of two concurrent linked lists: QSBR and lock-based.
3838
- [rcu\_queue](rcu_queue/): An efficient concurrent queue based on QSBR.
3939
- [thread-rcu](thread-rcu/): A Linux Kernel style thread-based simple RCU.
40+
- [cmap](cmap/): A concurrent map implementation based on RCU.
4041
* Applications
4142
- [httpd](httpd/): A multi-threaded web server.
4243
- [map-reduce](map-reduce/): word counting using MapReduce.

‎cmap/Makefile

+7
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
all:
2+
gcc -Wall -O2 -o test-cmap \
3+
cmap.c random.c rcu.c test-cmap.c \
4+
-lpthread
5+
6+
clean:
7+
rm -f test-cmap

‎cmap/cmap.c

+262
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,262 @@
1+
#include "cmap.h"
2+
#include <stdbool.h>
3+
#include <stddef.h>
4+
#include <stdlib.h>
5+
#include "locks.h"
6+
#include "util.h"
7+
8+
#define MAP_INITIAL_SIZE 512
9+
10+
struct cmap_entry {
11+
struct cmap_node *first;
12+
};
13+
14+
struct cmap_impl {
15+
struct cmap_entry *arr; /* Map entreis */
16+
size_t count; /* Number of elements in this */
17+
size_t max; /* Capacity of this */
18+
size_t utilization; /* Number of utialized entries */
19+
struct cond fence; /* Prevent new reads while old still exist */
20+
};
21+
22+
struct cmap_impl_pair {
23+
struct cmap_impl *old, *new;
24+
};
25+
26+
static void cmap_expand(struct cmap *cmap);
27+
static void cmap_expand_callback(void *args);
28+
static void cmap_destroy_callback(void *args);
29+
static size_t cmap_count__(const struct cmap *cmap);
30+
static void cmap_insert__(struct cmap_impl *, struct cmap_node *);
31+
32+
/* Only a single concurrent writer to cmap is allowed */
33+
static void cmap_insert__(struct cmap_impl *impl, struct cmap_node *node)
34+
{
35+
size_t i = node->hash & impl->max;
36+
node->next = impl->arr[i].first;
37+
if (!impl->arr[i].first)
38+
impl->utilization++;
39+
impl->arr[i].first = node;
40+
}
41+
42+
static void cmap_destroy_callback(void *args)
43+
{
44+
struct cmap_impl *impl = (struct cmap_impl *) args;
45+
cond_destroy(&impl->fence);
46+
free(impl);
47+
}
48+
49+
static struct cmap_impl *cmap_impl_init(size_t entry_num)
50+
{
51+
size_t size =
52+
sizeof(struct cmap_impl) + sizeof(struct cmap_entry) * entry_num;
53+
struct cmap_impl *impl = xmalloc(size);
54+
impl->max = entry_num - 1;
55+
impl->count = 0;
56+
impl->utilization = 0;
57+
impl->arr = OBJECT_END(struct cmap_entry *, impl);
58+
cond_init(&impl->fence);
59+
60+
for (int i = 0; i < entry_num; ++i)
61+
impl->arr[i].first = NULL;
62+
return impl;
63+
}
64+
65+
static void cmap_expand_callback(void *args)
66+
{
67+
struct cmap_impl_pair *pair = (struct cmap_impl_pair *) args;
68+
struct cmap_node *c, *n;
69+
70+
/* Rehash */
71+
for (int i = 0; i <= pair->old->max; i++) {
72+
for (c = pair->old->arr[i].first; c; c = n) {
73+
n = c->next;
74+
cmap_insert__(pair->new, c);
75+
}
76+
}
77+
78+
/* Remove fence */
79+
cond_unlock(&pair->new->fence);
80+
free(pair->old);
81+
free(pair);
82+
}
83+
84+
/* Only a single concurrent writer to cmap is allowed */
85+
static void cmap_expand(struct cmap *cmap)
86+
{
87+
struct rcu *impl_rcu = rcu_acquire(cmap->impl->p);
88+
struct cmap_impl_pair *pair = xmalloc(sizeof(*pair));
89+
pair->old = rcu_get(impl_rcu, struct cmap_impl *);
90+
91+
/* Do not allow two expansions in parallel */
92+
/* Prevent new reads while old still exist */
93+
while (cond_is_locked(&pair->old->fence)) {
94+
rcu_release(impl_rcu);
95+
cond_wait(&pair->old->fence);
96+
impl_rcu = rcu_acquire(cmap->impl->p);
97+
pair->old = rcu_get(impl_rcu, struct cmap_impl *);
98+
}
99+
100+
/* Initiate new rehash array */
101+
pair->new = cmap_impl_init((pair->old->max + 1) * 2);
102+
pair->new->count = pair->old->count;
103+
104+
/* Prevent new reads/updates while old reads still exist */
105+
cond_lock(&pair->new->fence);
106+
107+
rcu_postpone(impl_rcu, cmap_expand_callback, pair);
108+
rcu_release(impl_rcu);
109+
rcu_set(cmap->impl->p, pair->new);
110+
}
111+
112+
void cmap_init(struct cmap *cmap)
113+
{
114+
struct cmap_impl *impl = cmap_impl_init(MAP_INITIAL_SIZE);
115+
cmap->impl = xmalloc(sizeof(*cmap->impl));
116+
rcu_init(cmap->impl->p, impl);
117+
}
118+
119+
void cmap_destroy(struct cmap *cmap)
120+
{
121+
if (!cmap)
122+
return;
123+
124+
struct rcu *impl_rcu = rcu_acquire(cmap->impl->p);
125+
struct cmap_impl *impl = rcu_get(impl_rcu, struct cmap_impl *);
126+
rcu_postpone(impl_rcu, cmap_destroy_callback, impl);
127+
rcu_release(impl_rcu);
128+
rcu_destroy(impl_rcu);
129+
free(cmap->impl);
130+
}
131+
132+
static size_t cmap_count__(const struct cmap *cmap)
133+
{
134+
struct rcu *impl_rcu = rcu_acquire(cmap->impl->p);
135+
struct cmap_impl *impl = rcu_get(impl_rcu, struct cmap_impl *);
136+
size_t count = impl->count;
137+
rcu_release(impl_rcu);
138+
return count;
139+
}
140+
141+
double cmap_utilization(const struct cmap *cmap)
142+
{
143+
struct rcu *impl_rcu = rcu_acquire(cmap->impl->p);
144+
struct cmap_impl *impl = rcu_get(impl_rcu, struct cmap_impl *);
145+
double res = (double) impl->utilization / (impl->max + 1);
146+
rcu_release(impl_rcu);
147+
return res;
148+
}
149+
150+
size_t cmap_size(const struct cmap *cmap)
151+
{
152+
return cmap_count__(cmap);
153+
}
154+
155+
/* Only one concurrent writer */
156+
size_t cmap_insert(struct cmap *cmap, struct cmap_node *node, uint32_t hash)
157+
{
158+
node->hash = hash;
159+
160+
struct rcu *impl_rcu = rcu_acquire(cmap->impl->p);
161+
struct cmap_impl *impl = rcu_get(impl_rcu, struct cmap_impl *);
162+
cmap_insert__(impl, node);
163+
impl->count++;
164+
size_t count = impl->count;
165+
bool expand = impl->count > impl->max * 2;
166+
rcu_release(impl_rcu);
167+
168+
if (expand)
169+
cmap_expand(cmap);
170+
return count;
171+
}
172+
173+
/* Only one concurrent writer */
174+
size_t cmap_remove(struct cmap *cmap, struct cmap_node *node)
175+
{
176+
struct rcu *impl_rcu = rcu_acquire(cmap->impl->p);
177+
struct cmap_impl *impl = rcu_get(impl_rcu, struct cmap_impl *);
178+
size_t pos = node->hash & impl->max;
179+
struct cmap_entry *cmap_entry = &impl->arr[pos];
180+
size_t count = impl->count;
181+
182+
struct cmap_node **node_p = &cmap_entry->first;
183+
while (*node_p) {
184+
if (*node_p == node) {
185+
*node_p = node->next;
186+
count--;
187+
break;
188+
}
189+
node_p = &(*node_p)->next;
190+
}
191+
impl->count = count;
192+
rcu_release(impl_rcu);
193+
return count;
194+
}
195+
196+
struct cmap_state cmap_state_acquire(struct cmap *cmap)
197+
{
198+
struct cmap_state state = {.p = rcu_acquire(cmap->impl->p)};
199+
return state;
200+
}
201+
202+
void cmap_state_release(struct cmap_state state)
203+
{
204+
rcu_release(state.p);
205+
}
206+
207+
struct cmap_cursor cmap_find__(struct cmap_state state, uint32_t hash)
208+
{
209+
struct cmap_impl *impl = rcu_get(state.p, struct cmap_impl *);
210+
211+
/* Prevent new reads while old still exist */
212+
while (cond_is_locked(&impl->fence))
213+
cond_wait(&impl->fence);
214+
215+
struct cmap_cursor cursor = {
216+
.entry_idx = hash & impl->max,
217+
.node = impl->arr[hash & impl->max].first,
218+
.next = NULL,
219+
.accross_entries = false,
220+
};
221+
if (cursor.node)
222+
cursor.next = cursor.node->next;
223+
return cursor;
224+
}
225+
226+
struct cmap_cursor cmap_start__(struct cmap_state state)
227+
{
228+
struct cmap_cursor cursor = cmap_find__(state, 0);
229+
cursor.accross_entries = true;
230+
/* Don't start with an empty node */
231+
if (!cursor.node)
232+
cmap_next__(state, &cursor);
233+
return cursor;
234+
}
235+
236+
void cmap_next__(struct cmap_state state, struct cmap_cursor *cursor)
237+
{
238+
struct cmap_impl *impl = rcu_get(state.p, struct cmap_impl *);
239+
240+
cursor->node = cursor->next;
241+
if (cursor->node) {
242+
cursor->next = cursor->node->next;
243+
return;
244+
}
245+
246+
/* We got to the end of the current entry. Try to find
247+
* a valid node in next entries
248+
*/
249+
while (cursor->accross_entries) {
250+
cursor->entry_idx++;
251+
if (cursor->entry_idx > impl->max)
252+
break;
253+
cursor->node = impl->arr[cursor->entry_idx].first;
254+
if (cursor->node) {
255+
cursor->next = cursor->node->next;
256+
return;
257+
}
258+
}
259+
260+
cursor->node = NULL;
261+
cursor->next = NULL;
262+
}

‎cmap/cmap.h

+83
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
#pragma once
2+
3+
#include <stdbool.h>
4+
#include <stddef.h>
5+
#include <stdint.h>
6+
7+
#include "rcu.h"
8+
#include "util.h"
9+
10+
/* Concurrent cmap.
11+
* It supports multiple concurrent readers and a single concurrent writer.
12+
* To iterate, the user needs to acquire a "cmap state" (snapshot).
13+
*/
14+
struct cmap_node {
15+
struct cmap_node *next; /* Next node with same hash. */
16+
uint32_t hash;
17+
};
18+
19+
/* Used for going over all cmap nodes */
20+
struct cmap_cursor {
21+
struct cmap_node *node; /* Pointer to cmap_node */
22+
struct cmap_node *next; /* Pointer to cmap_node */
23+
size_t entry_idx; /* Current entry */
24+
bool accross_entries; /* Hold cursor accross cmap entries */
25+
};
26+
27+
/* Map state (snapshot), must be acquired before cmap iteration, and released
28+
* afterwards.
29+
*/
30+
struct cmap_state {
31+
struct rcu *p;
32+
};
33+
34+
/* Concurrent hash cmap. */
35+
struct cmap {
36+
struct cmap_state *impl;
37+
};
38+
39+
/* Initialization. */
40+
void cmap_init(struct cmap *);
41+
void cmap_destroy(struct cmap *);
42+
43+
/* Counters. */
44+
size_t cmap_size(const struct cmap *);
45+
double cmap_utilization(const struct cmap *cmap);
46+
47+
/* Insertion and deletion. Return the current count after the operation. */
48+
size_t cmap_insert(struct cmap *, struct cmap_node *, uint32_t hash);
49+
size_t cmap_remove(struct cmap *, struct cmap_node *);
50+
51+
/* Acquire/release cmap concurrent state. Use with iteration macros.
52+
* Each acquired state must be released. */
53+
struct cmap_state cmap_state_acquire(struct cmap *cmap);
54+
void cmap_state_release(struct cmap_state state);
55+
56+
/* Iteration macros. Usage example:
57+
*
58+
* struct {
59+
* struct cmap_node node;
60+
* int value;
61+
* } *data;
62+
* struct cmap_state *cmap_state = cmap_state_acquire(&cmap);
63+
* MAP_FOREACH(data, node, cmap_state) {
64+
* ...
65+
* }
66+
* cmap_state_release(cmap_state);
67+
*/
68+
#define MAP_FOREACH(NODE, MEMBER, STATE) \
69+
MAP_FOREACH__(NODE, MEMBER, MAP, cmap_start__(STATE), STATE)
70+
71+
#define MAP_FOREACH_WITH_HASH(NODE, MEMBER, HASH, STATE) \
72+
MAP_FOREACH__(NODE, MEMBER, MAP, cmap_find__(STATE, HASH), STATE)
73+
74+
/* Ieration, private methods. Use iteration macros instead */
75+
struct cmap_cursor cmap_start__(struct cmap_state state);
76+
struct cmap_cursor cmap_find__(struct cmap_state state, uint32_t hash);
77+
void cmap_next__(struct cmap_state state, struct cmap_cursor *cursor);
78+
79+
#define MAP_FOREACH__(NODE, MEMBER, MAP, START, STATE) \
80+
for (struct cmap_cursor cursor_ = START; \
81+
(cursor_.node ? (INIT_CONTAINER(NODE, cursor_.node, MEMBER), true) \
82+
: false); \
83+
cmap_next__(STATE, &cursor_))

0 commit comments

Comments
 (0)