////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2016 by EMC Corporation, 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.
/// You may obtain a copy of the License at
///
///     http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License 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.
///
/// Copyright holder is EMC Corporation
///
/// @author Andrey Abramov
/// @author Vasiliy Nabatchikov
////////////////////////////////////////////////////////////////////////////////

#include <array>

#include "gtest/gtest.h"
#include "utils/object_pool.hpp"
#include "utils/thread_utils.hpp"

using namespace std::chrono_literals;
namespace irs {

// Convenient class storing value and associated read-write lock
template<typename T>
class async_value {
 public:
  using value_type = T;

  template<typename... Args>
  explicit async_value(Args&&... args) : value_{std::forward<Args>(args)...} {}

  const value_type& value() const noexcept { return value_; }

  value_type& value() noexcept { return value_; }

  auto lock_read() { return std::shared_lock{lock_}; }

  auto lock_write() { return std::unique_lock{lock_}; }

 protected:
  value_type value_;
  std::shared_mutex lock_;
};

// A fixed size pool of objects.
// if the pool is empty then a new object is created via make(...)
// if an object is available in a pool then in is returned and no
// longer tracked by the pool
// when the object is released it is placed back into the pool if
// space in the pool is available
// pool may be safely destroyed even there are some produced objects
// alive.
// Object 'ptr' that evaluate to false when returnd back into the pool
// will be discarded instead.
template<typename T>
class unbounded_object_pool_volatile : public unbounded_object_pool_base<T> {
 private:
  struct generation {
    explicit generation(unbounded_object_pool_volatile* owner) noexcept
      : owner{owner} {}

    // current owner (null == stale generation)
    unbounded_object_pool_volatile* owner;
  };

  using base_t = unbounded_object_pool_base<T>;
  using generation_t = async_value<generation>;
  using generation_ptr_t = std::shared_ptr<generation_t>;
  using deleter_type = typename base_t::deleter_type;

 public:
  using element_type = typename base_t::element_type;
  using pointer = typename base_t::pointer;

 private:
  // Private because we want to disallow upcasts to std::unique_ptr<...>.
  class releaser final {
   public:
    explicit releaser(generation_ptr_t&& gen) noexcept : gen_{std::move(gen)} {}

    void operator()(pointer p) noexcept {
      IRS_ASSERT(p);     // Ensured by std::unique_ptr<...>
      IRS_ASSERT(gen_);  // Ensured by emplace(...)

      Finally release_gen = [this]() noexcept { gen_ = nullptr; };

      // do not remove scope!!!
      // variable 'lock' below must be destroyed before 'gen_'
      {
        auto lock = gen_->lock_read();

        if (auto* owner = gen_->value().owner; owner) {
          owner->release(p);
          return;
        }
      }

      // clear object oustide the lock if necessary
      deleter_type{}(p);
    }

   private:
    generation_ptr_t gen_;
  };

 public:
  // Represents a control object of unbounded_object_pool
  using ptr = pool_control_ptr<element_type, releaser>;

  explicit unbounded_object_pool_volatile(size_t size = 0)
    : base_t{size}, gen_{std::make_shared<generation_t>(this)} {}

  // FIXME check what if
  //
  // unbounded_object_pool_volatile p0, p1;
  // thread0: p0.clear();
  // thread1: unbounded_object_pool_volatile p1(std::move(p0));
  unbounded_object_pool_volatile(unbounded_object_pool_volatile&& rhs) noexcept
    : base_t{std::move(rhs)} {
    gen_ = std::atomic_load(&rhs.gen_);

    auto lock = gen_->lock_write();
    gen_->value().owner = this;  // change owner

    this->free_slots_ = std::move(rhs.free_slots_);
    this->free_objects_ = std::move(rhs.free_objects_);
  }

  ~unbounded_object_pool_volatile() noexcept {
    // prevent existing elements from returning into the pool
    // if pool doesn't own generation anymore
    {
      const auto gen = std::atomic_load(&gen_);
      auto lock = gen->lock_write();

      auto& value = gen->value();

      if (value.owner == this) {
        value.owner = nullptr;
      }
    }

    clear(false);
  }

  // Clears all cached objects and optionally prevents already created
  // objects from returning into the pool.
  // `new_generation` if true, prevents already created objects from
  // returning into the pool, otherwise just clears all cached objects.
  void clear(bool new_generation = false) {
    // prevent existing elements from returning into the pool
    if (new_generation) {
      {
        auto gen = std::atomic_load(&gen_);
        auto lock = gen->lock_write();
        gen->value().owner = nullptr;
      }

      // mark new generation
      std::atomic_store(&gen_, std::make_shared<generation_t>(this));
    }

    typename base_t::node* head = nullptr;

    // reset all cached instances
    while ((head = this->free_objects_.pop())) {
      auto p = std::exchange(head->value.value, nullptr);
      IRS_ASSERT(p);
      deleter_type{}(p);
      this->free_slots_.push(*head);
    }
  }

  template<typename... Args>
  ptr emplace(Args&&... args) {
    // retrieve before seek/instantiate
    auto gen = std::atomic_load(&gen_);
    auto value = this->acquire(std::forward<Args>(args)...);

    if (value) {
      return {value, releaser{std::move(gen)}};
    }

    return {nullptr, releaser{nullptr}};
  }

  size_t generation_size() const noexcept {
    const auto use_count = std::atomic_load(&gen_).use_count();
    IRS_ASSERT(use_count >= 2);
    return use_count - 2;  // -1 for temporary object, -1 for this->_gen
  }

 private:
  // disallow move assignment
  unbounded_object_pool_volatile& operator=(unbounded_object_pool_volatile&&) =
    delete;

  generation_ptr_t gen_;  // current generation
};

}  // namespace irs
namespace tests {

template<bool Shared, bool ReturnNull, size_t SleepSec>
struct test_object {
  using ptr = std::conditional_t<Shared, std::shared_ptr<test_object>,
                                 std::unique_ptr<test_object>>;

  static std::atomic<size_t> TOTAL_COUNT;  // # number of objects created
  static std::atomic<size_t> MAKE_COUNT;   // # number of objects created

  static ptr make(int i) {
    ++MAKE_COUNT;

    if constexpr (SleepSec > 0) {
      std::this_thread::sleep_for(std::chrono::seconds{SleepSec});
    }

    if constexpr (ReturnNull) {
      return nullptr;
    }

    return ptr{new test_object{i}};
  }

