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

#include "SystemFrameExchange.h"

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

SystemFrameExchange::SystemFrameExchange(const SystemFrameExchangeConfig& config)
{
	Configure(config);
}

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

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

	mCounters = SystemFrameExchangeMetrics();
	mCondition.notify_all();
}

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

bool SystemFrameExchange::AcquireForRender(SystemFrame& frame)
{
	std::lock_guard<std::mutex> lock(mMutex);
	if (!AcquireFreeLocked(frame))
	{
		frame = SystemFrame();
		++mCounters.acquireMisses;
		return false;
	}

	++mCounters.acquiredFrames;
	return true;
}

bool SystemFrameExchange::PublishCompleted(const SystemFrame& frame)
{
	std::lock_guard<std::mutex> lock(mMutex);
	if (!IsValidLocked(frame))
		return false;

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

	slot.state = SystemFrameSlotState::Completed;
	slot.frameIndex = frame.frameIndex;
	mCompletedIndices.push_back(frame.index);
	TrimCompletedLocked();
	++mCounters.completedFrames;
	mCondition.notify_all();
	return true;
}

bool SystemFrameExchange::ConsumeCompletedForSchedule(SystemFrame& frame)
{
	std::lock_guard<std::mutex> lock(mMutex);
	while (!mCompletedIndices.empty())
	{
		const std::size_t index = mCompletedIndices.front();
		mCompletedIndices.pop_front();
		if (index >= mSlots.size() || mSlots[index].state != SystemFrameSlotState::Completed)
			continue;

		mSlots[index].state = SystemFrameSlotState::Scheduled;
		FillFrameLocked(index, frame);
		++mCounters.scheduledFrames;
		return true;
	}

	frame = SystemFrame();
	++mCounters.completedPollMisses;
	return false;
}

bool SystemFrameExchange::ReleaseScheduledByBytes(void* bytes)
{
	if (bytes == nullptr)
		return false;

	std::lock_guard<std::mutex> lock(mMutex);
	for (std::size_t index = 0; index < mSlots.size(); ++index)
	{
		Slot& slot = mSlots[index];
		if (slot.bytes.empty() || slot.bytes.data() != bytes)
			continue;
		if (slot.state != SystemFrameSlotState::Scheduled)
			return false;

		slot.state = SystemFrameSlotState::Free;
		slot.frameIndex = 0;
		++slot.generation;
		mCondition.notify_all();
		return true;
	}

	return false;
}

bool SystemFrameExchange::WaitForCompletedDepth(std::size_t targetDepth, std::chrono::milliseconds timeout)
{
	std::unique_lock<std::mutex> lock(mMutex);
	return mCondition.wait_for(lock, timeout, [&]() {
		return CompletedCountLocked() >= targetDepth;
	});
}

bool SystemFrameExchange::WaitForStableCompletedDepth(
	std::size_t targetDepth,
	std::chrono::milliseconds stableDuration,
	std::chrono::milliseconds timeout)
{
	if (targetDepth == 0)
		return true;

	const auto deadline = std::chrono::steady_clock::now() + timeout;
	std::unique_lock<std::mutex> lock(mMutex);
	bool stableWindowStarted = false;
	std::chrono::steady_clock::time_point stableSince;

	while (true)
	{
		const auto now = std::chrono::steady_clock::now();
		if (now >= deadline)
			return false;

		if (CompletedCountLocked() >= targetDepth)
		{
			if (stableDuration <= std::chrono::milliseconds::zero())
				return true;

			if (!stableWindowStarted)
			{
				stableSince = now;
				stableWindowStarted = true;
			}

			const auto stableDeadline = stableSince + stableDuration;
			if (now >= stableDeadline)
				return true;

			mCondition.wait_until(lock, stableDeadline < deadline ? stableDeadline : deadline);
			continue;
		}

		stableWindowStarted = false;
		mCondition.wait_until(lock, deadline, [&]() {
			return CompletedCountLocked() >= targetDepth;
		});
	}
}

void SystemFrameExchange::Clear()
{
	std::lock_guard<std::mutex> lock(mMutex);
	mCompletedIndices.clear();
	for (Slot& slot : mSlots)
	{
		slot.state = SystemFrameSlotState::Free;
		slot.frameIndex = 0;
		++slot.generation;
	}
	mCondition.notify_all();
}

SystemFrameExchangeMetrics SystemFrameExchange::Metrics() const
{
	std::lock_guard<std::mutex> lock(mMutex);
	SystemFrameExchangeMetrics metrics = mCounters;
	metrics.capacity = mSlots.size();
	metrics.completedDepth = mCompletedIndices.size();

	for (const Slot& slot : mSlots)
	{
		switch (slot.state)
		{
		case SystemFrameSlotState::Free:
			++metrics.freeCount;
			break;
		case SystemFrameSlotState::Rendering:
			++metrics.renderingCount;
			break;
		case SystemFrameSlotState::Completed:
			++metrics.completedCount;
			break;
		case SystemFrameSlotState::Scheduled:
			++metrics.scheduledCount;
			break;
		}
	}

	return metrics;
}

bool SystemFrameExchange::AcquireFreeLocked(SystemFrame& frame)
{
	for (std::size_t index = 0; index < mSlots.size(); ++index)
	{
		Slot& slot = mSlots[index];
		if (slot.state != SystemFrameSlotState::Free)
			continue;

		slot.state = SystemFrameSlotState::Rendering;
		++slot.generation;
		FillFrameLocked(index, frame);
		return true;
	}

	return false;
}

bool SystemFrameExchange::DropOldestCompletedLocked()
{
	while (!mCompletedIndices.empty())
	{
		const std::size_t index = mCompletedIndices.front();
		mCompletedIndices.pop_front();
		if (index >= mSlots.size() || mSlots[index].state != SystemFrameSlotState::Completed)
			continue;

		Slot& slot = mSlots[index];
		slot.state = SystemFrameSlotState::Free;
		slot.frameIndex = 0;
		++slot.generation;
		++mCounters.completedDrops;
		mCondition.notify_all();
		return true;
	}

	return false;
}

void SystemFrameExchange::TrimCompletedLocked()
{
	if (mConfig.maxCompletedFrames == 0)
		return;
	while (CompletedCountLocked() > mConfig.maxCompletedFrames)
	{
		if (!DropOldestCompletedLocked())
			return;
	}
}

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

void SystemFrameExchange::FillFrameLocked(std::size_t index, SystemFrame& frame)
{
	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;
}

std::size_t SystemFrameExchange::CompletedCountLocked() const
{
	std::size_t count = 0;
	for (const Slot& slot : mSlots)
	{
		if (slot.state == SystemFrameSlotState::Completed)
			++count;
	}
	return count;
}

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