boost::interprocess::message_queue, where the underlying implementation uses a circular buffer

message_queue.hpp

//////////////////////////////////////////////////////////////////////////////
//
// (C) Copyright Ion Gaztanaga 2005-2009. Distributed under the Boost
// Software License, Version 1.0. (See accompanying file
// LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
// See http://www.boost.org/libs/interprocess for documentation.
//
//////////////////////////////////////////////////////////////////////////////

#ifndef BOOST_INTERPROCESS_MESSAGE_QUEUE_HPP
#define BOOST_INTERPROCESS_MESSAGE_QUEUE_HPP

#include <boost/interprocess/detail/config_begin.hpp>
#include <boost/interprocess/detail/workaround.hpp>

#include <boost/interprocess/windows_shared_memory.hpp>
//#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/detail/managed_open_or_create_impl.hpp>
#include <boost/interprocess/sync/interprocess_condition.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/interprocess/detail/utilities.hpp>
#include <boost/interprocess/offset_ptr.hpp>
#include <boost/interprocess/creation_tags.hpp>
#include <boost/interprocess/exceptions.hpp>
#include <boost/interprocess/permissions.hpp>
#include <boost/detail/no_exceptions_support.hpp>
#include <boost/interprocess/detail/type_traits.hpp>
#include <boost/pointer_to_other.hpp>
#include <boost/type_traits/make_unsigned.hpp>
#include <boost/type_traits/alignment_of.hpp>
#include <algorithm> //std::lower_bound
#include <cstddef>   //std::size_t
#include <cstring>   //memcpy


//!\file
//!Describes an inter-process message queue. This class allows sending
//!messages between processes and allows blocking, non-blocking and timed
//!sending and receiving.

namespace boost{  namespace interprocess{

//!A class that allows sending messages
//!between processes.
template<class VoidPointer>
class message_queue_t
{
   /// @cond
   //Blocking modes
   enum block_t   {  blocking,   timed,   non_blocking   };

   message_queue_t();
   /// @endcond

   public:
   typedef VoidPointer                                                 void_pointer;
   typedef typename boost::pointer_to_other<void_pointer, char>::type  char_ptr;
   typedef typename std::iterator_traits<char_ptr>::difference_type    difference_type;
   //typedef typename boost::make_unsigned<difference_type>::type        size_type;
   typedef __int32        size_type;

   //!Creates a process shared message queue with name "name". For this message queue,
   //!the maximum number of messages will be "max_num_msg" and the maximum message size
   //!will be "max_msg_size". Throws on error and if the queue was previously created.
   message_queue_t(create_only_t create_only,
                 const char *name, 
                 size_type max_num_msg, 
                 size_type max_msg_size,
                 const permissions &perm = permissions());

   //!Opens or creates a process shared message queue with name "name". 
   //!If the queue is created, the maximum number of messages will be "max_num_msg" 
   //!and the maximum message size will be "max_msg_size". If queue was previously 
   //!created the queue will be opened and "max_num_msg" and "max_msg_size" parameters
   //!are ignored. Throws on error.
   message_queue_t(open_or_create_t open_or_create,
                 const char *name, 
                 size_type max_num_msg, 
                 size_type max_msg_size,
                 const permissions &perm = permissions());

   //!Opens a previously created process shared message queue with name "name". 
   //!If the was not previously created or there are no free resources, 
   //!throws an error.
   message_queue_t(open_only_t open_only,
                 const char *name);

   //!Destroys *this and indicates that the calling process is finished using
   //!the resource. All opened message queues are still
   //!valid after destruction. The destructor function will deallocate
   //!any system resources allocated by the system for use by this process for
   //!this resource. The resource can still be opened again calling
   //!the open constructor overload. To erase the message queue from the system
   //!use remove().
   ~message_queue_t(); 

   //!Sends a message stored in buffer "buffer" with size "buffer_size" in the 
   //!message queue with priority "priority". If the message queue is full
   //!the sender is blocked. Throws interprocess_error on error.*/
   void send (const void *buffer,     size_type buffer_size, 
              unsigned int priority);

