Add shrink and capacity functions for unbounded queue

Author: odygrd

CHANGELOG.md

@@ -134,6 +134,10 @@
   example: [sink_formatter_override](https://github.com/odygrd/quill/blob/master/examples/sink_formatter_override.cpp).
 - Added `Frontend::remove_logger_blocking(...)`, which blocks the caller thread until the specified logger is fully
   removed.
+- Added `Frontend::shrink_thread_local_queue(capacity)` and `CustomFrontend::get_thread_local_queue_capacity()`.
+  These functions allow dynamic management of thread-local SPSC queues when using an unbounded queue configuration. They
+  enable on-demand shrinking of a queue that has grown due to bursty logging, helping to reduce memory usage, although
+  in typical scenarios they won't be required.
 - Added the `SyslogSink`, which logs messages to the system's syslog.
 
     ```c++

include/quill/Frontend.h

@@ -53,6 +53,64 @@ class FrontendImpl
     (void)spsc_queue_capacity;
   }
 
+  /**
+   * Shrink the thread-local SPSC queue to the specified target capacity.
+   *
+   * This function helps manage memory usage by reducing the size of the thread-local queue.
+   * In scenarios where a thread pool executes multiple jobs, one job might log a burst of messages
+   * that causes the queue to grow significantly. Subsequent jobs may not require such a large capacity,
+   * so you can call this function to explicitly shrink the queue to a smaller size.
+   *
+   * @note This function only applies when using the **UnboundedQueue** configuration. It will have no effect
+   *       if the BoundedQueue is enabled.
+   * @note The function will only shrink the queue if the provided target capacity is smaller than the current
+   *       queue capacity. If the target capacity is greater than or equal to the current capacity, no change is made.
+   * @warning The Logger object may maintain multiple thread-local queues. This function will only shrink the queue
+   *          associated with the calling thread, so it is important that the appropriate thread invokes it.
+   *
+   * @param capacity The desired new capacity for the thread-local SPSC queue.
+   */
+  static void shrink_thread_local_queue(size_t capacity)
+  {
+    if constexpr (logger_t::using_unbounded_queue)
+    {
+      detail::get_local_thread_context<TFrontendOptions>()
+        ->template get_spsc_queue<TFrontendOptions::queue_type>()
+        .shrink(capacity);
+    }
+  }
+
+  /**
+   * Retrieve the current capacity of the thread-local SPSC queue.
+   *
+   * This function returns the capacity of the SPSC queue that belongs to the calling thread.
+   * It is particularly useful for monitoring how much an UnboundedQueue has grown over time,
+   * while for a BoundedQueue, the capacity remains constant.
+   *
+   * @note When using an UnboundedQueue, the function returns the capacity as determined by the producer,
+   *       reflecting the dynamic growth of the queue. For a BoundedQueue, the returned capacity is fixed.
+   * @note Since the Logger object can maintain multiple thread-local queues, this function always returns
+   *       the capacity of the queue associated with the thread that calls it. Ensure that the correct thread
+   *       is invoking this function to check its own queue.
+   *
+   * @return The current capacity of the thread-local SPSC queue.
+   */
+  QUILL_NODISCARD static size_t get_thread_local_queue_capacity() noexcept
+  {
+    if constexpr (logger_t::using_unbounded_queue)
+    {
+      return detail::get_local_thread_context<TFrontendOptions>()
+        ->template get_spsc_queue<TFrontendOptions::queue_type>()
+        .producer_capacity();
+    }
+    else
+    {
+      return detail::get_local_thread_context<TFrontendOptions>()
+        ->template get_spsc_queue<TFrontendOptions::queue_type>()
+        .capacity();
+    }
+  }
+
   /**
    * @brief Creates a new sink or retrieves an existing one with the specified name.
    *

include/quill/Logger.h

@@ -50,6 +50,11 @@ class LoggerImpl : public detail::LoggerBase
 public:
   using frontend_options_t = TFrontendOptions;
 
+  static constexpr bool using_unbounded_queue =
+    (frontend_options_t::queue_type == QueueType::UnboundedUnlimited) ||
+    (frontend_options_t::queue_type == QueueType::UnboundedBlocking) ||
+    (frontend_options_t::queue_type == QueueType::UnboundedDropping);
+
   /***/
   LoggerImpl(LoggerImpl const&) = delete;
   LoggerImpl& operator=(LoggerImpl const&) = delete;
@@ -360,11 +365,7 @@ class LoggerImpl : public detail::LoggerBase
    */
   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)
+    if constexpr (using_unbounded_queue)
     {
       // MSVC doesn't like the template keyword, but every other compiler requires it
 #if defined(_MSC_VER)

include/quill/core/UnboundedSPSCQueue.h

@@ -138,7 +138,7 @@ class UnboundedSPSCQueue
   }
 
   /**
-   * Commit the write to notify the consumer bytes are ready to read
+   * Commit write to notify the consumer bytes are ready to read
    */
   QUILL_ATTRIBUTE_HOT void commit_write() noexcept { _producer->bounded_queue.commit_write(); }
 
@@ -151,6 +151,40 @@ class UnboundedSPSCQueue
     commit_write();
   }
 
+  /**
+   * Return the current buffer's capacity
+   * @note: producer only
+   * @return capacity
+   */
+  QUILL_NODISCARD size_t producer_capacity() const noexcept
+  {
+    return _producer->bounded_queue.capacity();
+  }
+
+  /**
+   * Shrinks the queue if capacity is a valid smaller power of 2.
+   * @param capacity New target capacity.
+   * @note: producer (frontend) is safe to call this - Do not call on the consumer (the backend worker)
+   */
+  void shrink(size_t capacity)
+  {
+    if (capacity > (_producer->bounded_queue.capacity() >> 1))
+    {
+      // We should only shrink if the new capacity is less or at least equal to the previous_power_of_2
+      return;
+    }
+
+    // We want to shrink the queue, we will create a new queue with a smaller size
+    // the consumer will switch to the newer queue after emptying and deallocating the older queue
+    auto const next_node = new Node{capacity, _producer->bounded_queue.huge_pages_policy()};
+
+    // store the new node pointer as next in the current node
+    _producer->next.store(next_node, std::memory_order_release);
+
+    // producer is now using the next node
+    _producer = next_node;
+  }
+
   /**
    * Prepare to read from the buffer
    * @error_notifier a callback used for notifications to the user
@@ -193,19 +227,22 @@ class UnboundedSPSCQueue
 
   /**
    * Return the current buffer's capacity
+   * @note: consumer only
    * @return capacity
    */
   QUILL_NODISCARD size_t capacity() const noexcept { return _consumer->bounded_queue.capacity(); }
 
   /**
    * checks if the queue is empty
+   * @note consumer only
    * @return true if empty, false otherwise
    */
   QUILL_NODISCARD QUILL_ATTRIBUTE_HOT bool empty() const noexcept
   {
     return _consumer->bounded_queue.empty() && (_consumer->next.load(std::memory_order_relaxed) == nullptr);
   }
 
+
 private:
   /***/
 #if defined(_MSC_VER)

test/integration_tests/CMakeLists.txt

@@ -87,6 +87,7 @@ quill_add_test(TEST_RotatingSinkDailyRotation RotatingSinkDailyRotationTest.cpp)
 quill_add_test(TEST_RotatingSinkKeepOldest RotatingSinkKeepOldestTest.cpp)
 quill_add_test(TEST_RotatingSinkOverwriteOldest RotatingSinkOverwriteOldestTest.cpp)
 quill_add_test(TEST_RuntimeMetadata RuntimeMetadataTest.cpp)
+quill_add_test(TEST_ShrinkThreadLocalQueueTest ShrinkThreadLocalQueueTest.cpp)
 quill_add_test(TEST_SignalHandler SignalHandlerTest.cpp)
 quill_add_test(TEST_SignalHandlerLogger SignalHandlerLoggerTest.cpp)
 quill_add_test(TEST_SingleFrontendThreadMultipleLoggers SingleFrontendThreadMultipleLoggersTest.cpp)

test/integration_tests/ShrinkThreadLocalQueueTest.cpp

@@ -0,0 +1,90 @@
+#include "doctest/doctest.h"
+
+#include "misc/TestUtilities.h"
+#include "quill/Backend.h"
+#include "quill/Frontend.h"
+#include "quill/LogMacros.h"
+#include "quill/sinks/FileSink.h"
+
+#include <cstdio>
+#include <string>
+#include <vector>
+
+using namespace quill;
+
+// Define custom Frontend Options
+struct CustomFrontendOptions
+{
+  static constexpr quill::QueueType queue_type = quill::QueueType::UnboundedBlocking;
+  static constexpr uint32_t initial_queue_capacity = 16 * 1024; // 16 KiB
+  static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
+  static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
+};
+
+using CustomFrontend = FrontendImpl<CustomFrontendOptions>;
+using CustomLogger = LoggerImpl<CustomFrontendOptions>;
+
+/***/
+TEST_CASE("shrink_thread_local_queue")
+{
+  static constexpr size_t number_of_messages = 5000;
+  static constexpr size_t iterations = 5;
+  static constexpr char const* filename = "shrink_thread_local_queue.log";
+  static std::string const logger_name = "logger";
+
+  // Start the logging backend thread
+  Backend::start();
+
+  // just for testing - call before logging anything
+  CustomFrontend::shrink_thread_local_queue(8 * 1024);
+  REQUIRE_EQ(CustomFrontend::get_thread_local_queue_capacity(), 8 * 1024);
+  
+  // Set writing logging to a file
+  auto file_sink = CustomFrontend::create_or_get_sink<FileSink>(
+    filename,
+    []()
+    {
+      FileSinkConfig cfg;
+      cfg.set_open_mode('w');
+
+      // For this test only we use the default buffer size, it should not make any difference it is just for testing the default behaviour and code coverage
+      cfg.set_write_buffer_size(0);
+
+      return cfg;
+    }(),
+    FileEventNotifier{});
+
+  CustomLogger* logger = CustomFrontend::create_or_get_logger(logger_name, std::move(file_sink));
+
+  REQUIRE_EQ(CustomFrontend::get_thread_local_queue_capacity(), 8 * 1024);
+  
+  size_t cnt{0};
+  for (size_t iter = 0; iter < iterations; ++iter)
+  {
+    for (size_t i = 0; i < number_of_messages; ++i)
+    {
+      LOG_INFO(logger, "This is message {}", cnt++);
+    }
+
+    CustomFrontend::shrink_thread_local_queue(16 * 1024);
+    REQUIRE_EQ(CustomFrontend::get_thread_local_queue_capacity(), 16 * 1024);
+  }
+
+  logger->flush_log();
+  CustomFrontend::remove_logger(logger);
+
+  // Wait until the backend thread stops for test stability
+  Backend::stop();
+
+  // Read file and check
+  std::vector<std::string> const file_contents = quill::testing::file_contents(filename);
+  REQUIRE_EQ(file_contents.size(), number_of_messages * iterations);
+
+  for (size_t i = 0; i < number_of_messages * iterations; ++i)
+  {
+    std::string expected_string = logger_name + "       This is message " + std::to_string(i);
+    REQUIRE(quill::testing::file_contains(file_contents, expected_string));
+  }
+
+  testing::remove_file(filename);
+}

test/unit_tests/UnboundedQueueTest.cpp

@@ -9,6 +9,63 @@ TEST_SUITE_BEGIN("UnboundedQueue");
 
 using namespace quill::detail;
 
+TEST_CASE("unbounded_queue_shrink")
+{
+  constexpr size_t CHUNK{256};
+  constexpr size_t INITIAL_SIZE{1024};
+
+  UnboundedSPSCQueue buffer{INITIAL_SIZE};
+  REQUIRE_EQ(buffer.producer_capacity(), INITIAL_SIZE);
+
+  // This queue will grow as we request 5 * 256
+  for (uint32_t i = 0; i < 5; ++i)
+  {
+#if defined(_MSC_VER)
+    auto* write_buffer = buffer.prepare_write(CHUNK, quill::QueueType::UnboundedBlocking);
+#else
+    auto* write_buffer = buffer.prepare_write<quill::QueueType::UnboundedBlocking>(CHUNK);
+#endif
+
+    REQUIRE(write_buffer);
+    buffer.finish_write(CHUNK);
+    buffer.commit_write();
+  }
+
+  {
+    UnboundedSPSCQueue::ReadResult res = buffer.prepare_read();
+    REQUIRE(res.read_pos);
+    buffer.finish_read(INITIAL_SIZE);
+    buffer.commit_read();
+  }
+
+  {
+    // Now read again the next part from the allocated queue
+    UnboundedSPSCQueue::ReadResult res = buffer.prepare_read();
+    REQUIRE(res.read_pos);
+    REQUIRE(res.allocation);
+    REQUIRE_EQ(res.previous_capacity, INITIAL_SIZE);
+    REQUIRE_EQ(res.new_capacity, INITIAL_SIZE * 2);
+    buffer.finish_read(CHUNK);
+    buffer.commit_read();
+  }
+
+  // Shrink the queue
+  REQUIRE_EQ(buffer.producer_capacity(), INITIAL_SIZE * 2);
+  buffer.shrink(INITIAL_SIZE);
+  REQUIRE_EQ(buffer.producer_capacity(), INITIAL_SIZE);
+
+  {
+    // On next read we should see the new queue - old is gone
+    UnboundedSPSCQueue::ReadResult res = buffer.prepare_read();
+    REQUIRE_EQ(res.read_pos, nullptr);
+    REQUIRE(res.allocation);
+    REQUIRE_EQ(res.previous_capacity, INITIAL_SIZE * 2);
+    REQUIRE_EQ(res.new_capacity, INITIAL_SIZE);
+    buffer.finish_read(CHUNK);
+    buffer.commit_read();
+  }
+}
+
 TEST_CASE("unbounded_queue_read_write_multithreaded_plain_ints")
 {
   UnboundedSPSCQueue buffer{1024};