|
@@ -0,0 +1,900 @@
|
|
|
|
+//////////////////////////////////////////////////////////////////////////////
|
|
|
|
+//
|
|
|
|
+// (C) Copyright Ion Gaztanaga 2005-2012. 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/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/sync/scoped_lock.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/intrusive/pointer_traits.hpp>
|
|
|
|
+#include <boost/type_traits/make_unsigned.hpp>
|
|
|
|
+#include <boost/type_traits/alignment_of.hpp>
|
|
|
|
+#include <boost/intrusive/pointer_traits.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::intrusive::
|
|
|
|
+ pointer_traits<void_pointer>::template
|
|
|
|
+ rebind_pointer<char>::type char_ptr;
|
|
|
|
+ typedef typename boost::intrusive::pointer_traits<char_ptr>::difference_type difference_type;
|
|
|
|
+ typedef typename boost::make_unsigned<difference_type>::type 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 queue 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<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::intrusive::
|
|
|
|
+ pointer_traits<void_pointer>::template
|
|
|
|
+ rebind_pointer<char>::type char_ptr;
|
|
|
|
+ typedef typename boost::intrusive::pointer_traits<char_ptr>::difference_type difference_type;
|
|
|
|
+ typedef typename boost::make_unsigned<difference_type>::type 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 functor is the predicate to order stored messages by priority
|
|
|
|
+template<class VoidPointer>
|
|
|
|
+class priority_functor
|
|
|
|
+{
|
|
|
|
+ typedef typename boost::intrusive::
|
|
|
|
+ pointer_traits<VoidPointer>::template
|
|
|
|
+ rebind_pointer<msg_hdr_t<VoidPointer> >::type msg_hdr_ptr_t;
|
|
|
|
+
|
|
|
|
+ public:
|
|
|
|
+ bool operator()(const msg_hdr_ptr_t &msg1,
|
|
|
|
+ const msg_hdr_ptr_t &msg2) const
|
|
|
|
+ { return msg1->priority < msg2->priority; }
|
|
|
|
+};
|
|
|
|
+
|
|
|
|
+//!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. Elements of this array are
|
|
|
|
+//! reordered in runtime in the following way:
|
|
|
|
+//!
|
|
|
|
+//! IF BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX is defined:
|
|
|
|
+//!
|
|
|
|
+//! When the current number of messages is "cur_num_msg", the array
|
|
|
|
+//! is treated like a circular buffer. Starting from position "cur_first_msg"
|
|
|
|
+//! "cur_num_msg" in a circular way, pointers point to inserted messages and the rest
|
|
|
|
+//! point to free messages. Those "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_first_msg + cur_num_msg-1)%max_num_msg]
|
|
|
|
+//! and the first free message ready to be used in a "send" operation is
|
|
|
|
+//! [cur_first_msg] if circular buffer is extended from front,
|
|
|
|
+//! [(cur_first_msg + cur_num_msg)%max_num_msg] otherwise.
|
|
|
|
+//!
|
|
|
|
+//! This transforms the index in a circular buffer with an embedded free
|
|
|
|
+//! message queue.
|
|
|
|
+//!
|
|
|
|
+//! ELSE (BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX is NOT defined):
|
|
|
|
+//!
|
|
|
|
+//! 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 the 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
|
|
|
|
+ : public ipcdetail::priority_functor<VoidPointer>
|
|
|
|
+{
|
|
|
|
+ typedef VoidPointer void_pointer;
|
|
|
|
+ typedef msg_hdr_t<void_pointer> msg_header;
|
|
|
|
+ typedef typename boost::intrusive::
|
|
|
|
+ pointer_traits<void_pointer>::template
|
|
|
|
+ rebind_pointer<msg_header>::type msg_hdr_ptr_t;
|
|
|
|
+ typedef typename boost::intrusive::pointer_traits
|
|
|
|
+ <msg_hdr_ptr_t>::difference_type difference_type;
|
|
|
|
+ typedef typename boost::make_unsigned<difference_type>::type size_type;
|
|
|
|
+ typedef typename boost::intrusive::
|
|
|
|
+ pointer_traits<void_pointer>::template
|
|
|
|
+ rebind_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)
|
|
|
|
+ #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
|
|
|
|
+ ,m_cur_first_msg(0u)
|
|
|
|
+ #endif
|
|
|
|
+ { this->initialize_memory(); }
|
|
|
|
+
|
|
|
|
+ //!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 top priority message and saves it in the free message list
|
|
|
|
+ void free_top_msg()
|
|
|
|
+ { --m_cur_num_msg; }
|
|
|
|
+
|
|
|
|
+ #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
|
|
|
|
+
|
|
|
|
+ typedef msg_hdr_ptr_t *iterator;
|
|
|
|
+
|
|
|
|
+ size_type end_pos() const
|
|
|
|
+ {
|
|
|
|
+ const size_type space_until_bufend = m_max_num_msg - m_cur_first_msg;
|
|
|
|
+ return space_until_bufend > m_cur_num_msg
|
|
|
|
+ ? m_cur_first_msg + m_cur_num_msg : m_cur_num_msg - space_until_bufend;
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ //!Returns the inserted message with top priority
|
|
|
|
+ msg_header &top_msg()
|
|
|
|
+ {
|
|
|
|
+ size_type pos = this->end_pos();
|
|
|
|
+ return *mp_index[pos ? --pos : m_max_num_msg - 1];
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ //!Returns the inserted message with bottom priority
|
|
|
|
+ msg_header &bottom_msg()
|
|
|
|
+ { return *mp_index[m_cur_first_msg]; }
|
|
|
|
+
|
|
|
|
+ iterator inserted_ptr_begin() const
|
|
|
|
+ { return &mp_index[m_cur_first_msg]; }
|
|
|
|
+
|
|
|
|
+ iterator inserted_ptr_end() const
|
|
|
|
+ { return &mp_index[this->end_pos()]; }
|
|
|
|
+
|
|
|
|
+ iterator lower_bound(const msg_hdr_ptr_t & value, priority_functor<VoidPointer> func)
|
|
|
|
+ {
|
|
|
|
+ iterator begin(this->inserted_ptr_begin()), end(this->inserted_ptr_end());
|
|
|
|
+ if(end < begin){
|
|
|
|
+ iterator idx_end = &mp_index[m_max_num_msg];
|
|
|
|
+ iterator ret = std::lower_bound(begin, idx_end, value, func);
|
|
|
|
+ if(idx_end == ret){
|
|
|
|
+ iterator idx_beg = &mp_index[0];
|
|
|
|
+ ret = std::lower_bound(idx_beg, end, value, func);
|
|
|
|
+ //sanity check, these cases should not call lower_bound (optimized out)
|
|
|
|
+ assert(ret != end);
|
|
|
|
+ assert(ret != begin);
|
|
|
|
+ return ret;
|
|
|
|
+ }
|
|
|
|
+ else{
|
|
|
|
+ return ret;
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+ else{
|
|
|
|
+ return std::lower_bound(begin, end, value, func);
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ msg_header & insert_at(iterator where)
|
|
|
|
+ {
|
|
|
|
+ iterator inserted_ptr_end = this->inserted_ptr_end();
|
|
|
|
+ iterator inserted_ptr_beg = this->inserted_ptr_begin();
|
|
|
|
+ if(where == inserted_ptr_end){
|
|
|
|
+ ++m_cur_num_msg;
|
|
|
|
+ return **inserted_ptr_end;
|
|
|
|
+ }
|
|
|
|
+ else if(where == inserted_ptr_beg){
|
|
|
|
+ //unsigned integer guarantees underflow
|
|
|
|
+ m_cur_first_msg = m_cur_first_msg ? m_cur_first_msg : m_max_num_msg;
|
|
|
|
+ --m_cur_first_msg;
|
|
|
|
+ ++m_cur_num_msg;
|
|
|
|
+ return *mp_index[m_cur_first_msg];
|
|
|
|
+ }
|
|
|
|
+ else{
|
|
|
|
+ size_type pos = where - &mp_index[0];
|
|
|
|
+ size_type circ_pos = pos >= m_cur_first_msg ? pos - m_cur_first_msg : pos + (m_max_num_msg - m_cur_first_msg);
|
|
|
|
+ //Check if it's more efficient to move back or move front
|
|
|
|
+ if(circ_pos < m_cur_num_msg/2){
|
|
|
|
+ //The queue can't be full so m_cur_num_msg == 0 or m_cur_num_msg <= pos
|
|
|
|
+ //indicates two step insertion
|
|
|
|
+ if(!pos){
|
|
|
|
+ pos = m_max_num_msg;
|
|
|
|
+ where = &mp_index[m_max_num_msg-1];
|
|
|
|
+ }
|
|
|
|
+ else{
|
|
|
|
+ --where;
|
|
|
|
+ }
|
|
|
|
+ const bool unique_segment = m_cur_first_msg && m_cur_first_msg <= pos;
|
|
|
|
+ size_type first_segment_beg = unique_segment ? m_cur_first_msg : 1u;
|
|
|
|
+ size_type first_segment_end = pos;
|
|
|
|
+ size_type second_segment_beg = unique_segment || !m_cur_first_msg ? m_max_num_msg : m_cur_first_msg;
|
|
|
|
+ size_type second_segment_end = m_max_num_msg;
|
|
|
|
+ const msg_hdr_ptr_t backup = *(&mp_index[0] + (unique_segment ? first_segment_beg : second_segment_beg) - 1);
|
|
|
|
+
|
|
|
|
+ //First segment
|
|
|
|
+ if(!unique_segment){
|
|
|
|
+ std::copy( &mp_index[0] + second_segment_beg
|
|
|
|
+ , &mp_index[0] + second_segment_end
|
|
|
|
+ , &mp_index[0] + second_segment_beg - 1);
|
|
|
|
+ mp_index[m_max_num_msg-1] = mp_index[0];
|
|
|
|
+ }
|
|
|
|
+ std::copy( &mp_index[0] + first_segment_beg
|
|
|
|
+ , &mp_index[0] + first_segment_end
|
|
|
|
+ , &mp_index[0] + first_segment_beg - 1);
|
|
|
|
+ *where = backup;
|
|
|
|
+ m_cur_first_msg = m_cur_first_msg ? m_cur_first_msg : m_max_num_msg;
|
|
|
|
+ --m_cur_first_msg;
|
|
|
|
+ ++m_cur_num_msg;
|
|
|
|
+ return **where;
|
|
|
|
+ }
|
|
|
|
+ else{
|
|
|
|
+ //The queue can't be full so end_pos < m_cur_first_msg
|
|
|
|
+ //indicates two step insertion
|
|
|
|
+ const size_type end_pos = this->end_pos();
|
|
|
|
+ const bool unique_segment = pos < end_pos;
|
|
|
|
+ size_type first_segment_beg = pos;
|
|
|
|
+ size_type first_segment_end = unique_segment ? end_pos : m_max_num_msg-1;
|
|
|
|
+ size_type second_segment_beg = 0u;
|
|
|
|
+ size_type second_segment_end = unique_segment ? 0u : end_pos;
|
|
|
|
+ const msg_hdr_ptr_t backup = *inserted_ptr_end;
|
|
|
|
+
|
|
|
|
+ //First segment
|
|
|
|
+ if(!unique_segment){
|
|
|
|
+ std::copy_backward( &mp_index[0] + second_segment_beg
|
|
|
|
+ , &mp_index[0] + second_segment_end
|
|
|
|
+ , &mp_index[0] + second_segment_end + 1);
|
|
|
|
+ mp_index[0] = mp_index[m_max_num_msg-1];
|
|
|
|
+ }
|
|
|
|
+ std::copy_backward( &mp_index[0] + first_segment_beg
|
|
|
|
+ , &mp_index[0] + first_segment_end
|
|
|
|
+ , &mp_index[0] + first_segment_end + 1);
|
|
|
|
+ *where = backup;
|
|
|
|
+ ++m_cur_num_msg;
|
|
|
|
+ return **where;
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ #else
|
|
|
|
+
|
|
|
|
+ typedef msg_hdr_ptr_t *iterator;
|
|
|
|
+
|
|
|
|
+ //!Returns the inserted message with top priority
|
|
|
|
+ msg_header &top_msg()
|
|
|
|
+ { return *mp_index[m_cur_num_msg-1]; }
|
|
|
|
+
|
|
|
|
+ //!Returns the inserted message with bottom priority
|
|
|
|
+ msg_header &bottom_msg()
|
|
|
|
+ { return *mp_index[0]; }
|
|
|
|
+
|
|
|
|
+ iterator inserted_ptr_begin() const
|
|
|
|
+ { return &mp_index[0]; }
|
|
|
|
+
|
|
|
|
+ iterator inserted_ptr_end() const
|
|
|
|
+ { return &mp_index[m_cur_num_msg]; }
|
|
|
|
+
|
|
|
|
+ iterator lower_bound(const msg_hdr_ptr_t & value, priority_functor<VoidPointer> func)
|
|
|
|
+ { return std::lower_bound(this->inserted_ptr_begin(), this->inserted_ptr_end(), value, func); }
|
|
|
|
+
|
|
|
|
+ msg_header & insert_at(iterator pos)
|
|
|
|
+ {
|
|
|
|
+ const msg_hdr_ptr_t backup = *inserted_ptr_end();
|
|
|
|
+ std::copy_backward(pos, inserted_ptr_end(), inserted_ptr_end()+1);
|
|
|
|
+ *pos = backup;
|
|
|
|
+ ++m_cur_num_msg;
|
|
|
|
+ return **pos;
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ #endif
|
|
|
|
+
|
|
|
|
+ //!Inserts the first free message in the priority queue
|
|
|
|
+ msg_header & queue_free_msg(unsigned int priority)
|
|
|
|
+ {
|
|
|
|
+ //Get priority queue's range
|
|
|
|
+ iterator it (inserted_ptr_begin()), it_end(inserted_ptr_end());
|
|
|
|
+ //Optimize for non-priority usage
|
|
|
|
+ if(m_cur_num_msg && priority > this->bottom_msg().priority){
|
|
|
|
+ //Check for higher priority than all stored messages
|
|
|
|
+ if(priority > this->top_msg().priority){
|
|
|
|
+ it = it_end;
|
|
|
|
+ }
|
|
|
|
+ else{
|
|
|
|
+ //Since we don't now which free message we will pick
|
|
|
|
+ //build a dummy header for searches
|
|
|
|
+ msg_header dummy_hdr;
|
|
|
|
+ dummy_hdr.priority = priority;
|
|
|
|
+
|
|
|
|
+ //Get free msg
|
|
|
|
+ msg_hdr_ptr_t dummy_ptr(&dummy_hdr);
|
|
|
|
+
|
|
|
|
+ //Check where the free message should be placed
|
|
|
|
+ it = this->lower_bound(dummy_ptr, static_cast<priority_functor<VoidPointer>&>(*this));
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ }
|
|
|
|
+ //Insert the free message in the correct position
|
|
|
|
+ return this->insert_at(it);
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ //!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(max_num_msg*sizeof(msg_hdr_ptr_t), 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<shared_memory_object>::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(m_max_num_msg*sizeof(msg_hdr_ptr_t), 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;
|
|
|
|
+ //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;
|
|
|
|
+ #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
|
|
|
|
+ //Current start offset in the circular index
|
|
|
|
+ size_type m_cur_first_msg;
|
|
|
|
+ #endif
|
|
|
|
+};
|
|
|
|
+
|
|
|
|
+
|
|
|
|
+//!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::intrusive::
|
|
|
|
+ pointer_traits<VoidPointer>::template
|
|
|
|
+ rebind_pointer<char>::type char_ptr;
|
|
|
|
+ typedef typename boost::intrusive::pointer_traits<char_ptr>::difference_type difference_type;
|
|
|
|
+ typedef typename boost::make_unsigned<difference_type>::type 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);
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ bool was_empty = false;
|
|
|
|
+ //---------------------------------------------
|
|
|
|
+ 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;
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ was_empty = p_hdr->is_empty();
|
|
|
|
+ //Insert the first free message in the priority queue
|
|
|
|
+ ipcdetail::msg_hdr_t<VoidPointer> &free_msg_hdr = p_hdr->queue_free_msg(priority);
|
|
|
|
+
|
|
|
|
+ //Sanity check, free msgs are always cleaned when received
|
|
|
|
+ assert(free_msg_hdr.priority == 0);
|
|
|
|
+ assert(free_msg_hdr.len == 0);
|
|
|
|
+
|
|
|
|
+ //Copy control data to the free message
|
|
|
|
+ free_msg_hdr.priority = priority;
|
|
|
|
+ free_msg_hdr.len = buffer_size;
|
|
|
|
+
|
|
|
|
+ //Copy user buffer to the message
|
|
|
|
+ std::memcpy(free_msg_hdr.data(), buffer, buffer_size);
|
|
|
|
+ } // Lock end
|
|
|
|
+
|
|
|
|
+ //Notify outside lock to avoid contention. This might produce some
|
|
|
|
+ //spurious wakeups, but it's usually far better than notifying inside.
|
|
|
|
+ //If this message changes the queue empty state, notify it to receivers
|
|
|
|
+ if (was_empty){
|
|
|
|
+ p_hdr->m_cond_recv.notify_one();
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ 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);
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ bool was_full = false;
|
|
|
|
+ //---------------------------------------------
|
|
|
|
+ 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;
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ //There is at least one message ready to pick, get the top one
|
|
|
|
+ ipcdetail::msg_hdr_t<VoidPointer> &top_msg = p_hdr->top_msg();
|
|
|
|
+
|
|
|
|
+ //Get data from the message
|
|
|
|
+ recvd_size = top_msg.len;
|
|
|
|
+ priority = top_msg.priority;
|
|
|
|
+
|
|
|
|
+ //Some cleanup to ease debugging
|
|
|
|
+ top_msg.len = 0;
|
|
|
|
+ top_msg.priority = 0;
|
|
|
|
+
|
|
|
|
+ //Copy data to receiver's bufers
|
|
|
|
+ std::memcpy(buffer, top_msg.data(), recvd_size);
|
|
|
|
+
|
|
|
|
+ was_full = p_hdr->is_full();
|
|
|
|
+
|
|
|
|
+ //Free top message and put it in the free message list
|
|
|
|
+ p_hdr->free_top_msg();
|
|
|
|
+ } //Lock end
|
|
|
|
+
|
|
|
|
+ //Notify outside lock to avoid contention. This might produce some
|
|
|
|
+ //spurious wakeups, but it's usually far better than notifying inside.
|
|
|
|
+ //If this reception changes the queue full state, notify senders
|
|
|
|
+ if (was_full){
|
|
|
|
+ p_hdr->m_cond_send.notify_one();
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ 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 shared_memory_object::remove(name); }
|
|
|
|
+
|
|
|
|
+/// @endcond
|
|
|
|
+
|
|
|
|
+}} //namespace boost{ namespace interprocess{
|
|
|
|
+
|
|
|
|
+#include <boost/interprocess/detail/config_end.hpp>
|
|
|
|
+
|
|
|
|
+#endif //#ifndef BOOST_INTERPROCESS_MESSAGE_QUEUE_HPP
|