dvthread Documentation

Author:

dvermeir@vub.ac.be

Download

download directory

Installation

Standard. See the INSTALL file in the top directory.

./configure takes a --enable-sig-test option, see Dv::Thread::Thread::sig_set_handler.

Description

This package provides two main classes: Dv::Thread::Thread, representing threads, and Dv::Thread::Monitor, to represent monitors. In addition, there is a trivial convenience class Dv::Thread::Lock that can be used to conveniently ensure exclusive access for member functions to *this (where *this is an instance of (a subclass of) Dv::Thread::Monitor. There is also a class Dv::Thread::logstream which is a thread-safe version of Dv::Util::logstream. The Dv::Thread::Barrier class wraps around pthread barriers.

Example program

The following class implements a buffer for concurrent access. Note the use of the Lock object which ensures that the get() and put() member functions have exclusive access to the Buffer object. Conditions are used to signal threads waiting to put or get items into (from) the buffer.

// $Id: test-buffer.h,v 1.3 2008/12/21 09:40:47 dvermeir Exp $
#ifndef DV_THREAD_BUFFER_H
#define DV_THREAD_BUFFER_H
// $Id: test-buffer.h,v 1.3 2008/12/21 09:40:47 dvermeir Exp $
#include <dvthread/lock.h>

// This class represents a buffer with synchronized access.
class Buffer: public Dv::Thread::Monitor {
  public:       
    static const std::string GET; // ("get");
    Buffer(): Dv::Thread::Monitor("buffer",2), // &std::cerr),
      n_items_(0) {}
    // Add an item to a buffer.
    void put(int i) {
      Dv::Thread::Lock lock(*this, "put"); // Get exclusive access to *this.
      while (n_items_==MAX)
        if (!wait(OK_TO_PUT,2000)) // Wait at most 2 secs.
          throw std::runtime_error("Buffer::put() timed out");
      std::cerr << "put " << i << " ";
      data_[n_items_++] = i; // Actually put the item.
      signal(OK_TO_GET);
    }
  
     // Obtain and remove the last item in the buffer.
    int get() {
      Dv::Thread::Lock lock(*this); // , &std::cerr); // Get exclusive access to *this.
      while (n_items_ == 0) {
        if (!wait(OK_TO_GET, 2000)) 
          throw std::runtime_error("Buffer::get() timed out");
      }
      int tmp = data_[--n_items_];
      std::cerr << GET  << tmp << " ";
      signal(OK_TO_PUT);
      return tmp;
    }

  private:
    // Buffer capacity.
    enum { MAX = 3 };
    // Names for conditions.
    enum { OK_TO_GET = 0, OK_TO_PUT = 1 }; /// conditions 
    // Number of items in the buffer.
    int n_items_;
    // Actual store for items.
    int data_[MAX];
};
#endif

The code below shows the implementation of a reader thread class for the above buffer. Each thread should implement a Thread::main() member function that describes the work to be done by the thread.

#ifndef DV_THREAD_READER_H
#define DV_THREAD_READER_H
// $Id: test-reader.h,v 1.3 2008/12/21 09:40:47 dvermeir Exp $
#include <unistd.h>
#include <dvthread/thread.h>
#include "test-buffer.h"


// Reader thread class. A reader simply retrieves items from a buffer.
class Reader: public Dv::Thread::Thread {
  public:
    // Constructor; n is the number of items to retrieve.
    Reader(Buffer& buf, unsigned int n): Thread(), buffer_(buf), n_(n) {}
    // The function executed by this tread.
    virtual int main() {
      try {
        std::cerr << "reader main, id = " << id() << std::endl;
        for (unsigned int i=0;(i<n_);++i)
          buffer_.get();
      }
      catch (std::exception& e) {
        std::cerr << "reader exception: " << e.what() << std::endl;
        return 1;
      }
      return 0;
    }
    // Destructor. Wait for this thread to finish before we detroy it.
    virtual ~Reader() { std::cerr << "Reader::~Reader" << std::endl; } 
  private:
    // Buffer from where items will be retrieved.
    Buffer&     buffer_;
    // Number of items to retrieve.
    unsigned int  n_;
};
#endif

A corresponding writer thread class is shown below.

#ifndef DV_THREAD_WRITER_H
#define DV_THREAD_WRITER_H
// $Id: test-writer.h,v 1.3 2008/12/21 09:40:47 dvermeir Exp $
#include <dvthread/thread.h>
#include <unistd.h>
#include "test-buffer.h"

// A Thread class that writes 100 items to a buffer.
class Writer: public Dv::Thread::Thread {
  public:       
     // Constructor; n is the number of items to put.
    Writer(Buffer& buf, unsigned int n): Thread(), buffer_(buf), n_(n) {}
    // Main function of a Writer thread.
    int main() {
      std::cerr << "writer main, id = " << id() << std::endl;
      try {
        for (unsigned int i=0; (i<n_);++i)
          buffer_.put(i);
      }
      catch (std::exception& e) {
        std::cerr << e.what() << std::endl;
        return 1;
      }
      return 0;
    }
    // Destructor, join this thread first.
    virtual ~Writer() { std::cerr << "Writer::~Writer" << std::endl; }
  private:
     // Buffer to which items will be written.
    Buffer& buffer_;
    // Number of items to write.
    unsigned int n_;
};
#endif

The main program creates a Buffer object, a Reader and a Writer thread. Note that the writer wants to put more items than will be read. Consequently, it will throw an exception (from within Buffer::put()).

// $Id: test-thread.C,v 1.10 2008/12/21 09:40:47 dvermeir Exp $

#include <iostream>
#include <dvthread/logstream.h>
#include "test-buffer.h"
#include "test-reader.h"
#include "test-writer.h"

const std::string Buffer::GET("get");
// Test thread program. The test consists of a reader and a writer
// thread that read/write items to a buffer.
int
main(int,char**) {
  try {
    Dv::Thread::logstream log(std::cerr, "test-thread");
    Dv::Thread::thread_debug.set_log(&log);
    Dv::Thread::thread_debug.set_debug(0);

    Buffer buf; // buffer containing items
    Reader r(buf,10); // takes items from buffer
    Writer w(buf,15); // writes items to buffer
  
    r.start();
    std::cerr << "reader started.." << std::endl;
    w.start();
    std::cerr << "writer started.." << std::endl;
  
    r.join();
    if (r.status() != 0)
      throw std::runtime_error("reader has unexpected status");
    w.join();
    if (w.status() != 1)
      throw std::runtime_error("reader has unexpected status");
  }
  catch (std::exception& e) {
    std::cerr << "exception: " << e.what() << std::endl;
    return 1;
  }
return 0;
}