26257287247

Committed 21 May 2026 10:40PM UTC coverage: 92.718% (-0.01%) from 92.731%

Build Type

push

github

Committed by MarkusB

Commit Message

lint

Coverage Stats

3 of 4 new or added lines in 2 files covered. (75.0%)

6468 of 6976 relevant lines covered (92.72%)

125.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.1

/include/mb/ul/basiccodesupport/thread/waitcircbuffer.hpp

//! \file

#ifndef WAITCIRCBUFFER_H_4Z894527892XH5
#define WAITCIRCBUFFER_H_4Z894527892XH5

#include "../error.hpp"
#include "../ignore_unused.hpp"
#include "../std/circular_buffer.hpp"
#include <condition_variable>
#include <mutex>
#include <queue>
#include <type_traits>

namespace mb::ul::thread {
namespace detail::waitcircbuf_impl_container {
template <typename T, size_t static_capacity>
class CapacityCompiletime {
protected:
    ul::CircularBuffer<T, static_capacity> buf_;
};

template <typename T>
class CapacityRuntime {
protected:
    ul::CircularBuffer<T> buf_;

    explicit CapacityRuntime(size_t capacity)
        : buf_{capacity} {
        UL_EXPECT(capacity > 0);
    }
};

template <typename T, size_t static_capacity>
using Base = std::conditional_t<static_capacity == 0, CapacityRuntime<T>, CapacityCompiletime<T, static_capacity>>;
} // namespace detail::waitcircbuf_impl_container

//! A queue that exclusively provides methods that are safe to use in a multi-producer / multi-consumer context.
/** Note, unlike the behavior of CircularBuffer this WaitCircularBuffer protects against overwrites.*/
template <typename T, size_t static_capacity = 0>
class WaitCircularBuffer : private detail::waitcircbuf_impl_container::Base<T, static_capacity> {
public:
    using Base = detail::waitcircbuf_impl_container::Base<T, static_capacity>;

    //! Expects capacity > 0.
    template <size_t capacity_enabled = static_capacity>
        requires(capacity_enabled == 0)
    explicit WaitCircularBuffer(size_t capacity)
        : Base{capacity} {
    }

    WaitCircularBuffer() noexcept = default;

    template <typename U>
    bool push(U&& elem) {
        static_assert(std::is_convertible_v<U, T>);

        {
            const std::unique_lock<std::mutex> lock(mutex_);

            if (Base::buf_.full())
                return false;

            Base::buf_.push(std::forward<U>(elem));
        }
        condition_variable_.notify_one();
        return true;
    }

    //! Don't use! Doesn't work yet. Implementation of CircularBuffer is the guilty one.
    /** (At least not working in the 'dynamic' staticCapacity == 0 case, whereas in the 'static' capacity != 0 case
        there is no difference to push.)*/
    template <typename U>
    bool emplace(U&& elem) {
        static_assert(std::is_convertible_v<U, T>);

        ul::ignore_unused(std::forward<U>(elem));
        throw NotImplemented{"WaitCircularBuffer::emplace"};
        // {
        //     std::unique_lock<std::mutex> lock(mutex_);
        //
        //     if (Base::buf_.full())
        //         return false;
        //
        //     Base::buf_.emplace(std::forward<U>(elem);
        // }
        // conditionVariable_.notify_one();
        // return true;
    }

    //! \return false if the queue has been stopped (this interrupts waiting even if the queue is empty).
    bool wait_and_pop(T& popped_elem) {
        std::unique_lock<std::mutex> lk(mutex_);
        while (Base::buf_.empty() && !stopped_)
            condition_variable_.wait(lk);

        if (stopped_)
            return false;

        if (!Base::buf_.try_pop(popped_elem)) {
            UL_ASSERT(false);
            return false;
        }

        return true;
    }

    void stop() {
        const std::unique_lock<std::mutex> lock(mutex_);
        stopped_ = true;
        condition_variable_.notify_all();
    }

    size_t size() const {
        std::scoped_lock<std::mutex> lock(mutex_);
        return Base::buf_.size();
    }

    constexpr size_t capacity() const noexcept {
        return Base::buf_.capacity();
    }

private:
    mutable std::mutex mutex_;
    std::condition_variable condition_variable_;
    bool stopped_{false};
};
} // namespace mb::ul::thread

#endif

