c++ - 使线程循环 5 次迭代； pthreads、互斥量和信号量

Question

我的类(class)示例中有这段代码，老师的指示是“让每个线程循环 5 次迭代”。我对如何使用以下代码执行此操作感到困惑:

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <iostream>
#include <iomanip>
#include <sstream>
#include <sys/utsname.h>

/* Symbolic Constants*/
#define NUM_THREADS 4
#define BUFFER_SIZE 10

/* Semaphore and Mutex lock */
sem_t cEmpty;
sem_t cFull;
pthread_mutex_t mutex;

/* Threads */
pthread_t tid; /* Thread ID */
pthread_attr_t attr; /* Thread attributes */

//prototypes
void *producer(void *param);
void *consumer(void *param);
int insert_item(int threadID);
int remove_item(int threadID);
void init();

/* Progress Counter and Thread IDs*/
int counter, pthreadID=0, cthreadID=0;

int main()
{
  /* Variables */
  int c1;

  /* Perform initialization */
  init();

  /* Create the producer threads */
  for(c1=0; c1<NUM_THREADS; c1++)
  {
    pthread_create(&tid, &attr, producer, NULL);
  }

  /* Create the consumer threads */
  for(c1=0; c1<NUM_THREADS; c1++)
  {
    pthread_create(&tid, &attr, consumer, NULL);
  }

  /* Ending it */
  sleep(2);

  printf("All threads are done.\n");

  /* Destroy the mutex and semaphors */
  pthread_mutex_destroy(&mutex);
  sem_destroy(&cEmpty);
  sem_destroy(&cFull);

  printf("Resources cleaned up.\n");

  exit(0);
}

void init()
{
  pthread_mutex_init(&mutex, NULL); /* Initialize mutex lock */
  pthread_attr_init(&attr); /* Initialize pthread attributes to default */
  sem_init(&cFull, 0, 0); /* Initialize full semaphore */
  sem_init(&cEmpty, 0, BUFFER_SIZE); /* Initialize empty semaphore */
  counter = 0; /* Initialize global counter */
}

void *producer(void *param)
{
  int x;
  for(x=0; x<5;)
  {
    sleep(1);

    sem_wait(&cEmpty); /* Lock empty semaphore if not zero */
    pthread_mutex_lock(&mutex);

    if(insert_item(pthreadID))
    {
      fprintf(stderr, "Producer error.");
    }
    else
    {
      pthreadID++;
      x++;
    }

    pthread_mutex_unlock(&mutex);
    sem_post(&cFull); /* Increment semaphore for # of full */
  }
return 0;
}

void *consumer(void *param)
{
  int y;
  for(y=0; y<5;)
  {
    sleep(1);

    sem_wait(&cFull); /* Lock empty semaphore if not zero */
    pthread_mutex_lock(&mutex);

    if(remove_item(cthreadID))
    {
      fprintf(stderr, "Consumer error.");
    }
    else
    {
      cthreadID++;
      y++;
    }
    pthread_mutex_unlock(&mutex);
    sem_post(&cEmpty); /* Increments semaphore for # of empty */
  }
return 0;
}

int insert_item(int threadID)
{
  if(counter < BUFFER_SIZE) /* Buffer has space */
  {
    counter++;
    printf("Producer %d inserted a cookie. Total:%d\n", threadID, counter);
    return 0;
  }
  else /* Buffer full */
  {
    return -1;
  }
}

int remove_item(int threadID)
{
  if(counter > 0) /* Buffer has something in it */
  {
    counter--;
    printf("Consumer %d removed a cookie. Total:%d\n", threadID, counter);
    return 0;
  }
  else /* Buffer empty */
  {
    return -1;
  }
}

有人知道我在哪里添加 for 循环以“使每个线程循环 5 次迭代”吗？非常感谢您。

更新:我将 while(1) 更改为具有 5 次迭代的 for 循环，但我仍然无法从 insert_item 和 remove_item 函数中获取消息来打印 5 次，仅打印一次。有人知道如何让它打印 5 次吗？

Answer 1

