多客户端Socket服务器Semaphore阻塞失效及客户端异常断开问题排查

首先，你的代码里有几个关键问题导致同步机制失效和客户端异常断开，我来一步步拆解并给出修复方案：

一、为什么Mutex/Semaphore完全不生效？ #

1. 全局变量state根本没法跨进程共享 #

你用fork()给每个客户端创建子进程，但子进程会复制父进程的整个地址空间——也就是说每个客户端的子进程里都有一个独立的state副本。当客户端A发送"BEG"把自己的state设为1时，客户端B的state还是0，完全感知不到这个变化，阻塞逻辑自然触发不了。

2. 信号量的逻辑完全错位 #

你现在的代码里，只有当state !=1时才对semID2做down/up，但state本身不共享，这个判断等于没用。而且你需要的是"排他访问"：发起BEG的客户端独占资源，其他客户端只能等END或者quit，但当前的信号量逻辑根本没绑定到全局共享的状态上，甚至如果semGET()创建的是进程私有信号量，子进程之间根本看不到同一个信号量。

二、客户端为啥会异常断开？ #

看你的main.c子进程循环里，每次处理完指令都调用了close(rfd);——rfd是服务器的监听套接字啊！子进程把监听套接字关了，后续的客户端连接会直接失败，而且当前客户端的连接也会因为这个错误的关闭操作提前断开。这个语句必须删掉！

三、具体修复方案 #

1. 用共享内存存储全局状态 #

既然子进程地址空间独立，就得把state（还有记录发起BEG的进程ID）放到共享内存里，让所有子进程都能访问同一个变量：

• 用shmget()创建共享内存段，shmat()映射到每个子进程的地址空间。
• 替换原来的全局int state为共享内存中的指针。

2. 重新设计信号量同步逻辑 #

你需要两个信号量：

• 一个互斥锁信号量：保护共享内存中state和beg_pid的读写，避免多个进程同时修改导致竞态。
• 一个阻塞信号量：当state=1时，让非发起BEG的客户端阻塞等待，直到发起者发送END。

3. 修正子进程的处理流程 #

删掉子进程里的close(rfd);，同时在处理指令时：

• 先检查是否是quit指令，是的话直接退出，同时如果当前是发起BEG的进程，要先把state设为0并唤醒其他客户端。
• 非quit指令先检查共享state：如果是1且自己不是发起者，就阻塞在信号量上；发起者则可以正常执行。

四、怎么验证信号量是否正常工作？ #

• 打印调试：在down()和up()函数里，打印当前进程ID和信号量的当前值（用semctl(semID, 0, GETVAL)获取），观察当客户端A发送BEG后，客户端B执行PUT/GET时是否会卡在down()上。
• 分步测试：
  1. 启动服务器，连接客户端A，发送BEG，确认服务器返回提示。
  2. 连接客户端B，发送PUT test 123，观察客户端B是否会卡住（没有响应）。
  3. 客户端A发送END，此时客户端B的指令应该被执行，收到响应。
  4. 客户端B发送quit，应该能正常退出，不受任何阻塞影响。

五、修正后的核心代码片段 #

共享内存与信号量初始化（main.c） #

// 创建共享内存存储state和发起BEG的PID
int shm_id = shmget(IPC_PRIVATE, sizeof(int) + sizeof(pid_t), IPC_CREAT | 0666);
void *shm_ptr = shmat(shm_id, NULL, 0);
int *shared_state = (int*)shm_ptr;
pid_t *beg_pid = (pid_t*)((char*)shm_ptr + sizeof(int));
*shared_state = 0;
*beg_pid = -1;

// 互斥锁信号量（保护共享内存读写）
semaphor mutex_sem = semGET();
unsigned short mutex_marker[1] = {1};
semctl(mutex_sem, 0, SETALL, mutex_marker);

// 阻塞信号量（让非发起者等待）
semaphor block_sem = semGET2();
unsigned short block_marker[1] = {1};
semctl(block_sem, 0, SETALL, block_marker);

子进程处理逻辑修正 #

