[Boost-commit] svn:boost r86316 - in trunk: boost/interprocess/ipc libs/interprocess/doc libs/interprocess/test

Subject: [Boost-commit] svn:boost r86316 - in trunk: boost/interprocess/ipc libs/interprocess/doc libs/interprocess/test
From: igaztanaga_at_[hidden]
Date: 2013-10-15 04:02:10

Author: igaztanaga
Date: 2013-10-15 04:02:09 EDT (Tue, 15 Oct 2013)
New Revision: 86316
URL: http://svn.boost.org/trac/boost/changeset/86316

Log:
Fixes #9221 ("message_queue deadlock on linux")

Text files modified: 
   trunk/boost/interprocess/ipc/message_queue.hpp      |   190 ++++++++++++++++++++++++++------------- 
   trunk/libs/interprocess/doc/interprocess.qbk        |    10 ++                                      
   trunk/libs/interprocess/test/message_queue_test.cpp |    84 +++++++++++++++++                       
   3 files changed, 220 insertions(+), 64 deletions(-)

Modified: trunk/boost/interprocess/ipc/message_queue.hpp
==============================================================================
--- trunk/boost/interprocess/ipc/message_queue.hpp	Tue Oct 15 01:23:53 2013	(r86315)
+++ trunk/boost/interprocess/ipc/message_queue.hpp	2013-10-15 04:02:09 EDT (Tue, 15 Oct 2013)	(r86316)
@@ -306,6 +306,8 @@
          m_cur_num_msg(0)
          #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
          ,m_cur_first_msg(0u)
+         ,m_blocked_senders(0u)
+         ,m_blocked_receivers(0u)
          #endif
       {  this->initialize_memory();  }
 
@@ -376,17 +378,17 @@
    {
       iterator it_inserted_ptr_end = this->inserted_ptr_end();
       iterator it_inserted_ptr_beg = this->inserted_ptr_begin();
-      if(where == it_inserted_ptr_end){
-         ++m_cur_num_msg;
-         return **it_inserted_ptr_end;
-      }
-      else if(where == it_inserted_ptr_beg){
+      if(where == it_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 if(where == it_inserted_ptr_end){
+         ++m_cur_num_msg;
+         return **it_inserted_ptr_end;
+      }
       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);
@@ -452,7 +454,7 @@
       }
    }
 
-   #else
+   #else //BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
 
    typedef msg_hdr_ptr_t *iterator;
 
@@ -482,7 +484,7 @@
       return **pos;
    }
 
-   #endif
+   #endif   //BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
 
    //!Inserts the first free message in the priority queue
    msg_header & queue_free_msg(unsigned int priority)
@@ -507,7 +509,7 @@
             //Check where the free message should be placed
             it = this->lower_bound(dummy_ptr, static_cast<priority_functor<VoidPointer>&>(*this));
          }
-         
+         assert(0);  
       }
       //Insert the free message in the correct position
       return this->insert_at(it);
@@ -577,6 +579,8 @@
    #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
    //Current start offset in the circular index
    size_type                  m_cur_first_msg;
+   size_type                  m_blocked_senders;
+   size_type                  m_blocked_receivers;
    #endif
 };
 
@@ -714,41 +718,67 @@
       throw interprocess_exception(size_error);
    }
 
-   bool was_empty = false;
+   #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+   bool notify_blocked_receivers = false;
+   #endif
    //---------------------------------------------
    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;
+         BOOST_TRY{
+            #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
+            ++p_hdr->m_blocked_senders;
+            #endif
+            switch(block){
+               case non_blocking :
+                  #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
+                  --p_hdr->m_blocked_senders;
+                  #endif
+                  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()){
+                           #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
+                           --p_hdr->m_blocked_senders;
+                           #endif
+                           return false;
+                        }
+                        break;
+                     }
                   }
-               }
-               while (p_hdr->is_full());
-            break;
-            default:
-            break;
+                  while (p_hdr->is_full());
+               break;
+               default:
+               break;
+            }
+            #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
+            --p_hdr->m_blocked_senders;
+            #endif
+         }
+         BOOST_CATCH(...){
+            #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
+            --p_hdr->m_blocked_senders;
+            #endif
+            BOOST_RETHROW;
          }