  static ptr make() { return make(0); }

  explicit test_object(int i) : id{i} { ++TOTAL_COUNT; }

  ~test_object() { --TOTAL_COUNT; }

  int id;
};

template<bool Shared, bool ReturnNull, size_t SleepSec>
std::atomic<size_t> test_object<Shared, ReturnNull, SleepSec>::TOTAL_COUNT{};

template<bool Shared, bool ReturnNull, size_t SleepSec>
std::atomic<size_t> test_object<Shared, ReturnNull, SleepSec>::MAKE_COUNT{};

using test_slow_sobject = test_object<true, false, 2>;
using test_slow_uobject = test_object<false, false, 2>;
using test_sobject = test_object<true, false, 0>;
using test_uobject = test_object<false, false, 0>;
using test_sobject_nullptr = test_object<true, true, 0>;
using test_uobject_nullptr = test_object<false, true, 0>;

}  // namespace tests

using namespace tests;

// -----------------------------------------------------------------------------
// --SECTION--                                                        test suite
// -----------------------------------------------------------------------------

TEST(bounded_object_pool_tests, check_total_number_of_instances) {
  const size_t MAX_COUNT = 2;
  irs::bounded_object_pool<test_slow_sobject> pool(MAX_COUNT);

  std::mutex mutex;
  std::condition_variable ready_cv;
  bool ready{false};

  std::atomic<size_t> id{};
  test_slow_sobject::TOTAL_COUNT = 0;

  auto job = [&mutex, &ready_cv, &pool, &ready, &id]() {
    // wait for all threads to be ready
    {
      auto lock = std::unique_lock(mutex);

      while (!ready) {
        ready_cv.wait(lock);
      }
    }

    pool.emplace(id++);
  };

  auto job_shared = [&mutex, &ready_cv, &pool, &ready, &id]() {
    // wait for all threads to be ready
    {
      auto lock = std::unique_lock(mutex);

      while (!ready) {
        ready_cv.wait(lock);
      }
    }

    pool.emplace(id++);
  };

  const size_t THREADS_COUNT = 32;
  std::vector<std::thread> threads;

  for (size_t i = 0; i < THREADS_COUNT / 2; ++i) {
    threads.emplace_back(job);
    threads.emplace_back(job_shared);
  }

  // ready
  {
    auto lock = std::unique_lock(mutex);
    ready = true;
  }
  ready_cv.notify_all();

  for (auto& thread : threads) {
    thread.join();
  }

  ASSERT_LE(test_slow_sobject::TOTAL_COUNT.load(), MAX_COUNT);
}

TEST(bounded_object_pool_tests, test_sobject_pool) {
  // block on full pool
  {
    std::condition_variable cond;
    std::mutex mutex;
    irs::bounded_object_pool<test_sobject> pool(1);
    auto obj = pool.emplace(1);

    {
      auto lock = std::unique_lock(mutex);
      std::atomic<bool> emplace(false);
      std::thread thread([&cond, &mutex, &pool, &emplace]() -> void {
        auto obj = pool.emplace(2);
        emplace = true;
        auto lock = std::unique_lock(mutex);
        cond.notify_all();
      });

      auto result =
        cond.wait_for(lock, 1000ms);  // assume thread blocks in 1000ms

      // As declaration for wait_for contains "It may also be unblocked
      // spuriously." for all platforms
      while (!emplace && result == std::cv_status::no_timeout)
        result = cond.wait_for(lock, 1000ms);

      ASSERT_EQ(std::cv_status::timeout, result);
      // ^^^ expecting timeout because pool should block indefinitely
      obj.reset();
      lock.unlock();
      thread.join();
    }
  }

  // null objects not considered part of pool
  {
    std::condition_variable cond;
    std::mutex mutex;
    irs::bounded_object_pool<test_sobject_nullptr> pool(2);
    test_sobject_nullptr::MAKE_COUNT = 0;
    auto obj = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(1, test_sobject_nullptr::MAKE_COUNT);

    {
      auto lock = std::unique_lock(mutex);
      std::atomic<bool> emplace(false);
      std::thread thread([&cond, &mutex, &pool, &emplace]() -> void {
        auto obj = pool.emplace();
        ASSERT_FALSE(obj);
        ASSERT_EQ(2, test_sobject_nullptr::MAKE_COUNT);
        emplace = true;
        auto lock = std::unique_lock(mutex);
        cond.notify_all();
      });

      auto result =
        cond.wait_for(lock, 1000ms);  // assume thread blocks in 1000ms

      // As declaration for wait_for contains "It may also be unblocked
      // spuriously." for all platforms
      while (!emplace && result == std::cv_status::no_timeout)
        result = cond.wait_for(lock, 1000ms);

      ASSERT_TRUE(emplace);

      obj.reset();
      lock.unlock();
      thread.join();
    }
  }

  // test object reuse
  {
    irs::bounded_object_pool<test_sobject> pool(1);
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    auto obj_shared = pool.emplace(2);
    ASSERT_EQ(1, obj_shared->id);
    ASSERT_EQ(obj_ptr, obj_shared.get());
  }

  // test visitation
  {
    irs::bounded_object_pool<test_sobject> pool(1);
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);
    std::condition_variable cond;
    std::mutex mutex;
    auto lock = std::unique_lock(mutex);
    std::atomic<bool> visit(false);
    std::thread thread([&cond, &mutex, &pool, &visit]() -> void {
      auto visitor = [](test_sobject&) -> bool { return true; };
      pool.visit(visitor);
      visit = true;
      auto lock = std::unique_lock(mutex);
      cond.notify_all();
    });
    auto result =
      cond.wait_for(lock, 1000ms);  // assume thread finishes in 1000ms

    // As declaration for wait_for contains "It may also be unblocked
    // spuriously." for all platforms
    while (!visit && result == std::cv_status::no_timeout)
      result = cond.wait_for(lock, 1000ms);

    obj.reset();
    ASSERT_FALSE(obj);

    if (lock) {
      lock.unlock();
    }

    thread.join();
    ASSERT_EQ(
      std::cv_status::timeout,
      result);  // check only after joining with thread to avoid early exit
    // ^^^ expecting timeout because pool should block indefinitely
  }
}

