producer/consumer using Pthreads and Semaphores

Question

I'm currently learning houw to use Pthreads and Semaphores and I've been working on an Implementation of a Producer/Consumer problem but the program just hangs. I know that it gets to the consumer code, runs the comparison once and then hangs after comparing p1_string and p2_string at default initialization values and I really don't understand what exactly I'm doing wrong.

Basically each producer thread is supposed to take a file of sorted lines and read a line into memory. Then the main thread is supposed to compare those two strings, and write them in sorted order to output.

#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) The = operator on a string copies by value. The following code is copying by value into the 'buffer' variable. However, later you are using p1_string in an infinite loop expecting it to update. Assigning a new value to the pthread_param.buffer does not change the value of px_string. Thus, in this case the strings will always equal their initial value and string 1 will always be less than string 2.

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

2) Consider the following race condition in the code. The getline() function in the producer and the if/cout code could both be accessing the buffer variable at the same time. Rearranging the order of (pseudo)code in the producer to the following may appear to be very similar, however, it does changes the behavior quite a bit:

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

Now the producers have to immediately block and wait until they get a signal from the consumer that it's done accessing the buffer variables. It's calling the wait() before the post and this has a very significant effect that I will try to describe. Previously both the producers and consumer call sem_post before sem_wait and the respective semaphore counts are both incremented. Thus, when the producer tries to wait() it just ends up decrementing the already existing count and it continues on. The same thing happens in the producer as a result of the consumer having incremented its semaphore already. Thus every iteration of the loop in both the producer and the consumer becomes an unpredictable race situation to use the buffer variable(s).