   //!Sends a message stored in buffer "buffer" with size "buffer_size" through the 
   //!message queue with priority "priority". If the message queue is full
   //!the sender is not blocked and returns false, otherwise returns true.
   //!Throws interprocess_error on error.
   bool try_send    (const void *buffer,     size_type buffer_size, 
                         unsigned int priority);

   //!Sends a message stored in buffer "buffer" with size "buffer_size" in the 
   //!message queue with priority "priority". If the message queue is full
   //!the sender retries until time "abs_time" is reached. Returns true if
   //!the message has been successfully sent. Returns false if timeout is reached.
   //!Throws interprocess_error on error.
   bool timed_send    (const void *buffer,     size_type buffer_size, 
                           unsigned int priority,  const boost::posix_time::ptime& abs_time);

   //!Receives a message from the message queue. The message is stored in buffer 
   //!"buffer", which has size "buffer_size". The received message has size 
   //!"recvd_size" and priority "priority". If the message queue is empty
   //!the receiver is blocked. Throws interprocess_error on error.
   void receive (void *buffer,           size_type buffer_size, 
                 size_type &recvd_size,unsigned int &priority);

   //!Receives a message from the message queue. The message is stored in buffer 
   //!"buffer", which has size "buffer_size". The received message has size 
   //!"recvd_size" and priority "priority". If the message queue is empty
   //!the receiver is not blocked and returns false, otherwise returns true.
   //!Throws interprocess_error on error.
   bool try_receive (void *buffer,           size_type buffer_size, 
                     size_type &recvd_size,unsigned int &priority);

   //!Receives a message from the message queue. The message is stored in buffer 
   //!"buffer", which has size "buffer_size". The received message has size 
   //!"recvd_size" and priority "priority". If the message queue is empty
   //!the receiver retries until time "abs_time" is reached. Returns true if
   //!the message has been successfully sent. Returns false if timeout is reached.
   //!Throws interprocess_error on error.
   bool timed_receive (void *buffer,           size_type buffer_size, 
                       size_type &recvd_size,unsigned int &priority,
                       const boost::posix_time::ptime &abs_time);

   //!Returns the maximum number of messages allowed by the queue. The message
   //!queue must be opened or created previously. Otherwise, returns 0. 
   //!Never throws
   size_type get_max_msg() const;

   //!Returns the maximum size of message allowed by the queue. The message
   //!queue must be opened or created previously. Otherwise, returns 0. 
   //!Never throws
   size_type get_max_msg_size() const;

   //!Returns the number of messages currently stored. 
   //!Never throws
   size_type get_num_msg();

   //!Removes the message queue from the system.
   //!Returns false on error. Never throws
   static bool remove(const char *name);

   /// @cond   
   private:
   typedef boost::posix_time::ptime ptime;
   bool do_receive(block_t block,
                   void *buffer,         size_type buffer_size, 
                   size_type &recvd_size, unsigned int &priority,
                   const ptime &abs_time);

   bool do_send(block_t block,
                const void *buffer,      size_type buffer_size, 
                unsigned int priority,   const ptime &abs_time);

   //!Returns the needed memory size for the shared message queue.
   //!Never throws
   static size_type get_mem_size(size_type max_msg_size, size_type max_num_msg);

   ipcdetail::managed_open_or_create_impl<windows_shared_memory,false> m_shmem;
// ipcdetail::managed_open_or_create_impl<shared_memory_object> m_shmem;
   /// @endcond
};

/// @cond

namespace ipcdetail {

//!This header is the prefix of each message in the queue
template<class VoidPointer>
class msg_hdr_t 
{
   typedef VoidPointer                                                    void_pointer;
   typedef typename boost::
      pointer_to_other<VoidPointer, char>::type                          char_ptr;
   typedef typename std::iterator_traits<char_ptr>::difference_type       difference_type;
   //typedef typename boost::make_unsigned<difference_type>::type           size_type;
   typedef typename __int32        size_type;

