diff --git a/doc/cn/eventqueue.md b/doc/cn/eventqueue.md index ff1a540..6f05d5a 100644 --- a/doc/cn/eventqueue.md +++ b/doc/cn/eventqueue.md @@ -134,6 +134,8 @@ bool process(); 注意：若 `process()` 被同时在多个线程中调用，事件队列中的事件也将只被处理一次。 +若监听器抛出异常，正在调度的事件会从队列中移除（它的监听器已经被调用，可能只调用了一部分，再次调度会再次调用这些监听器），尚未处理的事件按原来的顺序保留在队列中，异常传递给 `process()` 的调用者。 `processOne()` 、 `processIf()` 和 `processUntil()` 也是如此。若 `processIf()` 或 `processUntil()` 的 `predictor` 抛出异常，则不移除任何事件。 + #### processOne ```c++ diff --git a/doc/eventqueue.md b/doc/eventqueue.md index d2b1f86..5f6cd55 100644 --- a/doc/eventqueue.md +++ b/doc/eventqueue.md @@ -113,6 +113,7 @@ The listeners are called in the thread same as the caller of `process`. Any new events added to the queue during `process()` are not dispatched during current `process()`. `process()` is efficient in single thread event processing, it processes all events in the queue in current thread. To process events from multiple threads efficiently, use `processOne()`. Note: if `process()` is called from multiple threads simultaneously, the events in the event queue are guaranteed dispatched only once. +If a listener throws an exception, the event being dispatched is removed from the queue (its listeners were invoked, maybe partially, and dispatching it again would invoke them again), the events that were not processed yet are kept in the queue in their original order, and the exception propagates to the caller of `process()`. The same applies to `processOne()`, `processIf()` and `processUntil()`. If the `predictor` of `processIf()` or `processUntil()` throws, no event is removed. #### processOne diff --git a/doc/hetereventqueue.md b/doc/hetereventqueue.md index 50f231d..9b65fbb 100644 --- a/doc/hetereventqueue.md +++ b/doc/hetereventqueue.md @@ -84,6 +84,7 @@ The listeners are called in the thread same as the caller of `process`. Any new events added to the queue during `process()` are not dispatched during current `process()`. `process()` is efficient in single thread event processing, it processes all events in the queue in current thread. To process events from multiple threads efficiently, use `processOne()`. Note: if `process()` is called from multiple threads simultaneously, the events in the event queue are guaranteed dispatched only once. +If a listener throws an exception, the event being dispatched is removed from the queue (its listeners were invoked, maybe partially, and dispatching it again would invoke them again), the events that were not processed yet are kept in the queue in their original order, and the exception propagates to the caller of `process()`. The same applies to `processOne()` and `processIf()`. If the `func` of `processIf()` throws, no event is removed. #### processOne diff --git a/doc/introduction.md b/doc/introduction.md index a73be62..5f44d9f 100644 --- a/doc/introduction.md +++ b/doc/introduction.md @@ -82,4 +82,4 @@ Exceptions may be thrown by underlying code when, 2. The listeners (callbacks) throw exceptions during copying, moving, comparing, or invoking. Almost all operations guarantee strong exception safety, which means the underlying data remains original value on exception is thrown. -An except is `EventQueue::process`, on exception, the remaining events will not be dispatched, and the queue becomes empty. +If a listener throws while `EventQueue` is processing, the event being dispatched is removed from the queue and the events not processed yet are kept, see [EventQueue::process](eventqueue.md). diff --git a/include/eventpp/eventqueue.h b/include/eventpp/eventqueue.h index 507b013..20233e3 100644 --- a/include/eventpp/eventqueue.h +++ b/include/eventpp/eventqueue.h @@ -218,6 +218,7 @@ class EventQueueBase < { if(! queueList.empty()) { BufferedItemList tempList; + BufferedItemList idleList; // Use a counter to tell the queue list is not empty during processing // even though queueList is swapped to empty. @@ -229,12 +230,23 @@ class EventQueueBase < } if(! tempList.empty()) { - for(auto & item : tempList) { - doDispatchQueuedEvent( - item.get(), - typename MakeIndexSequence::Type() - ); - item.clear(); + // Don't move the processed items to idleList one by one here, the splice of a QueueList + // policy may be expensive (OrderedQueueList sorts on every splice). Only do it on exception. + for(auto it = tempList.begin(); it != tempList.end(); ++it) { + try { + doDispatchQueuedEvent( + it->get(), + typename MakeIndexSequence::Type() + ); + } + catch(...) { + while(tempList.begin() != it) { + idleList.splice(idleList.end(), tempList, tempList.begin()); + } + doRequeueOnException(tempList, idleList, it, true); + throw; + } + it->clear(); } std::lock_guard queueListLock(freeListMutex); @@ -251,6 +263,7 @@ class EventQueueBase < { if(! queueList.empty()) { BufferedItemList tempList; + BufferedItemList idleList; // Use a counter to tell the queue list is not empty during processing // even though queueList is swapped to empty. @@ -265,10 +278,16 @@ class EventQueueBase < if(! tempList.empty()) { auto & item = tempList.front(); - doDispatchQueuedEvent( - item.get(), - typename MakeIndexSequence::Type() - ); + try { + doDispatchQueuedEvent( + item.get(), + typename MakeIndexSequence::Type() + ); + } + catch(...) { + doRequeueOnException(tempList, idleList, tempList.begin(), true); + throw; + } item.clear(); std::lock_guard queueListLock(freeListMutex); @@ -299,15 +318,29 @@ class EventQueueBase < if(! tempList.empty()) { for(auto it = tempList.begin(); it != tempList.end(); ) { - if(doInvokeFuncWithQueuedEvent( + bool shouldProcess = false; + try { + shouldProcess = doInvokeFuncWithQueuedEvent( predictor, - it->get(), - typename MakeIndexSequence::Type()) - ) { - doDispatchQueuedEvent( it->get(), typename MakeIndexSequence::Type() ); + } + catch(...) { + doRequeueOnException(tempList, idleList, it, false); + throw; + } + if(shouldProcess) { + try { + doDispatchQueuedEvent( + it->get(), + typename MakeIndexSequence::Type() + ); + } + catch(...) { + doRequeueOnException(tempList, idleList, it, true); + throw; + } it->clear(); auto tempIt = it; @@ -354,18 +387,32 @@ class EventQueueBase < if(! tempList.empty()) { for(auto it = tempList.begin(); it != tempList.end(); ) { - if(doInvokeFuncWithQueuedEvent( + bool shouldStop = false; + try { + shouldStop = doInvokeFuncWithQueuedEvent( predictor, it->get(), - typename MakeIndexSequence::Type()) - ) { + typename MakeIndexSequence::Type() + ); + } + catch(...) { + doRequeueOnException(tempList, idleList, it, false); + throw; + } + if(shouldStop) { break; } else { - doDispatchQueuedEvent( - it->get(), - typename MakeIndexSequence::Type() - ); + try { + doDispatchQueuedEvent( + it->get(), + typename MakeIndexSequence::Type() + ); + } + catch(...) { + doRequeueOnException(tempList, idleList, it, true); + throw; + } it->clear(); auto tempIt = it; @@ -467,6 +514,33 @@ class EventQueueBase < return ! emptyQueue() && doCanNotifyQueueAvailable(); } + // Called when a listener or the predictor throws while processing the item at `it` in tempList. + // If `dispatched` is true, the listeners of the item were invoked (maybe partially), so the item is + // dropped, because processing it again would invoke the listeners again. Otherwise the item is kept. + // The remaining items in tempList are put back to the front of the queue to be processed by the next + // call, and the processed items in idleList are recycled. The caller must rethrow after calling this. + void doRequeueOnException(BufferedItemList & tempList, BufferedItemList & idleList, typename BufferedItemList::iterator it, const bool dispatched) + { + if(dispatched) { + it->clear(); + idleList.splice(idleList.end(), tempList, it); + } + + if(! tempList.empty()) { + std::lock_guard queueListLock(queueListMutex); + queueList.splice(queueList.begin(), tempList); + } + + if(! idleList.empty()) { + std::lock_guard queueListLock(freeListMutex); + freeList.splice(freeList.end(), idleList); + } + + if(doCanProcess()) { + queueListConditionVariable.notify_one(); + } + } + bool doCanNotifyQueueAvailable() const { return queueNotifyCounter.load(std::memory_order_acquire) == 0; diff --git a/include/eventpp/hetereventqueue.h b/include/eventpp/hetereventqueue.h index 0f8ee5d..36bcdd4 100644 --- a/include/eventpp/hetereventqueue.h +++ b/include/eventpp/hetereventqueue.h @@ -169,6 +169,7 @@ class HeterEventQueueBase : public HeterEventDispatcherBase< { if(! queueList.empty()) { BufferedItemList tempList; + BufferedItemList idleList; // Use a counter to tell the queue list is not empty during processing // even though queueList is swapped to empty. @@ -180,9 +181,19 @@ class HeterEventQueueBase : public HeterEventDispatcherBase< } if(! tempList.empty()) { - for(auto & item : tempList) { - doDispatchQueuedEvent(item.template get()); - item.clear(); + // Only move the processed items to idleList on exception, see EventQueue::process. + for(auto it = tempList.begin(); it != tempList.end(); ++it) { + try { + doDispatchQueuedEvent(it->template get()); + } + catch(...) { + while(tempList.begin() != it) { + idleList.splice(idleList.end(), tempList, tempList.begin()); + } + doRequeueOnException(tempList, idleList, it, true); + throw; + } + it->clear(); } std::lock_guard queueListLock(freeListMutex); @@ -199,6 +210,7 @@ class HeterEventQueueBase : public HeterEventDispatcherBase< { if(! queueList.empty()) { BufferedItemList tempList; + BufferedItemList idleList; // Use a counter to tell the queue list is not empty during processing // even though queueList is swapped to empty. @@ -213,7 +225,13 @@ class HeterEventQueueBase : public HeterEventDispatcherBase< if(! tempList.empty()) { auto & item = tempList.front(); - doDispatchQueuedEvent(item.template get()); + try { + doDispatchQueuedEvent(item.template get()); + } + catch(...) { + doRequeueOnException(tempList, idleList, tempList.begin(), true); + throw; + } item.clear(); std::lock_guard queueListLock(freeListMutex); @@ -262,6 +280,33 @@ class HeterEventQueueBase : public HeterEventDispatcherBase< return ! emptyQueue() && doCanNotifyQueueAvailable(); } + // Called when a listener or the predictor throws while processing the item at `it` in tempList. + // If `dispatched` is true, the listeners of the item were invoked (maybe partially), so the item is + // dropped, because processing it again would invoke the listeners again. Otherwise the item is kept. + // The remaining items in tempList are put back to the front of the queue to be processed by the next + // call, and the processed items in idleList are recycled. The caller must rethrow after calling this. + void doRequeueOnException(BufferedItemList & tempList, BufferedItemList & idleList, typename BufferedItemList::iterator it, const bool dispatched) + { + if(dispatched) { + it->clear(); + idleList.splice(idleList.end(), tempList, it); + } + + if(! tempList.empty()) { + std::lock_guard queueListLock(queueListMutex); + queueList.splice(queueList.begin(), tempList); + } + + if(! idleList.empty()) { + std::lock_guard queueListLock(freeListMutex); + freeList.splice(freeList.end(), idleList); + } + + if(doCanProcess()) { + queueListConditionVariable.notify_one(); + } + } + bool doCanNotifyQueueAvailable() const { return queueNotifyCounter.load(std::memory_order_acquire) == 0; @@ -313,12 +358,26 @@ class HeterEventQueueBase : public HeterEventDispatcherBase< ++it; continue; } - if(doInvokeFuncWithQueuedEvent( - func, - item, - typename MakeIndexSequence::value>::Type()) - ) { - doDispatchQueuedEvent(item); + bool shouldProcess = false; + try { + shouldProcess = doInvokeFuncWithQueuedEvent( + func, + item, + typename MakeIndexSequence::value>::Type() + ); + } + catch(...) { + doRequeueOnException(tempList, idleList, it, false); + throw; + } + if(shouldProcess) { + try { + doDispatchQueuedEvent(item); + } + catch(...) { + doRequeueOnException(tempList, idleList, it, true); + throw; + } it->clear(); auto tempIt = it; diff --git a/tests/unittest/CMakeLists.txt b/tests/unittest/CMakeLists.txt index e9f8598..7e46818 100644 --- a/tests/unittest/CMakeLists.txt +++ b/tests/unittest/CMakeLists.txt @@ -12,6 +12,7 @@ set(SRC_TEST test_queue_ctors.cpp test_queue_multithread.cpp test_queue_ordered_list.cpp + test_queue_exception.cpp test_hetercallbacklist_basic.cpp test_hetercallbacklist_ctors.cpp test_heterdispatcher_basic.cpp diff --git a/tests/unittest/test_queue_exception.cpp b/tests/unittest/test_queue_exception.cpp new file mode 100644 index 0000000..4c12ce0 --- /dev/null +++ b/tests/unittest/test_queue_exception.cpp @@ -0,0 +1,297 @@ +// eventpp library +// Copyright (C) 2018 Wang Qi (wqking) +// Github: https://github.com/wqking/eventpp +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "test.h" +#include "eventpp/eventqueue.h" +#include "eventpp/hetereventqueue.h" +#include "eventpp/utilities/orderedqueuelist.h" + +#include +#include + +// When a listener throws during processing, the event being dispatched is dropped (its listeners +// were invoked, maybe partially), the events not processed yet stay in the queue in their +// original order, and the exception propagates to the caller. + +namespace { + +struct ThrowOn +{ + explicit ThrowOn(const int value) : value(value) { + } + + void operator() (const int e) const { + if(e == value) { + throw std::runtime_error("listener"); + } + } + + int value; +}; + +struct OrderedListPolicies +{ + template + using QueueList = eventpp::OrderedQueueList; +}; + +} //unnamed namespace + +TEST_CASE("EventQueue, process, listener throws, remaining events are kept") +{ + eventpp::EventQueue queue; + + std::vector dataList; + queue.appendListener(0, [&dataList](const int e) { + dataList.push_back(e); + }); + queue.appendListener(0, ThrowOn(2)); + std::vector secondDataList; + queue.appendListener(0, [&secondDataList](const int e) { + secondDataList.push_back(e); + }); + + queue.enqueue(0, 1); + queue.enqueue(0, 2); + queue.enqueue(0, 3); + queue.enqueue(0, 4); + + REQUIRE_THROWS_AS(queue.process(), std::runtime_error); + + // 1 was fully processed, 2 reached the first listener and threw in the second, 3 and 4 were not processed + REQUIRE(dataList == std::vector{ 1, 2 }); + REQUIRE(secondDataList == std::vector{ 1 }); + REQUIRE(! queue.emptyQueue()); + + // events enqueued after the exception come after the kept events + queue.enqueue(0, 5); + + REQUIRE(queue.process()); + + // 2 is not processed again + REQUIRE(dataList == std::vector{ 1, 2, 3, 4, 5 }); + REQUIRE(secondDataList == std::vector{ 1, 3, 4, 5 }); + REQUIRE(queue.emptyQueue()); +} + +TEST_CASE("EventQueue, processOne, listener throws, the event is dropped") +{ + eventpp::EventQueue queue; + + std::vector dataList; + queue.appendListener(0, ThrowOn(2)); + queue.appendListener(0, [&dataList](const int e) { + dataList.push_back(e); + }); + + queue.enqueue(0, 1); + queue.enqueue(0, 2); + queue.enqueue(0, 3); + + REQUIRE(queue.processOne()); + REQUIRE_THROWS_AS(queue.processOne(), std::runtime_error); + REQUIRE(! queue.emptyQueue()); + REQUIRE(queue.processOne()); + REQUIRE(! queue.processOne()); + + REQUIRE(dataList == std::vector{ 1, 3 }); +} + +TEST_CASE("EventQueue, processIf, listener throws, remaining events are kept in order") +{ + eventpp::EventQueue queue; + + std::vector dataList; + queue.appendListener(0, ThrowOn(3)); + queue.appendListener(0, [&dataList](const int e) { + dataList.push_back(e); + }); + + // 1 and 2 are not selected by the predictor, 3 throws, 4 and 5 are not processed yet + for(int i = 1; i <= 5; ++i) { + queue.enqueue(0, i); + } + + REQUIRE_THROWS_AS(queue.processIf([](const int e) { return e >= 3; }), std::runtime_error); + REQUIRE(dataList.empty()); + REQUIRE(! queue.emptyQueue()); + + REQUIRE(queue.process()); + REQUIRE(dataList == std::vector{ 1, 2, 4, 5 }); + REQUIRE(queue.emptyQueue()); +} + +TEST_CASE("EventQueue, processIf, predictor throws, the event is kept") +{ + eventpp::EventQueue queue; + + std::vector dataList; + queue.appendListener(0, [&dataList](const int e) { + dataList.push_back(e); + }); + + queue.enqueue(0, 1); + queue.enqueue(0, 2); + queue.enqueue(0, 3); + + REQUIRE_THROWS_AS(queue.processIf([](const int e) -> bool { + if(e == 2) { + throw std::runtime_error("predictor"); + } + return true; + }), std::runtime_error); + + REQUIRE(dataList == std::vector{ 1 }); + + // nothing was dispatched for 2, so it's still in the queue + REQUIRE(queue.process()); + REQUIRE(dataList == std::vector{ 1, 2, 3 }); +} + +TEST_CASE("EventQueue, processUntil, listener throws, remaining events are kept") +{ + eventpp::EventQueue queue; + + std::vector dataList; + queue.appendListener(0, ThrowOn(2)); + queue.appendListener(0, [&dataList](const int e) { + dataList.push_back(e); + }); + + for(int i = 1; i <= 4; ++i) { + queue.enqueue(0, i); + } + + REQUIRE_THROWS_AS(queue.processUntil([](const int e) { return e == 4; }), std::runtime_error); + REQUIRE(dataList == std::vector{ 1 }); + + REQUIRE(queue.processUntil([](const int e) { return e == 4; })); + REQUIRE(dataList == std::vector{ 1, 3 }); + REQUIRE(! queue.emptyQueue()); + + REQUIRE(queue.process()); + REQUIRE(dataList == std::vector{ 1, 3, 4 }); +} + +TEST_CASE("EventQueue, process, listener throws, waiting thread is woken up for the kept events") +{ + eventpp::EventQueue queue; + + queue.appendListener(0, ThrowOn(1)); + queue.enqueue(0, 1); + queue.enqueue(0, 2); + + REQUIRE_THROWS_AS(queue.process(), std::runtime_error); + + // waitFor must return true immediately because event 2 is still in the queue + REQUIRE(queue.waitFor(std::chrono::milliseconds(0))); +} + +TEST_CASE("EventQueue, ordered list, process, listener throws, remaining events are kept in order") +{ + eventpp::EventQueue queue; + + std::vector dataList; + for(int e = 1; e <= 5; ++e) { + queue.appendListener(e, ThrowOn(3)); + queue.appendListener(e, [&dataList](const int e) { + dataList.push_back(e); + }); + } + + // enqueued out of order, the ordered list sorts them by event + queue.enqueue(5, 5); + queue.enqueue(3, 3); + queue.enqueue(1, 1); + queue.enqueue(4, 4); + queue.enqueue(2, 2); + + REQUIRE_THROWS_AS(queue.process(), std::runtime_error); + REQUIRE(dataList == std::vector{ 1, 2 }); + REQUIRE(! queue.emptyQueue()); + + REQUIRE(queue.process()); + REQUIRE(dataList == std::vector{ 1, 2, 4, 5 }); + REQUIRE(queue.emptyQueue()); +} + +TEST_CASE("HeterEventQueue, process, listener throws, remaining events are kept") +{ + eventpp::HeterEventQueue > queue; + + std::vector dataList; + queue.appendListener(0, ThrowOn(2)); + queue.appendListener(0, [&dataList](const int e) { + dataList.push_back(e); + }); + queue.appendListener(0, [&dataList](const int a, const int b) { + dataList.push_back(a * 10 + b); + }); + + queue.enqueue(0, 1); + queue.enqueue(0, 2); + queue.enqueue(0, 3, 4); + queue.enqueue(0, 5); + + REQUIRE_THROWS_AS(queue.process(), std::runtime_error); + REQUIRE(dataList == std::vector{ 1 }); + REQUIRE(! queue.emptyQueue()); + + REQUIRE(queue.process()); + REQUIRE(dataList == std::vector{ 1, 34, 5 }); + REQUIRE(queue.emptyQueue()); +} + +TEST_CASE("HeterEventQueue, processOne, listener throws, the event is dropped") +{ + eventpp::HeterEventQueue > queue; + + std::vector dataList; + queue.appendListener(0, ThrowOn(2)); + queue.appendListener(0, [&dataList](const int e) { + dataList.push_back(e); + }); + + queue.enqueue(0, 1); + queue.enqueue(0, 2); + queue.enqueue(0, 3); + + REQUIRE(queue.processOne()); + REQUIRE_THROWS_AS(queue.processOne(), std::runtime_error); + REQUIRE(queue.processOne()); + REQUIRE(! queue.processOne()); + + REQUIRE(dataList == std::vector{ 1, 3 }); +} + +TEST_CASE("HeterEventQueue, processIf, listener throws, remaining events are kept") +{ + eventpp::HeterEventQueue > queue; + + std::vector dataList; + queue.appendListener(0, ThrowOn(3)); + queue.appendListener(0, [&dataList](const int e) { + dataList.push_back(e); + }); + + for(int i = 1; i <= 5; ++i) { + queue.enqueue(0, i); + } + + REQUIRE_THROWS_AS(queue.processIf([](const int e) { return e >= 3; }), std::runtime_error); + REQUIRE(dataList.empty()); + + REQUIRE(queue.process()); + REQUIRE(dataList == std::vector{ 1, 2, 4, 5 }); +} +