基于信号量的10生产者10消费者FIFO同步问题求助

Hey there! Let's work through this classic synchronization issue—semaphores can feel tricky at first, but once you nail their roles, this problem clicks into place.

First, Clarify Semaphore & Sync Roles #

Let's get the basics straight, since misusing semaphores is usually the root of the problem:

• semPush (empty slots semaphore): Tracks how many open spots are in your FIFO. Initialize this to your FIFO's total capacity (e.g., if your queue holds 5 items, start at 5). Producers wait on this before adding items—they can't push if the queue is full.
• semPop (filled slots semaphore): Tracks how many items are ready to be consumed. Initialize this to 0, since the queue starts empty. Consumers wait on this before popping items—they can't pull if there's nothing there.
• Critical add-on: You need a mutex lock to protect the FIFO itself. Semaphores handle slot availability, but they don't stop multiple threads from modifying the queue (like writing to the same tail position or reading invalid data) at the same time.

Example Implementation (C with POSIX Threads) #

Here's a working example with 10 producers, 10 consumers, and a circular buffer FIFO. I've added comments to walk through each step:

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

#define FIFO_CAPACITY 5
#define NUM_PRODUCERS 10
#define NUM_CONSUMERS 10

// FIFO buffer setup
int fifo[FIFO_CAPACITY];
int head = 0; // Next position to read from
int tail = 0; // Next position to write to

// Synchronization primitives
sem_t semPush; // Counts empty slots
sem_t semPop;  // Counts filled slots
pthread_mutex_t fifo_mutex; // Protects FIFO access

// Producer thread logic
void* producer(void* arg) {
    int producer_id = *(int*)arg;
    free(arg); // Clean up the passed ID

    // Each producer generates 3 random numbers
    for (int i = 0; i < 3; i++) {
        int item = rand() % 100; // Generate random item

        // Wait for an empty slot to become available
        sem_wait(&semPush);
        // Lock the FIFO to prevent race conditions during modification
        pthread_mutex_lock(&fifo_mutex);

        // Add item to the FIFO
        fifo[tail] = item;
        tail = (tail + 1) % FIFO_CAPACITY;
        printf("Producer %d pushed %d\n", producer_id, item);

        // Unlock the FIFO after modification
        pthread_mutex_unlock(&fifo_mutex);
        // Signal that a new item is ready for consumption
        sem_post(&semPop);

        // Simulate work time (optional, for realistic threading behavior)
        sleep(rand() % 2);
    }
    return NULL;
}

// Consumer thread logic
void* consumer(void* arg) {
    int consumer_id = *(int*)arg;
    free(arg);

    // Each consumer takes 3 items
    for (int i = 0; i < 3; i++) {
        // Wait for an item to become available
        sem_wait(&semPop);
        // Lock the FIFO to safely read/modify its state
        pthread_mutex_lock(&fifo_mutex);

        // Remove item from the FIFO
        int item = fifo[head];
        head = (head + 1) % FIFO_CAPACITY;
        printf("Consumer %d popped %d\n", consumer_id, item);

        // Unlock the FIFO after modification
        pthread_mutex_unlock(&fifo_mutex);
        // Signal that an empty slot is now available
        sem_post(&semPush);

        // Simulate work time
        sleep(rand() % 2);
    }
    return NULL;
}

int main() {
    pthread_t producers[NUM_PRODUCERS];
    pthread_t consumers[NUM_CONSUMERS];

    // Initialize synchronization tools
    sem_init(&semPush, 0, FIFO_CAPACITY);
    sem_init(&semPop, 0, 0);
    pthread_mutex_init(&fifo_mutex, NULL);

    // Create producer threads
    for (int i = 0; i < NUM_PRODUCERS; i++) {
        int* id = malloc(sizeof(int));
        *id = i + 1; // Assign unique ID for debugging
        pthread_create(&producers[i], NULL, producer, id);
    }

    // Create consumer threads
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        int* id = malloc(sizeof(int));
        *id = i + 1;
        pthread_create(&consumers[i], NULL, consumer, id);
    }

    // Wait for all threads to finish execution
    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);
    }

    // Clean up resources
    sem_destroy(&semPush);
    sem_destroy(&semPop);
    pthread_mutex_destroy(&fifo_mutex);

    return 0;
}

Key Rules to Follow #

1. Order of operations matters:
   • Producers: Always wait on semPush before locking the mutex. If you lock first, you could block holding the mutex and freeze all queue access.
   • Consumers: Same rule—wait on semPop first, then lock the mutex.
2. Semaphore initialization is non-negotiable:
   • semPush starts at your FIFO's capacity (how many items it can hold).
   • semPop starts at 0 (no items to consume initially).
3. Mutex protects the FIFO: Never skip this—without it, multiple threads will corrupt your queue's state (e.g., two producers writing to the same spot).

Common Mistakes to Avoid #

• Forgetting the mutex: Semaphores don't handle queue state protection—they only manage slot availability.
• Reversing semaphore waits/posts: If you post semPop before adding the item to the queue, consumers might try to read data that doesn't exist yet.
• Initializing semaphores to wrong values: Starting semPop at 1 instead of 0 will let consumers try to pop an empty queue immediately.

Give this implementation a test, and if you hit specific snags with your code, feel free to share details!

内容的提问来源于stack exchange，提问作者Repi