cozis/lock_free_queue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
cozis
2024-03-23 22:53:10 +01:00
parent 8a05101f0d
commit 172462b023
7 changed files with 129 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -1 +1,2 @@
test_queue
test_queue.*
lock_free_queue.c
+104 -26
View File
@@ -3,6 +3,23 @@
#include <stdatomic.h>
#include "lock_free_queue.h"
#include <stdio.h>
#include <stdarg.h>
static int
kind_of_atomic_fprintf(FILE *stream, const char *fmt, ...)
{
char msg[256];
va_list args;
va_start(args, fmt);
int len = vsnprintf(msg, sizeof(msg), fmt, args);
va_end(args);
if (len < 0) return 0;
return fwrite(msg, 1, len, stream);
}
void
lock_free_queue_init(struct lock_free_queue *q,
void *arrptr, int arrlen,
@@ -12,7 +29,8 @@ lock_free_queue_init(struct lock_free_queue *q,
q->head = 0;
q->tail = 0;
q->temp = 0;
q->temp_head = 0;
q->temp_tail = 0;
q->size = arrlen / cell;
q->base = arrptr;
q->cell = cell;
@@ -21,24 +39,35 @@ lock_free_queue_init(struct lock_free_queue *q,
static void*
item_addr(struct lock_free_queue *q, int index)
{
assert(index >= 0 && index < q->size);
return q->base + q->cell * index;
return q->base + q->cell * (index % q->size);
}
static int
static uint64_t
acquire_push_location(struct lock_free_queue *q)
{
int old_temp;
int new_temp;
uint64_t old_temp_tail;
uint64_t new_temp_tail;
do {
old_temp = q->temp;
new_temp = (old_temp + 1) % q->size;
} while (!atomic_compare_exchange_strong(&q->temp, &old_temp, new_temp));
return old_temp;
uint64_t old_temp_head;
old_temp_tail = q->temp_tail;
old_temp_head = q->temp_head;
if (old_temp_head + q->size == old_temp_tail)
return UINT64_MAX; // Queue is full
assert(old_temp_head + q->size > old_temp_tail);
new_temp_tail = old_temp_tail + 1;
} while (!atomic_compare_exchange_weak(&q->temp_tail, &old_temp_tail, new_temp_tail));
return old_temp_tail;
}
static void
release_push_location(struct lock_free_queue *q, int index)
static void
release_push_location(struct lock_free_queue *q, uint64_t index)
{
// The current thread already inserted an
// item at position "index", which comes
@@ -51,34 +80,83 @@ release_push_location(struct lock_free_queue *q, int index)
// Before being able to move the tail over
// this element, we need to wait for other
// threads to do it.
while (q->tail != index);
// At this point only this thread is allowed
// to push the tail forward
q->tail++;
}
q->tail = (q->tail + 1) % q->size;
bool
lock_free_queue_try_push(struct lock_free_queue *q, void *src)
{
uint64_t index = acquire_push_location(q);
if (index == UINT64_MAX) return false;
void *dst = item_addr(q, index);
memcpy(dst, src, q->cell);
//kind_of_atomic_fprintf(stdout, "push at %d\n", index % q->size);
release_push_location(q, index);
return true;
}
void
lock_free_queue_push(struct lock_free_queue *q, void *src)
{
int index = acquire_push_location(q);
while (!lock_free_queue_try_push(q, src));
}
void *dst = item_addr(q, index);
static uint64_t
acquire_pop_location(struct lock_free_queue *q)
{
uint64_t old_head;
uint64_t new_head;
do {
uint64_t old_tail;
// It's important to get head before tail
// or head may be incremented over tail before
// querying it.
old_head = q->head;
old_tail = q->tail;
if (old_head == old_tail)
return UINT64_MAX;
assert(old_tail > old_head);
new_head = old_head + 1;
} while (!atomic_compare_exchange_weak(&q->head, &old_head, new_head));
return old_head;
}
static void
release_pop_location(struct lock_free_queue *q, uint64_t index)
{
while (q->temp_head != index);
q->temp_head++;
}
bool
lock_free_queue_try_pop(struct lock_free_queue *q, void *dst)
{
uint64_t index = acquire_pop_location(q);
if (index == UINT64_MAX)
return false;
void *src = item_addr(q, index);
memcpy(dst, src, q->cell);
release_push_location(q, index);
//kind_of_atomic_fprintf(stdout, "pop at %d\n", index % q->size);
release_pop_location(q, index);
return true;
}
void
lock_free_queue_pop(struct lock_free_queue *q, void *dst)
{
int old_head;
int new_head;
do {
old_head = q->head;
new_head = (old_head + 1) % q->size;
void *src = item_addr(q, old_head);
memcpy(dst, src, q->cell);
} while (!atomic_compare_exchange_strong(&q->head, &old_head, new_head));
while (!lock_free_queue_try_pop(q, dst));
}
lock_free_queue.h
+19 -5
View File
@@ -1,19 +1,33 @@
#include <stdint.h>
#include <stdbool.h>
/*
* +--------------+-----------------+----------------+-------------------+-------------+
* | unused | Pop in progress | Items | Push in progress | Unused |
* +--------------+-----------------+----------------+-------------------+-------------+
* ^ ^ ^ ^
* temp_head head tail temp_tail
*/
struct lock_free_queue {
_Atomic(int) head;
_Atomic(int) tail;
_Atomic(int) temp;
_Atomic(uint64_t) head;
_Atomic(uint64_t) tail;
_Atomic(uint64_t) temp_head;
_Atomic(uint64_t) temp_tail;
// The following aren't written to by
// push and pop operations, therefore
// don't need to be atomic.
int size; // Capacity of the queue (item count)
int cell; // Size of a single item (in bytes)
uint64_t size; // Capacity of the queue (item count)
uint64_t cell; // Size of a single item (in bytes)
char *base; // Pointer to the first item
};
void lock_free_queue_init(struct lock_free_queue *q, void *arrptr, int arrlen, int item_size);
void lock_free_queue_push(struct lock_free_queue *q, void *src);
void lock_free_queue_pop(struct lock_free_queue *q, void *dst);
bool lock_free_queue_try_push(struct lock_free_queue *q, void *src);
bool lock_free_queue_try_pop(struct lock_free_queue *q, void *dst);
#define lock_free_queue_INIT(q, arr) lock_free_queue_init(q, arr, sizeof(arr), sizeof((arr)[0]))
lock_free_queue_test.c
+4 -26
View File
@@ -2,17 +2,15 @@
#include <stdlib.h>
#include <stdarg.h>
#include "tinycthread.h"
#include "tinycsem.h"
#include "lock_free_queue.h"
#define NUM_THREADS 30
#define OPERATIONS_PER_WORKER 100
struct lock_free_queue queue;
struct tinycsem consumed;
struct tinycsem produced;
static int
static int
kind_of_atomic_fprintf(FILE *stream, const char *fmt, ...)
{
char msg[256];
@@ -27,8 +25,6 @@ kind_of_atomic_fprintf(FILE *stream, const char *fmt, ...)
return fwrite(msg, 1, len, stream);
}
#define OPERATIONS_PER_WORKER 10000
int consumer(void *arg)
{
(void) arg;
@@ -37,19 +33,12 @@ int consumer(void *arg)
for (int i = 0; i < OPERATIONS_PER_WORKER; i++) {
tinycsem_wait(&produced);
int value;
lock_free_queue_pop(&queue, &value);
sum += value;
kind_of_atomic_fprintf(stderr, "Thread consumed %d (i=%d)\n", value, i);
tinycsem_signal(&consumed);
}
kind_of_atomic_fprintf(stderr, "Consumer stopped (sum=%d)\n", sum);
return sum;
}
@@ -62,33 +51,24 @@ int producer(void *arg)
int sum = 0;
for (int i = 0; i < OPERATIONS_PER_WORKER; i++) {
tinycsem_wait(&consumed);
int value = rand() % 10000;
lock_free_queue_push(&queue, &value);
kind_of_atomic_fprintf(stderr, "Thread produced %d (i=%d)\n", value, i);
sum += value;
tinycsem_signal(&produced);
}
kind_of_atomic_fprintf(stderr, "Producer stopped (sum=%d)\n", sum);
return sum;
}
#define MAX_ITEMS 100
#define MAX_ITEMS 10
int main()
{
int items[MAX_ITEMS];
lock_free_queue_INIT(&queue, items);
tinycsem_init(&consumed, MAX_ITEMS, MAX_ITEMS);
tinycsem_init(&produced, 0, MAX_ITEMS);
thrd_t threads[NUM_THREADS];
int created = 0;
for (int i = 0; i < NUM_THREADS; i++) {
@@ -122,7 +102,5 @@ int main()
produced_sum, consumed_sum,
(produced_sum == consumed_sum) ? "OK" : "Don't match!!");
tinycsem_free(&consumed);
tinycsem_free(&produced);
return 0;
}
makefile
+1 -1
View File
@@ -1,3 +1,3 @@
all:
gcc lock_free_queue_test.c lock_free_queue.c tinycsem.c tinycthread.c -o test_queue -Wall -Wextra
gcc lock_free_queue_test.c lock_free_queue.c tinycthread.c -o test_queue -ggdb -Wall -Wextra #-fsanitize=thread
tinycsem.c
-45
View File
@@ -1,45 +0,0 @@
#include "tinycsem.h"
bool tinycsem_init(struct tinycsem *sem, int count, int max)
{
#ifdef _WIN32
SECURITY_ATTRIBUTES *attr = NULL; // Default
const char *name = NULL; // No name
HANDLE handle = CreateSemaphore(attr, count, max, name);
if (handle == NULL)
return false;
sem->data = handle;
return true;
#else
(void) max; // POSIX doesn't use this
return sem_init(&sem->data, 0, count) == 0;
#endif
}
bool tinycsem_free(struct tinycsem *sem)
{
#ifdef _WIN32
CloseHandle(sem->data);
return true;
#else
return sem_destroy(&sem->data) == 0;
#endif
}
bool tinycsem_wait(struct tinycsem *sem)
{
#ifdef _WIN32
return WaitForSingleObject(sem->data, INFINITE) == WAIT_OBJECT_0;
#else
return sem_wait(&sem->data) == 0;
#endif
}
bool tinycsem_signal(struct tinycsem *sem)
{
#ifdef _WIN32
return ReleaseSemaphore(sem->data, 1, NULL);
#else
return sem_post(&sem->data) == 0;
#endif
}
tinycsem.h
-47
View File
@@ -1,47 +0,0 @@
#include <stddef.h>
#include <stdbool.h>
#ifdef _WIN32
#include <windows.h>
#else
#include <semaphore.h>
#endif
/*
* Semaphore structure
*/
struct tinycsem {
#ifdef _WIN32
HANDLE data;
#else
sem_t data;
#endif
};
/*
* Initialize a semaphore with a value of "count".
*
* The count of the semaphore must always be lower
* than "max", else behaviour is undefined.
*/
bool tinycsem_init(struct tinycsem *sem, int count, int max);
/*
* Releases the resources held by a semaphore previously
* initialized with "tinycsem_init".
*
* After this function is used on a semaphore, you can't
* reuse it unless you initialize it again.
*/
bool tinycsem_free(struct tinycsem *sem);
/*
* The usual semaphore wait operation.
*/
bool tinycsem_wait(struct tinycsem *sem);
/*
* The usual semaphore signal operation.
*/
bool tinycsem_signal(struct tinycsem *sem);