Based on Concurrent FIFO in C++11 and my review I implemented a queue and its concurrent pendant.
Is there anything left to improve regarding clarity, usability, code-style, lock-times or general efficiency?
#ifndef FIFO_H
#define FIFO_H
#include <array>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <type_traits>
template<class T, std::size_t CAPACITY>
class ST_FIFO
{
static_assert(CAPACITY, "Needs to have non-zero capacity");
T data[CAPACITY + 1];
std::size_t input_index = 0;
std::size_t output_index = 0;
inline static constexpr std::size_t wrap_index(std::size_t index) noexcept
{ return index > CAPACITY ? index - CAPACITY - 1 : index; }
public:
static constexpr std::size_t capacity() noexcept { return CAPACITY; }
bool empty() const noexcept { return input_index == output_index; }
std::size_t size() const noexcept
{
return input_index >= output_index
? input_index - output_index
: input_index + CAPACITY + 1 - output_index;
}
template<class X>
auto push(X&& x) noexcept(noexcept(pop(*data), *data = std::forward<X>(x)))
-> decltype(*data = std::forward<X>(x), true)
{
if(size() == CAPACITY)
pop(data[input_index]);
data[input_index] = std::forward<X>(x);
input_index = wrap_index(input_index + 1);
return true;
}
template<class X>
auto try_push(X&& x) noexcept(noexcept(*data = std::forward<X>(x)))
-> decltype(*data = std::forward<X>(x), true)
{
if(size() == CAPACITY)
return false;
data[input_index] = std::forward<X>(x);
input_index = wrap_index(input_index + 1);
return true;
}
std::size_t multi_push(const T ts[], size_t count)
noexcept(noexcept(push(*ts)))
{
for (size_t i = 0; i < count; ++i)
push(ts[i]);
return count;
}
std::size_t try_multi_push(const T ts[], size_t count)
noexcept(noexcept(try_push(*ts)))
{
for (size_t i = 0; i < count; ++i)
if(!try_push(ts[i]))
return i;
return count;
}
bool pop(T &t) noexcept(noexcept(t = std::move(t)))
{
if (empty())
return false;
t = std::move(data[output_index]);
output_index = wrap_index(output_index + 1);
return true;
}
std::size_t multi_pop(T ts[], size_t count) noexcept(noexcept(pop(*ts)))
{
for (size_t i = 0; i < count; ++i)
if(!pop(ts[i]))
return i;
return count;
}
bool peek(std::size_t ind, T &t) const noexcept(noexcept(t = t))
{
if (ind >= size())
return false;
t = data[wrap_index(output_index + ind)];
return true;
}
};
template<class T, std::size_t CAPACITY>
class MT_FIFO : ST_FIFO<T, CAPACITY>
{
std::atomic_bool wait_flag = true;
mutable std::mutex mutex;
mutable std::condition_variable cv;
using base = ST_FIFO<T, CAPACITY>;
template<bool wait = false, class... X>
inline std::unique_lock<std::mutex> lock(X... x) const
{
std::unique_lock<std::mutex> lock(mutex, x...);
if(wait)
cv.wait(lock, [this]{return !(base::empty() && wait_flag);});
return lock;
}
inline MT_FIFO(const MT_FIFO& other, std::unique_lock<std::mutex>&&)
: base(other)
, wait_flag(other.wait_flag.load())
{}
template<bool all = false, class F>
inline auto locked(F f) noexcept(noexcept(f())) -> decltype(f())
{
auto result = (lock(), f());
if(result)
all ? cv.notify_all() : cv.notify_one();
return result;
}
public:
MT_FIFO() = default;
MT_FIFO(const MT_FIFO& o) : MT_FIFO(o, o.lock()) {}
MT_FIFO& operator=(const MT_FIFO& o) noexcept(noexcept(base::operator=(o)))
{
if(this == &o)
return *this;
auto a = lock(std::defer_lock);
auto b = o.lock(std::defer_lock);
std::lock(a, b);
base::operator=(o);
wait_flag = o.wait_flag;
return *this;
}
using base::capacity;
bool empty() const noexcept { return lock(), base::empty(); }
std::size_t size() const noexcept { return lock(), base::size(); }
template<class X>
auto push(X&& x) noexcept(noexcept(base::push(std::forward<X>(x))))
-> decltype(base::push(std::forward<X>(x)))
{ return locked([&]{return base::push(std::forward<X>(x));}); }
template<class X>
auto try_push(X&& x) noexcept(noexcept(base::try_push(std::forward<X>(x))))
-> decltype(base::try_push(std::forward<X>(x)))
{ return locked([&]{return base::try_push(std::forward<X>(x));}); }
std::size_t multi_push(const T ts[], size_t count)
noexcept(noexcept(base::multi_push(ts, count)))
{ return locked<true>([&]{return base::multi_push(ts, count);}); }
std::size_t try_multi_push(const T ts[], size_t count)
noexcept(noexcept(base::try_multi_push(ts, count)))
{ return locked<true>([&]{return base::try_multi_push(ts, count);}); }
bool pop(T &t) noexcept(noexcept(base::pop(t)))
{ return lock<true>(), base::pop(t); }
std::size_t multi_pop(T ts[], size_t count)
noexcept(noexcept(base::multi_pop(ts, count)))
{ return lock<true>(), base::multi_pop(ts, count); }
bool peek(std::size_t ind, T &t) const
noexcept(noexcept(base::peek(ind, t)))
{ return lock(), base::peek(ind, t); }
void wait_on() noexcept
{ lock(), wait_flag = true; }
void wait_off() noexcept
{ locked<true>([&]{return wait_flag = false;}); }
};
#endif // FIFO_H