共享内存数据损坏：生产者消费者程序校验和频繁不匹配求助

1. 共享内存大小与映射错误 #

• 问题：
  1. 创建共享内存时，ftruncate(shm_fd, shmSize)仅设置了shmSize字节的大小，但shmSize是缓冲区中Item的数量，每个Item占32字节，实际需要的大小应为shmSize * sizeof(Item)。
  2. 生产者线程中mmap(0, 32, ...)只映射了32字节（单个Item大小），而消费者映射了shmSize字节，两者映射的内存区域不匹配，导致数据写入/读取越界。
  3. 线程重复打开和映射共享内存，全局shm_ptr被多次覆盖，访问的内存区域混乱。
• 修复：
  • 在CreateSharedMemory中，将ftruncate的参数改为shmSize * sizeof(Item)。
  • 在main函数中完成共享内存的映射，线程直接使用全局的shm_ptr，无需重复打开映射。

2. 校验和计算长度错误 #

• 问题：
  生成和验证校验和时，使用了shmSize-2作为数据长度，这完全错误——校验和应该针对Item的data数组（固定22字节），而非共享内存大小。
• 修复：
  将CreateItem和consumer中的checksum调用参数从shmSize-2改为22。

3. 缓冲区指针初始化与边界判断错误 #

• 问题：
  1. main函数中错误地将out初始化为shmSize，导致消费者首次访问shm_ptr[out]时直接越界。
  2. 生产者中的while (((in + 1)%shmSize) == out)和消费者中的while (in == out)是空循环，完全多余——信号量已经保证了缓冲区有空闲/有数据，且互斥锁保护了临界区，这些循环会导致不必要的等待甚至死锁。
  3. 消费者中的if(out > 0)判断会跳过第一个Item（out=0时不处理），导致数据处理错位。
• 修复：
  • 将main中out的初始化改为out = 0。
  • 删除生产者和消费者中多余的空循环。
  • 删除消费者中的if(out > 0)判断，直接处理当前out指向的Item。

4. 信号量删除参数错误 #

• 问题：
  sem_unlink(&empty)和sem_unlink(&full)传入的是sem_t*指针，而sem_unlink需要的是信号量的名字字符串（如"/empty"），这会导致信号量无法正确删除，残留的信号量会影响后续程序运行。
• 修复：
  将sem_unlink的参数改为正确的信号量名称：sem_unlink("/empty")和sem_unlink("/full")。

5. 其他冗余代码 #

• 问题：
  CreateItem中的局部变量j每次调用都会重置为0，随后的++j完全无效，且生产者中会重新设置tempItem.seqNo = in，这段代码属于冗余。
• 修复：
  删除CreateItem中关于j的冗余代码，直接用in作为序列号。

修正后的完整代码 #

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/shm.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <sys/types.h>
#include <errno.h>
#include <stdint.h>
#include <semaphore.h>
#include <time.h>
#include <pthread.h>

typedef struct{
    int seqNo;
    uint16_t checksum;
    uint32_t timeStamp;
    uint8_t data[22];
} Item;

char* shm_name = "buffer";
int shm_fd;
uint8_t* shm_ptr;
pthread_t producers;
pthread_t consumers;
pthread_mutex_t mutex;
sem_t *empty, *full;
int shmSize;
int in = 0;
int out = 0;

