#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <unistd.h>
#include <sched.h>
#define NUM_PRODUCERS 2
#define NUM_CONSUMERS 2
#define ITEMS_PER_PRODUCER 10
// Sentinel value
#define NO_DATA -1 // Stack is empty (transitory)
struct stack_node {
int data;
struct stack_node *next;
};
// Global stack head
static _Atomic(struct stack_node *) stack_head = NULL;
// Atomic counter for active producers
static atomic_int num_producers = 0;
// Helper function to push an item onto the stack
void
push_item(int value)
{
struct stack_node *new_node = malloc(sizeof(struct stack_node));
new_node->data = value;
struct stack_node *old_head;
do {
old_head = atomic_load(&stack_head);
new_node->next = old_head;
} while (
!atomic_compare_exchange_weak(&stack_head, &old_head, new_node));
}
// Helper function to pop an item from the stack
int
pop_item(void)
{
while (1) {
struct stack_node *popped_node = atomic_load(&stack_head);
if (popped_node == NULL) {
return NO_DATA; // Stack is empty
}
struct stack_node *new_head = popped_node->next;
if (atomic_compare_exchange_weak(&stack_head, &popped_node,
new_head)) {
int value = popped_node->data;
free(popped_node);
return value;
}
// If CAS failed, another thread modified the stack, so we try
// again
}
}
void *
producer(void *arg)
{
int id = *(int *) arg;
for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
int value = id * 100 + i; // Just to make unique items
push_item(value);
printf("Producer %d pushed: %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 = pop_item();
if (value == NO_DATA) {
if (atomic_load(&num_producers) == 0) {
printf(
"Consumer %d exiting: no more producers\n",
id);
break;
}
sched_yield(); // Stack is empty, yield and try again
continue;
}
printf("Consumer %d popped: %d\n", id, value);
usleep(rand() % 200000); // Sleep up to 0.2 seconds
}
return NULL;
}
int
main()
{
pthread_t producers[NUM_PRODUCERS];
pthread_t consumers[NUM_CONSUMERS];
int producer_ids[NUM_PRODUCERS];
int consumer_ids[NUM_CONSUMERS];
// 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);
}
return 0;
}
