fix AudioOut pacing

Co-authored-by: Stefanos Costa <stefanos.costa1999@gmail.com>

tests/AudioOut2PortTests.cpp

@@ -19,6 +19,8 @@ namespace AudioOut2 = Libs::Audio::AudioOut2;
 std::mutex              g_device_mutex;
 std::condition_variable g_device_cv;
 std::vector<int>        g_live_devices;
+std::vector<int>        g_device_backed_handles;
+std::vector<bool>       g_output_blocking;
 int                     g_next_device  = 1;
 int                     g_open_waiters = 0;
 bool                    g_block_opens  = false;
@@ -61,6 +63,17 @@ struct PortState {
 	uint64_t reserved[6];
 };
 
+struct Attribute {
+	uint32_t    attribute_id;
+	int32_t     reserved;
+	const void* value;
+	size_t      value_size;
+};
+
+struct Pcm {
+	const void* data;
+};
+
 const auto* AsParam(const PortParam* param) {
 	return reinterpret_cast<const AudioOut2::AudioOut2PortParam*>(param);
 }
@@ -73,6 +86,10 @@ auto* AsState(PortState* state) {
 	return reinterpret_cast<AudioOut2::AudioOut2PortState*>(state);
 }
 
+const auto* AsAttribute(const Attribute* attribute) {
+	return reinterpret_cast<const AudioOut2::AudioOut2Attribute*>(attribute);
+}
+
 PortParam MakeParam(uint32_t data_format = 0x200) {
 	PortParam param {};
 	param.data_format   = data_format;
@@ -80,9 +97,9 @@ PortParam MakeParam(uint32_t data_format = 0x200) {
 	return param;
 }
 
-AudioOut2::AudioOut2ContextHandle CreateContext() {
+AudioOut2::AudioOut2ContextHandle CreateContext(uint32_t queue_depth = 4) {
 	ContextParam param {};
-	param.queue_depth                         = 4;
+	param.queue_depth                         = queue_depth;
 	param.num_grains                          = 512;
 	AudioOut2::AudioOut2ContextHandle context = 0;
 	Check(AudioOut2::AudioOut2ContextCreate(AsParam(&param), nullptr, 0, &context) == OK,
@@ -112,6 +129,23 @@ int LiveDeviceCount() {
 	return static_cast<int>(g_live_devices.size());
 }
 
+void SetPcm(AudioOut2::AudioOut2PortHandle port, const void* data) {
+	const Pcm       pcm {data};
+	const Attribute attribute {0, 0, &pcm, sizeof(pcm)};
+	Check(AudioOut2::AudioOut2PortSetAttributes(port, AsAttribute(&attribute), 1) == OK,
+	      "setting PCM failed");
+}
+
+void ResetOutputCalls() {
+	std::lock_guard lock(g_device_mutex);
+	g_output_blocking.clear();
+}
+
+std::vector<bool> OutputCalls() {
+	std::lock_guard lock(g_device_mutex);
+	return g_output_blocking;
+}
+
 void TestSlotReuse() {
 	const auto context = CreateContext();
 	const auto param   = MakeParam();
@@ -202,14 +236,83 @@ void TestContextDestroyCancelsPendingCreate() {
 	Check(LiveDeviceCount() == 0, "cancelled port create leaked a device");
 }
 
+void TestSynchronousDevicePushBypassesModelledQueue() {
+	const auto context = CreateContext(1);
+	const auto param   = MakeParam();
+	AudioOut2::AudioOut2PortHandle port = 0;
+	Check(AudioOut2::AudioOut2PortCreate(context, AsParam(&param), &port) == OK,
+	      "device port create failed");
+
+	uint32_t pcm[512] {};
+	SetPcm(port, pcm);
+	ResetOutputCalls();
+
+	Check(AudioOut2::AudioOut2ContextPush(context, 1) == OK, "first sync push failed");
+	Check(AudioOut2::AudioOut2ContextPush(context, 1) == OK,
+	      "device-paced sync push was blocked by modelled queue");
+	const auto calls = OutputCalls();
+	Check(calls.size() == 2, "sync pushes did not reach the device backend");
+	Check(calls[0] && calls[1], "sync pushes lost their blocking mode");
+
+	AudioOut2::AudioOut2PortDestroy(port);
+	AudioOut2::AudioOut2ContextDestroy(context);
+}
+
+void TestAsynchronousDevicePushKeepsQueueBounded() {
+	const auto context = CreateContext(1);
+	const auto param   = MakeParam();
+	AudioOut2::AudioOut2PortHandle port = 0;
+	Check(AudioOut2::AudioOut2PortCreate(context, AsParam(&param), &port) == OK,
+	      "device port create failed");
+
+	uint32_t pcm[512] {};
+	SetPcm(port, pcm);
+	ResetOutputCalls();
+
+	Check(AudioOut2::AudioOut2ContextPush(context, 0) == OK, "first async push failed");
+	Check(AudioOut2::AudioOut2ContextPush(context, 0) != OK,
+	      "full async queue accepted another buffer");
+	const auto calls = OutputCalls();
+	Check(calls.size() == 1 && !calls[0], "rejected async push reached the device backend");
+
+	uint32_t queued    = 0;
+	uint32_t available = 0;
+	Check(AudioOut2::AudioOut2ContextGetQueueLevel(context, &queued, &available) == OK,
+	      "queue-level query failed");
+	Check(queued == 1 && available == 0, "async queue level does not match accepted pushes");
+
+	AudioOut2::AudioOut2PortDestroy(port);
+	AudioOut2::AudioOut2ContextDestroy(context);
+}
+
+void TestHandleWithoutPcmDoesNotBypassQueue() {
+	const auto context = CreateContext(1);
+	const auto param   = MakeParam();
+	AudioOut2::AudioOut2PortHandle port = 0;
+	Check(AudioOut2::AudioOut2PortCreate(context, AsParam(&param), &port) == OK,
+	      "device port create failed");
+	ResetOutputCalls();
+
+	Check(AudioOut2::AudioOut2ContextPush(context, 1) == OK, "empty sync push failed");
+	Check(AudioOut2::AudioOut2ContextPush(context, 0) != OK,
+	      "handle without PCM bypassed queue backpressure");
+	Check(OutputCalls().empty(), "empty push reached the device backend");
+
+	AudioOut2::AudioOut2PortDestroy(port);
+	AudioOut2::AudioOut2ContextDestroy(context);
+}
+
 } // namespace
 
 namespace Libs::Audio::AudioInternal {
 
-int AudioOutOpen(int /*type*/, uint32_t /*samples_num*/, uint32_t /*freq*/, Format /*format*/) {
+int AudioOutOpen(int type, uint32_t /*samples_num*/, uint32_t /*freq*/, Format /*format*/) {
 	std::unique_lock lock(g_device_mutex);
 	const int        handle = g_next_device++;
 	g_live_devices.push_back(handle);
+	if (type != 10) {
+		g_device_backed_handles.push_back(handle);
+	}
 	g_open_waiters++;
 	g_device_cv.notify_all();
 	g_device_cv.wait(lock, []() { return !g_block_opens; });
@@ -222,9 +325,22 @@ void AudioOutClose(int handle) {
 	if (it != g_live_devices.end()) {
 		g_live_devices.erase(it);
 	}
+	const auto device_it =
+	    std::find(g_device_backed_handles.begin(), g_device_backed_handles.end(), handle);
+	if (device_it != g_device_backed_handles.end()) {
+		g_device_backed_handles.erase(device_it);
+	}
 }
 
-uint32_t AudioOutOutputs(const OutputParam* /*params*/, uint32_t /*num*/, bool /*blocking*/) {
+bool AudioOutHasDevice(int handle) {
+	std::lock_guard lock(g_device_mutex);
+	return std::find(g_device_backed_handles.begin(), g_device_backed_handles.end(), handle) !=
+	       g_device_backed_handles.end();
+}
+
+uint32_t AudioOutOutputs(const OutputParam* /*params*/, uint32_t /*num*/, bool blocking) {
+	std::lock_guard lock(g_device_mutex);
+	g_output_blocking.push_back(blocking);
 	return 0;
 }
 
@@ -244,6 +360,9 @@ int main() {
 	TestFullTableRecovers();
 	TestConcurrentCreates();
 	TestContextDestroyCancelsPendingCreate();
+	TestSynchronousDevicePushBypassesModelledQueue();
+	TestAsynchronousDevicePushKeepsQueueBounded();
+	TestHandleWithoutPcmDoesNotBypassQueue();
 	std::printf("AudioOut2PortTests: all cases passed\n");
 	return 0;
 }