video-shader-toys/apps/RenderCadenceCompositor/frames/InputFrameMailbox.cpp

#include "InputFrameMailbox.h"

#include <algorithm>
#include <chrono>
#include <cstring>

namespace
{
InputFrameMailboxConfig NormalizeConfig(InputFrameMailboxConfig config)
{
	if (config.rowBytes == 0)
		config.rowBytes = VideoIORowBytes(config.pixelFormat, config.width);
	return config;
}
}

InputFrameMailbox::InputFrameMailbox(const InputFrameMailboxConfig& config)
{
	Configure(config);
}

void InputFrameMailbox::Configure(const InputFrameMailboxConfig& config)
{
	std::lock_guard<std::mutex> lock(mMutex);
	mConfig = NormalizeConfig(config);
	mReadyIndices.clear();
	mSlots.clear();
	mSlots.resize(mConfig.capacity);

	const std::size_t byteCount = FrameByteCount();
	for (Slot& slot : mSlots)
	{
		slot.bytes.resize(byteCount);
		slot.state = InputFrameSlotState::Free;
		slot.frameIndex = 0;
		++slot.generation;
	}

	mCounters = InputFrameMailboxMetrics();
}

InputFrameMailboxConfig InputFrameMailbox::Config() const
{
	std::lock_guard<std::mutex> lock(mMutex);
	return mConfig;
}

bool InputFrameMailbox::SubmitFrame(const void* bytes, unsigned rowBytes, uint64_t frameIndex)
{
	if (bytes == nullptr || rowBytes == 0)
		return false;

	std::lock_guard<std::mutex> lock(mMutex);
	if (mSlots.empty() || mConfig.width == 0 || mConfig.height == 0)
		return false;

	std::size_t slotIndex = mSlots.size();
	for (std::size_t index = 0; index < mSlots.size(); ++index)
	{
		if (mSlots[index].state == InputFrameSlotState::Free)
		{
			slotIndex = index;
			break;
		}
	}

	if (slotIndex == mSlots.size())
	{
		if (!DropOldestReadyLocked())
		{
			++mCounters.submitMisses;
			return false;
		}

		for (std::size_t index = 0; index < mSlots.size(); ++index)
		{
			if (mSlots[index].state == InputFrameSlotState::Free)
			{
				slotIndex = index;
				break;
			}
		}
	}

	if (slotIndex == mSlots.size())
	{
		++mCounters.submitMisses;
		return false;
	}

	Slot& slot = mSlots[slotIndex];
	const std::size_t destinationRowBytes = mConfig.rowBytes;
	const std::size_t sourceRowBytes = static_cast<std::size_t>(rowBytes);
	const unsigned char* source = static_cast<const unsigned char*>(bytes);
	if (sourceRowBytes == destinationRowBytes)
	{
		std::memcpy(slot.bytes.data(), source, destinationRowBytes * static_cast<std::size_t>(mConfig.height));
	}
	else
	{
		const std::size_t copyRowBytes = (std::min)(sourceRowBytes, destinationRowBytes);
		for (unsigned y = 0; y < mConfig.height; ++y)
		{
			std::memcpy(
				slot.bytes.data() + static_cast<std::size_t>(y) * destinationRowBytes,
				source + static_cast<std::size_t>(y) * sourceRowBytes,
				copyRowBytes);
		}
	}

	slot.state = InputFrameSlotState::Ready;
	slot.frameIndex = frameIndex;
	++slot.generation;
	mReadyIndices.push_back(slotIndex);
	TrimReadyFramesLocked();
	++mCounters.submittedFrames;
	mCounters.latestFrameIndex = frameIndex;
	mCounters.hasSubmittedFrame = true;
	mLatestSubmitTime = std::chrono::steady_clock::now();
	return true;
}

bool InputFrameMailbox::TryAcquireLatest(InputFrame& frame)
{
	std::lock_guard<std::mutex> lock(mMutex);
	while (!mReadyIndices.empty())
	{
		const std::size_t index = mReadyIndices.back();
		mReadyIndices.pop_back();
		if (index >= mSlots.size() || mSlots[index].state != InputFrameSlotState::Ready)
			continue;

		while (!mReadyIndices.empty())
		{
			const std::size_t olderIndex = mReadyIndices.front();
			mReadyIndices.pop_front();
			if (olderIndex >= mSlots.size() || mSlots[olderIndex].state != InputFrameSlotState::Ready)
				continue;
			mSlots[olderIndex].state = InputFrameSlotState::Free;
			++mSlots[olderIndex].generation;
			++mCounters.droppedReadyFrames;
		}

		mSlots[index].state = InputFrameSlotState::Reading;
		FillFrameLocked(index, frame);
		++mCounters.consumedFrames;
		return true;
	}

	frame = InputFrame();
	++mCounters.consumeMisses;
	return false;
}

