queue.c

#include <stdlib.h>
#include <stdatomic.h>
#include <threads.h>

typedef struct QueueNode {
  void* data;
  struct QueueNode* next;
} QueueNode;

typedef struct SleepingThread {
  cnd_t condition;           // Condition variable for this specific thread
  void* reserved_item;       // Item reserved for this thread (NULL if none)
  struct SleepingThread* next;
} SleepingThread;

typedef struct {
  QueueNode* head;          // Pointer to the head of the queue
  QueueNode* tail;          // Pointer to the tail of the queue

  SleepingThread* sleep_head; // Head of sleeping threads queue (oldest sleeper)
  SleepingThread* sleep_tail; // Tail of sleeping threads queue (newest sleeper)

  atomic_int visit_count;          // Number of items that have been dequeued
  mtx_t mutex;              // Mutex for synchronizing access to the queue
} Queue;

static Queue queue;

void initQueue(void) {
  // Initialize all pointers to NULL
  queue.head = NULL;
  queue.tail = NULL;
  queue.sleep_head = NULL;
  queue.sleep_tail = NULL;
  
  // Initialize mutex and atomic counter
  mtx_init(&queue.mutex, mtx_plain);
  atomic_init(&queue.visit_count, 0);
}

void destroyQueue(void) { 
  // Clean up remaining data nodes
  QueueNode* current = queue.head;
  while (current) {
      QueueNode* next = current->next;
      free(current);
      current = next;
  }
  
  // Clean up remaining sleeping thread nodes
  SleepingThread* sleep_current = queue.sleep_head;
  while (sleep_current) {
      SleepingThread* sleep_next = sleep_current->next;
      cnd_destroy(&sleep_current->condition);
      free(sleep_current);
      sleep_current = sleep_next;
  }
  
  // Destroy mutex and reset state
  mtx_destroy(&queue.mutex);
  queue.head = NULL;
  queue.tail = NULL;
  queue.sleep_head = NULL;
  queue.sleep_tail = NULL;
}

void enqueue(void* item) {
  mtx_lock(&queue.mutex);
  
  // If there are sleeping threads, give item directly to oldest sleeping thread
  if (queue.sleep_head) {
      SleepingThread* to_wake = queue.sleep_head;
      
      // Reserve the item for this specific thread
      to_wake->reserved_item = item;
      
      // Remove this thread from sleep queue immediately to prevent double assignment
      queue.sleep_head = to_wake->next;
      if (!queue.sleep_head) {
          queue.sleep_tail = NULL;
      }
      
      // Signal the thread to wake up
      cnd_signal(&to_wake->condition);
  } else {
      // No sleeping threads - add to regular queue
      QueueNode* new_node = malloc(sizeof(QueueNode));
      new_node->data = item;
      new_node->next = NULL;
      
      // Add to end of queue (FIFO)
      if (queue.tail) {
          queue.tail->next = new_node;
      } else {
          // Queue was empty
          queue.head = new_node;
      }
      queue.tail = new_node;
  }
  
  mtx_unlock(&queue.mutex);
}

void* dequeue(void) {
  mtx_lock(&queue.mutex);
  
  // Check if there's an item available in the regular queue
  if (queue.head) {
      // Remove item from front of queue (FIFO)
      QueueNode* node_to_remove = queue.head;
      void* data = node_to_remove->data;
      
      queue.head = node_to_remove->next;
      if (!queue.head) {
          // Queue is now empty
          queue.tail = NULL;
      }
      
      free(node_to_remove);
      
      // Increment visited counter atomically
      atomic_fetch_add(&queue.visit_count, 1);
      
      mtx_unlock(&queue.mutex);
      return data;
  }
  
  // No items available in regular queue - need to sleep
  // Create sleeping thread node
  SleepingThread* sleep_node = malloc(sizeof(SleepingThread));
  cnd_init(&sleep_node->condition);
  sleep_node->reserved_item = NULL;  // No item reserved yet
  sleep_node->next = NULL;
  
  // Add to end of sleeping threads queue (FIFO order)
  if (queue.sleep_tail) {
      queue.sleep_tail->next = sleep_node;
  } else {
      queue.sleep_head = sleep_node;
  }
  queue.sleep_tail = sleep_node;
  
  // Wait until an item is reserved for us (need loop because of spurious wakeups)
  while (sleep_node->reserved_item == NULL) {
      cnd_wait(&sleep_node->condition, &queue.mutex);
  }
  
  // We have been woken up and have a reserved item
  void* data = sleep_node->reserved_item;
  
  // Note: We were already removed from the sleeping threads queue by enqueue()
  
  // Clean up sleeping thread node
  cnd_destroy(&sleep_node->condition);
  free(sleep_node);
  
  // Increment visited counter atomically
  atomic_fetch_add(&queue.visit_count, 1);
  
  mtx_unlock(&queue.mutex);
  return data;
}

size_t visited(void) {
  // Return atomic counter value without locking
  // This is lock-free as required by the assignment
  return atomic_load(&queue.visit_count);
}