Improvements to the queue for shuffling/playing all (#170)

Queue now has a separate 'ready' property to indicate it's ready to be used, which is independent from whether it's still loading tracks in. This also improves the response time for shuffling all tracks (we will initially pick a random track in the first 100 tracks whilst the rest of the tracks are loading). This should also fix issues where one song will start playing and then repeat itself when the queue finishes loading, and hopefully solve #160 as well (though I couldn't actually repro this myself).

Co-authored-by: jacqueline <me@jacqueline.id.au>
Reviewed-on: https://codeberg.org/cool-tech-zone/tangara-fw/pulls/170
Reviewed-by: cooljqln <cooljqln@noreply.codeberg.org>
Co-authored-by: ailurux <ailuruxx@gmail.com>
Co-committed-by: ailurux <ailuruxx@gmail.com>

lua/playing.lua

@@ -267,11 +267,15 @@ return screen:new {
           scrubber:set { value = pos / track.duration * 100 }
         end
       end),
-      playback.track:bind(function(track)
+      queue.ready:bind(function(ready)
-        if not track then
+        if not ready then
-          if queue.loading:get() then
           title:set { text = "Loading..." }
+          album:set{text=""}
+          artist:set{text=""}
         end
+      end),
+      playback.track:bind(function(track)
+        if not track then
           return
         end
         if track.duration then

luals-stubs/queue.lua

@@ -9,6 +9,8 @@
 --- @field position Property The index in the queue of the currently playing track. This may be zero if the queue is empty. Writeable.
 --- @field size Property The total number of tracks in the queue, including tracks which have already been played.
 --- @field repeat_mode Property The current repeat mode for the queue. Writeable.
+--- @field loading Property Whether or not the queue is currently loading tracks.
+--- @field ready Property Whether or not the queue is ready to be used.
 --- @field random Property Determines whether, when progressing to the next track in the queue, the next track will be chosen randomly. The random selection algorithm used is a Miller Shuffle, which guarantees that no repeat selections will be made until every item in the queue has been played. Writeable.
 local queue = {}
 

src/tangara/audio/audio_fsm.cpp

@@ -150,6 +150,7 @@ void AudioState::react(const SetTrack& ev) {
   if (std::holds_alternative<std::monostate>(ev.new_track)) {
     ESP_LOGI(kTag, "playback finished, awaiting drain");
     sDecoder->open({});
+    sStreamCues.clear();
     return;
   }
 

src/tangara/audio/stream_cues.cpp

@@ -43,6 +43,11 @@ auto StreamCues::addCue(std::shared_ptr<TrackInfo> track, uint32_t sample)
   }
 }
 
+auto StreamCues::clear() -> void {
+  upcoming_.clear();
+  current_ = {};
+}
+
 auto StreamCues::current() -> std::pair<std::shared_ptr<TrackInfo>, uint32_t> {
   if (!current_) {
     return {};

src/tangara/audio/stream_cues.hpp

@@ -34,6 +34,8 @@ class StreamCues {
 
   auto addCue(std::shared_ptr<TrackInfo>, uint32_t start_at) -> void;
 
+  auto clear() -> void;
+
  private:
   uint32_t now_;
 

src/tangara/audio/track_queue.cpp

@@ -37,11 +37,9 @@ namespace audio {
 
 using Reason = QueueUpdate::Reason;
 
-RandomIterator::RandomIterator()
+RandomIterator::RandomIterator() : seed_(0), pos_(0), size_(0) {}
-    : seed_(0), pos_(0), size_(0) {}
 
-RandomIterator::RandomIterator(size_t size)
+RandomIterator::RandomIterator(size_t size) : seed_(), pos_(0), size_(size) {
-    : seed_(), pos_(0), size_(size) {
   esp_fill_random(&seed_, sizeof(seed_));
 }
 
@@ -95,7 +93,9 @@ auto notifyPlayFrom(uint32_t start_from_position) -> void {
   events::Audio().Dispatch(ev);
 }
 
-TrackQueue::TrackQueue(tasks::WorkerPool& bg_worker, database::Handle db, drivers::NvsStorage& nvs)
+TrackQueue::TrackQueue(tasks::WorkerPool& bg_worker,
+                       database::Handle db,
+                       drivers::NvsStorage& nvs)
     : mutex_(),
       bg_worker_(bg_worker),
       db_(db),
@@ -103,10 +103,17 @@ TrackQueue::TrackQueue(tasks::WorkerPool& bg_worker, database::Handle db, driver
       playlist_(".queue.playlist"),
       position_(0),
       shuffle_(),
-      repeatMode_(static_cast<RepeatMode>(nvs.QueueRepeatMode())) {}
+      repeatMode_(static_cast<RepeatMode>(nvs.QueueRepeatMode())),
+      cancel_appending_async_(false),
+      appending_async_(false),
+      loading_(false),
+      ready_(true) {}
 
 auto TrackQueue::current() const -> TrackItem {
   const std::shared_lock<std::shared_mutex> lock(mutex_);
+  if (!ready_) {
+    return {};
+  }
   std::string val;
   if (opened_playlist_ && position_ < opened_playlist_->size()) {
     val = opened_playlist_->value();
@@ -205,6 +212,7 @@ auto TrackQueue::append(Item i) -> void {
       if (!filename.empty()) {
         playlist_.append(filename);
       }
+      ready_ = true;
       updateShuffler(was_queue_empty);
     }
     notifyChanged(current_changed, Reason::kExplicitUpdate);
@@ -214,6 +222,7 @@ auto TrackQueue::append(Item i) -> void {
       {
         const std::unique_lock<std::shared_mutex> lock(mutex_);
         playlist_.append(std::get<std::string>(i));
+        ready_ = true;
         updateShuffler(was_queue_empty);
       }
       notifyChanged(current_changed, Reason::kExplicitUpdate);
@@ -222,15 +231,64 @@ auto TrackQueue::append(Item i) -> void {
     // Iterators can be very large, and retrieving items from them often
     // requires disk i/o. Handle them asynchronously so that inserting them
     // doesn't block.
-    bg_worker_.Dispatch<void>([=, this]() {
+    appendAsync(std::get<database::TrackIterator>(i), was_queue_empty);
-      database::TrackIterator it = std::get<database::TrackIterator>(i);
+  }
+}
+
+auto TrackQueue::appendAsync(database::TrackIterator it, bool was_empty)
+    -> void {
+  // First, check whether or not an async append is already running. Grab the
+  // mutex first to avoid races where we check appending_async_ between the bg
+  // task looking at pending_async_iterators_ and resetting appending_async_.
+  {
+    const std::unique_lock<std::shared_mutex> lock(mutex_);
+    if (appending_async_) {
+      // We are already appending, so just add to the queue.
+      pending_async_iterators_.push_back(it);
+      return;
+    } else {
+      // We need to start a new task.
+      appending_async_ = true;
+      cancel_appending_async_ = false;
+    }
+  }
 
+  bg_worker_.Dispatch<void>([=, this]() mutable {
+    bool update_current = was_empty;
+    if (update_current) {
+      ready_ = false;
+    }
+    loading_ = true;
     size_t next_update_at = 10;
-      while (true) {
+
+    while (!cancel_appending_async_) {
       auto next = *it;
       if (!next) {
-          break;
+        // The current iterator is out of tracks. Is there another iterator for
+        // us to process?
+        {
+          const std::unique_lock<std::shared_mutex> lock(mutex_);
+          if (!pending_async_iterators_.empty()) {
+            // Yes. Grab it and continue.
+            it = pending_async_iterators_.front();
+            pending_async_iterators_.pop_front();
+            continue;
+          } else {
+            // No, time to finish up.
+            // First make sure the shuffler has the final count.
+            updateShuffler(update_current);
+
+            // Now reset all our state.
+            loading_ = false;
+            ready_ = true;
+            appending_async_ = false;
+            appending_async_.notify_all();
+            notifyChanged(update_current, Reason::kExplicitUpdate);
+            return;
           }
+        }
+      }
+
       // Keep this critical section small so that we're not blocking methods
       // like current().
       {
@@ -246,18 +304,44 @@ auto TrackQueue::append(Item i) -> void {
       // queue updates during them so that the user has an idea what's going
       // on.
       if (!--next_update_at) {
-          next_update_at = util::sRandom->RangeInclusive(10, 20);
         notifyChanged(false, Reason::kBulkLoadingUpdate);
+
+        if (update_current) {
+          if (shuffle_ && playlist_.size() >= 100) {
+            // Special case for shuffling a large amount of tracks. Because
+            // shuffling many tracks can be slow to wait for them all to load,
+            // we wait for 100 or so to load, then start initially with a random
+            // track from this first lot.
+            updateShuffler(true);
+            ready_ = true;
+            notifyChanged(true, Reason::kExplicitUpdate);
+            update_current = false;
+          } else if (!shuffle_ && playlist_.size() > 0) {
+            // If the queue was empty, then we want to start playing the first
+            // track without waiting for the entire queue to finish loading
+            ready_ = true;
+            notifyChanged(true, Reason::kExplicitUpdate);
+            update_current = false;
           }
+        } else {
+          // Make sure the shuffler gets updated periodically so that skipping
+          // tracks whilst we're still loading gives us the whole queue to play
+          // with.
+          updateShuffler(false);
         }
 
-      {
+        next_update_at = util::sRandom->RangeInclusive(10, 20);
-        const std::unique_lock<std::shared_mutex> lock(mutex_);
-        updateShuffler(was_queue_empty);
       }
-      notifyChanged(current_changed, Reason::kExplicitUpdate);
-    });
     }
+
+    // If we're here, then the async append must have been cancelled. Bail out
+    // immediately and rely on whatever cancelled us to reset our state. This
+    // is a little messy, but we would have to gain a lock on mutex_ to reset
+    // ourselves properly, and at the moment the only thing that can cancel us
+    // is clear().
+    appending_async_ = false;
+    appending_async_.notify_all();
+  });
 }
 
 auto TrackQueue::next() -> void {
@@ -347,17 +431,24 @@ auto TrackQueue::finish() -> void {
 }
 
 auto TrackQueue::clear() -> void {
+  if (appending_async_) {
+    cancel_appending_async_ = true;
+    appending_async_.wait(true);
+  }
+
   {
     const std::unique_lock<std::shared_mutex> lock(mutex_);
     position_ = 0;
+    ready_ = false;
+    loading_ = false;
+    pending_async_iterators_.clear();
     playlist_.clear();
     opened_playlist_.reset();
     if (shuffle_) {
       shuffle_->resize(0);
     }
+    notifyChanged(false, Reason::kTrackFinished);
   }
-
-  notifyChanged(true, Reason::kExplicitUpdate);
 }
 
 auto TrackQueue::random(bool en) -> void {
@@ -389,6 +480,16 @@ auto TrackQueue::repeatMode() const -> RepeatMode {
   return repeatMode_;
 }
 
+auto TrackQueue::isLoading() const -> bool {
+  const std::unique_lock<std::shared_mutex> lock(mutex_);
+  return loading_;
+}
+
+auto TrackQueue::isReady() const -> bool {
+  const std::unique_lock<std::shared_mutex> lock(mutex_);
+  return ready_;
+}
+
 auto TrackQueue::serialise() -> std::string {
   cppbor::Array tracks{};
   cppbor::Map encoded;

src/tangara/audio/track_queue.hpp

@@ -114,6 +114,9 @@ class TrackQueue {
   auto repeatMode(RepeatMode mode) -> void;
   auto repeatMode() const -> RepeatMode;
 
+  auto isLoading() const -> bool;
+  auto isReady() const -> bool;
+
   auto serialise() -> std::string;
   auto deserialise(const std::string&) -> void;
 
@@ -125,6 +128,7 @@ class TrackQueue {
   auto next(QueueUpdate::Reason r) -> void;
   auto goTo(size_t position) -> void;
   auto getFilepath(database::TrackId id) -> std::optional<std::string>;
+  auto appendAsync(database::TrackIterator i, bool was_empty) -> void;
 
   mutable std::shared_mutex mutex_;
 
@@ -140,6 +144,13 @@ class TrackQueue {
   std::optional<RandomIterator> shuffle_;
   RepeatMode repeatMode_;
 
+  std::atomic<bool> cancel_appending_async_;
+  std::atomic<bool> appending_async_;
+  std::list<database::TrackIterator> pending_async_iterators_;
+
+  bool loading_;
+  bool ready_;
+
   class QueueParseClient : public cppbor::ParseClient {
    public:
     QueueParseClient(TrackQueue& queue);

src/tangara/database/database.hpp

@@ -259,6 +259,7 @@ class TrackIterator {
 
   TrackIterator(const TrackIterator&) = default;
   TrackIterator& operator=(TrackIterator&& other) = default;
+  TrackIterator& operator=(const TrackIterator& other) = default;
 
   auto value() const -> std::optional<TrackId>;
   std::optional<TrackId> operator*() const { return value(); }

src/tangara/ui/ui_fsm.cpp

@@ -230,6 +230,7 @@ lua::Property UiState::sQueueRandom{false, [](const lua::LuaValue& val) {
                                       return true;
                                     }};
 lua::Property UiState::sQueueLoading{false};
+lua::Property UiState::sQueueReady{false};
 
 lua::Property UiState::sVolumeCurrentPct{
     0, [](const lua::LuaValue& val) {
@@ -446,12 +447,8 @@ void UiState::react(const audio::QueueUpdate& update) {
   sQueuePosition.setDirect(current_pos);
   sQueueRandom.setDirect(queue.random());
   sQueueRepeatMode.setDirect(queue.repeatMode());
-
+  sQueueLoading.setDirect(queue.isLoading());
-  if (update.reason == audio::QueueUpdate::Reason::kBulkLoadingUpdate) {
+  sQueueReady.setDirect(queue.isReady());
-    sQueueLoading.setDirect(true);
-  } else {
-    sQueueLoading.setDirect(false);
-  }
 }
 
 void UiState::react(const audio::PlaybackUpdate& ev) {
@@ -651,6 +648,7 @@ void Lua::entry() {
             {"repeat_mode", &sQueueRepeatMode},
             {"random", &sQueueRandom},
             {"loading", &sQueueLoading},
+            {"ready", &sQueueReady},
         });
     registry.AddPropertyModule("volume",
                                {

src/tangara/ui/ui_fsm.hpp

@@ -122,6 +122,7 @@ class UiState : public tinyfsm::Fsm<UiState> {
   static lua::Property sQueueRepeatMode;
   static lua::Property sQueueRandom;
   static lua::Property sQueueLoading;
+  static lua::Property sQueueReady;
 
   static lua::Property sVolumeCurrentPct;
   static lua::Property sVolumeCurrentDb;