pipeline memory management fixes + logging

src/audio/audio_decoder.cpp

@@ -19,6 +19,8 @@
 
 namespace audio {
 
+static const char* kTag = "DEC";
+
 static const std::size_t kSamplesPerChunk = 256;
 
 AudioDecoder::AudioDecoder()
@@ -60,14 +62,14 @@ auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info)
 
   // TODO: defer until first header read, so we can give better info about
   // sample rate, chunk size, etc.
-  auto downstream_info = StreamEvent::CreateStreamInfo(
+  StreamInfo downstream_info(info);
-      input_events_, std::make_unique<StreamInfo>(info));
+  downstream_info.bits_per_sample = 32;
-  downstream_info->stream_info->bits_per_sample = 32;
+  downstream_info.sample_rate = 48'000;
-  downstream_info->stream_info->sample_rate = 48'000;
   chunk_size_ = 128;
-  downstream_info->stream_info->chunk_size = chunk_size_;
+  downstream_info.chunk_size = chunk_size_;
 
-  SendOrBufferEvent(std::move(downstream_info));
+  auto event = StreamEvent::CreateStreamInfo(input_events_, downstream_info);
+  SendOrBufferEvent(std::unique_ptr<StreamEvent>(event));
 
   return {};
 }
@@ -86,17 +88,18 @@ auto AudioDecoder::ProcessChunk(const cpp::span<std::byte>& chunk)
 
 auto AudioDecoder::Process() -> cpp::result<void, AudioProcessingError> {
   if (has_samples_to_send_) {
+    ESP_LOGI(kTag, "sending samples");
     // Writing samples is relatively quick (it's just a bunch of memcopy's), so
     // do them all at once.
     while (has_samples_to_send_ && !IsOverBuffered()) {
-      auto buffer = StreamEvent::CreateChunkData(input_events_, chunk_size_);
+      auto chunk = std::unique_ptr<StreamEvent>(
+          StreamEvent::CreateChunkData(input_events_, chunk_size_));
       auto write_res =
-          current_codec_->WriteOutputSamples(buffer->chunk_data.bytes);
+          current_codec_->WriteOutputSamples(chunk->chunk_data.bytes);
-      buffer->chunk_data.bytes =
+      chunk->chunk_data.bytes = chunk->chunk_data.bytes.first(write_res.first);
-          buffer->chunk_data.bytes.first(write_res.first);
       has_samples_to_send_ = !write_res.second;
 
-      if (!SendOrBufferEvent(std::move(buffer))) {
+      if (!SendOrBufferEvent(std::move(chunk))) {
         return {};
       }
     }
@@ -105,6 +108,7 @@ auto AudioDecoder::Process() -> cpp::result<void, AudioProcessingError> {
   }
 
   if (!needs_more_input_) {
+    ESP_LOGI(kTag, "decoding frame");
     auto res = current_codec_->ProcessNextFrame();
     if (res.has_error()) {
       // todo

src/audio/audio_element.cpp

@@ -3,7 +3,7 @@
 namespace audio {
 
 IAudioElement::IAudioElement()
-    : input_events_(xQueueCreate(kEventQueueSize, sizeof(StreamEvent))),
+    : input_events_(xQueueCreate(kEventQueueSize, sizeof(void*))),
       output_events_(nullptr),
       unprocessed_output_chunks_(0),
       buffered_output_(),
@@ -37,7 +37,7 @@ auto IAudioElement::SendOrBufferEvent(std::unique_ptr<StreamEvent> event)
     return false;
   }
   StreamEvent* raw_event = event.release();
-  if (!xQueueSend(output_events_, raw_event, 0)) {
+  if (!xQueueSend(output_events_, &raw_event, 0)) {
     buffered_output_.emplace_front(raw_event);
     return false;
   }
@@ -48,7 +48,7 @@ auto IAudioElement::FlushBufferedOutput() -> bool {
   while (!buffered_output_.empty()) {
     StreamEvent* raw_event = buffered_output_.front().release();
     buffered_output_.pop_front();
-    if (!xQueueSend(output_events_, raw_event, 0)) {
+    if (!xQueueSend(output_events_, &raw_event, 0)) {
       buffered_output_.emplace_front(raw_event);
       return false;
     }

src/audio/audio_playback.cpp

@@ -56,11 +56,10 @@ AudioPlayback::~AudioPlayback() {
 }
 
 auto AudioPlayback::Play(const std::string& filename) -> void {
-  auto info = std::make_unique<StreamInfo>();
+  StreamInfo info;
-  info->path = filename;
+  info.path = filename;
-  auto event = StreamEvent::CreateStreamInfo(nullptr, std::move(info));
+  auto event = StreamEvent::CreateStreamInfo(input_handle_, info);
-
+  xQueueSend(input_handle_, &event, portMAX_DELAY);
-  xQueueSend(input_handle_, event.release(), portMAX_DELAY);
 }
 
 auto AudioPlayback::ConnectElements(IAudioElement* src, IAudioElement* sink)

src/audio/audio_task.cpp

@@ -65,28 +65,41 @@ void AudioTaskMain(void* args) {
            element->HasUnprocessedInput()) &&
           !element->IsOverBuffered();
 
+      if (has_work_to_do) {
+        ESP_LOGI(kTag, "checking for events");
+      } else {
+        ESP_LOGI(kTag, "waiting for events");
+      }
+
       // If we have no new events to process and the element has nothing left to
       // do, then just delay forever waiting for a new event.
       TickType_t ticks_to_wait = has_work_to_do ? 0 : portMAX_DELAY;
 
-      StreamEvent* event_ptr = nullptr;
+      StreamEvent* new_event = nullptr;
       bool has_event =
-          xQueueReceive(element->InputEventQueue(), &event_ptr, ticks_to_wait);
+          xQueueReceive(element->InputEventQueue(), &new_event, ticks_to_wait);
 
-      if (has_event && event_ptr != nullptr) {
+      if (has_event) {
-        std::unique_ptr<StreamEvent> event(event_ptr);
+        if (new_event->tag == StreamEvent::UNINITIALISED) {
-        if (event->tag == StreamEvent::CHUNK_NOTIFICATION) {
+          ESP_LOGE(kTag, "discarding invalid event!!");
+        } else if (new_event->tag == StreamEvent::CHUNK_NOTIFICATION) {
+          ESP_LOGI(kTag, "marking chunk as used");
           element->OnChunkProcessed();
         } else {
           // This isn't an event that needs to be actioned immediately. Add it
           // to our work queue.
-          pending_events.push_back(std::move(event));
+          pending_events.emplace_back(new_event);
+          ESP_LOGI(kTag, "deferring event");
         }
         // Loop again, so that we service all incoming events before doing our
         // possibly expensive processing.
         continue;
       }
 
+      if (element->HasUnflushedOutput()) {
+        ESP_LOGI(kTag, "flushing output");
+      }
+
       // We have no new events. Next, see if there's anything that needs to be
       // flushed.
       if (element->HasUnflushedOutput() && !element->FlushBufferedOutput()) {
@@ -99,6 +112,7 @@ void AudioTaskMain(void* args) {
       }
 
       if (element->HasUnprocessedInput()) {
+        ESP_LOGI(kTag, "processing input events");
         auto process_res = element->Process();
         if (!process_res.has_error() || process_res.error() != OUT_OF_DATA) {
           // TODO: log!
@@ -109,27 +123,29 @@ void AudioTaskMain(void* args) {
       // The element ran out of data, so now it's time to let it process more
       // input.
       while (!pending_events.empty()) {
-        auto event = std::move(pending_events.front());
+        auto& event = pending_events.front();
-        pending_events.pop_front();
+        ESP_LOGI(kTag, "processing event, tag %i", event->tag);
 
         if (event->tag == StreamEvent::STREAM_INFO) {
+          ESP_LOGI(kTag, "processing stream info");
           auto process_res = element->ProcessStreamInfo(*event->stream_info);
+          pending_events.pop_front();
           if (process_res.has_error()) {
             // TODO(jacqueline)
             ESP_LOGE(kTag, "failed to process stream info");
           }
         } else if (event->tag == StreamEvent::CHUNK_DATA) {
-          StreamEvent* callback = new StreamEvent();
+          ESP_LOGI(kTag, "processing chunk data");
-          callback->source = element->InputEventQueue();
+          auto callback =
-          callback->tag = StreamEvent::CHUNK_NOTIFICATION;
+              StreamEvent::CreateChunkNotification(element->InputEventQueue());
-          if (!xQueueSend(event->source, callback, 0)) {
+          if (!xQueueSend(event->source, &callback, 0)) {
-            // TODO: log? crash? hmm.
+            ESP_LOGW(kTag, "failed to send chunk notif");
-            pending_events.push_front(std::move(event));
             continue;
           }
 
           auto process_chunk_res =
               element->ProcessChunk(event->chunk_data.bytes);
+          pending_events.pop_front();
           if (process_chunk_res.has_error()) {
             // TODO(jacqueline)
             ESP_LOGE(kTag, "failed to process chunk");

src/audio/fatfs_audio_input.cpp

@@ -12,6 +12,7 @@
 #include "chunk.hpp"
 #include "stream_buffer.hpp"
 #include "stream_event.hpp"
+#include "stream_info.hpp"
 #include "stream_message.hpp"
 
 static const char* kTag = "SRC";
@@ -43,6 +44,7 @@ auto FatfsAudioInput::ProcessStreamInfo(const StreamInfo& info)
   if (!info.path) {
     return cpp::fail(UNSUPPORTED_STREAM);
   }
+  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) {
@@ -51,12 +53,11 @@ auto FatfsAudioInput::ProcessStreamInfo(const StreamInfo& info)
 
   is_file_open_ = true;
 
-  std::unique_ptr<StreamInfo> new_info = std::make_unique<StreamInfo>(info);
+  StreamInfo new_info(info);
-  new_info->chunk_size = kChunkSize;
+  new_info.chunk_size = kChunkSize;
 
-  auto event =
+  auto event = StreamEvent::CreateStreamInfo(input_events_, new_info);
-      StreamEvent::CreateStreamInfo(input_events_, std::move(new_info));
+  SendOrBufferEvent(std::unique_ptr<StreamEvent>(event));
-  SendOrBufferEvent(std::move(event));
 
   return {};
 }
@@ -68,17 +69,19 @@ auto FatfsAudioInput::ProcessChunk(const cpp::span<std::byte>& chunk)
 
 auto FatfsAudioInput::Process() -> cpp::result<void, AudioProcessingError> {
   if (is_file_open_) {
-    auto dest_event = StreamEvent::CreateChunkData(input_events_, kChunkSize);
+    auto dest_event = std::unique_ptr<StreamEvent>(
+        StreamEvent::CreateChunkData(input_events_, kChunkSize));
     UINT bytes_read = 0;
 
-    FRESULT result =
+    ESP_LOGI(kTag, "reading from file");
-        f_read(&current_file_, dest_event->chunk_data.raw_bytes.get(),
+    FRESULT result = f_read(&current_file_, dest_event->chunk_data.raw_bytes,
-               kChunkSize, &bytes_read);
+                            kChunkSize, &bytes_read);
     if (result != FR_OK) {
       ESP_LOGE(kTag, "file I/O error %d", result);
       return cpp::fail(IO_ERROR);
     }
 
+    ESP_LOGI(kTag, "sending file data");
     dest_event->chunk_data.bytes =
         dest_event->chunk_data.bytes.first(bytes_read);
     SendOrBufferEvent(std::move(dest_event));

src/audio/i2s_audio_output.cpp

@@ -52,6 +52,9 @@ auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info)
     return cpp::fail(UNSUPPORTED_STREAM);
   }
 
+  ESP_LOGI(kTag, "incoming audio stream: %u bpp @ %u Hz", *info.bits_per_sample,
+           *info.sample_rate);
+
   drivers::AudioDac::BitsPerSample bps;
   switch (*info.bits_per_sample) {
     case 16:
@@ -86,6 +89,7 @@ auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info)
 
 auto I2SAudioOutput::ProcessChunk(const cpp::span<std::byte>& chunk)
     -> cpp::result<std::size_t, AudioProcessingError> {
+  ESP_LOGI(kTag, "playing samples");
   SetSoftMute(false);
   // TODO(jacqueline): write smaller parts with a small delay so that we can
   // be responsive to pause and seek commands.
@@ -94,7 +98,6 @@ auto I2SAudioOutput::ProcessChunk(const cpp::span<std::byte>& chunk)
 
 auto I2SAudioOutput::Process() -> cpp::result<void, AudioProcessingError> {
   // TODO(jacqueline): Consider powering down the dac completely maybe?
-  SetSoftMute(true);
   return {};
 }
 

src/audio/include/audio_element.hpp

@@ -64,7 +64,7 @@ class IAudioElement {
    * be tuned according to the observed stack size of each element, as different
    * elements have fairly different stack requirements (particular decoders).
    */
-  virtual auto StackSizeBytes() const -> std::size_t { return 2048; };
+  virtual auto StackSizeBytes() const -> std::size_t { return 4096; };
 
   virtual auto InputMinChunkSize() const -> std::size_t { return 0; }
 

src/audio/include/i2s_audio_output.hpp

@@ -8,7 +8,6 @@
 
 #include "dac.hpp"
 #include "gpio_expander.hpp"
-#include "sys/_stdint.h"
 
 namespace audio {
 

src/audio/include/stream_event.hpp

@@ -11,13 +11,11 @@
 namespace audio {
 
 struct StreamEvent {
-  static auto CreateStreamInfo(QueueHandle_t source,
+  static auto CreateStreamInfo(QueueHandle_t source, const StreamInfo& payload)
-                               std::unique_ptr<StreamInfo> payload)
+      -> StreamEvent*;
-      -> std::unique_ptr<StreamEvent>;
   static auto CreateChunkData(QueueHandle_t source, std::size_t chunk_size)
-      -> std::unique_ptr<StreamEvent>;
+      -> StreamEvent*;
-  static auto CreateChunkNotification(QueueHandle_t source)
+  static auto CreateChunkNotification(QueueHandle_t source) -> StreamEvent*;
-      -> std::unique_ptr<StreamEvent>;
 
   StreamEvent();
   ~StreamEvent();
@@ -33,16 +31,10 @@ struct StreamEvent {
   } tag;
 
   union {
-    std::unique_ptr<StreamInfo> stream_info;
+    StreamInfo* stream_info;
-
-    // Scott Meyers says:
-    // `About the only situation I can conceive of when a std::unique_ptr<T[]>
-    // would make sense would be when you’re using a C-like API that returns a
-    // raw pointer to a heap array that you assume ownership of.`
-    // :-)
 
     struct {
-      std::unique_ptr<std::byte*> raw_bytes;
+      std::byte* raw_bytes;
       cpp::span<std::byte> bytes;
     } chunk_data;
 

src/audio/stream_event.cpp

@@ -1,37 +1,37 @@
 #include "stream_event.hpp"
 #include <cstddef>
 #include <memory>
+#include "stream_info.hpp"
 
 namespace audio {
 
 auto StreamEvent::CreateStreamInfo(QueueHandle_t source,
-                                   std::unique_ptr<StreamInfo> payload)
+                                   const StreamInfo& payload) -> StreamEvent* {
-    -> std::unique_ptr<StreamEvent> {
+  auto event = new StreamEvent;
-  auto event = std::make_unique<StreamEvent>();
   event->tag = StreamEvent::STREAM_INFO;
   event->source = source;
-  event->stream_info = std::move(payload);
+  event->stream_info = new StreamInfo(payload);
   return event;
 }
 
 auto StreamEvent::CreateChunkData(QueueHandle_t source, std::size_t chunk_size)
-    -> std::unique_ptr<StreamEvent> {
+    -> StreamEvent* {
-  auto event = std::make_unique<StreamEvent>();
+  auto event = new StreamEvent;
   event->tag = StreamEvent::CHUNK_DATA;
   event->source = source;
 
   auto raw_bytes =
       static_cast<std::byte*>(heap_caps_malloc(chunk_size, MALLOC_CAP_SPIRAM));
 
-  event->chunk_data.raw_bytes = std::make_unique<std::byte*>(raw_bytes);
+  event->chunk_data.raw_bytes = raw_bytes;
   event->chunk_data.bytes = cpp::span<std::byte>(raw_bytes, chunk_size);
 
   return event;
 }
 
 auto StreamEvent::CreateChunkNotification(QueueHandle_t source)
-    -> std::unique_ptr<StreamEvent> {
+    -> StreamEvent* {
-  auto event = std::make_unique<StreamEvent>();
+  auto event = new StreamEvent;
   event->tag = StreamEvent::CHUNK_NOTIFICATION;
   event->source = source;
   return event;
@@ -44,10 +44,10 @@ StreamEvent::~StreamEvent() {
     case UNINITIALISED:
       break;
     case STREAM_INFO:
-      stream_info.reset();
+      delete stream_info;
       break;
     case CHUNK_DATA:
-      chunk_data.raw_bytes.reset();
+      free(chunk_data.raw_bytes);
       break;
     case CHUNK_NOTIFICATION:
       break;
@@ -61,10 +61,12 @@ StreamEvent::StreamEvent(StreamEvent&& other) {
     case UNINITIALISED:
       break;
     case STREAM_INFO:
-      stream_info = std::move(other.stream_info);
+      stream_info = other.stream_info;
+      other.stream_info = nullptr;
       break;
     case CHUNK_DATA:
-      chunk_data = std::move(other.chunk_data);
+      chunk_data = other.chunk_data;
+      other.chunk_data = {};
       break;
     case CHUNK_NOTIFICATION:
       break;