shm_queue.h

Go to the documentation of this file.

#ifndef SHM_QUEUE_H
#define SHM_QUEUE_H 1

#include <pthread.h>

#include <string>
#include <iostream>
#include <sstream>


template<typename T>
class shm_queue_node
{
 public:
  void* operator new(size_t s, shm_allocator& a)
    {
      return a.alloc(s);
    }
  void operator delete(void* p, shm_allocator& a)
    {
      a.free(p);
    }
  shm_ptr<T> data;
  shm_ptr<shm_queue_node> prev;
  shm_ptr<shm_queue_node> next;
};

template<typename T>
class shm_queue_header
{
 public:
  shm_queue_header( size_t s = 0 ) : size(s)
  {
    pthread_mutexattr_t shared_attr; 
    pthread_mutexattr_t fast_attr; 
    pthread_condattr_t cattr; 
    int rc; 
    head.set(0);
    tail.set(0);
    /* initialize a mutex and it's condition variables */
    rc = pthread_mutexattr_init(&shared_attr); 
    rc = pthread_mutexattr_init(&fast_attr); 
    rc = pthread_mutexattr_setpshared(&shared_attr, PTHREAD_PROCESS_SHARED);
    std::cout 
      << "pthread_mutexattr_setpshared(&shared_attr, PTHREAD_PROCESS_SHARED) => "
      << rc
      << std::endl
      << std::flush;
    /*
    rc = pthread_mutexattr_settype(&fast_attr, PTHREAD_MUTEX_FAST_NP );
    std::cout 
      << "pthread_mutexattr_settype(&fast_attr, PTHREAD_MUTEX_FAST_NP ) => "
      << rc
      << std::endl
      << std::flush;
    */
    rc = pthread_mutex_init( &head_lock, &shared_attr );
    //    rc = pthread_mutex_init( &tail_lock, &mattr );
    rc = pthread_condattr_init(&cattr); 
    rc = pthread_condattr_setpshared(&cattr, PTHREAD_PROCESS_SHARED);
    std::cout 
      << "pthread_condattr_setpshared(&cattr, PTHREAD_PROCESS_SHARED) => "
      << rc
      << std::endl
      << std::flush;

    rc = pthread_cond_init( &ready_for_push, &cattr ); 
    rc = pthread_cond_init( &ready_for_pop, &cattr );     
    return ;
  }
  void* operator new(size_t s, shm_allocator& a)
  {
      return a.alloc(s);
  }
  void operator delete(void* p, shm_allocator& a)
  {
    a.free(p);
  }

  size_t size;
  shm_ptr<shm_queue_node<T> > head;
  shm_ptr<shm_queue_node<T> > tail;
  pthread_mutex_t lock;
  pthread_mutex_t head_lock;
  pthread_mutex_t tail_lock;
  pthread_cond_t ready_for_push;
  pthread_cond_t ready_for_pop;
};
 
template<typename T>
class shm_queue
{
 public:
  shm_queue( shm_allocator& allocator, const size_t maxsize ) :
    a(allocator),
    MaxQueueSize( maxsize ),
    seg( a.get_segment() ) 
  {
    shm_header.reset( new (a) shm_queue_header<T>, seg );
    return; 
  };

  shm_queue( shm_allocator& allocator ) :
    a(allocator),
    seg( a.get_segment() ) 
  {
    shm_queue_header<T>* queue_head = static_cast<shm_queue_header<T>* >
                                      ((shm_queue_header<T>*)
                                      ((long)(seg.get_base())+a.get_overhead()));
    
    shm_header.reset( queue_head, seg );
    return; 
  };

  ~shm_queue() { return ; };

  void* operator new(size_t s, shm_allocator& a)
  {
    return a.alloc(s);
  }
  void operator delete(void* p, shm_allocator& a)
  {
    a.free(p);
  }

  void push(shm_ptr<T> obj);
  shm_ptr<T> pop();

  void lock_head() 
  {
    shm_queue_header<T>* header = ((shm_queue_header<T> *)(shm_header.get(seg)));
    pthread_mutex_lock( &(header->head_lock) );
  }

