audio: stop AudioOut2 contexts from stalling each other (PPSA09809) (#1745)

audio: submit AudioOut2 pushes outside the state mutex so contexts stop stalling each other

Co-authored-by: nmzik <Nmzik@mail.ru>

src/libs/libAudio2.cpp

@@ -178,7 +178,6 @@ struct AudioOut2ContextState {
 	uint32_t               num_grains          = 512;
 	uint64_t               last_update         = 0;
 	int                    object_audio_handle = 0;
-	std::vector<float>     object_mix;
 };
 
 struct AudioOut2PortStateEntry {
@@ -212,8 +211,18 @@ struct AudioOut2LatencyState {
 	uint32_t latency_us = 0;
 };
 
+// Per-context copy of what a push hands the backend. The backend can sleep while pacing, so a push
+// holds only its context's mutex across it; closing the context's handles waits on that mutex.
+struct AudioOut2Submission {
+	Common::Mutex                                                   mutex;
+	std::array<std::vector<uint8_t>, AudioInternal::OUT_PORTS_MAX>  pcm;
+	std::array<std::array<float, 16>, AudioInternal::OUT_PORTS_MAX> gains;
+	std::vector<float>                                              object_mix;
+};
+
 static Common::Mutex                              g_audioout2_state_mutex;
 static std::array<AudioOut2ContextState, 16>      g_audioout2_contexts;
+static std::array<AudioOut2Submission, 16>        g_audioout2_submissions;
 static std::array<AudioOut2PortStateEntry, 256>   g_audioout2_ports;
 static std::vector<AudioOut2UserHandle>          g_audioout2_users;
 static Common::Mutex                              g_audioout2_speaker_array_mutex;
@@ -249,6 +258,10 @@ static AudioOut2ContextState* audioout2_find_context_locked(AudioOut2ContextHand
 	return nullptr;
 }
 
+static AudioOut2Submission& audioout2_submission(const AudioOut2ContextState& context) {
+	return g_audioout2_submissions[&context - g_audioout2_contexts.data()];
+}
+
 static uint32_t audioout2_grain_micros(uint32_t grains) {
 	const auto sample_count = (grains == 0 ? 512u : grains);
 	return std::max<uint32_t>((sample_count * 1000000u) / 48000u, 1000u);
@@ -372,11 +385,10 @@ static bool audioout2_context_has_queueable_device(const AudioOut2ContextState&
 	return false;
 }
 
-static void audioout2_mix_object(AudioOut2ContextState&         context,
+static void audioout2_mix_object(std::vector<float>& mix, uint32_t num_grains,
                                  const AudioOut2PortStateEntry& object) {
-	auto& mix = context.object_mix;
 	if (mix.empty()) {
-		mix.resize(static_cast<size_t>(context.num_grains) * 2, 0.0f);
+		mix.resize(static_cast<size_t>(num_grains) * 2, 0.0f);
 	}
 
 	// Center ordinary objects. First-order stereo ambisonics uses left/right-facing
@@ -393,7 +405,7 @@ static void audioout2_mix_object(AudioOut2ContextState&         context,
 	}
 	left *= object.gains[0];
 	right *= object.gains[0];
-	for (uint32_t frame = 0; frame < context.num_grains; frame++) {
+	for (uint32_t frame = 0; frame < num_grains; frame++) {
 		const float sample =
 		    object.audio_format == AudioInternal::Format::FloatMono
 		        ? reinterpret_cast<const float*>(object.pcm_data.data())[frame]
@@ -403,15 +415,16 @@ static void audioout2_mix_object(AudioOut2ContextState&         context,
 	}
 }
 
-static void audioout2_queue_context_audio(AudioOut2ContextHandle ctx, bool blocking) {
+static void audioout2_queue_context_audio(AudioOut2ContextHandle ctx, AudioOut2Submission& out,
-	// The backend consumes the buffers synchronously, but can block while pacing SDL's queue.
+                                          bool blocking) {
-	// Keep both PCM storage and port handles alive until it returns.
+	Common::LockGuard submit_lock(out.mutex);
-	Common::LockGuard lock(g_audioout2_state_mutex);
+	g_audioout2_state_mutex.Lock();
-	auto*             context = audioout2_find_context_locked(ctx);
+	auto* context = audioout2_find_context_locked(ctx);
 	if (context == nullptr) {
+		g_audioout2_state_mutex.Unlock();
 		return;
 	}
-	context->object_mix.clear();
+	out.object_mix.clear();
 	std::vector<AudioInternal::OutputParam> params;
 	params.reserve(AudioInternal::OUT_PORTS_MAX);
 	for (const auto& state: g_audioout2_ports) {
@@ -419,14 +432,18 @@ static void audioout2_queue_context_audio(AudioOut2ContextHandle ctx, bool block
 			continue;
 		}
 		if (audioout2_port_type_is_object(state.port_type)) {
-			audioout2_mix_object(*context, state);
+			audioout2_mix_object(out.object_mix, context->num_grains, state);
 		} else if (state.audio_handle > 0 && params.size() < AudioInternal::OUT_PORTS_MAX) {
-			params.push_back({state.audio_handle, state.pcm_data.data(), state.gains.data()});
+			const auto i = params.size();
+			out.pcm[i]   = state.pcm_data;
+			out.gains[i] = state.gains;
+			params.push_back({state.audio_handle, out.pcm[i].data(), out.gains[i].data()});
 		}
 	}
-	if (!context->object_mix.empty() && params.size() < AudioInternal::OUT_PORTS_MAX) {
+	if (!out.object_mix.empty() && params.size() < AudioInternal::OUT_PORTS_MAX) {
-		params.push_back({context->object_audio_handle, context->object_mix.data()});
+		params.push_back({context->object_audio_handle, out.object_mix.data()});
 	}
+	g_audioout2_state_mutex.Unlock();
 	if (!params.empty()) {
 		(void)AudioInternal::AudioOutOutputs(params.data(), static_cast<uint32_t>(params.size()),
 		                                     blocking);
@@ -528,7 +545,8 @@ int KYTY_SYSV_ABI AudioOut2ContextDestroy(AudioOut2ContextHandle ctx) {
 	size_t audio_handles_num = 0;
 
 	g_audioout2_state_mutex.Lock();
-	if (auto* state = audioout2_find_context_locked(ctx); state != nullptr) {
+	auto* state = audioout2_find_context_locked(ctx);
+	if (state != nullptr) {
 		if (state->object_audio_handle > 0) {
 			audio_handles[audio_handles_num++] = state->object_audio_handle;
 		}
@@ -544,6 +562,10 @@ int KYTY_SYSV_ABI AudioOut2ContextDestroy(AudioOut2ContextHandle ctx) {
 	}
 	g_audioout2_state_mutex.Unlock();
 
+	if (state != nullptr) {
+		// Wait for an in-flight push that may still use the handles.
+		Common::LockGuard wait(audioout2_submission(*state).mutex);
+	}
 	for (size_t i = 0; i < audio_handles_num; i++) {
 		audioout2_close_audio_handle(audio_handles[i]);
 	}
@@ -595,8 +617,9 @@ int KYTY_SYSV_ABI AudioOut2ContextPush(AudioOut2ContextHandle ctx, uint32_t bloc
 				if (state->queued < state->queue_depth) {
 					state->queued++;
 				}
+				auto& submission = audioout2_submission(*state);
 				g_audioout2_state_mutex.Unlock();
-				audioout2_queue_context_audio(ctx, blocking != 0);
+				audioout2_queue_context_audio(ctx, submission, blocking != 0);
 				return OK;
 			}
 		}
@@ -729,14 +752,20 @@ int KYTY_SYSV_ABI AudioOut2PortDestroy(AudioOut2PortHandle port) {
 	PRINT_NAME();
 	LOGF("\t port = 0x%016" PRIx64 "\n", port);
 
-	int audio_handle = 0;
+	int                          audio_handle = 0;
+	const AudioOut2ContextState* context      = nullptr;
 	g_audioout2_state_mutex.Lock();
 	if (auto* state = audioout2_find_port_locked(port); state != nullptr) {
 		audio_handle = state->audio_handle;
+		context      = audioout2_find_context_locked(state->context);
 		*state       = AudioOut2PortStateEntry {};
 	}
 	g_audioout2_state_mutex.Unlock();
 
+	if (audio_handle > 0 && context != nullptr) {
+		// Wait for an in-flight push that may still use the handle.
+		Common::LockGuard wait(audioout2_submission(*context).mutex);
+	}
 	audioout2_close_audio_handle(audio_handle);
 
 	return OK;

tests/AudioOut2PortTests.cpp

@@ -5,6 +5,7 @@
 #include <algorithm>
 #include <array>
 #include <atomic>
+#include <chrono>
 #include <cmath>
 #include <condition_variable>
 #include <cstdint>
@@ -32,6 +33,8 @@ size_t                            g_capture_bytes = 0;
 int                               g_next_device  = 1;
 int                               g_open_waiters = 0;
 bool                              g_block_opens  = false;
+bool                              g_block_outputs = false;
+std::vector<int>                  g_blocked_output;
 
 void Check(bool value, const char* text) {
 	if (!value) {
@@ -190,6 +193,22 @@ void ReleaseDeviceOpens() {
 	g_device_cv.notify_all();
 }
 
+void BlockOutputs() {
+	std::lock_guard lock(g_device_mutex);
+	g_block_outputs = true;
+}
+
+void WaitForBlockedOutput() {
+	std::unique_lock lock(g_device_mutex);
+	g_device_cv.wait(lock, []() { return !g_blocked_output.empty(); });
+}
+
+void ReleaseOutputs() {
+	std::lock_guard lock(g_device_mutex);
+	g_block_outputs = false;
+	g_device_cv.notify_all();
+}
+
 int LiveDeviceCount() {
 	std::lock_guard lock(g_device_mutex);
 	return static_cast<int>(g_live_devices.size());
@@ -605,6 +624,46 @@ void TestObjectBusRouting() {
 	Check(LiveDeviceCount() == 0, "object bus device leaked");
 }
 
+void TestBlockedPushOnlyHoldsItsContext() {
+	const auto blocked      = CreateContext(4, 1);
+	const auto other        = CreateContext();
+	const auto blocked_port = CreatePort(blocked, MakeParam());
+	const auto other_port   = CreatePort(other, MakeParam());
+	auto       object_param = MakeParam(0x100);
+	object_param.port_type  = 0x100;
+	std::vector<float> pcm(512 * 2, 0.25f);
+	SetPcm(blocked_port, pcm.data());
+	SetPcm(CreatePort(blocked, object_param), pcm.data());
+	CaptureOutputPcm(pcm.size() * sizeof(float));
+
+	BlockOutputs();
+	std::thread pusher([&]() { Check(AudioOut2::AudioOut2ContextPush(blocked, 1) == OK, "blocked push failed"); });
+	WaitForBlockedOutput();
+	std::fill(pcm.begin(), pcm.end(), 0.5f);
+	SetPcm(blocked_port, pcm.data());
+	SetAttribute(blocked_port, 1, std::array {0.5f, 0.5f});
+	SetPcm(other_port, pcm.data());
+	Check(AudioOut2::AudioOut2ContextPush(other, 1) == OK, "push stalled behind another context");
+	// Give each destroyer time to close a device of the blocked output; the backend fails if it does.
+	std::thread port_destroyer([&]() { AudioOut2::AudioOut2PortDestroy(blocked_port); });
+	std::this_thread::sleep_for(std::chrono::milliseconds(20));
+	std::thread context_destroyer([&]() { AudioOut2::AudioOut2ContextDestroy(blocked); });
+	std::this_thread::sleep_for(std::chrono::milliseconds(20));
+	ReleaseOutputs();
+	pusher.join();
+	port_destroyer.join();
+	context_destroyer.join();
+
+	const auto output = OutputPcm();
+	Check(output.size() == 3, "pushes did not reach the backend once each");
+	CheckSamples(output[0], {0.5f}, "other context missed its PCM");
+	CheckSamples(output[1], {0.25f}, "PCM changed while its push was in flight");
+	Check(g_output_gains[0] == std::array {1.0f, 1.0f}, "gains changed while their push was in flight");
+	Check(LiveDeviceCount() == 1, "destroyed port or context kept its device");
+	CaptureOutputPcm(0);
+	AudioOut2::AudioOut2ContextDestroy(other);
+}
+
 } // namespace
 
 namespace Libs::Audio::AudioInternal {
@@ -625,6 +684,8 @@ int AudioOutOpen(int type, uint32_t /*samples_num*/, uint32_t /*freq*/, Format /
 void AudioOutClose(int handle) {
 	std::lock_guard lock(g_device_mutex);
 	const auto      it = std::find(g_live_devices.begin(), g_live_devices.end(), handle);
+	Check(std::find(g_blocked_output.begin(), g_blocked_output.end(), handle) == g_blocked_output.end(),
+	      "device closed during its output");
 	if (it != g_live_devices.end()) {
 		g_live_devices.erase(it);
 	}
@@ -642,8 +703,16 @@ bool AudioOutHasDevice(int handle) {
 }
 
 uint32_t AudioOutOutputs(const OutputParam* params, uint32_t num, bool blocking) {
-	std::lock_guard lock(g_device_mutex);
+	std::unique_lock lock(g_device_mutex);
 	g_output_blocking.push_back(blocking);
+	if (g_block_outputs && g_blocked_output.empty()) {
+		for (uint32_t i = 0; i < num; i++) {
+			g_blocked_output.push_back(params[i].handle);
+		}
+		g_device_cv.notify_all();
+		g_device_cv.wait(lock, []() { return !g_block_outputs; });
+		g_blocked_output.clear();
+	}
 	g_output_gains.clear();
 	if (g_capture_bytes != 0) {
 		for (uint32_t i = 0; i < num; i++) {
@@ -683,6 +752,7 @@ int main() {
 	TestPortGainAndValidation();
 	TestObjectGainAndAmbisonics();
 	TestObjectBusRouting();
+	TestBlockedPushOnlyHoldsItsContext();
 	Check(AudioOut2::AudioOut2UserDestroy(g_user_handle) == OK, "test user destroy failed");
 	std::printf("AudioOut2PortTests: all cases passed\n");
 	return 0;