C++多线程生命游戏中Semaphore同步异常问题排查求助

问题概述 #

开发采用多线程与Semaphore实现同步的C++康威生命游戏模拟程序时，出现Semaphore控制失效，多个线程可同时访问缓冲区的异常行为，导致模拟逻辑出错。

相关代码片段 #

Cell类核心实现 #

class Cell {
private:
    struct CellMessage {
        Cell *neighbor;
        Cell *caller;
        bool state;
    };

    std::vector<bool> buffer;
    int capacity;
    sem_t empty;
    sem_t full;
    sem_t mutex;

    static void* send_message_neighbor(void* curr_cell) {
        CellMessage* c = static_cast<CellMessage*>(curr_cell);

        // wait if the buffer is full
        sem_wait(&c->neighbor->empty);
        sem_wait(&c->neighbor->mutex); 
        c->neighbor->buffer.push_back(c->state);
        sem_post(&c->neighbor->mutex);  // Release mutex
        // Signal that a new item is produced
        sem_post(&c->neighbor->full);

        return NULL;
    }

public:
    std::vector<Cell*> neighbors;
    int row = 0;
    int col = 0;
    bool state;

    Cell() {}

    Cell(bool s, int r, int c) : state(s), capacity(r + 1), row(r), col(c) {
        sem_init(&empty, 0, r+1);
        sem_init(&full, 0, 0);
        sem_init(&mutex, 0, 1); // Initialize mutex semaphore
    }

    void send_message_neighbors() {
        pthread_t tneighbors[neighbors.size()];

        for (size_t i = 0; i < neighbors.size(); ++i) {
            CellMessage* cl = new CellMessage();
            cl->neighbor = neighbors[i];
            cl->caller = this;
            cl->state = state;

            pthread_create(&tneighbors[i], NULL, send_message_neighbor, cl);
        }

        for (size_t i = 0; i < neighbors.size(); ++i) {
            pthread_join(tneighbors[i], NULL);
        }
    }
};

细胞状态计算方法 #

void compute_state() {
    int amount_alive_neighbors = 0;

    while (!buffer.empty()) {
        sem_wait(&full);

        if (*buffer.begin()) {
            amount_alive_neighbors++;
        }
        buffer.erase(buffer.begin());

        sem_post(&empty);
    }

    if (state) {
        if (amount_alive_neighbors > 3 || amount_alive_neighbors == 0) {
            state = false;
        }
    } else {
        if (amount_alive_neighbors == 3) {
            state = true;
        }
    }
}

棋盘世代运算触发方法 #

void compute_cell_neighbors() {
    for (int i = 0; i < MATRIX_SIZE; i++) {
        for (int j = 0; j < MATRIX_SIZE; j++) {
            m[i][j]->neighbors = identify_neighbors(i, j);
            m[i][j]->send_message_neighbors();
            m[i][j]->compute_state();
        }
    }
}

问题分析 #

1. 缓冲区访问未加互斥保护：compute_state中直接操作buffer（读取begin()、erase()）时未持有mutex信号量，与send_message_neighbor中的写操作形成数据竞争，导致多线程同时修改/读取缓冲区。
2. Semaphore容量初始化错误：Cell构造时用row+1作为empty信号量的初始值，但实际缓冲区需要容纳的是所有邻居的消息（最多8条），容量不匹配会导致信号量同步逻辑失效。
3. 循环条件存在竞态：while (!buffer.empty())的检查与后续sem_wait(&full)之间存在时间窗口，其他线程可能在此期间修改缓冲区状态，导致读取逻辑遗漏或重复处理消息。
4. 内存泄漏：send_message_neighbor中创建的CellMessage未释放，长期运行会耗尽内存。
5. Semaphore未销毁：Cell析构时未调用sem_destroy清理信号量资源，造成系统资源泄漏。

修复建议 #

1. 修复缓冲区访问的互斥保护 #

修改compute_state，确保所有对buffer的操作都在mutex保护下进行：

void compute_state() {
    int amount_alive_neighbors = 0;
    // 按邻居数量循环，确保读取所有邻居的消息
    for (size_t i = 0; i < neighbors.size(); ++i) {
        sem_wait(&full);
        // 操作缓冲区前加锁
        sem_wait(&mutex);
        if (*buffer.begin()) {
            amount_alive_neighbors++;
        }
        buffer.erase(buffer.begin());
        sem_post(&mutex);
        sem_post(&empty);
    }

    if (state) {
        if (amount_alive_neighbors > 3 || amount_alive_neighbors == 0) {
            state = false;
        }
    } else {
        if (amount_alive_neighbors == 3) {
            state = true;
        }
    }
}

2. 修正Semaphore容量初始化 #

将Semaphore的初始化移到邻居确定之后，确保容量与邻居数量匹配：

• 修改Cell构造函数，暂不初始化Semaphore：

Cell(bool s, int r, int c) : state(s), row(r), col(c) {}

• 在compute_cell_neighbors中设置邻居后初始化Semaphore：

void compute_cell_neighbors() {
    for (int i = 0; i < MATRIX_SIZE; i++) {
        for (int j = 0; j < MATRIX_SIZE; j++) {
            m[i][j]->neighbors = identify_neighbors(i, j);
            // 按邻居数量初始化缓冲区容量
            size_t neighbor_count = m[i][j]->neighbors.size();
            sem_init(&m[i][j]->empty, 0, neighbor_count);
            sem_init(&m[i][j]->full, 0, 0);
            sem_init(&m[i][j]->mutex, 0, 1);
            
            m[i][j]->send_message_neighbors();
            m[i][j]->compute_state();
            
            // 计算完成后销毁Semaphore
            sem_destroy(&m[i][j]->empty);
            sem_destroy(&m[i][j]->full);
            sem_destroy(&m[i][j]->mutex);
        }
    }
}

3. 修复内存泄漏 #

在send_message_neighbor末尾释放CellMessage：

static void* send_message_neighbor(void* curr_cell) {
    CellMessage* c = static_cast<CellMessage*>(curr_cell);

    sem_wait(&c->neighbor->empty);
    sem_wait(&c->neighbor->mutex); 
    c->neighbor->buffer.push_back(c->state);
    sem_post(&c->neighbor->mutex);
    sem_post(&c->neighbor->full);

    delete c; // 释放动态分配的消息对象
    return NULL;
}

4. 补充Semaphore销毁逻辑 #

如果Cell对象会被复用，建议在Cell析构函数中添加Semaphore销毁：

~Cell() {
    // 可通过额外标志位判断Semaphore是否已初始化，避免重复销毁
    sem_destroy(&empty);
    sem_destroy(&full);
    sem_destroy(&mutex);
}

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