Introduction.

In this assignment you are to write a progream whose initial thread creates three other threads. These three threads - a producer, and two consumers - are to communicate using shared memory and semaphores. The producer has a number of "tasks" to be performed. These tasks are to be given to the consumer threads. It doesn't matter which consumer gets which task. However, one and only one consumer should get each task. The nature of these tasks is described in section A) below.

The producer will insert each task, one at a time, into a shared bounded buffer. Each consumer will remove tasks, one at a time, from this buffer. A bound, bufSize, will be imposed on the buffer. That is, at most bufSize tasks can be in the buffer at a time. The total number of tasks to be done will be stored in a global variable, limit, and will generally be larger than bufSize. So the producer will have to wait if the buffer becomes full. Consumers always wait if the buffer is empty. The producer and consumer threads should continue until all tasks have been received. If n1 is the number of tasks removed by Consumer1 and n2, the number removed by Consumer2, then n1 + n2 should be equal to limit. However, the number of tasks removed by Consumer1 will generally not be the same as removed by Consumer2 and this may change each time the program is run.

As each task is removed from the buffer by a consumer, the consumer should build an item:

struct item
{
  // Public members
  int task;
  pthread_t which_thread;

  // Constructor
  item(int tsk=0, pthread_t tid=0) 
   { task = tsk; which_thread = tid; }
};

The task member should be set to the task number removed from the buffer. The which_thread member should be set to the thread id of the consumer constructing the item. Finally, the consumer should then insert the item into a shared list (unbounded), call results.

It is best if the initial thread waits for all three other threads to terminate. (See section D below.) The initial thread can then process the results list sequentially. It should print out the total number of items, the number of items inserted by Consumer1 and the number inserted by Consumer2, and any task was not completed.

Here is a diagram of the threads, and their interaction through the the bounded buffer and the results list:

The producer can presumably terminate when it has copied the last task and one of the consumers has accepted it. However, both consumers should eventually terminate, so the producer may need to do a bit more communication to make sure of this. That is, you are allowed to modify the producer, if necessary, in order to help the consumers know when to stop.

How does a consumer know if a removed task is the last task? How does a consumer know if there are any more tasks to be done? If there are no more tasks to be done, should a Consumer execute a sem_wait? If so, and the Consumer blocks, who will unblock it?

The following details are discussed below:

• Description of the problem
• Remarks on shared memory
• Remarks on the semaphores
• Waiting for the producer/consumers
• Initial program code.
• What to turn in.

A. Description of the problem.

A task will be represented by an integer.

The initial thread should:

• Prompt the user for the total number of tasks (limit).
• Prompt the user for the bounded buffer size. This can be much smaller than limit.
• Initialize all semaphores to be used by the threads.
• Create the three threads. Each of the three threads should wait until the initial thread has finished creating all the threads. (Use a semaphore for this synchronization.)
• Unblock each of the three waiting threads. (Using the same semaphore as the previous item.)
• Wait for all three threads to finish.
• Process the results list.
  1. Check that the list contains 'limit' number of items.
  2. Count the number of items in the results list for each consumer thread and print the counts.
  3. Check that every task, 1 to limit, was in some item inserted in the results list. If some task was not in the list, print the task number and that it was not inserted.

The producer thread should:

• Wait until the initial thread has finished creating all three threads. (Use the same semphore the initial thread is using for this synchronization.)
• For each task = 1 to limit, insert the task in the bounded buffer. For each insertion, wait if the buffer is full, then gain exclusive control over the buffer and insert the one task. Release control over the buffer and notify the consumers that a new item has been added to the buffer.

Each consumer thread should:

1. While not all the tasks have been "consumed" remove a task. For each removal, wait if the buffer is empty, then gain exclusive control over the buffer and remove one task. Release control over the buffer and notify the producer, that a new empty slot is now available in the buffer.
2. After an task has been removed from the buffer and control of the buffer is released, the consumer should store the task (integer) and its own thread id into an "item". The consumer should then gain exclusive control over the results list and insert this item into the that list. The consumer should then release control over the results list.

You may want to print the values of the result list for debugging purposes. This feature may result in too much output for large values of limit. You should be able to turn this feature on or off using a command line argument. This will be discussed in class.