TEST(bounded_object_pool_tests, test_uobject_pool) {
  // block on full pool
  {
    std::condition_variable cond;
    std::mutex mutex;
    irs::bounded_object_pool<test_uobject> pool(1);
    auto obj = pool.emplace(1);

    {
      auto lock = std::unique_lock(mutex);
      std::atomic<bool> emplace(false);
      std::thread thread([&cond, &mutex, &pool, &emplace]() -> void {
        auto obj = pool.emplace(2);
        emplace = true;
        auto lock = std::unique_lock(mutex);
        cond.notify_all();
      });

      auto result =
        cond.wait_for(lock, 1000ms);  // assume thread blocks in 1000ms

      // As declaration for wait_for contains "It may also be unblocked
      // spuriously." for all platforms
      while (!emplace && result == std::cv_status::no_timeout)
        result = cond.wait_for(lock, 1000ms);

      ASSERT_EQ(std::cv_status::timeout, result);
      // ^^^ expecting timeout because pool should block indefinitely
      obj.reset();
      obj.reset();
      lock.unlock();
      thread.join();
    }
  }

  // null objects not considered part of pool
  {
    std::condition_variable cond;
    std::mutex mutex;
    irs::bounded_object_pool<test_uobject_nullptr> pool(2);
    test_uobject_nullptr::MAKE_COUNT = 0;
    auto obj = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(1, test_uobject_nullptr::MAKE_COUNT);

    {
      auto lock = std::unique_lock(mutex);
      std::thread thread([&cond, &mutex, &pool]() -> void {
        auto lock = std::unique_lock(mutex);
        auto obj = pool.emplace();
        ASSERT_FALSE(obj);
        ASSERT_EQ(2, test_uobject_nullptr::MAKE_COUNT);
        cond.notify_all();
      });

      ASSERT_EQ(std::cv_status::no_timeout, cond.wait_for(lock, 1000ms));
      obj.reset();
      lock.unlock();
      thread.join();
    }
  }

  // test object reuse
  {
    irs::bounded_object_pool<test_uobject> pool(1);
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    obj = pool.emplace(2);
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());
  }

  // test visitation
  {
    irs::bounded_object_pool<test_uobject> pool(1);
    auto obj = pool.emplace(1);
    std::condition_variable cond;
    std::mutex mutex;
    auto lock = std::unique_lock(mutex);
    std::atomic<bool> visit(false);
    std::thread thread([&cond, &mutex, &pool, &visit]() -> void {
      auto visitor = [](test_uobject&) -> bool { return true; };
      pool.visit(visitor);
      visit = true;
      auto lock = std::unique_lock(mutex);
      cond.notify_all();
    });
    auto result =
      cond.wait_for(lock, 1000ms);  // assume thread finishes in 1000ms

    // As declaration for wait_for contains "It may also be unblocked
    // spuriously." for all platforms
    while (!visit && result == std::cv_status::no_timeout)
      result = cond.wait_for(lock, 1000ms);

    obj.reset();

    if (lock) {
      lock.unlock();
    }

    thread.join();
    ASSERT_EQ(std::cv_status::timeout, result);
    // ^^^ expecting timeout because pool should block indefinitely
  }
}

TEST(unbounded_object_pool_tests, construct) {
  irs::unbounded_object_pool<test_uobject> pool(42);
  ASSERT_EQ(42, pool.size());
}

TEST(unbounded_object_pool_tests, check_total_number_of_cached_instances) {
  constexpr size_t MAX_COUNT = 8;
  irs::unbounded_object_pool<test_uobject> pool(MAX_COUNT);

  std::mutex mutex;
  std::condition_variable ready_cv;
  bool ready{false};

  std::atomic<size_t> id{};
  test_uobject::TOTAL_COUNT = 0;

  auto job = [&mutex, &ready_cv, &pool, &ready, &id]() {
    // wait for all threads to be ready
    {
      auto lock = std::unique_lock(mutex);

      while (!ready) {
        ready_cv.wait(lock);
      }
    }

    for (size_t i = 0; i < 100000; ++i) {
      auto p = pool.emplace(id++);
      ASSERT_TRUE(p->id >= 0);
    }
  };

  auto job_shared = [&mutex, &ready_cv, &pool, &ready, &id]() {
    // wait for all threads to be ready
    {
      auto lock = std::unique_lock(mutex);

      while (!ready) {
        ready_cv.wait(lock);
      }
    }

    for (size_t i = 0; i < 100000; ++i) {
      auto p = pool.emplace(id++);
      ASSERT_TRUE(p->id >= 0);
    }
  };

  constexpr size_t THREADS_COUNT = 32;
  std::vector<std::thread> threads;

  for (size_t i = 0; i < THREADS_COUNT / 2; ++i) {
    threads.emplace_back(job);
    threads.emplace_back(job_shared);
  }

  // ready
  {
    auto lock = std::unique_lock(mutex);
    ready = true;
  }
  ready_cv.notify_all();

  for (auto& thread : threads) {
    thread.join();
  }

  ASSERT_LE(test_slow_uobject::TOTAL_COUNT.load(), MAX_COUNT);
}

