Merge pull request arximboldi#157 from arximboldi/exception-safety

Fix exception safety of basic event loop queues

lager/event_loop/manual.hpp

@@ -26,20 +26,28 @@ struct with_manual_event_loop
     template <typename Fn>
     void async(Fn&& fn)
     {
-        LAGER_THROW(std::logic_error{"manual_event_loop does not support async()"});
+        LAGER_THROW(
+            std::logic_error{"manual_event_loop does not support async()"});
     }
 
     template <typename Fn>
     void post(Fn&& fn)
     {
-        auto is_root = queue_.empty();
         queue_.push_back(std::forward<Fn>(fn));
+        auto is_root = i_ == 0;
         if (is_root) {
-            for (auto i = std::size_t{}; i < queue_.size(); ++i) {
-                auto f = std::move(queue_[i]);
-                f();
+            try {
+                while (i_ < queue_.size()) {
+                    auto f = std::move(queue_[i_++]);
+                    std::move(f)();
+                }
+            } catch (...) {
+                queue_.erase(queue_.begin(), queue_.begin() + i_);
+                i_ = 0;
+                throw;
             }
             queue_.clear();
+            i_ = 0;
         }
     }
 
@@ -51,6 +59,7 @@ struct with_manual_event_loop
     using post_fn_t = std::function<void()>;
 
     std::vector<post_fn_t> queue_;
+    std::size_t i_ = {};
 };
 
 } // namespace lager

lager/event_loop/queue.hpp

@@ -35,14 +35,20 @@ struct queue_event_loop
         LAGER_THROW(std::logic_error{"not implemented!"});
     }
 
+    // If there is an exception, the step() function needs to be re-run for the
+    // queue to be fully processed.
     void step()
     {
-        auto is_root = !queue_.empty();
-        if (is_root) {
-            for (auto i = std::size_t{}; i < queue_.size(); ++i)
-                queue_[i]();
-            queue_.clear();
+        for (auto i = std::size_t{}; i < queue_.size();) {
+            try {
+                auto f = std::move(queue_[i++]);
+                std::move(f)();
+            } catch (...) {
+                queue_.erase(queue_.begin(), queue_.begin() + i);
+                throw;
+            }
         }
+        queue_.clear();
     }
 
 private:

lager/event_loop/safe_queue.hpp

@@ -47,6 +47,8 @@ struct safe_queue_event_loop
         LAGER_THROW(std::logic_error{"not implemented!"});
     }
 
+    // If there is an exception, the step() function needs to be re-run for the
+    // queue to be fully processed.
     void step()
     {
         assert(thread_id_ == std::this_thread::get_id());
@@ -72,9 +74,15 @@ struct safe_queue_event_loop
 
     void run_local_queue_()
     {
-        for (auto i = std::size_t{}; i < local_queue_.size(); ++i) {
-            auto fn = local_queue_[i];
-            fn();
+        for (auto i = std::size_t{}; i < local_queue_.size();) {
+            try {
+                auto fn = std::move(local_queue_[i++]);
+                std::move(fn)();
+            } catch (...) {
+                local_queue_.erase(local_queue_.begin(),
+                                   local_queue_.begin() + i);
+                throw;
+            }
         }
         local_queue_.clear();
     }

test/event_loop/manual.cpp

@@ -0,0 +1,52 @@
+//
+// lager - library for functional interactive c++ programs
+// Copyright (C) 2017 Juan Pedro Bolivar Puente
+//
+// This file is part of lager.
+//
+// lager is free software: you can redistribute it and/or modify
+// it under the terms of the MIT License, as detailed in the LICENSE
+// file located at the root of this source code distribution,
+// or here: <https://github.com/arximboldi/lager/blob/master/LICENSE>
+//
+
+#include <catch.hpp>
+
+#include <lager/event_loop/manual.hpp>
+#include <lager/store.hpp>
+
+TEST_CASE("exception safety")
+{
+    auto loop = lager::with_manual_event_loop{};
+
+    SECTION("basic")
+    {
+        CHECK_THROWS(loop.post([] { throw std::runtime_error{"noo!"}; }));
+
+        auto called = 0;
+        loop.post([&] { ++called; });
+        CHECK(called == 1);
+    }
+
+    SECTION("recursive")
+    {
+        auto called_a = 0;
+        auto called_b = 0;
+        auto called_c = 0;
+        CHECK_THROWS(loop.post([&] {
+            loop.post([&] { ++called_a; });
+            loop.post([&] { throw std::runtime_error{"noo!"}; });
+            loop.post([&] { ++called_b; });
+            CHECK(called_a == 0);
+            CHECK(called_b == 0);
+        }));
+
+        CHECK(called_a == 1);
+        CHECK(called_b == 0); // not called yet, queue was interrupted
+
+        loop.post([&] { ++called_c; });
+        CHECK(called_b == 1); // queue continued
+        CHECK(called_c == 1);
+        CHECK(called_a == 1);
+    }
+}

test/event_loop/queue.cpp

@@ -29,3 +29,18 @@ TEST_CASE("basic")
     queue.step();
     CHECK(store->value == 1);
 }
+
+TEST_CASE("exception")
+{
+    auto called = 0;
+    auto loop   = lager::queue_event_loop{};
+
+    loop.post([&] { throw std::runtime_error{"noo!"}; });
+    loop.post([&] { ++called; });
+
+    CHECK_THROWS(loop.step());
+    CHECK(called == 0);
+
+    loop.step();
+    CHECK(called == 1);
+}

test/event_loop/safe_queue.cpp

@@ -50,3 +50,18 @@ TEST_CASE("threads")
 
     CHECK(store->value == 200);
 }
+
+TEST_CASE("exception")
+{
+    auto called = 0;
+    auto loop   = lager::safe_queue_event_loop{};
+
+    loop.post([&] { throw std::runtime_error{"noo!"}; });
+    loop.post([&] { ++called; });
+
+    CHECK_THROWS(loop.step());
+    CHECK(called == 0);
+
+    loop.step();
+    CHECK(called == 1);
+}