Optimize resolver collection performance in CommitProxyServer

Author: Dan Lambright

fdbserver/CommitProxyServer.actor.cpp

@@ -154,18 +154,24 @@ struct ResolutionRequestBuilder {
 		for (int idx = 0; idx < trIn.read_conflict_ranges.size(); ++idx) {
 			const auto& r = trIn.read_conflict_ranges[idx];
 			auto ranges = self->keyResolvers.intersectingRanges(r);
-			std::set<int> resolvers;
+			std::vector<int> resolvers;
+			resolvers.reserve(self->resolvers.size());
 			for (auto& ir : ranges) {
 				auto& version_resolver = ir.value();
 				for (int i = version_resolver.size() - 1; i >= 0; i--) {
-					resolvers.insert(version_resolver[i].second);
+					int resolver_id = version_resolver[i].second;
+					if (std::find(resolvers.begin(), resolvers.end(), resolver_id) == resolvers.end()) {
+						resolvers.push_back(resolver_id);
+					}
 					if (version_resolver[i].first < trIn.read_snapshot)
 						break;
 				}
 			}
 			if (SERVER_KNOBS->PROXY_USE_RESOLVER_PRIVATE_MUTATIONS && systemKeys.intersects(r)) {
 				for (int k = 0; k < self->resolvers.size(); k++) {
-					resolvers.insert(k);
+					if (std::find(resolvers.begin(), resolvers.end(), k) == resolvers.end()) {
+						resolvers.push_back(k);
+					}
 				}
 			}
 			ASSERT(resolvers.size());
@@ -181,16 +187,22 @@ struct ResolutionRequestBuilder {
 	void addWriteConflictRanges(CommitTransactionRef& trIn) {
 		for (auto& r : trIn.write_conflict_ranges) {
 			auto ranges = self->keyResolvers.intersectingRanges(r);
-			std::set<int> resolvers;
+			std::vector<int> resolvers;
+			resolvers.reserve(20); // Optimized for small resolver counts
 			for (auto& ir : ranges) {
 				auto& version_resolver = ir.value();
 				if (!version_resolver.empty()) {
-					resolvers.insert(version_resolver.back().second);
+					int resolver_id = version_resolver.back().second;
+					if (std::find(resolvers.begin(), resolvers.end(), resolver_id) == resolvers.end()) {
+						resolvers.push_back(resolver_id);
+					}
 				}
 			}
 			if (SERVER_KNOBS->PROXY_USE_RESOLVER_PRIVATE_MUTATIONS && systemKeys.intersects(r)) {
 				for (int k = 0; k < self->resolvers.size(); k++) {
-					resolvers.insert(k);
+					if (std::find(resolvers.begin(), resolvers.end(), k) == resolvers.end()) {
+						resolvers.push_back(k);
+					}
 				}
 			}
 			ASSERT(resolvers.size());
@@ -1619,11 +1631,25 @@ void applyMetadataEffect(CommitBatchContext* self) {
 		     transactionIndex < self->resolution[0].stateMutations[versionIndex].size() && !self->forceRecovery;
 		     transactionIndex++) {
 			bool committed = true;
-			for (int resolver = 0; resolver < self->resolution.size(); resolver++) {
-				committed =
-				    committed && self->resolution[resolver].stateMutations[versionIndex][transactionIndex].committed;
+			auto& res = self->resolution;
+			auto isCommitted = [&](int i) { return res[i].stateMutations[versionIndex][transactionIndex].committed; };
+			// Unroll the loop for the common number of resolvers
+			switch (res.size()) {
+			case 2:
+				committed = isCommitted(0) && isCommitted(1);
+				break;
+			case 4:
+				committed = isCommitted(0) && isCommitted(1) && isCommitted(2) && isCommitted(3);
+				break;
+			case 8:
+				committed = isCommitted(0) && isCommitted(1) && isCommitted(2) && isCommitted(3) && isCommitted(4) &&
+				            isCommitted(5) && isCommitted(6) && isCommitted(7);
+				break;
+			default:
+				for (int resolver = 0; resolver < res.size(); resolver++) {
+					committed = committed && isCommitted(resolver);
+				}
 			}
-
 			if (committed && self->pProxyCommitData->getTenantMode() == TenantMode::REQUIRED) {
 				auto& tenantIds = self->resolution[0].stateMutations[versionIndex][transactionIndex].tenantIds;
 				ASSERT(tenantIds.present());
@@ -1672,7 +1698,8 @@ void applyMetadataEffect(CommitBatchContext* self) {
 			}
 		}
 
-		// These changes to txnStateStore will be committed by the other proxy, so we simply discard the commit message
+		// These changes to txnStateStore will be committed by the other proxy, so we simply discard the commit
+		// message
 		auto fcm = self->pProxyCommitData->logAdapter->getCommitMessage();
 		self->storeCommits.emplace_back(fcm, self->pProxyCommitData->txnStateStore->commit());
 
@@ -1735,8 +1762,8 @@ void determineCommittedTransactions(CommitBatchContext* self) {
 	}
 }
 
-// This first pass through committed transactions deals with "metadata" effects (modifications of txnStateStore, changes
-// to storage servers' responsibilities)
+// This first pass through committed transactions deals with "metadata" effects (modifications of txnStateStore,
+// changes to storage servers' responsibilities)
 ACTOR Future<Void> applyMetadataToCommittedTransactions(CommitBatchContext* self) {
 	state ProxyCommitData* const pProxyCommitData = self->pProxyCommitData;
 	state std::unordered_set<int64_t> rawAccessTenantIds;
@@ -2025,7 +2052,8 @@ void addAccumulativeChecksumMutations(CommitBatchContext* self) {
 		self->toCommit.writeTypedMessage(acsMutation);
 	}
 }
-// RangeLock takes effect only when the feature flag is on and database is unlocked and the mutation is not encrypted
+// RangeLock takes effect only when the feature flag is on and database is unlocked and the mutation is not
+// encrypted
 void rejectMutationsForReadLockOnRange(CommitBatchContext* self) {
 	ASSERT(self->rangeLockEnabled());
 	ProxyCommitData* const pProxyCommitData = self->pProxyCommitData;
@@ -2328,6 +2356,7 @@ ACTOR Future<Void> postResolution(CommitBatchContext* self) {
 	self->isMyFirstBatch = !pProxyCommitData->version.get();
 	self->previousCoordinators = pProxyCommitData->txnStateStore->readValue(coordinatorsKey).get();
 
+	// debug mode only?
 	assertResolutionStateMutationsSizeConsistent(self->resolution);
 
 	applyMetadataEffect(self);
@@ -2694,7 +2723,8 @@ ACTOR Future<Void> reply(CommitBatchContext* self) {
 
 	if (SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {
 		// We have received a reply from the sequencer, so all versions prior to the current version have been
-		// made durable and we can consider the current transaction to be committed - advance min commit version now.
+		// made durable and we can consider the current transaction to be committed - advance min commit version
+		// now.
 		if (self->prevVersion && self->commitVersion - self->prevVersion < SERVER_KNOBS->MAX_VERSIONS_IN_FLIGHT / 2) {
 			//TraceEvent("CPAdvanceMinVersion", self->pProxyCommitData->dbgid).detail("PrvVersion", self->prevVersion).detail("CommitVersion", self->commitVersion).detail("Master", self->pProxyCommitData->master.id().toString()).detail("TxSize", self->trs.size());
 			debug_advanceMinCommittedVersion(UID(), self->commitVersion);
@@ -2845,15 +2875,16 @@ ACTOR Future<Void> commitBatchImpl(CommitBatchContext* pContext) {
 	pContext->stage = INITIALIZE;
 	getCurrentLineage()->modify(&TransactionLineage::operation) = TransactionLineage::Operation::Commit;
 
-	// Active load balancing runs at a very high priority (to obtain accurate estimate of memory used by commit batches)
-	// so we need to downgrade here
+	// Active load balancing runs at a very high priority (to obtain accurate estimate of memory used by commit
+	// batches) so we need to downgrade here
 	wait(delay(0, TaskPriority::ProxyCommit));
 
 	pContext->pProxyCommitData->lastVersionTime = pContext->startTime;
 	++pContext->pProxyCommitData->stats.commitBatchIn;
 	pContext->setupTraceBatch();
 
-	/////// Phase 1: Pre-resolution processing (CPU bound except waiting for a version # which is separately pipelined
+	/////// Phase 1: Pre-resolution processing (CPU bound except waiting for a version # which is separately
+	/// pipelined
 	/// and *should* be available by now (unless empty commit); ordered; currently atomic but could yield)
 	pContext->stage = PRE_RESOLUTION;
 	wait(CommitBatch::preresolutionProcessing(pContext));
@@ -2866,12 +2897,14 @@ ACTOR Future<Void> commitBatchImpl(CommitBatchContext* pContext) {
 	pContext->stage = RESOLUTION;
 	wait(CommitBatch::getResolution(pContext));
 
-	////// Phase 3: Post-resolution processing (CPU bound except for very rare situations; ordered; currently atomic but
+	////// Phase 3: Post-resolution processing (CPU bound except for very rare situations; ordered; currently atomic
+	/// but
 	/// doesn't need to be)
 	pContext->stage = POST_RESOLUTION;
 	wait(CommitBatch::postResolution(pContext));
 
-	/////// Phase 4: Logging (network bound; pipelined up to MAX_READ_TRANSACTION_LIFE_VERSIONS (limited by loop above))
+	/////// Phase 4: Logging (network bound; pipelined up to MAX_READ_TRANSACTION_LIFE_VERSIONS (limited by loop
+	/// above))
 	pContext->stage = TRANSACTION_LOGGING;
 	wait(CommitBatch::transactionLogging(pContext));