diff --git a/.gitignore b/.gitignore index d9a9331..73ddc0e 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ +test_queue test_queue.* \ No newline at end of file diff --git a/lock_free_queue.c b/lock_free_queue.c index 04638fd..f08d941 100644 --- a/lock_free_queue.c +++ b/lock_free_queue.c @@ -3,6 +3,23 @@ #include #include "lock_free_queue.h" +#include +#include +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)); } diff --git a/lock_free_queue.h b/lock_free_queue.h index ac5825d..7c46021 100644 --- a/lock_free_queue.h +++ b/lock_free_queue.h @@ -1,19 +1,33 @@ +#include +#include + +/* + * +--------------+-----------------+----------------+-------------------+-------------+ + * | 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])) \ No newline at end of file diff --git a/lock_free_queue_test.c b/lock_free_queue_test.c index 72de7f5..9410ca1 100644 --- a/lock_free_queue_test.c +++ b/lock_free_queue_test.c @@ -2,17 +2,15 @@ #include #include #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; } \ No newline at end of file diff --git a/makefile b/makefile index d5e2345..75b4da7 100644 --- a/makefile +++ b/makefile @@ -1,3 +1,3 @@ all: - gcc lock_free_queue_test.c lock_free_queue.c tinycsem.c tinycthread.c -o test_queue -Wall -Wextra \ No newline at end of file + gcc lock_free_queue_test.c lock_free_queue.c tinycthread.c -o test_queue -ggdb -Wall -Wextra #-fsanitize=thread \ No newline at end of file diff --git a/tinycsem.c b/tinycsem.c deleted file mode 100644 index 9cd5a17..0000000 --- a/tinycsem.c +++ /dev/null @@ -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 -} \ No newline at end of file diff --git a/tinycsem.h b/tinycsem.h deleted file mode 100644 index c0f99e6..0000000 --- a/tinycsem.h +++ /dev/null @@ -1,47 +0,0 @@ - -#include -#include - -#ifdef _WIN32 -#include -#else -#include -#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); \ No newline at end of file