Improve contract of ProcessEvents

flow/ProcessEvents.cpp

@@ -34,29 +34,79 @@ struct EventImpl {
 	  : names(std::move(names)), callback(std::move(callback)) {
 		addEvent();
 	}
-	~EventImpl() { removeEvent(); }
 	Id id() const { return reinterpret_cast<intptr_t>(this); }
 	void addEvent();
 	void removeEvent();
 };
 
 struct ProcessEventsImpl {
-	std::unordered_map<StringRef, std::unordered_map<EventImpl::Id, EventImpl*>> events;
+	struct Triggering {
+		bool& value;
+		explicit Triggering(bool& value) : value(value) {
+			ASSERT(!value);
+			value = true;
+		}
+		~Triggering() { value = false; }
+	};
+	using EventMap = std::unordered_map<StringRef, std::unordered_map<EventImpl::Id, EventImpl*>>;
+	bool triggering = false;
+	EventMap events;
+	std::map<EventImpl::Id, std::vector<StringRef>> toRemove;
+	EventMap toInsert;
 
-	void trigger(StringRef name, StringRef msg, Error const& e) const {
+	void trigger(StringRef name, StringRef msg, Error const& e) {
+		Triggering _(triggering);
 		auto iter = events.find(name);
 		// strictly speaking this isn't a bug, but having callbacks that aren't caught
 		// by anyone could mean that something was misspelled. Therefore, the safe thing
 		// to do is to abort.
 		ASSERT(iter != events.end());
-		std::unordered_map<EventImpl::Id, EventImpl*> callbacks;
-		// These two iterations are necessary, since some callbacks might appear in multiple maps
-		for (auto const& c : iter->second) {
-			callbacks.insert(c);
-		}
+		std::unordered_map<EventImpl::Id, EventImpl*> callbacks = iter->second;
 		// after we collected all unique callbacks we can call each
 		for (auto const& c : callbacks) {
-			c.second->callback(name, msg, e);
+			try {
+				// it's possible that the callback has been removed in
+				// which case attempting to call it will be undefined
+				// behavior.
+				if (toRemove.count(c.first) > 0) {
+					c.second->callback(name, msg, e);
+				}
+			} catch (...) {
+				// callbacks are not allowed to throw
+				UNSTOPPABLE_ASSERT(false);
+			}
+		}
+		// merge modifications back into the event map
+		for (auto const& p : toRemove) {
+			for (auto const& n : p.second) {
+				events[n].erase(p.first);
+			}
+		}
+		toRemove.clear();
+		for (auto const& p : toInsert) {
+			events[p.first].insert(p.second.begin(), p.second.end());
+		}
+		toInsert.clear();
+	}
+
+	void add(StringRef const& name, EventImpl* event) {
+		EventMap& m = triggering ? toInsert : events;
+		m[name].emplace(event->id(), event);
+	}
+
+	void add(std::vector<StringRef> const& names, EventImpl* event) {
+		for (auto const& name : names) {
+			add(name, event);
+		}
+	}
+
+	void remove(std::vector<StringRef> names, EventImpl::Id id) {
+		if (triggering) {
+			toRemove.emplace(id, std::move(names));
+		} else {
+			for (auto const& name : names) {
+				events[name].erase(id);
+			}
 		}
 	}
 };
@@ -64,15 +114,11 @@ struct ProcessEventsImpl {
 ProcessEventsImpl impl;
 
 void EventImpl::addEvent() {
-	for (auto const& name : names) {
-		impl.events[name].emplace(this->id(), this);
-	}
+	impl.add(names, this);
 }
 
 void EventImpl::removeEvent() {
-	for (auto const& name : names) {
-		impl.events[name].erase(this->id());
-	}
+	impl.remove(names, this->id());
 }
 
 } // namespace
@@ -90,7 +136,9 @@ Event::Event(std::vector<StringRef> names, Callback callback) {
 	impl = new EventImpl(std::move(names), std::move(callback));
 }
 Event::~Event() {
-	delete reinterpret_cast<EventImpl*>(impl);
+	auto ptr = reinterpret_cast<EventImpl*>(impl);
+	ptr->removeEvent();
+	delete ptr;
 }
 
 } // namespace ProcessEvents

flow/include/flow/ProcessEvents.h

@@ -25,6 +25,8 @@
 
 namespace ProcessEvents {
 
+// A callback is never allowed to throw. Since std::function can't
+// take noexcept signatures, this is enforced at runtime
 using Callback = std::function<void(StringRef, StringRef, Error const&)>;
 
 class Event : NonCopyable {