Fix SIGSEGV in lock_free_queue::guard_ptr

Author: dobord

include/coro/detail/lockfree_queue.hpp

@@ -69,30 +69,26 @@ class guard_ptr
     {
         using namespace std::chrono_literals;
 
-        node<T>* current = ref.load(std::memory_order::relaxed);
         do
         {
-            node<T>* result = current;
-            m_node_ptr      = current;
-            if (m_node_ptr)
+            m_node_ptr = ref.load(std::memory_order::acquire);
+            if (!m_node_ptr)
             {
-                m_node_ptr->ref_count.fetch_add(1, std::memory_order::acq_rel);
-                if (m_node_ptr->is_removed.load(std::memory_order::acquire))
-                {
-                    decrement();
-                    m_is_removed = true;
-                    return;
-                }
+                return;
             }
-            current = ref.load(std::memory_order::acquire);
-            if (result == current)
-                break;
-
-            if (m_node_ptr)
+            if (m_node_ptr->is_removed.load(std::memory_order::acquire))
             {
-                decrement();
+                m_is_removed = true;
+                m_node_ptr   = nullptr;
+                return;
+            }
+            node<T>* current = ref.load(std::memory_order::acquire);
+            if (m_node_ptr == current)
+            {
+                break;
             }
         } while (true);
+        m_node_ptr->ref_count.fetch_add(1, std::memory_order::acq_rel);
     }
 
     void remove()
@@ -222,33 +218,33 @@ class lockfree_queue
         using namespace std::chrono_literals;
 
         auto new_node = m_alloc.allocate(value);
+        assert(new_node->next.load() == nullptr);
         while (true)
         {
             guard_type tail_guard(m_tail, [this](node_type* n) { dispose_node(n); });
             if (tail_guard.is_removed())
             {
                 continue;
             }
-            auto       tail_node = tail_guard.value();
-            guard_type next_guard(tail_node->next, [this](node_type* n) { dispose_node(n); });
-            if (next_guard.is_removed())
-            {
-                continue;
-            }
-            auto next_node = next_guard.value();
+            auto tail_node = tail_guard.value();
+            assert(tail_node != new_node);
+            auto next_node = tail_node->next.load(std::memory_order::acquire);
 
             if (next_node != nullptr)
             {
+                assert(tail_node != next_node);
+
                 // Tail is misplaced, advance it
                 m_tail.compare_exchange_weak(
                     tail_node, next_node, std::memory_order::release, std::memory_order::relaxed);
                 continue;
             }
 
-            node_type* expected{};
+            node_type* expected = nullptr;
             if (tail_node->next.compare_exchange_strong(
                     expected, new_node, std::memory_order::release, std::memory_order::relaxed))
             {
+                assert(tail_node != new_node);
                 m_tail.compare_exchange_strong(
                     tail_node, new_node, std::memory_order::release, std::memory_order::relaxed);
                 return;
@@ -291,12 +287,16 @@ class lockfree_queue
 
             if (head_node == tail_node)
             {
+                assert(tail_node != next_node);
+
                 // It is needed to help push()
                 m_tail.compare_exchange_strong(
                     tail_node, next_node, std::memory_order::release, std::memory_order::relaxed);
                 continue;
             }
 
+            assert(head_node != next_node);
+
             if (m_head.compare_exchange_strong(
                     head_node, next_node, std::memory_order::acquire, std::memory_order::relaxed))
             {

src/condition_variable.cpp

@@ -72,7 +72,8 @@ auto detail::strategy_based_on_io_scheduler::timeout_task(
     using namespace std::chrono_literals;
 
     assert(!m_scheduler.expired());
-    auto deadline = steady_clock::now() + timeout;
+    auto deadline   = steady_clock::now() + timeout;
+    auto stop_token = stop_source.get_token();
 
     while ((steady_clock::now() < deadline) && !stop_source.stop_requested())
     {
@@ -84,7 +85,10 @@ auto detail::strategy_based_on_io_scheduler::timeout_task(
         }
     }
 
-    extract_waiter(wo.get());
+    if (!stop_token.stop_requested())
+    {
+        extract_waiter(wo.get());
+    }
     co_return timeout_status::timeout;
 }
 

test/test_condition_variable.cpp

@@ -245,6 +245,7 @@ TEST_CASE("condition_variable for thread-safe-queue between producers and consum
         {
             {
                 auto ulock = co_await bp.m.lock();
+                REQUIRE_FALSE(bp.m.try_lock());
                 auto value = bp.next.fetch_add(1, std::memory_order::acq_rel);
 
                 // limit for end of test
@@ -268,7 +269,14 @@ TEST_CASE("condition_variable for thread-safe-queue between producers and consum
         while (true)
         {
             auto ulock = co_await bp.m.lock();
-            co_await bp.cv.wait(ulock, [&bp]() { return bp.q.size() || bp.cancel.load(std::memory_order::acquire); });
+            co_await bp.cv.wait(
+                ulock,
+                [&bp]()
+                {
+                    REQUIRE_FALSE(bp.m.try_lock());
+                    return bp.q.size() || bp.cancel.load(std::memory_order::acquire);
+                });
+            REQUIRE_FALSE(bp.m.try_lock());
             if (bp.cancel.load(std::memory_order::acquire))
             {
                 break;
@@ -294,6 +302,7 @@ TEST_CASE("condition_variable for thread-safe-queue between producers and consum
         {
             {
                 auto ulock = co_await bp.m.lock();
+                REQUIRE_FALSE(bp.m.try_lock());
                 if ((bp.next.load(std::memory_order::acquire) >= bp.max_value) && bp.q.empty())
                 {
                     break;