haiwan
/
ck
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ck_ring: Replace MPMC variant with SPMC variant for starters.

Browse Source 
Immediate use-case is SPMC. Next is MPSC, then I will generalize
to MPMC (unfortunately, I don't have an algorithm that avoids CAS2
yet for MPMC).
ck_pring
Samy Al Bahra 13 years ago
parent bb48f602ff
commit d705e02b76
3 changed files with 111 additions and 119 deletions
Show Stats Download Patch File Download Diff File

140
include/ck_ring.h
Unescape View File

@ -31,9 +31,7 @@
#include <ck_md.h>
#include <ck_pr.h>
#include <stdbool.h>
#ifndef CK_F_RING
#define CK_F_RING
#include <string.h>
/*
* Concurrent ring buffer.
@ -159,77 +157,17 @@ ck_ring_capacity(struct ck_ring *ring)
return ring->size;
}
/* XXX: MPMC variant is incorrect, replacement in works. */
CK_CC_INLINE static bool
ck_ring_enqueue_mpmc(struct ck_ring *ring, void *entry)
{
unsigned int consumer, producer, delta;
bool success;
void *r;
producer = ck_pr_load_uint(&ring->p_tail);
do {
consumer = ck_pr_load_uint(&ring->c_head);
delta = (producer + 1) & ring->mask;
if (delta == consumer)
return false;
/* Speculate slot availability. */
r = ck_pr_load_ptr(&ring->ring[producer]);
success = ck_pr_cas_ptr(&ring->ring[producer], r, entry);
/* Publish value before publishing counter update. */
ck_pr_fence_store();
/* This is the linearization point. */
ck_pr_cas_uint_value(&ring->p_tail,
producer,
delta,
&producer);
} while (success == false);
return true;
}
CK_CC_INLINE static bool
ck_ring_dequeue_mpmc(struct ck_ring *ring, void *data)
{
unsigned int consumer, producer;
void *r;
consumer = ck_pr_load_uint(&ring->c_head);
do {
producer = ck_pr_load_uint(&ring->p_tail);
if (consumer == producer)
return false;
ck_pr_fence_load();
r = ck_pr_load_ptr(&ring->ring[consumer]);
/* Serialize load with respect to head update. */
ck_pr_fence_memory();
} while (ck_pr_cas_uint_value(&ring->c_head,
consumer,
(consumer + 1) & ring->mask,
&consumer) == false);
ck_pr_store_ptr(data, r);
return true;
}
CK_CC_INLINE static bool
ck_ring_enqueue_spsc(struct ck_ring *ring, void *entry)
{
unsigned int consumer, producer;
unsigned int consumer, producer, delta;
consumer = ck_pr_load_uint(&ring->c_head);
producer = ring->p_tail;
delta = (producer + 1) & ring->mask;
if (((producer + 1) & ring->mask) == consumer)
return (false);
if (delta == consumer)
return false;
ring->ring[producer] = entry;
@ -238,8 +176,8 @@ ck_ring_enqueue_spsc(struct ck_ring *ring, void *entry)
* that the slot is available for consumption.
*/
ck_pr_fence_store();
ck_pr_store_uint(&ring->p_tail, (producer + 1) & ring->mask);
return (true);
ck_pr_store_uint(&ring->p_tail, delta);
return true;
}
/*
@ -275,17 +213,69 @@ ck_ring_dequeue_spsc(struct ck_ring *ring, void *data)
return (true);
}
CK_CC_INLINE static bool
ck_ring_enqueue_spmc(struct ck_ring *ring, void *entry)
{
unsigned int consumer, producer, delta;
consumer = ck_pr_load_uint(&ring->c_head) & ring->mask;
producer = ring->p_tail;
delta = (producer + 1) & ring->mask;
if (delta == consumer)
return false;
ring->ring[producer] = entry;
/*
* Make sure to update slot value before indicating
* that the slot is available for consumption.
*/
ck_pr_fence_store();
ck_pr_store_uint(&ring->p_tail, delta);
return true;
}
CK_CC_INLINE static bool
ck_ring_dequeue_spmc(struct ck_ring *ring, void *data)
{
unsigned int consumer, producer, position;
void *r;
consumer = ck_pr_load_uint(&ring->c_head);
do {
position = consumer & ring->mask;
producer = ck_pr_load_uint(&ring->p_tail);
if (position == producer)
return false;
ck_pr_fence_load();
r = ck_pr_load_ptr(&ring->ring[position]);
/* Serialize load with respect to head update. */
ck_pr_fence_memory();
} while (ck_pr_cas_uint_value(&ring->c_head,
consumer,
consumer + 1,
&consumer) == false);
ck_pr_store_ptr(data, r);
return true;
}
CK_CC_INLINE static void
ck_ring_init(struct ck_ring *ring, void *buffer, unsigned int size)
{
ck_pr_store_uint(&ring->size, size);
ck_pr_store_uint(&ring->mask, size - 1);
ck_pr_store_uint(&ring->p_tail, 0);
ck_pr_store_uint(&ring->c_head, 0);
ck_pr_store_ptr(&ring->ring, buffer);
memset(buffer, 0, sizeof(void *) * size);
ring->size = size;
ring->mask = size - 1;
ring->p_tail = 0;
ring->c_head = 0;
ring->ring = buffer;
return;
}
#endif /* CK_F_RING */
#endif /* _CK_RING_H */

