pthread C 中的池线程

Question

在我的 Advance OS 工作中，我需要从文件中读取数据并将其分配给线程以进行进一步处理..

这是我从文件中读取数据并将其传递给线程函数的代码

int main() {
FILE *fp = fopen(fileName, "r");
char str_pass[80][MAX_CHAR];

if (fp == NULL){
    printf("Could not open file");
    return 1;
}
int i=0;
pthread_t thread[THREADS];
int rc;

for(int th=1; th<=THREADS; th++)
{
if(fgets(str_pass[i], MAX_CHAR, fp) != NULL){
    //printf("Hello, Thread %d\n",i);

    rc = pthread_create(&thread[i], NULL, create_thread, (void*)str_pass[i]);
    pthread_join(thread[i],NULL);

    if(rc)
    {
        printf("ERROR; return code from pthread_create() is %d\n",rc);
        exit(-1);
    }
    pthread_join(thread[i],NULL);
    i++;
    }
else{
    printf("End of File");
    exit(0);
}
}
pthread_exit(NULL);
fclose(fp);
return 0;
}

这是我的线程代码；

void * create_thread(void *hash_string)
{

    gen_combinations(hash_string);

    //sleep(1);
    pthread_exit(NULL);
    return NULL;
}

这段代码运行良好，它创建的线程与我在THREADS 变量中定义的值一样多，除非它在文件中找不到任何记录。但现在我必须用线程池概念来做这件事。因为我无法生成与文件中的数据一样多的线程。

所以我需要使用线程池来实现多线程。我做了 对其进行了一些搜索，但没有获得任何许可。而现在我 完全卡在这里，不知道从哪里开始， 这个工作怎么做？？？

我们将不胜感激。

Answer 1

如果您还不熟悉“生产者/消费者问题”，特别是多个消费者的变化，请查看它。您的主线程扮演生产者的角色，从输入文件中读取数据和参数，并将它们打包成整齐的工作单元——这些是正在生产的小部件。工作线程扮演消费者的角色，接受工作单元并通过实际执行所描述的工作来“消费”它们。那是一个线程池。

您的特定实现可能不需要是通用的。它可以被适当地设计和调整以专门解决手头的问题。

Answer 2

rc = pthread_create(&thread[i], NULL, create_thread, (void*)str_pass[i]);
pthread_join(thread[i],NULL);

if(rc)
{
    printf("ERROR; return code from pthread_create() is %d\n",rc);
    exit(-1);
}
pthread_join(thread[i],NULL);

这里有两个主要错误：

1. 在创建线程后立即调用pthread_join。这意味着您等待线程完成。那么创建线程有什么意义呢？

2. 一旦pthread_join 返回，线程不再存在，其 ID 现在无效。然而，您再次调用pthread_join 并将之前的ID 传递给它，但现在是垃圾。这样做的后果是完全无法预测的。

Answer 3

在您的示例中，您一次只有 一个 线程处于活动状态。您启动它并等待它结束，然后再从文件中读取下一行。它是由线程创建之后的 pthread_join 引起的。相反，您需要将 join 语句移出循环以加入您创建的所有线程。

现在，您可以在不创建池的情况下限制线程数。您只需要一个原子计数器，您将在线程启动之前递增并在线程完成时递减。这样，您可以在创建下一个线程之前检查计数器的值并等待条件变量。这将是一种信号量。

大致如下：

counter = 0;

while {
   lock counter;
   while (counter > 8)
      wait_on_conditional_variable
   counter ++;
   unlock counter;
   run thread;
}
join all remaining threads.

在线程中

    do work;
    lock counter;
    counter --;
    signal cond var;
    unlock counter;
    return;

对于池，您需要启动多个线程。每个线程都有一个循环，它等待某个作业可用。很可能是从fifo那里得到的。它必须等待一个条件变量来检查。

thread
    do {
        lock fifo;
        while (fifo.size == 0) 
           wait on conditional variable.
        read job from fifo;
        unlock fifo;
        do work;

    } while (!exiting);

在阅读文件时，应完成以下操作

    while ... {
         lock fifo;
         push line into fifo;
         signal var;
         unlock fifo;
    }
    set-exit-condition;
    join the pools.

我希望这会有所帮助。 但是您可以在那里进行多种方式和优化。