问题是在 main 的末尾，我调用了 sleep(2)。这不足以让所有线程打印其输出。我也没有将正确的索引传递给我的 add_item 和 remove_item 函数。此外，我需要所有线程的连接命令而不是 sleep 命令，并且连接命令确保所有线程在程序退出之前完成。这是更新和更正的代码。希望这可以帮助尝试做类似事情的人!

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <iostream>
#include <iomanip>
#include <sstream>
#include <sys/utsname.h>

// Symbolic Constants
#define NUM_THREADS 4
#define BUFFER_SIZE 10

// Semaphore and Mutex lock
sem_t cEmpty;
sem_t cFull;
pthread_mutex_t mutex;

// Threads
pthread_t tid[NUM_THREADS]; //Thread ID
pthread_t tid2[NUM_THREADS]; //Thread ID
pthread_attr_t attr; //Thread attributes

//prototypes
void *producer(void *param);
void *consumer(void *param);
int insert_item(long threadID);
int remove_item(long threadID);
void init();

//Progress Counter and Thread IDs
int counter=0;

int main()
{
  //Variables
  long c1;

  //Perform initialization
  init();

  //Create the producer threads
  for(c1=0; c1<NUM_THREADS; c1++)
  {
    pthread_create(&tid[c1], &attr, producer, (void *)c1);
    pthread_create(&tid2[c1], &attr, consumer, (void *)c1);
  }

  //Ending it
  for(c1=0; c1<NUM_THREADS; c1++)
  {
    pthread_join(tid[c1], NULL);
    pthread_join(tid2[c1],NULL);
  }

  printf("All threads are done.\n");

  //Destroy the mutex and semaphors
  pthread_mutex_destroy(&mutex);
  sem_destroy(&cEmpty);
  sem_destroy(&cFull);

  printf("Resources cleaned up.\n");

  exit(0);
}

//This function performs initialization
void init()
{
  pthread_mutex_init(&mutex, NULL); //Initialize mutex lock
  pthread_attr_init(&attr); //Initialize pthread attributes to default
  sem_init(&cFull, 0, 0); //Initialize full semaphore
  sem_init(&cEmpty, 0, BUFFER_SIZE); //Initialize empty semaphore
  counter = 0; //Initialize global counter
}

//This function creates the producer thread
void *producer(void *param)
{
  long index = (long)param;

  for(int x = 0; x<5; x++)
  {
    sleep(1);

    sem_wait(&cEmpty); //Lock empty semaphore if not zero
    pthread_mutex_lock(&mutex);

    //check to see if item inserted correctly; print error on fail
    if(insert_item(index))
    {
      fprintf(stderr, "Producer error.");
    }

    pthread_mutex_unlock(&mutex);
    sem_post(&cFull); //Increment semaphore for # of full
  }
pthread_exit(NULL);
return 0;
}

//This function created the consumer thread
void *consumer(void *param)
{
  long index = (long)param;

  for(int x = 0; x<5; x++)
  {
    sleep(1);

    sem_wait(&cFull); //Lock empty semaphore if not zero
    pthread_mutex_lock(&mutex);

    //print error if cookie not decremented correctly
    if(remove_item(index))
    {
      fprintf(stderr, "Consumer error.");
    }

    pthread_mutex_unlock(&mutex);
    sem_post(&cEmpty); //Increments semaphore for # of empty
  }
pthread_exit(NULL);
return 0;
}
//Insert item function to increment the cookie count and print thread message
int insert_item(long threadID)
{
  if(counter < BUFFER_SIZE) //Buffer has space
  {
    counter++;
    printf("Producer %ld inserted a cookie. Total:%d\n", threadID, counter);
    return 0;
  }
  else //Buffer full
  {
    return -1;
  }
}

//Remove item function to decrement the cookie count and print thread message
int remove_item(long threadID)
{
  if(counter > 0) //Buffer has something in it
  {
    counter--;
    printf("Consumer %ld removed a cookie. Total:%d\n", threadID, counter);
    return 0;
  }
  else //Buffer empty
  {
    return -1;
  }
}