bool InputFrameMailbox::TryAcquireOldest(InputFrame& frame)
{
	std::lock_guard<std::mutex> lock(mMutex);
	while (!mReadyIndices.empty())
	{
		const std::size_t index = mReadyIndices.front();
		mReadyIndices.pop_front();
		if (index >= mSlots.size() || mSlots[index].state != InputFrameSlotState::Ready)
			continue;

		mSlots[index].state = InputFrameSlotState::Reading;
		FillFrameLocked(index, frame);
		++mCounters.consumedFrames;
		return true;
	}

	frame = InputFrame();
	++mCounters.consumeMisses;
	return false;
}

bool InputFrameMailbox::Release(const InputFrame& frame)
{
	std::lock_guard<std::mutex> lock(mMutex);
	if (!IsValidLocked(frame))
		return false;

	Slot& slot = mSlots[frame.index];
	if (slot.state != InputFrameSlotState::Reading)
		return false;

	slot.state = InputFrameSlotState::Free;
	slot.frameIndex = 0;
	++slot.generation;
	return true;
}

void InputFrameMailbox::Clear()
{
	std::lock_guard<std::mutex> lock(mMutex);
	mReadyIndices.clear();
	for (Slot& slot : mSlots)
	{
		slot.state = InputFrameSlotState::Free;
		slot.frameIndex = 0;
		++slot.generation;
	}
}

InputFrameMailboxMetrics InputFrameMailbox::Metrics() const
{
	std::lock_guard<std::mutex> lock(mMutex);
	InputFrameMailboxMetrics metrics = mCounters;
	metrics.capacity = mSlots.size();
	if (metrics.hasSubmittedFrame)
	{
		metrics.latestFrameAgeMilliseconds = std::chrono::duration_cast<std::chrono::duration<double, std::milli>>(
			std::chrono::steady_clock::now() - mLatestSubmitTime).count();
	}

	for (const Slot& slot : mSlots)
	{
		switch (slot.state)
		{
		case InputFrameSlotState::Free:
			++metrics.freeCount;
			break;
		case InputFrameSlotState::Ready:
			++metrics.readyCount;
			break;
		case InputFrameSlotState::Reading:
			++metrics.readingCount;
			break;
		}
	}

	return metrics;
}

bool InputFrameMailbox::IsValidLocked(const InputFrame& frame) const
{
	return frame.index < mSlots.size() && mSlots[frame.index].generation == frame.generation;
}

void InputFrameMailbox::FillFrameLocked(std::size_t index, InputFrame& frame) const
{
	const Slot& slot = mSlots[index];
	frame.bytes = slot.bytes.empty() ? nullptr : slot.bytes.data();
	frame.rowBytes = static_cast<long>(mConfig.rowBytes);
	frame.width = mConfig.width;
	frame.height = mConfig.height;
	frame.pixelFormat = mConfig.pixelFormat;
	frame.index = index;
	frame.generation = slot.generation;
	frame.frameIndex = slot.frameIndex;
}

bool InputFrameMailbox::DropOldestReadyLocked()
{
	while (!mReadyIndices.empty())
	{
		const std::size_t index = mReadyIndices.front();
		mReadyIndices.pop_front();
		if (index >= mSlots.size() || mSlots[index].state != InputFrameSlotState::Ready)
			continue;

		mSlots[index].state = InputFrameSlotState::Free;
		mSlots[index].frameIndex = 0;
		++mSlots[index].generation;
		++mCounters.droppedReadyFrames;
		return true;
	}

	return false;
}

void InputFrameMailbox::TrimReadyFramesLocked()
{
	if (mConfig.maxReadyFrames == 0)
		return;
	while (mReadyIndices.size() > mConfig.maxReadyFrames)
		DropOldestReadyLocked();
}

std::size_t InputFrameMailbox::FrameByteCount() const
{
	return static_cast<std::size_t>(mConfig.rowBytes) * static_cast<std::size_t>(mConfig.height);
}