Tidy up pipeline and use arena capacity to test for overruns

src/audio/CMakeLists.txt

@@ -1,8 +1,7 @@
 idf_component_register(
   SRCS "audio_decoder.cpp" "audio_task.cpp" "chunk.cpp" "fatfs_audio_input.cpp"
   "stream_message.cpp" "i2s_audio_output.cpp" "stream_buffer.cpp"
-  "audio_playback.cpp" "audio_element_handle.cpp" "stream_event.cpp"
+  "audio_playback.cpp" "stream_event.cpp" "audio_element.cpp"
-  "audio_element.cpp"
   INCLUDE_DIRS "include"
   REQUIRES "codecs" "drivers" "cbor" "result" "tasks" "span" "memory")
 

src/audio/audio_decoder.cpp

@@ -37,15 +37,18 @@ auto AudioDecoder::HasUnprocessedInput() -> bool {
   return !needs_more_input_ || has_samples_to_send_;
 }
 
-auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info)
+auto AudioDecoder::IsOverBuffered() -> bool {
-    -> cpp::result<void, AudioProcessingError> {
+  return arena_.BlocksFree() == 0;
+}
+
+auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info) -> void {
   stream_info_ = info;
 
   if (info.chunk_size) {
     chunk_reader_.emplace(info.chunk_size.value());
   } else {
     ESP_LOGE(kTag, "no chunk size given");
-    return cpp::fail(UNSUPPORTED_STREAM);
+    return;
   }
 
   // Reuse the existing codec if we can. This will help with gapless playback,
@@ -54,7 +57,7 @@ auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info)
   if (current_codec_ != nullptr &&
       current_codec_->CanHandleFile(info.path.value_or(""))) {
     current_codec_->ResetForNewStream();
-    return {};
+    return;
   }
 
   auto result = codecs::CreateCodecForFile(info.path.value_or(""));
@@ -62,28 +65,23 @@ auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info)
     current_codec_ = std::move(result.value());
   } else {
     ESP_LOGE(kTag, "no codec for this file");
-    return cpp::fail(UNSUPPORTED_STREAM);
+    return;
   }
 
   stream_info_ = info;
   has_sent_stream_info_ = false;
-
-  return {};
 }
 
-auto AudioDecoder::ProcessChunk(const cpp::span<std::byte>& chunk)
+auto AudioDecoder::ProcessChunk(const cpp::span<std::byte>& chunk) -> void {
-    -> cpp::result<size_t, AudioProcessingError> {
   if (current_codec_ == nullptr || !chunk_reader_) {
     // Should never happen, but fail explicitly anyway.
     ESP_LOGW(kTag, "received chunk without chunk size or codec");
-    return cpp::fail(UNSUPPORTED_STREAM);
+    return;
   }
 
   ESP_LOGI(kTag, "received new chunk (size %u)", chunk.size());
   current_codec_->SetInput(chunk_reader_->HandleNewData(chunk));
   needs_more_input_ = false;
-
-  return {};
 }
 
 auto AudioDecoder::ProcessEndOfStream() -> void {
@@ -95,7 +93,7 @@ auto AudioDecoder::ProcessEndOfStream() -> void {
       StreamEvent::CreateEndOfStream(input_events_)));
 }
 
