fix liveness / performance bug in PooledThreadExecutor

src/aws-cpp-sdk-core/include/aws/core/utils/threading/Executor.h

@@ -8,7 +8,6 @@
 #define AWS_EXECUTOR_H
 
 #include <aws/core/Core_EXPORTS.h>
-
 #include <functional>
 
 namespace Aws

src/aws-cpp-sdk-core/include/aws/core/utils/threading/PooledThreadExecutor.h

@@ -9,7 +9,7 @@
 
 #include <aws/core/utils/memory/stl/AWSQueue.h>
 #include <aws/core/utils/memory/stl/AWSVector.h>
-#include <aws/core/utils/threading/Semaphore.h>
+#include <condition_variable>
 #include <functional>
 #include <mutex>
 #include <atomic>
@@ -57,7 +57,7 @@ namespace Aws
             private:
                 Aws::Queue<std::function<void()>*> m_tasks;
                 mutable std::mutex m_queueLock;
-                Aws::Utils::Threading::Semaphore m_sync;
+                std::condition_variable m_sync;
                 Aws::Vector<ThreadTask*> m_threadTaskHandles;
                 size_t m_poolSize = 0;
                 OverflowPolicy m_overflowPolicy = OverflowPolicy::QUEUE_TASKS_EVENLY_ACROSS_THREADS;

src/aws-cpp-sdk-core/source/utils/threading/PooledThreadExecutor.cpp

@@ -31,12 +31,16 @@ void PooledThreadExecutor::WaitUntilStopped()
         std::lock_guard<std::mutex> locker(m_queueLock);
         m_stopped = true;
     }
-    for(auto threadTask : m_threadTaskHandles)
     {
-        threadTask->StopProcessingWork();
+        std::lock_guard lock(m_queueLock);
+
+        for(auto threadTask : m_threadTaskHandles)
+        {
+            threadTask->StopProcessingWork();
+        }
     }
 
-    m_sync.ReleaseAll();
+    m_sync.notify_all();
 
     for (auto threadTask : m_threadTaskHandles)
     {
@@ -73,15 +77,13 @@ bool PooledThreadExecutor::SubmitToThread(std::function<void()>&& fn)
         m_tasks.push(fnCpy);
     }
 
-    m_sync.Release();
+    m_sync.notify_one();
 
     return true;
 }
 
 std::function<void()>* PooledThreadExecutor::PopTask()
 {
-    std::lock_guard<std::mutex> locker(m_queueLock);
-
     if (m_tasks.size() > 0)
     {
         std::function<void()>* fn = m_tasks.front();
@@ -97,6 +99,5 @@ std::function<void()>* PooledThreadExecutor::PopTask()
 
 bool PooledThreadExecutor::HasTasks() const
 {
-    std::lock_guard<std::mutex> locker(m_queueLock);
     return m_tasks.size() > 0;
 }

src/aws-cpp-sdk-core/source/utils/threading/ThreadTask.cpp

@@ -21,22 +21,23 @@ ThreadTask::~ThreadTask()
 
 void ThreadTask::MainTaskRunner()
 {
+    std::unique_lock lock(m_executor.m_queueLock);
+
     while (m_continue)
     {        
-        while (m_continue && m_executor.HasTasks())
+        if (m_executor.HasTasks())
         {      
             auto fn = m_executor.PopTask();
+            lock.unlock();
             if(fn)
             {
                 (*fn)();
                 Aws::Delete(fn);               
             }
+            lock.lock();
         }
-
-        if(m_continue)
-        {
-            m_executor.m_sync.WaitOne();
-        }
+
+        m_executor.m_sync.wait(lock, [this] { return !m_continue || m_executor.HasTasks(); });
     }
 }