Skip to content

Commit

Permalink
use constexpr in unbounded queue
Browse files Browse the repository at this point in the history
  • Loading branch information
@odygrd
odygrd authored Sep 19, 2024
1 parent 33ab1fa commit c29664a
Show file tree
Hide file tree
Showing 4 changed files with 86 additions and 82 deletions.
52 changes: 28 additions & 24 deletions include/quill/Logger.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -114,21 +114,7 @@ class LoggerImpl : public detail::LoggerBase
total_size += sizeof(dynamic_log_level);
}

constexpr bool is_unbounded_queue = (frontend_options_t::queue_type == QueueType::UnboundedUnlimited) ||
(frontend_options_t::queue_type == QueueType::UnboundedBlocking) ||
(frontend_options_t::queue_type == QueueType::UnboundedDropping);

std::byte* write_buffer;

if constexpr (is_unbounded_queue)
{
write_buffer = thread_context->get_spsc_queue<frontend_options_t::queue_type>().prepare_write(
total_size, frontend_options_t::queue_type);
}
else
{
write_buffer = thread_context->get_spsc_queue<frontend_options_t::queue_type>().prepare_write(total_size);
}
std::byte* write_buffer = _prepare_write_buffer(total_size);

if constexpr (frontend_options_t::queue_type == QueueType::UnboundedUnlimited)
{
Expand Down Expand Up @@ -166,15 +152,7 @@ class LoggerImpl : public detail::LoggerBase
}

// not enough space to push to queue, keep trying
if constexpr (is_unbounded_queue)
{
write_buffer = thread_context->get_spsc_queue<frontend_options_t::queue_type>().prepare_write(
total_size, frontend_options_t::queue_type);
}
else
{
write_buffer = thread_context->get_spsc_queue<frontend_options_t::queue_type>().prepare_write(total_size);
}
write_buffer = _prepare_write_buffer(total_size);
} while (write_buffer == nullptr);
}
}
Expand Down Expand Up @@ -357,6 +335,32 @@ class LoggerImpl : public detail::LoggerBase

return write_buffer;
}

/**
* Prepares a write buffer for the given context and size.
*/
QUILL_NODISCARD QUILL_ATTRIBUTE_HOT std::byte* _prepare_write_buffer(size_t total_size)
{
constexpr bool is_unbounded_queue = (frontend_options_t::queue_type == QueueType::UnboundedUnlimited) ||
(frontend_options_t::queue_type == QueueType::UnboundedBlocking) ||
(frontend_options_t::queue_type == QueueType::UnboundedDropping);

if constexpr (is_unbounded_queue)
{
// MSVC doesn't like the template keyword, but every other compiler requires it
#if defined(_MSC_VER)
return thread_context->get_spsc_queue<frontend_options_t::queue_type>().prepare_write<frontend_options_t::queue_type>(
total_size);
#else
return thread_context->get_spsc_queue<frontend_options_t::queue_type>()
.template prepare_write<frontend_options_t::queue_type>(total_size);
#endif
}
else
{
return thread_context->get_spsc_queue<frontend_options_t::queue_type>().prepare_write(total_size);
}
}
};

using Logger = LoggerImpl<FrontendOptions>;
Expand Down
18 changes: 9 additions & 9 deletions include/quill/core/UnboundedSPSCQueue.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,8 @@ class UnboundedSPSCQueue
* making it visible to the consumer.
* @return a valid point to the buffer
*/
QUILL_NODISCARD QUILL_ATTRIBUTE_HOT std::byte* prepare_write(size_t nbytes, QueueType queue_type)
template <quill::QueueType queue_type>
QUILL_NODISCARD QUILL_ATTRIBUTE_HOT std::byte* prepare_write(size_t nbytes)
{
// Try to reserve the bounded queue
std::byte* write_pos = _producer->bounded_queue.prepare_write(nbytes);
Expand All @@ -123,11 +124,11 @@ class UnboundedSPSCQueue
capacity = capacity * 2ull;
}

size_t constexpr max_bounded_queue_size = 2ull * 1024 * 1024 * 1024; // 2 GB

if (QUILL_UNLIKELY(capacity > max_bounded_queue_size))
if constexpr ((queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedDropping))
{
if ((queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedDropping))
size_t constexpr max_bounded_queue_size = 2ull * 1024 * 1024 * 1024; // 2 GB

if (QUILL_UNLIKELY(capacity > max_bounded_queue_size))
{
if (nbytes > max_bounded_queue_size)
{
Expand All @@ -146,13 +147,12 @@ class UnboundedSPSCQueue
std::to_string(max_bounded_queue_size) + " bytes"});
}

// we reached the unbounded queue limit of 2147483648 bytes (~2GB) we won't be allocating
// anymore and instead return nullptr to block or drop
// we reached the max_bounded_queue_size we won't be allocating more
// instead return nullptr to block or drop
return nullptr;
}

// else the UnboundedUnlimited queue has no limits
}
// else the UnboundedUnlimited queue has no limits

// commit previous write to the old queue before switching
_producer->bounded_queue.commit_write();
Expand Down
94 changes: 47 additions & 47 deletions test/unit_tests/UnboundedQueueLimitTest.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,34 +9,33 @@ TEST_SUITE_BEGIN("UnboundedQueue");

using namespace quill::detail;

constexpr static uint64_t half_gb = 500u * 1024u * 1024u;
constexpr static uint64_t two_gb = 2u * 1024u * 1024u * 1024u - 1;
constexpr static uint64_t three_gb = 3u * 1024u * 1024u * 1024u;