-auto AudioDecoder::Process() -> cpp::result<void, AudioProcessingError> {
+auto AudioDecoder::Process() -> void {
   if (has_samples_to_send_) {
     // Writing samples is relatively quick (it's just a bunch of memcopy's), so
     // do them all at once.
@@ -115,7 +113,7 @@ auto AudioDecoder::Process() -> cpp::result<void, AudioProcessingError> {
 
       auto block = arena_.Acquire();
       if (!block) {
-        return {};
+        return;
       }
 
       auto write_res =
@@ -126,18 +124,17 @@ auto AudioDecoder::Process() -> cpp::result<void, AudioProcessingError> {
       auto chunk = std::unique_ptr<StreamEvent>(
           StreamEvent::CreateArenaChunk(input_events_, *block));
       if (!SendOrBufferEvent(std::move(chunk))) {
-        return {};
+        return;
       }
     }
     // We will process the next frame during the next call to this method.
-    return {};
   }
 
   if (!needs_more_input_) {
     auto res = current_codec_->ProcessNextFrame();
     if (res.has_error()) {
-      // todo
+      // TODO(jacqueline): Handle errors.
-      return {};
+      return;
     }
     needs_more_input_ = res.value();
     has_samples_to_send_ = true;
@@ -146,8 +143,6 @@ auto AudioDecoder::Process() -> cpp::result<void, AudioProcessingError> {
       chunk_reader_->HandleBytesUsed(current_codec_->GetInputPosition());
     }
   }
-
-  return {};
 }
 
 }  // namespace audio

src/audio/audio_element.cpp

@@ -6,9 +6,7 @@ namespace audio {
 IAudioElement::IAudioElement()
     : input_events_(xQueueCreate(kEventQueueSize, sizeof(void*))),
       output_events_(nullptr),
-      unprocessed_output_chunks_(0),
+      buffered_output_() {}
-      buffered_output_(),
-      current_state_(STATE_RUN) {}
 
 IAudioElement::~IAudioElement() {
   // Ensure we don't leak any memory from events leftover in the queue.
@@ -28,9 +26,6 @@ IAudioElement::~IAudioElement() {
 
 auto IAudioElement::SendOrBufferEvent(std::unique_ptr<StreamEvent> event)
     -> bool {
-  if (event->tag == StreamEvent::ARENA_CHUNK) {
-    unprocessed_output_chunks_++;
-  }
   if (!buffered_output_.empty()) {
     // To ensure we send data in order, don't try to send if we've already
     // failed to send something.

src/audio/audio_element_handle.cpp

@@ -1,80 +0,0 @@
-#include "audio_element_handle.hpp"
-#include "audio_element.hpp"
-#include "freertos/projdefs.h"
-#include "freertos/task.h"
-
-namespace audio {
-
-AudioElementHandle::AudioElementHandle(std::unique_ptr<TaskHandle_t> task,
-                                       std::shared_ptr<IAudioElement> element)
-    : task_(std::move(task)), element_(std::move(element)) {}
-
-AudioElementHandle::~AudioElementHandle() {
-  Quit();
-}
-
-auto AudioElementHandle::CurrentState() -> ElementState {
-  return element_->ElementState();
-}
-
-auto AudioElementHandle::PlayPause(enum PlayPause state) -> void {
-  ElementState s = CurrentState();
-  if (state == PLAY && s == STATE_PAUSE) {
-    // Ensure we actually finished any previous pause command.
-    // TODO: really?
-    PauseSync();
-    SetStateAndWakeUp(STATE_RUN);
-    return;
-  }
-  if (state == PAUSE && s == STATE_RUN) {
-    element_->ElementState(STATE_PAUSE);
-    SetStateAndWakeUp(STATE_PAUSE);
-    return;
-  }
-}
-
-auto AudioElementHandle::Quit() -> void {
-  SetStateAndWakeUp(STATE_QUIT);
-}
-
-auto AudioElementHandle::PauseSync() -> void {
-  PlayPause(PAUSE);
-  MonitorUntilState(eSuspended);
-}
-
-auto AudioElementHandle::QuitSync() -> void {
-  Quit();
-  MonitorUntilState(eDeleted);
-}
-
-auto AudioElementHandle::MonitorUntilState(eTaskState desired) -> void {
-  while (eTaskGetState(*task_) != desired) {
-    WakeUpTask();
-    vTaskDelay(pdMS_TO_TICKS(1));
-  }
-}
-
-auto AudioElementHandle::SetStateAndWakeUp(ElementState state) -> void {
-  element_->ElementState(state);
-  WakeUpTask();
-}
-
-auto AudioElementHandle::WakeUpTask() -> void {
-  // TODO: various races where the task isn't blocked yet, but there is a block
-  // between now and its next element state check. Also think about chunk blocks
-  // nested in element bodies.
-  // Maybe we need a big mutex or semaphore somewhere in here.
-  switch (eTaskGetState(*task_)) {
-    case eBlocked:
-      // TODO: when is this safe?
-      xTaskAbortDelay(*task_);
-      break;
-    case eSuspended:
-      vTaskResume(*task_);
-      break;
-    default:
-      return;
-  }
-}
-
-}  // namespace audio

src/audio/audio_playback.cpp

@@ -4,11 +4,13 @@
 #include <cstdint>
 #include <memory>
 #include <string_view>
+
+#include "freertos/portmacro.h"
+
 #include "audio_decoder.hpp"
 #include "audio_task.hpp"
 #include "chunk.hpp"
 #include "fatfs_audio_input.hpp"
-#include "freertos/portmacro.h"
 #include "gpio_expander.hpp"
 #include "i2s_audio_output.hpp"
 #include "storage.hpp"
@@ -38,9 +40,9 @@ auto AudioPlayback::create(drivers::GpioExpander* expander,
   playback->ConnectElements(codec.get(), sink.get());
 
   // Launch!
-  playback->element_handles_.push_back(StartAudioTask("src", {}, source));
+  StartAudioTask("src", {}, source);
-  playback->element_handles_.push_back(StartAudioTask("dec", {}, codec));
+  StartAudioTask("dec", {}, codec);
-  playback->element_handles_.push_back(StartAudioTask("sink", 0, sink));
+  StartAudioTask("sink", 0, sink);
 
   playback->input_handle_ = source->InputEventQueue();
 
@@ -49,11 +51,7 @@ auto AudioPlayback::create(drivers::GpioExpander* expander,
 
 AudioPlayback::AudioPlayback() {}
 
-AudioPlayback::~AudioPlayback() {
+AudioPlayback::~AudioPlayback() {}
-  for (auto& element : element_handles_) {
-    element->Quit();
-  }
-}
 
 auto AudioPlayback::Play(const std::string& filename) -> void {
   StreamInfo info;

src/audio/audio_task.cpp

@@ -6,8 +6,6 @@
 #include <deque>
 #include <memory>
 
-#include "arena.hpp"
-#include "audio_element_handle.hpp"
 #include "cbor.h"
 #include "esp_heap_caps.h"
 #include "esp_log.h"
@@ -16,6 +14,7 @@
 #include "freertos/queue.h"
 #include "span.hpp"
 
+#include "arena.hpp"
 #include "audio_element.hpp"
 #include "chunk.hpp"
 #include "stream_event.hpp"
@@ -30,8 +29,7 @@ static const char* kTag = "task";
 
 auto StartAudioTask(const std::string& name,
                     std::optional<BaseType_t> core_id,
-                    std::shared_ptr<IAudioElement> element)
+                    std::shared_ptr<IAudioElement> element) -> void {
-    -> std::unique_ptr<AudioElementHandle> {
   auto task_handle = std::make_unique<TaskHandle_t>();
 
   // Newly created task will free this.
@@ -46,8 +44,6 @@ auto StartAudioTask(const std::string& name,
     xTaskCreate(&AudioTaskMain, name.c_str(), element->StackSizeBytes(), args,
                 kTaskPriorityAudio, task_handle.get());
   }
-
-  return std::make_unique<AudioElementHandle>(std::move(task_handle), element);
 }
 
 void AudioTaskMain(void* args) {
@@ -62,7 +58,8 @@ void AudioTaskMain(void* args) {
     // processed.
     std::deque<std::unique_ptr<StreamEvent>> pending_events;
 
-    while (element->ElementState() != STATE_QUIT) {
+    // TODO(jacqueline): quit event
+    while (true) {
       // First, we pull events from our input queue into pending_events. This
       // keeps us responsive to any events that need to be handled immediately.
       // Then we check if there's any events to flush downstream.
@@ -91,8 +88,6 @@ void AudioTaskMain(void* args) {
         if (new_event->tag == StreamEvent::UNINITIALISED) {
           ESP_LOGE(kTag, "discarding invalid event!!");
         } else if (new_event->tag == StreamEvent::CHUNK_NOTIFICATION) {
-          ESP_LOGD(kTag, "marking chunk as used");
-          element->OnChunkProcessed();
           delete new_event;
         } else if (new_event->tag == StreamEvent::LOG_STATUS) {
           element->ProcessLogStatus();
@@ -129,10 +124,7 @@ void AudioTaskMain(void* args) {
 
       if (element->HasUnprocessedInput()) {
         ESP_LOGD(kTag, "processing input events");
-        auto process_res = element->Process();
+        element->Process();
-        if (!process_res.has_error() || process_res.error() != OUT_OF_DATA) {
-          // TODO: log!
-        }
         continue;
       }
 
@@ -146,13 +138,12 @@ void AudioTaskMain(void* args) {
 
         if (event->tag == StreamEvent::STREAM_INFO) {
           ESP_LOGD(kTag, "processing stream info");
-          auto process_res = element->ProcessStreamInfo(*event->stream_info);
+
-          if (process_res.has_error()) {
+          element->ProcessStreamInfo(*event->stream_info);
-            // TODO(jacqueline)
+
-            ESP_LOGE(kTag, "failed to process stream info");
-          }
         } else if (event->tag == StreamEvent::ARENA_CHUNK) {
           ESP_LOGD(kTag, "processing arena data");
+
           memory::ArenaRef ref(event->arena_chunk);
           auto callback =
               StreamEvent::CreateChunkNotification(element->InputEventQueue());
@@ -163,13 +154,7 @@ void AudioTaskMain(void* args) {
 
           // TODO(jacqueline): Consider giving the element a full ArenaRef here,
           // so that it can hang on to it and potentially save an alloc+copy.
-          auto process_chunk_res =
+          element->ProcessChunk({ref.ptr.start, ref.ptr.used_size});
-              element->ProcessChunk({ref.ptr.start, ref.ptr.used_size});
-          if (process_chunk_res.has_error()) {
-            // TODO(jacqueline)
-            ESP_LOGE(kTag, "failed to process chunk");
-            continue;
-          }
 
           // TODO: think about whether to do the whole queue
           break;

src/audio/fatfs_audio_input.cpp

@@ -36,22 +36,27 @@ auto FatfsAudioInput::HasUnprocessedInput() -> bool {
   return is_file_open_;
 }
 
-auto FatfsAudioInput::ProcessStreamInfo(const StreamInfo& info)
+auto FatfsAudioInput::IsOverBuffered() -> bool {
-    -> cpp::result<void, AudioProcessingError> {
+  return arena_.BlocksFree() == 0;
+}
+
+auto FatfsAudioInput::ProcessStreamInfo(const StreamInfo& info) -> void {
   if (is_file_open_) {
     f_close(&current_file_);
     is_file_open_ = false;
   }
 
   if (!info.path) {
-    return cpp::fail(UNSUPPORTED_STREAM);
+    // TODO(jacqueline): Handle errors.
+    return;
   }
   ESP_LOGI(kTag, "opening file %s", info.path->c_str());
   std::string path = *info.path;
   FRESULT res = f_open(&current_file_, path.c_str(), FA_READ);
   if (res != FR_OK) {
     ESP_LOGE(kTag, "failed to open file! res: %i", res);
-    return cpp::fail(IO_ERROR);
+    // TODO(jacqueline): Handle errors.
+    return;
   }
 
   is_file_open_ = true;
@@ -62,14 +67,9 @@ auto FatfsAudioInput::ProcessStreamInfo(const StreamInfo& info)
 
   auto event = StreamEvent::CreateStreamInfo(input_events_, new_info);
   SendOrBufferEvent(std::unique_ptr<StreamEvent>(event));
-
-  return {};
 }
 
-auto FatfsAudioInput::ProcessChunk(const cpp::span<std::byte>& chunk)
+auto FatfsAudioInput::ProcessChunk(const cpp::span<std::byte>& chunk) -> void {}
-    -> cpp::result<size_t, AudioProcessingError> {
-  return cpp::fail(UNSUPPORTED_STREAM);
-}
 
 auto FatfsAudioInput::ProcessEndOfStream() -> void {
   if (is_file_open_) {
@@ -80,18 +80,19 @@ auto FatfsAudioInput::ProcessEndOfStream() -> void {
   }
 }
 
-auto FatfsAudioInput::Process() -> cpp::result<void, AudioProcessingError> {
+auto FatfsAudioInput::Process() -> void {
   if (is_file_open_) {
     auto dest_block = memory::ArenaRef::Acquire(&arena_);
     if (!dest_block) {
-      return {};
+      return;
     }
 
     FRESULT result = f_read(&current_file_, dest_block->ptr.start,
                             dest_block->ptr.size, &dest_block->ptr.used_size);
     if (result != FR_OK) {
       ESP_LOGE(kTag, "file I/O error %d", result);
-      return cpp::fail(IO_ERROR);
+      // TODO(jacqueline): Handle errors.
+      return;
     }
 
     if (dest_block->ptr.used_size < dest_block->ptr.size ||
@@ -105,7 +106,6 @@ auto FatfsAudioInput::Process() -> cpp::result<void, AudioProcessingError> {
 
     SendOrBufferEvent(std::move(dest_event));
   }
-  return {};
 }
 
 }  // namespace audio

src/audio/i2s_audio_output.cpp

@@ -40,8 +40,6 @@ I2SAudioOutput::I2SAudioOutput(drivers::GpioExpander* expander,
                                std::unique_ptr<drivers::AudioDac> dac)
     : expander_(expander),
       dac_(std::move(dac)),
-      volume_(255),
-      is_soft_muted_(false),
       chunk_reader_(),
       latest_chunk_() {}
 
@@ -51,18 +49,21 @@ auto I2SAudioOutput::HasUnprocessedInput() -> bool {
   return latest_chunk_.size() > 0;
 }
 
-auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info)
+auto I2SAudioOutput::IsOverBuffered() -> bool {
-    -> cpp::result<void, AudioProcessingError> {
+  return false;
+}
+
+auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info) -> void {
   // TODO(jacqueline): probs do something with the channel hey
 
   if (!info.bits_per_sample || !info.sample_rate) {
     ESP_LOGE(kTag, "audio stream missing bits or sample rate");
-    return cpp::fail(UNSUPPORTED_STREAM);
+    return;
   }
 
   if (!info.chunk_size) {
     ESP_LOGE(kTag, "audio stream missing chunk size");
-    return cpp::fail(UNSUPPORTED_STREAM);
+    return;
   }
   chunk_reader_.emplace(*info.chunk_size);
 
@@ -82,7 +83,7 @@ auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info)
       break;
     default:
       ESP_LOGE(kTag, "dropping stream with unknown bps");
-      return cpp::fail(UNSUPPORTED_STREAM);
+      return;
   }
 
   drivers::AudioDac::SampleRate sample_rate;
@@ -95,18 +96,14 @@ auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info)
       break;
     default:
       ESP_LOGE(kTag, "dropping stream with unknown rate");
-      return cpp::fail(UNSUPPORTED_STREAM);
+      return;
   }
 
   dac_->Reconfigure(bps, sample_rate);
-
-  return {};
 }
 
-auto I2SAudioOutput::ProcessChunk(const cpp::span<std::byte>& chunk)
+auto I2SAudioOutput::ProcessChunk(const cpp::span<std::byte>& chunk) -> void {
-    -> cpp::result<std::size_t, AudioProcessingError> {
   latest_chunk_ = chunk_reader_->HandleNewData(chunk);
-  return 0;
 }
 
 auto I2SAudioOutput::ProcessEndOfStream() -> void {
@@ -119,8 +116,9 @@ auto I2SAudioOutput::ProcessLogStatus() -> void {
   dac_->LogStatus();
 }
 
-auto I2SAudioOutput::Process() -> cpp::result<void, AudioProcessingError> {
+auto I2SAudioOutput::Process() -> void {
-  // Note: no logging here!
+  // Note: avoid logging here! We need to get bytes from the chunk buffer into
+  // the I2S DMA buffer as fast as possible, to avoid running out of samples.
   std::size_t bytes_written = dac_->WriteData(latest_chunk_);
   if (bytes_written == latest_chunk_.size_bytes()) {
     latest_chunk_ = cpp::span<std::byte>();
@@ -128,26 +126,11 @@ auto I2SAudioOutput::Process() -> cpp::result<void, AudioProcessingError> {
   } else {
     latest_chunk_ = latest_chunk_.subspan(bytes_written);
   }
-  return {};
+  return;
 }
 
 auto I2SAudioOutput::SetVolume(uint8_t volume) -> void {
-  volume_ = volume;
+  dac_->WriteVolume(volume);
-  if (!is_soft_muted_) {
-    dac_->WriteVolume(volume);
-  }
-}
-
-auto I2SAudioOutput::SetSoftMute(bool enabled) -> void {
-  if (enabled == is_soft_muted_) {
-    return;
-  }
-  is_soft_muted_ = enabled;
-  if (is_soft_muted_) {
-    dac_->WriteVolume(255);
-  } else {
-    dac_->WriteVolume(volume_);
-  }
 }
 
 }  // namespace audio

src/audio/include/audio_decoder.hpp

@@ -24,21 +24,13 @@ class AudioDecoder : public IAudioElement {
 
   auto StackSizeBytes() const -> std::size_t override { return 10 * 1024; };
 
-  auto InputMinChunkSize() const -> std::size_t override {
-    // 128 kbps MPEG-1 @ 44.1 kHz is approx. 418 bytes according to the
-    // internet.
-    // TODO(jacqueline): tune as more codecs are added.
-    return 1024;
-  }
-
   auto HasUnprocessedInput() -> bool override;
+  auto IsOverBuffered() -> bool override;
 
-  auto ProcessStreamInfo(const StreamInfo& info)
+  auto ProcessStreamInfo(const StreamInfo& info) -> void override;
-      -> cpp::result<void, AudioProcessingError> override;
+  auto ProcessChunk(const cpp::span<std::byte>& chunk) -> void override;
-  auto ProcessChunk(const cpp::span<std::byte>& chunk)
-      -> cpp::result<std::size_t, AudioProcessingError> override;
   auto ProcessEndOfStream() -> void override;
-  auto Process() -> cpp::result<void, AudioProcessingError> override;
+  auto Process() -> void override;
 
   AudioDecoder(const AudioDecoder&) = delete;
   AudioDecoder& operator=(const AudioDecoder&) = delete;
@@ -50,7 +42,6 @@ class AudioDecoder : public IAudioElement {
   std::optional<ChunkReader> chunk_reader_;
 
   bool has_sent_stream_info_;
-  std::size_t chunk_size_;
   bool has_samples_to_send_;
   bool needs_more_input_;
 };

src/audio/include/audio_element.hpp

@@ -20,25 +20,6 @@
 
 namespace audio {
 
-enum ElementState {
-  STATE_RUN,
-  STATE_PAUSE,
-  STATE_QUIT,
-};
-
-/*
- * Errors that may be returned by any of the Process* methods of an audio
- * element.
- */
-enum AudioProcessingError {
-  // Indicates that this element is unable to handle the upcoming chunks.
-  UNSUPPORTED_STREAM,
-  // Indicates an error with reading or writing stream data.
-  IO_ERROR,
-  // Indicates that the element has run out of data to process.
-  OUT_OF_DATA,
-};
-
 static const size_t kEventQueueSize = 8;
 
 /*
@@ -66,35 +47,24 @@ class IAudioElement {
    */
   virtual auto StackSizeBytes() const -> std::size_t { return 4096; };
 
-  virtual auto InputMinChunkSize() const -> std::size_t { return 0; }
-
   /* Returns this element's input buffer. */
   auto InputEventQueue() const -> QueueHandle_t { return input_events_; }
-
   /* Returns this element's output buffer. */
   auto OutputEventQueue() const -> QueueHandle_t { return output_events_; }
-
   auto OutputEventQueue(const QueueHandle_t q) -> void { output_events_ = q; }
 
-  auto HasUnflushedOutput() -> bool { return !buffered_output_.empty(); }
-
   virtual auto HasUnprocessedInput() -> bool = 0;
 
-  auto IsOverBuffered() -> bool { return unprocessed_output_chunks_ > 4; }
+  virtual auto IsOverBuffered() -> bool { return false; }
 
+  auto HasUnflushedOutput() -> bool { return !buffered_output_.empty(); }
   auto FlushBufferedOutput() -> bool;
 
-  auto ElementState() const -> ElementState { return current_state_; }
-  auto ElementState(enum ElementState e) -> void { current_state_ = e; }
-
-  virtual auto OnChunkProcessed() -> void { unprocessed_output_chunks_--; }
-
   /*
    * Called when a StreamInfo message is received. Used to configure this
    * element in preperation for incoming chunks.
    */
-  virtual auto ProcessStreamInfo(const StreamInfo& info)
+  virtual auto ProcessStreamInfo(const StreamInfo& info) -> void = 0;
-      -> cpp::result<void, AudioProcessingError> = 0;
 
   /*
    * Called when a ChunkHeader message is received. Includes the data associated
@@ -102,8 +72,7 @@ class IAudioElement {
    * bytes in this chunk that were actually used; leftover bytes will be
    * prepended to the next call.
    */
-  virtual auto ProcessChunk(const cpp::span<std::byte>& chunk)
+  virtual auto ProcessChunk(const cpp::span<std::byte>& chunk) -> void = 0;
-      -> cpp::result<std::size_t, AudioProcessingError> = 0;
 
   virtual auto ProcessEndOfStream() -> void = 0;
 
@@ -114,7 +83,7 @@ class IAudioElement {
    * time. This could be used to synthesize output, or to save memory by
    * releasing unused resources.
    */
-  virtual auto Process() -> cpp::result<void, AudioProcessingError> = 0;
+  virtual auto Process() -> void = 0;
 
  protected:
   auto SendOrBufferEvent(std::unique_ptr<StreamEvent> event) -> bool;
@@ -125,15 +94,8 @@ class IAudioElement {
   // if we're not yet in a pipeline.
   // FIXME: it would be nicer if this was non-nullable.
   QueueHandle_t output_events_;
-
-  // The number of output chunks that we have generated, but have not yet been
-  // processed by the next element in the pipeline. This includes any chunks
-  // that are currently help in buffered_output_.
-  int unprocessed_output_chunks_;
   // Output events that have been generated, but are yet to be sent downstream.
   std::deque<std::unique_ptr<StreamEvent>> buffered_output_;
-
-  enum ElementState current_state_;
 };
 
 }  // namespace audio

src/audio/include/audio_element_handle.hpp

@@ -1,41 +0,0 @@
-#pragma once
-
-#include <memory>
-#include "audio_element.hpp"
-
-namespace audio {
-
-class AudioElementHandle {
- public:
-  AudioElementHandle(std::unique_ptr<TaskHandle_t> task,
-                     std::shared_ptr<IAudioElement> element);
-  ~AudioElementHandle();
-
-  auto CurrentState() -> ElementState;
-
-  // TODO: think about this contract. Would it ever make sense to pause and
-  // then walk away? Things could keep running for a whole loop if data comes
-  // through, so probably not?
-  enum PlayPause {
-    PLAY,
-    PAUSE,
-  };
-  auto PlayPause(PlayPause state) -> void;
-  auto Quit() -> void;
-
-  auto PauseSync() -> void;
-  auto QuitSync() -> void;
-
-  AudioElementHandle(const AudioElementHandle&) = delete;
-  AudioElementHandle& operator=(const AudioElementHandle&) = delete;
-
- private:
-  std::unique_ptr<TaskHandle_t> task_;
-  std::shared_ptr<IAudioElement> element_;
-
-  auto MonitorUntilState(eTaskState desired) -> void;
-  auto SetStateAndWakeUp(ElementState state) -> void;
-  auto WakeUpTask() -> void;
-};
-
-}  // namespace audio

src/audio/include/audio_playback.hpp

@@ -5,12 +5,12 @@
 #include <string>
 #include <vector>
 
-#include "audio_element.hpp"
-#include "audio_element_handle.hpp"
 #include "esp_err.h"
-#include "gpio_expander.hpp"
 #include "result.hpp"
 #include "span.hpp"
+
+#include "audio_element.hpp"
+#include "gpio_expander.hpp"
 #include "storage.hpp"
 #include "stream_buffer.hpp"
 
@@ -45,7 +45,6 @@ class AudioPlayback {
 
  private:
   auto ConnectElements(IAudioElement* src, IAudioElement* sink) -> void;
-  std::vector<std::unique_ptr<AudioElementHandle>> element_handles_;
 
   QueueHandle_t input_handle_;
 };

src/audio/include/audio_task.hpp

@@ -5,7 +5,6 @@
 #include <string>
 
 #include "audio_element.hpp"
-#include "audio_element_handle.hpp"
 #include "freertos/portmacro.h"
 
 namespace audio {
@@ -16,8 +15,7 @@ struct AudioTaskArgs {
 
 auto StartAudioTask(const std::string& name,
                     std::optional<BaseType_t> core_id,
-                    std::shared_ptr<IAudioElement> element)
+                    std::shared_ptr<IAudioElement> element) -> void;
-    -> std::unique_ptr<AudioElementHandle>;
 
 void AudioTaskMain(void* args);
 

src/audio/include/fatfs_audio_input.hpp

@@ -24,13 +24,12 @@ class FatfsAudioInput : public IAudioElement {
   ~FatfsAudioInput();
 
   auto HasUnprocessedInput() -> bool override;
+  auto IsOverBuffered() -> bool override;
 
-  auto ProcessStreamInfo(const StreamInfo& info)
+  auto ProcessStreamInfo(const StreamInfo& info) -> void override;
-      -> cpp::result<void, AudioProcessingError> override;
+  auto ProcessChunk(const cpp::span<std::byte>& chunk) -> void override;
-  auto ProcessChunk(const cpp::span<std::byte>& chunk)
-      -> cpp::result<std::size_t, AudioProcessingError> override;
   auto ProcessEndOfStream() -> void override;
-  auto Process() -> cpp::result<void, AudioProcessingError> override;
+  auto Process() -> void override;
 
   FatfsAudioInput(const FatfsAudioInput&) = delete;
   FatfsAudioInput& operator=(const FatfsAudioInput&) = delete;

src/audio/include/i2s_audio_output.hpp

@@ -23,28 +23,23 @@ class I2SAudioOutput : public IAudioElement {
   ~I2SAudioOutput();
 
   auto HasUnprocessedInput() -> bool override;
+  auto IsOverBuffered() -> bool override;
 
-  auto ProcessStreamInfo(const StreamInfo& info)
+  auto ProcessStreamInfo(const StreamInfo& info) -> void override;
-      -> cpp::result<void, AudioProcessingError> override;
+  auto ProcessChunk(const cpp::span<std::byte>& chunk) -> void override;
-  auto ProcessChunk(const cpp::span<std::byte>& chunk)
-      -> cpp::result<std::size_t, AudioProcessingError> override;
   auto ProcessEndOfStream() -> void override;
   auto ProcessLogStatus() -> void override;
-  auto Process() -> cpp::result<void, AudioProcessingError> override;
+  auto Process() -> void override;
 
   I2SAudioOutput(const I2SAudioOutput&) = delete;
   I2SAudioOutput& operator=(const I2SAudioOutput&) = delete;
 
  private:
   auto SetVolume(uint8_t volume) -> void;
-  auto SetSoftMute(bool enabled) -> void;
 
   drivers::GpioExpander* expander_;
   std::unique_ptr<drivers::AudioDac> dac_;
 
-  uint8_t volume_;
-  bool is_soft_muted_;
-
   std::optional<ChunkReader> chunk_reader_;
   cpp::span<std::byte> latest_chunk_;
 };

src/codecs/mad.cpp

@@ -121,7 +121,7 @@ auto MadMp3Decoder::WriteOutputSamples(cpp::span<std::byte> output)
       uint32_t sample_24 =
           scaleTo24Bits(synth_.pcm.samples[channel][current_sample_]);
       output[output_byte++] = static_cast<std::byte>((sample_24 >> 8) & 0xFF);
-      output[output_byte++] = static_cast<std::byte>((sample_24) & 0xFF);
+      output[output_byte++] = static_cast<std::byte>((sample_24)&0xFF);
     }
     current_sample_++;
   }

src/drivers/dac.cpp

@@ -50,19 +50,19 @@ auto AudioDac::create(GpioExpander* expander)
   i2s_std_config_t i2s_config = {
       .clk_cfg = dac->clock_config_,
       .slot_cfg = dac->slot_config_,
-      .gpio_cfg = {
+      .gpio_cfg =
-        // TODO: investigate running in three wire mode for less noise
+          {// TODO: investigate running in three wire mode for less noise
-        .mclk = GPIO_NUM_0,
+           .mclk = GPIO_NUM_0,
-                   .bclk = GPIO_NUM_26,
+           .bclk = GPIO_NUM_26,
-                   .ws = GPIO_NUM_27,
+           .ws = GPIO_NUM_27,
-                   .dout = GPIO_NUM_5,
+           .dout = GPIO_NUM_5,
-                   .din = I2S_GPIO_UNUSED,
+           .din = I2S_GPIO_UNUSED,
-                   .invert_flags =
+           .invert_flags =
-                       {
+               {
-                           .mclk_inv = false,
+                   .mclk_inv = false,
-                           .bclk_inv = false,
+                   .bclk_inv = false,
-                           .ws_inv = false,
+                   .ws_inv = false,
-                       }},
+               }},
   };
 
   if (esp_err_t err =
@@ -89,11 +89,12 @@ auto AudioDac::create(GpioExpander* expander)
   dac->WriteRegister(Register::DAC_CLOCK_SOURCE, 0b11 << 5);
 
   // Enable auto clocking, and do your best to carry on despite errors.
-  //dac->WriteRegister(Register::CLOCK_ERRORS, 0b1111101);
+  // dac->WriteRegister(Register::CLOCK_ERRORS, 0b1111101);
 
   i2s_channel_enable(dac->i2s_handle_);
 
-  dac->WaitForPowerState([](bool booted, PowerState state) { return state == STANDBY; });
+  dac->WaitForPowerState(
+      [](bool booted, PowerState state) { return state == STANDBY; });
 
   return dac;
 }
@@ -208,16 +209,11 @@ auto AudioDac::Stop() -> void {
 }
 
 #define BYTE_TO_BINARY_PATTERN "%c%c%c%c%c%c%c%c"
-#define BYTE_TO_BINARY(byte)  \
+#define BYTE_TO_BINARY(byte)                                \
-  (byte & 0x80 ? '1' : '0'), \
+  (byte & 0x80 ? '1' : '0'), (byte & 0x40 ? '1' : '0'),     \
-  (byte & 0x40 ? '1' : '0'), \
+      (byte & 0x20 ? '1' : '0'), (byte & 0x10 ? '1' : '0'), \
-  (byte & 0x20 ? '1' : '0'), \
+      (byte & 0x08 ? '1' : '0'), (byte & 0x04 ? '1' : '0'), \
-  (byte & 0x10 ? '1' : '0'), \
+      (byte & 0x02 ? '1' : '0'), (byte & 0x01 ? '1' : '0')
-  (byte & 0x08 ? '1' : '0'), \
-  (byte & 0x04 ? '1' : '0'), \
-  (byte & 0x02 ? '1' : '0'), \
-  (byte & 0x01 ? '1' : '0') 
-
 
 auto AudioDac::LogStatus() -> void {
   uint8_t res;

src/memory/arena.cpp

@@ -23,7 +23,8 @@ Arena::Arena(std::size_t block_size,
 }
 
 Arena::~Arena() {
-  // TODO: assert queue is full?
+  // We shouldn't have any blocks in use when destroying an arena.
+  assert(uxQueueSpacesAvailable(free_blocks_) == 0);
   vQueueDelete(free_blocks_);
   free(pool_);
 }
@@ -44,6 +45,10 @@ auto Arena::Return(ArenaPtr ptr) -> void {
   xQueueSend(free_blocks_, &ptr.start, 0);
 }
 
+auto Arena::BlocksFree() -> std::size_t {
+  return uxQueueMessagesWaiting(free_blocks_);
+}
+
 auto ArenaRef::Acquire(Arena* a) -> std::optional<ArenaRef> {
   auto ptr = a->Acquire();
   if (ptr) {