c - 为什么我的程序在调用 sem_wait 时不等待？

Question

基本上，我的程序创建了 3 个线程。 “服务器”和 2 个“ worker ”。工作人员旨在对 1000 行文件(每个线程 500 个数字)中的 3 位正整数求和。在每个 worker 对自己的部分求和后，服务器打印每个 worker 的总数。唯一的问题是我的信号量似乎不起作用。

这是我的程序的一个版本:

// simple c program to simulate POSIX thread and semaphore
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <semaphore.h>
// define semaphores
sem_t s1;
FILE *file;
int sum1 = 0, sum2 = 0, num1 = 0, num2 = 0;
// file name
char fileName[10] = "data1.dat";
// server routine
void* server_routine()
{
    printf("Server sent signal to worker thread 1\n");
    printf("Server sent signal to worker thread 2\n");
    sem_wait(&s1);
    printf("Server recieved completion signal from worker thread 1\n");
    sem_wait(&s1);
    printf("Server recieved completion signal from worker thread 2\n\n");
    // print the final results
    printf("The sum of the first 500 numbers in the file is: %d\n", sum1);
    printf("The sum of the last 500 numbers in the file is: %d\n\n", sum2);
    pthread_exit(NULL);
}
// thread 1 reoutine
void* t1_routine()
{
    printf("Thread 1 recieved signal from server\n");
    file = fopen(fileName, "r");
    for(int i = 0; i < 500; i++)
    {
        fscanf(file, "%d", &num1);
        sum1 += num1;
    }
    printf("sum in thread 1: %d\n", sum1);
    printf("Thread 1 sends completion signal to server\n");
    sem_post(&s1);
    pthread_exit(NULL);
}
// thread 2 routine
void* t2_routine()
{
    printf("Thread 2 recieved signal from server\n");
    file = fopen(fileName, "r");
    fseek(file, 500 * 5, SEEK_SET);
    for(int i = 0; i < 500; i++)
    {
        fscanf(file, "%d", &num2);
        sum2 += num2;
    }
    printf("sum in thread 2: %d\n", sum2);
    printf("Thread 2 sends completion signal to server\n");
    sem_post(&s1);
    pthread_exit(NULL);
}
// main function
int main(int argc, char *argv[])
{
    // define threads
    pthread_t server, t1, t2;
    // initialize the semaphore
    sem_init(&s1, 0, 0);
    
    if(pthread_create(&server, NULL, &server_routine, NULL) != 0)
    {
        return 1;
    }

    if(pthread_create(&t1, NULL, &t1_routine, NULL) != 0)
    {
        return 2;
    }
    if(pthread_create(&t2, NULL, &t2_routine, NULL) != 0)
    {
        return 3;
    }

    if(pthread_join(server, NULL) != 0)
    {
        return 4;
    }

    if(pthread_join(t1, NULL) != 0)
    {
        return 5;
    }
    if(pthread_join(t2, NULL) != 0)
    {
        return 6;
    }
    // destroy semaphores
    sem_close(&s1);
    // exit thread
    pthread_exit(NULL);
    // end
    return 0;
}

我也用更少的线程测试了更多的信号量，但运气不佳。我尝试过不同的初始信号量值。我唯一可以获得正确输出的时间是手动等待 sleep(5);但这违背了这个项目的目的。

Answer 1

几个问题...

1. 每个客户端线程都有自己的/私有(private)的fopen 但是FILE *file; 是global 所以它们覆盖彼此重视。
2. 我们需要使这个变量具有函数作用域，这样每个线程都有自己的私有(private)指针。
3. 没有 fclose 调用。
4. pthread_exit 应该不 由主 线程完成。它仅适用于使用 pthread_create 创建的线程。

否则……

