RedisConnection: simplify query prioritization logic

As opposed to the previous version which used a complex data structure
to correctly manage the query priorities, this version uses a
`std::deque` for the write queue and a simple mechanism to insert
high-priority items at the front. By default, items are processed in FIFO
order, but if someone wants to immediately send a high-priority query it
will be placed at the front of the queue (remember std::deque allows
efficient insertion at both ends), an will overtake any normal priority
items already queued. However, if there are already high-priority items
in the queue, the new high-priority item will be inserted after them but
still before any normal priority items, ensuring that all high-priority
items are processed in the order they were enqueued.

Author: yhabteab

lib/icingadb/icingadb-objects.cpp

@@ -46,8 +46,6 @@
 
 using namespace icinga;
 
-using Prio = RedisConnection::QueryPriority;
-
 std::map<Type::Ptr, RedisKey> IcingaDB::m_CustomVarKeys;
 std::unordered_set<Type*> IcingaDB::m_IndexedTypes;
 
@@ -197,7 +195,9 @@ void IcingaDB::ConfigStaticInitialize()
 
 void IcingaDB::UpdateAllConfigObjects()
 {
-	m_Rcon->FireAndForgetQuery({"XADD", "icinga:schema", "MAXLEN", "1", "*", "version", "6"}, Prio::Heartbeat);
+	// This function performs an initial dump of all configuration objects into Redis, thus there are no
+	// previously enqueued queries on m_Rcon that we need to wait for. So, no Sync() call is necessary here.
+	m_Rcon->FireAndForgetQuery({"XADD", "icinga:schema", "MAXLEN", "1", "*", "version", "6"}, {}, true);
 
 	Log(LogInformation, "IcingaDB") << "Starting initial config/status dump";
 	double startTime = Utility::GetTime();
@@ -215,7 +215,7 @@ void IcingaDB::UpdateAllConfigObjects()
 	std::vector<Type::Ptr> types = GetTypes();
 
 	// Add a new type=* state=wip entry to the stream and remove all previous entries (MAXLEN 1).
-	m_Rcon->FireAndForgetQuery({"XADD", "icinga:dump", "MAXLEN", "1", "*", "key", "*", "state", "wip"}, Prio::Config);
+	m_Rcon->FireAndForgetQuery({"XADD", "icinga:dump", "MAXLEN", "1", "*", "key", "*", "state", "wip"});
 
 	const std::vector<String> globalKeys = {
 		m_PrefixConfigObject + "customvar",
@@ -232,8 +232,8 @@ void IcingaDB::UpdateAllConfigObjects()
 		m_PrefixConfigObject + "redundancygroup",
 		m_PrefixConfigObject + "redundancygroup:state",
 	};
-	DeleteKeys(m_Rcon, globalKeys, Prio::Config);
-	DeleteKeys(m_Rcon, {"icinga:nextupdate:host", "icinga:nextupdate:service"}, Prio::Config);
+	DeleteKeys(m_Rcon, globalKeys);
+	DeleteKeys(m_Rcon, {"icinga:nextupdate:host", "icinga:nextupdate:service"});
 	m_Rcon->Sync();
 
 	Defer resetDumpedGlobals ([this]() {
@@ -253,7 +253,7 @@ void IcingaDB::UpdateAllConfigObjects()
 		auto& rcon (m_Rcons.at(ctype));
 
 		std::vector<String> keys = GetTypeOverwriteKeys(lcType);
-		DeleteKeys(rcon, keys, Prio::Config);
+		DeleteKeys(rcon, keys);
 
 		WorkQueue upqObjectType(25000, Configuration::Concurrency, LogNotice);
 		upqObjectType.SetName("IcingaDB:ConfigDump:" + lcType);
@@ -265,9 +265,7 @@ void IcingaDB::UpdateAllConfigObjects()
 			String cursor = "0";
 
 			do {
-				Array::Ptr res = rcon->GetResultOfQuery({
-					"HSCAN", configCheckSum, cursor, "COUNT", "1000"
-				}, Prio::Config);
+				Array::Ptr res = rcon->GetResultOfQuery({"HSCAN", configCheckSum, cursor, "COUNT", "1000"});
 
 				AddKvsToMap(res->Get(1), redisCheckSums);
 
@@ -410,7 +408,7 @@ void IcingaDB::UpdateAllConfigObjects()
 			setChecksum.clear();
 			setObject.clear();
 
-			rcon->FireAndForgetQueries(std::move(transaction), Prio::Config, {affectedConfig});
+			rcon->FireAndForgetQueries(std::move(transaction), {affectedConfig});
 		});
 
 		auto flushDels ([&]() {
@@ -429,7 +427,7 @@ void IcingaDB::UpdateAllConfigObjects()
 			delChecksum.clear();
 			delObject.clear();
 
-			rcon->FireAndForgetQueries(std::move(transaction), Prio::Config, {affectedConfig});
+			rcon->FireAndForgetQueries(std::move(transaction), {affectedConfig});
 		});
 
 		auto setOne ([&]() {
@@ -490,7 +488,7 @@ void IcingaDB::UpdateAllConfigObjects()
 		}
 
 		for (auto& key : GetTypeDumpSignalKeys(type)) {
-			rcon->FireAndForgetQuery({"XADD", "icinga:dump", "*", "key", key, "state", "done"}, Prio::Config);
+			rcon->FireAndForgetQuery({"XADD", "icinga:dump", "*", "key", key, "state", "done"});
 		}
 		rcon->Sync();
 		rcon->Disconnect(); // We're done with this connection, so close it.
@@ -513,14 +511,14 @@ void IcingaDB::UpdateAllConfigObjects()
 	}
 
 	for (auto& key : globalKeys) {
-		m_Rcon->FireAndForgetQuery({"XADD", "icinga:dump", "*", "key", key, "state", "done"}, Prio::Config);
+		m_Rcon->FireAndForgetQuery({"XADD", "icinga:dump", "*", "key", key, "state", "done"});
 	}
 
-	m_Rcon->FireAndForgetQuery({"XADD", "icinga:dump", "*", "key", "*", "state", "done"}, Prio::Config);
+	m_Rcon->FireAndForgetQuery({"XADD", "icinga:dump", "*", "key", "*", "state", "done"});
 
 	// enqueue a callback that will notify us once all previous queries were executed and wait for this event
 	std::promise<void> p;
-	m_Rcon->EnqueueCallback([&p](boost::asio::yield_context& yc) { p.set_value(); }, Prio::Config);
+	m_Rcon->EnqueueCallback([&p](boost::asio::yield_context&) { p.set_value(); });
 	p.get_future().wait();
 
 	auto endTime (Utility::GetTime());
@@ -553,13 +551,13 @@ std::vector<std::vector<intrusive_ptr<ConfigObject>>> IcingaDB::ChunkObjects(std
 	return chunks;
 }
 
-void IcingaDB::DeleteKeys(const RedisConnection::Ptr& conn, const std::vector<String>& keys, RedisConnection::QueryPriority priority) {
+void IcingaDB::DeleteKeys(const RedisConnection::Ptr& conn, const std::vector<String>& keys) {
 	std::vector<String> query = {"DEL"};
 	for (auto& key : keys) {
 		query.emplace_back(key);
 	}
 
-	conn->FireAndForgetQuery(std::move(query), priority);
+	conn->FireAndForgetQuery(std::move(query));
 }
 
 std::vector<String> IcingaDB::GetTypeOverwriteKeys(const String& type)
@@ -1318,7 +1316,7 @@ void IcingaDB::UpdateState(const Checkable::Ptr& checkable, int mode)
 		m_Rcon->FireAndForgetQueries({
 			{"HSET", redisStateKey, objectKey, JsonEncode(stateAttrs)},
 			{"HSET", redisChecksumKey, objectKey, JsonEncode(new Dictionary({{"checksum", checksum}}))},
-		}, Prio::RuntimeStateSync);
+		});
 	}
 
 	if (mode & RuntimeState) {
@@ -1336,7 +1334,7 @@ void IcingaDB::UpdateState(const Checkable::Ptr& checkable, int mode)
 			streamadd.emplace_back(IcingaToStreamValue(kv.second));
 		}
 
