#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdbool.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)
{
while (1) {
queue_state old_state = {.head = 0, .count = 0};
uint64_t 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
}
uint32_t tail =
(old_state.head + old_state.count) % BUFFER_SIZE;
// Attempt to claim the empty slot
int expected = EMPTY_SLOT;
if (!atomic_compare_exchange_strong(&buffer[tail], &expected,
value)) {
continue; // Slot was filled by another thread, try
// again
}
// Update the state
queue_state new_state = old_state;
new_state.count++;
uint64_t new_state_val;
memcpy(&new_state_val, &new_state, sizeof(uint64_t));
if (atomic_compare_exchange_weak(&state, &old_state_val,
new_state_val)) {
return true;
}
// If state update failed, revert the buffer change
atomic_store(&buffer[tail], EMPTY_SLOT);
}
}
bool
dequeue(int *value)
{
while (1) {
queue_state old_state = {.head = 0, .count = 0};
uint64_t 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
}
if (!atomic_compare_exchange_strong(&buffer[old_state.head],
value, EMPTY_SLOT)) {
continue; // Slot was emptied by another thread, try
// again
}
queue_state new_state = old_state;
new_state.count--;
new_state.head = (old_state.head + 1) % BUFFER_SIZE;
uint64_t new_state_val;
memcpy(&new_state_val, &new_state, sizeof(uint64_t));
if (atomic_compare_exchange_weak(&state, &old_state_val,
new_state_val)) {
return true;
}
// If state update failed, revert the buffer change
atomic_store(&buffer[old_state.head], *value);
}
}
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;
}