1. 执行fopen last 的线程将设置最终值。
2. 因此，存在竞争条件，效果与 main 线程(在 pthread_create 调用之前)完成了单个 fopen，违背了每个线程执行自己的 fopen 的目的。
3. 更糟糕的是，给定的线程可能首先 fopen，然后从fscanf 开始并拥有它的文件 当 second 线程替换 file 值时值发生变化，因此每个线程都会发生奇怪的事情，因为它们正在 fseek/fscanf < em>相同 FILE * 实例。
4. 进行上述 fclose 调用会使问题更加明显。

---

这是重构后的代码。它被注释:

// simple c program to simulate POSIX thread and semaphore
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <semaphore.h>

// define semaphores
sem_t s1;
// NOTE/BUG: each thread opens a different stream, so this must be function
// scoped
#if 0
FILE *file;
#endif
int sum1 = 0,
    sum2 = 0,
    num1 = 0,
    num2 = 0;

// file name
char fileName[10] = "data1.dat";

// server routine
void *
server_routine()
{
    printf("Server sent signal to worker thread 1\n");
    printf("Server sent signal to worker thread 2\n");
    sem_wait(&s1);

    printf("Server recieved completion signal from worker thread 1\n");
    sem_wait(&s1);
    printf("Server recieved completion signal from worker thread 2\n\n");

    // print the final results
    printf("The sum of the first 500 numbers in the file is: %d\n", sum1);
    printf("The sum of the last 500 numbers in the file is: %d\n\n", sum2);
    pthread_exit(NULL);
}

// thread 1 reoutine
void *
t1_routine()
{
// NOTE/FIX: this must be function scoped (i.e. private to this thread)
#if 1
    FILE *file;
#endif
    printf("Thread 1 recieved signal from server\n");
    file = fopen(fileName, "r");
    for (int i = 0; i < 500; i++) {
        fscanf(file, "%d", &num1);
        sum1 += num1;
    }
    printf("sum in thread 1: %d\n", sum1);
    printf("Thread 1 sends completion signal to server\n");
    sem_post(&s1);
#if 1
    fclose(file);
#endif
    pthread_exit(NULL);
}

// thread 2 routine
void *
t2_routine()
{
// NOTE/FIX: this must be function scoped (i.e. private to this thread)
#if 1
    FILE *file;
#endif
    printf("Thread 2 recieved signal from server\n");
    file = fopen(fileName, "r");
    fseek(file, 500 * 5, SEEK_SET);
    for (int i = 0; i < 500; i++) {
        fscanf(file, "%d", &num2);
        sum2 += num2;
    }
    printf("sum in thread 2: %d\n", sum2);
    printf("Thread 2 sends completion signal to server\n");
    sem_post(&s1);
#if 1
    fclose(file);
#endif
    pthread_exit(NULL);
}

// main function
int
main(int argc, char *argv[])
{
    // define threads
    pthread_t server, t1, t2;

    // initialize the semaphore
    sem_init(&s1, 0, 0);

    if (pthread_create(&server, NULL, &server_routine, NULL) != 0) {
        return 1;
    }

    if (pthread_create(&t1, NULL, &t1_routine, NULL) != 0) {
        return 2;
    }
    if (pthread_create(&t2, NULL, &t2_routine, NULL) != 0) {
        return 3;
    }

    if (pthread_join(server, NULL) != 0) {
        return 4;
    }

    if (pthread_join(t1, NULL) != 0) {
        return 5;
    }
    if (pthread_join(t2, NULL) != 0) {
        return 6;
    }

    // destroy semaphores
    sem_close(&s1);

    // exit thread
// NOTE/BUG: only a subthread should do this
#if 0
    pthread_exit(NULL);
#endif

    // end
    return 0;
}

---

在上面的代码中，我使用了 cpp 条件来表示旧代码与新代码:

#if 0
// old code
#else
// new code
#endif

#if 1
// new code
#endif

注意:这可以通过 unifdef -k

运行文件来清除

---

更新:

