11.3 Maintaining the Stream Metaphor

Besides using interface classes to simplify and to create new fat interfaces for the parallel and message-passing libraries, we may also want to extend existing interfaces. For instance, the object-oriented stream metaphor can be extended to pipes, fifos, and the message passing libraries like PVM and MPI. These components are used to accomplish IPC (Inter-Process Communication), ITC (Inter-Thread Communication), and in some cases OTOC (Object-to-Object Communicaton). When communication occurs between concurrently executing threads or processes, then the communication channel may be a critical section. That is, if two or more routines are attempting to update the same pipe, fifo, or message buffer at the same time, then a race condition is present. If we are going to extend the object-oriented stream interface to include components from the PVM or MPI library, then we need to make sure that the streams we design are concurrency safe. Here is where our synchronization components that were designed as interface classes come in handy. Let's look at a simple pvm_stream class.

Example 11.12 Declaration of pvm_stream class that inherits mios class.

 class pvm_stream : public mios{ protected:    int TaskId;    int MessageId;    mutex Mutex;    //... public:    void taskId(int Tid);    void messageId(int Mid);    pvm_stream(int Coding = PvmDataDefault);    void reset(int Coding = PvmDataDefault);    pvm_stream &operator<<(string &Data);    pvm_stream &operator>>(string &Data);    pvm_stream &operator>>(int &Data);    pvm_stream &operator<<(int &Data);    //... }; 

This stream class will be used to encapsulate the state of the active buffer in a PVM task. The inserter operator << and the extractor operator >> will be used to send and retrieve messages between PVM processes. Here we only show operators for string and int types. The interface for this class is far from complete. Since this class could possibly be used with any datatype, we have to expand the inserter and extractor definitions. Since we plan to use the pvm_stream class within a multithreaded program, we want to make sure that the pvm_stream object is thread safe. Therefore, we include a mutex class as a member of our pvm_stream class. The pvm_stream class also encapsulates the active buffer for the PVM task. The stream can direct the message to a particular PVM task. The goal is to use the ostream and istream classes as a guide for the type of functionality that the pvm_stream class should have. Recall that istream and ostream classes are translator classes. They translate datatypes into generic streams of bytes for output and from generic streams of bytes to specific datatypes on input. Using the istream and ostream classes, the programmer does not have to get bogged down in details of what datatype is being inserted or extracted from a stream. We want the pvm_stream to behave in the same manner. The PVM has a different function for every type that needs to be packed into or unpacked from a send or receive buffer. For instance:

 pvm_pkdouble() pvm_pkint() pvm_pkfloat() 

are used to pack double s, int s, and float s. There are similar functions for the other datatypes that C++ uses. We would like to retain our stream metaphor where input and output is seen as a generic stream of bytes moving into or out of the program. Therefore, we define the << inserter operator and the >> extractor operator for every type we wish to exchange between PVM tasks . Furthermore, we also model the stream state after istream and ostream classes. The istream and ostream classes have an ios component that is used to hold the state of the stream. The stream may be in an error state, or the stream may be in one of several different numeric states such as octal, decimal, and hexadecimal. The stream may be in a good state, a locked state, an end-of-file state, and so on. The pvm_stream class should have a component that maintains the state of the stream and should have methods that set, reset, and report the PVM stream state. Our pvm_stream class contains a mios component for this purpose. The mios component maintains the state of the stream and the active send and receive buffer. Figure 11-4 contains a class diagram showing the relationships between the major classes in the iostream class library and how the pvm_stream class compares .

Figure 11-4. The class diagram showing the relationship between the major classes in the iostream class library and the class diagram for the pvm_stream class.

Notice that the istream and ostream classes inherit an ios class. The ios class maintains the stream state and buffer state of the istream and ostream class. Our mios class plays the same role for the pvm_stream class. The istream and ostream classes contain the definitions for the inserter << and extractor >> operators. The operators are defined by our pvm_stream class. So although our pvm_stream class is not related to the iostream classes by inheritance, it is related by interface. We use the interface of the iostream classes to dictate a semi-fat interface for the pvm_stream and mios classes. Notice in Figure 11-4 that the mios class is inherited by the pvm_stream class. We want to maintain the stream metaphor with the pvm_stream class. The notion of an interface class is used to accomplish this.

11.3.1 Overloading the << , >> Operators for PVM Streams

Let's take a look at how the << inserter operators and >> extractor operators are defined for the pvm_stream class. The << inserter operator is used to wrap the pvm_send and pvm_pk routines. A method definition that looks something like:

Example 11.13 Definition of << operator for the pvm_stream class.

 pvm_stream &pvm_stream::operator<<(int Data) {    //...    reset();    pvm_pkint(&Data,1,1);    pvm_send(TaskId,MessageId);    //...    return(*this); } 

is provided for each type that will be used with the pvm_stream class. The reset() method is inherited from the mios class. This method is used to clear or initialize the send buffer. TaskId and MessageId are data members of the pvm_stream class and are set with the taskId() and messageId() methods. The inserter method allows us to send data to a PVM task with the standard stream notation:

 int  Value = 2004; pvm_stream   MyStream; //... MyStream   << Value; //... 

The >> extractor operators are used in a similar manner to receive messages from PVM tasks. The >> operator is used to wrap the pvm_recv() and pvmupk routines. Extractor operators can be defined as:

Example 11.14 Definition of operator >> for the pvm_stream class.

 pvm_stream &pvm_stream::operator>>(int &Data) {    int BufId;    //...    BufId = pvm_recv(TaskId,MessageId);    StreamState = pvm_upkint(&Data,1,1);    //...    return(*this); } 

This type of definition will allow us to receive messages from PVM tasks using the extractor operator.

 int Value; pvm_stream MyStream; MyStream >> Value; 

Because the operator returns a reference to the pvm_stream , the insertion and extraction operators may be strung together:

 Mystream << Value1 << Value2; Mystream >> Value3 >> Value4; 

Using this syntax, the programmer is isolated from the more cumbersome syntax of the pvm_send , pvm_pk , pvm_upk , and pvm_recv routines and the more familiar object-oriented stream metaphor can be used. In this case, the stream represents a message buffer and the items that are inserted and extracted from the streams represent messages that are being exchanged between PVM processes. Recall that each PVM process has a separate address space. So not only do the << insertion and >> extraction operators disguise the pvm_send and pvm_recv calls, they also mask the underlying socket activity. Since the pvm_stream class might be used in a multithreaded environment, the insertion and extraction operators need to be thread safe.

The class diagram in Figure 11-4 shows that the pvm_stream class contains a mutex class. The mutex class can be used to protect the critical sections that are present in the pvm_stream class. The pvm_stream class encapsulates access to the send buffer and the receive buffer. Figure 11-5 shows how threads and the pvm_stream class interact with the pvm_send and pvm_receive buffers.

Figure 11-5. The interaction between the threads, the pvm_stream class, and the pvm_send and pvm_receive buffers.

Not only are the send and receive buffers critical sections, the mios class used to store the state of the pvm_stream class is also a critical section. The mutex class can be used to protect this component as well.

The Mutex object can be used in the call to the insertion and extraction operators.

Example 11.15 Definition of operator << and operator >> for the pvm_stream class.

 pvm_stream &pvm_stream::operator<<(int Data) {    //...    Mutex.lock();    reset();    pvm_pkint(&Data,1,1);    pvm_send(TaskId,MessageId);    Mutex.unlock();    //...    return(*this); } pvm_stream &pvm_stream::operator>>(int &Data) {    int BufId;    //...    Mutex.lock();    BufId = pvm_recv(TaskId,MessageId);    StreamState = pvm_upkint(&Data,1,1);    Mutex.unlock();    //...    return(*this); } 

This kind of scheme can be used to make the pvm_stream class thread safe. We don't show the exception handling code or the extra processing that would be included to prevent indefinite postponement or deadlock. The idea here is to focus on the components and architectures that can be used to support concurrency. The mutex interface class and the pvm_stream class can be reused and both support concurrency programming. For our purposes, the pvm_stream objects are assumed to be used by the receiving and the sending PVM task. However, this is not a strict requirement. In order for the user to use the pvm_stream class concept with user-defined classes, the insertion operator ( << ) and the extraction operator ( >> ) must be defined for the user-defined class.