RWTPCValQueueGuarded<Type>

Threads Module Reference Guide
Rogue Wave web site: Home Page | Main Documentation Page

RWTPCValQueueGuarded<Type>

Module: Threads Package: Interthread Communication


RWTPCValQueueGuarded<Type>  
    RWTPCValBufferBaseGuarded<Type,RWTGuardDecorator<Type>>

Local Index
Header File
Description
Example
Public Constructor
Public Destructor
See Also

Local Index

Members

RWTPCValQueueGuarded()

~RWTPCValQueueGuarded()

Header File

#include <rw/itc/RWTPCValQueueGuarded.h>

Description

RWTPCValQueueGuarded<Type> is a first-in-first-out (FIFO) queue that provides producer-consumer synchronization semantics for exchanging guarded values between cooperating threads.

In the producer-consumer synchronization model, reader threads (consumers) are blocked while the queue is empty, and writer threads (producers) are blocked while the queue is full. The queue is considered full when the number of unread entries equals or exceeds some user-specified maximum capacity.

The write operations inherited by this class bind a guard functor to each value prior to storing that value in an internal buffer. A guard functor is used during read operations to determine whether the associated value is currently eligible for retrieval from the buffer.

Example

#include <rw/itc/RWTPCValQueueGuarded.h>
#include <rw/thread/RWThreadFunction.h>  // for RWThreadFunction
#include <rw/functor/functorR0.h>        // for RWTFunctorR0<>
#include <iostream.h>

enum Command { RUN, STOP, PROCESS, EXIT };
RWTPCValQueueGuarded<Command>  pcQueue;
bool isRunning = false;

// The guard function
bool canProcess() { return isRunning; }

void reader(void)
{
   // Get commands from queue, update state, and write to stdout
   Command command;
   while (EXIT != (command = pcQueue.read())) {
     switch(command) {
       case RUN: cout<<"RUN"<< endl; isRunning=true; break;
       case STOP: cout<<"STOP"<<endl; isRunning=false; break;
       case PROCESS: cout<<"PROCESS"<<endl; break;
     }
   }
   cout<<"EXIT"<<endl;
}
     
int main()
{
   RWThread thread = rwtMakeThreadFunctionG(void,reader);
   RWTFunctorR0<bool> guard =
             rwtMakeFunctorR0G(bool,bool,canProcess);

   thread.start();
   pcQueue.write(PROCESS,guard);  // PROCESS commands initially
   pcQueue.write(PROCESS,guard);  // blocked by guards...
   pcQueue.write(PROCESS,guard);
   pcQueue.write(RUN);            // Until the state is changed
   pcQueue.write(PROCESS,guard);  // These won't block...
   pcQueue.write(PROCESS,guard);
   pcQueue.write(PROCESS,guard);
   pcQueue.write(STOP);          // Start blocking PROCESS again
   pcQueue.write(PROCESS,guard); // These get blocked...
   pcQueue.write(PROCESS,guard);
   pcQueue.write(PROCESS,guard);
   pcQueue.write(RUN);           // Let the reader read the rest
   pcQueue.write(EXIT);
   thread.join();
   return 0;
}

This code produces the following output:

RUN
PROCESS
PROCESS
PROCESS
PROCESS
PROCESS
PROCESS
STOP
RUN
PROCESS
PROCESS
PROCESS
EXIT

Public Constructor

RWTPCValQueueGuarded(size_t maxCapacity=0, 
                      bool isOpen=true);

Constructs a value-based, guarded, producer-consumer queue instance.

The parameter maxCapacity specifies the maximum number of unread entries allowed to accumulate within the queue. Once the number of entries in the queue equals this number, any thread attempting to write an additional entry is blocked until an entry is removed by a read operation, or until the capacity is increased. A capacity of zero, the default, indicates that the queue has no size limit, except as imposed by memory limitations, and that all write operations should complete without blocking.

The parameter isOpen is a bool value that specifies whether the queue should be initialized in the open state (true, the default) or the closed state (false).

Public Destructor

~RWTPCValQueueGuarded();

Destructor.