Thank you for the response Craig. I have implemented your suggestions to my own code but nothing seemed to change. I then decided to copy paste your updated code into a c file to test it out and I got the same result. It is as follows (in a separate comment since the output is too long): – Max

很难比较结果，因为我们使用的是不同的数据集。我创建了一个 perl 脚本来创建一些数据。

最重要的是给定工作人员报告的总和与服务器看到的该工作人员任务的总和相匹配。

然后，如果我们知道文件的每个线程部分的总和应该是多少，那就是另一回事了。

每行终止是关键的(例如):CRLF 与 LF(见下文)

worker sem_post 和终止的实际顺序并不重要。它可以因系统而异，甚至可以因调用而异。重要的是服务器线程在打印任何总和之前等待 N 个线程(即)N 个 sem_wait 调用。

我在下面制作了一个更新版本。

1. 服务器不向工作人员发出“信号”。工作人员通过执行 sem_post 向服务器“发送信号”，服务器通过执行 sem_wait
来“接收”它
2. 我已经创建了一个任务/线程 struct 来保存总和、线程 ID 等。
3. 我已将代码概括为允许 N 个线程。
4. 添加了对 \n 位置的检查(即线宽)。也就是说，在 linux/POSIX 下，四位数字后跟 LF(换行符)，长度为 5。但是，在 Windows 下，它将是 CRLF(回车/换行符)，长度为 6。
5. 添加了文件大小检查以确保它恰好是所需/预期的长度。
6. 一些额外的诊断。

这是更新后的代码:

// simple c program to simulate POSIX thread and semaphore
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <semaphore.h>
#include <sys/stat.h>

// number of bytes per line
// 5: 4 digits + LF
// 6: 4 digits + CRLF
#ifndef LINEWID
#define LINEWID     (4 + 1)
#endif

// number of items / task
#ifndef COUNT
#define COUNT       500
#endif

// define semaphores
sem_t s1;

#if 0
int sum1 = 0,
    sum2 = 0,
    num1 = 0,
    num2 = 0;
#endif

// file name
#if 0
char fileName[10] = "data1.dat";
#else
const char *fileName = "data1.dat";
#endif

// task control
typedef struct {
    pthread_t tid;                  // thread ID
    int tno;                        // thread index/offset
    int sum;                        // sum
} tsk_t;

#define TSKMAX  50
int tskmax;                         // actual number of tasks
tsk_t tsklist[TSKMAX];              // list of tasks

// loop through all task blocks
#define TSKFORALL \
    tsk_t *tsk = &tsklist[0];  tsk < &tsklist[tskmax];  ++tsk

// server routine
void *
server_routine(void *vp)
{
// NOTE/BUG: server does _not_ signal worker
#if 0
    printf("Server sent signal to worker thread 1\n");
    printf("Server sent signal to worker thread 2\n");
#endif

    for (TSKFORALL) {
        printf("Server waiting ...\n");
        sem_wait(&s1);
        printf("Server complete ...\n");
    }

    // print the final results
    for (TSKFORALL)
        printf("The sum of task %d is %d\n",tsk->tno,tsk->sum);

    return (void *) 0;
}

// thread 1 reoutine
void *
worker_routine(void *vp)
{
    FILE *file;
    tsk_t *tsk = vp;

    printf("Thread %d running ...\n",tsk->tno);

    file = fopen(fileName, "r");
    fseek(file,tsk->tno * COUNT * LINEWID,SEEK_SET);

    tsk->sum = 0;

    int num1;
    int first = -1;
    int last = -1;
    for (int i = 0; i < COUNT; i++) {
        if (fscanf(file, "%d", &num1) != 1) {
            printf("Thread %d fscan error\n",tsk->tno);
            break;
        }

        if (i == 0)
            first = num1;
        if (i == (COUNT - 1))
            last = num1;

        tsk->sum += num1;
    }

    printf("sum in thread %d: %d (first %d, last %d)\n",
        tsk->tno, tsk->sum, first, last);
    sem_post(&s1);

#if 1
    fclose(file);
#endif

    return (void *) 0;
}