TEST(unbounded_object_pool_tests, test_uobject_pool_1) {
  // create new untracked object on full pool
  {
    std::condition_variable cond;
    std::mutex mutex;
    irs::unbounded_object_pool<test_uobject> pool(1);
    std::shared_ptr<test_uobject> obj = pool.emplace(1);

    {
      auto lock = std::unique_lock(mutex);
      std::thread thread([&cond, &mutex, &pool]() -> void {
        auto obj = pool.emplace(2);
        auto lock = std::unique_lock(mutex);
        cond.notify_all();
      });
      ASSERT_EQ(
        std::cv_status::no_timeout,
        cond.wait_for(lock, 1000ms));  // assume threads start within 1000msec
      lock.unlock();
      thread.join();
    }
  }

  // test empty pool
  {
    irs::unbounded_object_pool<test_uobject> pool;
    ASSERT_EQ(0, pool.size());
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    auto obj_shared = pool.emplace(2);
    ASSERT_TRUE(bool(obj_shared));
    ASSERT_EQ(2, obj_shared->id);
  }

  // null objects not considered part of pool
  {
    irs::unbounded_object_pool<test_uobject_nullptr> pool(2);
    test_uobject_nullptr::MAKE_COUNT = 0;
    auto obj = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(1, test_uobject_nullptr::MAKE_COUNT);
    obj.reset();
    std::shared_ptr<test_uobject_nullptr> obj_shared = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(2, test_uobject_nullptr::MAKE_COUNT);
    obj.reset();
  }

  // test object reuse
  {
    irs::unbounded_object_pool<test_uobject> pool(1);
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    std::shared_ptr<test_uobject> obj_shared{pool.emplace(2)};
    ASSERT_TRUE(bool(obj_shared));
    ASSERT_EQ(1, obj_shared->id);
    ASSERT_EQ(obj_ptr, obj_shared.get());
  }

  // ensure untracked object is not placed back in the pool
  {
    irs::unbounded_object_pool<test_uobject> pool(1);
    auto obj0 = pool.emplace(1);
    ASSERT_TRUE(obj0);
    std::shared_ptr<test_uobject> obj1{pool.emplace(2)};
    ASSERT_TRUE(bool(obj1));
    auto* obj0_ptr = obj0.get();

    ASSERT_EQ(1, obj0->id);
    ASSERT_EQ(2, obj1->id);
    ASSERT_NE(obj0_ptr, obj1.get());
    obj0.reset();  // will be placed back in pool first
    ASSERT_FALSE(obj0);
    ASSERT_EQ(nullptr, obj0.get());
    obj1.reset();  // will push obj1 out of the pool
    ASSERT_FALSE(obj1);
    ASSERT_EQ(nullptr, obj1.get());

    std::shared_ptr<test_uobject> obj2{pool.emplace(3)};
    ASSERT_TRUE(bool(obj2));
    auto obj3 = pool.emplace(4);
    ASSERT_TRUE(obj3);
    ASSERT_EQ(1, obj2->id);
    ASSERT_EQ(4, obj3->id);
    ASSERT_EQ(obj0_ptr, obj2.get());
    ASSERT_NE(obj0_ptr, obj3.get());
    // obj3 may have been allocated in the same addr as obj1, so can't safely
    // validate
  }

  // test pool clear
  {
    irs::unbounded_object_pool<test_uobject> pool(1);
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());

    pool.clear();  // will clear objects inside the pool only
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());  // same object

    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    pool.clear();           // will clear objects inside the pool only
    obj = pool.emplace(3);  // 'obj' should not be reused
    ASSERT_TRUE(obj);
    ASSERT_EQ(3, obj->id);
  }
}

TEST(unbounded_object_pool_tests, test_uobject_pool_2) {
  // create new untracked object on full pool
  {
    std::condition_variable cond;
    std::mutex mutex;
    irs::unbounded_object_pool<test_uobject> pool(1);
    std::shared_ptr<test_uobject> obj = pool.emplace(1);

    {
      auto lock = std::unique_lock(mutex);
      std::thread thread([&cond, &mutex, &pool]() -> void {
        auto obj = pool.emplace(2);
        auto lock = std::unique_lock(mutex);
        cond.notify_all();
      });
      ASSERT_EQ(
        std::cv_status::no_timeout,
        cond.wait_for(lock, 1000ms));  // assume threads start within 1000msec
      lock.unlock();
      thread.join();
    }
  }

  // null objects not considered part of pool
  {
    irs::unbounded_object_pool<test_uobject_nullptr> pool(2);
    test_uobject_nullptr::MAKE_COUNT = 0;
    auto obj = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(1, test_uobject_nullptr::MAKE_COUNT);
    obj.reset();
    std::shared_ptr<test_uobject_nullptr> obj_shared = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(2, test_uobject_nullptr::MAKE_COUNT);
    obj.reset();
  }

  // test object reuse
  {
    irs::unbounded_object_pool<test_uobject> pool(1);
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    std::shared_ptr<test_uobject> obj_shared{pool.emplace(2)};
    ASSERT_TRUE(bool(obj_shared));
    ASSERT_EQ(1, obj_shared->id);
    ASSERT_EQ(obj_ptr, obj_shared.get());
  }

  // ensure untracked object is not placed back in the pool
  {
    irs::unbounded_object_pool<test_uobject> pool(1);
    auto obj0 = pool.emplace(1);
    ASSERT_TRUE(obj0);
    std::shared_ptr<test_uobject> obj1{pool.emplace(2)};
    ASSERT_TRUE(bool(obj1));
    auto* obj0_ptr = obj0.get();

    ASSERT_EQ(1, obj0->id);
    ASSERT_EQ(2, obj1->id);
    ASSERT_NE(obj0_ptr, obj1.get());
    obj0.reset();  // will be placed back in pool first
    ASSERT_FALSE(obj0);
    ASSERT_EQ(nullptr, obj0.get());
    obj1.reset();  // will push obj1 out of the pool
    ASSERT_FALSE(obj1);
    ASSERT_EQ(nullptr, obj1.get());

    std::shared_ptr<test_uobject> obj2{pool.emplace(3)};
    ASSERT_TRUE(bool(obj2));
    auto obj3 = pool.emplace(4);
    ASSERT_TRUE(obj3);
    ASSERT_EQ(1, obj2->id);
    ASSERT_EQ(4, obj3->id);
    ASSERT_EQ(obj0_ptr, obj2.get());
    ASSERT_NE(obj0_ptr, obj3.get());
    // obj3 may have been allocated in the same addr as obj1, so can't safely
    // validate
  }

  // test pool clear
  {
    irs::unbounded_object_pool<test_uobject> pool(1);
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());

    pool.clear();  // will clear objects inside the pool only
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());  // same object

    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    pool.clear();           // will clear objects inside the pool only
    obj = pool.emplace(3);  // 'obj' should not be reused
    ASSERT_TRUE(obj);
    ASSERT_EQ(3, obj->id);
  }
}

TEST(unbounded_object_pool_tests, control_objectb_move) {
  irs::unbounded_object_pool<test_uobject> pool(2);
  ASSERT_EQ(2, pool.size());

  // move constructor
  {
    auto moved = pool.emplace(1);
    ASSERT_TRUE(moved);
    ASSERT_NE(nullptr, moved.get());
    ASSERT_EQ(1, moved->id);

    auto obj = std::move(moved);
    ASSERT_FALSE(moved);
    ASSERT_EQ(nullptr, moved);
    ASSERT_EQ(moved, nullptr);
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
  }

  // move assignment
  {
    auto moved = pool.emplace(1);
    ASSERT_TRUE(moved);
    ASSERT_NE(nullptr, moved);
    ASSERT_NE(moved, nullptr);
    ASSERT_EQ(1, moved->id);
    auto* moved_ptr = moved.get();

    auto obj = pool.emplace(2);
    ASSERT_TRUE(obj);
    ASSERT_EQ(2, obj->id);

    obj = std::move(moved);
    ASSERT_TRUE(obj);
    ASSERT_FALSE(moved);
    ASSERT_EQ(nullptr, moved.get());
    ASSERT_EQ(obj.get(), moved_ptr);
    ASSERT_EQ(1, obj->id);
  }
}

