Names pipes are created with mkfifo():

Synposis

#include <sys/types.h>
#include <sys/stat.h>

int mkfifo(const char *pathname, mode_t mode);
int unlink(const char *pathname);

mkfifo() creates a named pipe using pathname as the name of the FIFO with permission specified by mode. mode comprises the file permission bits. They are as listed previously in Table 7-3.

mkfifo() is created with O_CREAT | O_EXCL flags, which means it creates a new named pipe with the name specified if it does not exist. If is does exist, an error EEXIST is returned. So, if you want to open an already existing named pipe, call the function, and check for this error. If the error occurs then use open() instead of mkfifo().

The unlink() removes the filename pathname from the filesystem. The program in Listing 7-1 creates a named pipe with mkfifo.

Listings 7-1 and 7-2 are listings of programs that demonstrate how a named pipe can be used the transfer data from one process to another unrelated process. Listing 7-1 contains the program of the writer, and Listing 7-2 contains the program for the reader.

// Listing 7-1 A program that creates a named pipe with mkfifo().

 1  using namespace std;
 2  #include <iostream>
 3  #include <fstream>
 4  #include <sys/wait.h>
 5  #include <sys/types.h>
 6  #include <sys/stat.h>
 7

8  int main(int argc,char *argv[],char *envp[])
 9  {
10
11     fstream Pipe;
12
13     if(mkfifo("Channel-one",S_IRUSR | S_IWUSR
14                                     | S_IRGRP
15                                     | S_IWGRP) == −1){
16        cerr << "could not make fifo" << endl;
17     }
18
19     Pipe.open("Channel-one",ios::out);
20     if(Pipe.bad()){
21        cerr << "could not open fifo" << endl;
22     }
23     else{
24            Pipe << "2 3 4 5 6 7 " << endl;
25
26     }
27
28     return(0);
29  }

The program in Listing 7-1 creates a named pipe with the mkfifo() system call. The program then opens the pipe with an fstream object called Pipe in Line 19. Notice that Pipe has been opened for output using the ios::out flag. If the Pipe is not in a bad() state after the call to open, then the Pipe is ready for writing data to Channel-one. Although Pipe is ready for input, it blocks (waits) until another process has opened Channel-one for reading. When using the iostreams with pipes, it is important to remember that either the writer or reader must be opened for both input and output using ios::in | ios::out. Opening either the reader or the writer in this manner will prevent deadlock. In this case, we open the reader (Listing 7-2) for both. The program in this listing is called a reader because it reads the information from the pipe. The writer then writes a line of input to Pipe.

// Listing 7-2 A program that reads from a named pipe.

 1  using namespace std;
 2  #include <iostream>
 3  #include <fstream>
 4  #include <string>
 5  #include <sys/wait.h>
 6  #include <sys/types.h>
 7  #include <sys/stat.h>
 8
 9
10  int main(int argc, char *argv[])
11  {
12     int type;
13     fstream NamedPipe;
14     string Input;
15
16     NamedPipe.open("Channel-one",ios::in | ios::out);

17
18     if(NamedPipe.bad()){
19        cerr << "could not open Channel-one" << endl;
20     }
21
22     while(!NamedPipe.eof() && NamedPipe.good()){
23
24        getline(NamedPipe,Input);
25        cout << Input << endl;
26     }
27     NamedPipe.close();
28     unlink("Channel-one");
29     return(0);
30
31  }

The program in Listing 7-1 uses the << operator to write data into the pipe. Here the reader also has to open the pipe by using an fstream open, using the name of the named pipe Channel-one, and opening the pipe for input and output in Line 16. If the NamedPipe is not in a bad state after opening, then the data is read from the NamedPipe while NamedPipe is not eof() and is still good. The data is read from the pipe and stored in a string Input that is sent to cout. NamedPiped is then closed and the pipe is unlinked. These are unrelated processes. To run, each is launched separately.

Here is Program Profile 7-1 for Listings 7-1 and 7-2.

program7-1.cc (Listing 7-1)

program7-2.cc (Listing 7-2)

Listing 7-1 creates a pipe and opens the pipe with a fstream object. Before it writes a string to the pipe, it waits until another process opens the pipe for reading (Listing 7-2). Once the reader process opens the pipe for reading, the writer program writes a string to the pipe, closes the pipe, and then exits. The reader pipe reads the string from the pipe and displays the string to standard out. Run the reader and then the writer.

None

<iostream> <fstream> <string> <sys/stat.h>

c++ -o program7-1 program7-1.cc program7-2.cc

Solaris 10, gcc 3.4.3 and 3.4.6

Opteron, UltraSparc T1

./program7-1
./program7-2

Run each program in separate terminals.

The use of fstream simplifies the IPC making the named pipe easier to access. All of the functionality of streams comes into play for this example:

mkfifo("Channel-one",...);
vector<int> X( 2,3,4,5,6,7);
ofstream OPipe("Channel-one",ios::out);
ostream_iterator<int> Optr(OPipe,"
");
copy(X.begin(),X.end(),Optr);

Here a vector is used to hold all the data. An ofstream object is used this time instead of an fstream object to open the named pipe. An ostream iterator is declared and points to the named pipe (Channel-one). Now instead of successive insertion in a loop, the copy algorithm can copy all the data to the pipe. This is a convenience if have hundreds, thousands, or even more numbers to write to the pipe.

Besides simplifying the use of IPC mechanisms by using iostreams, iterators, and algorithm, you can also simplify use by encapsulating the FIFO into an FIFO interface class. Remember that in doing so you are modeling a FIFO structure. The FIFO class is a model for the communication between two or more processes. It transmits some form of information between them. The information is translated into a sequence of data, inserted into the pipe, and then retrieved by a process on the other side of the pipe. The data is then reassembled by the retrieving process. There must be somewhere for the data to be stored while it is in transit from process A to process B. This storage area for the data is called a buffer. Insertion and extraction operations are used to place the data into or extract the data from this buffer. Before performing insertions or extractions into or from the data buffer, the data buffer must exist. Once communication has been completed, the data buffer is no longer needed. So, your model must be able to remove the data buffer when it is no longer necessary. As indicated, a pipe has two ends, one end for inserting data and the other end for extracting data, and these ends can be accessed from different processes. Therefore the model should also include an input port and an output port that can be connected to separate processes. Here are the basic components of the FIFO model:

For this example, there are only two processes involved in communication. But if there are multiple processes that can read/write to/ from the named pipe stream, synchronization is required. So, this class also requires a mutex object. Example 7-1 shows the beginnings of the FIFO class:

// Example 7-1 Declaration of fifo class.

 class fifo{
    mutex Mutex;
    //...
 protected:
    string  Name;
 public:
    fifo &operator<<(fifo &In, int X);
    fifo &operator<<(fifo &In, char X);
    fifo &operator>>(fifo &Out, float X);
    //...
};

Using this technique, you can easily create fifos in the constructor. You can pass them easily as parameters and return values. You can use them in conjunction with the standard container classes and so on. The construction of such a component greatly reduces the amount of code needed to use FIFOs, provides opportunities for type safety, and generally allows the programmer to work at a higher level.

The message queue is created with mq_open():

Synopsis

#include <mqueue.h>

mqd_t mq_open(const char *name, int oflag,mode_t mode,
              struct mq_attr *attr);
int mq_close(mqd_t mqdes);
int mq_unlink(const char *name);

mq_open() creates a message queue with the specified name. The message queue uses oflag with these possible values to specify the access modes:

These flags can be ORred with the following:

The function returns a message queue descriptor of type mq_dt.

The last parameter is a struct mq_attr *attr. This is an attribute structure that describes the properties of the message queue:

struct mq_attr {
   long mq_flags;       //flags
   long mq_maxmsg;      //maximum number of messages allowed
   long mq_msgsize;     //maximum size of message
   long mq_curmsgs;     //number of messages currently in queue
}

mq_close() closes the message queue, but the message queue still exists in the kernel. However, the calling function can no longer use the descriptor. If the process terminates, all message queues associated with the process also close. The data is still retained in the queue.

unlink() removes the message queue specified by name from the system. The number of references to the message queue is tracked, but the queue name can still be removed from the system even if the count is greater than 0. The queue is not destroyed until all processes that utilized the queue have closed or called mq_close().

There are two functions to set and return the attribute object, as shown in the following code synopsis:

Synopsis

#include <mqueue.h>

int mq_getattr(mqd_t mqdes,struct mq_attr *attr);
int mq_setattr(mqd_t mqdes,struct mq_attr *attr,struct mq_attr *oattr);

When you are setting the attribute with mq_setattr, only the mq_flags are set in the attr structure. Other attributes are not affected. mq_maxmsg and mq_msgsize are set when the message queue is created. mq_curmsg can be returned and not set. oattr contains the previous values for the attributes.

To send or write a message to the queue, use these functions:

Synopsis

#include <mqueue.h>

int mq_send(mqd_t mqdes, const char *ptr, size_t len,
            unsigned int prio);
ssize_t mq_receive(mqd_t mqdes, const char *ptr, size_t len,
            unsigned int priop);

For mq_receive(), the len must be at least the maximum size of the message. The returned message is stored in *ptr.

posix_queue is a simple class that models some of the functionality of a message queue. It encapsulates the basic functions and the message queue attributes. Listing 7-3 shows the declaration of the posix_queue class.

// Listing 7-3 Declaration of the posix_queue class.

 1  #ifndef __POSIX_QUEUE
 2  #define __POSIX_QUEUE
 3  using namespace std;
 4  #include <string>
 5  #include <mqueue.h>
 6  #include <errno.h>
 7  #include <iostream>
 8  #include <sstream>

9  #include <sys/stat.h>
10
11
12  class posix_queue{
13  protected:
14     mqd_t  PosixQueue;
15     mode_t  OMode;
16     int   QueueFlags;
17     string QueueName;
18     struct mq_attr  QueueAttr;
19     int  QueuePriority;
20     int MaximumNoMessages;
21     int MessageSize;
22     int ReceivedBytes;
23     void setQueueAttr(void);
24  public:
25     posix_queue(void);
26     posix_queue(string QName);
27     posix_queue(string QName,int MaxMsg, int MsgSize);
28     ~posix_queue(void);
29
30     mode_t  openMode(void);
31     void openMode(mode_t OPmode);
32
33     int queueFlags(void);
34     void queueFlags(int X);
35
36     int queuePriority(void);
37     void queuePriority(int X);
38
39     int maxMessages(void);
40     void maxMessages(int X);
41     int messageSize(void);
42     void messageSize(int X);
43
44     void queueName(string X);
45     string queueName(void);
46
47     bool open(void);
48     int send(string Msg);
49     int receive(string &Msg);
50     int remove(void);
51     int close(void);
52
53
54  };
55


#endif

The basic functions performed by a message queue are encapsulated in the posix_queue class:

47     bool open(void);
48     int send(string Msg);
49     int receive(string &Msg);
50     int remove(void);
51     int close(void);

We have discussed what each these functions does already. Examples 7-2 through 7-6 show the definitions of these methods. Example 7-2 is the definition of open():

// Example 7-2 The definition of open().

122  bool posix_queue::open(void)
123  {
124     bool Success = true;
125     int RetCode;
126     PosixQueue = mq_open(QueueName.c_str(),QueueFlags,OMode,&QueueAttr);
127     if(errno == EACCES){
128        cerr  << "Permission denied to created " << QueueName << endl;
129        Success = false;
130     }
131     RetCode = mq_getattr(PosixQueue,&QueueAttr);
132     if(errno == EBADF){
133        cerr << "PosixQueue is not a valid message descriptor" << endl;
134        Success = false;
135        close();
136
137     }
138     if(RetCode == −1){
139        cerr << "unknown error in mq_getattr() " << endl;
140        Success = false;
141        close();
142     }
143     return(Success);
144  }

After the call to mq_open() is made in Line #126, errno is checked to see if message queue failed to open because a message queue by the name QueueName.c_str() already exists. If it does, then the call is not successful. bool Success is returned with a value of false.

In Line 131, the queue attribute structure is returned by mq_getattr(). To ensure that the message queue was opened and successfully initialized its attribute, errno is checked again in Line 132. EBADF means the descriptor was not a valid message queue descriptor. The return code is checked in Line 138.

Example 7-3 is the definition of send().

// Example 7-3 The definition of send().

146  int posix_queue::send(string Msg)
147  {
148
149    int StatusCode = 0;
150    if(Msg.size() > QueueAttr.mq_msgsize){
151       cerr << "message to be sent is larger than max queue
                   message size " << endl;
152       StatusCode = −1;
153    }
154    StatusCode = mq_send(PosixQueue,Msg.c_str(),Msg.size(),0);
155    if(errno == EAGAIN){
156       StatusCode = errno;
157       cerr << "O_NONBLOCK not set and the queue is full " << endl;
158    }
159    if(errno == EBADF){
160       StatusCode = errno;
161       cerr << "PosixQueue is not a valid descriptor open for
                   writing" << endl;
162    }
163    if(errno == EINVAL){
164       StatusCode = errno;
165       cerr << "msgprio is out side of the priority range for the
                   message queue or " << endl;
166       cerr << "Thread my block causing a timing conflict with
                   time out" << endl;
167    }
168
169    if(errno == EMSGSIZE){
170       StatusCode = errno;
171       cerr << "message size exceeds maximum size of message
                   parameter on message queue" << endl;
172
173    }
174    if(errno == ETIMEDOUT){
175       StatusCode = errno;
176       cerr << "The O_NONBlock flag was not set, but the time expired
                   before the  message " << endl;
177       cerr << "could be added to the queue " << endl;
178    }
179    if(StatusCode == −1){
180       cerr << "unknown error in mq_send() " << endl;
181    }
182    return(StatusCode);
183
184  }

In Line 150, the message is checked to ensure that its size does not exceed the allowable size for a message. In Line 154 the call to mq_send() is made. All other code checks for errors.

Example 7-4 is the definition of receive().

//Example 7-4 The definition of receive().

187  int posix_queue::receive(string &Msg)
188  {
189
190     int StatusCode = 0;
191     char QueueBuffer[QueueAttr.mq_msgsize];
192     ReceivedBytes = mq_receive(PosixQueue,QueueBuffer,
                                   QueueAttr.mq_msgsize,NULL);
193     if(errno == EAGAIN){
194        StatusCode = errno;
195        cerr << "O_NONBLOCK not set and the queue is full " << endl;
196
197     }
198     if(errno == EBADF){
199        StatusCode = errno;
200        cerr << "PosixQueue is not a valid descriptor open for writing"
                << endl;
201     }
202     if(errno == EINVAL){
203        StatusCode = errno;
204        cerr << "msgprio is out side of the priority range for the message
                    queue or " << endl;
205        cerr << "Thread my block causing a timing conflict with time out"
                << endl;
206     }
207     if(errno == EMSGSIZE){
208        StatusCode = errno;
209        cerr << "message size exceeds maximum size of message parameter on
                    message queue" << endl;
210     }
211     if (errno == ETIMEDOUT){
212        StatusCode = errno;
213        cerr << "The O_NONBlock flag was not set, but the time expired
                    before the message " << endl;
214        cerr << "could be added to the queue " << endl;
215     }
216     string XMessage(QueueBuffer,QueueAttr.mq_msgsize);
217     Msg = XMessage;
218     return(StatusCode);
219
220  }

In Line 191, a buffer QueueBuffer with the maximum size of a message is created. mq_receive() is called. The message returned is stored in QueueBuffer, and the number of bytes is returned and stored in ReceivedBytes. In Line 216, the message is extracted from QueueBuffer and assigned to the string in Line 217.

In Example 7-5 is the definition for remove().

//Example 7-5 The definition for remove().

221  int  posix_queue::remove(void)
222  {
223     int StatusCode = 0;
224     StatusCode = mq_unlink(QueueName.c_str());
225     if(StatusCode != 0){
226        cerr << "Did not unlink " << QueueName << endl;
227     }
228     return(StatusCode);
229  }
230

In Line 224, mq_unlink() is called to remove the message queue from the system.

Example 7-6 provides the definition for close().

//Example 7-6 The definition for close().

231  int posix_queue::close(void)
232 {
233
234     int StatusCode = 0;
235     StatusCode = mq_close(PosixQueue);
236     if(errno == EBADF){
237        StatusCode = errno;
238        cerr << "PosixQueue is not a valid descriptor open for
                    writing" << endl;
239     }
240    if(StatusCode == −1){
241        cerr << "unknown error in mq_close() " << endl;
242    }
243    return(StatusCode);
244
245  }

In Line 235, mq_close() is called to close the message queue.

Return briefly to Listing 7-3, notice that the bolded methods from Examples 7-2 through 7-6 encapsulate the message queue's attributes to set and return the properties of the message queue:

33  int queueFlags(void);
34  void queueFlags(int X);
35
36  int queuePriority(void);
37  void queuePriority(int X);
38

39  int maxMessages(void);
40  void maxMessages(int X);
41  int messageSize(void);
42  void messageSize(int X);

Some of these attributes are can also be set in the constructor. There are three constructors in Listing 7-3:

25  posix_queue(void);
26  posix_queue(string QName);
27  posix_queue(string QName,int MaxMsg, int MsgSize);

At Line 25 is the default constructor. Example 7-7 shows the definition.

// Example 7-7 The definition of the default constructor.

 4  posix_queue::posix_queue(void)
 5  {
 6
 7
 8     QueueFlags = O_RDWR | O_CREAT | O_EXCL;
 9     OMode = S_IRUSR | S_IWUSR;
10     QueueName.assign("none");
11     QueuePriority = 0;
12     MaximumNoMessages = 10;
13     MessageSize = 8192;
14     ReceivedBytes = 0;
15     setQueueAttr();
16
17
18  }

This posix_queue class is a simple model of the message queue. All functionality has not been included here, but you can see a message queue class makes the message queue easier to use. The posix_queue class performs error checking for all of the major functions of the IPC mechanism. What should be added is the mq_notify function. With notification signaling, a process is signaled when the empty message queue has a message. This class does not have synchronization capabilities. If multiple processes want to use the posix_queue to write to it, a built-in mutex should be implemented and used when messages are sent or received.

program7-6.cc (Listing 7-6)

For this program, there is a global variable Answer declared in the main line in the file program7-6.cc. It is declared as extern, so it can have global scope in thread_tasks.cc. Answer is to be processed by ThreadA, ThreadB, and then ThreadC. They are to perform their calculations, which requires synchronization. The correct answer is 165. Although a pipeline model of data communication is what was desired, there is no synchronization in place for it to be executed.

libpthread

<iostream> <pthread.h> "thread_tasks.h"

c++ -o program7-6 program7-6.cc thread_tasks.cc -lpthread

Solaris 10, gcc 3.4.3 and 3.4.6

Opteron, UltraSparc T1

./program7-6

None

program7-8.cc (Listing 7-8)

The program in Listing 7-8 shows the code for the main thread. It declares the two global queues used for input and output. The strings are read in from a file into string queue QText, the data source queue for ThreadA. The main thread then calls a join on ThreadA and waits for it to return. When ThreadA returns, ThreadB uses the global queue SourceText just populated by ThreadA. When ThreadB returns, the strings in the global queue FilteredText are sent to cout by the main thread.

libpthread

<iostream> <pthread.h> <fstream> <queue> <string> "thread_tasks.h"

c++ -o program7-8 program7-8.cc thread_tasks.cc -lpthread

Solaris 10, gcc 3.4.3 and 3.4.6

Opteron, UltraSparc T1

./program7-8

None

With processes, command-line arguments are passed by using the exec family of functions or posix_spawn, as discussed in Chapter 5. The command-line arguments are restricted to simple data types such as numbers and characters. The parameters passed to processes are one-way communication. The child process simply copies the parameter values. Any modifications made to the data will not be reflected in the parent. With threads, the parameter is not a copy but an address to some data location. Any modification made by the thread to that data can be seen by any thread that uses it.

But this type of transparency may not be what you desire. A thread can keep its own copy of data passed to it. It can copy it to its stack, but what's on its stack will come and go. A thread can be called several times performing the same task over and over again. By using thread-specific data, data can be associated with a thread and made private and persistent.

The concurrent and exclusive read-write algorithms can be combined into the following types of algorithm combinations that are possible for read-write access:

These algorithms can be viewed as access policies implemented by the tasks sharing the data. Figure 7-7 shows these access policies. EREW means access to the shared memory is serialized where only one task at a time is given access to the shared memory whether it is access to write or to read. An example of EREW access policy is the producer-consumer. The program in Chapter 5 in Listing 5-7 has an EREW access policy with the shared posix_queue between processes. One process writes the name of a file another process is to search for the code in. Access to the queue that contained the filenames was restricted to exclusive write by the producer and exclusive read by the consumer. Only one task was allowed access to the queue at any given time.

CREW access policy allows multiple reads of the shared memory and exclusive writes. There are no restrictions on how many tasks can read the shared memory concurrently, but only one task can write to the shared memory. Concurrent reads can occur while an exclusive write is taking place. With this access policy, each reading task may read a different value while other task is writing. The next task that reads the shared memory will see different data than some other task. This may be intended, but it also may not. ERCW access policy is direct reverse of CREW. Only one task can read the shared data, but concurrent writes are allowed. CRCW access policy allows concurrent reads and concurrent writes.

Figure 7.7. Figure 7-7

Each of these four algorithm types requires different levels and types of synchronization. They can be analyzed on a continuum with the access policy that requires the least amount of synchronization to implement on one end and the access policy that requires the most amount of synchronization at the other end. EREW is the policy that is the simplest to implement because EREW essentially forces sequential processing. You may think that CRCW is the simplest, but it presents the most challenges. It may appear that memory can be accessed without restriction. But this is the most difficult to implement and requires the most synchronization in order to meet the goal to implement a synchronization process that maintains data integrity and satisfactory system performance.

Synchronization is also needed to coordinate the order of execution of concurrent tasks. Order of execution was important in the program in Listings 7-5 and 7-6. If the tasks were executed out of order, the final value for Answer would be wrong. In the program in Listings 7-7 and 7-8, if task1 did not complete, task2 would attempt to read from an empty queue. Synchronization is required to coordinate these tasks so that work can progress or so that the correct results can be produced. Data synchronization (access synchronization) and task synchronization (sequence synchronization) are two types of synchronization required when executing multiple concurrent tasks. Task synchronization enforces preconditions and postconditions of logical processes.

There are four basic synchronization relationships between any two tasks in a single process or between any two processes within a single application:

These four basic relationships characterize the coordination of work between threads and processes. Figure 7-8 shows activity diagrams for each synchronization relationship.

Figure 7.8. Figure 7-8

In a start-to-start synchronization, one task cannot start until another task starts. One task may start before the other but never after. For example, say that you have a program that implements an Embedded Conversational Agent (ECA). The ECA is a computer-generated talking head, which provides a kind of personality for software. The program that implements the ECA has several threads. Here, the focus is on the threads that controls the animation of the eyes (ECA does not have a mouth, the eyes animate) and the thread that controls the sound or voice. You want to give the illusion that the sound and eyes animation are synchronized. Ideally, they should execute at precisely the same moment. With multiple processor cores, both threads may start simultaneously. The threads have a start-to-start relationship. Because of timing conditions, the thread that produces the audio (Thread A) is allowed to start slightly before the thread that starts the animation (Thread B), but not much before for the illusion's sake. It takes a little longer for the audio to initialize, so it can start a bit early. Graphics load much faster. Figure 7-9 shows images of our ECA.

Figure 7.9. Figure 7-9

In a finish-to-start synchronization, Task A cannot finish until Task B starts. This type of relationship is common with parent-child processes. The parent process cannot complete execution of some operation until it spawns a child process or it receives a communication from the child process that it has started its operation. The child process continues to execute once it has signaled the parent or supplied the needed information. The parent process is then free to complete its operation.

A start-to-finish synchronization relationship is the reverse of the finish-to-start relationship. In a start-to-finish synchronization relationship, one task cannot start until another task finishes. Task A cannot start execution until Task B finishes executing or completes a certain operation. The program in Listing 7-7 and 7-8 had a start-to-finish synchronization. task2 could not start until task1 completed. The main thread used a join for synchronization. The main thread blocked until task1 returned then it created a thread that executed task2.

If process A is reading from a pipe connected to process B, process B must first write to the pipe before process A reads from it. Process B must complete at least one operation — write a single element to the pipe — before process A starts. With the pipe, there is limited synchronization built in by using blocking. But if there are multiple readers and writers, then more elaborate synchronization is required.

A finish-to-finish synchronization relationship means one task cannot finish until another task finishes. Task A cannot finish until Task B finishes. This again can describe the relationship between parent and child processes. The parent process must wait until all its child processes have terminated before it is allowed to terminate. If the parent process terminates before its child processes, those terminated child processes become zombied. The parent process calls a wait() for each of its child processes (like join for threads) or waits for a mutex or condition variable that is broadcast by child threads.

Another example of a finish-to-finish relationship is the boss-worker concurrency model. It is the boss's job to delegate work to the workers. It would be undesirable for the boss to terminate before the worker. New requests to the system would not be processed, existing threads would have no work to perform, and no new threads would be created. If the boss were a primary thread and it terminated, the process would terminate along with all the worker threads. In a peer-to-peer model, if thread A dynamically allocates an object passed to thread B and thread A terminates, the object is destroyed along with thread A. If this is done before thread B has a chance to use it, a segmentation fault or data access violation occurs. In order to prevent these kinds of errors with threads, termination of threads is synchronized by using the pthread_join(). This creates a finish-to-finish synchronization.

A semaphore can be accessed only by specific operations. There are two operations that can be performed on a semaphore: P() and V() operations. The P() operation decrements the semaphore and the V() operation increments the semaphore:

P(Mutex)
if(Mutex > 0){
  Mutex — ;
}
else {
    Block on Mutex;
}

V(Mutex)
if(Blocked on Mutex N processes){
  pass on Mutex;
}
else{
    Mutex++;
}

Here Mutex is the semaphore. The actual implementation will be system dependent. These operations are indivisible. This means that once the operation is in progress, it cannot be preempted. If several tasks attempt to make a call to the P() operation, only one task is allowed to proceed. If Mutex has already been decremented, then the task blocks and is placed in a queue. The V() operation is called by the task that owns Mutex. If there are other tasks waiting on Mutex, it is given to the next task in the queue according the scheduling policy. If no tasks are waiting, then Mutex is incremented.

Semaphore operations can go by other names such as:

P() operation
lock()
wait()
own()


V() operation
unlock()
post()
unown()

The value of the semaphore depends on the type of semaphore it is. There are several types of semaphores. For example:

The POSIX standard defines several types of semaphores. Some of these semaphores are used by threads only, and others can be used by processes or threads. Any operating system that is compliant with the Single Unix Specification or POSIX standard can supply an implementation of these semaphores. They are apart of the libpthread library, and the functions are declared in the pthread.h header.

The POSIX semaphore defines a named binary semaphore. The name corresponds to a pathname in the filesystem. Table 7-6 lists the basic functions for using a semaphore along with a brief description.

Listing 7-9 shows how semaphores can be used between multiple processes.

// Listing 7-9 A process using a semaphore on an output file.

 1  using namespace std;
 2  #include <semaphore.h>
 3  #include <iostream>
 4  #include <fstream>
 5
 6
 7  int main(int argc, char *argv[])
 8  {
 9
10     int Loop, PN;
11     sem_t *Sem;
12     const char *Name;
13     ofstream Outfile("out_text.txt",ios::app);
14
15     PN = atoi(argv[1]);
16     Loop = atoi(argv[2]);
17     Name = argv[3];
18
19     Sem = sem_open(Name,O_CREAT,O_RDWR,1);
20     sem_unlink(Name);
21
22
23     for (int Count = 1; Count < Loop; ++Count) {
24        sem_wait(Sem);
25        Outfile << "Process:" << PN << " counting " << Count << endl;
26        sem_post(Sem);
27
28     }
29     Outfile.close();
30
31     exit(0);
32
33
34
35  }

The program in Listing 7-9 opens a semaphore of type sem_t in Line 11. The named semaphore Sem is opened with Name typed in as the third argument on the command line. O_CREATE and O_RDWR are flags that specify how the semaphore is opened. In this case, the semaphore is opened only if is does not exist. With the O_RDWR flag set, the semaphore is opened with read and write permissions. Sem is initialized with a value of 1. The sem_wait and sem_post operations encapsulate the access to Outfile. During the execution of Line 25, no other processes should access the file. All processes that use this file for input or output should use the same semaphore. In Listing 7-10, a process that reads the file also uses the semaphore.

// Listing 7-10 A process using a semaphore on an input file.

 1  using namespace std;
 2  #include <semaphore.h>
 3  #include <iostream>
 4  #include <fstream>
 5  #include <string>
 6
 7
 8  int main(int argc, char *argv[])
 9  {
10
11     string Str;
12     const char *Name;
13     sem_t *Sem;
14     ifstream Infile("out_text.txt");
15
16     if(Infile.is_open()){
17        Name = argv[1];
18        Sem = sem_open(Name,O_CREAT,O_RDWR,1);
19        sem_unlink(Name);
20
21        while(!Infile.eof() && Infile.good()){
22           sem_wait(Sem);
23           getline(Infile,Str);
24           cout << Str << endl;
25           sem_post(Sem);
26
27        }
28        cout << " -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  - " << endl;
29
30        Infile.close();
31
32     }
33
34     exit(0);
35
36
37
38  }

The program in Listing 7-10 the named semaphore Sem is opened with Name typed in as the first argument on the command line. O_CREATE and O_RDWR are flags that specify how the semaphore is opened, as in Listing 7-9. The sem_wait and sem_post operations encapsulate the access to Infile. During the execution of Line 23, the process in Listing 7-9 cannot write to the file.

Here is Program Profile 7-4 for Listings 7-9 and 7-10.

program7-9.cc (Listing 7-9)

program7-10.cc (Listing 7-10)

The program7-9 in Listing 7-9 opens a semaphore of type sem_t named semaphore Sem. It is opened with Name typed in as the third argument on the command line. Sem is initialized with a value of 1. The sem_wait and sem_post operations encapsulate the access to Outfile. During the execution no other processes should access the file. All processes that use this file for input or output should use the same semaphore. In Listing 7-10, a process that reads the file also uses the semaphore. From the command line the program looks for the process number, a loop invariant, and the name of the semaphore. The program writes process, process number, and the loop iteration number to the file.

In program7-10 in Listing 7-10 the named semaphore Sem is opened with Name typed in as the first argument on the command line. It should have the same name as the semaphore in program7-9 in order to coordinate access to out_text.txt. The sem_wait and sem_post operations encapsulate the access to Infile. During the execution program7-9 cannot write to the file. The program requires the name of the semaphore as a command-line argument. It uses the named semaphore. It opens the file out_text.txt and writes its contents to stout.

librt

<semaphore.h> <iostream> <fstream> <string> <fcntl.h>

c++ -o program7-9 program7-9.cc -lrt
c++ -o program7-10 program7-10.cc -lrt

Solaris 10, gcc 3.4.3 and 3.4.6

Opteron, UltraSparc T1

These programs require command-line arguments. For program7-9, the first argument is the process number, the second argument is the loop invariant, and the third argument is the name of the semaphore. program7-10 requires the name of the semaphore.

./program7-9 3 4 /sugi & ./program7-10 /sugi

Make sure that the name of the semaphore contains a "/". These programs are to execute at the same time.

The POSIX standard defines a mutex semaphore of type pthread_mutex_t that can be used by threads and processes. Mutex means mutual exclusion. A mutex is a type semaphore, but there is a difference between them. A mutex must always be unlocked by the thread that locked it. With a semaphore, a post (or unlock) can be performed by a thread other than the thread that performed the wait (or unlock). So, one thread or process can call wait() and another process/thread can call post() on the same semaphore.

pthread_mutex_t provides the basic operations necessary to make it a practical synchronization mechanism:

Table 7-7 lists the pthread_mutex_t functions that are used to perform these basic operations. The initialization allocates memory required to hold the mutex semaphore and to give the memory some initial values. A binary semaphore has an initial value of 0 or 1. A counting semaphore has a value that represents the number of resources the semaphore is to track. It can represent the request limit a program is capable of processing in a single session. In contrast to regular variables, there is no guarantee that the initialization operation of a mutex will occur. Be sure to take precautions to ensure that the mutex was initialized by checking the return value or checking the errno value. The system fails to create the mutex if the space set aside for mutexes has been used, the number of allowable semaphores has been exceeded, the named semaphore already exists, or there is some other memory allocation problem.

int pthread_mutex_init(pthread_mutex_t *restrict
mutex, const pthread_mutexattr_t *restrict attr);

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

<time.h

>int pthread_mutex_lock(pthread_mutex_t *mutex);

int pthread_mutex_timedlock(pthread_mutex_t *restrict
mutex, const struct timespec *restrict abs_timeout);

int pthread_mutex_unlock (pthread_mutex_t *mutex);

int pthread_mutex_trylock(pthread_mutex_t *mutex);

int pthread_mutex_destroy(pthread_mutex_t *mutex);

The pthread mutex has an attribute object that encapsulates all the attributes of the mutex. This attribute object is used similarly to the attribute object for a thread. This difference is that whereas the attribute object of a thread is set, the attribute for a mutex has no set group of properties associated with it. We will discuss this later in the chapter. What is important to understand for now is that the attribute object can be passed to the initialization function creating a mutex with attributes of those set in the object. If no attribute object is used, the mutex is initialized with default values. The pthread_mutex_t is initially unlocked and private. A private mutex is shared between threads of the same process, whereas a shared mutex is shared between threads of multiple processes. If default attributes are to be used, the mutex can be initialized statically by using the macro:

pthread_mutext Mutex = PTHREAD_MUTEX_INITIALIZER;

This creates a statically allocated mutex object. This method uses less overhead but performs no error checking.

The request ownership operation grants ownership of the mutex to the calling process or thread. The mutex is either owned or unowned. Once owned, the thread or process owning it has exclusive access to the resource. If there is any attempt to own the mutex (by calling this operation) by any other processes or threads, they are blocked until the mutex is made available. When the mutex is released, this causes the next process or thread that has blocked to unblock and obtain ownership of the mutex. With the pthread_mutex_lock() the thread granted ownership of a given mutex is the only thread that can release the mutex.

The try ownership operation tests the mutex to see if it is already owned. The function returns some value if it is owned. The advantage of this operation is the thread or process does not block. It can continue executing code. If the mutex is not owned, then ownership is granted.

The destruction operation frees the memory associated with the mutex. The memory cannot be destroyed or closed if it is owned or a thread or process is waiting for the mutex.

The pthread_mutex_t has an attribute object used in a similar way as the thread attribute. As previously indicated, the attribute object encapsulates the attributes of a mutex object. Once initialized, it can be used by multiple mutex objects when passed to pthread_mutex_init(). Also, as previously indicated, in contrast to the thread attribute function, there are no mandatory attributes associated with the object. The functions that can be used to set the mutex attributes have to do with the following:

These functions are listed in Table 7-8 with a brief description.

int pthread_mutexattr_init(pthread_mutexattr_t * attr);

int pthread_mutexattr_destroy(pthread_mutexattr_t * attr);

int
pthread_mutexattr_
setprioceiling(pthread_
mutexattr_t * attr,int
prioceiling);

int
pthread_mutexattr_
getprioceiling(const pthread_
mutexattr_t*restrict attr,int
*restrict prioceiling);

int
pthread_mutexattr_
setprotocol(pthread_
mutexattr_t * attr,int
protocol);

int
pthread_mutexattr_
getprotocol(const pthread_
mutexattr_t *restrict attr,
int *restrict protocol);

PTHREAD_PRIO_NONE

PTHREAD_PRIO_INHERIT

PTHREAD_PRIO_PROTECT

int pthread_mutexattr_
setpshared(pthread_mutexattr_
t * attr,int pshared);

int pthread_mutexattr_
getpshared(const pthread_
mutexattr_t *restrict
attr,int *restrict pshared);

PTHREAD_PROCESS_SHARED

PTHREAD_PROCESS_PRIVATE

int pthread_mutexattr_
settype(pthread_mutexattr_t *
attr,int type);

int pthread_mutexattr_
gettype(const pthread_
mutexattr_t *restrict
attr,int *restrict type);

PTHREAD_MUTEX_DEFAULT

PTHREAD_MUTEX_RECURSIVE

PTHREAD_MUTEX_ERRORCHECK

PTHREAD_MUTEX_NORMAL

To use a phtread mutex between threads of different processes requires the shared attribute. This attribute determines if the mutex is private or shared. Private mutexes are shared only among threads of the same process. They can be declared as global or a handle can be passed between threads. Shared mutexes, however, are used by any thread that has access to the mutex memory, and this includes threads of different processes. To do this, use pthread_mutexattr_setshared() and set the attribute to PTHREAD_PROCESS_SHARED as follows:

pthread_mutexattr_setpshared(&MutexAttr,PTHREAD_PROCESS_SHARED);

This allows Mutex to be shared by threads of different processes.

Figure 7-11 contrasts the idea of private and shared mutexes. If threads of different processes are to share a mutex, that mutex must be allocated in memory shared between processes. We discussed shared memory earlier in this chapter. Mutexes between processes can be used to protect critical sections that access files, pipes, shared memory, and devices.

Figure 7.11. Figure 7-11

Mutexes can be used to manage critical sections of processes and threads in order to control race conditions. Mutexes avoid race conditions by serializing access to the critical section. Example 7-8 shows code for the new tasks defined in Listing 7-5. Mutexes are used to protect their critical sections.

// Example 7-8 New code for tasks in Listing 7-5.

 3  void *task1(void *X)
 4  {
 5     pthread_mutex_lock(&Mutex);
 6     Answer = Answer * 32; //critical section
 7     pthread_mutex_unlock(&Mutex);
 8     cout << "thread A Answer = " << Answer << endl;
 9
10  }

In Example 7-8, task1 now uses a mutex when it modifies the global variable Answer. In Line 8 the task sends the new value of Answer to cout. This is the critical section for the task. Now you can have each task execute the same code except each task sends to stout its thread name. So, now this is the output:

Before threads Answer = 10
thread 1 Answer = 320
thread 2 Answer = 160
thread 3 Answer = 165
After threads Answer = 165

Read-write locks can be used to implement a CREW access policy. Several tasks can be granted concurrent reads, but only one task is granted write access. Using read-write locks can keep concurrent reads from occurring with the exclusive write. Example 7-9 contains tasks using read-write locks to protect critical sections.

// Example 7-9 Threads using read-write locks.
//...

pthread_t ThreadA,ThreadB,ThreadC,ThreadD;
pthread_rwlock_t RWLock;

void producer1(void *X)
{
   pthread_rwlock_wrlock(&RWLock);

//critical section
   pthread_rwlock_unlock(&RWLock);
}

void producer2(void *X)
{
   pthread_rwlock_wrlock(&RWLock);
 //critical section
   pthread_rwlock_unlock(&RWLock);
}

void consumer1(void *X)
{
   pthread_rwlock_rdlock(&RWLock);
 //critical section
   pthread_rwlock_unlock(&RWLock);
}

void consumer2(void *X)
{
   pthread_rwlock_rdlock(&RWLock);
 //critical section
   pthread_rwlock_unlock(&RWLock);
}

int main(void)
{

   pthread_rwlock_init(&RWLock,NULL);
   //set mutex attributes
   pthread_create(&ThreadA,NULL,producer1,NULL);
   pthread_create(&ThreadB,NULL,consumer1,NULL);
   pthread_create(&ThreadC,NULL,producer2,NULL);
   pthread_create(&ThreadD,NULL,consumer2,NULL);
   //...
   return(0);
}

In Example 7-9 four threads are created. Two threads are producers, ThreadA and ThreadC, and two threads are consumers, ThreadB and ThreadD. All the threads have a critical section protected by the read-write lock, RWLock. ThreadB and ThreadD can enter their critical sections concurrently or serially, but neither thread can enter its critical section if either ThreadA or ThreadC is in its. ThreadA and ThreadC cannot enter their critical sections concurrently. Table 7-11 shows part of the decision table for this program.

Listing 7-11 is the declaration of an object-oriented mutex class.

// Listing 7-11 Declaration of an object-oriented mutex class.


 1  #ifndef _PERMIT_H
 2  #define_PERMIT_H
 3  #include <pthread.h>
 4  #include <time.h>
 5  class permit{
 6  protected:
 7     pthread_mutex_t Permit;
 8     pthread_mutexattr_t PermitAttr;
 9  public:
10     permit(void);
11     bool available(void);
12     bool not_in_use(void);
13     bool checkAvailability(void);
14     bool  availableFor(int secs,int nanosecs);
15  };
16
17
18  #endif/* _PERMIT_H */

The permit class provides the basic operations for a mutex class. Listing 7-12 contains the definition of the permit class.

// Listing 7-12 Definition of the permit class.

 1  #include "permit.h"
 2
 3
 4  permit:: permit(void)
 5  {
 6     int AValue,MValue;
 7     AValue = pthread_mutexattr_init(&PermitAttr);
 8     MValue = pthread_mutex_init(&Permit,&PermitAttr);
 9  }
10  bool permit::available(void)
11  {
12     int RC;
13     RC = pthread_mutex_lock(&Permit);
14     return(true);
15
16  }
17  bool permit::not_in_use(void)
18  {
19     int RC;
20     RC = pthread_mutex_unlock(&Permit);
21     return(true);
22
23  }
24  bool permit::checkAvailability(void)
25  {
26     int RC;
27     RC = pthread_mutex_trylock(&Permit);
28     return(true);
29  }
30  bool permit::availableFor(int secs,int nanosecs)
31  {
32     //...
33     struct timespec Time;
34     return(true);
35
36  }

Listing 7-12 shows only the basic operations. For the class to be a fully viable mutex class, error checking would have to be added, as was the case in the posix_queue class in Example 7-3 earlier in the chapter.

Here is Program Profile 7-5 for Listings 7-11 and 7-12.

permit.h (Listing 7-11)

permit.cc (Listing 7-12)

Listing 7-11 contains the header for permit.h, and Listing 7-12 contains permit.cc.

libpthread

<pthread.h> <time.h> "permit.h"

c++ -c permit.cc

Solaris 10, gcc 3.4.3 and 3.4.6

Opteron, UltraSparc T1

N/A

None

The condition variable and the mutex can be used to implement the synchronization relationships mentioned earlier in the chapter:

These relationships can exist between threads of the same processes or different processes. Listing 7-13 contains a program demonstrating how to implement SF synchronization relationship. There are two mutexes used in each example. One mutex is used to synchronize access to the shared data, and the other mutex is used with the condition variable.

// Listing 7-13 SF synchronization relationship implemented with
// condition variables and mutexes.

 1  using namespace std;
 2  #include <iostream>
 3  #include <pthread.h>
 4
 5  int Number;
 6  pthread_t ThreadA,ThreadB;
 7  pthread_mutex_t Mutex,EventMutex;
 8  pthread_cond_t Event;
 9
10  void *worker1(void *X)
11  {
12     for(int Count = 1;Count < 10;Count++){
13        pthread_mutex_lock(&Mutex);
14        Number++;
15        pthread_mutex_unlock(&Mutex);
16        cout << "worker1: Number = " << Number << endl;
17        if(Number == 7){
18           pthread_cond_signal(&Event);
19        }
20     }
21
22     return(0);
23  }
24
25  void *worker2(void *X)
26  {
27     pthread_mutex_lock(&EventMutex);
28     pthread_cond_wait(&Event,&EventMutex);
29     pthread_mutex_unlock(&EventMutex);
30     for(int Count = 1;Count < 10;Count++){
31        pthread_mutex_lock(&Mutex);
32        Number = Number + 20;
33        cout << "worker2: Number = " << Number << endl;
34        pthread_mutex_unlock(&Mutex);
35
36     }
37
38     return(0);
39  }
40
41
42  int main(int argc, char *argv[])
43  {
44     pthread_mutex_init(&Mutex,NULL);

45     pthread_mutex_init(&EventMutex,NULL);
46     pthread_cond_init(&Event,NULL);
47     pthread_create(&ThreadA,NULL,worker1,NULL);
48     pthread_create(&ThreadB,NULL,worker2,NULL);
49
50     pthread_join(ThreadA,NULL);
51     pthread_join(ThreadB,NULL);
52
53     return (0);
54  }

In Listing 7-13, the SF synchronization relationship is implemented. ThreadB cannot start until ThreadA finishes. ThreadA signals to ThreadB once Number has a value of 7. It can then continue execution until finished. ThreadB cannot start its computation until it gets a signal from ThreadA. Both use the EventMutex with the condition variable Event. Mutex is used to synchronize write access to the shared data Number. A task can use several mutexes to synchronize different critical sections and synchronize different events. These techniques can easily be used to synchronize order of execution between processes.

Here is Program Profile 7-6 for Listing 7-13.

program7-13.cc. (Listing 7-13)

program7-13.cc (Listing 7-13) has a SF synchronization relationship. ThreadB cannot start until ThreadA finishes. ThreadA signals to ThreadB once Number has a value of 7. It can then continue execution until finished. ThreadB cannot start its computation until it gets a signal from ThreadA. Both use the EventMutex with the condition variable Event. Mutex is used to synchronize write access to the shared data Number. ThreadA and ThreadB send Number to stout. ThreadA adds 1 to the value of Number and ThreadB adds 20 to Number at each iteration through the loop.

libpthread

<iostream> <pthread.h>

c++ -o program7-13 program7-13.cc -lpthread

Solaris 10, gcc 3.4.3 and 3.4.6

Opteron, UltraSparc T1

./program7-13

None