Add function to skip enqueued work in thread_pool

Author: martincapello

base/task.cpp

@@ -33,10 +33,33 @@ task_token& task::start(thread_pool& pool)
   m_state = state::ENQUEUED;
   m_token.reset();
 
-  pool.execute([this] { in_worker_thread(); });
+  m_token.m_work = pool.execute([this] { in_worker_thread(); });
   return m_token;
 }
 
+bool task::try_skip(thread_pool& pool)
+{
+  bool skipped = pool.try_skip(m_token.m_work);
+  if (skipped) {
+    m_token.m_canceled = true;
+    call_finished();
+  }
+
+  return skipped;
+}
+
+void task::call_finished()
+{
+  if (m_finished) {
+    try {
+      m_finished(m_token);
+    }
+    catch (const std::exception& ex) {
+      LOG(ERROR, "Exception executing 'finished' callback: %s\n", ex.what());
+    }
+  }
+}
+
 void task::in_worker_thread()
 {
   m_state = state::RUNNING;
@@ -50,14 +73,7 @@ void task::in_worker_thread()
 
   m_state = state::FINISHED;
 
-  if (m_finished) {
-    try {
-      m_finished(m_token);
-    }
-    catch (const std::exception& ex) {
-      LOG(ERROR, "Exception executing 'finished' callback: %s\n", ex.what());
-    }
-  }
+  call_finished();
 }
 
 } // namespace base

base/task.h

@@ -27,7 +27,6 @@ class task_token {
 
   bool canceled() const { return m_canceled; }
   float progress() const { return m_progress; }
-  bool finished() const { return m_canceled || m_progress == m_progress_max; }
 
   void cancel() { m_canceled = true; }
   void set_progress(float p)
@@ -51,6 +50,7 @@ class task_token {
   std::atomic<bool> m_canceled;
   std::atomic<float> m_progress;
   float m_progress_min, m_progress_max;
+  const thread_pool::work* m_work = nullptr;
 };
 
 class task {
@@ -71,6 +71,7 @@ class task {
   void on_finished(func_t&& f) { m_finished = std::move(f); }
 
   task_token& start(thread_pool& pool);
+  bool try_skip(thread_pool& pool);
 
   bool running() const { return m_state == state::RUNNING; }
 
@@ -86,6 +87,7 @@ class task {
 
 private:
   void in_worker_thread();
+  void call_finished();
 
   std::atomic<state> m_state;
   task_token m_token;

base/thread_pool.cpp

@@ -26,12 +26,27 @@ thread_pool::~thread_pool()
   join_all();
 }
 
-void thread_pool::execute(std::function<void()>&& func)
+const thread_pool::work* thread_pool::execute(std::function<void()>&& func)
 {
+  thread_pool::work_ptr work = std::make_unique<thread_pool::work>(std::move(func));
+  const thread_pool::work* result = work.get();
   const std::unique_lock lock(m_mutex);
   ASSERT(m_running);
-  m_work.push(std::move(func));
+  m_work.push_back(std::move(work));
   m_cv.notify_one();
+  return result;
+}
+
+bool thread_pool::try_skip(const work* w)
+{
+  std::unique_lock<std::mutex> lock(m_mutex);
+  for (auto it = m_work.begin(); it != m_work.end(); ++it) {
+    if (w == it->get()) {
+      m_work.erase(it);
+      return true;
+    }
+  }
+  return false;
 }
 
 void thread_pool::wait_all()
@@ -79,9 +94,9 @@ void thread_pool::worker()
       m_cv.wait(lock, [this]() -> bool { return !m_running || !m_work.empty(); });
       running = m_running;
       if (m_running && !m_work.empty()) {
-        func = std::move(m_work.front());
+        func = std::move(m_work.front()->m_func);
         ++m_doingWork;
-        m_work.pop();
+        m_work.pop_front();
       }
     }
     try {

base/thread_pool.h

@@ -9,20 +9,37 @@
 #pragma once
 
 #include <condition_variable>
+#include <deque>
 #include <functional>
 #include <mutex>
-#include <queue>
 #include <thread>
 #include <vector>
 
 namespace base {
 
 class thread_pool {
 public:
+  class work {
+    friend class thread_pool;
+
+  public:
+    work(std::function<void()>&& func) { m_func = std::move(func); }
+
+  private:
+    std::function<void()> m_func = nullptr;
+  };
+
+  typedef std::unique_ptr<work> work_ptr;
+
   thread_pool(const size_t n);
   ~thread_pool();
 
-  void execute(std::function<void()>&& func);
+  const work* execute(std::function<void()>&& func);
+
+  // Tries to skip the work if it was not started yet, in other words, it
+  // removes the specified work from the queue if possible. Returns true if it
+  // was able to do so, or false otherwise.
+  bool try_skip(const work* w);
 
   // Waits until the queue is empty.
   void wait_all();
@@ -39,7 +56,7 @@ class thread_pool {
   std::mutex m_mutex;
   std::condition_variable m_cv;
   std::condition_variable m_cvWait;
-  std::queue<std::function<void()>> m_work;
+  std::deque<work_ptr> m_work;
   int m_doingWork;
 };