  void unlock_head() 
  {    
    shm_queue_header<T>* header = ((shm_queue_header<T> *)(shm_header.get(seg)));
    pthread_mutex_unlock( &(header->head_lock) );
  }

  void lock_tail() 
  {
    shm_queue_header<T>* header = ((shm_queue_header<T> *)(shm_header.get(seg)));
    pthread_mutex_lock( &(header->tail_lock) );
  }

  void unlock_tail() 
  {    
    shm_queue_header<T>* header = ((shm_queue_header<T> *)(shm_header.get(seg)));
    pthread_mutex_unlock( &(header->tail_lock) );
  }

  size_t size() const 
  {
    shm_queue_header<T>* header = ((shm_queue_header<T> *)(shm_header.get(seg)));
    return header->size;
  }
 private:
  size_t MaxQueueSize;
  //  shm_queue_header<T>* header;
  shm_ptr< shm_queue_header<T> > shm_header;

  shm_allocator& a;
  const shm_segment& seg;
 
  //disallowed (not implemented)
  shm_queue(const shm_queue& copy);
  shm_queue& operator=(const shm_queue& rhs);
};
 
 
template<typename T>
void shm_queue<T>::push(shm_ptr<T> obj)
{
  shm_queue_header<T>* header = static_cast<shm_queue_header<T>* >
                                      ((shm_queue_header<T>*)
                                      ((long)(seg.get_base())+a.get_overhead()));
  scoped_lock guard(&(header->head_lock)); 
  shm_queue_node<T>* node = new (a) shm_queue_node<T>;
  node->data = obj;
  node->prev.set(0);

  // queue head is not empty
  if ( header->head.off() != 0) {
    shm_queue_node<T>* cart = header->head.get( seg );

    std::cout 
      << "previous queue head : "
      << cart
      << std::endl
      << std::flush;

    cart->prev.reset( node, seg );
    node->next.reset( cart, seg );
  } 

  header->head.reset( node, seg );

  // queue tail is empty
  if ( header->tail.off() == 0) {
    header->tail.reset( node, seg );
  }

  header->size += 1;

  std::cout 
    << "queue size : "
    << header->size
    << std::endl
    << std::flush;

  // signal any waiting processes
  pthread_cond_signal( &header->ready_for_pop );
}

template<typename T>
shm_ptr<T> shm_queue<T>::pop() {
  shm_queue_header<T>* header = static_cast<shm_queue_header<T>* >
                                      ((shm_queue_header<T>*)
                                      ((long)(seg.get_base())+a.get_overhead()));
  scoped_lock guard( &(header->tail_lock) );
  if ( 1 > header->size ) {
    std::cout 
      << "Queue empty : wait until ready for pop."
      << header->size
      << std::endl
      << std::flush;
    guard.wait( &(header->ready_for_pop) );
  }
  // get object at tail of queue
  shm_ptr<shm_queue_node<T> > obj = header->tail;
  shm_queue_node<T>* node = obj.get(seg);
  // head and tail are the same (single queue node)
  if ( header->tail.off() == header->head.off() ) {
    std::cout 
      << "Queue head and tail offsets are identical."
      << std::endl
      << std::flush;
    header->head.set(0);
    header->tail.set(0);
    header->size = 0;

  } else { 
    // set tail to the prior queue node
    header->tail = node->prev;
    header->size -= 1;
  }
  if ( header->tail.off() == 0) {

    std::cout 
      << "Queue is now empty."
      << std::endl
      << std::flush;

  }    
  pthread_cond_signal( &header->ready_for_push );
  return node->data;
}

#endif

/*
 * $Log: shm_queue.h,v $
 * Revision 1.2  2006/09/02 18:47:28  oommoo
 * documentation enhancement
 *
 * Revision 1.1.1.1  2006/08/30 12:14:59  oommoo
 * Initial checkin of project library
 *
 * SharedMemory notes from :
 *
 * http://home.earthlink.net/~joshwalker1/writing/SharedMemory.html
 *
 */

---