   public:
   size_type               len;     // Message length
   unsigned int            priority;// Message priority
   //!Returns the data buffer associated with this this message
   void * data(){ return this+1; }  //
};



//!This header is placed in the beginning of the shared memory and contains 
//!the data to control the queue. This class initializes the shared memory 
//!in the following way: in ascending memory address with proper alignment
//!fillings:
//!
//!-> mq_hdr_t: 
//!   Main control block that controls the rest of the elements
//!
//!-> offset_ptr<msg_hdr_t> index [max_num_msg]
//!   An array of pointers with size "max_num_msg" called index. Each pointer 
//!   points to a preallocated message. The elements of this array are 
//!   reordered in runtime in the following way:
//!
//!   When the current number of messages is "cur_num_msg", the first 
//!   "cur_num_msg" pointers point to inserted messages and the rest
//!   point to free messages. The first "cur_num_msg" pointers are
//!   ordered by the priority of the pointed message and by insertion order 
//!   if two messages have the same priority. So the next message to be 
//!   used in a "receive" is pointed by index [cur_num_msg-1] and the first free
//!   message ready to be used in a "send" operation is index [cur_num_msg].
//!   This transforms index in a fixed size priority queue with an embedded free
//!   message queue.
//!
//!-> struct message_t
//!   {  
//!      msg_hdr_t            header;
//!      char[max_msg_size]   data;
//!   } messages [max_num_msg];
//!
//!   An array of buffers of preallocated messages, each one prefixed with the
//!   msg_hdr_t structure. Each of this message is pointed by one pointer of
//!   the index structure.
template<class VoidPointer>
class mq_hdr_t

{   
   typedef VoidPointer                                                    void_pointer;
   typedef msg_hdr_t<void_pointer>                                        msg_header;
   typedef typename boost::
      pointer_to_other<void_pointer, msg_header>::type                   msg_hdr_ptr_t;
   typedef typename std::iterator_traits<msg_hdr_ptr_t>::difference_type  difference_type;
   //typedef typename boost::make_unsigned<difference_type>::type           size_type;
   typedef  __int32   size_type;
   typedef typename boost::
      pointer_to_other<void_pointer, msg_hdr_ptr_t>::type                msg_hdr_ptr_ptr_t;

   public:
   //!Constructor. This object must be constructed in the beginning of the 
   //!shared memory of the size returned by the function "get_mem_size".
   //!This constructor initializes the needed resources and creates
   //!the internal structures like the priority index. This can throw.*/
   mq_hdr_t(size_type max_num_msg, size_type max_msg_size)
      : m_max_num_msg(max_num_msg), 
         m_max_msg_size(max_msg_size),
         m_cur_num_msg(0)
      {  this->initialize_memory();  }

   //!Returns the inserted message with top priority, the next message to be read
   msg_header * top_msg()
      {  return mp_index[m_first_msg_index].get();   }

   //!Returns true if the message queue is full
   bool is_full() const
      {  return m_cur_num_msg == m_max_num_msg;  }

   //!Returns true if the message queue is empty
   bool is_empty() const
      {  return !m_cur_num_msg;  }

   //!Frees the location of the message that was read.
   void free_top_msg()
      {  
		--m_cur_num_msg;  
		m_first_msg_index = (m_first_msg_index + 1) % m_max_num_msg; // cyclic buffer, slide over to the start if neccesary
	  }


	//!Returns the first free msg of the free message queue, to write to.
   msg_header * free_msg()
   {  
		size_t last_msg_index = (m_first_msg_index + m_cur_num_msg) % m_max_num_msg; // cyclic buffer
		++m_cur_num_msg;
		return mp_index[last_msg_index].get();  
   }	  


   //!Returns the number of bytes needed to construct a message queue with 
   //!"max_num_size" maximum number of messages and "max_msg_size" maximum 
   //!message size. Never throws.
   static size_type get_mem_size
      (size_type max_msg_size, size_type max_num_msg)
   {
      const size_type 
		 msg_hdr_align  = ::boost::alignment_of<msg_header>::value,
		 index_align    = ::boost::alignment_of<msg_hdr_ptr_t>::value,
         r_hdr_size     = ipcdetail::ct_rounded_size<sizeof(mq_hdr_t), index_align>::value,
         r_index_size   = ipcdetail::get_rounded_size((size_type)sizeof(msg_hdr_ptr_t)*max_num_msg, (size_type)msg_hdr_align),
         r_max_msg_size = ipcdetail::get_rounded_size(max_msg_size, msg_hdr_align) + sizeof(msg_header);
      return r_hdr_size + r_index_size + (max_num_msg*r_max_msg_size) + 
         ipcdetail::managed_open_or_create_impl<windows_shared_memory, false>::ManagedOpenOrCreateUserOffset;
   }

