使用 pthreads 逐行读取文件...意外退出

Question

我有以下代码：

    /*//not important
    FILE * INFILE;
    list_file = optarg;
    if( ( INFILE = fopen( list_file, "a+" ) ) == NULL ) {
        fprintf( stderr, "Can't open input file\n");
        exit(0);
    }
    */

    pthread_mutex_t input_queue;
    pthread_mutex_init(&input_queue, NULL);

    for( i = 0 ; i < number_thread; i++)
    {
        if( pthread_create( &thread_id[i], NULL, &work, NULL) != 0 )
        {
            i--;
            fprintf(stderr, RED "\nError in creating thread\n" NONE);
        }
    }
    for( i = 0 ; i < number_thread; i++)
        if( pthread_join( thread_id[i], NULL) != 0 )
        {
            fprintf(stderr, RED "\nError in joining thread\n" NONE);
        }




    void * work(void * data)
    {
        unsigned long line;
        char buf[512];
        while ( !feof(INFILE) )
        {
            pthread_mutex_lock(&input_queue);
            fgets((char *)&buf, sizeof(buf), INFILE);
            if (buf[strlen (buf) - 1] == '\n')
                buf[strlen (buf) - 1] = '\0';
            line = (unsigned long)buf;
            pthread_mutex_unlock(&input_queue);
            do_work( line );
        }
        fclose(INFILE);
        return NULL;
    }

它从文件中读取行，但一段时间后它意外退出，没有错误消息。 我想我搞砸了。

如何使用 pthreads 逐行读取文件，同时尽可能保持代码不变（我的意思是不要搞乱整个程序）？

Answer 1

您将在遇到 EOF 的第一个线程中关闭 INFILE。之后其他线程将在关闭的文件上调用feof()——可能还有fclose()——这将破坏堆并且几乎肯定会导致崩溃。此外，您的换行代码可能会在 EOF 处耗尽您的缓冲区，请参阅下面的备注。

要解决此问题，请使用相同的互斥锁保护 feof() 和 fclose()，并将 INFILE 设置为 NULL。获取互斥体后，检查 INFILE 是否为 NULL，如果是则立即返回：

for (;;) {
  pthread_mutex_lock(&input_queue);
  if (!INFILE) {
    pthread_mutex_unlock(&input_queue);
    break;
  }
  if (feof(INFILE)) {
    INFILE = NULL;
    pthread_mutex_unlock(&input_queue);
    break;
  }

  fgets(buf, sizeof(buf), INFILE);
  pthread_mutex_unlock(&input_queue);

  // ...strip newline, do_work...
}

几点说明：

• 您的代码写入buf[strlen(buf) - 1] 而不检查strlen(buf) 是否为零。 buf 在 EOF 时将为空，因此这不是理论上的问题，每次执行都会发生一次。

• line 的类型为 unsigned long，但您为其分配了一个指针值。这将在long 不包含指针的平台上失败，例如 Win64。将line 和do_work 的参数声明为char *（或void *，如果它必须接受其他指针类型）。

• 避免将互斥体称为“队列”；在多线程编程队列中指的是producer-consumer aware FIFO。

• 您不需要使用互斥锁来保护单独的 stdio 函数，例如 fgets。根据 POSIX 的要求，它们是 MT 安全的。 （但是，在我修改后的代码中，fgets() 确实需要受互斥体保护，因为在互斥体未被持有时，INFILE 可能会失效。）

• (char *) &buf 没有意义。由于buf 是一个char 数组，它已经衰减为指向其第一个成员的指针，因此您可以简单地将buf 发送到fgets。如果你坚持使用address-of操作符，正确的表达式是&buf[0]。

• 正如 Carl Norum 所暗示的，feof() 可能不是您想要的，因为它只检测 fgets() 已经遇到的 EOF 条件。检查 EOF 的正确方法是测试 fgets() 是否返回空字符串 - 在剥离换行符之前。

Answer 2

如果 INFILE 是一个全局变量，那么你已经关闭了线程函数中的引用，如果你创建了多个线程，那么其他线程中的 flcose(INFILE) 即 fclose(NULL) 预计会崩溃。无法猜测您要对多个线程执行什么操作，但是当您确定 INFILE 将不再被任何其他线程访问时，最好在最后关闭它。我认为您应该在所有线程之后关闭 main 中的 INFILE 参考加入 main 并完成他们的处理。

#include<stdio.h>
#include<pthread.h>
#include<string.h>
#include<stdlib.h>
#define number_thread 10

FILE * INFILE;
char *list_file = "test_thread";
pthread_mutex_t input_queue;

void do_work(unsigned long buf)
{
    printf("working on %u\n",buf);
}

void * work(void * data)
{
    unsigned long line;
    char buf[512];
    printf("IAM NEW THREAD\n" );

    while ( !feof(INFILE) )
      {
        pthread_mutex_lock(&input_queue);
        fgets((char *)&buf, sizeof(buf), INFILE);
        if (buf[strlen (buf) - 1] == '\n')
            buf[strlen (buf) - 1] = '\0';
        line = (unsigned long)buf;
        pthread_mutex_unlock(&input_queue);
        do_work( line );
      }

    return NULL;
}

int main()
{
    printf("IAM MAIN THREAD\n")
    pthread_mutex_init(&input_queue, NULL);
    if( ( INFILE = fopen( list_file, "a+" ) ) == NULL ) {
        fprintf( stderr, "Can't open input file\n");
        exit(0);
    }
    pthread_t thread_id[10];

    int i=0;
    for( i = 0 ; i < number_thread; i++)
      { 
        if( pthread_create( &thread_id[i], NULL, &work, NULL) != 0 )
          {
            i--;
            fprintf(stderr,  "\nError in creating thread\n");
          }
      }

    for( i = 0 ; i < number_thread; i++)
        if( pthread_join( thread_id[i], NULL) != 0 )
          {
            fprintf(stderr,  "\nError in joining thread\n" );
          }

    fclose(INFILE);
}