+         BOOST_CATCH_END
       }
 
-      was_empty = p_hdr->is_empty();
+      #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+      notify_blocked_receivers = 0 != p_hdr->m_blocked_receivers;
+      #endif
       //Insert the first free message in the priority queue
       ipcdetail::msg_hdr_t<VoidPointer> &free_msg_hdr = p_hdr->queue_free_msg(priority);
 
@@ -767,9 +797,13 @@
    //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){
+   #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+   if (notify_blocked_receivers){
       p_hdr->m_cond_recv.notify_one();
    }
+   #else
+   p_hdr->m_cond_recv.notify_one();
+   #endif
 
    return true;
 }
@@ -811,42 +845,70 @@
       throw interprocess_exception(size_error);
    }
 
-   bool was_full = false;
+   #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+   bool notify_blocked_senders = false;
+   #endif
    //---------------------------------------------
    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;
+         BOOST_TRY{
+            #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+            ++p_hdr->m_blocked_receivers;
+            #endif
+            switch(block){
+               case non_blocking :
+                  #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+                  --p_hdr->m_blocked_receivers;
+                  #endif
+                  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()){
+                           #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+                           --p_hdr->m_blocked_receivers;
+                           #endif
+                           return false;
+                        }
+                        break;
+                     }
                   }
-               }
-               while (p_hdr->is_empty());
-            break;
-
-            //Paranoia check
-            default:
-            break;
+                  while (p_hdr->is_empty());
+               break;
+
+               //Paranoia check
+               default:
+               break;
+            }
+            #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+            --p_hdr->m_blocked_receivers;
+            #endif
+         }
+         BOOST_CATCH(...){
+            #if defined(BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX)
+            --p_hdr->m_blocked_receivers;
+            #endif
+            BOOST_RETHROW;
          }
+         BOOST_CATCH_END
       }
 
+      #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
+      notify_blocked_senders = 0 != p_hdr->m_blocked_senders;
+      #endif
+
       //There is at least one message ready to pick, get the top one
       ipcdetail::msg_hdr_t<VoidPointer> &top_msg = p_hdr->top_msg();
 
@@ -861,8 +923,6 @@
       //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
@@ -870,9 +930,13 @@
    //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){
+   #ifdef BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX
+   if (notify_blocked_senders){
       p_hdr->m_cond_send.notify_one();
    }
+   #else
+   p_hdr->m_cond_send.notify_one();
+   #endif
 
    return true;
 }

Modified: trunk/libs/interprocess/doc/interprocess.qbk
==============================================================================
--- trunk/libs/interprocess/doc/interprocess.qbk	Tue Oct 15 01:23:53 2013	(r86315)
+++ trunk/libs/interprocess/doc/interprocess.qbk	2013-10-15 04:02:09 EDT (Tue, 15 Oct 2013)	(r86316)
@@ -6713,6 +6713,16 @@
 
 [section:release_notes Release Notes]
 
