More eagerly reset shared state in async_rw_mutex

pika-org · May 8, 2023 · 76f8e93 · 76f8e93
1 parent b8d0ffc
commit 76f8e93
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 0 deletions.
diff --git a/libs/pika/synchronization/include/pika/synchronization/async_rw_mutex.hpp b/libs/pika/synchronization/include/pika/synchronization/async_rw_mutex.hpp
@@ -458,6 +458,7 @@ namespace pika::execution::experimental {
                         // add a continuation to be triggered when the previous
                         // state is released.
                         p->add_continuation(PIKA_MOVE(continuation));
+                        os.state.reset();
                         os.prev_state.reset();
                     }
                     else
@@ -653,6 +654,7 @@ namespace pika::execution::experimental {
                         // add a continuation to be triggered when the previous
                         // state is released.
                         p->add_continuation(PIKA_MOVE(continuation));
+                        os.state.reset();
                         os.prev_state.reset();
                     }
                     else

diff --git a/libs/pika/synchronization/tests/unit/async_rw_mutex.cpp b/libs/pika/synchronization/tests/unit/async_rw_mutex.cpp
@@ -24,6 +24,7 @@ using pika::execution::experimental::start_detached;
 using pika::execution::experimental::then;
 using pika::execution::experimental::thread_pool_scheduler;
 using pika::execution::experimental::transfer;
+using pika::execution::experimental::when_all;
 using pika::this_thread::experimental::sync_wait;
 
 unsigned int seed = std::random_device{}();
@@ -259,6 +260,31 @@ void test_read_sender_copyable(async_rw_mutex<ReadWriteT, ReadT> rwm)
     PIKA_TEST_EQ(read_accesses, std::size_t(4));
 }
 
+template <typename ReadWriteT, typename ReadT = ReadWriteT>
+void test_multiple_when_all(async_rw_mutex<ReadWriteT, ReadT> rwm)
+{
+    {
+        auto s1 = rwm.readwrite() | then([](auto) {});
+        auto s2 = rwm.readwrite() | then([](auto) {});
+        auto s3 = rwm.readwrite() | then([](auto) {});
+        sync_wait(when_all(std::move(s1), std::move(s2), std::move(s3)));
+    }
+
+    {
+        auto s1 = rwm.readwrite() | then([](auto) {});
+        auto s2 = rwm.readwrite() | then([](auto) {});
+        auto s3 = rwm.readwrite() | then([](auto) {});
+        sync_wait(when_all(std::move(s3), std::move(s2), std::move(s1)));
+    }
+
+    {
+        auto s1 = rwm.readwrite() | then([](auto) {});
+        auto s2 = rwm.readwrite() | then([](auto) {});
+        auto s3 = rwm.readwrite() | then([](auto) {});
+        sync_wait(when_all(std::move(s3), std::move(s1), std::move(s2)));
+    }
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 int pika_main(pika::program_options::variables_map& vm)
 {
@@ -292,6 +318,10 @@ int pika_main(pika::program_options::variables_map& vm)
     test_read_sender_copyable(async_rw_mutex<std::size_t>{0});
     test_read_sender_copyable(async_rw_mutex<mytype, mytype_base>{mytype{}});
 
+    test_multiple_when_all(async_rw_mutex<void>{});
+    test_multiple_when_all(async_rw_mutex<std::size_t>{0});
+    test_multiple_when_all(async_rw_mutex<mytype, mytype_base>{mytype{}});
+
     return pika::finalize();
 }