B. Remarks on shared memory

When using shared memory and semaphores, you will be using threads. So you will need to include the thread.h header:

#include <thread.h>

but no additional header file is needed to use shared memory with threads, since all threads within a process share the same address space and all have access to the global variables.

You should use the standard c++ list class for the results list. I have provided a BoundedBuffer class for the buffer. y:

#include <list>

int limit;

list<item, malloc_alloc> results; // where item is as declared above

list<int, malloc_alloc> buffer;

int bufsize;

The program should prompt for the buffer size, bufSize, and for the number of tasks, limit. All of this should be done by the initial thread before creating the producer and two consumer threads.

You can also use a small number of other shared (global) memory variables to communicate information if you think it would be useful in solving the termination problem for the two consumer threads.

The producer should only hold exclusive use of the buffer while inserting one item. That is, the producer should not get exclusive use and hold it while it fills up the buffer. This might cause the consumers to wait when the buffer is not empty. Each time the producer gets exclusive control of the buffer, it should only insert ONE task. To insert another task, it must first release control and then reacquire control for the next task.

C. Remarks on semaphores

As noted in above, when using shared memory and semaphores, you will be using threads. So you will need to include the thread.h header. To use semaphores for threads, you will also need to include the semaphore.h header:

#include <thread.h>
#include <sys/semaphore.h>

You will need to use sem_init, sem_wait, and sem_post. These functions are briefly described by:

C.1 Semaphore declaration

sem_t s;  /* declaration for the data structure for a semaphore
*/

C.2 Semaphore initialization

      int sem_init(sem_t *sem, int pshared, unsigned int value);

	sem = address of a sem_t structure
        pshared = should be 0 (can't be shared with threads in other processes)
	value = the initial value for the semaphore, must be >= 0.


Returns 0 if successful, and non-zero otherwise.

	if ( sem_init(&s, 0, 5) != 0)
	  {
	    printf("Semaphore not initialized\n");
	    exit(0);
          }

C.3 P operation ("decrement") on semaphores.

int sem_wait(sem_t *sp); 

	sp = address of a sem_t structure

Example:

	sem_wait(&s);

C.4 V operation ("increment") on semaphores.

int sem_post(sem_t *sp); 

	sp = address of a sem_t structure

Example:

	sem_post(&s);

D. Waiting for the producer and consumers

You should have the initial thread execute pthread_join(th_id, 0) statement for each of the threads, where th_id is the thread id of the thread for which you are waiting.

	pthread_t th_id;

but the type pthread_t is really just unsigned integer.

E. Initial code.

The initial code provides:

• Sample code declaring the data structures for the bounded buffer and the results list. (Both use the C++ template list class. However, they each use the malloc_alloc allocator to ensure that memory allocation for these two data structures is "thread safe" as multiple threads sometimes will access these two separate lists simulataneously or interleaved.
• Sample declarations for semaphores you may need.
• Sample code initializing two of these semaphores, goSem and printMutex

  The goSem semaphore is used to make the producer/consumer threads wait until the initial thread has finished creating all the threads.

  The printMutex thread is to guarantee only one thread prints at a time (see the DEBUG function next)

• void DEBUG(char debugChar, char *msg)

  This function uses the printMutex semaphore to print msg. However, DEBUG only prints msg provided the character value of debugChar was entered as a command line argument when prog1sample is run.

• Sample code in main to create one producer and one consumer.
• Skeleton producer function, that shows how the producer blocks itself using the goSem semaphore in order to wait until the initial thread has created all the other threads.
• Similar Skeleton consumer function, with the same method of blocking itself.
• Comments labeled // YOU WRITE CODE HERE with some comments indicate where you should add code to this initial version in order to complete the program requirements.

F. What to turn in.

I will run your program. You should put your source code file(s) in a subdirectory named "prog1src" of your unix login directory. You should give me (and only me) access permissions to that directory (r-x) and read permission on your source code files for project 1 (r--). Use the access control list command, setacl, to set these permissions. (See the Jan19 lecture notes and/or the man pages for setacl.) Submit a file to COL for this project with the message that your program is ready and the permissions have been changed to give me access. You may also include any additional comments you wish to provide on running your program.