// main function
int
main(int argc, char **argv)
{

    --argc;
    ++argv;

    setlinebuf(stdout);
    setlinebuf(stderr);

    if (argc != 1)
        tskmax = 2;
    else
        tskmax = atoi(*argv);

    if (tskmax > TSKMAX)
        tskmax = TSKMAX;

    // define threads
    pthread_t server;

    printf("main: %d tasks\n",tskmax);
    printf("main: %d count\n",COUNT);

    FILE *file = fopen(fileName,"r");
    if (file == NULL) {
        printf("main: fopen failure\n");
        exit(96);
    }

    // check alignment
    char chr;
    fseek(file,LINEWID - 1,0);
    fread(&chr,1,1,file);
    if (chr != '\n') {
        printf("main: newline mismatch -- chr=%2.2X\n",chr);
        exit(97);
    }

    // get the file size
    struct stat st;
    if (fstat(fileno(file),&st) < 0) {
        printf("main: fstat fault\n");
        exit(97);
    }

    // ensure the file has the correct size
    off_t size = tskmax * LINEWID * COUNT;
    if (st.st_size != size)
        printf("main: wrong file size -- st_size=%llu size=%llu\n",
            (unsigned long long) st.st_size,
            (unsigned long long) size);

    fclose(file);

    // initialize the semaphore
    sem_init(&s1, 0, 0);

    // set the offsets
    int tno = 0;
    for (TSKFORALL, ++tno)
        tsk->tno = tno;

    if (pthread_create(&server, NULL, &server_routine, NULL) != 0)
        return 98;

    for (TSKFORALL) {
        if (pthread_create(&tsk->tid,NULL,worker_routine,tsk) != 0)
            return 1 + tsk->tno;
    }

    if (pthread_join(server, NULL) != 0) {
        return 99;
    }

    for (TSKFORALL) {
        if (pthread_join(tsk->tid, NULL) != 0) {
            return 5;
        }
    }

    // destroy semaphores
    sem_close(&s1);

    // end
    return 0;
}

这是我用来生成数据的 perl 脚本输出:

number of tasks 2
element count per task 500
line separater 0A
section 0 sum 124750
section 1 sum 125250

程序输出如下:

main: 2 tasks
Server waiting ...
Thread 0 running ...
Thread 1 running ...
sum in thread 1: 125250 (first 1, last 500)
sum in thread 0: 124750 (first 0, last 499)
Server complete ...
Server waiting ...
Server complete ...
The sum of task 0 is 124750
The sum of task 1 is 125250

这是 perl 脚本:

#!/usr/bin/perl
# gendata -- generate data
#
# arguments:
#   1 - number of tasks (DEFAULT: 2)
#   2 - number of items / task (DEFAULT: 500)
#   3 - line separater (DEFAULT: \n)

master(@ARGV);
exit(0);

# master -- master control
sub master
{
    my(@argv) = @_;

    $tskmax = shift(@argv);
    $tskmax //= 2;
    printf(STDERR "number of tasks %d\n",$tskmax);

    $count = shift(@argv);
    $count //= 500;
    printf(STDERR "element count per task %d\n",$count);

    $sep = shift(@argv);
    $sep //= "\n";
    printf(STDERR "line separater");
    foreach $chr (split(//,$sep)) {
        $hex = ord($chr);
        printf(STDERR " %2.2X",$hex);
    }
    printf(STDERR "\n");

    for ($itsk = 0;  $itsk < $tskmax;  ++$itsk) {
        $val = $itsk;
        $sum = 0;
        for ($lno = 1;  $lno <= $count;  ++$lno, ++$val) {
            printf("%4d%s",$val,$sep);
            $sum += $val;
        }
        printf(STDERR "section %d sum %d\n",$itsk,$sum);
    }
}