   //!Initializes the memory structures to preallocate messages and constructs the
   //!message index. Never throws.
   void initialize_memory()
   {
      const size_type 
		  msg_hdr_align  = ::boost::alignment_of<msg_header>::value,
		  index_align    = ::boost::alignment_of<msg_hdr_ptr_t>::value,
         r_hdr_size     = ipcdetail::ct_rounded_size<sizeof(mq_hdr_t), index_align>::value,
         r_index_size   = ipcdetail::get_rounded_size((size_type)sizeof(msg_hdr_ptr_t)*m_max_num_msg, (size_type)msg_hdr_align),
         r_max_msg_size = ipcdetail::get_rounded_size(m_max_msg_size, msg_hdr_align) + sizeof(msg_header);

      //Pointer to the index
      msg_hdr_ptr_t *index =  reinterpret_cast<msg_hdr_ptr_t*>
                                 (reinterpret_cast<char*>(this)+r_hdr_size);

      //Pointer to the first message header
      msg_header *msg_hdr   =  reinterpret_cast<msg_header*>
                                 (reinterpret_cast<char*>(this)+r_hdr_size+r_index_size);  

      //Initialize the pointer to the index
      mp_index             = index;

      //Initialize the index so each slot points to a preallocated message
      for(size_type i = 0; i < m_max_num_msg; ++i){
         index[i] = msg_hdr;
         msg_hdr  = reinterpret_cast<msg_header*>
                        (reinterpret_cast<char*>(msg_hdr)+r_max_msg_size);
      }
   }

   public:
   //Pointer to the index
   msg_hdr_ptr_ptr_t          mp_index;
   //Maximum number of messages of the queue
   const size_type            m_max_num_msg;
   //Maximum size of messages of the queue
   const size_type            m_max_msg_size;
   //Current number of messages
   size_type                  m_cur_num_msg;
   //The location of the first message, the one that should be received next
   size_type				  m_first_msg_index;   
   //Mutex to protect data structures
   interprocess_mutex         m_mutex;
   //Condition block receivers when there are no messages
   interprocess_condition     m_cond_recv;
   //Condition block senders when the queue is full
   interprocess_condition     m_cond_send;
};


//!This is the atomic functor to be executed when creating or opening 
//!shared memory. Never throws
template<class VoidPointer>
class initialization_func_t
{
   public:
   typedef typename boost::pointer_to_other<VoidPointer, char>::type   char_ptr;
   typedef typename std::iterator_traits<char_ptr>::difference_type    difference_type;
   //typedef typename boost::make_unsigned<difference_type>::type        size_type;
   typedef __int32        size_type;

   initialization_func_t(size_type maxmsg = 0, 
                         size_type maxmsgsize = 0)
      : m_maxmsg (maxmsg), m_maxmsgsize(maxmsgsize) {}

