feat(runqueue): Add id for tracking function execution (#401)

* feat(runqueue): Add id for tracking function execution
* Update runqueue to generate and store an ID associated with each function, returning the id when functions are added to the queue
* Allow querying of the ids currently in the queue or running
* Allow removal of functions by id from the queue
* Update example to test and showcase new APIs

Allows callers to track if their function is currently in the queue or has finished executing (no longer in the queue or running). Also allows callers to remove the function from the queue if it is no longer needed to run.

* Build and run `runqueue/example` on QtPy ESP32s3 and ensure it works.

* update readme

* update example and logging

Author: finger563

components/runqueue/CMakeLists.txt

@@ -2,3 +2,6 @@ idf_component_register(
   INCLUDE_DIRS "include"
   SRC_DIRS "src"
   REQUIRES base_component task)
+
+# add compiler flags to enable rtti
+target_compile_options(${COMPONENT_LIB} PRIVATE -frtti)

components/runqueue/example/README.md

@@ -23,4 +23,5 @@ See the Getting Started Guide for full steps to configure and use ESP-IDF to bui
 
 ## Example Output
 
-![CleanShot 2025-03-06 at 14 38 32](https://github.com/user-attachments/assets/57dbbe6c-dc46-448b-b86f-750a19142658)
+![CleanShot 2025-03-07 at 22 13 10@2x](https://github.com/user-attachments/assets/dedff19e-c40f-4451-a0e1-94e2b4a8835f)
+

components/runqueue/example/main/runqueue_example.cpp

@@ -41,24 +41,90 @@ extern "C" void app_main(void) {
     auto task0 = std::bind(task, 0);
     auto task1 = std::bind(task, 1);
     auto task2 = std::bind(task, 2);
-    runqueue.add_function(
+    auto task3 = std::bind(task, 3);
+    auto task0_id = runqueue.add_function(
         task0,
         espp::RunQueue::MIN_PRIORITY); // this will run first because it was first to be added
-    runqueue.add_function(task1, espp::RunQueue::MIN_PRIORITY + 1); // this will run third
-    runqueue.add_function(task2, espp::RunQueue::MIN_PRIORITY + 2); // this will run second
-    runqueue.add_function(stop_task, espp::RunQueue::MIN_PRIORITY); // this will run last
+    auto task1_id =
+        runqueue.add_function(task1, espp::RunQueue::MIN_PRIORITY + 1); // this will run fourth
+    auto task3_id =
+        runqueue.add_function(task3, espp::RunQueue::MAX_PRIORITY); // this will run second
+    auto task2_id =
+        runqueue.add_function(task2, espp::RunQueue::MIN_PRIORITY + 2); // this will run third
+    auto stop_task_id =
+        runqueue.add_function(stop_task, espp::RunQueue::MIN_PRIORITY); // this will run last
     logger.info("All tasks scheduled!");
 
+    // print our the task ids
+    logger.info("Task 0 id: {}", task0_id);
+    logger.info("Task 1 id: {}", task1_id);
+    logger.info("Task 2 id: {}", task2_id);
+    logger.info("Task 3 id: {}", task3_id);
+    logger.info("Stop task id: {}", stop_task_id);
+
     // check the API for queue_size
     logger.info("Queue size: {}", runqueue.queue_size());
 
     // check the API for is_running
     logger.info("RunQueue is running: {}", runqueue.is_running());
 
+    // check the API for is_function_queued
+    logger.info("Task 0 is queued: {}", runqueue.is_function_queued(task0_id));
+    logger.info("Task 1 is queued: {}", runqueue.is_function_queued(task1_id));
+    logger.info("Task 2 is queued: {}", runqueue.is_function_queued(task2_id));
+    logger.info("Task 3 is queued: {}", runqueue.is_function_queued(task3_id));
+    logger.info("Stop task is queued: {}", runqueue.is_function_queued(stop_task_id));
+
+    // check the API for removing an invalid id
+    if (runqueue.remove_function(espp::RunQueue::INVALID_ID)) {
+      logger.error("Incorrectly Removed invalid id!");
+    } else {
+      logger.info("Correctly failed to remove invalid id!");
+    }
+
+    // check the api for removing a valid but non-existent id
+    if (runqueue.remove_function(999)) {
+      logger.error("Incorrectly Removed non-existent id!");
+    } else {
+      logger.info("Correctly failed to remove non-existent id!");
+    }
+
+    // check the api for removing the currently running id (which should fail)
+    if (runqueue.remove_function(runqueue.get_running_id().value())) {
+      logger.error("Incorrectly Removed currently running id!");
+    } else {
+      logger.info("Correctly failed to remove currently running id!");
+    }
+
+    // NOTE: in the next example (below) we'll check removing a valid ID
+
+    // check the API for get_queued_ids(bool include_running)
+    auto queued_ids = runqueue.get_queued_ids(true);
+    logger.info("Queued ids (including running): {}", queued_ids);
+
+    // check the API for get_running_id
+    auto running_id = runqueue.get_running_id();
+    logger.info("Running id: {}", running_id);
+
     logger.info("Waiting for stop task to complete...");
     std::unique_lock lock(done_mutex);
     done_cv.wait(lock, [&done] { return done; });
 
+    // check the API for get_running_id again (should return nullopt)
+    running_id = runqueue.get_running_id();
+    logger.info("Running id: {}", running_id);
+
+    // check the api for get_queued_ids again
+    queued_ids = runqueue.get_queued_ids(true);
+    logger.info("Queued ids (including running): {}", queued_ids);
+
+    // check the api for is_function_queued again
+    logger.info("Task 0 is queued: {}", runqueue.is_function_queued(task0_id));
+    logger.info("Task 1 is queued: {}", runqueue.is_function_queued(task1_id));
+    logger.info("Task 2 is queued: {}", runqueue.is_function_queued(task2_id));
+    logger.info("Task 3 is queued: {}", runqueue.is_function_queued(task3_id));
+    logger.info("Stop task is queued: {}", runqueue.is_function_queued(stop_task_id));
+
     logger.info("All tasks done!");
     //! [runqueue example]
   }
@@ -99,6 +165,8 @@ extern "C" void app_main(void) {
         espp::RunQueue::MIN_PRIORITY); // this will run first because it was first to be added
     runqueue0.add_function(task1, espp::RunQueue::MIN_PRIORITY + 1); // this will run third
     runqueue0.add_function(task2, espp::RunQueue::MIN_PRIORITY + 2); // this will run second
+    auto id_to_remove = runqueue0.add_function(stop_task,
+                                               espp::RunQueue::MIN_PRIORITY); // this will run last
 
     runqueue1.add_function(
         task0,
@@ -108,6 +176,13 @@ extern "C" void app_main(void) {
     runqueue1.add_function(stop_task, espp::RunQueue::MIN_PRIORITY); // this will run last
     logger.info("All tasks scheduled!");
 
+    // now remove the stop task from runqueue0
+    if (!runqueue0.remove_function(id_to_remove)) {
+      logger.error("Failed to remove task from runqueue0!");
+    } else {
+      logger.info("Removed task from runqueue0!");
+    }
+
     logger.info("Waiting for stop task to complete...");
     std::unique_lock lock(done_mutex);
     done_cv.wait(lock, [&done] { return done; });

components/runqueue/include/runqueue.hpp

@@ -63,18 +63,25 @@ class RunQueue : public espp::BaseComponent {
   using Priority = std::uint8_t;
 
   /// The minimum priority a function can have.
-  static constexpr int MIN_PRIORITY = std::numeric_limits<Priority>::min();
+  static constexpr Priority MIN_PRIORITY = std::numeric_limits<Priority>::min();
 
   /// The maximum priority a function can have.
-  static constexpr int MAX_PRIORITY = std::numeric_limits<Priority>::max();
+  static constexpr Priority MAX_PRIORITY = std::numeric_limits<Priority>::max();
+
+  /// The type used to represent the id of a function.
+  using Id = std::uint32_t;
+
+  /// The invalid id value. This is used to know if an id is valid or not.
+  static constexpr Id INVALID_ID = std::numeric_limits<Id>::min();
 
   /// A function that takes no arguments and returns void.
   using Function = std::function<void(void)>;
   // typedef std::function<void(void)> Function;
 
   /// A pair of a priority and a function.
   struct PriorityFunction {
-    Priority priority; ///< The priority of the function.
+    Priority priority; ///< The priority of the function. Lower values have lower priority.
+    Id id;             ///< The id of the function. Can be provided or auto-generated.
     Function function; ///< The function.
   };
 
@@ -94,6 +101,14 @@ class RunQueue : public espp::BaseComponent {
     return lhs.priority > rhs.priority;
   }
 
+  /// Equality operator for PriorityFunction.
+  /// \param lhs The left hand side of the comparison.
+  /// \param rhs The right hand side of the comparison.
+  /// \return True if the left and right hand side have the same priority and id.
+  friend bool operator==(const PriorityFunction &lhs, const PriorityFunction &rhs) {
+    return lhs.priority == rhs.priority && lhs.id == rhs.id;
+  }
+
   /// Configuration struct for the RunQueue
   struct Config {
     bool auto_start = true;                  ///< Whether the RunQueue should start automatically.
@@ -121,12 +136,57 @@ class RunQueue : public espp::BaseComponent {
   void start();
 
   /// Stop the run queue.
+  /// \note This must wait until the currently running function (if any) has
+  ///       completed before stopping the run queue.
   void stop();
 
   /// Add a function to the queue.
   /// \param function The function to add.
   /// \param priority The priority of the function. Defaults to MIN_PRIORITY.
-  void add_function(const Function &function, Priority priority = MIN_PRIORITY);
+  /// \return The id of the function. This will be auto-generated and can be
+  ///         used to query or remove the function from the queue.
+  Id add_function(const Function &function, Priority priority = MIN_PRIORITY);
+
+  /// Remove a function from the queue.
+  /// \param id The id of the function to remove. Cannot be INVALID_ID.
+  /// \return True if the function was removed.
+  /// \note This will not remove or stop the currently running function (if
+  ///       any).
+  /// \note This will return false if the id is INVALID_ID.
+  /// \note This will return false if the id is the id of the currently running
+  ///       function.
+  bool remove_function(Id id);
+
+  /// Check if a function is queued or running.
+  /// \param id The id of the function to check. Cannot be INVALID_ID.
+  /// \return True if the function is queued or is currently running.
+  /// \note This will return false if the the id is INVALID_ID.
+  bool is_function_queued(Id id);
+
+  /// Remove all functions from the queue.
+  /// \note This will not remove or stop the currently running function (if
+  ///       any).
+  void clear_queue();
+
+  /// Get the ids of all functions in the queue in order of priority.
+  /// \param include_running Whether to include the id of the currently running
+  ///                        function (if any) in the returned vector.
+  /// \return A vector of the ids of all functions in the queue, optionally
+  ///         including the id of the currently running function. Maybe empty if
+  ///         there are no functions queued or running.
+  /// \note The vector will be in order of priority, with the highest priority
+  ///       function at the end of the vector. This means that if include_running
+  ///       is true, the id of the currently running function will be the last
+  ///       element in the vector.
+  std::vector<Id> get_queued_ids(bool include_running = false);
+
+  /// Get the id of the currently running function.
+  /// \return The id of the currently running function, or std::nullopt if no
+  ///         function is currently running.
+  /// \note This may return nullopt if the currently running function has
+  ///       completed but the runner task has not yet fetched the next function
+  ///       from the queue.
+  std::optional<Id> get_running_id();
 
 protected:
   /// Manage the run queue.
@@ -140,6 +200,8 @@ class RunQueue : public espp::BaseComponent {
   bool task_fn(std::mutex &m, std::condition_variable &cv, bool &task_notified);
 
   std::unique_ptr<espp::Task> runner_;
+  std::atomic<Id> id_counter_{INVALID_ID};
+  std::atomic<Id> running_id_{INVALID_ID};
   std::mutex queue_mutex_;
   std::condition_variable queue_cv_;
   bool queue_notified_ = false;

components/runqueue/src/runqueue.cpp

@@ -36,12 +36,89 @@ void RunQueue::stop() {
   runner_->stop();
 }
 
-void RunQueue::add_function(const Function &function, Priority priority) {
-  logger_.debug("Adding function to queue with priority: {}", priority);
+RunQueue::Id RunQueue::add_function(const Function &function, Priority priority) {
+  // generate an Id for this function
+  Id id = ++id_counter_; // pre-increment so the first id is 1
+  if (id == INVALID_ID) {
+    // if the id is min, then increment again. this should be rare, and only
+    // happen if the id_counter_ overflows
+    id = ++id_counter_;
+  }
   std::unique_lock lock(queue_mutex_);
-  priority_function_queue_.insert({priority, function});
+  priority_function_queue_.insert({priority, id, function});
   // notify the queue so that the runner wakes up and runs the function
   queue_cv_.notify_all();
+  return id;
+}
+
+bool RunQueue::remove_function(RunQueue::Id id) {
+  // return false if the id is invalid
+  if (id == INVALID_ID) {
+    logger_.debug("Cannot remove function with id: {}, it is invalid", id);
+    return false;
+  }
+  // return false if the function is the one currently running, as we cannot
+  // remove the running function
+  if (id == running_id_) {
+    logger_.debug("Cannot remove function with id: {}, it is running", id);
+    return false;
+  }
+  logger_.debug("Removing function from queue with id: {}", id);
+  std::unique_lock lock(queue_mutex_);
+  auto it = std::find_if(priority_function_queue_.begin(), priority_function_queue_.end(),
+                         [&](const auto &pf) { return pf.id == id; });
+  if (it != priority_function_queue_.end()) {
+    priority_function_queue_.erase(it);
+    return true;
+  }
+  logger_.debug("Function with id: {} not found in queue", id);
+  return false;
+}
+
+bool RunQueue::is_function_queued(RunQueue::Id id) {
+  // return false if the id is invalid
+  if (id == INVALID_ID) {
+    return false;
+  }
+  // return true if the function is the one currently running
+  if (id == running_id_) {
+    return true;
+  }
+  // otherwise, check if the function is in the queue
+  std::unique_lock lock(queue_mutex_);
+  auto it = std::find_if(priority_function_queue_.begin(), priority_function_queue_.end(),
+                         [&](const auto &pf) { return pf.id == id; });
+  // return whether or not the function was found
+  return it != priority_function_queue_.end();
+}
+
+void RunQueue::clear_queue() {
+  logger_.debug("Clearing queue");
+  std::unique_lock lock(queue_mutex_);
+  priority_function_queue_.clear();
+}
+
+std::vector<RunQueue::Id> RunQueue::get_queued_ids(bool include_running) {
+  std::vector<Id> ids;
+  // Note: the vector will be in order of priority, with the highest priority
+  // function at the end of the vector
+  {
+    std::unique_lock lock(queue_mutex_);
+    for (const auto &pf : priority_function_queue_) {
+      ids.push_back(pf.id);
+    }
+  }
+  if (include_running && running_id_ != INVALID_ID) {
+    ids.push_back(running_id_);
+  }
+  return ids;
+}
+
+std::optional<RunQueue::Id> RunQueue::get_running_id() {
+  if (running_id_ != INVALID_ID) {
+    return running_id_;
+  }
+  return std::nullopt;
 }
 
 bool RunQueue::manage_queue() {
@@ -62,10 +139,16 @@ bool RunQueue::manage_queue() {
     priority_function_queue_.erase(std::next(top).base());
   }
 
+  // set the running id
+  running_id_ = highest_priority_function.id;
+
   logger_.debug("Running function with priority: {}", highest_priority_function.priority);
   // run the function
   highest_priority_function.function();
 
+  // clear the running id
+  running_id_ = INVALID_ID;
+
   // return whether or not there are more functions in the queue
   std::unique_lock lock(queue_mutex_);
   return !priority_function_queue_.empty();