media: add core AVRecorder

Generic interface to audio and video encoders.

src/media/CMakeLists.txt

@@ -1,5 +1,9 @@
 target_sources(SRB2SDL2 PRIVATE
 	audio_encoder.hpp
+	avrecorder.cpp
+	avrecorder.hpp
+	avrecorder_impl.hpp
+	avrecorder_queue.cpp
 	cfile.cpp
 	cfile.hpp
 	container.hpp

src/media/avrecorder.cpp

@@ -0,0 +1,211 @@
+// RING RACERS
+//-----------------------------------------------------------------------------
+// Copyright (C) 2023 by James Robert Roman
+//
+// This program is free software distributed under the
+// terms of the GNU General Public License, version 2.
+// See the 'LICENSE' file for more details.
+//-----------------------------------------------------------------------------
+
+#include <algorithm>
+#include <chrono>
+#include <exception>
+#include <iterator>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <utility>
+
+#include "../cxxutil.hpp"
+#include "../i_time.h"
+#include "../m_fixed.h"
+#include "avrecorder_impl.hpp"
+#include "webm_container.hpp"
+
+using namespace srb2::media;
+
+using Impl = AVRecorder::Impl;
+
+namespace
+{
+
+constexpr auto kBufferMethod = VideoFrame::BufferMethod::kEncoderAllocatedRGBA8888;
+
+}; // namespace
+
+Impl::Impl(Config cfg) :
+	max_size_(cfg.max_size),
+	max_duration_(cfg.max_duration),
+
+	container_(std::make_unique<WebmContainer>(MediaContainer::Config {
+		.file_name = cfg.file_name,
+	})),
+
+	audio_encoder_(make_audio_encoder(cfg)),
+	video_encoder_(make_video_encoder(cfg)),
+
+	epoch_(I_GetTime()),
+
+	thread_([this] { worker(); })
+{
+}
+
+std::unique_ptr<AudioEncoder> Impl::make_audio_encoder(const Config cfg) const
+{
+	if (!cfg.audio)
+	{
+		return nullptr;
+	}
+
+	const Config::Audio& a = *cfg.audio;
+
+	return container_->make_audio_encoder({
+		.channels = 2,
+		.sample_rate = a.sample_rate,
+	});
+}
+
+std::unique_ptr<VideoEncoder> Impl::make_video_encoder(const Config cfg) const
+{
+	if (!cfg.video)
+	{
+		return nullptr;
+	}
+
+	const Config::Video& v = *cfg.video;
+
+	return container_->make_video_encoder({
+		.width = v.width,
+		.height = v.height,
+		.frame_rate = v.frame_rate,
+		.buffer_method = kBufferMethod,
+	});
+}
+
+Impl::~Impl()
+{
+	valid_ = false;
+	wake_up_worker();
+	thread_.join();
+
+	try
+	{
+		// Finally flush encoders, unless queues were finished
+		// already due to time or size constraints.
+
+		if (!audio_queue_.finished())
+		{
+			audio_encoder_->flush();
+		}
+
+		if (!video_queue_.finished())
+		{
+			video_encoder_->flush();
+		}
+	}
+	catch (const std::exception& ex)
+	{
+		CONS_Alert(CONS_ERROR, "AVRecorder::Impl::~Impl: %s\n", ex.what());
+		return;
+	}
+}
+
+std::optional<int> Impl::advance_video_pts()
+{
+	auto _ = queue_guard();
+
+	// Don't let this queue grow out of hand. It's normal
+	// for encoding time to vary by a small margin and
+	// spend longer than one frame rate on a single
+	// frame. It should normalize though.
+
+	if (video_queue_.vec_.size() >= 3)
+	{
+		return {};
+	}
+
+	SRB2_ASSERT(video_encoder_ != nullptr);
+
+	const float tic_pts = video_encoder_->frame_rate() / static_cast<float>(TICRATE);
+	const int pts = ((I_GetTime() - epoch_) + FixedToFloat(g_time.timefrac)) * tic_pts;
+
+	if (!video_queue_.advance(pts, 1))
+	{
+		return {};
+	}
+
+	return pts;
+}
+
+void Impl::worker()
+{
+	for (;;)
+	{
+		QueueState qs;
+
+		try
+		{
+			while ((qs = encode_queues()) == QueueState::kFlushed)
+				;
+		}
+		catch (const std::exception& ex)
+		{
+			CONS_Alert(CONS_ERROR, "AVRecorder::Impl::worker: %s\n", ex.what());
+			break;
+		}
+
+		if (qs != QueueState::kFinished && valid_)
+		{
+			std::unique_lock lock(queue_mutex_);
+
+			queue_cond_.wait(lock);
+		}
+		else
+		{
+			break;
+		}
+	}
+
+	// Breaking out of the loop ensures invalidation!
+	valid_ = false;
+}
+
+AVRecorder::AVRecorder(const Config config) : impl_(std::make_unique<Impl>(config))
+{
+}
+
+AVRecorder::~AVRecorder()
+{
+	// impl_ is destroyed in a background thread so it doesn't
+	// block the thread AVRecorder was destroyed in.
+	//
+	// TODO: Save into a thread pool instead of detaching so
+	//       the thread could be joined at program exit and
+	//       not possibly terminate before fully destroyed?
+
+	std::thread([_ = std::move(impl_)] {}).detach();
+}
+
+void AVRecorder::push_audio_samples(audio_buffer_t buffer)
+{
+	const auto _ = impl_->queue_guard();
+
+	auto& q = impl_->audio_queue_;
+
+	if (!q.advance(q.pts(), buffer.size()))
+	{
+		return;
+	}
+
+	using T = const float;
+	tcb::span<T> p(reinterpret_cast<T*>(buffer.data()), buffer.size() * 2); // 2 channels
+
+	std::copy(p.begin(), p.end(), std::back_inserter(q.vec_));
+
+	impl_->wake_up_worker();
+}
+
+bool AVRecorder::invalid() const
+{
+	return !impl_->valid_;
+}

src/media/avrecorder.hpp

@@ -0,0 +1,76 @@
+// RING RACERS
+//-----------------------------------------------------------------------------
+// Copyright (C) 2023 by James Robert Roman
+//
+// This program is free software distributed under the
+// terms of the GNU General Public License, version 2.
+// See the 'LICENSE' file for more details.
+//-----------------------------------------------------------------------------
+
+#ifndef __SRB2_MEDIA_AVRECORDER_HPP__
+#define __SRB2_MEDIA_AVRECORDER_HPP__
+
+#include <array>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include <tcb/span.hpp>
+
+#include "../audio/sample.hpp"
+
+namespace srb2::media
+{
+
+class AVRecorder
+{
+public:
+	using audio_sample_t = srb2::audio::Sample<2>;
+	using audio_buffer_t = tcb::span<const audio_sample_t>;
+
+	class Impl;
+
+	struct Config
+	{
+		struct Audio
+		{
+			int sample_rate;
+		};
+
+		struct Video
+		{
+			int width;
+			int height;
+			int frame_rate;
+		};
+
+		std::string file_name;
+
+		std::optional<std::size_t> max_size; // file size limit
+		std::optional<std::chrono::duration<float>> max_duration;
+
+		std::optional<Audio> audio;
+		std::optional<Video> video;
+	};
+
+	AVRecorder(Config config);
+	~AVRecorder();
+
+	void push_audio_samples(audio_buffer_t buffer);
+
+	// True if this instance has terminated. Continuing to use
+	// this interface is useless and the object should be
+	// destructed immediately.
+	bool invalid() const;
+
+private:
+	std::unique_ptr<Impl> impl_;
+};
+
+}; // namespace srb2::media
+
+#endif // __SRB2_MEDIA_AVRECORDER_HPP__

src/media/avrecorder_impl.hpp

@@ -0,0 +1,164 @@
+// RING RACERS
+//-----------------------------------------------------------------------------
+// Copyright (C) 2023 by James Robert Roman
+//
+// This program is free software distributed under the
+// terms of the GNU General Public License, version 2.
+// See the 'LICENSE' file for more details.
+//-----------------------------------------------------------------------------
+
+#ifndef __SRB2_MEDIA_AVRECORDER_IMPL_HPP__
+#define __SRB2_MEDIA_AVRECORDER_IMPL_HPP__
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstddef>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <thread>
+#include <vector>
+
+#include "../i_time.h"
+#include "avrecorder.hpp"
+#include "container.hpp"
+
+namespace srb2::media
+{
+
+class AVRecorder::Impl
+{
+public:
+	template <typename T>
+	class Queue
+	{
+	public:
+		// The use of typename = void is a GCC bug.
+		// https://gcc.gnu.org/bugzilla/show_bug.cgi?id=85282
+		// Explicit specialization inside of a class still
+		// does not work as of 12.2.1.
+
+		template <typename, typename = void>
+		struct Traits
+		{
+		};
+
+		template <typename _>
+		struct Traits<AudioEncoder, _>
+		{
+			using frame_type = float;
+		};
+
+		template <typename _>
+		struct Traits<VideoEncoder, _>
+		{
+			using frame_type = VideoFrame::instance_t;
+		};
+
+		std::vector<typename Traits<T>::frame_type> vec_;
+
+		// This number only decrements once a frame has
+		// actually been written to container.
+		std::size_t queued_frames_ = 0;
+
+		Queue(const std::unique_ptr<T>& encoder, Impl& impl) : encoder_(encoder.get()), impl_(&impl) {}
+
+		// This method handles validation of the queue,
+		// finishing the queue and advancing PTS. Returns true
+		// if PTS was advanced.
+		bool advance(int pts, int duration);
+
+		// True if no more data may be queued.
+		bool finished() const { return finished_; }
+
+		// Presentation Time Stamp; one frame for video
+		// encoders, one sample for audio encoders.
+		int pts() const { return pts_; }
+
+	private:
+		using time_unit_t = std::chrono::duration<float>;
+
+		T* const encoder_;
+		Impl* const impl_;
+
+		bool finished_ = (encoder_ == nullptr);
+		int pts_ = -1; // valid pts starts at 0
+
+		// Actual duration of PTS unit.
+		time_unit_t time_scale() const;
+	};
+
+	const std::optional<std::size_t> max_size_;
+	std::optional<std::chrono::duration<float>> max_duration_;
+
+	// max_duration_ may be readjusted in case a queue
+	// finishes early for any reason. max_duration_config_ is
+	// the original, unmodified value.
+	const decltype(max_duration_) max_duration_config_ = max_duration_;
+
+	std::unique_ptr<MediaContainer> container_;
+	std::unique_ptr<AudioEncoder> audio_encoder_;
+	std::unique_ptr<VideoEncoder> video_encoder_;
+
+	Queue<AudioEncoder> audio_queue_ {audio_encoder_, *this};
+	Queue<VideoEncoder> video_queue_ {video_encoder_, *this};
+
+	// This class becomes invalid if:
+	//
+	// 1) an exception occurred
+	// 2) the object has begun destructing
+	std::atomic<bool> valid_ = true;
+
+	Impl(Config config);
+	~Impl();
+
+	// Returns valid PTS if enough time has passed.
+	std::optional<int> advance_video_pts();
+
+	// Use before accessing audio_queue_ or video_queue_.
+	auto queue_guard() { return std::lock_guard(queue_mutex_); }
+
+	// Use to notify worker thread if queues were modified.
+	void wake_up_worker() { queue_cond_.notify_one(); }
+
+private:
+	enum class QueueState
+	{
+		kEmpty,	   // all queues are empty
+		kFlushed,  // a queue was flushed but more data may be waiting
+		kFinished, // all queues are finished -- no more data may be queued
+	};
+
+	const tic_t epoch_;
+
+	std::thread thread_;
+	mutable std::recursive_mutex queue_mutex_; // guards audio and video queues
+	std::condition_variable_any queue_cond_;
+
+	std::unique_ptr<AudioEncoder> make_audio_encoder(const Config cfg) const;
+	std::unique_ptr<VideoEncoder> make_video_encoder(const Config cfg) const;
+
+	QueueState encode_queues();
+
+	void worker();
+};
+
+template <>
+inline AVRecorder::Impl::Queue<AudioEncoder>::time_unit_t AVRecorder::Impl::Queue<AudioEncoder>::time_scale() const
+{
+	return time_unit_t(1.f / encoder_->sample_rate());
+}
+
+template <>
+inline AVRecorder::Impl::Queue<VideoEncoder>::time_unit_t AVRecorder::Impl::Queue<VideoEncoder>::time_scale() const
+{
+	return time_unit_t(1.f / encoder_->frame_rate());
+}
+
+extern template class AVRecorder::Impl::Queue<AudioEncoder>;
+extern template class AVRecorder::Impl::Queue<VideoEncoder>;
+
+}; // namespace srb2::media
+
+#endif // __SRB2_MEDIA_AVRECORDER_IMPL_HPP__

src/media/avrecorder_queue.cpp

@@ -0,0 +1,142 @@
+// RING RACERS
+//-----------------------------------------------------------------------------
+// Copyright (C) 2023 by James Robert Roman
+//
+// This program is free software distributed under the
+// terms of the GNU General Public License, version 2.
+// See the 'LICENSE' file for more details.
+//-----------------------------------------------------------------------------
+
+#include <chrono>
+#include <cstddef>
+#include <mutex>
+#include <utility>
+
+#include "avrecorder_impl.hpp"
+
+using namespace srb2::media;
+
+using Impl = AVRecorder::Impl;
+
+template <typename T>
+bool Impl::Queue<T>::advance(int new_pts, int duration)
+{
+	if (!impl_->valid_ || finished())
+	{
+		return false;
+	}
+
+	new_pts += duration;
+
+	// PTS must only advance.
+	if (new_pts <= pts())
+	{
+		return false;
+	}
+
+	auto finish = [this]
+	{
+		finished_ = true;
+
+		const auto t = impl_->container_->duration();
+
+		// Tracks are ultimately cut to the shortest among
+		// them, therefore it would be pointless for another
+		// queue to continue beyond this point.
+		//
+		// This is relevant if finishing due to size
+		// constraint; in that case, another queue might be
+		// far behind this one in terms of size and would
+		// continue in vain.
+
+		if (!impl_->max_duration_ || t < impl_->max_duration_)
+		{
+			impl_->max_duration_ = t;
+		}
+
+		impl_->wake_up_worker();
+	};
+
+	if (impl_->max_duration_)
+	{
+		const int final_pts = *impl_->max_duration_ / time_scale();
+
+		if (new_pts > final_pts)
+		{
+			return finish(), false;
+		}
+	}
+
+	if (impl_->max_size_)
+	{
+		const MediaEncoder::BitRate est = encoder_->estimated_bit_rate();
+
+		const float br = est.bits / 8.f;
+
+		// count size of already queued frames too
+		const float t = ((duration + queued_frames_) * time_scale()) / est.period;
+
+		if ((impl_->container_->size() + (t * br)) > *impl_->max_size_)
+		{
+			return finish(), false;
+		}
+	}
+
+	pts_ = new_pts;
+	queued_frames_ += duration;
+
+	return true;
+}
+
+Impl::QueueState Impl::encode_queues()
+{
+	bool remain = false;
+	bool flushed = false;
+
+	auto check = [&, this](auto& q, auto encode)
+	{
+		std::unique_lock lock(queue_mutex_);
+
+		if (!q.finished())
+		{
+			remain = true;
+		}
+
+		if (!q.vec_.empty())
+		{
+			const std::size_t n = q.queued_frames_;
+
+			auto copy = std::move(q.vec_);
+
+			lock.unlock();
+			encode(std::move(copy));
+			lock.lock();
+
+			q.queued_frames_ -= n;
+
+			flushed = true;
+		}
+	};
+
+	auto encode_audio = [this](auto copy) { audio_encoder_->encode(copy); };
+	auto encode_video = [this](auto copy) {};
+
+	check(audio_queue_, encode_audio);
+	check(video_queue_, encode_video);
+
+	if (flushed)
+	{
+		return QueueState::kFlushed;
+	}
+	else if (remain)
+	{
+		return QueueState::kEmpty;
+	}
+	else
+	{
+		return QueueState::kFinished;
+	}
+}
+
+template class Impl::Queue<AudioEncoder>;
+template class Impl::Queue<VideoEncoder>;