Merge branch 'v0.22.x'

src/io/FileDescriptor.cxx

@@ -172,7 +172,15 @@ FileDescriptor::CreatePipe(FileDescriptor &r, FileDescriptor &w) noexcept
 #endif
 }
 
-#ifndef _WIN32
+#ifdef _WIN32
+
+void
+FileDescriptor::SetBinaryMode() noexcept
+{
+	_setmode(fd, _O_BINARY);
+}
+
+#else // !_WIN32
 
 bool
 FileDescriptor::CreatePipeNonBlock(FileDescriptor &r,

src/io/FileDescriptor.hxx

@@ -147,10 +147,13 @@ public:
 #ifdef _WIN32
 	void EnableCloseOnExec() noexcept {}
 	void DisableCloseOnExec() noexcept {}
+	void SetBinaryMode() noexcept;
 #else
 	static bool CreatePipeNonBlock(FileDescriptor &r,
 				       FileDescriptor &w) noexcept;
 
+	void SetBinaryMode() noexcept {}
+
 	/**
 	 * Enable non-blocking mode on this file descriptor.
 	 */

src/mixer/plugins/WasapiMixerPlugin.cxx

@@ -44,7 +44,11 @@ public:
 	void Close() noexcept override {}
 
 	int GetVolume() override {
-		auto future = COMWorker::Async([&]() -> int {
+		auto com_worker = wasapi_output_get_com_worker(output);
+		if (!com_worker)
+			return -1;
+
+		auto future = com_worker->Async([&]() -> int {
 			HRESULT result;
 			float volume_level;
 
@@ -55,9 +59,9 @@ public:
 				result = endpoint_volume->GetMasterVolumeLevelScalar(
 					&volume_level);
 				if (FAILED(result)) {
-					throw FormatHResultError(result,
+					throw MakeHResultError(result,
-								 "Unable to get master "
+							       "Unable to get master "
-								 "volume level");
+							       "volume level");
 				}
 			} else {
 				auto session_volume =
@@ -65,7 +69,7 @@ public:
 
 				result = session_volume->GetMasterVolume(&volume_level);
 				if (FAILED(result)) {
-					throw FormatHResultError(
+					throw MakeHResultError(
 						result, "Unable to get master volume");
 				}
 			}
@@ -76,7 +80,11 @@ public:
 	}
 
 	void SetVolume(unsigned volume) override {
-		COMWorker::Async([&]() {
+		auto com_worker = wasapi_output_get_com_worker(output);
+		if (!com_worker)
+			throw std::runtime_error("Cannot set WASAPI volume");
+
+		com_worker->Async([&]() {
 			HRESULT result;
 			const float volume_level = volume / 100.0f;
 
@@ -87,7 +95,7 @@ public:
 				result = endpoint_volume->SetMasterVolumeLevelScalar(
 					volume_level, nullptr);
 				if (FAILED(result)) {
-					throw FormatHResultError(
+					throw MakeHResultError(
 						result,
 						"Unable to set master volume level");
 				}
@@ -98,7 +106,7 @@ public:
 				result = session_volume->SetMasterVolume(volume_level,
 									 nullptr);
 				if (FAILED(result)) {
-					throw FormatHResultError(
+					throw MakeHResultError(
 						result, "Unable to set master volume");
 				}
 			}

src/output/plugins/wasapi/AudioClient.hxx

@@ -33,8 +33,8 @@ GetBufferSizeInFrames(IAudioClient &client)
 
 	HRESULT result = client.GetBufferSize(&buffer_size_in_frames);
 	if (FAILED(result))
-		throw FormatHResultError(result,
+		throw MakeHResultError(result,
-					 "Unable to get audio client buffer size");
+				       "Unable to get audio client buffer size");
 
 	return buffer_size_in_frames;
 }
@@ -46,8 +46,8 @@ GetCurrentPaddingFrames(IAudioClient &client)
 
 	HRESULT result = client.GetCurrentPadding(&padding_frames);
 	if (FAILED(result))
-		throw FormatHResultError(result,
+		throw MakeHResultError(result,
-					 "Failed to get current padding");
+				       "Failed to get current padding");
 
 	return padding_frames;
 }
@@ -59,7 +59,7 @@ GetMixFormat(IAudioClient &client)
 
 	HRESULT result = client.GetMixFormat(&f);
 	if (FAILED(result))
-		throw FormatHResultError(result, "GetMixFormat failed");
+		throw MakeHResultError(result, "GetMixFormat failed");
 
 	return ComHeapPtr{f};
 }
@@ -69,7 +69,7 @@ Start(IAudioClient &client)
 {
 	HRESULT result = client.Start();
 	if (FAILED(result))
-		throw FormatHResultError(result, "Failed to start client");
+		throw MakeHResultError(result, "Failed to start client");
 }
 
 inline void
@@ -77,7 +77,7 @@ Stop(IAudioClient &client)
 {
 	HRESULT result = client.Stop();
 	if (FAILED(result))
-		throw FormatHResultError(result, "Failed to stop client");
+		throw MakeHResultError(result, "Failed to stop client");
 }
 
 inline void
@@ -85,7 +85,7 @@ SetEventHandle(IAudioClient &client, HANDLE h)
 {
 	HRESULT result = client.SetEventHandle(h);
 	if (FAILED(result))
-		throw FormatHResultError(result, "Unable to set event handle");
+		throw MakeHResultError(result, "Unable to set event handle");
 }
 
 template<typename T>
@@ -95,7 +95,7 @@ GetService(IAudioClient &client)
 	T *p = nullptr;
 	HRESULT result = client.GetService(IID_PPV_ARGS(&p));
 	if (FAILED(result))
-		throw FormatHResultError(result, "Unable to get service");
+		throw MakeHResultError(result, "Unable to get service");
 
 	return ComPtr{p};
 }

src/output/plugins/wasapi/Device.hxx

@@ -33,8 +33,8 @@ GetDefaultAudioEndpoint(IMMDeviceEnumerator &e)
 	HRESULT result = e.GetDefaultAudioEndpoint(eRender, eMultimedia,
 						   &device);
 	if (FAILED(result))
-		throw FormatHResultError(result,
+		throw MakeHResultError(result,
-					 "Unable to get default device for multimedia");
+				       "Unable to get default device for multimedia");
 
 	return ComPtr{device};
 }
@@ -47,7 +47,7 @@ EnumAudioEndpoints(IMMDeviceEnumerator &e)
 	HRESULT result = e.EnumAudioEndpoints(eRender, DEVICE_STATE_ACTIVE,
 					      &dc);
 	if (FAILED(result))
-		throw FormatHResultError(result, "Unable to enumerate devices");
+		throw MakeHResultError(result, "Unable to enumerate devices");
 
 	return ComPtr{dc};
 }
@@ -59,7 +59,7 @@ GetCount(IMMDeviceCollection &dc)
 
 	HRESULT result = dc.GetCount(&count);
 	if (FAILED(result))
-		throw FormatHResultError(result, "Collection->GetCount failed");
+		throw MakeHResultError(result, "Collection->GetCount failed");
 
 	return count;
 }
@@ -71,7 +71,7 @@ Item(IMMDeviceCollection &dc, UINT i)
 
 	auto result = dc.Item(i, &device);
 	if (FAILED(result))
-		throw FormatHResultError(result, "Collection->Item failed");
+		throw MakeHResultError(result, "Collection->Item failed");
 
 	return ComPtr{device};
 }
@@ -83,7 +83,7 @@ GetState(IMMDevice &device)
 
 	HRESULT result = device.GetState(&state);;
 	if (FAILED(result))
-		throw FormatHResultError(result, "Unable to get device status");
+		throw MakeHResultError(result, "Unable to get device status");
 
 	return state;
 }
@@ -96,7 +96,7 @@ Activate(IMMDevice &device)
 	HRESULT result = device.Activate(__uuidof(T), CLSCTX_ALL,
 					 nullptr, (void **)&p);
 	if (FAILED(result))
-		throw FormatHResultError(result, "Unable to activate device");
+		throw MakeHResultError(result, "Unable to activate device");
 
 	return ComPtr{p};
 }
@@ -108,8 +108,8 @@ OpenPropertyStore(IMMDevice &device)
 
 	HRESULT result = device.OpenPropertyStore(STGM_READ, &property_store);
 	if (FAILED(result))
-		throw FormatHResultError(result,
+		throw MakeHResultError(result,
-					 "Device->OpenPropertyStore failed");
+				       "Device->OpenPropertyStore failed");
 
 	return ComPtr{property_store};
 }