TEST(unbounded_object_pool_tests, control_object_move_pools) {
  irs::unbounded_object_pool<test_uobject> pool(2);
  test_uobject::MAKE_COUNT = 0;
  {
    irs::unbounded_object_pool<test_uobject> pool_other(2);
    auto from_other = pool_other.emplace(1);
    ASSERT_EQ(1, test_uobject::MAKE_COUNT);
    {
      auto from_this = pool.emplace(1);
      ASSERT_EQ(2, test_uobject::MAKE_COUNT);
      from_other = std::move(from_this);
    }
    // from_other should be returned to pool_other now
    // and no new make should be called
    auto from_other2 = pool_other.emplace(1);
    ASSERT_EQ(2, test_uobject::MAKE_COUNT);
  }
}

TEST(unbounded_object_pool_volatile_tests, construct) {
  irs::unbounded_object_pool_volatile<test_uobject> pool(42);
  ASSERT_EQ(42, pool.size());
  ASSERT_EQ(0, pool.generation_size());
}

TEST(unbounded_object_pool_volatile_tests, move) {
  irs::unbounded_object_pool_volatile<test_uobject> moved(2);
  ASSERT_EQ(2, moved.size());
  ASSERT_EQ(0, moved.generation_size());

  auto obj0 = moved.emplace(1);
  ASSERT_EQ(1, moved.generation_size());
  ASSERT_TRUE(obj0);
  ASSERT_NE(nullptr, obj0.get());
  ASSERT_EQ(1, obj0->id);

  irs::unbounded_object_pool_volatile<test_uobject> pool(std::move(moved));
  ASSERT_EQ(2, moved.generation_size());
  ASSERT_EQ(2, pool.generation_size());

  auto obj1 = pool.emplace(2);
  ASSERT_EQ(3, pool.generation_size());  // +1 for moved
  ASSERT_EQ(3, moved.generation_size());
  ASSERT_TRUE(obj1);
  ASSERT_NE(nullptr, obj1.get());
  ASSERT_EQ(2, obj1->id);

  // insert via moved pool
  auto obj2 = moved.emplace(3);
  ASSERT_EQ(4, pool.generation_size());
  ASSERT_EQ(4, moved.generation_size());
  ASSERT_TRUE(obj2);
  ASSERT_NE(nullptr, obj2.get());
  ASSERT_EQ(3, obj2->id);
}

TEST(unbounded_object_pool_volatile_tests, control_object_move) {
  irs::unbounded_object_pool_volatile<test_uobject> pool(2);
  ASSERT_EQ(2, pool.size());
  ASSERT_EQ(0, pool.generation_size());

  // move constructor
  {
    auto moved = pool.emplace(1);
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(moved);
    ASSERT_NE(nullptr, moved);
    ASSERT_EQ(1, moved->id);

    auto obj = std::move(moved);
    ASSERT_FALSE(moved);
    ASSERT_EQ(nullptr, moved);
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
  }

  // move assignment
  {
    auto moved = pool.emplace(1);
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(moved);
    ASSERT_NE(nullptr, moved);
    ASSERT_NE(moved, nullptr);
    ASSERT_EQ(1, moved->id);
    auto* moved_ptr = moved.get();

    auto obj = pool.emplace(2);
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_EQ(2, obj->id);

    obj = std::move(moved);
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_FALSE(moved);
    ASSERT_EQ(nullptr, moved);
    ASSERT_EQ(moved, nullptr);
    ASSERT_EQ(obj.get(), moved_ptr);
    ASSERT_EQ(1, obj->id);
  }

  // move between two identical pools
  {
    irs::unbounded_object_pool_volatile<test_uobject> pool_other(2);
    auto from_other = pool_other.emplace(1);
    ASSERT_EQ(1, pool_other.generation_size());
    {
      auto from_this = pool.emplace(3);
      ASSERT_EQ(1, pool.generation_size());
      from_other = std::move(from_this);
    }
    // from_other should be returned to pool_other now
    ASSERT_EQ(0, pool_other.generation_size());
  }

  ASSERT_EQ(0, pool.generation_size());
}

