Made sure detail::condition_variable can be safely destroyed

hpx/lcos/detail/future_data.hpp

@@ -244,6 +244,11 @@ namespace detail
 
                 // handle all threads waiting for the block to become full
                 cond_.notify_all(l, ec);
+
+                // Note: cond_.notify_all() may cause this shared state to be
+                // destroyed. This is, however, safe as the scoped lock above
+                // will not touch the wrapped mutex if it was unlocked inside
+                // notify_all().
             }
 
             // invoke the callback (continuation) function

hpx/lcos/local/counting_semaphore.hpp

@@ -129,6 +129,8 @@ namespace hpx { namespace lcos { namespace local
                 value_ += count;
                 for (boost::int64_t i = 0; value_ >= 0 && i < count; ++i)
                 {
+                    // notify_one() returns false if no more threads are
+                    // waiting
                     if (!cond_.notify_one(l))
                         break;
                 }

hpx/lcos/local/detail/condition_variable.hpp

@@ -74,7 +74,9 @@ namespace hpx { namespace lcos { namespace local { namespace detail
         {
             if (!queue_.empty())
             {
-                LERR_(fatal) << "~condition_variable: queue is not empty, aborting threads";
+                LERR_(fatal)
+                    << "~condition_variable: queue is not empty, "
+                       "aborting threads";
 
                 local::no_mutex no_mtx;
                 abort_all(no_mtx);
@@ -97,6 +99,9 @@ namespace hpx { namespace lcos { namespace local { namespace detail
             return queue_.size();
         }
 
+        // Return false if no more threads are waiting. If it returns false
+        // the lock is left unlocked, otherwise it will be relocked before
+        // returning.
         template <typename Lock>
         bool notify_one(Lock& lock, error_code& ec = throws)
         {
@@ -115,20 +120,36 @@ namespace hpx { namespace lcos { namespace local { namespace detail
                 queue_.front().id_ = threads::invalid_thread_id_repr;
                 queue_.pop_front();
 
-                util::scoped_unlock<Lock> unlock(lock);
+                if (!queue_.empty())
+                {
+                    util::scoped_unlock<Lock> unlock(lock);
 
-                threads::set_thread_state(threads::thread_id_type(
-                    reinterpret_cast<threads::thread_data_base*>(id)),
-                    threads::pending, threads::wait_timeout,
-                    threads::thread_priority_default, ec);
-                if (!ec) return true;
+                    threads::set_thread_state(threads::thread_id_type(
+                        reinterpret_cast<threads::thread_data_base*>(id)),
+                        threads::pending, threads::wait_timeout,
+                        threads::thread_priority_default, ec);
+                    if (!ec) return true;
+                }
+                else
+                {
+                    // Since this is the last thread waiting on this condition
+                    // variable it could happen that this instance will be
+                    // destroyed before set_thread_state() returns. We have to
+                    // make sure that this instance is not touched anymore.
+                    lock.unlock();
+
+                    threads::set_thread_state(threads::thread_id_type(
+                        reinterpret_cast<threads::thread_data_base*>(id)),
+                        threads::pending, threads::wait_timeout,
+                        threads::thread_priority_default, ec);
+                }
             }
-
             return false;
         }
 
+        // This leaves the lock unlocked
         template <typename Lock>
-        void notify_all(Lock& lock, error_code& ec = throws) // leaves the lock unlocked
+        void notify_all(Lock& lock, error_code& ec = throws)
         {
             HPX_ASSERT_OWNS_LOCK(lock);
 

hpx/lcos/local/promise.hpp

@@ -129,10 +129,10 @@ namespace hpx { namespace lcos { namespace local
                     return;
                 }
 
+                has_result_ = true;
+
                 shared_state_->set_result(std::forward<T>(value), ec);
                 if (ec) return;
-
-                has_result_ = true;
             }
 
 #ifndef BOOST_NO_CXX11_EXPLICIT_CONVERSION_OPERATORS

tests/regressions/lcos/CMakeLists.txt

@@ -1,4 +1,4 @@
-# Copyright (c) 2007-2013 Hartmut Kaiser
+# Copyright (c) 2007-2015 Hartmut Kaiser
 # Copyright (c) 2011-2012 Bryce Adelstein-Lelbach
 #
 # Distributed under the Boost Software License, Version 1.0. (See accompanying
@@ -18,6 +18,7 @@ set(tests
     lifetime_588
     lifetime_588_1
     promise_leak_996
+    safely_destroy_cv_1481
     set_hpx_limit_798
     shared_stated_leaked_1211
    )
@@ -30,7 +31,7 @@ set(tests ${tests}
     dataflow_future_swap
     dataflow_791
    )
- if(HPX_WITH_CXX11_LAMBDAS)
+if(HPX_WITH_CXX11_LAMBDAS)
   set(tests ${tests}
       dataflow_const_functor_773
       dataflow_future_swap2
@@ -56,6 +57,9 @@ if(HPX_WITH_CXX11_LAMBDAS)
     future_hang_on_wait_with_callback_629)
 endif()
 
+set(safely_destroy_cv_1481_PARAMETERS THREADS_PER_LOCALITY 2)
+
+# Create test cases
 foreach(test ${tests})
   set(sources
       ${test}.cpp)

tests/regressions/lcos/safely_destroy_cv_1481.cpp

@@ -0,0 +1,44 @@
+//  Copyright 2015 (c) Hartmut Kaiser
+//
+//  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)
+
+// This test case demonstrates the issue described in #1481:
+// Sync primitives safe destruction
+
+#include <hpx/hpx.hpp>
+#include <hpx/hpx_main.hpp>
+#include <hpx/include/local_lcos.hpp>
+#include <hpx/util/lightweight_test.hpp>
+
+void test_safe_destruction()
+{
+    hpx::thread t;
+    hpx::future<void> outer;
+
+    {
+        hpx::lcos::local::promise<void> p;
+        hpx::shared_future<void> inner = p.get_future().share();
+
+        // Delay returning from p.set_value() below to destroy the promise
+        // before set_value returns.
+        outer = inner.then(
+            [](hpx::shared_future<void> &&)
+            {
+                hpx::this_thread::sleep_for(boost::chrono::milliseconds(100));
+            });
+
+        // create a thread which will make the inner future ready
+        t = hpx::thread([&p]() { p.set_value(); });
+        inner.get();
+    }
+
+    outer.get();
+    t.join();
+}
+
+int main()
+{
+    test_safe_destruction();
+    return hpx::util::report_errors();
+}