This chapter is from the book
This chapter is from the book
This chapter is from the book
4.11 Dividing Your Program into Multiple Threads
Earlier in this chapter we discussed the delegation of work according to a specific strategy or approach called a thread model. Those thread models were:
-
delegation (boss–worker)
-
peer-to-peer
-
pipeline
-
producer–consumer
Each model has its own WBS (Work Breakdown Structure) that determines who is responsible for thread creation and under what conditions threads are created. In this section we will show an example of a program for each model using Pthread library functions.
4.11.1 Using the Delegation Model
We discussed two approaches that can be used to implement the delegation approach to dividing a program into threads. To recall, in the delegation model, a single thread (boss) creates the threads (workers) and assigns each a task. The boss thread delegates the task each worker thread is to perform by specifying a function. With one approach, the boss thread creates threads as a result of requests made to the system. The boss thread processes each type of request in an event loop. As events occur, thread workers are created and assigned their duties. Example 4.5 shows the event loop in the boss thread and the worker threads in pseudocode.
Example 4.5 Approach 1: Skeleton program of boss and worker thread model.
//...
pthread_mutex_t Mutex = PTHREAD_MUTEX_INITIALIZER
int AvailableThreads
pthread_t Thread[Max_Threads]
void decrementThreadAvailability(void)
void incrementThreadAvailability(void)
int threadAvailability(void);
// boss thread
{
//...
if(sysconf(_SC_THREAD_THREADS_MAX) > 0){
AvailableThreads = sysconf(_SC_THREAD_THREADS_MAX)
}
else{
AvailableThreads = Default
}
int Count = 1;
loop while(Request Queue is not empty)
if(threadAvailability()){
Count++
decrementThreadAvailability()
classify request
switch(request type)
{
case X : pthread_create(&(Thread[Count])...taskX...)
case Y : pthread_create(&(Thread[Count])...taskY...)
case Z : pthread_create(&(Thread[Count])...taskZ...)
//...
}
}
else{
//free up thread resources
}
end loop
}
void *taskX(void *X)
{
// process X type request
incrementThreadAvailability()
return(NULL)
}
void *taskY(void *Y)
{
// process Y type request
incrementThreadAvailability()
return(NULL)
}
void *taskZ(void *Z)
{
// process Z type request
decrementThreadAvailability()
return(NULL)
}
//...
In Example 4.5, the boss thread dynamically creates a thread to process each new request that enters the system, but there are a maximum number of threads that will be created. There are n number of tasks to process n request types. To be sure the maximum number of threads per process will not be exceeded, these additional functions can be defined:
threadAvailability() incrementThreadAvailability() decrementThreadAvailability()
Example 4.6 shows pseudocode for these functions.
Example 4.6 Functions that manage thread availability count.
void incrementThreadAvailability(void)
{
//...
pthread_mutex_lock(&Mutex)
AvailableThreads++
pthread_mutex_unlock(&Mutex)
}
void decrementThreadAvailability(void)
{
//...
pthread_mutex_lock(&Mutex)
AvailableThreads—
pthread_mutex_unlock(&Mutex)
}
int threadAvailability(void)
{
//...
pthread_mutex_lock(&Mutex)
if(AvailableThreads > 1)
return 1
else
return 0
pthread_mutex_unlock(&Mutex)
}
The threadAvailability() function will return 1 if the maximum number of threads allowed per process has not been reached. This function accesses a global variable ThreadAvailability that stores the number of threads still available for the process. The boss thread calls the decrementThreadAvailability() function, which decrements the global variable before the boss thread creates a thread. The worker threads call incrementThreadAvailability(), which increments the global variable before a worker thread exits. Both functions contain a call to pthread_mutex_lock() before accessing the variable and a call to pthread_mutex_unlock() after accessing the global variable. If the maximum number of threads are exceeded, then the boss thread can cancel threads if possible or spawn another process, if necessary. taskX(), taskY(), and taskZ() execute code that processes their type of request.
The other approach to the delegation model is to have the boss thread create a pool of threads that are reassigned new requests instead of creating a new thread per request. The boss thread creates a number of threads during initialization and then each thread is suspended until a request is added to the queue. The boss thread will still contain an event loop to extract requests from the queue. But instead of creating a new thread per request, the boss thread signals the appropriate thread to process the request. Example 4.7 shows the boss thread and the worker threads in pseudocode for this approach to the delegation model.
Example 4.7 Approach 2: Skeleton program of boss and worker thread model.
//...
pthread_t Thread[N]
// boss thread
{
pthread_create(&(Thread[1]...taskX...);
pthread_create(&(Thread[2]...taskY...);
pthread_create(&(Thread[3]...taskZ...);
//...
loop while(Request Queue is not empty
get request
classify request
switch(request type)
{
case X :
enqueue request to XQueue
signal Thread[1]
case Y :
enqueue request to YQueue
signal Thread[2]
case Z :
enqueue request to ZQueue
signal Thread[3]
//...
}
end loop
}
void *taskX(void *X)
{
loop
suspend until awaken by boss
loop while XQueue is not empty
dequeue request
process request
end loop
until done
{
void *taskY(void *Y)
{
loop
suspend until awaken by boss
loop while YQueue is not empty
dequeue request
process request
end loop
until done
}
void *taskZ(void *Z)
{
loop
suspend until awaken by boss
loop while (ZQueue is not empty)
dequeue request
process request
end loop
until done
}
//...
In Example 4.7, the boss thread creates N number of threads, one thread for each task to be executed. Each task is associated with processing a request type. In the event loop, the boss thread dequeues a request from the request queue, determines the request type, enqueues the request to the appropriate request queue, then signals the thread that processes the request in that queue. The functions also contain an event loop. The thread is suspended until it receives a signal from the boss that there is a request in its queue. Once awakened, in the inner loop, the thread processes all the requests in the queue until it is empty.
4.11.2 Using the Peer-to-Peer Model
In the peer-to-peer model, a single thread initially creates all the threads needed to perform all the tasks called peers. The peer threads process requests from their own input stream. Example 4.8. shows a skeleton program of the peer-to-peer approach of dividing a program into threads.
Example 4.8 Skeleton program using the peer-to-peer model
//...
pthread_t Thread[N]
// initial thread
{
pthread_create(&(Thread[1]...taskX...);
pthread_create(&(Thread[2]...taskY...);
pthread_create(&(Thread[3]...taskZ...);
//...
}
void *taskX(void *X)
{
loop while (Type XRequests are available)
extract Request
process request
end loop
return(NULL)
}
//...
In the peer-to-peer model, each thread is responsible for its own stream of input. The input can be extracted from a database, file, and so on.
4.11.3 Using the Pipeline Model
In the pipeline model, there is a stream of input processed in stages. At each stage, work is performed on a unit of input by a thread. The input continues to move to each stage until the input has completed processing. This approach allows multiple inputs to be processed simultaneously. Each thread is responsible for producing its interim results or output, making them available to the next stage or next thread in the pipeline. Example 4.9 shows the skeleton program for the pipeline model.
Example 4.9 Skeleton program using the pipeline model.
//...
pthread_t Thread[N]
Queues[N]
// initial thread
{
place all input into stage1's queue
pthread_create(&(Thread[1]...stage1...);
pthread_create(&(Thread[2]...stage2...);
pthread_create(&(Thread[3]...stage3...);
//...
}
void *stageX(void *X)
{
loop
suspend until input unit is in queue
loop while XQueue is not empty
dequeue input unit
process input unit
enqueue input unit into next stage's queue
end loop
until done
return(NULL)
}
//...
In Example 4.9, N queues are declared for N stages. The initial thread enqueues all the input into stage 1's queue. The initial thread then creates all the threads needed to execute each stage. Each stage has an event loop. The thread sleeps until an input unit has been enqueued. The inner loop continues to iterate until its queue is empty. The input unit is dequeued, processed, then that unit is then enqueued into the queue of the next stage.
4.11.4 Using the Producer–Consumer Model
In the producer-consumer model, the producer thread produces data consumed by the consumer thread or threads. The data is stored in a block memory shared between the producer and consumer threads. This model was used in Programs 4.5, 4.6, and 4.7. Example 4.10 shows the skeleton program for the producer-consumer model.
Example 4.10 Skeleton program using the producer–consumer model.
pthread_mutex_t Mutex = PTHREAD_MUTEX_INITIALIZER
pthread_t Thread[2]
Queue
// initial thread
{
pthread_create(&(Thread[1]...producer...);
pthread_create(&(Thread[2]...consumer...);
//...
}
void *producer(void *X)
{
loop
perform work
pthread_mutex_lock(&Mutex)
enqueue data
pthread_mutex_unlock(&Mutex)
signal consumer
//...
until done
}
void *consumer(void *X)
{
loop
suspend until signaled
loop while(Data Queue not empty)
pthread_mutex_lock(&Mutex)
dequeue data
pthread_mutex_unlock(&Mutex)
perform work
end loop
until done
}
In Example 4.9, an initial thread creates the producer and consumer threads. The producer thread executes a loop in which it performs work then locks a mutex on the shared queue in order to enqueue the data it has produced. The producer unlocks the mutex then signals the consumer thread that there is data in the queue. The producer iterates through the loop until all work is done. The consumer thread also executes a loop in which it suspends itself until it is signaled. In the inner loop, the consumer thread processes all the data until the queue is empty. It locks the mutex on the shared queue before it dequeues any data and unlocks the mutex after the data has been dequeued. It then performs work on that data. In Program 4.6, the consumer thread enqueues its results to a file. The results could have been inserted into another data structure. This is often done by consumer threads in which it plays both the role of consumer and producer. It plays the role of consumer of the unprocessed data produced by the producer thread, then it plays the role of producer when it processes data stored in another shared queue consumed by another thread.
4.11.5 Creating Multithreaded Objects
The delegation, peer-to-peer, pipeline. and producer–consumer models demonstrate approaches to dividing a program into multiple threads along function lines. When using objects, member functions can create threads to perform multiple tasks. Threads can be used to execute code on behalf of the object: free-floating functions and other member functions.
In either case, the threads are declared within the object and created by one of the member functions (e.g., the constructor). The threads can then execute some free-floating functions (function defined outside the object), which invokes member functions of the object that are global. This is one approach to making an object multithreaded. Example 4.10 contains an example of a multithreaded object.
Example 4.10 Declaration and definition of multithreading an object.
#include <pthread.h>
#include <iostream>
#include <unistd.h>
void *task1(void *);
void *task2(void *);
class multithreaded_object
{
pthread_t Thread1,Thread2;
public:
multithreaded_object(void);
int c1(void);
int c2(void);
//...
};
multithreaded_object::multithreaded_object(void)
{
//...
pthread_create(&Thread1,NULL,task1,NULL);
pthread_create(&Thread2,NULL,task2,NULL);
pthread_join(Thread1,NULL);
pthread_join(Thread2,NULL);
//...
}
int multithreaded_object::c1(void)
{
// do work
return(1);
}
int multithreaded_object::c2(void)
{
// do work
return(1);
}
multithreaded_object MObj;
void *task1(void *)
{
//...
MObj.c1();
return(NULL);
}
void *task2(void *)
{
//...
MObj.c2();
return(NULL);
}
In Example 4.10, the class multithread_object declares two threads. From the constructor of the class, the threads are created and joined. Thread1 executes task1 and Thread2 executes task2. task1 and task2, then invokes member functions of the global object MObj.