ResourceManager.h

 
#pragma once
 
#include <aws/core/utils/memory/stl/AWSVector.h>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <cassert>
 
namespace Aws
{
    namespace Utils
    {
        template< typename RESOURCE_TYPE>
        class ExclusiveOwnershipResourceManager
        {
        public:
            ExclusiveOwnershipResourceManager() : m_shutdown(false) {}
 
            RESOURCE_TYPE Acquire()
            {
                std::unique_lock<std::mutex> locker(m_queueLock);
                while(!m_shutdown.load() && m_resources.size() == 0)
                {
                    m_semaphore.wait(locker, [&](){ return m_shutdown.load() || m_resources.size() > 0; });
                }
 
                assert(!m_shutdown);
 
                RESOURCE_TYPE resource = m_resources.back();
                m_resources.pop_back();
 
                return resource;
            }
 
            RESOURCE_TYPE TryAcquire(typename std::enable_if<std::is_pointer<RESOURCE_TYPE>::value >::type* = 0)
            {
                std::unique_lock<std::mutex> locker(m_queueLock);
                while(!m_shutdown.load() && m_resources.size() == 0)
                {
                    m_semaphore.wait(locker, [&](){ return m_shutdown.load() || m_resources.size() > 0; });
                }
 
                if (m_shutdown) {
                    return nullptr;
                }
 
                RESOURCE_TYPE resource = m_resources.back();
                m_resources.pop_back();
 
                return resource;
            }
 
            template <typename std::enable_if<std::is_pointer<RESOURCE_TYPE>::value>::type* = nullptr>
            RESOURCE_TYPE TryAcquire(const uint64_t timeoutMs) {
              std::unique_lock<std::mutex> locker(m_queueLock);
              bool hasResource = m_shutdown.load() || !m_resources.empty();
              if (!hasResource) {
                hasResource = m_semaphore.wait_for(locker, std::chrono::milliseconds(timeoutMs),
                                                   [&]() { return m_shutdown.load() || !m_resources.empty(); });
              }
 
              if (m_shutdown || !hasResource) {
                return nullptr;
              }
 
              RESOURCE_TYPE resource = m_resources.back();
              m_resources.pop_back();
 
              return resource;
            }
 
            bool HasResourcesAvailable()
            {
                std::lock_guard<std::mutex> locker(m_queueLock);
                return m_resources.size() > 0 && !m_shutdown.load();
            }
 
            void Release(RESOURCE_TYPE resource)
            {
                std::unique_lock<std::mutex> locker(m_queueLock);
                m_resources.push_back(resource);
                locker.unlock();
                m_semaphore.notify_one();
            }
 
            void PutResource(RESOURCE_TYPE resource)
            {
                m_resources.push_back(resource);
            }
 
            Aws::Vector<RESOURCE_TYPE> ShutdownAndWait(size_t resourceCount)
            {
                std::unique_lock<std::mutex> locker(m_queueLock);
                m_shutdown = true;
                m_semaphore.notify_all();
 
                //wait for all acquired resources to be released.
                while (m_resources.size() < resourceCount)
                {
                    m_semaphore.wait(locker, [&]() { return m_resources.size() == resourceCount; });
                }
 
                Aws::Vector<RESOURCE_TYPE> resources{std::move(m_resources)};
                m_semaphore.notify_one();
                return resources;
            }
 
        private:
            Aws::Vector<RESOURCE_TYPE> m_resources;
            std::mutex m_queueLock;
            std::condition_variable m_semaphore;
            std::atomic<bool> m_shutdown;
        };
    }
}