Merge pull request #6413 from ygorshenin/fix-possible-worker-thread-deadlock

[base] Fixed possible deadlock on WorkerThread shutdown.

base/base_tests/worker_thread_tests.cpp

@@ -3,6 +3,7 @@
 #include "base/worker_thread.hpp"
 
 #include <condition_variable>
+#include <future>
 #include <mutex>
 
 using namespace base;
@@ -67,4 +68,22 @@ UNIT_TEST(WorkerThread_SimpleFlush)
   }
   TEST_EQUAL(value, 1024, ());
 }
+
+UNIT_TEST(WorkerThread_PushFromPendingTask)
+{
+  // promise - future pair is used as a socketpair here to pass a
+  // signal from the main thread to the worker thread.
+  promise<void> p;
+  auto f = p.get_future();
+
+  WorkerThread thread;
+  bool const rv = thread.Push([&f, &thread]() {
+    f.get();
+    bool const rv = thread.Push([]() { TEST(false, ("This task should not be executed")); });
+    TEST(!rv, ());
+  });
+  TEST(rv, ());
+  thread.Shutdown(WorkerThread::Exit::ExecPending);
+  p.set_value();
+}
 }  // namespace

base/worker_thread.cpp

@@ -20,14 +20,12 @@ void WorkerThread::ProcessTasks()
 {
   queue<Task> pending;
 
-  unique_lock<mutex> lk(m_mu, defer_lock);
-
   while (true)
   {
     Task task;
 
     {
-      lk.lock();
+      unique_lock<mutex> lk(m_mu);
       m_cv.wait(lk, [this]() { return m_shutdown || !m_queue.empty(); });
 
       if (m_shutdown)
@@ -46,7 +44,6 @@ void WorkerThread::ProcessTasks()
       CHECK(!m_queue.empty(), ());
       task = move(m_queue.front());
       m_queue.pop();
-      lk.unlock();
     }
 
     task();