Merge pull request #419 from AdenKoperczak/max_time_step_queue_size

scwx-qt/scwx-qt.cmake

@@ -372,6 +372,7 @@ set(HDR_UTIL source/scwx/qt/util/color.hpp
              source/scwx/qt/util/q_color_modulate.hpp
              source/scwx/qt/util/q_file_buffer.hpp
              source/scwx/qt/util/q_file_input_stream.hpp
+             source/scwx/qt/util/queue_counter.hpp
              source/scwx/qt/util/time.hpp
              source/scwx/qt/util/tooltip.hpp)
 set(SRC_UTIL source/scwx/qt/util/color.cpp
@@ -385,6 +386,7 @@ set(SRC_UTIL source/scwx/qt/util/color.cpp
              source/scwx/qt/util/q_color_modulate.cpp
              source/scwx/qt/util/q_file_buffer.cpp
              source/scwx/qt/util/q_file_input_stream.cpp
+             source/scwx/qt/util/queue_counter.cpp
              source/scwx/qt/util/time.cpp
              source/scwx/qt/util/tooltip.cpp)
 set(HDR_VIEW source/scwx/qt/view/level2_product_view.hpp
@@ -692,6 +694,7 @@ target_link_libraries(scwx-qt PUBLIC Qt${QT_VERSION_MAJOR}::Widgets
                                      Qt${QT_VERSION_MAJOR}::Svg
                                      Boost::json
                                      Boost::timer
+                                     Boost::atomic
                                      QMapLibre::Core
                                      $<$<CXX_COMPILER_ID:MSVC>:opengl32>
                                      $<$<CXX_COMPILER_ID:MSVC>:SetupAPI>

scwx-qt/source/scwx/qt/manager/timeline_manager.cpp

@@ -1,6 +1,7 @@
 #include <scwx/qt/manager/timeline_manager.hpp>
 #include <scwx/qt/manager/radar_product_manager.hpp>
 #include <scwx/qt/settings/general_settings.hpp>
+#include <scwx/qt/util/queue_counter.hpp>
 #include <scwx/util/logger.hpp>
 #include <scwx/util/map.hpp>
 #include <scwx/util/time.hpp>
@@ -31,6 +32,8 @@ enum class Direction
 
 // Wait up to 5 seconds for radar sweeps to update
 static constexpr std::chrono::seconds kRadarSweepMonitorTimeout_ {5};
+// Only allow for 3 steps to be queued at any time
+static constexpr size_t kMaxQueuedSteps_ {3};
 
 class TimelineManager::Impl
 {
@@ -80,6 +83,8 @@ public:
    boost::asio::thread_pool playThreadPool_ {1};
    boost::asio::thread_pool selectThreadPool_ {1};
 
+   util::QueueCounter stepCounter_ {kMaxQueuedSteps_};
+
    std::size_t                           mapCount_ {0};
    std::string                           radarSite_ {"?"};
    std::string                           previousRadarSite_ {"?"};
@@ -256,7 +261,7 @@ void TimelineManager::AnimationStepEnd()
    if (p->viewType_ == types::MapTime::Live)
    {
       // If the selected view type is live, select the current products
-      p->SelectTime();
+      p->SelectTimeAsync();
    }
    else
    {
@@ -395,8 +400,9 @@ void TimelineManager::Impl::UpdateCacheLimit(
 {
    // Calculate the number of volume scans in the loop
    auto [startTime, endTime] = GetLoopStartAndEndTimes();
-   auto startIter = util::GetBoundedElementIterator(volumeTimes, startTime);
+   auto startIter =
-   auto endIter   = util::GetBoundedElementIterator(volumeTimes, endTime);
+      scwx::util::GetBoundedElementIterator(volumeTimes, startTime);
+   auto endIter = scwx::util::GetBoundedElementIterator(volumeTimes, endTime);
    std::size_t numVolumeScans = std::distance(startIter, endIter) + 1;
 
    // Dynamically update maximum cached volume scans to the lesser of
@@ -571,7 +577,8 @@ std::pair<bool, bool> TimelineManager::Impl::SelectTime(
    UpdateCacheLimit(radarProductManager, volumeTimes);
 
    // Find the best match bounded time
-   auto elementPtr = util::GetBoundedElementPointer(volumeTimes, selectedTime);
+   auto elementPtr =
+      scwx::util::GetBoundedElementPointer(volumeTimes, selectedTime);
 
    // The timeline is no longer live
    Q_EMIT self_->LiveStateUpdated(false);
@@ -612,6 +619,12 @@ std::pair<bool, bool> TimelineManager::Impl::SelectTime(
 
 void TimelineManager::Impl::StepAsync(Direction direction)
 {
+   // Prevent too many steps from being added to the queue
+   if (!stepCounter_.add())
+   {
+      return;
+   }
+
    boost::asio::post(selectThreadPool_,
                      [=, this]()
                      {
@@ -623,6 +636,7 @@ void TimelineManager::Impl::StepAsync(Direction direction)
                         {
                            logger_->error(ex.what());
                         }
+                        stepCounter_.remove();
                      });
 }
 

scwx-qt/source/scwx/qt/util/queue_counter.cpp

@@ -0,0 +1,49 @@
+#include <scwx/qt/util/queue_counter.hpp>
+
+#include <boost/atomic/atomic.hpp>
+
+namespace scwx::qt::util
+{
+
+class QueueCounter::Impl
+{
+public:
+   explicit Impl(size_t maxCount) : maxCount_ {maxCount} {}
+
+   const size_t          maxCount_;
+   boost::atomic<size_t> count_ {0};
+};
+
+QueueCounter::QueueCounter(size_t maxCount) :
+    p {std::make_unique<Impl>(maxCount)}
+{
+}
+
+QueueCounter::~QueueCounter() = default;
+
+bool QueueCounter::add()
+{
+   const size_t count = p->count_.fetch_add(1);
+   // Must be >= (not ==) to avoid race conditions
+   if (count >= p->maxCount_)
+   {
+      p->count_.fetch_sub(1);
+      return false;
+   }
+   else
+   {
+      return true;
+   }
+}
+
+void QueueCounter::remove()
+{
+   p->count_.fetch_sub(1);
+}
+
+bool QueueCounter::is_lock_free()
+{
+   return p->count_.is_lock_free();
+}
+
+} // namespace scwx::qt::util

scwx-qt/source/scwx/qt/util/queue_counter.hpp

@@ -0,0 +1,64 @@
+#pragma once
+
+#include <memory>
+#include <boost/atomic/atomic.hpp>
+
+namespace scwx::qt::util
+{
+
+class QueueCounter
+{
+public:
+   /**
+    * Counts the number of items in a queue, and prevents it from exceeding a
+    * count in a thread safe manor. This is lock free, assuming
+    * std::atomic<size_t> supports lock free fetch_add and fetch_sub.
+    */
+
+   /**
+    * Construct a QueueCounter with a given maximum count
+    *
+    * @param maxCount The maximum number of items in the queue
+    */
+   explicit QueueCounter(size_t maxCount);
+
+   ~QueueCounter();
+   QueueCounter(const QueueCounter&)            = delete;
+   QueueCounter(QueueCounter&&)                 = delete;
+   QueueCounter& operator=(const QueueCounter&) = delete;
+   QueueCounter& operator=(QueueCounter&&)      = delete;
+
+   /**
+    * Called before adding an item. If it returns true, it is ok to add. If it
+    * returns false, it should not be added
+    *
+    * @return true if it is ok to add, false if the queue is full
+    */
+   bool add();
+
+   /**
+    * Called when item is removed from the queue. Should only be called after a
+    * corresponding and successful call to add.
+    */
+   void remove();
+
+   /**
+    * Tells if this instance is lock free
+    *
+    * @return true if it is lock free, false otherwise
+    */
+   bool is_lock_free();
+
+   /**
+    * Tells if this class is always lock free. True if it is lock free, false
+    * otherwise
+    */
+   static constexpr bool is_always_lock_free =
+      boost::atomic<size_t>::is_always_lock_free;
+
+private:
+   class Impl;
+   std::unique_ptr<Impl> p;
+};
+
+} // namespace scwx::qt::util