TEST(unbounded_object_pool_volatile_tests, test_uobject_pool_1) {
  // create new untracked object on full pool
  {
    std::condition_variable cond;
    std::mutex mutex;
    irs::unbounded_object_pool_volatile<test_uobject> pool(1);
    ASSERT_EQ(0, pool.generation_size());
    std::shared_ptr<test_uobject> obj = pool.emplace(1);
    ASSERT_EQ(1, pool.generation_size());

    {
      auto lock = std::unique_lock(mutex);
      std::thread thread([&cond, &mutex, &pool]() -> void {
        auto obj = pool.emplace(2);
        auto lock = std::unique_lock(mutex);
        cond.notify_all();
      });
      // assume threads start within 1000msec
      ASSERT_EQ(std::cv_status::no_timeout, cond.wait_for(lock, 1000ms));
      lock.unlock();
      thread.join();
    }

    ASSERT_EQ(1, pool.generation_size());
  }

  // test empty pool
  {
    irs::unbounded_object_pool_volatile<test_uobject> pool;
    ASSERT_EQ(0, pool.size());
    auto obj = pool.emplace(1);
    ASSERT_TRUE(obj);

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    auto obj_shared = pool.emplace(2);
    ASSERT_TRUE(bool(obj_shared));
    ASSERT_EQ(2, obj_shared->id);
  }

  // null objects not considered part of pool
  {
    irs::unbounded_object_pool_volatile<test_uobject_nullptr> pool(2);
    test_uobject_nullptr::MAKE_COUNT = 0;
    auto obj = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(1, test_uobject_nullptr::MAKE_COUNT);
    obj.reset();
    std::shared_ptr<test_uobject_nullptr> obj_shared = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(2, test_uobject_nullptr::MAKE_COUNT);
    obj.reset();
  }

  // test object reuse
  {
    irs::unbounded_object_pool_volatile<test_uobject> pool(1);
    ASSERT_EQ(0, pool.generation_size());
    auto obj = pool.emplace(1);
    ASSERT_EQ(1, pool.generation_size());
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    ASSERT_EQ(0, pool.generation_size());
    std::shared_ptr<test_uobject> obj_shared{pool.emplace(2)};
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_EQ(1, obj_shared->id);
    ASSERT_EQ(obj_ptr, obj_shared.get());
  }

  // ensure untracked object is not placed back in the pool
  {
    irs::unbounded_object_pool_volatile<test_uobject> pool(1);
    ASSERT_EQ(0, pool.generation_size());
    auto obj0 = pool.emplace(1);
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(obj0);
    std::shared_ptr<test_uobject> obj1 = pool.emplace(2);
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(bool(obj1));
    auto* obj0_ptr = obj0.get();

    ASSERT_EQ(1, obj0->id);
    ASSERT_EQ(2, obj1->id);
    ASSERT_NE(obj0_ptr, obj1.get());
    obj0.reset();  // will be placed back in pool first
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_FALSE(obj0);
    ASSERT_EQ(nullptr, obj0.get());
    obj1.reset();  // will push obj1 out of the pool
    ASSERT_EQ(0, pool.generation_size());
    ASSERT_FALSE(obj1);
    ASSERT_EQ(nullptr, obj1.get());

    std::shared_ptr<test_uobject> obj2 = pool.emplace(3);
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(bool(obj2));
    auto obj3 = pool.emplace(4);
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(obj3);
    ASSERT_EQ(1, obj2->id);
    ASSERT_EQ(4, obj3->id);
    ASSERT_EQ(obj0_ptr, obj2.get());
    ASSERT_NE(obj0_ptr, obj3.get());
    // obj3 may have been allocated in the same addr as obj1, so can't safely
    // validate
  }

  // test pool clear
  {
    irs::unbounded_object_pool_volatile<test_uobject> pool(1);
    ASSERT_EQ(0, pool.generation_size());
    auto obj_noreuse = pool.emplace(-1);
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(obj_noreuse);
    auto obj = pool.emplace(1);
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(obj);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());

    pool.clear();  // clear existing in a pool
    ASSERT_EQ(2, pool.generation_size());
    obj.reset();
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);  // may return same memory address as obj_ptr, but
                            // constructor would have been called
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());

    pool.clear(true);  // clear existing in a pool and prevent external object
                       // from returning back
    ASSERT_EQ(0, pool.generation_size());
    obj.reset();
    ASSERT_EQ(0, pool.generation_size());
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);  // may return same memory address as obj_ptr, but
                            // constructor would have been called
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_EQ(2, obj->id);

    obj_noreuse.reset();  // reset value from previuos generation
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_FALSE(obj_noreuse);
    ASSERT_EQ(nullptr, obj_noreuse.get());
    obj = pool.emplace(3);  // 'obj_noreuse' should not be reused
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_EQ(3, obj->id);
  }
}

TEST(unbounded_object_pool_volatile_tests, return_object_after_pool_destroyed) {
  auto pool =
    std::make_unique<irs::unbounded_object_pool_volatile<test_uobject>>(1);
  ASSERT_EQ(0, pool->generation_size());
  ASSERT_NE(nullptr, pool);

  auto obj = pool->emplace(42);
  ASSERT_EQ(1, pool->generation_size());
  ASSERT_TRUE(obj);
  ASSERT_EQ(42, obj->id);
  std::shared_ptr<test_uobject> objShared = pool->emplace(442);
  ASSERT_EQ(2, pool->generation_size());
  ASSERT_NE(nullptr, objShared);
  ASSERT_EQ(442, objShared->id);

  // destroy pool
  pool.reset();
  ASSERT_EQ(nullptr, pool);

  // ensure objects are still there
  ASSERT_EQ(42, obj->id);
  ASSERT_EQ(442, objShared->id);
}

TEST(unbounded_object_pool_volatile_tests, test_uobject_pool_2) {
  // create new untracked object on full pool
  {
    std::condition_variable cond;
    std::mutex mutex;
    irs::unbounded_object_pool_volatile<test_uobject> pool(1);
    auto obj = pool.emplace(1);

    {
      auto lock = std::unique_lock(mutex);
      std::thread thread([&cond, &mutex, &pool]() -> void {
        auto obj = pool.emplace(2);
        auto lock = std::unique_lock(mutex);
        cond.notify_all();
      });
      // assume threads start within 1000msec
      ASSERT_EQ(std::cv_status::no_timeout, cond.wait_for(lock, 1000ms));
      lock.unlock();
      thread.join();
    }
  }

  // null objects not considered part of pool
  {
    irs::unbounded_object_pool_volatile<test_uobject_nullptr> pool(2);
    test_uobject_nullptr::MAKE_COUNT = 0;
    auto obj = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(1, test_uobject_nullptr::MAKE_COUNT);
    obj.reset();
    std::shared_ptr<test_uobject_nullptr> obj_shared = pool.emplace();
    ASSERT_FALSE(obj);
    ASSERT_EQ(2, test_uobject_nullptr::MAKE_COUNT);
    obj.reset();
  }

  // test object reuse
  {
    irs::unbounded_object_pool_volatile<test_uobject> pool(1);
    auto obj = pool.emplace(1);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());
  }

  // ensure untracked object is not placed back in the pool
  {
    irs::unbounded_object_pool_volatile<test_uobject> pool(1);
    auto obj0 = pool.emplace(1);
    auto obj1 = pool.emplace(2);
    auto* obj0_ptr = obj0.get();
    auto* obj1_ptr = obj1.get();

    ASSERT_EQ(1, obj0->id);
    ASSERT_EQ(2, obj1->id);
    ASSERT_NE(obj0_ptr, obj1.get());
    obj1.reset();  // will be placed back in pool first
    ASSERT_FALSE(obj1);
    ASSERT_EQ(nullptr, obj1.get());
    obj0.reset();  // will push obj1 out of the pool
    ASSERT_FALSE(obj0);
    ASSERT_EQ(nullptr, obj0.get());

    auto obj2 = pool.emplace(3);
    auto obj3 = pool.emplace(4);
    ASSERT_EQ(2, obj2->id);
    ASSERT_EQ(4, obj3->id);
    ASSERT_EQ(obj1_ptr, obj2.get());
    ASSERT_NE(obj1_ptr, obj3.get());
    // obj3 may have been allocated in the same addr as obj1, so can't safely
    // validate
  }

  // test pool clear
  {
    irs::unbounded_object_pool_volatile<test_uobject> pool(1);
    ASSERT_EQ(0, pool.generation_size());
    auto obj_noreuse = pool.emplace(-1);
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(obj_noreuse);
    auto obj = pool.emplace(1);
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(obj);
    auto* obj_ptr = obj.get();

    ASSERT_EQ(1, obj->id);
    obj.reset();
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());

    pool.clear();  // clear existing in a pool
    ASSERT_EQ(2, pool.generation_size());
    obj.reset();
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);  // may return same memory address as obj_ptr, but
                            // constructor would have been called
    ASSERT_EQ(2, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_EQ(1, obj->id);
    ASSERT_EQ(obj_ptr, obj.get());

    pool.clear(true);  // clear existing in a pool and prevent external object
                       // from returning back
    ASSERT_EQ(0, pool.generation_size());
    obj.reset();
    ASSERT_EQ(0, pool.generation_size());
    ASSERT_FALSE(obj);
    ASSERT_EQ(nullptr, obj.get());
    obj = pool.emplace(2);  // may return same memory address as obj_ptr, but
                            // constructor would have been called
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_TRUE(obj);
    ASSERT_EQ(2, obj->id);

    obj_noreuse.reset();  // reset value from previuos generation
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_FALSE(obj_noreuse);
    ASSERT_EQ(nullptr, obj_noreuse.get());
    obj = pool.emplace(3);  // 'obj_noreuse' should not be reused
    ASSERT_EQ(1, pool.generation_size());
    ASSERT_EQ(3, obj->id);
  }
}

