(Optionally Concurrent) FIFO

Question

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

Answer 1

Your popped elements will not be destructed. You move them but that isn't the same as being destructed and it doesn't guarantee that memory will be released.

Furthermore your container requires T to be default constructible which prevents immutable objects from being used with the queue.

See this question for how to resolve both: Implementation of fixed size queue using a ring (cyclic) buffer.

Your push function is not exception safe.

I dislike all capital names as typically all capital identifiers are used for macros.

The name FIFO is poor in my opinion because the concept you're modelling is a queue or a pipe (these are similar but different). First in first out is just a description of how data enters and leaves the container. FIFO doesn't allude to the fact that it will overwrite on push if the container is full which is a bit surprising if you're not familiar with the container. Personally I do not like this behaviour as it violates the principle of least surprise.

Also I'm not a fan of the ST_ prefix which I assume means single threaded... If you're adding that prefix to this class, you should add it to all classes which are not concurrent safe, which quickly becomes obnoxious. As for the MT_ prefix, that is acceptable but as it is an attribute of the queue, I'd rather see it as a suffix. I would prefer naming the class something like concurrent_fixed_pipe.