使用 C 中的 pthread 保持线程数恒定

Question

我试图找到一个解决方案，以便使用 pthreads 在 C 中的 linux 下保持工作线程数恒定，但我似乎无法完全理解以下代码有什么问题：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

#define MAX_JOBS 50
#define MAX_THREADS 5

pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
int jobs = MAX_JOBS;
int worker = 0;
int counter = 0;

void *functionC() {
  pthread_mutex_lock(&mutex1);
  worker++;
  counter++;
  printf("Counter value: %d\n",counter);
  pthread_mutex_unlock(&mutex1);

  // Do something...
  sleep(4);

  pthread_mutex_lock(&mutex1);
  jobs--;
  worker--;
  printf(" >>> Job done: %d\n",jobs);
  pthread_mutex_unlock(&mutex1);
}

int main(int argc, char *argv[]) {
  int i=0, j=0;
  pthread_t thread[MAX_JOBS];

  // Create threads if the number of working threads doesn't exceed MAX_THREADS
  while (1) {
    if (worker > MAX_THREADS) {
      printf(" +++ In queue: %d\n", worker);
      sleep(1);
    } else {
      //printf(" +++ Creating new thread: %d\n", worker);
      pthread_create(&thread[i], NULL, &functionC, NULL);
      //printf("%d",worker);
      i++;
    }
    if (i == MAX_JOBS) break;
  }

  // Wait all threads to finish
  for (j=0;j<MAX_JOBS;j++) {
    pthread_join(thread[j], NULL);
  }

  return(0);
}

A while (1) 循环不断创建线程，如果工作线程数低于一定阈值。每次工作线程的全局计数器递增（线程创建）和递减（作业完成）时，互斥体应该锁定关键部分。我认为它可以很好地工作，并且在大多数情况下都可以，但是奇怪的事情发生了......

例如，如果我评论（如本片段中所示） printf //printf(" +++ 创建新线程： % d\n"，工人）； while (1) 似乎一次生成一个随机数（根据我的经验为 18-25）线程（functionC 打印出“计数器值：从 1 到 18-25”...），而不是尊重 IF 条件环形。如果我包含 printf，则循环似乎“几乎”以正确的方式运行...这似乎暗示缺少“互斥”条件，我应该将其添加到 main() 中的循环中，以便在 MAX_THREADS 时有效锁定线程已达到，但在过去几天更改此代码很多次之后，我现在有点迷失了。我缺少什么？

请让我知道我应该更改什么以保持线程数恒定，似乎我距离解决方案并不太远...希望...:-)

提前致谢！

Answer 1

您的问题是，在新线程实际启动并开始运行之前，worker 不会递增 - 与此同时，主线程循环，检查workers，发现它没有没有改变，并启动另一个线程。它可以重复此操作多次，从而创建太多线程。

因此，当您决定创建一个新线程时，您需要在主线程中增加 worker 。

您还有另一个问题 - 您应该使用条件变量让主线程休眠，直到它启动另一个线程，而不是使用其中带有 sleep(1); 的忙等待循环。完整的固定代码如下所示：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#define MAX_JOBS 50
#define MAX_THREADS 5

pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond1 = PTHREAD_COND_INITIALIZER;
int jobs = MAX_JOBS;
int workers = 0;
int counter = 0;

void *functionC() {
  pthread_mutex_lock(&mutex1);
  counter++;
  printf("Counter value: %d\n",counter);
  pthread_mutex_unlock(&mutex1);

  // Do something...
  sleep(4);

  pthread_mutex_lock(&mutex1);
  jobs--;
  printf(" >>> Job done: %d\n",jobs);

  /* Worker is about to exit, so decrement count and wakeup main thread */
  workers--;
  pthread_cond_signal(&cond1);

  pthread_mutex_unlock(&mutex1);
  return NULL;
}

int main(int argc, char *argv[]) {
  int i=0, j=0;
  pthread_t thread[MAX_JOBS];

  // Create threads if the number of working threads doesn't exceed MAX_THREADS
  while (i < MAX_JOBS) {
    /* Block on condition variable until there are insufficient workers running */
    pthread_mutex_lock(&mutex1);
    while (workers >= MAX_THREADS)
        pthread_cond_wait(&cond1, &mutex1);

    /* Another worker will be running shortly */
    workers++;
    pthread_mutex_unlock(&mutex1);

    pthread_create(&thread[i], NULL, &functionC, NULL);
    i++;
  }

  // Wait all threads to finish
  for (j=0;j<MAX_JOBS;j++) {
    pthread_join(thread[j], NULL);
  }

  return(0);
}

请注意，尽管这有效，但它并不理想 - 最好预先创建您想要的线程数量，并让它们循环，等待工作。这是因为创建和销毁线程会产生很大的开销，而且它通常会简化资源管理。重写后的代码版本如下所示：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#define MAX_JOBS 50
#define MAX_THREADS 5

pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
int jobs = MAX_JOBS;
int counter = 0;

void *functionC()
{
  int running_job;

  pthread_mutex_lock(&mutex1);
  counter++;
  printf("Counter value: %d\n",counter);

  while (jobs > 0) {
    running_job = jobs--;
    pthread_mutex_unlock(&mutex1);

    printf(" >>> Job starting: %d\n", running_job);

    // Do something...
    sleep(4);

    printf(" >>> Job done: %d\n", running_job);

    pthread_mutex_lock(&mutex1);
  }
  pthread_mutex_unlock(&mutex1);

  return NULL;
}

int main(int argc, char *argv[]) {
  int i;
  pthread_t thread[MAX_THREADS];

  for (i = 0; i < MAX_THREADS; i++)
    pthread_create(&thread[i], NULL, &functionC, NULL);

  // Wait all threads to finish
  for (i = 0; i < MAX_THREADS; i++)
    pthread_join(thread[i], NULL);

  return 0;
}