TEST(unbounded_object_pool_volatile_tests,
     check_total_number_of_cached_instances) {
  constexpr size_t MAX_COUNT = 2;
  irs::unbounded_object_pool_volatile<test_uobject> pool(MAX_COUNT);

  std::mutex mutex;
  std::condition_variable ready_cv;
  bool ready{false};

  std::atomic<size_t> id{};
  test_uobject::TOTAL_COUNT = 0;

  auto job = [&mutex, &ready_cv, &pool, &ready, &id]() {
    // wait for all threads to be ready
    {
      auto lock = std::unique_lock(mutex);

      while (!ready) {
        ready_cv.wait(lock);
      }
    }

    for (size_t i = 0; i < 100000; ++i) {
      auto p = pool.emplace(id++);
      ASSERT_TRUE(p->id >= 0);
    }
  };

  auto job_shared = [&mutex, &ready_cv, &pool, &ready, &id]() {
    // wait for all threads to be ready
    {
      auto lock = std::unique_lock(mutex);

      while (!ready) {
        ready_cv.wait(lock);
      }
    }

    for (size_t i = 0; i < 100000; ++i) {
      auto p = pool.emplace(id++);
      ASSERT_TRUE(p->id >= 0);
    }
  };

  constexpr size_t THREADS_COUNT = 32;
  std::vector<std::thread> threads;

  for (size_t i = 0; i < THREADS_COUNT / 2; ++i) {
    threads.emplace_back(job);
    threads.emplace_back(job_shared);
  }

  // ready
  {
    auto lock = std::unique_lock(mutex);
    ready = true;
  }
  ready_cv.notify_all();

  for (auto& thread : threads) {
    thread.join();
  }

  ASSERT_LE(test_slow_uobject::TOTAL_COUNT.load(), MAX_COUNT);
}

TEST(concurrent_linked_list_test, push_pop) {
  typedef irs::concurrent_stack<size_t> stack;
  typedef stack::node_type node_type;

  std::vector<node_type> nodes(10);

  size_t v = 0;
  for (auto& node : nodes) {
    node.value = v++;
  }

  stack list;
  ASSERT_TRUE(list.empty());
  ASSERT_EQ(nullptr, list.pop());

  for (auto& node : nodes) {
    list.push(node);
  }
  ASSERT_FALSE(list.empty());

  node_type* node = 0;
  auto rbegin = nodes.rbegin();
  while ((node = list.pop())) {
    ASSERT_EQ(&*rbegin, node);
    list.push(*node);
    node = list.pop();
    ASSERT_EQ(&*rbegin, node);
    ++rbegin;
  }
  ASSERT_EQ(nodes.rend(), rbegin);
}

TEST(concurrent_linked_list_test, push) {
  typedef irs::concurrent_stack<size_t> stack;
  typedef stack::node_type node_type;

  std::vector<node_type> nodes(10);
  stack list;
  ASSERT_TRUE(list.empty());
  ASSERT_EQ(nullptr, list.pop());

  for (auto& node : nodes) {
    list.push(node);
  }
  ASSERT_FALSE(list.empty());

  size_t count = 0;
  while (auto* head = list.pop()) {
    ++count;
  }
  ASSERT_TRUE(list.empty());

  ASSERT_EQ(nodes.size(), count);
}

TEST(concurrent_linked_list_test, move) {
  typedef irs::concurrent_stack<size_t> stack;

  std::array<stack::node_type, 10> nodes;
  stack moved;
  ASSERT_TRUE(moved.empty());

  size_t i = 0;
  for (auto& node : nodes) {
    node.value = i;
    moved.push(node);
  }
  ASSERT_FALSE(moved.empty());

  stack list(std::move(moved));
  ASSERT_TRUE(moved.empty());
  ASSERT_FALSE(list.empty());

  auto rbegin = nodes.rbegin();
  while (auto* node = list.pop()) {
    ASSERT_EQ(rbegin->value, node->value);
    ++rbegin;
  }

  ASSERT_EQ(nodes.rend(), rbegin);
  ASSERT_TRUE(list.empty());
}