if (pid == 0) {
    bzero(input, sizeof(input));
    bytes_read = read(cfd, input, BUFSIZE);
    strncat(input, " ", strlen(" "));
    input[strcspn(input, "\r\n")] = 0;
    
    pid_t my_pid = getpid();
    
    while (bytes_read > 0) {
        eingabe = strtok(input, delimiter);
        int i = 0;
        while (eingabe != NULL) {
            inputBuffer[i++] = eingabe;
            eingabe = strtok(NULL, delimiter);
        }
        
        // 处理quit指令
        if (strncmp("quit", inputBuffer[0], 4) == 0) { 
            close(cfd); 
            // 如果是发起BEG的进程，清理状态
            down(mutex_sem, 0);
            if (*shared_state == 1 && *beg_pid == my_pid) {
                *shared_state = 0;
                *beg_pid = -1;
                up(block_sem, 0); // 唤醒等待的客户端
            }
            up(mutex_sem, 0);
            break; 
        }
        
        // 检查是否需要阻塞
        down(mutex_sem, 0);
        int current_state = *shared_state;
        pid_t current_beg_pid = *beg_pid;
        up(mutex_sem, 0);
        
        if (current_state == 1 && current_beg_pid != my_pid) {
            // 非发起者，阻塞等待
            down(block_sem, 0);
            // 唤醒后重新获取状态
            down(mutex_sem, 0);
            current_state = *shared_state;
            up(mutex_sem, 0);
        }
        
        // 执行指令
        down(mutex_sem, 0);
        conditions(inputBuffer[0], inputBuffer[1], inputBuffer[2], cfd, shmID, shared_state, beg_pid, my_pid, mutex_sem, block_sem);
        up(mutex_sem, 0);
        
        bzero(input, sizeof(input));
        bytes_read = read(cfd, input, BUFSIZE);
        strncat(input, " ", strlen(" "));
        input[strcspn(input, "\r\n")] = 0;
        
        // 删掉这个错误的close(rfd)!
        // close(rfd);
    }
    
    // 子进程退出前解除共享内存映射
    shmdt(shm_ptr);
    exit(0);
}

修改后的conditions函数 #

void conditions(char *eingabehalter1, char *eingabehalter2, char *eingabehalter3, int cfd, int shmID, int *shared_state, pid_t *beg_pid, pid_t my_pid, semaphor mutex_sem, semaphor block_sem) {
    if (strncmp("PUT", eingabehalter1, 3) == 0) { 
        put(eingabehalter2, eingabehalter3, cfd, shmID); 
    } else if (strncmp("GET", eingabehalter1, 3) == 0) { 
        get(eingabehalter2, cfd, shmID); 
    } else if (strncmp("DEL", eingabehalter1, 3) == 0) { 
        del(eingabehalter2, cfd, shmID); 
    } else if (strncmp("BEG", eingabehalter1, 3) == 0) { 
        *shared_state = 1;
        *beg_pid = my_pid;
        write(cfd, "BEG accepted: other clients blocked\n", strlen("BEG accepted: other clients blocked\n"));
    } else if (strncmp("END", eingabehalter1, 3) == 0) { 
        if (*beg_pid == my_pid) {
            *shared_state = 0;
            *beg_pid = -1;
            up(block_sem, 0);
            write(cfd, "END accepted: other clients unblocked\n", strlen("END accepted: other clients unblocked\n"));
        } else {
            write(cfd, "Error: Only BEG initiator can send END\n", strlen("Error: Only BEG initiator can send END\n"));
        }
    } else { 
        write(cfd, "Error: Unknown command\n", strlen("Error: Unknown command\n")); 
    }
}

六、额外注意点 #

• 确保semGET()创建的是系统共享信号量（比如用IPC_PRIVATE或者固定键值），不能是进程私有信号量，否则子进程之间看不到同一个信号量。
• 服务器退出时，要清理共享内存和信号量：用shmctl(shm_id, IPC_RMID, NULL)和semctl(mutex_sem, 0, IPC_RMID)、semctl(block_sem, 0, IPC_RMID)，避免资源泄漏。
• 处理read()返回0的情况（客户端正常关闭连接），避免无限循环。

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