1	//! \file
2
3	#ifndef WAITCIRCBUFFER_H_4Z894527892XH5
4	#define WAITCIRCBUFFER_H_4Z894527892XH5
5
6	#include "../error.hpp"
7	#include "../ignore_unused.hpp"
8	#include "../std/circular_buffer.hpp"
9	#include <condition_variable>
10	#include <mutex>
11	#include <queue>
12	#include <type_traits>
13
14	namespace mb::ul::thread {
15	namespace detail::waitcircbuf_impl_container {
16	template <typename T, size_t static_capacity>
17	class CapacityCompiletime {
18	protected:
19	ul::CircularBuffer<T, static_capacity> buf_;
20	};
21
22	template <typename T>
23	class CapacityRuntime {
24	protected:
25	ul::CircularBuffer<T> buf_;
26
27	explicit CapacityRuntime(size_t capacity)	3✔
28	: buf_{capacity} {	3✔
29	UL_EXPECT(capacity > 0);
30	}	3✔
31	};
32
33	template <typename T, size_t static_capacity>
34	using Base = std::conditional_t<static_capacity == 0, CapacityRuntime<T>, CapacityCompiletime<T, static_capacity>>;
35	} // namespace detail::waitcircbuf_impl_container
36
37	//! A queue that exclusively provides methods that are safe to use in a multi-producer / multi-consumer context.
38	/** Note, unlike the behavior of CircularBuffer this WaitCircularBuffer protects against overwrites.*/
39	template <typename T, size_t static_capacity = 0>
40	class WaitCircularBuffer : private detail::waitcircbuf_impl_container::Base<T, static_capacity> {
41	public:
42	using Base = detail::waitcircbuf_impl_container::Base<T, static_capacity>;
43
44	//! Expects capacity > 0.
45	template <size_t capacity_enabled = static_capacity>
46	requires(capacity_enabled == 0)
47	explicit WaitCircularBuffer(size_t capacity)	3✔
48	: Base{capacity} {	3✔
49	}	3✔
50
51	WaitCircularBuffer() noexcept = default;
52
53	template <typename U>
54	bool push(U&& elem) {	104✔
55	static_assert(std::is_convertible_v<U, T>);
56
57	{
58	const std::unique_lock<std::mutex> lock(mutex_);	104✔
59
60	if (Base::buf_.full())	104✔
61	return false;	1✔
62
63	Base::buf_.push(std::forward<U>(elem));	206✔
64	}	104✔
65	condition_variable_.notify_one();	103✔
66	return true;	103✔
67	}
68
69	//! Don't use! Doesn't work yet. Implementation of CircularBuffer is the guilty one.
70	/** (At least not working in the 'dynamic' staticCapacity == 0 case, whereas in the 'static' capacity != 0 case
71	there is no difference to push.)*/
72	template <typename U>
73	bool emplace(U&& elem) {
74	static_assert(std::is_convertible_v<U, T>);
75
76	ul::ignore_unused(std::forward<U>(elem));
77	throw NotImplemented{"WaitCircularBuffer::emplace"};
78	// {
79	// std::unique_lock<std::mutex> lock(mutex_);
80	//
81	// if (Base::buf_.full())
82	// return false;
83	//
84	// Base::buf_.emplace(std::forward<U>(elem);
85	// }
86	// conditionVariable_.notify_one();
87	// return true;
88	}
89
90	//! \return false if the queue has been stopped (this interrupts waiting even if the queue is empty).
91	bool wait_and_pop(T& popped_elem) {	104✔
92	std::unique_lock<std::mutex> lk(mutex_);	104✔
93	while (Base::buf_.empty() && !stopped_)	104✔
94	condition_variable_.wait(lk);	×
95
96	if (stopped_)	104✔
97	return false;	1✔
98
99	if (!Base::buf_.try_pop(popped_elem)) {	103✔
100	UL_ASSERT(false);
NEW 101	return false;	×
102	}
103
104	return true;	103✔
105	}	104✔
106
107	void stop() {	1✔
108	const std::unique_lock<std::mutex> lock(mutex_);	1✔
109	stopped_ = true;	1✔
110	condition_variable_.notify_all();	1✔
111	}	1✔
112
113	size_t size() const {
114	std::scoped_lock<std::mutex> lock(mutex_);
115	return Base::buf_.size();
116	}
117
118	constexpr size_t capacity() const noexcept {
119	return Base::buf_.capacity();
120	}
121
122	private:
123	mutable std::mutex mutex_;
124	std::condition_variable condition_variable_;
125	bool stopped_{false};
126	};
127	} // namespace mb::ul::thread
128
129	#endif

devmarkusb / util / 26257287247

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous