dmaax/ladybird
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

LibCore+LibIPC: Remove SharedSingleProducerCircularQueue

Browse source 
This went unused.
This commit is contained in:
Jelle Raaijmakers 2026-04-30 09:49:51 +02:00 committed by Jelle Raaijmakers
parent f32bcbc5c9
commit 23a3ce1f2f
8 changed files with 0 additions and 441 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
AK/Debug.h.in
View file

@ -250,10 +250,6 @@
# cmakedefine01 RSA_PARSE_DEBUG
#endif
#ifndef SHARED_QUEUE_DEBUG
# cmakedefine01 SHARED_QUEUE_DEBUG
#endif
#ifndef SPAM_DEBUG
# cmakedefine01 SPAM_DEBUG
#endif

206
Libraries/LibCore/SharedCircularQueue.h
View file

@ -1,206 +0,0 @@
/*
* Copyright (c) 2022, kleines Filmröllchen <filmroellchen@serenityos.org>
* Copyright (c) 2024, stasoid <stasoid@yahoo.com>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <AK/AtomicRefCounted.h>
#include <AK/BuiltinWrappers.h>
#include <AK/ByteString.h>
#include <AK/Debug.h>
#include <AK/Function.h>
#include <LibCore/AnonymousBuffer.h>
namespace Core {
// A circular lock-free queue (or a buffer) with a single producer,
// residing in shared memory and designed to be accessible to multiple processes.
// This implementation makes use of the fact that any producer-related code can be sure that
// it's the only producer-related code that is running, which simplifies a bunch of the synchronization code.
// The exclusivity and liveliness for critical sections in this class is proven to be correct
// under the assumption of correct synchronization primitives, i.e. atomics.
// In many circumstances, this is enough for cross-process queues.
// This class is designed to be transferred over IPC and mmap()ed into multiple processes' memory.
// It is a synthetic pointer to the actual shared memory, which is abstracted away from the user.
// FIXME: Make this independent of shared memory, so that we can move it to AK.
template<typename T, size_t Size = 32>
// Size must be a power of two, which speeds up the modulus operations for indexing.
requires(popcount(Size) == 1)
class SharedSingleProducerCircularQueue final {
public:
using ValueType = T;
enum class QueueStatus : u8 {
Invalid = 0,
Full,
Empty,
};
SharedSingleProducerCircularQueue() = default;
SharedSingleProducerCircularQueue(SharedSingleProducerCircularQueue<ValueType, Size>& queue) = default;
SharedSingleProducerCircularQueue(SharedSingleProducerCircularQueue&& queue) = default;
SharedSingleProducerCircularQueue& operator=(SharedSingleProducerCircularQueue&& queue) = default;
// Allocates a new circular queue in shared memory.
static ErrorOr<SharedSingleProducerCircularQueue<T, Size>> create()
{
auto anon_buf = TRY(AnonymousBuffer::create_with_size(sizeof(SharedMemorySPCQ)));
auto shared_queue = new (anon_buf.data<void>()) SharedMemorySPCQ;
return create_internal(anon_buf, shared_queue);
}
// Uses an existing circular queue from given shared memory.
static ErrorOr<SharedSingleProducerCircularQueue<T, Size>> create(int fd)
{
auto anon_buf = TRY(AnonymousBuffer::create_from_anon_fd(fd, sizeof(SharedMemorySPCQ)));
auto shared_queue = (SharedMemorySPCQ*)anon_buf.data<void>();
return create_internal(anon_buf, shared_queue);
}
constexpr size_t size() const { return Size; }
// These functions are provably inconsistent and should only be used as hints to the actual capacity and used count.
ALWAYS_INLINE size_t weak_remaining_capacity() const { return Size - weak_used(); }
ALWAYS_INLINE size_t weak_used() const
{
auto volatile head = m_queue->m_queue->m_tail.load(AK::MemoryOrder::memory_order_relaxed);
auto volatile tail = m_queue->m_queue->m_head.load(AK::MemoryOrder::memory_order_relaxed);
return head - tail;
}
ALWAYS_INLINE constexpr int fd() const { return m_queue->fd(); }
ALWAYS_INLINE constexpr bool is_valid() const { return !m_queue.is_null(); }
ALWAYS_INLINE constexpr size_t weak_head() const { return m_queue->m_queue->m_head.load(AK::MemoryOrder::memory_order_relaxed); }
ALWAYS_INLINE constexpr size_t weak_tail() const { return m_queue->m_queue->m_tail.load(AK::MemoryOrder::memory_order_relaxed); }
ErrorOr<void, QueueStatus> enqueue(ValueType to_insert)
{
VERIFY(!m_queue.is_null());
if (!can_enqueue())
return QueueStatus::Full;
auto our_tail = m_queue->m_queue->m_tail.load() % Size;
m_queue->m_queue->m_data[our_tail] = to_insert;
m_queue->m_queue->m_tail.fetch_add(1);
return {};
}
ALWAYS_INLINE bool can_enqueue() const
{
return ((head() - 1) % Size) != (m_queue->m_queue->m_tail.load() % Size);
}
// Repeatedly try to enqueue, using the wait_function to wait if it's not possible
ErrorOr<void> blocking_enqueue(ValueType to_insert, Function<void()> wait_function)
{
ErrorOr<void, QueueStatus> result;
while (true) {
result = enqueue(to_insert);
if (!result.is_error())
break;
if (result.error() != QueueStatus::Full)
return Error::from_string_literal("Unexpected error while enqueuing");
wait_function();
}
return {};
}
ErrorOr<ValueType, QueueStatus> dequeue()
{
VERIFY(!m_queue.is_null());
while (true) {
// This CAS only succeeds if nobody is currently dequeuing.
auto size_max = NumericLimits<size_t>::max();
if (m_queue->m_queue->m_head_protector.compare_exchange_strong(size_max, m_queue->m_queue->m_head.load())) {
auto old_head = m_queue->m_queue->m_head.load();
// This check looks like it's in a weird place (especially since we have to roll back the protector), but it's actually protecting against a race between multiple dequeuers.
if (old_head >= m_queue->m_queue->m_tail.load()) {
m_queue->m_queue->m_head_protector.store(NumericLimits<size_t>::max(), AK::MemoryOrder::memory_order_release);
return QueueStatus::Empty;
}
auto data = move(m_queue->m_queue->m_data[old_head % Size]);
m_queue->m_queue->m_head.fetch_add(1);
m_queue->m_queue->m_head_protector.store(NumericLimits<size_t>::max(), AK::MemoryOrder::memory_order_release);
return { move(data) };
}
}
}
// The "real" head as seen by the outside world. Don't use m_head directly unless you know what you're doing.
size_t head() const
{
return min(m_queue->m_queue->m_head.load(), m_queue->m_queue->m_head_protector.load());
}
private:
struct SharedMemorySPCQ {
SharedMemorySPCQ() = default;
SharedMemorySPCQ(SharedMemorySPCQ const&) = delete;
SharedMemorySPCQ(SharedMemorySPCQ&&) = delete;
~SharedMemorySPCQ() = default;
// Invariant: tail >= head
// Invariant: head and tail are monotonically increasing
// Invariant: tail always points to the next free location where an enqueue can happen.
// Invariant: head always points to the element to be dequeued next.
// Invariant: tail is only modified by enqueue functions.
// Invariant: head is only modified by dequeue functions.
// An empty queue is signalled with: tail = head
// A full queue is signalled with: head - 1 mod size = tail mod size (i.e. head and tail point to the same index in the data array)
// FIXME: These invariants aren't proven to be correct after each successful completion of each operation where it is relevant.
// The work could be put in but for now I think the algorithmic correctness proofs of the functions are enough.
AK_CACHE_ALIGNED Atomic<size_t, AK::MemoryOrder::memory_order_seq_cst> m_tail { 0 };
AK_CACHE_ALIGNED Atomic<size_t, AK::MemoryOrder::memory_order_seq_cst> m_head { 0 };
AK_CACHE_ALIGNED Atomic<size_t, AK::MemoryOrder::memory_order_seq_cst> m_head_protector { NumericLimits<size_t>::max() };
alignas(ValueType) Array<ValueType, Size> m_data;
};
class RefCountedSharedMemorySPCQ
: public AtomicRefCounted<RefCountedSharedMemorySPCQ>
, public AnonymousBuffer {
friend class SharedSingleProducerCircularQueue;
public:
SharedMemorySPCQ* m_queue;
ByteString m_name;
~RefCountedSharedMemorySPCQ()
{
dbgln_if(SHARED_QUEUE_DEBUG, "destructed SSPCQ at {:p} named {}, shared mem: {:p}", this, m_name, m_queue);
}
private:
RefCountedSharedMemorySPCQ(AnonymousBuffer anon_buf, SharedMemorySPCQ* shared_queue, ByteString name)
: AnonymousBuffer(anon_buf)
, m_queue(shared_queue)
, m_name(move(name))
{
}
};
static ErrorOr<SharedSingleProducerCircularQueue<T, Size>> create_internal(AnonymousBuffer anon_buf, SharedMemorySPCQ* shared_queue)
{
if (!shared_queue)
return Error::from_string_literal("Unexpected error when creating shared queue from raw memory");
auto name = ByteString::formatted("SharedSingleProducerCircularQueue@{:x}", anon_buf.fd());
dbgln_if(SHARED_QUEUE_DEBUG, "successfully mmapped {} at {:p}", name, shared_queue);
auto ref_counted = new (nothrow) RefCountedSharedMemorySPCQ(anon_buf, shared_queue, move(name));
return SharedSingleProducerCircularQueue<T, Size> { adopt_ref(*ref_counted) };
}
SharedSingleProducerCircularQueue(RefPtr<RefCountedSharedMemorySPCQ> queue)
: m_queue(queue)
{
}
RefPtr<RefCountedSharedMemorySPCQ> m_queue;
};
}

9
Libraries/LibIPC/Concepts.h
View file

@ -11,7 +11,6 @@
#include <AK/Span.h>
#include <AK/Variant.h>
#include <AK/Vector.h>
#include <LibCore/SharedCircularQueue.h>
// These concepts are used to help the compiler distinguish between specializations that would be
// ambiguous otherwise. For example, if the specializations for int and Vector<T> were declared as
@ -45,11 +44,6 @@ constexpr inline bool IsHashMap = false;
template<typename K, typename V, typename KeyTraits, typename ValueTraits, bool IsOrdered>
constexpr inline bool IsHashMap<HashMap<K, V, KeyTraits, ValueTraits, IsOrdered>> = true;
template<typename T>
constexpr inline bool IsSharedSingleProducerCircularQueue = false;
template<typename T, size_t Size>
constexpr inline bool IsSharedSingleProducerCircularQueue<Core::SharedSingleProducerCircularQueue<T, Size>> = true;
}
template<typename T>
@ -64,9 +58,6 @@ concept Span = SpecializationOf<T, AK::Span>;
template<typename T>
concept HashMap = Detail::IsHashMap<T>;
template<typename T>
concept SharedSingleProducerCircularQueue = Detail::IsSharedSingleProducerCircularQueue<T>;
template<typename T>
concept Optional = SpecializationOf<T, AK::Optional>;

8
Libraries/LibIPC/Decoder.h
View file

@ -19,7 +19,6 @@
#include <AK/TypeList.h>
#include <AK/Variant.h>
#include <LibCore/Forward.h>
#include <LibCore/SharedCircularQueue.h>
#include <LibIPC/Attachment.h>
#include <LibIPC/Concepts.h>
#include <LibIPC/File.h>
@ -201,13 +200,6 @@ ErrorOr<T> decode(Decoder& decoder)
return hashmap;
}
template<Concepts::SharedSingleProducerCircularQueue T>
ErrorOr<T> decode(Decoder& decoder)
{
auto anon_file = TRY(decoder.decode<IPC::File>());
return T::create(anon_file.take_fd());
}
template<Concepts::Optional T>
ErrorOr<T> decode(Decoder& decoder)
{

8
Libraries/LibIPC/Encoder.h
View file

@ -13,7 +13,6 @@
#include <AK/StdLibExtras.h>
#include <AK/Variant.h>
#include <LibCore/Forward.h>
#include <LibCore/SharedCircularQueue.h>
#include <LibIPC/Attachment.h>
#include <LibIPC/Concepts.h>
#include <LibIPC/File.h>
@ -195,13 +194,6 @@ ErrorOr<void> encode(Encoder& encoder, T const& hashmap)
return {};
}
template<Concepts::SharedSingleProducerCircularQueue T>
ErrorOr<void> encode(Encoder& encoder, T const& queue)
{
TRY(encoder.encode(TRY(IPC::File::clone_fd(queue.fd()))));
return {};
}
template<Concepts::Optional T>
ErrorOr<void> encode(Encoder& encoder, T const& optional)
{

1
Meta/CMake/all_the_debug_macros.cmake
View file

@ -58,7 +58,6 @@ set(REQUESTSERVER_DEBUG ON)
set(REQUESTSERVER_WIRE_DEBUG ON)
set(RESOURCE_DEBUG ON)
set(RSA_PARSE_DEBUG ON)
set(SHARED_QUEUE_DEBUG ON)
set(SPAM_DEBUG ON)
set(STYLE_INVALIDATION_DEBUG ON)
set(SYNTAX_HIGHLIGHTING_DEBUG ON)

2
Tests/LibCore/CMakeLists.txt
View file

@ -6,7 +6,6 @@ set(TEST_SOURCES
TestLibCoreMappedFile.cpp
TestLibCoreMimeType.cpp
TestLibCorePromise.cpp
TestLibCoreSharedSingleProducerCircularQueue.cpp
TestLibCoreStream.cpp
)
@ -30,4 +29,3 @@ if(NOT WIN32)
endif()
target_link_libraries(TestLibCoreEventLoop PRIVATE LibThreading)
target_link_libraries(TestLibCoreSharedSingleProducerCircularQueue PRIVATE LibThreading)

203
Tests/LibCore/TestLibCoreSharedSingleProducerCircularQueue.cpp
View file

@ -1,203 +0,0 @@
/*
* Copyright (c) 2022, kleines Filmröllchen <filmroellchen@serenityos.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#include <LibCore/SharedCircularQueue.h>
#include <LibTest/TestCase.h>
#include <LibThreading/Thread.h>
#include <sched.h>
using TestQueue = Core::SharedSingleProducerCircularQueue<int>;
using QueueError = ErrorOr<int, TestQueue::QueueStatus>;
Function<intptr_t()> dequeuer(TestQueue& queue, Atomic<size_t>& dequeue_count, size_t test_count);
// These first two cases don't multithread at all.
TEST_CASE(simple_enqueue)
{
auto queue = MUST(TestQueue::create());
for (size_t i = 0; i < queue.size() - 1; ++i)
MUST(queue.enqueue((int)i));
auto result = queue.enqueue(0);
EXPECT(result.is_error());
EXPECT_EQ(result.release_error(), TestQueue::QueueStatus::Full);
}
TEST_CASE(simple_dequeue)
{
auto queue = MUST(TestQueue::create());
auto const test_count = 10;
for (int i = 0; i < test_count; ++i)
(void)queue.enqueue(i);
for (int i = 0; i < test_count; ++i) {
// TODO: This could be TRY_OR_FAIL(), if someone implements Formatter<SharedSingleProducerCircularQueue::QueueStatus>.
auto const element = MUST(queue.dequeue());
EXPECT_EQ(element, i);
}
}
// There is one parallel consumer, but nobody is producing at the same time.
TEST_CASE(simple_multithread)
{
IGNORE_USE_IN_ESCAPING_LAMBDA auto queue = MUST(TestQueue::create());
auto const test_count = 10;
for (int i = 0; i < test_count; ++i)
(void)queue.enqueue(i);
auto second_thread = Threading::Thread::construct("QueueConsumer"sv, [&queue]() {
auto copied_queue = queue;
for (int i = 0; i < test_count; ++i) {
QueueError result = TestQueue::QueueStatus::Invalid;
do {
result = copied_queue.dequeue();
if (!result.is_error())
EXPECT_EQ(result.value(), i);
} while (result.is_error() && result.error() == TestQueue::QueueStatus::Empty);
if (result.is_error())
FAIL("Unexpected error while dequeueing.");
}
return 0;
});
second_thread->start();
(void)second_thread->join();
EXPECT_EQ(queue.weak_used(), (size_t)0);
}
// There is one parallel consumer and one parallel producer.
TEST_CASE(producer_consumer_multithread)
{
IGNORE_USE_IN_ESCAPING_LAMBDA auto queue = MUST(TestQueue::create());
// Ensure that we have the possibility of filling the queue up.
auto const test_count = queue.size() * 4;
IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<bool> other_thread_running { false };
auto second_thread = Threading::Thread::construct("QueueConsumer"sv, [&queue, &other_thread_running]() {
auto copied_queue = queue;
other_thread_running.store(true);
for (size_t i = 0; i < test_count; ++i) {
QueueError result = TestQueue::QueueStatus::Invalid;
do {
result = copied_queue.dequeue();
if (!result.is_error())
EXPECT_EQ(result.value(), (int)i);
} while (result.is_error() && result.error() == TestQueue::QueueStatus::Empty);
if (result.is_error())
FAIL("Unexpected error while dequeueing.");
}
return 0;
});
second_thread->start();
while (!other_thread_running.load())
;
for (size_t i = 0; i < test_count; ++i) {
ErrorOr<void, TestQueue::QueueStatus> result = TestQueue::QueueStatus::Invalid;
do {
result = queue.enqueue((int)i);
} while (result.is_error() && result.error() == TestQueue::QueueStatus::Full);
if (result.is_error())
FAIL("Unexpected error while enqueueing.");
}
(void)second_thread->join();
EXPECT_EQ(queue.weak_used(), (size_t)0);
}
// There are multiple parallel consumers, but nobody is producing at the same time.
TEST_CASE(multi_consumer)
{
auto queue = MUST(TestQueue::create());
// This needs to be divisible by 4!
size_t const test_count = queue.size() - 4;
Atomic<size_t> dequeue_count = 0;
auto threads = {
Threading::Thread::construct("Dequeuer"sv, dequeuer(queue, dequeue_count, test_count)),
Threading::Thread::construct("Dequeuer"sv, dequeuer(queue, dequeue_count, test_count)),
Threading::Thread::construct("Dequeuer"sv, dequeuer(queue, dequeue_count, test_count)),
Threading::Thread::construct("Dequeuer"sv, dequeuer(queue, dequeue_count, test_count)),
};
for (size_t i = 0; i < test_count; ++i)
(void)queue.enqueue((int)i);
for (auto thread : threads)
thread->start();
for (auto thread : threads)
(void)thread->join();
EXPECT_EQ(queue.weak_used(), (size_t)0);
EXPECT_EQ(dequeue_count.load(), (size_t)test_count);
}
// There are multiple parallel consumers and one parallel producer.
TEST_CASE(single_producer_multi_consumer)
{
auto queue = MUST(TestQueue::create());
// Choose a higher number to provoke possible race conditions.
size_t const test_count = queue.size() * 8;
Atomic<size_t> dequeue_count = 0;
auto threads = {
Threading::Thread::construct("Dequeuer"sv, dequeuer(queue, dequeue_count, test_count)),
Threading::Thread::construct("Dequeuer"sv, dequeuer(queue, dequeue_count, test_count)),
Threading::Thread::construct("Dequeuer"sv, dequeuer(queue, dequeue_count, test_count)),
Threading::Thread::construct("Dequeuer"sv, dequeuer(queue, dequeue_count, test_count)),
};
for (auto thread : threads)
thread->start();
for (size_t i = 0; i < test_count; ++i) {
ErrorOr<void, TestQueue::QueueStatus> result = TestQueue::QueueStatus::Invalid;
do {
result = queue.enqueue((int)i);
// After we put something in the first time, let's wait while nobody has dequeued yet.
while (dequeue_count.load() == 0)
;
// Give others time to do something.
sched_yield();
} while (result.is_error() && result.error() == TestQueue::QueueStatus::Full);
if (result.is_error())
FAIL("Unexpected error while enqueueing.");
}
for (auto thread : threads)
(void)thread->join();
EXPECT_EQ(queue.weak_used(), (size_t)0);
EXPECT_EQ(dequeue_count.load(), (size_t)test_count);
}
Function<intptr_t()> dequeuer(TestQueue& queue, Atomic<size_t>& dequeue_count, size_t const test_count)
{
return [&queue, &dequeue_count, test_count]() {
auto copied_queue = queue;
for (size_t i = 0; i < test_count / 4; ++i) {
QueueError result = TestQueue::QueueStatus::Invalid;
do {
result = copied_queue.dequeue();
if (!result.is_error())
dequeue_count.fetch_add(1);
// Give others time to do something.
sched_yield();
} while (result.is_error() && result.error() == TestQueue::QueueStatus::Empty);
if (result.is_error())
FAIL("Unexpected error while dequeueing.");
}
return (intptr_t)0;
};
}