-		m_Rcon->FireAndForgetQuery(std::move(streamadd), Prio::RuntimeStateStream, {0, 1});
+		m_Rcon->FireAndForgetQuery(std::move(streamadd), {0, 1});
 	}
 }
 
@@ -1410,8 +1408,8 @@ void IcingaDB::UpdateDependenciesState(const Checkable::Ptr& checkable, const De
 			queries.emplace_back(std::move(query));
 		}
 
-		m_Rcon->FireAndForgetQueries(std::move(queries), Prio::RuntimeStateSync);
-		m_Rcon->FireAndForgetQueries(std::move(streamStates), Prio::RuntimeStateStream, {0, 1});
+		m_Rcon->FireAndForgetQueries(std::move(queries));
+		m_Rcon->FireAndForgetQueries(std::move(streamStates), {0, 1});
 	}
 }
 
@@ -2416,17 +2414,15 @@ void IcingaDB::SendNextUpdate(const Checkable::Ptr& checkable)
 				dynamic_pointer_cast<Service>(checkable) ? "icinga:nextupdate:service" : "icinga:nextupdate:host",
 				Convert::ToString(checkable->GetNextUpdate()),
 				GetObjectIdentifier(checkable)
-			},
-			Prio::CheckResult
+			}
 		);
 	} else if (!checkable->GetEnableActiveChecks() || checkable->GetExtension("ConfigObjectDeleted")) {
 		m_Rcon->FireAndForgetQuery(
 			{
 				"ZREM",
 				dynamic_pointer_cast<Service>(checkable) ? "icinga:nextupdate:service" : "icinga:nextupdate:host",
 				GetObjectIdentifier(checkable)
-			},
-			Prio::CheckResult
+			}
 		);
 	}
 }