TEST(concurrent_linked_list_test, move_assignment) {
  typedef irs::concurrent_stack<size_t> stack;

  std::array<stack::node_type, 10> nodes;
  stack list0;
  ASSERT_TRUE(list0.empty());
  stack list1;
  ASSERT_TRUE(list1.empty());

  size_t i = 0;

  for (; i < nodes.size() / 2; ++i) {
    auto& node = nodes[i];
    node.value = i;
    list0.push(node);
  }
  ASSERT_FALSE(list0.empty());

  for (; i < nodes.size(); ++i) {
    auto& node = nodes[i];
    node.value = i;
    list1.push(node);
  }
  ASSERT_FALSE(list1.empty());

  list0 = std::move(list1);

  auto rbegin = nodes.rbegin();
  while (auto* node = list0.pop()) {
    ASSERT_EQ(rbegin->value, node->value);
    ++rbegin;
  }
  ASSERT_TRUE(list0.empty());
  ASSERT_TRUE(list1.empty());
}

TEST(concurrent_linked_list_test, concurrent_pop) {
  const size_t NODES = 10000;
  const size_t THREADS = 16;

  typedef irs::concurrent_stack<size_t> stack;
  std::vector<stack::node_type> nodes(NODES);

  // build-up a list
  stack list;
  ASSERT_TRUE(list.empty());
  size_t size = 0;
  for (auto& node : nodes) {
    node.value = size++;
    list.push(node);
  }
  ASSERT_FALSE(list.empty());

  std::vector<std::vector<size_t>> threads_data(THREADS);
  std::vector<std::thread> threads;
  threads.reserve(THREADS);

  std::mutex mutex;
  std::condition_variable ready_cv;
  bool ready = false;

  auto wait_for_all = [&mutex, &ready, &ready_cv]() {
    // wait for all threads to be registered
    std::unique_lock<std::remove_reference<decltype(mutex)>::type> lock(mutex);
    while (!ready) {
      ready_cv.wait(lock);
    }
  };

  // start threads
  {
    auto lock = std::unique_lock(mutex);
    for (size_t i = 0; i < threads_data.size(); ++i) {
      auto& thread_data = threads_data[i];
      threads.emplace_back([&list, &wait_for_all, &thread_data]() {
        wait_for_all();

        while (auto* head = list.pop()) {
          thread_data.push_back(head->value);
        }
      });
    }
  }

  // all threads are registered... go, go, go...
  {
    std::lock_guard<decltype(mutex)> lock(mutex);
    ready = true;
    ready_cv.notify_all();
  }

  for (auto& thread : threads) {
    thread.join();
  }

  ASSERT_TRUE(list.empty());

  std::set<size_t> results;
  for (auto& thread_data : threads_data) {
    for (auto value : thread_data) {
      ASSERT_TRUE(results.insert(value).second);
    }
  }
  ASSERT_EQ(NODES, results.size());
}

TEST(concurrent_linked_list_test, concurrent_push) {
  const size_t NODES = 10000;
  const size_t THREADS = 16;

  typedef irs::concurrent_stack<size_t> stack;

  // build-up a list
  stack list;
  ASSERT_TRUE(list.empty());

  std::vector<std::vector<stack::node_type>> threads_data;
  threads_data.reserve(THREADS);
  for (size_t i = 0; i < THREADS; ++i) {
    threads_data.emplace_back(NODES);
  }

  std::vector<std::thread> threads;
  threads.reserve(THREADS);

  std::mutex mutex;
  std::condition_variable ready_cv;
  bool ready = false;

  auto wait_for_all = [&mutex, &ready, &ready_cv]() {
    // wait for all threads to be registered
    auto lock = std::unique_lock(mutex);
    while (!ready) {
      ready_cv.wait(lock);
    }
  };

  // start threads
  {
    auto lock = std::unique_lock(mutex);
    for (size_t i = 0; i < threads_data.size(); ++i) {
      auto& thread_data = threads_data[i];
      threads.emplace_back([&list, &wait_for_all, &thread_data]() {
        wait_for_all();

        size_t idx = 0;
        for (auto& node : thread_data) {
          node.value = idx++;
          list.push(node);
        }
      });
    }
  }

  // all threads are registered... go, go, go...
  {
    std::lock_guard<decltype(mutex)> lock(mutex);
    ready = true;
    ready_cv.notify_all();
  }

  for (auto& thread : threads) {
    thread.join();
  }

  ASSERT_FALSE(list.empty());

  std::vector<size_t> results(NODES);

  while (auto* node = list.pop()) {
    ASSERT_TRUE(node->value < results.size());
    ++results[node->value];
  }

  ASSERT_TRUE(results.front() == THREADS &&
              irs::irstd::all_equal(results.begin(), results.end()));
}

TEST(concurrent_linked_list_test, concurrent_pop_push) {
  static const size_t NODES = 10000;
  const size_t THREADS = 16;

  struct data {
    std::atomic_flag visited = ATOMIC_FLAG_INIT;
    std::atomic<size_t> num_owners{};
    size_t value{};
  };

  typedef irs::concurrent_stack<data> stack;
  std::vector<stack::node_type> nodes(NODES);

  // build-up a list
  stack list;
  ASSERT_TRUE(list.empty());
  for (auto& node : nodes) {
    list.push(node);
  }
  ASSERT_FALSE(list.empty());

  std::vector<std::thread> threads;
  threads.reserve(THREADS);

  std::mutex mutex;
  std::condition_variable ready_cv;
  bool ready = false;

  auto wait_for_all = [&mutex, &ready, &ready_cv]() {
    // wait for all threads to be registered
    auto lock = std::unique_lock(mutex);
    while (!ready) {
      ready_cv.wait(lock);
    }
  };

  // start threads
  {
    auto lock = std::unique_lock(mutex);
    for (size_t i = 0; i < THREADS; ++i) {
      threads.emplace_back([&list, &wait_for_all]() {
        wait_for_all();

        // no more than NODES
        size_t processed = 0;

        while (auto* head = list.pop()) {
          ++processed;
          EXPECT_LE(processed, 2 * NODES);

          auto& value = head->value;

          if (!value.visited.test_and_set()) {
            ++value.num_owners;
            ++value.value;
            list.push(*head);
          } else {
            --value.num_owners;
            ASSERT_EQ(0, value.num_owners);
          }
        }
      });
    }
  }

  // all threads are registered... go, go, go...
  {
    auto lock = std::unique_lock(mutex);
    ready = true;
    ready_cv.notify_all();
  }

  for (auto& thread : threads) {
    thread.join();
  }

  ASSERT_TRUE(list.empty());

  for (auto& node : nodes) {
    ASSERT_EQ(1, node.value.value);
    ASSERT_EQ(true, node.value.visited.test_and_set());
    ASSERT_EQ(0, node.value.num_owners);
  }
}
