生产者/消费者使用 Pthread 和信号量

Question

我目前正在学习如何使用 Pthreads 和 Semaphores，并且我一直在研究生产者/消费者问题的实现，但程序只是挂起。我知道它会到达消费者代码，运行一次比较，然后在默认初始化值比较 p1_string 和 p2_string 后挂起，我真的不明白我到底做错了什么。

基本上，每个生产者线程都应该获取一个已排序行的文件并将一行读入内存。然后主线程应该比较这两个字符串，并按排序顺序将它们写入输出。

#include <iostream>
#include <fstream>
#include <string>
#include <cstdlib>
#include <cstring>
#include <cerrno>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
using namespace std;

sem_t p1_empty,p2_empty,p1_full,p2_full;
string p1_string="atest", p2_string="btest";

typedef struct {
    char* filename;
    string buffer;
    sem_t empty;
    sem_t full;
} pthread_param;

void* producer(void* arg) {
  pthread_param* ptp = (pthread_param*)arg;
    ifstream input(ptp->filename);
    if (not input) {
        cerr << "Can't open file \"" << ptp->filename << "\".\n";
        exit(EXIT_FAILURE);
    }
    while(getline(input,ptp->buffer)) {
        sem_post(&ptp->full);
        sem_wait(&ptp->empty);
    }
    ptp->buffer = "\x7f";
}

int main(int argc, char* argv[]) {
    if (argc != 3) {
        cerr << "Syntax: " << argv[0] << " filename filename\n";
        exit(EXIT_FAILURE);
    }
    //init threads, variables and semaphores
    sem_init(&p1_empty,0,0);
    sem_init(&p2_empty,0,0);
    sem_init(&p1_full,0,0);
    sem_init(&p2_full,0,0);
    pthread_t p1_thread, p2_thread;
    pthread_param pt1_param;
    pthread_param pt2_param;
    pt1_param.filename = argv[1];
    pt2_param.filename = argv[2];
    pt1_param.buffer = p1_string;
    pt2_param.buffer = p2_string;
    pt1_param.empty = p1_empty;
    pt2_param.empty = p2_empty;
    pt1_param.full = p1_full;
    pt2_param.full = p2_full;
    pthread_create(&p1_thread,nullptr,producer,&pt1_param);
    pthread_create(&p2_thread,nullptr,producer,&pt2_param);

    /* testing to make sure producer reads correctly
  pthread_param* ptp = &pt1_param;
    ifstream input(ptp->filename);
    if (not input) {
        cerr << "Can't open file \"" << ptp->filename << "\".\n";
        exit(EXIT_FAILURE);
    }
    while(getline(input,ptp->buffer)) {
      cout<<ptp->buffer<<endl;
    }
    ptp->buffer = "\x7f";
 */

    //consumer
    while(pt1_param.buffer != "\x7f" && pt2_param.buffer != "\x7f"){
      if(pt1_param.buffer <= pt2_param.buffer) {
        cout<<pt1_param.buffer<<endl;
        sem_post(&p1_empty);
        sem_wait(&p1_full);
      }
      else {
        cout << pt2_param.buffer <<endl;
        sem_post(&p2_empty);
        sem_wait(&p2_full);
      }
    }

    //delete threads/semaphores
    pthread_join(p1_thread,nullptr);
    pthread_join(p2_thread,nullptr);
    sem_destroy(&p1_empty);
    sem_destroy(&p2_empty);
    sem_destroy(&p2_full);
    sem_destroy(&p2_full);
    return 0;
}

Answer 1

1) 字符串上的 = 运算符按值复制。以下代码按值复制到“缓冲区”变量中。但是，稍后您在无限循环中使用 p1_string 并期望它更新。为 pthread_param.buffer 分配新值不会更改 px_string 的值。因此，在这种情况下，字符串将始终等于其初始值，并且字符串 1 将始终小于字符串 2。

pt1_param.buffer = p1_string; //assignment by value
pt2_param.buffer = p2_string; //assignment by value

2) 考虑代码中的以下竞争条件。生产者中的 getline() 函数和 if/cout 代码可以同时访问缓冲区变量。将生产者中的（伪）代码的顺序重新排列为以下可能看起来非常相似，但是，它确实改变了相当多的行为：

while (1)
{
    sem_wait()
    if (!getline(buffer))
        break;
    sem_post()
}

现在生产者必须立即阻塞并等待，直到他们从消费者那里得到信号，表明它已经完成了对缓冲区变量的访问。它在发布之前调用 wait() ，这具有非常重要的效果，我将尝试描述。以前，生产者和消费者都在 sem_wait 之前调用 sem_post，并且各自的信号量计数都增加了。因此，当生产者尝试 wait() 时，它只会减少已经存在的计数并继续。由于消费者已经增加了它的信号量，所以在生产者中也会发生同样的事情。因此，生产者和消费者中循环的每次迭代都成为使用缓冲区变量的不可预知的竞争情况。