Fixed EventQueue and HeterEventQueue losing the queued events when a listener throws

doc/cn/eventqueue.md

@@ -134,6 +134,8 @@ bool process();
 
 注意：若 `process()` 被同时在多个线程中调用，事件队列中的事件也将只被处理一次。
 
+若监听器抛出异常，正在调度的事件会从队列中移除（它的监听器已经被调用，可能只调用了一部分，再次调度会再次调用这些监听器），尚未处理的事件按原来的顺序保留在队列中，异常传递给 `process()` 的调用者。 `processOne()` 、 `processIf()` 和 `processUntil()` 也是如此。若 `processIf()` 或 `processUntil()` 的 `predictor` 抛出异常，则不移除任何事件。
+
 #### processOne
 
 ```c++

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
 

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
 

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).

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<sizeof...(Args)>::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<sizeof...(Args)>::Type()
+						);
+					}
+					catch(...) {
+						while(tempList.begin() != it) {
+							idleList.splice(idleList.end(), tempList, tempList.begin());
+						}
+						doRequeueOnException(tempList, idleList, it, true);
+						throw;
+					}
+					it->clear();
 				}
 
 				std::lock_guard<Mutex> 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<sizeof...(Args)>::Type()
-				);
+				try {
+					doDispatchQueuedEvent(
+						item.get(),
+						typename MakeIndexSequence<sizeof...(Args)>::Type()
+					);
+				}
+				catch(...) {
+					doRequeueOnException(tempList, idleList, tempList.begin(), true);
+					throw;
+				}
 				item.clear();
 
 				std::lock_guard<Mutex> 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<sizeof...(Args)>::Type())
-						) {
-						doDispatchQueuedEvent(
 							it->get(),
 							typename MakeIndexSequence<sizeof...(Args)>::Type()
 						);
+					}
+					catch(...) {
+						doRequeueOnException(tempList, idleList, it, false);
+						throw;
+					}
+					if(shouldProcess) {
+						try {
+							doDispatchQueuedEvent(
+								it->get(),
+								typename MakeIndexSequence<sizeof...(Args)>::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<sizeof...(Args)>::Type())
-						) {
+							typename MakeIndexSequence<sizeof...(Args)>::Type()
+						);
+					}
+					catch(...) {
+						doRequeueOnException(tempList, idleList, it, false);
+						throw;
+					}
+					if(shouldStop) {
 						break;
 					}
 					else {
-						doDispatchQueuedEvent(
-							it->get(),
-							typename MakeIndexSequence<sizeof...(Args)>::Type()
-						);
+						try {
+							doDispatchQueuedEvent(
+								it->get(),
+								typename MakeIndexSequence<sizeof...(Args)>::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<Mutex> queueListLock(queueListMutex);
+			queueList.splice(queueList.begin(), tempList);
+		}
+
+		if(! idleList.empty()) {
+			std::lock_guard<Mutex> queueListLock(freeListMutex);
+			freeList.splice(freeList.end(), idleList);
+		}
+
+		if(doCanProcess()) {
+			queueListConditionVariable.notify_one();
+		}
+	}
+
 	bool doCanNotifyQueueAvailable() const
 	{
 		return queueNotifyCounter.load(std::memory_order_acquire) == 0;

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<QueuedItemBase>());
-					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<QueuedItemBase>());
+					}
+					catch(...) {
+						while(tempList.begin() != it) {
+							idleList.splice(idleList.end(), tempList, tempList.begin());
+						}
+						doRequeueOnException(tempList, idleList, it, true);
+						throw;
+					}
+					it->clear();
 				}
 
 				std::lock_guard<Mutex> 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<QueuedItemBase>());
+				try {
+					doDispatchQueuedEvent(item.template get<QueuedItemBase>());
+				}
+				catch(...) {
+					doRequeueOnException(tempList, idleList, tempList.begin(), true);
+					throw;
+				}
 				item.clear();
 
 				std::lock_guard<Mutex> 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<Mutex> queueListLock(queueListMutex);
+			queueList.splice(queueList.begin(), tempList);
+		}
+
+		if(! idleList.empty()) {
+			std::lock_guard<Mutex> 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<std::tuple_size<ArgsTuple>::value>::Type())
-					) {
-					doDispatchQueuedEvent(item);
+				bool shouldProcess = false;
+				try {
+					shouldProcess = doInvokeFuncWithQueuedEvent(
+						func,
+						item,
+						typename MakeIndexSequence<std::tuple_size<ArgsTuple>::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;

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