#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <assert.h>
#include <stdint.h>
#include <string.h>
#include <unistd.h>
#include <sched.h>
#define BUFFER_SIZE 32
#define NUM_PRODUCERS 2
#define NUM_CONSUMERS 2
#define ITEMS_PER_PRODUCER 10
#define EMPTY_SLOT -1
typedef struct {
uint32_t head;
uint32_t count;
} queue_state;
_Static_assert(sizeof(queue_state) == sizeof(uint64_t),
"queue_state must be 64 bits");
static _Atomic int buffer[BUFFER_SIZE];
static _Atomic uint64_t state;
static atomic_int num_producers = 0;
static atomic_int sum_of_produced = 0;
bool
enqueue(int value)
{
queue_state old_state;
uint32_t tail;
int expect_empty;
uint64_t old_state_val, cur_state_val, new_state_val;
do {
old_state_val = atomic_load(&state);
memcpy(&old_state, &old_state_val, sizeof(uint64_t));
if (old_state.count == BUFFER_SIZE) {
return false; // Queue is full
}
tail = (old_state.head + old_state.count) % BUFFER_SIZE;
expect_empty = EMPTY_SLOT;
// Attempt to claim the empty slot
// Loop back if the slot was filled by another thread
} while (
!atomic_compare_exchange_weak(&buffer[tail], &expect_empty, value));
// Progress point, we've actually dequeued something, but we need to
// update the state
cur_state_val = old_state_val;
do {
// Update the state (an enqueue should be impossible until
// we're done)
queue_state cur_state;
memcpy(&cur_state, &cur_state_val, sizeof(uint64_t));
// Sanity check
int cur_tail = (cur_state.head + cur_state.count) % BUFFER_SIZE;
assert(tail == cur_tail);
// Update the state
queue_state new_state = cur_state;
new_state.count++;
memcpy(&new_state_val, &new_state, sizeof(uint64_t));
} while (!atomic_compare_exchange_weak(&state, &cur_state_val,
new_state_val));
return true;
}
bool
dequeue(int *value)
{
queue_state old_state;
uint64_t old_state_val, cur_state_val, new_state_val;
do {
old_state_val = atomic_load(&state);
memcpy(&old_state, &old_state_val, sizeof(uint64_t));
if (old_state.count == 0) {
return false; // Queue is empty
}
*value = atomic_load(&buffer[old_state.head]);
// Attempt to claim the value
// Loop back if the slot was emptied by another thread
} while (!atomic_compare_exchange_weak(&buffer[old_state.head], value,
EMPTY_SLOT));
// Progress point, we've actually dequeued something, but we need to
// update the state
cur_state_val = old_state_val;
do {
// Update the state (a dequeue should be impossible until
// we're done)
queue_state cur_state;
memcpy(&cur_state, &cur_state_val, sizeof(uint64_t));
// Sanity check
assert(cur_state.count > 0 && cur_state.head == old_state.head);
// Update the state
queue_state new_state = cur_state;
new_state.count--;
new_state.head = (old_state.head + 1) % BUFFER_SIZE;
memcpy(&new_state_val, &new_state, sizeof(uint64_t));
} while (!atomic_compare_exchange_weak(&state, &cur_state_val,
new_state_val));
return true;
}
void *
producer(void *arg)
{
int id = *(int *) arg;
for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
int value =
id * 100 + i + 1; // Ensure we never produce EMPTY_SLOT
while (!enqueue(value)) {
sched_yield(); // Queue is full, yield and try again
}
printf("Producer %d enqueued: %d\n", id, value);
usleep(rand() % 100000); // Sleep up to 0.1 seconds
}
atomic_fetch_sub(&num_producers, 1);
printf("Producer %d finished\n", id);
return NULL;
}
void *
consumer(void *arg)
{
int id = *(int *) arg;
while (1) {
int value;
if (dequeue(&value)) {
printf("Consumer %d dequeued: %d\n", id, value);
atomic_fetch_add(&sum_of_produced, value);
usleep(rand() % 200000); // Sleep up to 0.2 seconds
} else {
if (atomic_load(&num_producers) == 0) {
printf(
"Consumer %d exiting: no more producers\n",
id);
break;
}
sched_yield(); // Queue is empty, yield and try again
}
}
return NULL;
}
int
main()
{
pthread_t producers[NUM_PRODUCERS];
pthread_t consumers[NUM_CONSUMERS];
int producer_ids[NUM_PRODUCERS];
int consumer_ids[NUM_CONSUMERS];
// Initialize state and buffer
queue_state initial_state = {.head = 0, .count = 0};
atomic_store(&state, *(uint64_t *) &initial_state);
for (int i = 0; i < BUFFER_SIZE; i++) {
atomic_store(&buffer[i], EMPTY_SLOT);
}
// Set initial number of producers
atomic_store(&num_producers, NUM_PRODUCERS);
// Create producer threads
for (int i = 0; i < NUM_PRODUCERS; i++) {
producer_ids[i] = i;
pthread_create(&producers[i], NULL, producer, &producer_ids[i]);
}
// Create consumer threads
for (int i = 0; i < NUM_CONSUMERS; i++) {
consumer_ids[i] = i;
pthread_create(&consumers[i], NULL, consumer, &consumer_ids[i]);
}
// Wait for all threads to complete
for (int i = 0; i < NUM_PRODUCERS; i++) {
pthread_join(producers[i], NULL);
}
for (int i = 0; i < NUM_CONSUMERS; i++) {
pthread_join(consumers[i], NULL);
}
int final_sum = atomic_load(&sum_of_produced);
printf("Final sum of produced values: %d\n", final_sum);
return 0;
}