src/output/plugins/wasapi/ForMixer.hxx

@@ -20,10 +20,13 @@
 #ifndef MPD_WASAPI_OUTPUT_FOR_MIXER_HXX
 #define MPD_WASAPI_OUTPUT_FOR_MIXER_HXX
 
+#include <memory>
+
 struct IMMDevice;
 struct IAudioClient;
 class AudioOutput;
 class WasapiOutput;
+class COMWorker;
 
 [[gnu::pure]]
 WasapiOutput &
@@ -33,6 +36,10 @@ wasapi_output_downcast(AudioOutput &output) noexcept;
 bool
 wasapi_is_exclusive(WasapiOutput &output) noexcept;
 
+[[gnu::pure]]
+std::shared_ptr<COMWorker>
+wasapi_output_get_com_worker(WasapiOutput &output) noexcept;
+
 [[gnu::pure]]
 IMMDevice *
 wasapi_output_get_device(WasapiOutput &output) noexcept;

src/output/plugins/wasapi/WasapiOutputPlugin.cxx

@@ -61,7 +61,9 @@
 namespace {
 static constexpr Domain wasapi_output_domain("wasapi_output");
 
-gcc_const constexpr uint32_t GetChannelMask(const uint8_t channels) noexcept {
+constexpr uint32_t
+GetChannelMask(const uint8_t channels) noexcept
+{
 	switch (channels) {
 	case 1:
 		return KSAUDIO_SPEAKER_MONO;
@@ -86,18 +88,9 @@ gcc_const constexpr uint32_t GetChannelMask(const uint8_t channels) noexcept {
 }
 
 template <typename Functor>
-inline bool SafeTry(Functor &&functor) {
+inline bool
-	try {
+SafeSilenceTry(Functor &&functor) noexcept
-		functor();
+{
-		return true;
-	} catch (...) {
-		FormatError(std::current_exception(), "%s");
-		return false;
-	}
-}
-
-template <typename Functor>
-inline bool SafeSilenceTry(Functor &&functor) {
 	try {
 		functor();
 		return true;
@@ -106,7 +99,9 @@ inline bool SafeSilenceTry(Functor &&functor) {
 	}
 }
 
-std::vector<WAVEFORMATEXTENSIBLE> GetFormats(const AudioFormat &audio_format) noexcept {
+std::vector<WAVEFORMATEXTENSIBLE>
+GetFormats(const AudioFormat &audio_format) noexcept
+{
 #ifdef ENABLE_DSD
 	if (audio_format.format == SampleFormat::DSD) {
 		AudioFormat dop_format = audio_format;
@@ -152,57 +147,154 @@ std::vector<WAVEFORMATEXTENSIBLE> GetFormats(const AudioFormat &audio_format) no
 }
 
 #ifdef ENABLE_DSD
-void SetDSDFallback(AudioFormat &audio_format) noexcept {
+void
+SetDSDFallback(AudioFormat &audio_format) noexcept
+{
 	audio_format.format = SampleFormat::FLOAT;
 	audio_format.sample_rate = 384000;
 }
 #endif
 
-inline constexpr const unsigned int kErrorId = -1;
-
 } // namespace
 
-class WasapiOutputThread : public Thread {
+class WasapiOutputThread {
-public:
+	Thread thread{BIND_THIS_METHOD(Work)};
-	enum class Status : uint32_t { FINISH, PLAY, PAUSE };
-	WasapiOutputThread(IAudioClient *_client,
-			   ComPtr<IAudioRenderClient> &&_render_client,
-			   const UINT32 _frame_size, const UINT32 _buffer_size_in_frames,
-			   bool _is_exclusive)
-	: Thread(BIND_THIS_METHOD(Work)), client(_client),
-	  render_client(std::move(_render_client)), frame_size(_frame_size),
-	  buffer_size_in_frames(_buffer_size_in_frames), is_exclusive(_is_exclusive),
-	  spsc_buffer(_buffer_size_in_frames * 4 * _frame_size) {}
-	void Finish() noexcept { return SetStatus(Status::FINISH); }
-	void Play() noexcept { return SetStatus(Status::PLAY); }
-	void Pause() noexcept { return SetStatus(Status::PAUSE); }
-	void WaitDataPoped() noexcept { data_poped.Wait(); }
-	void CheckException() {
-		if (error.occur.load()) {
-			auto err = std::exchange(error.ptr, nullptr);
-			error.thrown.Set();
-			std::rethrow_exception(err);
-		}
-	}
-
-private:
-	friend class WasapiOutput;
 	WinEvent event;
 	WinEvent data_poped;
-	IAudioClient *client;
+	IAudioClient &client;
 	ComPtr<IAudioRenderClient> render_client;
 	const UINT32 frame_size;
 	const UINT32 buffer_size_in_frames;
-	bool is_exclusive;
+	const bool is_exclusive;
+
+	/**
+	 * This flag is only used by the calling thread
+	 * (i.e. #OutputThread), and specifies whether the
+	 * WasapiOutputThread has been told to play via Play().  This
+	 * variable is somewhat redundant because we already have
+	 * "state", but using this variable saves some overhead for
+	 * atomic operations.
+	 */
+	bool playing = false;
+
+	bool started = false;
+
+	std::atomic_bool cancel = false;
+
+	std::atomic_bool empty = true;
+
+	enum class Status : uint32_t { FINISH, PLAY, PAUSE };
+
 	alignas(BOOST_LOCKFREE_CACHELINE_BYTES) std::atomic<Status> status =
 		Status::PAUSE;
 	alignas(BOOST_LOCKFREE_CACHELINE_BYTES) struct {
 		std::atomic_bool occur = false;
 		std::exception_ptr ptr = nullptr;
-		WinEvent thrown;
 	} error;
 	boost::lockfree::spsc_queue<BYTE> spsc_buffer;
 
+public:
+	WasapiOutputThread(IAudioClient &_client,
+			   ComPtr<IAudioRenderClient> &&_render_client,
+			   const UINT32 _frame_size, const UINT32 _buffer_size_in_frames,
+			   bool _is_exclusive)
+		:client(_client),
+		 render_client(std::move(_render_client)), frame_size(_frame_size),
+		 buffer_size_in_frames(_buffer_size_in_frames), is_exclusive(_is_exclusive),
+		 spsc_buffer(_buffer_size_in_frames * 4 * _frame_size)
+	{
+		SetEventHandle(client, event.handle());
+		thread.Start();
+	}
+
+	void Finish() noexcept {
+		SetStatus(Status::FINISH);
+		thread.Join();
+	}
+
+	void Play() noexcept {
+		playing = true;
+		SetStatus(Status::PLAY);
+	}
+
+	void Pause() noexcept {
+		if (!playing)
+			return;
+
+		playing = false;
+		SetStatus(Status::PAUSE);
+	}
+
+	std::size_t Push(ConstBuffer<void> input) noexcept {
+		empty.store(false);
+
+		std::size_t consumed =
+			spsc_buffer.push(static_cast<const BYTE *>(input.data),
+					 input.size);
+
+		if (!playing) {
+			playing = true;
+			Play();
+		}
+
+		return consumed;
+	}
+
+	/**
+	 * Check if the buffer is empty, and if not, wait a bit.
+	 *
+	 * Throws on error.
+	 *
+	 * @return true if the buffer is now empty
+	 */
+	bool Drain() {
+		if (empty)
+			return true;
+
+		CheckException();
+		Wait();
+		CheckException();
+
+		return empty;
+	}
+
+	/**
+	 * Instruct the thread to discard the buffer (and wait for
+	 * completion).  This needs to be done inside this thread,
+	 * because only the consumer thread is allowed to do that.
+	 */
+	void Cancel() noexcept {
+		cancel.store(true);
+		event.Set();
+
+		while (cancel.load() && !error.occur.load())
+			Wait();
+
+		/* not rethrowing the exception here via
+		   CheckException() because this method must be
+		   "noexcept"; the next WasapiOutput::Play() call will
+		   throw */
+	}
+
+	/**
+	 * Wait for the thread to finish some work (e.g. until some
+	 * buffer space becomes available).
+	 */
+	void Wait() noexcept {
+		data_poped.Wait();
+	}
+
+	void InterruptWaiter() noexcept {
+		data_poped.Set();
+	}
+
+	void CheckException() {
+		if (error.occur.load()) {
+			std::rethrow_exception(error.ptr);
+		}
+	}
+
+private:
 	void SetStatus(Status s) noexcept {
 		status.store(s);
 		event.Set();
@@ -211,30 +303,61 @@ private:
 };
 
 class WasapiOutput final : public AudioOutput {
+	const bool is_exclusive;
+	const bool enumerate_devices;
+#ifdef ENABLE_DSD
+	const bool dop_setting;
+#endif
+
+	/**
+	 * Only valid if the output is open.
+	 */
+	bool paused;
+
+	std::atomic_flag not_interrupted = true;
+
+	const std::string device_config;
+
+	std::shared_ptr<COMWorker> com_worker;
+	ComPtr<IMMDevice> device;
+	ComPtr<IAudioClient> client;
+	WAVEFORMATEXTENSIBLE device_format;
+	std::optional<WasapiOutputThread> thread;
+	std::size_t watermark;
+	std::optional<PcmExport> pcm_export;
+
 public:
 	static AudioOutput *Create(EventLoop &, const ConfigBlock &block);
 	WasapiOutput(const ConfigBlock &block);
+
+	auto GetComWorker() noexcept {
+		// TODO: protect access to the shard_ptr
+		return com_worker;
+	}
+
 	void Enable() override {
-		COMWorker::Aquire();
+		com_worker = std::make_shared<COMWorker>();
 
 		try {
-			COMWorker::Async([&]() { OpenDevice(); }).get();
+			com_worker->Async([&]() { ChooseDevice(); }).get();
 		} catch (...) {
-			COMWorker::Release();
+			com_worker.reset();
 			throw;
 		}
 	}
 	void Disable() noexcept override {
-		COMWorker::Async([&]() { DoDisable(); }).get();
+		com_worker->Async([&]() { DoDisable(); }).get();
-		COMWorker::Release();
+		com_worker.reset();
 	}
 	void Open(AudioFormat &audio_format) override {
-		COMWorker::Async([&]() { DoOpen(audio_format); }).get();
+		com_worker->Async([&]() { DoOpen(audio_format); }).get();
+		paused = false;
 	}
 	void Close() noexcept override;
 	std::chrono::steady_clock::duration Delay() const noexcept override;
 	size_t Play(const void *chunk, size_t size) override;
 	void Drain() override;
+	void Cancel() noexcept override;
 	bool Pause() override;
 	void Interrupt() noexcept override;
 
@@ -245,22 +368,6 @@ public:
 	}
 
 private:
-	std::atomic_flag not_interrupted = true;
-	bool is_started = false;
-	bool is_exclusive;
-	bool enumerate_devices;
-#ifdef ENABLE_DSD
-	bool dop_setting;
-#endif
-	std::string device_config;
-	ComPtr<IMMDeviceEnumerator> enumerator;
-	ComPtr<IMMDevice> device;
-	ComPtr<IAudioClient> client;
-	WAVEFORMATEXTENSIBLE device_format;
-	std::optional<WasapiOutputThread> thread;
-	std::size_t watermark;
-	std::optional<PcmExport> pcm_export;
-
 	friend bool wasapi_is_exclusive(WasapiOutput &output) noexcept;
 	friend IMMDevice *wasapi_output_get_device(WasapiOutput &output) noexcept;
 	friend IAudioClient *wasapi_output_get_client(WasapiOutput &output) noexcept;
@@ -268,126 +375,179 @@ private:
 	void DoDisable() noexcept;
 	void DoOpen(AudioFormat &audio_format);
 
-	void OpenDevice();
+	void ChooseDevice();
 	bool TryFormatExclusive(const AudioFormat &audio_format);
 	void FindExclusiveFormatSupported(AudioFormat &audio_format);
 	void FindSharedFormatSupported(AudioFormat &audio_format);
-	void EnumerateDevices();
+	static void EnumerateDevices(IMMDeviceEnumerator &enumerator);
-	ComPtr<IMMDevice> GetDevice(unsigned int index);
+	static ComPtr<IMMDevice> GetDevice(IMMDeviceEnumerator &enumerator,
-	ComPtr<IMMDevice> SearchDevice(std::string_view name);
+					   unsigned index);
+	static ComPtr<IMMDevice> SearchDevice(IMMDeviceEnumerator &enumerator,
+					      std::string_view name);
 };
 
-WasapiOutput &wasapi_output_downcast(AudioOutput &output) noexcept {
+WasapiOutput &
+wasapi_output_downcast(AudioOutput &output) noexcept
+{
 	return static_cast<WasapiOutput &>(output);
 }
 
-bool wasapi_is_exclusive(WasapiOutput &output) noexcept { return output.is_exclusive; }
+bool
+wasapi_is_exclusive(WasapiOutput &output) noexcept
+{
+	return output.is_exclusive;
+}
 
-IMMDevice *wasapi_output_get_device(WasapiOutput &output) noexcept {
+std::shared_ptr<COMWorker>
+wasapi_output_get_com_worker(WasapiOutput &output) noexcept
+{
+	return output.GetComWorker();
+}
+
+IMMDevice *
+wasapi_output_get_device(WasapiOutput &output) noexcept
+{
 	return output.device.get();
 }
 
-IAudioClient *wasapi_output_get_client(WasapiOutput &output) noexcept {
+IAudioClient *
+wasapi_output_get_client(WasapiOutput &output) noexcept
+{
 	return output.client.get();
 }
 
-void WasapiOutputThread::Work() noexcept {
+inline void
+WasapiOutputThread::Work() noexcept
+try {
 	SetThreadName("Wasapi Output Worker");
 	FormatDebug(wasapi_output_domain, "Working thread started");
-	COM com{true};
+	COM com;
-	while (true) {
-		try {
-			event.Wait();
 
-			Status current_state = status.load();
+	AtScopeExit(this) {
-			if (current_state == Status::FINISH) {
+		if (started) {
-				FormatDebug(wasapi_output_domain,
+			try {
-					    "Working thread stopped");
+				Stop(client);
-				return;
+			} catch (...) {
+				LogError(std::current_exception());
 			}
-
-			UINT32 write_in_frames = buffer_size_in_frames;
-			if (!is_exclusive) {
-				UINT32 data_in_frames =
-					GetCurrentPaddingFrames(*client);
-
-				if (data_in_frames >= buffer_size_in_frames) {
-					continue;
-				}
-				write_in_frames -= data_in_frames;
-			}
-
-			BYTE *data;
-			DWORD mode = 0;
-
-			if (HRESULT result =
-				    render_client->GetBuffer(write_in_frames, &data);
-			    FAILED(result)) {
-				throw FormatHResultError(result, "Failed to get buffer");
-			}
-
-			AtScopeExit(&) {
-				render_client->ReleaseBuffer(write_in_frames, mode);
-			};
-
-			if (current_state == Status::PLAY) {
-				const UINT32 write_size = write_in_frames * frame_size;
-				UINT32 new_data_size = 0;
-				new_data_size = spsc_buffer.pop(data, write_size);
-				std::fill_n(data + new_data_size,
-					    write_size - new_data_size, 0);
-				data_poped.Set();
-			} else {
-				mode = AUDCLNT_BUFFERFLAGS_SILENT;
-				FormatDebug(wasapi_output_domain,
-					    "Working thread paused");
-			}
-		} catch (...) {
-			error.ptr = std::current_exception();
-			error.occur.store(true);
-			error.thrown.Wait();
 		}
+	};
+
+	while (true) {
+		event.Wait();
+
+		if (cancel.load()) {
+			spsc_buffer.consume_all([](auto &&) {});
+			cancel.store(false);
+			empty.store(true);
+			InterruptWaiter();
+		}
+
+		Status current_state = status.load();
+		switch (current_state) {
+		case Status::FINISH:
+			FormatDebug(wasapi_output_domain,
+				    "Working thread stopped");
+			return;
+
+		case Status::PAUSE:
+			if (!started)
+				/* don't bother starting the
+				   IAudioClient if we're paused */
+				continue;
+
+			/* stop the IAudioClient while paused; it will
+			   be restarted as soon as we're asked to
+			   resume playback */
+			Stop(client);
+			started = false;
+			continue;
+
+		case Status::PLAY:
+			break;
+		}
+
+		UINT32 write_in_frames = buffer_size_in_frames;
+		if (!is_exclusive) {
+			UINT32 data_in_frames =
+				GetCurrentPaddingFrames(client);
+			if (data_in_frames >= buffer_size_in_frames) {
+				continue;
+			}
+			write_in_frames -= data_in_frames;
+		}
+
+		BYTE *data;
+		DWORD mode = 0;
+
+		if (HRESULT result =
+		    render_client->GetBuffer(write_in_frames, &data);
+		    FAILED(result)) {
+			throw MakeHResultError(result, "Failed to get buffer");
+		}
+
+		AtScopeExit(&) {
+			render_client->ReleaseBuffer(write_in_frames, mode);
+
+			if (!started) {
+				Start(client);
+				started = true;
+			}
+		};
+
+		const UINT32 write_size = write_in_frames * frame_size;
+		UINT32 new_data_size = 0;
+		new_data_size = spsc_buffer.pop(data, write_size);
+		if (new_data_size == 0)
+			empty.store(true);
+
+		std::fill_n(data + new_data_size,
+			    write_size - new_data_size, 0);
+		InterruptWaiter();
 	}
+} catch (...) {
+	error.ptr = std::current_exception();
+	error.occur.store(true);
+
+	/* wake up the client thread which may be inside Wait() */
+	InterruptWaiter();
 }
 
-AudioOutput *WasapiOutput::Create(EventLoop &, const ConfigBlock &block) {
+AudioOutput *
+WasapiOutput::Create(EventLoop &, const ConfigBlock &block)
+{
 	return new WasapiOutput(block);
 }
 
 WasapiOutput::WasapiOutput(const ConfigBlock &block)
-: AudioOutput(FLAG_ENABLE_DISABLE | FLAG_PAUSE),
+	:AudioOutput(FLAG_ENABLE_DISABLE | FLAG_PAUSE),
-  is_exclusive(block.GetBlockValue("exclusive", false)),
+	 is_exclusive(block.GetBlockValue("exclusive", false)),
-  enumerate_devices(block.GetBlockValue("enumerate", false)),
+	 enumerate_devices(block.GetBlockValue("enumerate", false)),
 #ifdef ENABLE_DSD
-  dop_setting(block.GetBlockValue("dop", false)),
+	 dop_setting(block.GetBlockValue("dop", false)),
 #endif
-  device_config(block.GetBlockValue("device", "")) {
+	 device_config(block.GetBlockValue("device", ""))
+{
 }
 
 /// run inside COMWorkerThread
-void WasapiOutput::DoDisable() noexcept {
+void
-	if (thread) {
+WasapiOutput::DoDisable() noexcept
-		try {
+{
-			thread->Finish();
+	assert(!thread);
-			thread->Join();
+
-		} catch (std::exception &err) {
-			FormatError(wasapi_output_domain, "exception while disabling: %s",
-				    err.what());
-		}
-		thread.reset();
-		client.reset();
-	}
 	device.reset();
-	enumerator.reset();
 }
 
 /// run inside COMWorkerThread
-void WasapiOutput::DoOpen(AudioFormat &audio_format) {
+void
+WasapiOutput::DoOpen(AudioFormat &audio_format)
+{
 	client.reset();
 
 	if (GetState(*device) != DEVICE_STATE_ACTIVE) {
 		device.reset();
-		OpenDevice();
+		ChooseDevice();
 	}
 
 	client = Activate<IAudioClient>(*device);
@@ -444,7 +604,7 @@ void WasapiOutput::DoOpen(AudioFormat &audio_format) {
 	if (HRESULT result =
 		    client->GetDevicePeriod(&default_device_period, &min_device_period);
 	    FAILED(result)) {
-		throw FormatHResultError(result, "Unable to get device period");
+		throw MakeHResultError(result, "Unable to get device period");
 	}
 	FormatDebug(wasapi_output_domain,
 		    "Default device period: %I64u ns, Minimum device period: "
@@ -492,8 +652,7 @@ void WasapiOutput::DoOpen(AudioFormat &audio_format) {
 			}
 
 			if (FAILED(result)) {
-				throw FormatHResultError(
+				throw MakeHResultError(result, "Unable to initialize audio client");
-					result, "Unable to initialize audio client");
 			}
 		}
 	} else {
@@ -502,8 +661,8 @@ void WasapiOutput::DoOpen(AudioFormat &audio_format) {
 			    buffer_duration, 0,
 			    reinterpret_cast<WAVEFORMATEX *>(&device_format), nullptr);
 		    FAILED(result)) {
-			throw FormatHResultError(result,
+			throw MakeHResultError(result,
-						 "Unable to initialize audio client");
+					       "Unable to initialize audio client");
 		}
 	}
 
@@ -512,56 +671,49 @@ void WasapiOutput::DoOpen(AudioFormat &audio_format) {
 	const UINT32 buffer_size_in_frames = GetBufferSizeInFrames(*client);
 
 	watermark = buffer_size_in_frames * 3 * FrameSize();
-	thread.emplace(client.get(), std::move(render_client), FrameSize(),
+	thread.emplace(*client, std::move(render_client), FrameSize(),
 		       buffer_size_in_frames, is_exclusive);
 
-	SetEventHandle(*client, thread->event.handle());
+	paused = false;
-
-	thread->Start();
 }
 
-void WasapiOutput::Close() noexcept {
+void
+WasapiOutput::Close() noexcept
+{
 	assert(thread);
 
 	try {
-		COMWorker::Async([&]() {
-			Stop(*client);
-		}).get();
 		thread->CheckException();
-	} catch (std::exception &err) {
+	} catch (...) {
-		FormatError(wasapi_output_domain, "exception while stoping: %s",
+		FormatError(std::current_exception(),
-			    err.what());
+			    "exception while stopping");
 	}
-	is_started = false;
 	thread->Finish();
-	thread->Join();
+	com_worker->Async([&]() {
-	COMWorker::Async([&]() {
 		thread.reset();
 		client.reset();
 	}).get();
 	pcm_export.reset();
 }
 
-std::chrono::steady_clock::duration WasapiOutput::Delay() const noexcept {
+std::chrono::steady_clock::duration
-	if (!is_started) {
+WasapiOutput::Delay() const noexcept
+{
+	if (paused) {
 		// idle while paused
 		return std::chrono::seconds(1);
 	}
 
-	assert(thread);
+	return std::chrono::steady_clock::duration::zero();
-
-	const size_t data_size = thread->spsc_buffer.read_available();
-	const size_t delay_size = std::max(data_size, watermark) - watermark;
-
-	using s = std::chrono::seconds;
-	using duration = std::chrono::steady_clock::duration;
-	auto result = duration(s(delay_size)) / device_format.Format.nAvgBytesPerSec;
-	return result;
 }
 
-size_t WasapiOutput::Play(const void *chunk, size_t size) {
+size_t
+WasapiOutput::Play(const void *chunk, size_t size)
+{
 	assert(thread);
 
+	paused = false;
+
 	not_interrupted.test_and_set();
 
 	ConstBuffer<void> input(chunk, size);
@@ -572,25 +724,17 @@ size_t WasapiOutput::Play(const void *chunk, size_t size) {
 		return size;
 
 	do {
-		const size_t consumed_size = thread->spsc_buffer.push(
+		const size_t consumed_size = thread->Push({chunk, size});
-			static_cast<const BYTE *>(input.data), input.size);
+
 		if (consumed_size == 0) {
-			assert(is_started);
+			thread->Wait();
-			thread->WaitDataPoped();
+			thread->CheckException();
 			if (!not_interrupted.test_and_set()) {
 				throw AudioOutputInterrupted{};
 			}
 			continue;
 		}
 
-		if (!is_started) {
-			is_started = true;
-			thread->Play();
-			COMWorker::Async([&]() {
-				Start(*client);
-			}).wait();
-		}
-
 		thread->CheckException();
 
 		if (pcm_export) {
@@ -600,58 +744,82 @@ size_t WasapiOutput::Play(const void *chunk, size_t size) {
 	} while (true);
 }
 
-bool WasapiOutput::Pause() {
+bool
-	if (is_started) {
+WasapiOutput::Pause()
-		thread->Pause();
+{
-		is_started = false;
+	paused = true;
-	}
+	thread->Pause();
 	thread->CheckException();
 	return true;
 }
 
-void WasapiOutput::Interrupt() noexcept {
+void
+WasapiOutput::Interrupt() noexcept
+{
 	if (thread) {
 		not_interrupted.clear();
-		thread->data_poped.Set();
+		thread->InterruptWaiter();
 	}
 }
 
-void WasapiOutput::Drain() {
+void
+WasapiOutput::Drain()
+{
 	assert(thread);
 
-	thread->spsc_buffer.consume_all([](auto &&) {});
+	not_interrupted.test_and_set();
-	thread->CheckException();
+
+	while (!thread->Drain()) {
+		if (!not_interrupted.test_and_set())
+			throw AudioOutputInterrupted{};
+	}
+
+	/* TODO: this needs to wait until the hardware has really
+	   finished playing */
+}
+
+void
+WasapiOutput::Cancel() noexcept
+{
+	assert(thread);
+
+	thread->Cancel();
 }
 
 /// run inside COMWorkerThread
-void WasapiOutput::OpenDevice() {
+void
+WasapiOutput::ChooseDevice()
+{
+	ComPtr<IMMDeviceEnumerator> enumerator;
 	enumerator.CoCreateInstance(__uuidof(MMDeviceEnumerator), nullptr,
 				    CLSCTX_INPROC_SERVER);
 
 	if (enumerate_devices) {
 		try {
-			EnumerateDevices();
+			EnumerateDevices(*enumerator);
 		} catch (...) {
 			LogError(std::current_exception());
 		}
 	}
 
-	unsigned int id = kErrorId;
 	if (!device_config.empty()) {
+		unsigned int id;
 		if (!SafeSilenceTry([this, &id]() { id = std::stoul(device_config); })) {
-			device = SearchDevice(device_config);
+			device = SearchDevice(*enumerator, device_config);
 			if (!device)
 				throw FormatRuntimeError("Device '%s' not found",
 							 device_config.c_str());
 		} else
-			device = GetDevice(id);
+			device = GetDevice(*enumerator, id);
 	} else {
 		device = GetDefaultAudioEndpoint(*enumerator);
 	}
 }
 
 /// run inside COMWorkerThread
-bool WasapiOutput::TryFormatExclusive(const AudioFormat &audio_format) {
+bool
+WasapiOutput::TryFormatExclusive(const AudioFormat &audio_format)
+{
 	for (auto test_format : GetFormats(audio_format)) {
 		HRESULT result = client->IsFormatSupported(
 			AUDCLNT_SHAREMODE_EXCLUSIVE,
@@ -675,7 +843,9 @@ bool WasapiOutput::TryFormatExclusive(const AudioFormat &audio_format) {
 }
 
 /// run inside COMWorkerThread
-void WasapiOutput::FindExclusiveFormatSupported(AudioFormat &audio_format) {
+void
+WasapiOutput::FindExclusiveFormatSupported(AudioFormat &audio_format)
+{
 	for (uint8_t channels : {0, 2, 6, 8, 7, 1, 4, 5, 3}) {
 		if (audio_format.channels == channels) {
 			continue;
@@ -735,7 +905,9 @@ void WasapiOutput::FindExclusiveFormatSupported(AudioFormat &audio_format) {
 }
 
 /// run inside COMWorkerThread
-void WasapiOutput::FindSharedFormatSupported(AudioFormat &audio_format) {
+void
+WasapiOutput::FindSharedFormatSupported(AudioFormat &audio_format)
+{
 	HRESULT result;
 
 	// In shared mode, different sample rate is always unsupported.
@@ -760,7 +932,7 @@ void WasapiOutput::FindSharedFormatSupported(AudioFormat &audio_format) {
 	}
 
 	if (FAILED(result) && result != AUDCLNT_E_UNSUPPORTED_FORMAT) {
-		throw FormatHResultError(result, "IsFormatSupported failed");
+		throw MakeHResultError(result, "IsFormatSupported failed");
 	}
 
 	switch (result) {
@@ -789,7 +961,7 @@ void WasapiOutput::FindSharedFormatSupported(AudioFormat &audio_format) {
 				    result_string.c_str());
 		}
 		if (FAILED(result)) {
-			throw FormatHResultError(result, "Format is not supported");
+			throw MakeHResultError(result, "Format is not supported");
 		}
 		break;
 	case S_FALSE:
@@ -838,8 +1010,10 @@ void WasapiOutput::FindSharedFormatSupported(AudioFormat &audio_format) {
 }
 
 /// run inside COMWorkerThread
-void WasapiOutput::EnumerateDevices() {
+void
-	const auto device_collection = EnumAudioEndpoints(*enumerator);
+WasapiOutput::EnumerateDevices(IMMDeviceEnumerator &enumerator)
+{
+	const auto device_collection = EnumAudioEndpoints(enumerator);
 
 	const UINT count = GetCount(*device_collection);
 	for (UINT i = 0; i < count; ++i) {
@@ -860,17 +1034,18 @@ void WasapiOutput::EnumerateDevices() {
 
 /// run inside COMWorkerThread
 ComPtr<IMMDevice>
-WasapiOutput::GetDevice(unsigned int index)
+WasapiOutput::GetDevice(IMMDeviceEnumerator &enumerator, unsigned index)
 {
-	const auto device_collection = EnumAudioEndpoints(*enumerator);
+	const auto device_collection = EnumAudioEndpoints(enumerator);
 	return Item(*device_collection, index);
 }
 
 /// run inside COMWorkerThread
 ComPtr<IMMDevice>
-WasapiOutput::SearchDevice(std::string_view name)
+WasapiOutput::SearchDevice(IMMDeviceEnumerator &enumerator,
+			   std::string_view name)
 {
-	const auto device_collection = EnumAudioEndpoints(*enumerator);
+	const auto device_collection = EnumAudioEndpoints(enumerator);
 
 	const UINT count = GetCount(*device_collection);
 	for (UINT i = 0; i < count; ++i) {
@@ -885,7 +1060,11 @@ WasapiOutput::SearchDevice(std::string_view name)
 	return nullptr;
 }
 
-static bool wasapi_output_test_default_device() { return true; }
+static bool
+wasapi_output_test_default_device()
+{
+	return true;
+}
 
 const struct AudioOutputPlugin wasapi_output_plugin = {
 	"wasapi",

src/win32/Com.hxx

@@ -29,17 +29,9 @@
 class COM {
 public:
 	COM() {
-		if (HRESULT result = CoInitializeEx(nullptr, COINIT_MULTITHREADED);
+		if (HRESULT result = CoInitializeEx(nullptr, COINIT_APARTMENTTHREADED|COINIT_DISABLE_OLE1DDE);
 		    FAILED(result)) {
-			throw FormatHResultError(
+			throw MakeHResultError(
-				result,
-				"Unable to initialize COM with COINIT_MULTITHREADED");
-		}
-	}
-	COM(bool) {
-		if (HRESULT result = CoInitializeEx(nullptr, COINIT_APARTMENTTHREADED);
-		    FAILED(result)) {
-			throw FormatHResultError(
 				result,
 				"Unable to initialize COM with COINIT_APARTMENTTHREADED");
 		}

src/win32/ComPtr.hxx

@@ -85,7 +85,7 @@ public:
 			::CoCreateInstance(class_id, unknown_outer, class_context,
 					   __uuidof(T), reinterpret_cast<void **>(&ptr));
 		if (FAILED(result)) {
-			throw FormatHResultError(result, "Unable to create instance");
+			throw MakeHResultError(result, "Unable to create instance");
 		}
 	}
 

src/win32/ComWorker.cxx

@@ -21,13 +21,11 @@
 #include "Com.hxx"
 #include "thread/Name.hxx"
 
-Mutex COMWorker::mutex;
+void
-unsigned int COMWorker::reference_count = 0;
+COMWorker::Work() noexcept
-std::optional<COMWorker::COMWorkerThread> COMWorker::thread;
+{
-
-void COMWorker::COMWorkerThread::Work() noexcept {
 	SetThreadName("COM Worker");
-	COM com{true};
+	COM com;
 	while (true) {
 		if (!running_flag.test_and_set()) {
 			return;

src/win32/ComWorker.hxx

@@ -22,76 +22,46 @@
 
 #include "WinEvent.hxx"
 #include "thread/Future.hxx"
-#include "thread/Mutex.hxx"
 #include "thread/Thread.hxx"
 
 #include <boost/lockfree/spsc_queue.hpp>
-#include <mutex>
-#include <optional>
 
 #include <windows.h>
 
 // Worker thread for all COM operation
 class COMWorker {
-private:
+	Thread thread{BIND_THIS_METHOD(Work)};
-	class COMWorkerThread : public Thread {
-	public:
-		COMWorkerThread() : Thread{BIND_THIS_METHOD(Work)} {}
 
-	private:
+	boost::lockfree::spsc_queue<std::function<void()>> spsc_buffer{32};
-		friend class COMWorker;
+	std::atomic_flag running_flag = true;
-		void Work() noexcept;
+	WinEvent event{};
-		void Finish() noexcept {
-			running_flag.clear();
-			event.Set();
-		}
-		void Push(const std::function<void()> &function) {
-			spsc_buffer.push(function);
-			event.Set();
-		}
-
-		boost::lockfree::spsc_queue<std::function<void()>> spsc_buffer{32};
-		std::atomic_flag running_flag = true;
-		WinEvent event{};
-	};
 
 public:
-	static void Aquire() {
+	COMWorker() {
-		std::unique_lock locker(mutex);
+		thread.Start();
-		if (reference_count == 0) {
-			thread.emplace();
-			thread->Start();
-		}
-		++reference_count;
-	}
-	static void Release() noexcept {
-		std::unique_lock locker(mutex);
-		--reference_count;
-		if (reference_count == 0) {
-			thread->Finish();
-			thread->Join();
-			thread.reset();
-		}
 	}
 
-	template <typename Function, typename... Args>
+	~COMWorker() noexcept {
-	static auto Async(Function &&function, Args &&...args) {
+		Finish();
-		using R = std::invoke_result_t<std::decay_t<Function>,
+		thread.Join();
-					       std::decay_t<Args>...>;
+	}
+
+	COMWorker(const COMWorker &) = delete;
+	COMWorker &operator=(const COMWorker &) = delete;
+
+	template<typename Function>
+	auto Async(Function &&function) {
+		using R = std::invoke_result_t<std::decay_t<Function>>;
 		auto promise = std::make_shared<Promise<R>>();
 		auto future = promise->get_future();
-		thread->Push([function = std::forward<Function>(function),
+		Push([function = std::forward<Function>(function),
-			      args = std::make_tuple(std::forward<Args>(args)...),
 			      promise = std::move(promise)]() mutable {
 			try {
 				if constexpr (std::is_void_v<R>) {
-					std::apply(std::forward<Function>(function),
+					std::invoke(std::forward<Function>(function));
-						   std::move(args));
 					promise->set_value();
 				} else {
-					promise->set_value(std::apply(
+					promise->set_value(std::invoke(std::forward<Function>(function)));
-						std::forward<Function>(function),
-						std::move(args)));
 				}
 			} catch (...) {
 				promise->set_exception(std::current_exception());
@@ -101,9 +71,17 @@ public:
 	}
 
 private:
-	static Mutex mutex;
+	void Finish() noexcept {
-	static unsigned int reference_count;
+		running_flag.clear();
-	static std::optional<COMWorkerThread> thread;
+		event.Set();
+	}
+
+	void Push(const std::function<void()> &function) {
+		spsc_buffer.push(function);
+		event.Set();
+	}
+
+	void Work() noexcept;
 };
 
 #endif

src/win32/HResult.cxx

@@ -18,6 +18,7 @@
  */
 
 #include "HResult.hxx"
+#include "system/Error.hxx"
 
 #include <cassert>
 #include <cstdarg>
@@ -27,11 +28,21 @@
 std::string
 HResultCategory::message(int Errcode) const
 {
+	char buffer[256];
+
+	/* FormatMessage() supports some HRESULT values (depending on
+	   the Windows version) */
+	if (FormatMessageA(FORMAT_MESSAGE_FROM_SYSTEM |
+			   FORMAT_MESSAGE_IGNORE_INSERTS,
+			   nullptr, Errcode, 0,
+			   buffer, sizeof(buffer),
+			   nullptr))
+		return buffer;
+
 	const auto msg = HRESULTToString(Errcode);
 	if (!msg.empty())
 		return std::string(msg);
 
-	char buffer[11]; // "0x12345678\0"
 	int size = snprintf(buffer, sizeof(buffer), "0x%1x", Errcode);
 	assert(2 <= size && size <= 10);
 	return std::string(buffer, size);

src/win32/HResult.hxx

@@ -50,6 +50,8 @@ case x:
 		C(AUDCLNT_E_SERVICE_NOT_RUNNING);
 		C(AUDCLNT_E_UNSUPPORTED_FORMAT);
 		C(AUDCLNT_E_WRONG_ENDPOINT_TYPE);
+		C(AUDCLNT_E_NOT_INITIALIZED);
+		C(AUDCLNT_E_NOT_STOPPED);
 		C(CO_E_NOTINITIALIZED);
 		C(E_INVALIDARG);
 		C(E_OUTOFMEMORY);
@@ -74,6 +76,13 @@ static inline const std::error_category &hresult_category() noexcept {
 	return hresult_category_instance;
 }
 
+inline std::system_error
+MakeHResultError(HRESULT result, const char *msg) noexcept
+{
+	return std::system_error(std::error_code(result, hresult_category()),
+				 msg);
+}
+
 gcc_printf(2, 3) std::system_error
 FormatHResultError(HRESULT result, const char *fmt, ...) noexcept;
 

test/run_convert.cxx

@@ -29,6 +29,7 @@
 #include "pcm/Convert.hxx"
 #include "fs/Path.hxx"
 #include "fs/NarrowPath.hxx"
+#include "io/FileDescriptor.hxx"
 #include "util/ConstBuffer.hxx"
 #include "util/StaticFifoBuffer.hxx"
 #include "util/OptionDef.hxx"
@@ -101,26 +102,21 @@ public:
 	}
 };
 
-int
+static void
-main(int argc, char **argv)
+RunConvert(PcmConvert &convert, size_t in_frame_size,
-try {
+	   FileDescriptor in_fd, FileDescriptor out_fd)
-	const auto c = ParseCommandLine(argc, argv);
+{
+	in_fd.SetBinaryMode();
+	out_fd.SetBinaryMode();
 
-	SetLogThreshold(c.verbose ? LogLevel::DEBUG : LogLevel::INFO);
+	StaticFifoBuffer<std::byte, 4096> buffer;
-	const GlobalInit init(c.config_path);
-
-	const size_t in_frame_size = c.in_audio_format.GetFrameSize();
-
-	PcmConvert state(c.in_audio_format, c.out_audio_format);
-
-	StaticFifoBuffer<uint8_t, 4096> buffer;
 
 	while (true) {
 		{
 			const auto dest = buffer.Write();
 			assert(!dest.empty());
 
-			ssize_t nbytes = read(0, dest.data, dest.size);
+			ssize_t nbytes = in_fd.Read(dest.data, dest.size);
 			if (nbytes <= 0)
 				break;
 
@@ -136,20 +132,31 @@ try {
 
 		buffer.Consume(src.size);
 
-		auto output = state.Convert({src.data, src.size});
+		auto output = convert.Convert({src.data, src.size});
-
+		out_fd.FullWrite(output.data, output.size);
-		[[maybe_unused]] ssize_t ignored = write(1, output.data,
-						   output.size);
 	}
 
 	while (true) {
-		auto output = state.Flush();
+		auto output = convert.Flush();
 		if (output.IsNull())
 			break;
 
-		[[maybe_unused]] ssize_t ignored = write(1, output.data,
+		out_fd.FullWrite(output.data, output.size);
-						   output.size);
 	}
+}
+
+int
+main(int argc, char **argv)
+try {
+	const auto c = ParseCommandLine(argc, argv);
+
+	SetLogThreshold(c.verbose ? LogLevel::DEBUG : LogLevel::INFO);
+	const GlobalInit init(c.config_path);
+
+	PcmConvert state(c.in_audio_format, c.out_audio_format);
+	RunConvert(state, c.in_audio_format.GetFrameSize(),
+		   FileDescriptor(STDIN_FILENO),
+		   FileDescriptor(STDOUT_FILENO));
 
 	return EXIT_SUCCESS;
 } catch (...) {

test/run_input.cxx

@@ -164,6 +164,8 @@ static int
 dump_input_stream(InputStream &is, FileDescriptor out,
 		  offset_type seek, size_t chunk_size)
 {
+	out.SetBinaryMode();
+
 	std::unique_lock<Mutex> lock(is.mutex);
 
 	if (seek > 0)

test/run_output.cxx

@@ -31,6 +31,7 @@
 #include "util/StringBuffer.hxx"
 #include "util/RuntimeError.hxx"
 #include "util/ScopeExit.hxx"
+#include "util/StaticFifoBuffer.hxx"
 #include "util/PrintException.hxx"
 #include "LogBackend.hxx"
 
@@ -113,8 +114,11 @@ LoadAudioOutput(const ConfigData &config, EventLoop &event_loop,
 }
 
 static void
-run_output(AudioOutput &ao, AudioFormat audio_format)
+RunOutput(AudioOutput &ao, AudioFormat audio_format,
+	  FileDescriptor in_fd)
 {
+	in_fd.SetBinaryMode();
+
 	/* open the audio output */
 
 	ao.Enable();
@@ -126,33 +130,40 @@ run_output(AudioOutput &ao, AudioFormat audio_format)
 	fprintf(stderr, "audio_format=%s\n",
 		ToString(audio_format).c_str());
 
-	size_t frame_size = audio_format.GetFrameSize();
+	const size_t in_frame_size = audio_format.GetFrameSize();
 
 	/* play */
 
-	size_t length = 0;
+	StaticFifoBuffer<std::byte, 4096> buffer;
-	char buffer[4096];
+
 	while (true) {
-		if (length < sizeof(buffer)) {
+		{
-			ssize_t nbytes = read(0, buffer + length,
+			const auto dest = buffer.Write();
-					      sizeof(buffer) - length);
+			assert(!dest.empty());
+
+			ssize_t nbytes = in_fd.Read(dest.data, dest.size);
 			if (nbytes <= 0)
 				break;
 
-			length += (size_t)nbytes;
+			buffer.Append(nbytes);
 		}
 
-		size_t play_length = (length / frame_size) * frame_size;
+		auto src = buffer.Read();
-		if (play_length > 0) {
+		assert(!src.empty());
-			size_t consumed = ao.Play(buffer, play_length);
 
-			assert(consumed <= length);
+		src.size -= src.size % in_frame_size;
-			assert(consumed % frame_size == 0);
+		if (src.empty())
+			continue;
 
-			length -= consumed;
+		size_t consumed = ao.Play(src.data, src.size);
-			memmove(buffer, buffer + consumed, length);
+
-		}
+		assert(consumed <= src.size);
+		assert(consumed % in_frame_size == 0);
+
+		buffer.Consume(consumed);
 	}
+
+	ao.Drain();
 }
 
 int main(int argc, char **argv)
@@ -174,7 +185,7 @@ try {
 
 	/* do it */
 
-	run_output(*ao, c.audio_format);
+	RunOutput(*ao, c.audio_format, FileDescriptor(STDIN_FILENO));
 
 	/* cleanup and exit */