16.2 Bounded Buffer Protected by Mutex Locks

Figure 16.2 shows a diagram of a circular buffer with eight slots that might be used as a holding area between producers and consumers. The buffer has three data items, and the remaining five slots are free. The bufout variable has the slot number of the next data item to be removed, and the bufin variable has the number of the next slot to be filled.

Figure 16.2. Circular buffer.

Program 16.1 is an initial version of a circular buffer implemented as a shared object. The data structures for the buffer have internal linkage because the static qualifier limits their scope. (See Appendix A.5 for a discussion of the two meanings of the static qualifier in C.) The code is in a separate file so that the program can access the buffer only through getitem and putitem . The header file, buffer.h , contains the definitions of BUFSIZE and buffer_t . The functions of Program 16.1 follow the preferred POSIX error-handling semantics and return 0 if successful or a nonzero error code if unsuccessful .

Program 16.1 bufferbad.c

A flawed circular buffer protected by mutex locks .

 #include <pthread.h> #include "buffer.h" static buffer_t buffer[BUFSIZE]; static pthread_mutex_t  bufferlock = PTHREAD_MUTEX_INITIALIZER; static int bufin = 0; static int bufout = 0; int getitem(buffer_t *itemp) {  /* remove item from buffer and put in *itemp */    int error;    if (error = pthread_mutex_lock(&bufferlock))         /* no mutex, give up */       return error;    *itemp = buffer[bufout];    bufout = (bufout + 1) % BUFSIZE;    return pthread_mutex_unlock(&bufferlock); } int putitem(buffer_t item) {                    /* insert item in the buffer */    int error;    if (error = pthread_mutex_lock(&bufferlock))         /* no mutex, give up */       return error;    buffer[bufin] = item;    bufin = (bufin + 1) % BUFSIZE;    return pthread_mutex_unlock(&bufferlock); } 

Exercise 16.1

The following code segment uses the circular buffer defined in Program 16.1. What happens when it executes?

 int myitem; if (getitem(&myitem) == 0)    printf("retrieved %d from the buffer\n", myitem); 

Answer:

The result cannot be predicted . The getitem returns an error only when the locking fails, but it does not keep track of the number of items in the buffer. If a consumer executes this code before a producer calls putitem , the value retrieved for myitem will not be meaningful.

Exercise 16.2

The following code segment uses the circular buffer defined in Program 16.1. What happens when it executes?

 int i; for (i = 0; i < 10; i++)    if (putitem(i))       break; 

Answer:

The buffer has only 8 slots, but this code segment calls putitem 10 times. The putitem does not keep track of how many empty slots are available, so it does not report an error if full slots are overwritten. If a consumer does not call getitem , the code overwrites the first items in the buffer.

Program 16.1 is flawed because the code does not protect the buffer from overflows or underflows. Program 16.2 is a revised implementation that keeps track of the number of items actually in the buffer. If successful, getitem and putitem return 0. If unsuccessful, these functions return a nonzero error code. In particular, getitem returns EAGAIN if the buffer is empty, and putitem returns EAGAIN if the buffer is full.

Example 16.3

The following code segment attempts to retrieve at most 10 items from the buffer of Program 16.2.

 int error; int i; int item; for (i = 0; i < 10; i++) {    while((error = getitem(&item)) && (error == EAGAIN)) ;    if (error)                      /* real error occurred */       break;    printf("Retrieved item %d: %d\n", i, item); } 

Program 16.2 buffer.c

A circular buffer implementation that does not allow overwriting of full slots or retrieval of empty slots .

 #include <errno.h> #include <pthread.h> #include "buffer.h" static buffer_t buffer[BUFSIZE]; static pthread_mutex_t  bufferlock = PTHREAD_MUTEX_INITIALIZER; static int bufin = 0; static int bufout = 0; static int totalitems = 0; int getitem(buffer_t *itemp) {  /* remove item from buffer and put in *itemp */    int error;    int erroritem = 0;    if (error = pthread_mutex_lock(&bufferlock))         /* no mutex, give up */       return error;    if (totalitems > 0) {                   /* buffer has something to remove */       *itemp = buffer[bufout];        bufout = (bufout + 1) % BUFSIZE;        totalitems--;    } else        erroritem = EAGAIN;    if (error = pthread_mutex_unlock(&bufferlock))       return error;                /* unlock error more serious than no item */    return erroritem; } int putitem(buffer_t item) {                    /* insert item in the buffer */    int error;    int erroritem = 0;    if (error = pthread_mutex_lock(&bufferlock))         /* no mutex, give up */       return error;    if (totalitems < BUFSIZE) {           /* buffer has room for another item */       buffer[bufin] = item;       bufin = (bufin + 1) % BUFSIZE;       totalitems++;    } else       erroritem = EAGAIN;    if (error = pthread_mutex_unlock(&bufferlock))       return error;                /* unlock error more serious than no slot */    return erroritem; } 

The while loop of Example 16.3 uses busy waiting. The implementation is worse than you might imagine. Not only does busy waiting waste CPU time, but consumers executing this code segment block the producers, resulting in even more delay. Depending on the thread-scheduling algorithm, a busy-waiting consumer could prevent a producer from ever obtaining the CPU.