8
regressions/ck_ring/validate/Makefile
Unescape View File

@ -1,13 +1,13 @@
.PHONY: check clean distribution
OBJECTS=ck_ring_spsc ck_ring_spsc_template ck_ring_mpmc
OBJECTS=ck_ring_spsc ck_ring_spsc_template ck_ring_spmc
all: $(OBJECTS)
check: all
./ck_ring_spsc $(CORES) 1 65536
./ck_ring_spsc_template $(CORES) 1 65536
./ck_ring_mpmc $(CORES) 1 65536
./ck_ring_spmc $(CORES) 1 65536
ck_ring_spsc_template: ck_ring_spsc_template.c ../../../include/ck_ring.h
$(CC) $(CFLAGS) -o ck_ring_spsc_template ck_ring_spsc_template.c
@ -15,8 +15,8 @@ ck_ring_spsc_template: ck_ring_spsc_template.c ../../../include/ck_ring.h
ck_ring_spsc: ck_ring_spsc.c ../../../include/ck_ring.h
$(CC) $(CFLAGS) -o ck_ring_spsc ck_ring_spsc.c
ck_ring_mpmc: ck_ring_mpmc.c ../../../include/ck_ring.h
$(CC) $(CFLAGS) -o ck_ring_mpmc ck_ring_mpmc.c
ck_ring_spmc: ck_ring_spmc.c ../../../include/ck_ring.h
$(CC) $(CFLAGS) -o ck_ring_spmc ck_ring_spmc.c
clean:
rm -rf *~ *.o $(OBJECTS) *.dSYM

82
regressions/ck_ring/validate/ck_ring_mpmc.c → regressions/ck_ring/validate/ck_ring_spmc.c
Unescape View File

