1//////////////////////////////////////////////////////////////////////////////
2//
3// (C) Copyright Ion Gaztanaga 2005-2012. Distributed under the Boost
4// Software License, Version 1.0. (See accompanying file
5// LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6//
7// See http://www.boost.org/libs/interprocess for documentation.
8//
9//////////////////////////////////////////////////////////////////////////////
10
11#ifndef BOOST_INTERPROCESS_MESSAGE_QUEUE_HPP
12#define BOOST_INTERPROCESS_MESSAGE_QUEUE_HPP
13
14#ifndef BOOST_CONFIG_HPP
15# include <boost/config.hpp>
16#endif
17#
18#if defined(BOOST_HAS_PRAGMA_ONCE)
19# pragma once
20#endif
21
22#include <boost/interprocess/detail/config_begin.hpp>
23#include <boost/interprocess/detail/workaround.hpp>
24
25#include <boost/interprocess/shared_memory_object.hpp>
26#include <boost/interprocess/detail/managed_open_or_create_impl.hpp>
27#include <boost/interprocess/sync/interprocess_condition.hpp>
28#include <boost/interprocess/sync/interprocess_mutex.hpp>
29#include <boost/interprocess/sync/scoped_lock.hpp>
30#include <boost/interprocess/detail/utilities.hpp>
31#include <boost/interprocess/timed_utils.hpp>
32#include <boost/interprocess/offset_ptr.hpp>
33#include <boost/interprocess/creation_tags.hpp>
34#include <boost/interprocess/exceptions.hpp>
35#include <boost/interprocess/permissions.hpp>
36#include <boost/core/no_exceptions_support.hpp>
37#include <boost/interprocess/detail/type_traits.hpp>
38#include <boost/intrusive/pointer_traits.hpp>
39#include <boost/move/detail/type_traits.hpp> //make_unsigned, alignment_of
40#include <boost/intrusive/pointer_traits.hpp>
41#include <boost/move/detail/force_ptr.hpp>
42#include <boost/assert.hpp>
43#include <algorithm> //std::lower_bound
44#include <cstddef> //std::size_t
45#include <cstring> //memcpy
46
47
48//!\file
49//!Describes an inter-process message queue. This class allows sending
50//!messages between processes and allows blocking, non-blocking and timed
51//!sending and receiving.
52
53namespace boost{ namespace interprocess{
54
55namespace ipcdetail
56{
57 template<class VoidPointer>
58 class msg_queue_initialization_func_t;
59
60}
61
62//Blocking modes
63enum mqblock_types { blocking, timed, non_blocking };
64
65//!A class that allows sending messages
66//!between processes.
67template<class VoidPointer>
68class message_queue_t
69{
70 #if !defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
71
72 message_queue_t();
73 #endif //#ifndef BOOST_INTERPROCESS_DOXYGEN_INVOKED
74
75 public:
76 typedef VoidPointer void_pointer;
77 typedef typename boost::intrusive::
78 pointer_traits<void_pointer>::template
79 rebind_pointer<char>::type char_ptr;
80 typedef typename boost::intrusive::pointer_traits<char_ptr>::difference_type difference_type;
81 typedef typename boost::container::dtl::make_unsigned<difference_type>::type size_type;
82
83 //!Creates a process shared message queue with name "name". For this message queue,
84 //!the maximum number of messages will be "max_num_msg" and the maximum message size
85 //!will be "max_msg_size". Throws on error and if the queue was previously created.
86 message_queue_t(create_only_t,
87 const char *name,
88 size_type max_num_msg,
89 size_type max_msg_size,
90 const permissions &perm = permissions());
91
92 //!Opens or creates a process shared message queue with name "name".
93 //!If the queue is created, the maximum number of messages will be "max_num_msg"
94 //!and the maximum message size will be "max_msg_size". If queue was previously
95 //!created the queue will be opened and "max_num_msg" and "max_msg_size" parameters
96 //!are ignored. Throws on error.
97 message_queue_t(open_or_create_t,
98 const char *name,
99 size_type max_num_msg,
100 size_type max_msg_size,
101 const permissions &perm = permissions());
102
103 //!Opens a previously created process shared message queue with name "name".
104 //!If the queue was not previously created or there are no free resources,
105 //!throws an error.
106 message_queue_t(open_only_t, const char *name);
107
108 #if defined(BOOST_INTERPROCESS_WCHAR_NAMED_RESOURCES) || defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
109
110 //!Creates a process shared message queue with name "name". For this message queue,
111 //!the maximum number of messages will be "max_num_msg" and the maximum message size
112 //!will be "max_msg_size". Throws on error and if the queue was previously created.
113 //!
114 //!Note: This function is only available on operating systems with
115 //! native wchar_t APIs (e.g. Windows).
116 message_queue_t(create_only_t,
117 const wchar_t *name,
118 size_type max_num_msg,
119 size_type max_msg_size,
120 const permissions &perm = permissions());
121
122 //!Opens or creates a process shared message queue with name "name".
123 //!If the queue is created, the maximum number of messages will be "max_num_msg"
124 //!and the maximum message size will be "max_msg_size". If queue was previously
125 //!created the queue will be opened and "max_num_msg" and "max_msg_size" parameters
126 //!are ignored. Throws on error.
127 //!
128 //!Note: This function is only available on operating systems with
129 //! native wchar_t APIs (e.g. Windows).
130 message_queue_t(open_or_create_t,
131 const wchar_t *name,
132 size_type max_num_msg,
133 size_type max_msg_size,
134 const permissions &perm = permissions());
135
136 //!Opens a previously created process shared message queue with name "name".
137 //!If the queue was not previously created or there are no free resources,
138 //!throws an error.
139 //!
140 //!Note: This function is only available on operating systems with
141 //! native wchar_t APIs (e.g. Windows).
142 message_queue_t(open_only_t, const wchar_t *name);
143
144 #endif //defined(BOOST_INTERPROCESS_WCHAR_NAMED_RESOURCES) || defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
145
146 //!Destroys *this and indicates that the calling process is finished using
147 //!the resource. All opened message queues are still
148 //!valid after destruction. The destructor function will deallocate
149 //!any system resources allocated by the system for use by this process for
150 //!this resource. The resource can still be opened again calling
151 //!the open constructor overload. To erase the message queue from the system
152 //!use remove().
153 ~message_queue_t();
154
155 //!Sends a message stored in buffer "buffer" with size "buffer_size" in the
156 //!message queue with priority "priority". If the message queue is full
157 //!the sender is blocked. Throws interprocess_error on error.
158 void send (const void *buffer, size_type buffer_size,
159 unsigned int priority);
160
161 //!Sends a message stored in buffer "buffer" with size "buffer_size" through the
162 //!message queue with priority "priority". If the message queue is full
163 //!the sender is not blocked and returns false, otherwise returns true.
164 //!Throws interprocess_error on error.
165 bool try_send (const void *buffer, size_type buffer_size,
166 unsigned int priority);
167
168 //!Sends a message stored in buffer "buffer" with size "buffer_size" in the
169 //!message queue with priority "priority". If the message queue is full
170 //!the sender retries until time "abs_time" is reached. Returns true if
171 //!the message has been successfully sent. Returns false if timeout is reached.
172 //!Throws interprocess_error on error.
173 template<class TimePoint>
174 bool timed_send (const void *buffer, size_type buffer_size,
175 unsigned int priority, const TimePoint& abs_time);
176
177 //!Receives a message from the message queue. The message is stored in buffer
178 //!"buffer", which has size "buffer_size". The received message has size
179 //!"recvd_size" and priority "priority". If the message queue is empty
180 //!the receiver is blocked. Throws interprocess_error on error.
181 void receive (void *buffer, size_type buffer_size,
182 size_type &recvd_size,unsigned int &priority);
183
184 //!Receives a message from the message queue. The message is stored in buffer
185 //!"buffer", which has size "buffer_size". The received message has size
186 //!"recvd_size" and priority "priority". If the message queue is empty
187 //!the receiver is not blocked and returns false, otherwise returns true.
188 //!Throws interprocess_error on error.
189 bool try_receive (void *buffer, size_type buffer_size,
190 size_type &recvd_size,unsigned int &priority);
191
192 //!Receives a message from the message queue. The message is stored in buffer
193 //!"buffer", which has size "buffer_size". The received message has size
194 //!"recvd_size" and priority "priority". If the message queue is empty
195 //!the receiver retries until time "abs_time" is reached. Returns true if
196 //!the message has been successfully sent. Returns false if timeout is reached.
197 //!Throws interprocess_error on error.
198 template<class TimePoint>
199 bool timed_receive (void *buffer, size_type buffer_size,
200 size_type &recvd_size,unsigned int &priority,
201 const TimePoint &abs_time);
202
203 //!Returns the maximum number of messages allowed by the queue. The message
204 //!queue must be opened or created previously. Otherwise, returns 0.
205 //!Never throws
206 size_type get_max_msg() const;
207
208 //!Returns the maximum size of message allowed by the queue. The message
209 //!queue must be opened or created previously. Otherwise, returns 0.
210 //!Never throws
211 size_type get_max_msg_size() const;
212
213 //!Returns the number of messages currently stored.
214 //!Never throws
215 size_type get_num_msg() const;
216
217 //!Removes the message queue from the system.
218 //!Returns false on error. Never throws
219 static bool remove(const char *name);
220
221 #if defined(BOOST_INTERPROCESS_WCHAR_NAMED_RESOURCES) || defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
222
223 //!Removes the message queue from the system.
224 //!Returns false on error. Never throws
225 //!
226 //!Note: This function is only available on operating systems with
227 //! native wchar_t APIs (e.g. Windows).
228 static bool remove(const wchar_t *name);
229
230 #endif
231
232 #if !defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
233 private:
234
235 friend class ipcdetail::msg_queue_initialization_func_t<VoidPointer>;
236
237 template<mqblock_types Block, class TimePoint>
238 bool do_receive(void *buffer, size_type buffer_size,
239 size_type &recvd_size, unsigned int &priority,
240 const TimePoint &abs_time);
241
242 template<mqblock_types Block, class TimePoint>
243 bool do_send(const void *buffer, size_type buffer_size,
244 unsigned int priority, const TimePoint &abs_time);
245
246 //!Returns the needed memory size for the shared message queue.
247 //!Never throws
248 static size_type get_mem_size(size_type max_msg_size, size_type max_num_msg);
249 typedef ipcdetail::managed_open_or_create_impl<shared_memory_object, 0, true, false> open_create_impl_t;
250 open_create_impl_t m_shmem;
251
252 template<class Lock, class TimePoint>
253 static bool do_cond_wait(ipcdetail::bool_<true>, interprocess_condition &cond, Lock &lock, const TimePoint &abs_time)
254 { return cond.timed_wait(lock, abs_time); }
255
256 template<class Lock, class TimePoint>
257 static bool do_cond_wait(ipcdetail::bool_<false>, interprocess_condition &cond, Lock &lock, const TimePoint &)
258 { cond.wait(lock); return true; }
259
260 #endif //#ifndef BOOST_INTERPROCESS_DOXYGEN_INVOKED
261};
262
263#if !defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
264
265namespace ipcdetail {
266
267//!This header is the prefix of each message in the queue
268template<class VoidPointer>
269class msg_hdr_t
270{
271 typedef VoidPointer void_pointer;
272 typedef typename boost::intrusive::
273 pointer_traits<void_pointer>::template
274 rebind_pointer<char>::type char_ptr;
275 typedef typename boost::intrusive::pointer_traits<char_ptr>::difference_type difference_type;
276 typedef typename boost::container::dtl::make_unsigned<difference_type>::type size_type;
277
278 public:
279 size_type len; // Message length
280 unsigned int priority;// Message priority
281 //!Returns the data buffer associated with this this message
282 void * data(){ return this+1; } //
283};
284
285//!This functor is the predicate to order stored messages by priority
286template<class VoidPointer>
287class priority_functor
288{
289 typedef typename boost::intrusive::
290 pointer_traits<VoidPointer>::template
291 rebind_pointer<msg_hdr_t<VoidPointer> >::type msg_hdr_ptr_t;
292
293 public:
294 bool operator()(const msg_hdr_ptr_t &msg1,
295 const msg_hdr_ptr_t &msg2) const
296 { return msg1->priority < msg2->priority; }
297};
298
299//!This header is placed in the beginning of the shared memory and contains
300//!the data to control the queue. This class initializes the shared memory
301//!in the following way: in ascending memory address with proper alignment
302//!fillings:
303//!
304//!-> mq_hdr_t:
305//! Main control block that controls the rest of the elements
306//!
307//!-> offset_ptr<msg_hdr_t> index [max_num_msg]
308//! An array of pointers with size "max_num_msg" called index. Each pointer
309//! points to a preallocated message. Elements of this array are
310//! reordered in runtime in the following way:
311//!
312//! IF BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX is defined:
313//!
314//! When the current number of messages is "cur_num_msg", the array
315//! is treated like a circular buffer. Starting from position "cur_first_msg"
316//! "cur_num_msg" in a circular way, pointers point to inserted messages and the rest
317//! point to free messages. Those "cur_num_msg" pointers are
318//! ordered by the priority of the pointed message and by insertion order
319//! if two messages have the same priority. So the next message to be
320//! used in a "receive" is pointed by index [(cur_first_msg + cur_num_msg-1)%max_num_msg]
321//! and the first free message ready to be used in a "send" operation is
322//! [cur_first_msg] if circular buffer is extended from front,
323//! [(cur_first_msg + cur_num_msg)%max_num_msg] otherwise.
324//!
325//! This transforms the index in a circular buffer with an embedded free
326//! message queue.
327//!
328//! ELSE (BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX is NOT defined):
329//!
330//! When the current number of messages is "cur_num_msg", the first
331//! "cur_num_msg" pointers point to inserted messages and the rest
332//! point to free messages. The first "cur_num_msg" pointers are
333//! ordered by the priority of the pointed message and by insertion order
334//! if two messages have the same priority. So the next message to be
335//! used in a "receive" is pointed by index [cur_num_msg-1] and the first free
336//! message ready to be used in a "send" operation is index [cur_num_msg].
337//!
338//! This transforms the index in a fixed size priority queue with an embedded free
339//! message queue.
340//!
341//!-> struct message_t
342//! {
343//! msg_hdr_t header;
344//! char[max_msg_size] data;
345//! } messages [max_num_msg];
346//!
347//! An array of buffers of preallocated messages, each one prefixed with the
348//! msg_hdr_t structure. Each of this message is pointed by one pointer of
349//! the index structure.
350template<class VoidPointer>
351class mq_hdr_t
352 : public ipcdetail::priority_functor<VoidPointer>
353{
354 typedef VoidPointer void_pointer;
355 typedef msg_hdr_t<void_pointer> msg_header;
356 typedef typename boost::intrusive::
357 pointer_traits<void_pointer>::template
358 rebind_pointer<msg_header>::type msg_hdr_ptr_t;
359 typedef typename boost::intrusive::pointer_traits
360 <msg_hdr_ptr_t>::difference_type difference_type;
361 typedef typename boost::container::
362 dtl::make_unsigned<difference_type>::type size_type;
363 typedef typename boost::intrusive::
364 pointer_traits<void_pointer>::template
365 rebind_pointer<msg_hdr_ptr_t>::type msg_hdr_ptr_ptr_t;
366 typedef ipcdetail::managed_open_or_create_impl<shared_memory_object, 0, true, false> open_create_impl_t;
367
368 public:
369 //!Constructor. This object must be constructed in the beginning of the
370 //!shared memory of the size returned by the function "get_mem_size".
371 //!This constructor initializes the needed resources and creates
372 //!the internal structures like the priority index. This can throw.
373 mq_hdr_t(size_type max_num_msg, size_type max_msg_size)
374 : m_max_num_msg(max_num_msg),
375 m_max_msg_size(max_msg_size),
376 m_cur_num_msg(0)
377 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
378 ,m_cur_first_msg(0u)
379 ,m_blocked_senders(0u)
380 ,m_blocked_receivers(0u)
381 #endif
382 { this->initialize_memory(); }
383
384 //!Returns true if the message queue is full
385 bool is_full() const
386 { return m_cur_num_msg == m_max_num_msg; }
387
388 //!Returns true if the message queue is empty
389 bool is_empty() const
390 { return !m_cur_num_msg; }
391
392 //!Frees the top priority message and saves it in the free message list
393 void free_top_msg()
394 { --m_cur_num_msg; }
395
396 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
397
398 typedef msg_hdr_ptr_t *iterator;
399
400 size_type end_pos() const
401 {
402 const size_type space_until_bufend = m_max_num_msg - m_cur_first_msg;
403 return space_until_bufend > m_cur_num_msg
404 ? m_cur_first_msg + m_cur_num_msg : m_cur_num_msg - space_until_bufend;
405 }
406
407 //!Returns the inserted message with top priority
408 msg_header &top_msg()
409 {
410 size_type pos = this->end_pos();
411 return *mp_index[difference_type(pos ? --pos : m_max_num_msg - 1)];
412 }
413
414 //!Returns the inserted message with bottom priority
415 msg_header &bottom_msg()
416 { return *mp_index[difference_type(m_cur_first_msg)]; }
417
418 iterator inserted_ptr_begin() const
419 { return &mp_index[difference_type(m_cur_first_msg)]; }
420
421 iterator inserted_ptr_end() const
422 { return &mp_index[difference_type(this->end_pos())]; }
423
424 iterator lower_bound(const msg_hdr_ptr_t & value, priority_functor<VoidPointer> func)
425 {
426 iterator begin(this->inserted_ptr_begin()), end(this->inserted_ptr_end());
427 if(end < begin){
428 iterator idx_end = &mp_index[difference_type(m_max_num_msg)];
429 iterator ret = std::lower_bound(begin, idx_end, value, func);
430 if(idx_end == ret){
431 iterator idx_beg = &mp_index[0];
432 ret = std::lower_bound(idx_beg, end, value, func);
433 //sanity check, these cases should not call lower_bound (optimized out)
434 BOOST_ASSERT(ret != end);
435 BOOST_ASSERT(ret != begin);
436 return ret;
437 }
438 else{
439 return ret;
440 }
441 }
442 else{
443 return std::lower_bound(begin, end, value, func);
444 }
445 }
446
447 msg_header & insert_at(iterator where)
448 {
449 iterator it_inserted_ptr_end = this->inserted_ptr_end();
450 iterator it_inserted_ptr_beg = this->inserted_ptr_begin();
451 if(where == it_inserted_ptr_beg){
452 //unsigned integer guarantees underflow
453 m_cur_first_msg = m_cur_first_msg ? m_cur_first_msg : m_max_num_msg;
454 --m_cur_first_msg;
455 ++m_cur_num_msg;
456 return *mp_index[difference_type(m_cur_first_msg)];
457 }
458 else if(where == it_inserted_ptr_end){
459 ++m_cur_num_msg;
460 return **it_inserted_ptr_end;
461 }
462 else{
463 size_type pos = size_type(where - &mp_index[0]);
464 size_type circ_pos = pos >= m_cur_first_msg ? pos - m_cur_first_msg : pos + (m_max_num_msg - m_cur_first_msg);
465 //Check if it's more efficient to move back or move front
466 if(circ_pos < m_cur_num_msg/2){
467 //The queue can't be full so m_cur_num_msg == 0 or m_cur_num_msg <= pos
468 //indicates two step insertion
469 if(!pos){
470 pos = m_max_num_msg;
471 where = &mp_index[difference_type(m_max_num_msg-1u)];
472 }
473 else{
474 --where;
475 }
476 const bool unique_segment = m_cur_first_msg && m_cur_first_msg <= pos;
477 const size_type first_segment_beg = unique_segment ? m_cur_first_msg : 1u;
478 const size_type first_segment_end = pos;
479 const size_type second_segment_beg = unique_segment || !m_cur_first_msg ? m_max_num_msg : m_cur_first_msg;
480 const size_type second_segment_end = m_max_num_msg;
481 const msg_hdr_ptr_t backup = *(&mp_index[0] + (unique_segment ? first_segment_beg : second_segment_beg) - 1);
482
483 //First segment
484 if(!unique_segment){
485 std::copy( &mp_index[0] + second_segment_beg
486 , &mp_index[0] + second_segment_end
487 , &mp_index[0] + second_segment_beg - 1);
488 mp_index[difference_type(m_max_num_msg-1u)] = mp_index[0];
489 }
490 std::copy( &mp_index[0] + first_segment_beg
491 , &mp_index[0] + first_segment_end
492 , &mp_index[0] + first_segment_beg - 1);
493 *where = backup;
494 m_cur_first_msg = m_cur_first_msg ? m_cur_first_msg : m_max_num_msg;
495 --m_cur_first_msg;
496 ++m_cur_num_msg;
497 return **where;
498 }
499 else{
500 //The queue can't be full so end_pos < m_cur_first_msg
501 //indicates two step insertion
502 const size_type pos_end = this->end_pos();
503 const bool unique_segment = pos < pos_end;
504 const size_type first_segment_beg = pos;
505 const size_type first_segment_end = unique_segment ? pos_end : m_max_num_msg-1;
506 const size_type second_segment_beg = 0u;
507 const size_type second_segment_end = unique_segment ? 0u : pos_end;
508 const msg_hdr_ptr_t backup = *it_inserted_ptr_end;
509
510 //First segment
511 if(!unique_segment){
512 std::copy_backward( &mp_index[0] + second_segment_beg
513 , &mp_index[0] + second_segment_end
514 , &mp_index[0] + second_segment_end + 1u);
515 mp_index[0] = mp_index[difference_type(m_max_num_msg-1u)];
516 }
517 std::copy_backward( &mp_index[0] + first_segment_beg
518 , &mp_index[0] + first_segment_end
519 , &mp_index[0] + first_segment_end + 1u);
520 *where = backup;
521 ++m_cur_num_msg;
522 return **where;
523 }
524 }
525 }
526
527 #else //BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
528
529 typedef msg_hdr_ptr_t *iterator;
530
531 //!Returns the inserted message with top priority
532 msg_header &top_msg()
533 { return *mp_index[difference_type(m_cur_num_msg-1u)]; }
534
535 //!Returns the inserted message with bottom priority
536 msg_header &bottom_msg()
537 { return *mp_index[0]; }
538
539 iterator inserted_ptr_begin() const
540 { return &mp_index[0]; }
541
542 iterator inserted_ptr_end() const
543 { return &mp_index[difference_type(m_cur_num_msg)]; }
544
545 iterator lower_bound(const msg_hdr_ptr_t & value, priority_functor<VoidPointer> func)
546 { return std::lower_bound(this->inserted_ptr_begin(), this->inserted_ptr_end(), value, func); }
547
548 msg_header & insert_at(iterator pos)
549 {
550 const msg_hdr_ptr_t backup = *inserted_ptr_end();
551 std::copy_backward(pos, inserted_ptr_end(), inserted_ptr_end()+1);
552 *pos = backup;
553 ++m_cur_num_msg;
554 return **pos;
555 }
556
557 #endif //BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
558
559 //!Inserts the first free message in the priority queue
560 msg_header & queue_free_msg(unsigned int priority)
561 {
562 //Get priority queue's range
563 iterator it (inserted_ptr_begin()), it_end(inserted_ptr_end());
564 //Optimize for non-priority usage
565 if(m_cur_num_msg && priority > this->bottom_msg().priority){
566 //Check for higher priority than all stored messages
567 if(priority > this->top_msg().priority){
568 it = it_end;
569 }
570 else{
571 //Since we don't now which free message we will pick
572 //build a dummy header for searches
573 msg_header dummy_hdr;
574 dummy_hdr.priority = priority;
575
576 //Get free msg
577 msg_hdr_ptr_t dummy_ptr(&dummy_hdr);
578
579 //Check where the free message should be placed
580 it = this->lower_bound(value: dummy_ptr, func: static_cast<priority_functor<VoidPointer>&>(*this));
581 }
582 }
583 //Insert the free message in the correct position
584 return this->insert_at(where: it);
585 }
586
587 //!Returns the number of bytes needed to construct a message queue with
588 //!"max_num_size" maximum number of messages and "max_msg_size" maximum
589 //!message size. Never throws.
590 static size_type get_mem_size
591 (size_type max_msg_size, size_type max_num_msg)
592 {
593 const size_type
594 msg_hdr_align = ::boost::container::dtl::alignment_of<msg_header>::value,
595 index_align = ::boost::container::dtl::alignment_of<msg_hdr_ptr_t>::value,
596 r_hdr_size = ipcdetail::ct_rounded_size<sizeof(mq_hdr_t), index_align>::value,
597 r_index_size = ipcdetail::get_rounded_size<size_type>(max_num_msg*sizeof(msg_hdr_ptr_t), msg_hdr_align),
598 r_max_msg_size = ipcdetail::get_rounded_size<size_type>(max_msg_size, msg_hdr_align) + sizeof(msg_header);
599 return r_hdr_size + r_index_size + (max_num_msg*r_max_msg_size) +
600 open_create_impl_t::ManagedOpenOrCreateUserOffset;
601 }
602
603 //!Initializes the memory structures to preallocate messages and constructs the
604 //!message index. Never throws.
605 void initialize_memory()
606 {
607 const size_type
608 msg_hdr_align = ::boost::container::dtl::alignment_of<msg_header>::value,
609 index_align = ::boost::container::dtl::alignment_of<msg_hdr_ptr_t>::value,
610 r_hdr_size = ipcdetail::ct_rounded_size<sizeof(mq_hdr_t), index_align>::value,
611 r_index_size = ipcdetail::get_rounded_size<size_type>(m_max_num_msg*sizeof(msg_hdr_ptr_t), msg_hdr_align),
612 r_max_msg_size = ipcdetail::get_rounded_size<size_type>(m_max_msg_size, msg_hdr_align) + sizeof(msg_header);
613
614 //Pointer to the index
615 msg_hdr_ptr_t *index = move_detail::force_ptr<msg_hdr_ptr_t*>
616 (reinterpret_cast<char*>(this)+r_hdr_size);
617
618 //Pointer to the first message header
619 msg_header *msg_hdr = move_detail::force_ptr<msg_header*>
620 (reinterpret_cast<char*>(this)+r_hdr_size+r_index_size);
621
622 //Initialize the pointer to the index
623 mp_index = index;
624
625 //Initialize the index so each slot points to a preallocated message
626 for(size_type i = 0; i < m_max_num_msg; ++i){
627 index[i] = msg_hdr;
628 msg_hdr = move_detail::force_ptr<msg_header*>
629 (reinterpret_cast<char*>(msg_hdr)+r_max_msg_size);
630 }
631 }
632
633 public:
634 //Pointer to the index
635 msg_hdr_ptr_ptr_t mp_index;
636 //Maximum number of messages of the queue
637 const size_type m_max_num_msg;
638 //Maximum size of messages of the queue
639 const size_type m_max_msg_size;
640 //Current number of messages
641 size_type m_cur_num_msg;
642 //Mutex to protect data structures
643 interprocess_mutex m_mutex;
644 //Condition block receivers when there are no messages
645 interprocess_condition m_cond_recv;
646 //Condition block senders when the queue is full
647 interprocess_condition m_cond_send;
648 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
649 //Current start offset in the circular index
650 size_type m_cur_first_msg;
651 size_type m_blocked_senders;
652 size_type m_blocked_receivers;
653 #endif
654};
655
656
657//!This is the atomic functor to be executed when creating or opening
658//!shared memory. Never throws
659template<class VoidPointer>
660class msg_queue_initialization_func_t
661{
662 public:
663 typedef typename boost::intrusive::
664 pointer_traits<VoidPointer>::template
665 rebind_pointer<char>::type char_ptr;
666 typedef typename boost::intrusive::pointer_traits<char_ptr>::
667 difference_type difference_type;
668 typedef typename boost::container::dtl::
669 make_unsigned<difference_type>::type size_type;
670
671 msg_queue_initialization_func_t(size_type maxmsg = 0,
672 size_type maxmsgsize = 0)
673 : m_maxmsg (maxmsg), m_maxmsgsize(maxmsgsize) {}
674
675 bool operator()(void *address, size_type, bool created)
676 {
677 char *mptr;
678
679 if(created){
680 mptr = reinterpret_cast<char*>(address);
681 //Construct the message queue header at the beginning
682 BOOST_TRY{
683 new (mptr) mq_hdr_t<VoidPointer>(m_maxmsg, m_maxmsgsize);
684 }
685 BOOST_CATCH(...){
686 return false;
687 } BOOST_CATCH_END
688 }
689 return true;
690 }
691
692 std::size_t get_min_size() const
693 {
694 return mq_hdr_t<VoidPointer>::get_mem_size(m_maxmsgsize, m_maxmsg)
695 - message_queue_t<VoidPointer>::open_create_impl_t::ManagedOpenOrCreateUserOffset;
696 }
697
698 const size_type m_maxmsg;
699 const size_type m_maxmsgsize;
700};
701
702} //namespace ipcdetail {
703
704template<class VoidPointer>
705inline message_queue_t<VoidPointer>::~message_queue_t()
706{}
707
708template<class VoidPointer>
709inline typename message_queue_t<VoidPointer>::size_type message_queue_t<VoidPointer>::get_mem_size
710 (size_type max_msg_size, size_type max_num_msg)
711{ return ipcdetail::mq_hdr_t<VoidPointer>::get_mem_size(max_msg_size, max_num_msg); }
712
713template<class VoidPointer>
714inline message_queue_t<VoidPointer>::message_queue_t(create_only_t,
715 const char *name,
716 size_type max_num_msg,
717 size_type max_msg_size,
718 const permissions &perm)
719 //Create shared memory and execute functor atomically
720 : m_shmem(create_only,
721 name,
722 get_mem_size(max_msg_size, max_num_msg),
723 read_write,
724 static_cast<void*>(0),
725 //Prepare initialization functor
726 ipcdetail::msg_queue_initialization_func_t<VoidPointer> (max_num_msg, max_msg_size),
727 perm)
728{}
729
730template<class VoidPointer>
731inline message_queue_t<VoidPointer>::message_queue_t(open_or_create_t,
732 const char *name,
733 size_type max_num_msg,
734 size_type max_msg_size,
735 const permissions &perm)
736 //Create shared memory and execute functor atomically
737 : m_shmem(open_or_create,
738 name,
739 get_mem_size(max_msg_size, max_num_msg),
740 read_write,
741 static_cast<void*>(0),
742 //Prepare initialization functor
743 ipcdetail::msg_queue_initialization_func_t<VoidPointer> (max_num_msg, max_msg_size),
744 perm)
745{}
746
747template<class VoidPointer>
748inline message_queue_t<VoidPointer>::message_queue_t(open_only_t, const char *name)
749 //Create shared memory and execute functor atomically
750 : m_shmem(open_only,
751 name,
752 read_write,
753 static_cast<void*>(0),
754 //Prepare initialization functor
755 ipcdetail::msg_queue_initialization_func_t<VoidPointer> ())
756{}
757
758#if defined(BOOST_INTERPROCESS_WCHAR_NAMED_RESOURCES) || defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
759
760template<class VoidPointer>
761inline message_queue_t<VoidPointer>::message_queue_t(create_only_t,
762 const wchar_t *name,
763 size_type max_num_msg,
764 size_type max_msg_size,
765 const permissions &perm)
766 //Create shared memory and execute functor atomically
767 : m_shmem(create_only,
768 name,
769 get_mem_size(max_msg_size, max_num_msg),
770 read_write,
771 static_cast<void*>(0),
772 //Prepare initialization functor
773 ipcdetail::msg_queue_initialization_func_t<VoidPointer> (max_num_msg, max_msg_size),
774 perm)
775{}
776
777template<class VoidPointer>
778inline message_queue_t<VoidPointer>::message_queue_t(open_or_create_t,
779 const wchar_t *name,
780 size_type max_num_msg,
781 size_type max_msg_size,
782 const permissions &perm)
783 //Create shared memory and execute functor atomically
784 : m_shmem(open_or_create,
785 name,
786 get_mem_size(max_msg_size, max_num_msg),
787 read_write,
788 static_cast<void*>(0),
789 //Prepare initialization functor
790 ipcdetail::msg_queue_initialization_func_t<VoidPointer> (max_num_msg, max_msg_size),
791 perm)
792{}
793
794template<class VoidPointer>
795inline message_queue_t<VoidPointer>::message_queue_t(open_only_t, const wchar_t *name)
796 //Create shared memory and execute functor atomically
797 : m_shmem(open_only,
798 name,
799 read_write,
800 static_cast<void*>(0),
801 //Prepare initialization functor
802 ipcdetail::msg_queue_initialization_func_t<VoidPointer> ())
803{}
804
805#endif //defined(BOOST_INTERPROCESS_WCHAR_NAMED_RESOURCES) || defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
806
807template<class VoidPointer>
808inline void message_queue_t<VoidPointer>::send
809 (const void *buffer, size_type buffer_size, unsigned int priority)
810{ this->do_send<blocking>(buffer, buffer_size, priority, 0); }
811
812template<class VoidPointer>
813inline bool message_queue_t<VoidPointer>::try_send
814 (const void *buffer, size_type buffer_size, unsigned int priority)
815{ return this->do_send<non_blocking>(buffer, buffer_size, priority, 0); }
816
817template<class VoidPointer>
818template<class TimePoint>
819inline bool message_queue_t<VoidPointer>::timed_send
820 (const void *buffer, size_type buffer_size
821 ,unsigned int priority, const TimePoint &abs_time)
822{
823 if(ipcdetail::is_pos_infinity(abs_time)){
824 this->send(buffer, buffer_size, priority);
825 return true;
826 }
827 return this->do_send<timed>(buffer, buffer_size, priority, abs_time);
828}
829
830template<class VoidPointer>
831template<mqblock_types Block, class TimePoint>
832inline bool message_queue_t<VoidPointer>::do_send(
833 const void *buffer, size_type buffer_size,
834 unsigned int priority, const TimePoint &abs_time)
835{
836 ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
837 //Check if buffer is smaller than maximum allowed
838 if (buffer_size > p_hdr->m_max_msg_size) {
839 throw interprocess_exception(size_error);
840 }
841
842 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
843 bool notify_blocked_receivers = false;
844 #endif
845 //---------------------------------------------
846 scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
847 //---------------------------------------------
848 {
849 //If the queue is full execute blocking logic
850 if (p_hdr->is_full()) {
851 BOOST_TRY{
852 #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
853 ++p_hdr->m_blocked_senders;
854 #endif
855 switch(Block){
856 case non_blocking :
857 #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
858 --p_hdr->m_blocked_senders;
859 #endif
860 return false;
861 break;
862
863 case blocking :
864 do{
865 (void)do_cond_wait(ipcdetail::bool_<false>(), p_hdr->m_cond_send, lock, abs_time);
866 }
867 while (p_hdr->is_full());
868 break;
869
870 case timed :
871 do{
872 if(!do_cond_wait(ipcdetail::bool_<Block == timed>(), p_hdr->m_cond_send, lock, abs_time)) {
873 if(p_hdr->is_full()){
874 #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
875 --p_hdr->m_blocked_senders;
876 #endif
877 return false;
878 }
879 break;
880 }
881 }
882 while (p_hdr->is_full());
883 break;
884 default:
885 break;
886 }
887 #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
888 --p_hdr->m_blocked_senders;
889 #endif
890 }
891 BOOST_CATCH(...){
892 #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
893 --p_hdr->m_blocked_senders;
894 #endif
895 BOOST_RETHROW;
896 } BOOST_CATCH_END
897 }
898
899 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
900 notify_blocked_receivers = 0 != p_hdr->m_blocked_receivers;
901 #endif
902 //Insert the first free message in the priority queue
903 ipcdetail::msg_hdr_t<VoidPointer> &free_msg_hdr = p_hdr->queue_free_msg(priority);
904
905 //Sanity check, free msgs are always cleaned when received
906 BOOST_ASSERT(free_msg_hdr.priority == 0);
907 BOOST_ASSERT(free_msg_hdr.len == 0);
908
909 //Copy control data to the free message
910 free_msg_hdr.priority = priority;
911 free_msg_hdr.len = buffer_size;
912
913 //Copy user buffer to the message
914 std::memcpy(dest: free_msg_hdr.data(), src: buffer, n: buffer_size);
915 } // Lock end
916
917 //Notify outside lock to avoid contention. This might produce some
918 //spurious wakeups, but it's usually far better than notifying inside.
919 //If this message changes the queue empty state, notify it to receivers
920 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
921 if (notify_blocked_receivers){
922 p_hdr->m_cond_recv.notify_one();
923 }
924 #else
925 p_hdr->m_cond_recv.notify_one();
926 #endif
927
928 return true;
929}
930
931template<class VoidPointer>
932inline void message_queue_t<VoidPointer>::receive(void *buffer, size_type buffer_size,
933 size_type &recvd_size, unsigned int &priority)
934{ this->do_receive<blocking>(buffer, buffer_size, recvd_size, priority, 0); }
935
936template<class VoidPointer>
937inline bool
938 message_queue_t<VoidPointer>::try_receive(void *buffer, size_type buffer_size,
939 size_type &recvd_size, unsigned int &priority)
940{ return this->do_receive<non_blocking>(buffer, buffer_size, recvd_size, priority, 0); }
941
942template<class VoidPointer>
943template<class TimePoint>
944inline bool
945 message_queue_t<VoidPointer>::timed_receive(void *buffer, size_type buffer_size,
946 size_type &recvd_size, unsigned int &priority,
947 const TimePoint &abs_time)
948{
949 if(ipcdetail::is_pos_infinity(abs_time)){
950 this->receive(buffer, buffer_size, recvd_size, priority);
951 return true;
952 }
953 return this->do_receive<timed>(buffer, buffer_size, recvd_size, priority, abs_time);
954}
955
956template<class VoidPointer>
957template<mqblock_types Block, class TimePoint>
958inline bool
959 message_queue_t<VoidPointer>::do_receive(
960 void *buffer, size_type buffer_size,
961 size_type &recvd_size, unsigned int &priority,
962 const TimePoint &abs_time)
963{
964 ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
965 //Check if buffer is big enough for any message
966 if (buffer_size < p_hdr->m_max_msg_size) {
967 throw interprocess_exception(size_error);
968 }
969
970 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
971 bool notify_blocked_senders = false;
972 #endif
973 //---------------------------------------------
974 scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
975 //---------------------------------------------
976 {
977 //If there are no messages execute blocking logic
978 if (p_hdr->is_empty()) {
979 BOOST_TRY{
980 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
981 ++p_hdr->m_blocked_receivers;
982 #endif
983 switch(Block){
984 case non_blocking :
985 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
986 --p_hdr->m_blocked_receivers;
987 #endif
988 return false;
989 break;
990
991 case blocking :
992 do{
993 (void)do_cond_wait(ipcdetail::bool_<false>(), p_hdr->m_cond_recv, lock, abs_time);
994 }
995 while (p_hdr->is_empty());
996 break;
997
998 case timed :
999 do{
1000 if(!do_cond_wait(ipcdetail::bool_<Block == timed>(), p_hdr->m_cond_recv, lock, abs_time)) {
1001 if(p_hdr->is_empty()){
1002 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
1003 --p_hdr->m_blocked_receivers;
1004 #endif
1005 return false;
1006 }
1007 break;
1008 }
1009 }
1010 while (p_hdr->is_empty());
1011 break;
1012
1013 //Paranoia check
1014 default:
1015 break;
1016 }
1017 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
1018 --p_hdr->m_blocked_receivers;
1019 #endif
1020 }
1021 BOOST_CATCH(...){
1022 #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
1023 --p_hdr->m_blocked_receivers;
1024 #endif
1025 BOOST_RETHROW;
1026 } BOOST_CATCH_END
1027 }
1028
1029 #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
1030 notify_blocked_senders = 0 != p_hdr->m_blocked_senders;
1031 #endif
1032
1033 //There is at least one message ready to pick, get the top one
1034 ipcdetail::msg_hdr_t<VoidPointer> &top_msg = p_hdr->top_msg();
1035
1036 //Get data from the message
1037 recvd_size = top_msg.len;
1038 priority = top_msg.priority;
1039
1040 //Some cleanup to ease debugging
1041 top_msg.len = 0;
1042 top_msg.priority = 0;
1043
1044 //Copy data to receiver's bufers
1045 std::memcpy(dest: buffer, src: top_msg.data(), n: recvd_size);
1046
1047 //Free top message and put it in the free message list
1048 p_hdr->free_top_msg();
1049 } //Lock end
1050
1051 //Notify outside lock to avoid contention. This might produce some
1052 //spurious wakeups, but it's usually far better than notifying inside.
1053 //If this reception changes the queue full state, notify senders
1054 #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
1055 if (notify_blocked_senders){
1056 p_hdr->m_cond_send.notify_one();
1057 }
1058 #else
1059 p_hdr->m_cond_send.notify_one();
1060 #endif
1061
1062 return true;
1063}
1064
1065template<class VoidPointer>
1066inline typename message_queue_t<VoidPointer>::size_type message_queue_t<VoidPointer>::get_max_msg() const
1067{
1068 ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
1069 return p_hdr ? p_hdr->m_max_num_msg : 0; }
1070
1071template<class VoidPointer>
1072inline typename message_queue_t<VoidPointer>::size_type message_queue_t<VoidPointer>::get_max_msg_size() const
1073{
1074 ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
1075 return p_hdr ? p_hdr->m_max_msg_size : 0;
1076}
1077
1078template<class VoidPointer>
1079inline typename message_queue_t<VoidPointer>::size_type message_queue_t<VoidPointer>::get_num_msg() const
1080{
1081 ipcdetail::mq_hdr_t<VoidPointer> *p_hdr = static_cast<ipcdetail::mq_hdr_t<VoidPointer>*>(m_shmem.get_user_address());
1082 if(p_hdr){
1083 //---------------------------------------------
1084 scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
1085 //---------------------------------------------
1086 return p_hdr->m_cur_num_msg;
1087 }
1088
1089 return 0;
1090}
1091
1092template<class VoidPointer>
1093inline bool message_queue_t<VoidPointer>::remove(const char *name)
1094{ return shared_memory_object::remove(filename: name); }
1095
1096#if defined(BOOST_INTERPROCESS_WCHAR_NAMED_RESOURCES) || defined(BOOST_INTERPROCESS_DOXYGEN_INVOKED)
1097
1098template<class VoidPointer>
1099inline bool message_queue_t<VoidPointer>::remove(const wchar_t *name)
1100{ return shared_memory_object::remove(name); }
1101
1102#endif
1103
1104#else
1105
1106//!Typedef for a default message queue
1107//!to be used between processes
1108typedef message_queue_t<offset_ptr<void> > message_queue;
1109
1110#endif //#ifndef BOOST_INTERPROCESS_DOXYGEN_INVOKED
1111
1112}} //namespace boost{ namespace interprocess{
1113
1114#include <boost/interprocess/detail/config_end.hpp>
1115
1116#endif //#ifndef BOOST_INTERPROCESS_MESSAGE_QUEUE_HPP
1117

source code of boost/libs/interprocess/include/boost/interprocess/ipc/message_queue.hpp