// 初始化共享内存
void CreateSharedMemory(){
    shm_fd = shm_open(shm_name, O_CREAT | O_RDWR, 0644);
    if (shm_fd == -1) {
        fprintf(stderr, "Error unable to create shared memory, '%s, errno = %d (%s)\n", shm_name,
                errno, strerror(errno));
        exit(EXIT_FAILURE);
    }

    // 设置共享内存大小为Item数量 * 单个Item大小
    size_t total_size = shmSize * sizeof(Item);
    if (ftruncate(shm_fd, total_size) == -1) {
        fprintf(stderr, "Error configure shared memory, '%s, errno = %d (%s)\n", shm_name,
                errno, strerror(errno));
        shm_unlink(shm_name);
        exit(EXIT_FAILURE);
    }

    // 映射共享内存到进程地址空间
    shm_ptr = (uint8_t*)mmap(0, total_size, PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
    if (shm_ptr == MAP_FAILED) {
        fprintf(stderr, "Error mapping shared memory, errno = %d (%s)\n", errno, strerror(errno));
        shm_unlink(shm_name);
        exit(EXIT_FAILURE);
    }

    printf("Shared memory created successfully, shm_fd = %d, total size = %zu bytes\n", shm_fd, total_size);
}

uint16_t checksum(char *addr, uint32_t count)
{
    register uint32_t sum = 0;
    uint16_t *buf = (uint16_t *)addr;

    // 主求和循环
    while(count > 1)
    {
        sum += *(buf)++;
        count -= 2;
    }

    // 处理剩余的单个字节
    if (count > 0)
        sum += *addr;

    // 将32位和折叠为16位
    while (sum >> 16)
        sum = (sum & 0xFFFF) + (sum >> 16);

    return (~sum);
}

Item CreateItem(){
    Item item;
    time_t seconds = time(NULL);

    item.timeStamp = (uint32_t)seconds;
    // 生成随机数据
    for(int i = 0; i < 22; ++i){
        item.data[i] = rand() % 256;
    }
    // 计算data的校验和（长度为22）
    item.checksum = checksum((char*)&item.data[0], 22);

    return item;
}

void* producer() {
    while(1) {
        Item tempItem = CreateItem();
        tempItem.seqNo = in;

        sem_wait(empty);
        pthread_mutex_lock(&mutex);

        // 将Item写入共享内存的对应位置
        memcpy(&shm_ptr[in * sizeof(Item)], &tempItem, sizeof(Item));
        printf("Producer: wrote item seqNo %d, checksum = %x\n", tempItem.seqNo, tempItem.checksum);

        in = (in + 1) % shmSize;

        pthread_mutex_unlock(&mutex);
        sem_post(full);
        sleep(1);
    }
}

void* consumer() {
    while(1) {
        sem_wait(full);
        pthread_mutex_lock(&mutex);

        Item tempItem;
        memcpy(&tempItem, &shm_ptr[out * sizeof(Item)], sizeof(Item));

        uint16_t cksum1 = tempItem.checksum;
        uint16_t cksum2 = checksum((char*)&tempItem.data[0], 22);

        if (cksum1 != cksum2) {
            printf("Checksum mismatch: expected %04x, received %04x (seqNo %d)\n", cksum2, cksum1, tempItem.seqNo);
        } else {
            printf("Consumer: verified item seqNo %d, checksums match\n", tempItem.seqNo);
        }

        out = (out + 1) % shmSize;

        pthread_mutex_unlock(&mutex);
        sem_post(empty);
        sleep(1);
    }
}

int main(int argc, char **argv){
    if (argc != 2) {
        fprintf(stderr, "Usage: %s <buffer_size>\n", argv[0]);
        return EXIT_FAILURE;
    }

    // 清理残留的信号量和共享内存
    sem_unlink("/empty");
    sem_unlink("/full");
    shm_unlink(shm_name);

    shmSize = atoi(argv[1]);
    if(shmSize <= 0){
        fprintf(stderr, "Error: Buffer size must be a positive integer.\n");
        return EXIT_FAILURE;
    }

    pthread_mutex_init(&mutex, NULL);
    empty = sem_open("/empty", O_CREAT | O_EXCL, 0644, shmSize);
    full = sem_open("/full", O_CREAT | O_EXCL, 0644, 0);

    if (empty == SEM_FAILED || full == SEM_FAILED) {
        fprintf(stderr, "Error creating semaphores, errno = %d (%s)\n", errno, strerror(errno));
        return EXIT_FAILURE;
    }

    CreateSharedMemory();

    // 初始化随机数种子
    srand((unsigned int)time(NULL));

    pthread_create(&producers, NULL, producer, NULL);
    pthread_create(&consumers, NULL, consumer, NULL);

    pthread_join(producers, NULL);
    pthread_join(consumers, NULL);

    // 清理资源（实际不会执行到这里，因为线程是无限循环）
    pthread_mutex_destroy(&mutex);
    sem_close(empty);
    sem_close(full);
    sem_unlink("/empty");
    sem_unlink("/full");
    munmap(shm_ptr, shmSize * sizeof(Item));
    shm_unlink(shm_name);

    return EXIT_SUCCESS;
}

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