A message queue made by6 ChatGPT

message_queue.c

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#define MAX_QUEUE_SIZE 10

typedef struct {
  int tag;
  char *message;
} Message;

typedef struct {
  Message *queue;
  int head;
  int tail;
  int size;
  pthread_mutex_t lock;
  pthread_cond_t not_full;
  pthread_cond_t not_empty;
} MessageQueue;

void init_queue(MessageQueue *queue) {
  queue->queue = malloc(MAX_QUEUE_SIZE * sizeof(Message));
  queue->head = 0;
  queue->tail = 0;
  queue->size = 0;
  pthread_mutex_init(&queue->lock, NULL);
  pthread_cond_init(&queue->not_full, NULL);
  pthread_cond_init(&queue->not_empty, NULL);
}

void enqueue(MessageQueue *queue, Message message) {
  pthread_mutex_lock(&queue->lock);

  // Wait until the queue is not full
  while (queue->size == MAX_QUEUE_SIZE) {
    pthread_cond_wait(&queue->not_full, &queue->lock);
  }

  queue->queue[queue->tail] = message;
  queue->tail = (queue->tail + 1) % MAX_QUEUE_SIZE;
  queue->size++;

  // Signal that the queue is not empty
  pthread_cond_signal(&queue->not_empty);
  pthread_mutex_unlock(&queue->lock);
}

Message dequeue(MessageQueue *queue, int tag) {
  pthread_mutex_lock(&queue->lock);

  // Wait until the queue is not empty or we find a message with the correct tag
  while (queue->size == 0 || queue->queue[queue->head].tag != tag) {
    pthread_cond_wait(&queue->not_empty, &queue->lock);
  }

  Message message = queue->queue[queue->head];
  queue->head = (queue->head + 1) % MAX_QUEUE_SIZE;
  queue->size--;

  // Signal that the queue is not full
  pthread_cond_signal(&queue->not_full);
  pthread_mutex_unlock(&queue->lock);

  return message;
}

void *producer(void *arg) {
  MessageQueue *queue = (MessageQueue *)arg;

  while (1) {
    // Generate a message with a random tag
    Message message;
    message.tag = rand() % 10;
    message.message = "Hello, world!";

    enqueue(queue, message);
    printf("Producer enqueued message with tag %d\n", message.tag);
  }
}

void *consumer(void *arg) {
  MessageQueue *queue = (MessageQueue *)
(MessageQueue *)arg;
  int tag = *((int *)arg);

  while (1) {
    sleep(15);
    printf("Consumer waking up to fetch messages with tag %d\n", tag);

    // Fetch all messages with the given tag
    while (1) {
      Message message = dequeue(queue, tag);
      printf("Consumer dequeued message with tag %d: %s\n", message.tag, message.message);
    }
  }
}

int main() {
  MessageQueue queue;
  init_queue(&queue);

  pthread_t producer_threads[3];
  for (int i = 0; i < 3; i++) {
    pthread_create(&producer_threads[i], NULL, producer, &queue);
  }

  int tag = 5;
  pthread_t consumer_thread;
  pthread_create(&consumer_thread, NULL, consumer, &tag);

  pthread_exit(NULL);
}