Merge pull request #2 from awslabs/ThreadPool

Thread pool based executor implementation

Author: bretambrose

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

@@ -15,19 +15,22 @@
 
 #pragma once
 
-
 #include <aws/core/Core_EXPORTS.h>
-
-
-#include <functional>
 #include <aws/core/utils/memory/stl/AWSFunction.h>
+#include <aws/core/utils/memory/stl/AWSQueue.h>
+#include <aws/core/utils/memory/stl/AWSVector.h>
+#include <functional>
+#include <mutex>
+
 
 namespace Aws
 {
     namespace Utils
     {
         namespace Threading
         {
+            class ThreadTask;
+
             /**
             * Interface for implementing an Executor, to implement a custom thread execution strategy, inherit from this class
             * and override SubmitToThread().
@@ -63,7 +66,52 @@ namespace Aws
             public:
                 DefaultExecutor() {}
             protected:
-                bool SubmitToThread(std::function<void()>&&);
+                bool SubmitToThread(std::function<void()>&&) override;
+            };
+
+            enum class OverflowPolicy
+            {
+                QUEUE_TASKS_EVENLY_ACCROSS_THREADS,
+                REJECT_IMMEDIATELY
+            };
+
+            /**
+            * Thread Pool Executor implementation.
+            */
+            class AWS_CORE_API PooledThreadExecutor : public Executor
+            {
+            public:
+                PooledThreadExecutor(size_t poolSize, OverflowPolicy overflowPolicy = OverflowPolicy::QUEUE_TASKS_EVENLY_ACCROSS_THREADS);
+                ~PooledThreadExecutor();
+
+                /**
+                * Rule of 5 stuff.
+                * Don't copy or move
+                */
+                PooledThreadExecutor(const PooledThreadExecutor&) = delete;
+                PooledThreadExecutor& operator =(const PooledThreadExecutor&) = delete;
+                PooledThreadExecutor(PooledThreadExecutor&&) = delete;
+                PooledThreadExecutor& operator =(PooledThreadExecutor&&) = delete;
+
+            protected:
+                bool SubmitToThread(std::function<void()>&&) override;
+
+            private:
+                Aws::Queue<std::function<void()>*> m_tasks;
+                std::mutex m_queueLock;
+                std::mutex m_syncPointLock;
+                std::condition_variable m_syncPoint;
+                Aws::Vector<ThreadTask*> m_threadTaskHandles;
+                size_t m_poolSize;
+                OverflowPolicy m_overflowPolicy;
+
+                /**
+                 * Once you call this, you are responsible for freeing the memory pointed to by task.
+                 */
+                std::function<void()>* PopTask();
+                bool HasTasks();
+
+                friend class ThreadTask;
             };
 
 

aws-cpp-sdk-core/include/aws/core/utils/threading/ThreadTask.h

@@ -0,0 +1,58 @@
+/*
+* Copyright 2010-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+*
+* Licensed under the Apache License, Version 2.0 (the "License").
+* You may not use this file except in compliance with the License.
+* A copy of the License is located at
+*
+*  http://aws.amazon.com/apache2.0
+*
+* or in the "license" file accompanying this file. This file is distributed
+* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+* express or implied. See the License for the specific language governing
+* permissions and limitations under the License.
+*/
+
+#pragma once
+
+#include <aws/core/Core_EXPORTS.h>
+#include <functional>
+#include <thread>
+#include <atomic>
+
+namespace Aws
+{
+    namespace Utils
+    {
+        namespace Threading
+        {
+            class PooledThreadExecutor;
+
+            class AWS_CORE_API ThreadTask
+            {
+            public:
+                ThreadTask(PooledThreadExecutor& executor);
+                ~ThreadTask();
+
+                /**
+                * Rule of 5 stuff.
+                * Don't copy or move
+                */
+                ThreadTask(const ThreadTask&) = delete;
+                ThreadTask& operator =(const ThreadTask&) = delete;
+                ThreadTask(ThreadTask&&) = delete;
+                ThreadTask& operator =(ThreadTask&&) = delete;
+
+                void StopProcessingWork();                
+
+            protected:
+                void MainTaskRunner();
+
+            private:                
+                std::atomic<bool> m_continue;
+                PooledThreadExecutor& m_executor;
+                std::thread m_thread;
+            };
+        }
+    }
+}

aws-cpp-sdk-core/source/utils/logging/AWSLogging.cpp

@@ -15,7 +15,6 @@
 
 
 #include <aws/core/utils/logging/AWSLogging.h>
-
 #include <aws/core/utils/logging/LogSystemInterface.h>
 
 #include <memory>

aws-cpp-sdk-core/source/utils/threading/Executor.cpp

@@ -14,15 +14,96 @@
   */
 
 #include <aws/core/utils/threading/Executor.h>
-
-
+#include <aws/core/utils/threading/ThreadTask.h>
 #include <thread>
 
+static const char* POOLED_CLASS_TAG = "PooledThreadExecutor";
+
 using namespace Aws::Utils::Threading;
 
 bool DefaultExecutor::SubmitToThread(std::function<void()>&&  fx)
 {
     std::thread t(fx);
     t.detach();
     return true;
-}
+}
+
+PooledThreadExecutor::PooledThreadExecutor(size_t poolSize, OverflowPolicy overflowPolicy) :
+    m_poolSize(poolSize), m_overflowPolicy(overflowPolicy)
+{
+    for (size_t index = 0; index < m_poolSize; ++index)
+    {
+        m_threadTaskHandles.push_back(Aws::New<ThreadTask>(POOLED_CLASS_TAG, *this));
+    }
+}
+
+PooledThreadExecutor::~PooledThreadExecutor()
+{
+    for(auto threadTask : m_threadTaskHandles)
+    {
+        threadTask->StopProcessingWork();       
+    }
+
+    m_syncPoint.notify_all();
+
+    for (auto threadTask : m_threadTaskHandles)
+    {
+        Aws::Delete(threadTask);
+    }
+
+    while(m_tasks.size() > 0)
+    {
+        std::function<void()>* fn = m_tasks.front();
+        m_tasks.pop();
+
+        if(fn)
+        {
+            Aws::Delete(fn);
+        }
+    }
+
+}
+
+bool PooledThreadExecutor::SubmitToThread(std::function<void()>&& fn)
+{
+    //avoid the need to do copies inside the lock. Instead lets do a pointer push.
+    std::function<void()>* fnCpy = Aws::New<std::function<void()>>(POOLED_CLASS_TAG, std::forward<std::function<void()>>(fn));
+
+    {
+        std::lock_guard<std::mutex> locker(m_queueLock);
+
+        if (m_overflowPolicy == OverflowPolicy::REJECT_IMMEDIATELY && m_poolSize >= m_tasks.size())
+        {
+            return false;
+        }
+
+        m_tasks.push(fnCpy);
+    }
+
+    m_syncPoint.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();
+        if (fn)
+        {           
+            m_tasks.pop();
+            return fn;
+        }
+    }
+
+    return nullptr;
+}
+
+bool PooledThreadExecutor::HasTasks()
+{
+    std::lock_guard<std::mutex> locker(m_queueLock);
+    return m_tasks.size() > 0;
+}

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

@@ -0,0 +1,57 @@
+/*
+* Copyright 2010-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+*
+* Licensed under the Apache License, Version 2.0 (the "License").
+* You may not use this file except in compliance with the License.
+* A copy of the License is located at
+*
+*  http://aws.amazon.com/apache2.0
+*
+* or in the "license" file accompanying this file. This file is distributed
+* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+* express or implied. See the License for the specific language governing
+* permissions and limitations under the License.
+*/
+
+#include <aws/core/utils/threading/ThreadTask.h>
+#include <aws/core/utils/threading/Executor.h>
+
+using namespace Aws::Utils;
+using namespace Aws::Utils::Threading;
+
+ThreadTask::ThreadTask(PooledThreadExecutor& executor) : m_continue(true), m_executor(executor), m_thread(std::bind(&ThreadTask::MainTaskRunner, this))
+{
+}
+
+ThreadTask::~ThreadTask()
+{
+    StopProcessingWork();
+    m_thread.join();
+}
+
+void ThreadTask::MainTaskRunner()
+{
+    while (m_continue)
+    {        
+        while (m_continue && m_executor.HasTasks())
+        {      
+            auto fn = m_executor.PopTask();
+            if(fn)
+            {
+                (*fn)();
+                Aws::Delete(fn);               
+            }
+        }
+     
+        std::unique_lock<std::mutex> locker(m_executor.m_syncPointLock);
+        if(m_continue)
+        {
+            m_executor.m_syncPoint.wait(locker);    
+        }
+    }
+}
+
+void ThreadTask::StopProcessingWork()
+{
+    m_continue = false;
+}

aws-cpp-sdk-dynamodb-integration-tests/TableOperationTest.cpp

@@ -18,13 +18,15 @@
 #include <aws/core/client/CoreErrors.h>
 #include <aws/core/auth/AWSCredentialsProviderChain.h>
 #include <aws/core/http/HttpTypes.h>
-#include <aws/dynamodb/DynamoDBClient.h>
-#include <aws/dynamodb/DynamoDBErrors.h>
+#include <aws/core/utils/memory/AWSMemory.h>
 #include <aws/core/utils/UnreferencedParam.h>
 #include <aws/core/utils/Outcome.h>
 #include <aws/core/utils/memory/stl/AWSSet.h>
 #include <aws/core/utils/memory/stl/AWSStringStream.h>
 #include <aws/core/utils/ratelimiter/DefaultRateLimiter.h>
+#include <aws/core/utils/threading/Executor.h>
+#include <aws/dynamodb/DynamoDBClient.h>
+#include <aws/dynamodb/DynamoDBErrors.h>
 #include <aws/dynamodb/model/CreateTableRequest.h>
 #include <aws/dynamodb/model/DeleteTableRequest.h>
 #include <aws/dynamodb/model/DescribeTableRequest.h>
@@ -159,6 +161,7 @@ class TableOperationTest : public ::testing::Test {
         config.readRateLimiter = m_limiter;
         config.writeRateLimiter = m_limiter;
         config.httpLibOverride = transferType;
+        config.executor = Aws::MakeShared<Aws::Utils::Threading::PooledThreadExecutor>(ALLOCATION_TAG, 25);
 
         //to test proxy functionality, uncomment the next two lines.
         //config.proxyHost = "localhost";