@ -31,6 +31,7 @@
#include <pthread.h>
#include <ck_ring.h>
#include <ck_spinlock.h>
#include "../../common.h"
#ifndef ITERATIONS
@ -44,6 +45,7 @@ struct context {
};
struct entry {
unsigned long value_long;
unsigned int magic;
unsigned int ref;
int tid;
@ -52,18 +54,18 @@ struct entry {
static int nthr;
static ck_ring_t *ring;
static ck_ring_t ring_mpmc;
static ck_ring_t ring_spmc;
static struct affinity a;
static int size;
static volatile int barrier;
static int eb;
static void *
test_mpmc(void *c)
test_spmc(void *c)
{
struct entry *entry;
unsigned int observed = 0, foreign = 0;
int i, j;
unsigned int observed = 0;
int i, j, tid;
unsigned long previous = 0;
(void)c;
if (aff_iterate(&a)) {
@ -71,40 +73,22 @@ test_mpmc(void *c)
exit(EXIT_FAILURE);
}
ck_pr_inc_int(&eb);
while (ck_pr_load_int(&eb) != nthr);
tid = ck_pr_faa_int(&eb, 1);
while (ck_pr_load_int(&eb) != nthr - 1);
for (i = 0; i < ITERATIONS; i++) {
for (j = 0; j < size; j++) {
struct entry *o;
entry = malloc(sizeof(*entry));
assert(entry != NULL);
entry->magic = 0xdead;
entry->tid = j;
entry->value = j;
entry->ref = 0;
if (ck_ring_enqueue_mpmc(&ring_mpmc, entry) == false) {
free(entry);
j--;
if (j < 0)
j = 0;
}
/* Keep trying until we encounter at least one node. */
if (ck_ring_dequeue_mpmc(&ring_mpmc, &o) == false) {
j--;
if (j < 0)
j = 0;
continue;
}
while (ck_ring_dequeue_spmc(&ring_spmc, &o) == false);
observed++;
foreign += entry != o;
if (o->value < 0 || o->value >= size || o->value != o->tid || o->magic != 0xdead) {
fprintf(stderr, "[%p] (%x) (%d, %d) >< (0, %d)\n",
if (o->value < 0
|| o->value != o->tid
|| o->magic != 0xdead
|| (previous != 0 && previous >= o->value_long)) {
fprintf(stderr, "[0x%p] (%x) (%d, %d) >< (0, %d)\n",
(void *)o, o->magic, o->tid, o->value, size);
exit(EXIT_FAILURE);
}
@ -112,6 +96,7 @@ test_mpmc(void *c)
o->magic = 0xbeef;
o->value = -31337;
o->tid = -31338;
previous = o->value_long;
if (ck_pr_faa_uint(&o->ref, 1) != 0) {
fprintf(stderr, "[%p] We dequeued twice.\n", (void *)o);
@ -122,7 +107,7 @@ test_mpmc(void *c)
}
}
fprintf(stderr, "Observed %u / Foreign %u\n", observed, foreign);
fprintf(stderr, "[%d] Observed %u\n", tid, observed);
return NULL;
}
@ -155,7 +140,7 @@ test(void *c)
entries[i].value = i;
entries[i].tid = 0;
r = ck_ring_enqueue_mpmc(ring, entries + i);
r = ck_ring_enqueue_spmc(ring, entries + i);
assert(r != false);
}
@ -178,7 +163,7 @@ test(void *c)
for (i = 0; i < ITERATIONS; i++) {
for (j = 0; j < size; j++) {
while (ck_ring_dequeue_mpmc(ring + context->previous, &entry) == false);
while (ck_ring_dequeue_spmc(ring + context->previous, &entry) == false);
if (context->previous != (unsigned int)entry->tid) {
fprintf(stderr, "[%u:%p] %u != %u\n",
@ -193,7 +178,7 @@ test(void *c)
}
entry->tid = context->tid;
r = ck_ring_enqueue_mpmc(ring + context->tid, entry);
r = ck_ring_enqueue_spmc(ring + context->tid, entry);
assert(r == true);
}
}
@ -206,6 +191,7 @@ main(int argc, char *argv[])
{
int i, r;
void *buffer;
unsigned long l;
struct context *context;
pthread_t *thread;
@ -259,18 +245,34 @@ main(int argc, char *argv[])
pthread_join(thread[i], NULL);
fprintf(stderr, " done\n");
fprintf(stderr, "MPMC test:\n");
fprintf(stderr, "SPMC test:\n");
buffer = malloc(sizeof(void *) * (size + 1));
assert(buffer);
memset(buffer, 0, sizeof(void *) * (size + 1));
ck_ring_init(&ring_mpmc, buffer, size + 1);
for (i = 0; i < nthr; i++) {
r = pthread_create(thread + i, NULL, test_mpmc, context + i);
ck_ring_init(&ring_spmc, buffer, size + 1);
for (i = 0; i < nthr - 1; i++) {
r = pthread_create(thread + i, NULL, test_spmc, context + i);
assert(r == 0);
}
for (i = 0; i < nthr; i++)
for (l = 0; l < (unsigned long)size * ITERATIONS * (nthr - 1) ; l++) {
struct entry *entry = malloc(sizeof *entry);
assert(entry != NULL);
entry->value_long = l;
entry->value = (int)l;
entry->tid = (int)l;
entry->magic = 0xdead;
entry->ref = 0;
/* Wait until queue is not full. */
while (ck_ring_enqueue_spmc(&ring_spmc, entry) == false)
ck_pr_stall();
}
for (i = 0; i < nthr - 1; i++)
pthread_join(thread[i], NULL);
return (0);
}
Write Preview
Loading…
Cancel
Save