resourcePool.hpp

/*
  ANCH Framework: ANother C++ Hack is a C++ framework based on C++11 standard
  Copyright (C) 2012 Vincent Lachenal

  This file is part of ANCH Framework.

  ANCH Framework is free software: you can redistribute it and/or modify
  it under the terms of the GNU Lesser General Public License as published by
  the Free Software Foundation, either version 3 of the License, or
  (at your option) any later version.

  ANCH Framework is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public License
  along with ANCH Framework.  If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _ANCH_RES_POOL_H_
#define _ANCH_RES_POOL_H_

#include <memory>
#include <mutex>
#include <chrono>
#include <condition_variable>
#include <deque>
#include <atomic>


namespace anch {

  class TimeoutException: std::exception {

    // Attributes +
  private:
    std::chrono::milliseconds _timeout;

    std::string _msg;
    // Attributes -

    // Constructors +
  public:
    TimeoutException(const std::chrono::milliseconds& timeout, const std::string msg) noexcept: std::exception(),
      _timeout(timeout),
      _msg(msg) {
    }
    // Constructors -

    // Destructor +
  public:
    virtual ~TimeoutException() {
      // nothing to do
    }
    // Destructor -

    // Accessors +
  public:
    virtual const char* what() const noexcept {
      return _msg.data();
    }

    inline const std::chrono::milliseconds& getTimeout() const noexcept {
      return _timeout;
    }
    // Accessors -

  };

  template<typename T, typename C, std::shared_ptr<T>(*make_ptr)(const C&) = std::make_shared<T> >
  class ResourcePool {

    class PoolableResource {

      // Attributes +
    private:
      ResourcePool<T, C, make_ptr>& _pool;

      std::shared_ptr<T> _ptr;
      // Attributes -

      // Constructors +
    public:
      PoolableResource(ResourcePool<T, C, make_ptr>& pool, std::shared_ptr<T> ptr): _pool(pool), _ptr(ptr) {
        // Nothing to do
      }
      // Constructors -

      // Destructor +
    public:
      virtual ~PoolableResource() {
        if(_ptr.get()->isValid()) {
          _pool.returnResource(_ptr);
        } else {
          _pool.invalidateResource(_ptr);
        }
      }
      // Destructor -

      // Accessors +
    public:
      inline T& get() {
        return *(_ptr.get());
      }
      // Accessors -
    };

    // Attributes +
  private:
    std::mutex _mutex;

    std::condition_variable _wait;

    std::mutex _waitex;

    std::deque<std::shared_ptr<T> > _availables;

    std::size_t _maxSize;

    std::atomic_size_t _used;

    C _config;

    std::chrono::milliseconds _timeout;
    // Attributes -

    // Constructors +
  public:
    ResourcePool(const C& config,
                 std::size_t maxSize,
                 std::size_t initialiSize = 0,
                 std::chrono::milliseconds timeout = std::chrono::milliseconds(100)):
      _mutex(),
      _wait(),
      _waitex(),
      _availables(),
      _maxSize(maxSize),
      _used(0),
      _config(config),
      _timeout(timeout) {
      for(std::size_t i = 0 ; i < initialiSize && i < maxSize ; ++i) {
        try {
          std::shared_ptr<T> ptr = make_ptr(_config);
          _availables.push_back(ptr);
        } catch(...) {
          // Nothing to do => resources will be instanciated later ... or not
        }
      }
    }

    ResourcePool(const ResourcePool&) = delete;
    // Constructors -

    // Destructor +
  public:
    virtual ~ResourcePool() {
      while(!_availables.empty()) {
        _availables.front().reset();
        _availables.pop_front();
      }
    }
    // Destructor -

    // Methods +
  public:
    PoolableResource borrowResource() {
      _mutex.lock();
      std::shared_ptr<T> ptr;
      if(_availables.empty()) {
        if(_used.load() < _maxSize) {
          try {
            ptr = make_ptr(_config);
          } catch(...) {
            _mutex.unlock();
            throw;
          }
          _used.fetch_add(1);
          _mutex.unlock();

        } else {
          _mutex.unlock();
          std::unique_lock<std::mutex> lock(_waitex);
          if(_wait.wait_for(lock, _timeout, [this](){return !this->_availables.empty();})) {
            _mutex.lock();
            ptr = _availables.front();
            _availables.pop_front();
            _used.fetch_add(1);
            _mutex.unlock();
          } else {
            throw TimeoutException(_timeout, "Resource could not be retrieved.");
          }
        }

      } else {
        ptr = _availables.front();
        _availables.pop_front();
        _used.fetch_add(1);
        _mutex.unlock();
      }
      return PoolableResource(*this, ptr);
    }

  private:
    void returnResource(std::shared_ptr<T> res) {
      std::lock_guard<std::mutex> lock(_mutex);
      _availables.push_back(res);
      _used.fetch_sub(1);
      _wait.notify_one();
    }

    void invalidateResource(std::shared_ptr<T> res) {
      res.reset();
      _used.fetch_sub(1);
      try {
        _availables.push_back(make_ptr(_config));
        _wait.notify_one();
      } catch(...) {
        // Nothing to do => until resource will be instanciate other waiters will fail on timeout
      }
    }
    // Methods -

    // Accessors +
  public:
    inline void setTimeout(const std::chrono::milliseconds timeout) {
      _timeout = timeout;
    }
    // Accessors -

  };

}

#endif // _ANCH_RES_POOL_H_