Function commitMessages

fdbserver/LogRouter.actor.cpp:207–256

};

void commitMessages(LogRouterData* self, Version version, const std::vector<TagsAndMessage>& taggedMessages) {
	if (!taggedMessages.size()) {
		return;
	}

	int msgSize = 0;
	for (const auto& i : taggedMessages) {
		msgSize += i.message.size();
	}

	// Grab the last block in the blocks list so we can share its arena
	// We pop all of the elements of it to create a "fresh" vector that starts at the end of the previous vector
	Standalone<VectorRef<uint8_t>> block;
	if (self->messageBlocks.empty()) {
		block = Standalone<VectorRef<uint8_t>>();
		block.reserve(block.arena(), std::max<int64_t>(SERVER_KNOBS->TLOG_MESSAGE_BLOCK_BYTES, msgSize));
	} else {
		block = self->messageBlocks.back().second;
	}

	block.pop_front(block.size());

	for (const auto& msg : taggedMessages) {
		if (msg.message.size() > block.capacity() - block.size()) {
			self->messageBlocks.emplace_back(version, block);
			block = Standalone<VectorRef<uint8_t>>();
			block.reserve(block.arena(), std::max<int64_t>(SERVER_KNOBS->TLOG_MESSAGE_BLOCK_BYTES, msgSize));
		}

		block.append(block.arena(), msg.message.begin(), msg.message.size());
		for (const auto& tag : msg.tags) {
			auto tagData = self->getTagData(tag);
			if (!tagData) {
				tagData = self->createTagData(tag, 0, 0);
			}

			if (version >= tagData->popped) {
				tagData->version_messages.emplace_back(
				    version, LengthPrefixedStringRef((uint32_t*)(block.end() - msg.message.size())));
				if (tagData->version_messages.back().second.expectedSize() > SERVER_KNOBS->MAX_MESSAGE_SIZE) {
					TraceEvent(SevWarnAlways, "LargeMessage")
					    .detail("Size", tagData->version_messages.back().second.expectedSize());
				}
			}
		}

		msgSize -= msg.message.size();
	}
	self->messageBlocks.emplace_back(version, block);
}

ACTOR Future<Void> waitForVersion(LogRouterData* self, Version ver) {
	// The only time the log router should allow a gap in versions larger than MAX_READ_TRANSACTION_LIFE_VERSIONS is