TEST_CASE("unbounded_queue_max_limit")
{
UnboundedSPSCQueue buffer{1024};

auto* write_buffer_a = buffer.prepare_write(half_gb, quill::QueueType::UnboundedUnlimited);
constexpr static uint64_t half_gb = 500u * 1024u * 1024u;
constexpr static uint64_t two_gb = 2u * 1024u * 1024u * 1024u - 1;

auto* write_buffer_a = buffer.prepare_write<quill::QueueType::UnboundedUnlimited>(half_gb);
REQUIRE(write_buffer_a);
buffer.finish_write(half_gb);
buffer.commit_write();

auto* write_buffer_b = buffer.prepare_write(two_gb, quill::QueueType::UnboundedUnlimited);
auto* write_buffer_b = buffer.prepare_write<quill::QueueType::UnboundedUnlimited>(two_gb);
REQUIRE(write_buffer_b);
buffer.finish_write(two_gb);
buffer.commit_write();

// Buffer is filled with two GB here, we can try to reserve more to allocate another queue
auto* write_buffer_c = buffer.prepare_write(two_gb, quill::QueueType::UnboundedBlocking);
auto* write_buffer_c = buffer.prepare_write<quill::QueueType::UnboundedBlocking>(two_gb);
REQUIRE_FALSE(write_buffer_c);

write_buffer_c = buffer.prepare_write(two_gb, quill::QueueType::UnboundedDropping);
write_buffer_c = buffer.prepare_write<quill::QueueType::UnboundedDropping>(two_gb);
REQUIRE_FALSE(write_buffer_c);

// Buffer is filled with two GB here, we can try to reserve more to allocate another queue
// for the UnboundedLimit queue
write_buffer_c = buffer.prepare_write(two_gb, quill::QueueType::UnboundedUnlimited);
write_buffer_c = buffer.prepare_write<quill::QueueType::UnboundedUnlimited>(two_gb);
REQUIRE(write_buffer_c);
buffer.finish_write(two_gb);
buffer.commit_write();
Expand All @@ -59,43 +58,44 @@ TEST_CASE("unbounded_queue_max_limit")
REQUIRE(buffer.empty());
}

TEST_CASE("unbounded_queue_unbounded_unlimited")
{
UnboundedSPSCQueue buffer{1024};

// Try to allocate over 2GB
auto func = [&buffer]()
{
auto* write_buffer_z = buffer.prepare_write(three_gb, quill::QueueType::UnboundedUnlimited);
return write_buffer_z;
};
REQUIRE_NOTHROW(func());
}

TEST_CASE("unbounded_queue_unbounded_blocking")
{
UnboundedSPSCQueue buffer{1024};

// Try to allocate over 2GB
auto func = [&buffer]()
{
auto* write_buffer_z = buffer.prepare_write(three_gb, quill::QueueType::UnboundedBlocking);
return write_buffer_z;
};
REQUIRE_THROWS_AS(func(), quill::QuillError);
}

TEST_CASE("unbounded_queue_unbounded_dropping")
{
UnboundedSPSCQueue buffer{1024};

// Try to allocate over 2GB
auto func = [&buffer]()
{
auto* write_buffer_z = buffer.prepare_write(three_gb, quill::QueueType::UnboundedDropping);
return write_buffer_z;
};
REQUIRE_THROWS_AS(func(), quill::QuillError);
}
// These tests work but sometimes can randomly crash on CI
// TEST_CASE("unbounded_queue_unbounded_unlimited")
//{
// UnboundedSPSCQueue buffer{1024};
//
// // Try to allocate over 2GB
// auto func = [&buffer]()
// {
// auto* write_buffer_z = buffer.prepare_write<quill::QueueType::UnboundedUnlimited>(three_gb);
// return write_buffer_z;
// };
// REQUIRE_NOTHROW(func());
//}
//
// TEST_CASE("unbounded_queue_unbounded_blocking")
//{
// UnboundedSPSCQueue buffer{1024};
//
// // Try to allocate over 2GB
// auto func = [&buffer]()
// {
// auto* write_buffer_z = buffer.prepare_write<quill::QueueType::UnboundedBlocking>(three_gb);
// return write_buffer_z;
// };
// REQUIRE_THROWS_AS(func(), quill::QuillError);
//}
//
// TEST_CASE("unbounded_queue_unbounded_dropping")
//{
// UnboundedSPSCQueue buffer{1024};
//
// // Try to allocate over 2GB
// auto func = [&buffer]()
// {
// auto* write_buffer_z = buffer.prepare_write<quill::QueueType::UnboundedDropping>(three_gb);
// return write_buffer_z;
// };
// REQUIRE_THROWS_AS(func(), quill::QuillError);
//}

TEST_SUITE_END();
4 changes: 2 additions & 2 deletions test/unit_tests/UnboundedQueueTest.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,12 +20,12 @@ TEST_CASE("unbounded_queue_read_write_multithreaded_plain_ints")
{
for (uint32_t i = 0; i < 8192; ++i)
{
auto* write_buffer = buffer.prepare_write(sizeof(uint32_t), quill::QueueType::UnboundedBlocking);
auto* write_buffer = buffer.prepare_write<quill::QueueType::UnboundedBlocking>(sizeof(uint32_t));

while (!write_buffer)
{
std::this_thread::sleep_for(std::chrono::microseconds{2});
write_buffer = buffer.prepare_write(sizeof(uint32_t), quill::QueueType::UnboundedBlocking);
write_buffer = buffer.prepare_write<quill::QueueType::UnboundedBlocking>(sizeof(uint32_t));
}

std::memcpy(write_buffer, &i, sizeof(uint32_t));
Expand Down

0 comments on commit c29664a

Please sign in to comment.