@@ -2592,7 +2588,7 @@ void IcingaDB::ForwardHistoryEntries()
 
 			if (m_HistoryCon && m_HistoryCon->IsConnected()) {
 				try {
-					m_HistoryCon->GetResultsOfQueries(haystack, Prio::History, {0, 0, haystack.size()});
+					m_HistoryCon->GetResultsOfQueries(haystack, {0, 0, haystack.size()});
 					break;
 				} catch (const std::exception& ex) {
 					logFailure(ex.what());
@@ -3225,7 +3221,7 @@ void IcingaDB::DeleteRelationship(const String& id, const String& redisKeyWithou
 		"redis_key", redisKey, "id", id, "runtime_type", "delete"
 	});
 
-	m_Rcon->FireAndForgetQueries(queries, Prio::Config);
+	m_Rcon->FireAndForgetQueries(queries);
 }
 
 void IcingaDB::DeleteRelationship(const String& id, RedisKey redisKey, bool hasChecksum)
@@ -3352,7 +3348,7 @@ void IcingaDB::DeleteState(const String& id, RedisKey redisKey, bool hasChecksum
 	}
 	hdels.emplace_back(RedisConnection::Query{"HDEL", m_PrefixConfigObject + redisKeyWithoutPrefix, id});
 
-	m_Rcon->FireAndForgetQueries(std::move(hdels), Prio::RuntimeStateSync);
+	m_Rcon->FireAndForgetQueries(std::move(hdels));
 	// TODO: This is currently purposefully commented out due to how Icinga DB (Go) handles runtime state
 	//       upsert and delete events. See https://github.com/Icinga/icingadb/pull/894 for more details.
 	/*m_Rcon->FireAndForgetQueries({{
@@ -3439,11 +3435,11 @@ void IcingaDB::ExecuteRedisTransaction(const RedisConnection::Ptr& rcon, std::ma
 	if (transaction.size() > 1) {
 		transaction.emplace_back(RedisConnection::Query{"EXEC"});
 		if (!runtimeUpdates.empty()) {
-			rcon->FireAndForgetQueries(std::move(transaction), Prio::Config, {1});
+			rcon->FireAndForgetQueries(std::move(transaction), {1});
 		} else {
 			// This is likely triggered by the initial Redis config dump, so a) we don't need to record the number of
 			// affected objects and b) we don't really know how many objects are going to be affected by this tx.
-			rcon->FireAndForgetQueries(std::move(transaction), Prio::Config);
+			rcon->FireAndForgetQueries(std::move(transaction));
 		}
 	}
 }

lib/icingadb/icingadb-worker.cpp

@@ -216,8 +216,7 @@ void IcingaDB::ProcessPendingItem(const PendingConfigItem& citem)
 					"runtime_type",
 					"delete"
 				}
-			},
-			RedisConnection::QueryPriority::Config
+			}
 		);
 	}
 

lib/icingadb/icingadb.cpp

@@ -22,8 +22,6 @@ using namespace icinga;
 
 #define MAX_EVENTS_DEFAULT 5000
 
-using Prio = RedisConnection::QueryPriority;
-
 String IcingaDB::m_EnvironmentId;
 std::mutex IcingaDB::m_EnvironmentIdInitMutex;
 
@@ -184,7 +182,7 @@ void IcingaDB::PublishStats()
 		}
 	}
 
-	m_HistoryCon->FireAndForgetQuery(std::move(query), Prio::Heartbeat);
+	m_HistoryCon->FireAndForgetQuery(std::move(query), {}, true /* high priority */);
 }
 
 void IcingaDB::Stop(bool runtimeRemoved)

lib/icingadb/icingadb.hpp

@@ -275,7 +275,7 @@ class IcingaDB : public ObjectImpl<IcingaDB>
 	/* config & status dump */
 	void UpdateAllConfigObjects();
 	std::vector<std::vector<intrusive_ptr<ConfigObject>>> ChunkObjects(std::vector<intrusive_ptr<ConfigObject>> objects, size_t chunkSize);
-	void DeleteKeys(const RedisConnection::Ptr& conn, const std::vector<String>& keys, RedisConnection::QueryPriority priority);
+	void DeleteKeys(const RedisConnection::Ptr& conn, const std::vector<String>& keys);
 	std::vector<String> GetTypeOverwriteKeys(const String& type);
 	std::vector<String> GetTypeDumpSignalKeys(const Type::Ptr& type);
 	void InsertCheckableDependencies(const Checkable::Ptr& checkable, std::map<String, RedisConnection::Query>& hMSets,

lib/icingadb/icingadbchecktask.cpp

@@ -126,7 +126,8 @@ void IcingadbCheckTask::ScriptFunc(const Checkable::Ptr& checkable, const CheckR
 					"0-0", "0-0", "0-0", "0-0", "0-0", "0-0",
 				}
 			},
-			RedisConnection::QueryPriority::Heartbeat
+			{},
+			true /* high priority */
 		));
 
 		redisTime = std::move(replies.at(0));