+[section:release_notes_boost_1_56_00 Boost 1.56 Release]
+*  Fixed bugs: 
+   * [@https://svn.boost.org/trac/boost/ticket/9221 #9221 ("message_queue deadlock on linux")].
+      
+*  [*ABI breaking]: [@https://svn.boost.org/trac/boost/ticket/9221 #9221] showed
+   that `BOOST_INTERPROCESS_MSG_QUEUE_CIRCULAR_INDEX` option of message queue,
+   was completely broken so a ABI break was necessary to have a working implementation.
+
+[endsect]
+
 [section:release_notes_boost_1_55_00 Boost 1.55 Release]
 
 *  Fixed bugs [@https://svn.boost.org/trac/boost/ticket/7156 #7156],

Modified: trunk/libs/interprocess/test/message_queue_test.cpp
==============================================================================
--- trunk/libs/interprocess/test/message_queue_test.cpp	Tue Oct 15 01:23:53 2013	(r86315)
+++ trunk/libs/interprocess/test/message_queue_test.cpp	2013-10-15 04:02:09 EDT (Tue, 15 Oct 2013)	(r86316)
@@ -17,6 +17,7 @@
 #include <boost/interprocess/allocators/node_allocator.hpp>
 #include <boost/interprocess/detail/os_thread_functions.hpp>
 #include <vector>
+#include <iostream>
 #include <cstddef>
 #include <limits>
 #include <memory>
@@ -222,12 +223,12 @@
    return true;
 }
 //]
+
 static const int MsgSize = 10;
 static const int NumMsg  = 1000;
 static char msgsend [10];
 static char msgrecv [10];
 
-
 static boost::interprocess::message_queue *pmessage_queue;
 
 void receiver()
@@ -267,6 +268,83 @@
    return true;
 }
 
+
+//////////////////////////////////////////////////////////////////////////////
+//
+// test_multi_sender_receiver is based on Alexander (aalutov's)
+// testcase for ticket #9221. Many thanks.
+//
+//////////////////////////////////////////////////////////////////////////////
+
+static boost::interprocess::message_queue *global_queue = 0;
+//We'll send MULTI_NUM_MSG_PER_SENDER messages per sender
+static const int MULTI_NUM_MSG_PER_SENDER = 10000;
+//Message queue message capacity
+static const int MULTI_QUEUE_SIZE = (MULTI_NUM_MSG_PER_SENDER - 1)/MULTI_NUM_MSG_PER_SENDER + 1;
+//We'll launch MULTI_THREAD_COUNT senders and MULTI_THREAD_COUNT receivers 
+static const int MULTI_THREAD_COUNT = 10;
+
+static void multisend()
+{
+	char buff;
+	for (int i = 0; i < MULTI_NUM_MSG_PER_SENDER; i++) {
+		global_queue->send(&buff, 1, 0);
+	}
+	global_queue->send(&buff, 0, 0);
+	//std::cout<<"writer thread complete"<<std::endl;
+}
+
+static void multireceive()
+{
+	char buff;
+	size_t size;
+   int received_msgs = 0;
+	unsigned int priority;
+	do {
+		global_queue->receive(&buff, 1, size, priority);
+      ++received_msgs;
+	} while (size > 0);
+   --received_msgs;
+	//std::cout << "reader thread complete, read msgs: " << received_msgs << std::endl;
+}
+
+
+bool test_multi_sender_receiver()
+{
+   bool ret = true;
+   //std::cout << "Testing multi-sender / multi-receiver " << std::endl;
+	try {
+		boost::interprocess::message_queue::remove(test::get_process_id_name());
+		boost::interprocess::message_queue mq
+         (boost::interprocess::open_or_create, test::get_process_id_name(), MULTI_QUEUE_SIZE, 1);
+      global_queue = &mq;
+      std::vector<boost::interprocess::ipcdetail::OS_thread_t> threads(MULTI_THREAD_COUNT*2);
+
+      //Launch senders receiver thread
+		for (int i = 0; i < MULTI_THREAD_COUNT; i++) {
+         boost::interprocess::ipcdetail::thread_launch
+            (threads[i], &multisend);
+		}
+
+		for (int i = 0; i < MULTI_THREAD_COUNT; i++) {
+         boost::interprocess::ipcdetail::thread_launch
+            (threads[MULTI_THREAD_COUNT+i], &multireceive);
+		}
+
+		for (int i = 0; i < MULTI_THREAD_COUNT*2; i++) {
+         boost::interprocess::ipcdetail::thread_join(threads[i]);
+         //std::cout << "Joined thread " << i << std::endl;
+		}
+	}
+   catch (std::exception &e) {
+		std::cout << "error " << e.what() << std::endl;
+      ret = false;
+	}
+   boost::interprocess::message_queue::remove(test::get_process_id_name());
+	return ret;
+}
+
+
 int main ()
 {
    if(!test_priority_order()){
@@ -281,6 +359,10 @@
       return 1;
    }
 
+   if(!test_multi_sender_receiver()){
+      return 1;
+   }
+
    return 0;
 }