Bugfix: Removed deadlock when shutting down active Reader

Author: joto

include/osmium/io/detail/queue_util.hpp

@@ -101,13 +101,11 @@ namespace osmium {
             class queue_wrapper {
 
                 future_queue_type<T>& m_queue;
-                bool m_has_reached_end_of_data;
 
             public:
 
                 explicit queue_wrapper(future_queue_type<T>& queue) :
-                    m_queue(queue),
-                    m_has_reached_end_of_data(false) {
+                    m_queue(queue) {
                 }
 
                 queue_wrapper(const queue_wrapper&) = delete;
@@ -117,33 +115,30 @@ namespace osmium {
                 queue_wrapper& operator=(queue_wrapper&&) = delete;
 
                 ~queue_wrapper() noexcept {
-                    drain();
+                    try {
+                        shutdown();
+                    } catch (...) {
+                    }
                 }
 
-                void drain() {
-                    while (!m_queue.empty()) {
-                        try {
-                            std::future<T> data_future;
-                            m_queue.try_pop(data_future);
-                        } catch (...) {
-                            // Ignore any exceptions.
-                        }
-                    }
+                void shutdown() {
+                    m_queue.shutdown();
                 }
 
                 bool has_reached_end_of_data() const noexcept {
-                    return m_has_reached_end_of_data;
+                    return !m_queue.in_use();
                 }
 
                 T pop() {
                     T data;
-                    if (!m_has_reached_end_of_data) {
+                    if (m_queue.in_use()) {
                         std::future<T> data_future;
                         m_queue.wait_and_pop(data_future);
-                        assert(data_future.valid());
-                        data = std::move(data_future.get());
-                        if (at_end_of_data(data)) {
-                            m_has_reached_end_of_data = true;
+                        if (data_future.valid()) {
+                            data = std::move(data_future.get());
+                            if (at_end_of_data(data)) {
+                                m_queue.shutdown();
+                            }
                         }
                     }
                     return data;

include/osmium/io/detail/write_thread.hpp

@@ -97,7 +97,7 @@ namespace osmium {
                     } catch (...) {
                         m_notification->store(true);
                         m_promise.set_exception(std::current_exception());
-                        m_queue.drain();
+                        m_queue.shutdown();
                     }
                 }
 

include/osmium/io/reader.hpp

@@ -386,7 +386,7 @@ namespace osmium {
 
                 m_read_thread_manager.stop();
 
-                m_osmdata_queue_wrapper.drain();
+                m_osmdata_queue_wrapper.shutdown();
 
                 try {
                     m_read_thread_manager.close();

include/osmium/thread/queue.hpp

@@ -33,6 +33,7 @@ DEALINGS IN THE SOFTWARE.
 
 */
 
+#include <atomic>
 #include <chrono>
 #include <condition_variable>
 #include <cstddef>
@@ -42,7 +43,6 @@ DEALINGS IN THE SOFTWARE.
 #include <utility> // IWYU pragma: keep
 
 #ifdef OSMIUM_DEBUG_QUEUE_SIZE
-# include <atomic>
 # include <iostream>
 #endif
 
@@ -73,6 +73,8 @@ namespace osmium {
             /// Used to signal producers when queue is not full.
             std::condition_variable m_space_available;
 
+            std::atomic<bool> m_in_use{true};
+
 #ifdef OSMIUM_DEBUG_QUEUE_SIZE
             /// The largest size the queue has been so far.
             std::size_t m_largest_size;
@@ -145,6 +147,9 @@ namespace osmium {
              * this call will block if the queue is full.
              */
             void push(T value) {
+                if (!m_in_use) {
+                    return;
+                }
                 constexpr const std::chrono::milliseconds max_wait{10};
 #ifdef OSMIUM_DEBUG_QUEUE_SIZE
                 ++m_push_counter;
@@ -181,7 +186,7 @@ namespace osmium {
                 }
 #endif
                 m_data_available.wait(lock, [this] {
-                    return !m_queue.empty();
+                    return !m_in_use || !m_queue.empty();
                 });
                 if (!m_queue.empty()) {
                     value = std::move(m_queue.front());
@@ -224,6 +229,19 @@ namespace osmium {
                 return m_queue.size();
             }
 
+            bool in_use() const noexcept {
+                return m_in_use;
+            }
+
+            void shutdown() {
+                m_in_use = false;
+                std::lock_guard<std::mutex> lock{m_mutex};
+                while (!m_queue.empty()) {
+                    m_queue.pop();
+                }
+                m_data_available.notify_all();
+            }
+
         }; // class Queue
 
     } // namespace thread

test/t/thread/test_queue.cpp

@@ -18,3 +18,30 @@ TEST_CASE("Queue can have max elements and can be named") {
     osmium::thread::Queue<int> queue{100, "Queue of max size 100"};
 }
 
+TEST_CASE("When queue is shut down, nothing goes in or out") {
+    osmium::thread::Queue<std::string> queue;
+    REQUIRE(queue.in_use());
+    REQUIRE(queue.empty());
+    queue.push("foo");
+    queue.push("bar");
+    queue.push("baz");
+    REQUIRE(queue.size() == 3);
+
+    std::string value;
+
+    queue.wait_and_pop(value);
+    REQUIRE(value == "foo");
+    REQUIRE(queue.size() == 2);
+    REQUIRE(queue.in_use());
+    queue.shutdown();
+    REQUIRE_FALSE(queue.in_use());
+    REQUIRE(queue.empty());
+    queue.push("lost");
+    REQUIRE(queue.empty());
+
+    value.clear();
+    queue.try_pop(value);
+    REQUIRE(value.empty());
+    queue.wait_and_pop(value);
+    REQUIRE(value.empty());
+}