Producer-consumer code

Question

The idea I have in my head is this:

I am not sure if I need the index variable in consumer and producer.

Moreover, I want to know if it is OK like this (it seems to work), or it is just some stupid working code. Can it be more clever/efficient? Leave the readability for me.

My code will use the POSIX semaphore mutex in order to access the CS (which is writing/reading from the shared buffer of words).

full and empty are there to indicate if the buffer is full, or empty, so that we don't write a word if the buffer is full (we will overwrite the first word if we do that, before consumer reads it). Similarly, we don't want to read from any empty buffer!

When the producer is done, it sets a shared flag (initialized in 0). However, in the consumer side I had to check not only for the flag but for the values of full and empty.

The size of the buffer is 15. n (and the typo m) in the picture below is in our case 15.

Producer:

int index = 0;
size_t str_len;
while (fgets(buf, sizeof(buf), fp) != NULL) {
  str_len = strlen(buf);
  if(buf[str_len - 1] != '\n')
    printf("Didn't read whole line!\n");
  else {
    buf[str_len - 1] = '\0';
    sem_wait(&(sh->empty));
    sem_wait(&(sh->mutex));
    printf("|%s|\n", buf);
    strcpy(sh->words[index], buf);
    index = (index + 1) % N;
    sem_post(&(sh->mutex));
    sem_post(&(sh->full));
  }
}
sem_wait(&(sh->mutex));
sh->done = 1;
sem_post(&(sh->mutex));
int full_val;
sem_getvalue(&(sh->full), &full_val);
while(full_val++ <= 0) { // 'if' instead of 'while' should be OK too
  sem_post(&(sh->full));
}

Consumer:

int index = 0;
while(1) {
  sem_wait(&(sh->full));
  sem_wait(&(sh->mutex));
  int full_val, empty_val;
  sem_getvalue(&(sh->full), &full_val);
  sem_getvalue(&(sh->empty), &empty_val);
  if(empty_val != 15)
    printf("got: %s\n", sh->words[index]);
  if(sh->done == 1 && full_val == 0)
    break;
  index = (index + 1) % N;
  sem_post(&(sh->mutex));
  sem_post(&(sh->empty));
}

I am obliged to use sem_wait() and sem_post() by the way.

---

After the answers, I think that the main question is that if I can avoid the extra post in full semaphore after the while of producer is ended.

The problem if I remove the posting of full is in the following case:

1. Producer puts all the words in the buffer.
2. Scheduler gives the CPU to consumer.
3. Consumer reads all the words and gets suspended because on the full semaphore.
4. Now, scheduler gives the CPU to producer which now sets the shared flag to 1.
5. Scheduler goes back to consumer, which is suspended on full semaphore, forever.

Answer 1

Robust code does not assume the first char in buf is not '\0'.

This is a pedantic point, but user IO is a hacker paradise - best to code defensively.

Instead of:

size_t str_len;
while (fgets(buf, sizeof(buf), fp) != NULL) {
  str_len = strlen(buf);
  if(buf[str_len - 1] != '\n')
    printf("Didn't read whole line!\n");
  else {
    buf[str_len - 1] = '\0';

use to prevent undefined behavior (UB) with buf[str_len - 1].

while (fgets(buf, sizeof(buf), fp) != NULL) {
  size_t str_len = strlen(buf);
  if (str_len == 0 || buf[str_len - 1] != '\n') {
    printf("Didn't read whole line or embedded null character detected!\n");
  } else {
    buf[str_len - 1] = '\0';

Answer 2

Check error status:

Well the first thing you should do is check the error state of all system calls.

From: sem_wait.html

The sem_wait() function locks the semaphore referenced by sem by performing a semaphore lock operation on that semaphore. If the semaphore value is currently zero, then the calling thread will not return from the call to sem_wait() until it either locks the semaphore or the call is interrupted by a signal.

Usage should be more like this:

int errorCode;
do
{
    errorCode = sem_wait(&sh->full);
    if (errorCode != 0) && (errno == EINTR)
    {
         /* Signal. We can try again */
         continue;
    }
    if (errorCode != 0)
    {
        /* Unrecoverable Error */
        exit(1);
    }
} while(errorCode != 0);

Index

Both the consumer and producer use index to access the queue. If these are distinct values unique to consumer/producer then it will work but it is definately a bad naming scheme. Maybe better is consumerIndex and produerIndex to indicate the read/write position.

Semaphore Simplification.

I don't think you need three semaphores. You should be able to use two (one for the producers one for the consumers). You only need the mutex if there are multiple producers or multiple consumers.

Answer 3

The producer posts the full semaphore on depositing a message. The consumer waits on a full semaphore to remove the message. Therefore its value is a number of messages in the queue. The

while(full_val++ <= 0) { // 'if' instead of 'while' here is OK too
  sem_post(&(sh->full));
}

loop clearly violate the invariant. If your program doesn't properly terminate without it, there must be a bug somewhere.

My impression is that the bug stems from the index variable. A state of the queue requires two indices: one for producer (to put the message to) and one for consumer (to fetch the message from).

And yes, you need a shared resource to convey a producer termination event to the consumer.