   bool operator()(void *address, size_type, bool created)
   {
      char      *mptr;

      if(created){
         mptr     = reinterpret_cast<char*>(address);
         //Construct the message queue header at the beginning
         BOOST_TRY{
            new (mptr) mq_hdr_t<VoidPointer>(m_maxmsg, m_maxmsgsize);
         }
         BOOST_CATCH(...){
            return false;   
         }
         BOOST_CATCH_END
      }
      return true;
   }
   const size_type m_maxmsg;
   const size_type m_maxmsgsize;
};

}  //namespace ipcdetail {

template<class VoidPointer>
inline message_queue_t<VoidPointer>::~message_queue_t()
{}

template<class VoidPointer>
inline typename message_queue_t<VoidPointer>::size_type message_queue_t<VoidPointer>::get_mem_size
   (size_type max_msg_size, size_type max_num_msg)
{  return ipcdetail::mq_hdr_t<VoidPointer>::get_mem_size(max_msg_size, max_num_msg);   }

template<class VoidPointer>
inline message_queue_t<VoidPointer>::message_queue_t(create_only_t create_only,
                                    const char *name, 
                                    size_type max_num_msg, 
                                    size_type max_msg_size,
                                    const permissions &perm)
      //Create shared memory and execute functor atomically
   :  m_shmem(create_only, 
              name, 
           get_mem_size(max_msg_size, max_num_msg),
              read_write,
              static_cast<void*>(0),
              //Prepare initialization functor
              ipcdetail::initialization_func_t<VoidPointer> (max_num_msg, max_msg_size),
              perm)
{}

template<class VoidPointer>
inline message_queue_t<VoidPointer>::message_queue_t(open_or_create_t open_or_create,
                                    const char *name, 
                                    size_type max_num_msg, 
                                    size_type max_msg_size,
                                    const permissions &perm)
      //Create shared memory and execute functor atomically
   :  m_shmem(open_or_create, 
              name, 
              get_mem_size(max_msg_size, max_num_msg),
              read_write,
              static_cast<void*>(0),
              //Prepare initialization functor
              ipcdetail::initialization_func_t<VoidPointer> (max_num_msg, max_msg_size),
              perm)
{}

template<class VoidPointer>
inline message_queue_t<VoidPointer>::message_queue_t(open_only_t open_only,
                                    const char *name)
   //Create shared memory and execute functor atomically
   :  m_shmem(open_only, 
              name,
              read_write,
              static_cast<void*>(0),
              //Prepare initialization functor
              ipcdetail::initialization_func_t<VoidPointer> ())
{}

template<class VoidPointer>
inline void message_queue_t<VoidPointer>::send
   (const void *buffer, size_type buffer_size, unsigned int priority)
{  this->do_send(blocking, buffer, buffer_size, priority, ptime()); }

template<class VoidPointer>
inline bool message_queue_t<VoidPointer>::try_send
   (const void *buffer, size_type buffer_size, unsigned int priority)
{  return this->do_send(non_blocking, buffer, buffer_size, priority, ptime()); }

template<class VoidPointer>
inline bool message_queue_t<VoidPointer>::timed_send
   (const void *buffer, size_type buffer_size
   ,unsigned int priority, const boost::posix_time::ptime &abs_time)
{
   if(abs_time == boost::posix_time::pos_infin){
      this->send(buffer, buffer_size, priority);
      return true;
   }
   return this->do_send(timed, buffer, buffer_size, priority, abs_time);
}

template<class VoidPointer>
inline bool message_queue_t<VoidPointer>::do_send(block_t block,
                                const void *buffer,      size_type buffer_size, 
                                unsigned int priority,   const boost::posix_time::ptime &abs_time)
{
   ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
   //Check if buffer is smaller than maximum allowed
   if (buffer_size > p_hdr->m_max_msg_size) {
      throw interprocess_exception(size_error);
   }

   //---------------------------------------------
   scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
   //---------------------------------------------
   {
      //If the queue is full execute blocking logic
      if (p_hdr->is_full()) {

         switch(block){
            case non_blocking :
               return false;
            break;

            case blocking :
               do{
                  p_hdr->m_cond_send.wait(lock);
               }
               while (p_hdr->is_full());
            break;

            case timed :
               do{
                  if(!p_hdr->m_cond_send.timed_wait(lock, abs_time)){
                     if(p_hdr->is_full())
                        return false;
                     break;
                  }
               }
               while (p_hdr->is_full());
            break;
            default:
            break;
         }
      }
      
      //Get the first free message from free message queue
      ipcdetail::msg_hdr_t<VoidPointer> *free_msg = p_hdr->free_msg();
      if (free_msg == 0) {
         throw interprocess_exception("boost::interprocess::message_queue corrupted");
      }

      //Copy control data to the free message
      free_msg->priority = priority;
      free_msg->len      = buffer_size;

      //Copy user buffer to the message
      std::memcpy(free_msg->data(), buffer, buffer_size);

      bool was_empty = p_hdr->is_empty();
      //Insert the first free message in the priority queue

      
      //If this message changes the queue empty state, notify it to receivers
      if (was_empty){
		 // return true;      // PERF tweak by omer. we'll be doing timed checks if buffer is empty or not, so no need for a cond variable.UNCOMMENT FOR PERF
         p_hdr->m_cond_recv.notify_one();
      }
   }  // Lock end

   return true;
}

template<class VoidPointer>
inline void message_queue_t<VoidPointer>::receive(void *buffer,        size_type buffer_size, 
                        size_type &recvd_size,   unsigned int &priority)
{  this->do_receive(blocking, buffer, buffer_size, recvd_size, priority, ptime()); }

template<class VoidPointer>
inline bool
   message_queue_t<VoidPointer>::try_receive(void *buffer,              size_type buffer_size, 
                              size_type &recvd_size,   unsigned int &priority)
{  return this->do_receive(non_blocking, buffer, buffer_size, recvd_size, priority, ptime()); }

template<class VoidPointer>
inline bool
   message_queue_t<VoidPointer>::timed_receive(void *buffer,            size_type buffer_size, 
                                size_type &recvd_size,   unsigned int &priority,
                                const boost::posix_time::ptime &abs_time)
{
   if(abs_time == boost::posix_time::pos_infin){
      this->receive(buffer, buffer_size, recvd_size, priority);
      return true;
   }
   return this->do_receive(timed, buffer, buffer_size, recvd_size, priority, abs_time);
}

template<class VoidPointer>
inline bool
   message_queue_t<VoidPointer>::do_receive(block_t block,
                          void *buffer,            size_type buffer_size, 
                          size_type &recvd_size,   unsigned int &priority,
                          const boost::posix_time::ptime &abs_time)
{
   ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
   //Check if buffer is big enough for any message
   if (buffer_size < p_hdr->m_max_msg_size) {
      throw interprocess_exception(size_error);
   }

   //---------------------------------------------
   scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
   //---------------------------------------------
   {
      //If there are no messages execute blocking logic
      if (p_hdr->is_empty()) {
         switch(block){
            case non_blocking :
               return false;
            break;

            case blocking :
               do{
                  p_hdr->m_cond_recv.wait(lock);
               }
               while (p_hdr->is_empty());
            break;

            case timed :
               do{
                  if(!p_hdr->m_cond_recv.timed_wait(lock, abs_time)){
                     if(p_hdr->is_empty())
                        return false;
                     break;
                  }
               }
               while (p_hdr->is_empty());
            break;

            //Paranoia check
            default:
            break;
         }
      }

      //Thre is at least message ready to pick, get the top one
     ipcdetail::msg_hdr_t<VoidPointer> *top_msg = p_hdr->top_msg();

      //Paranoia check
      if (top_msg == 0) {
         throw interprocess_exception("boost::interprocess::message_queue corrupted");
      }

      //Get data from the message
      recvd_size     = top_msg->len;
      priority       = top_msg->priority;

      //Copy data to receiver's bufers
     std::memcpy(buffer, top_msg->data(), recvd_size);

      bool was_full = p_hdr->is_full();

      //Free top message and put it in the free message list
      p_hdr->free_top_msg();

      //If this reception changes the queue full state, notify senders
      if (was_full){
         p_hdr->m_cond_send.notify_one();
      }
   }  //Lock end

   return true;
}

template<class VoidPointer>
inline typename message_queue_t<VoidPointer>::size_type message_queue_t<VoidPointer>::get_max_msg() const
{  
   ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
   return p_hdr ? p_hdr->m_max_num_msg : 0;  }

template<class VoidPointer>
inline typename message_queue_t<VoidPointer>::size_type message_queue_t<VoidPointer>::get_max_msg_size() const
{  
   ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
   return p_hdr ? p_hdr->m_max_msg_size : 0;  
}

template<class VoidPointer>
inline typename message_queue_t<VoidPointer>::size_type message_queue_t<VoidPointer>::get_num_msg()
{  
   ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
   if(p_hdr){
      //---------------------------------------------
      scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
      //---------------------------------------------
      return p_hdr->m_cur_num_msg;
   }

   return 0;  
}

template<class VoidPointer>
inline bool message_queue_t<VoidPointer>::remove(const char *name)
{  return true;;  }

/// @endcond

}} //namespace boost{  namespace interprocess{



#include <boost/interprocess/detail/config_end.hpp>

#endif   //#ifndef BOOST_INTERPROCESS_MESSAGE_QUEUE_HPP

This code needs to be cleaned up - all mention of message priority should be removed, the new implementation completely ignores it.