Shrink TransitEventBuffer in backend when frontend unbounded queue is explicitly shrunk (#734)

Author: odygrd

include/quill/backend/BackendOptions.h

@@ -64,8 +64,8 @@ struct BackendOptions
    * If a frontend threads continuously push messages to the queue (e.g., logging in a loop),
    * no logs can ever be processed.
    *
-   * When the soft limit is reached (default: 800), the backend worker thread will try to process
-   * a batch of cached transit events all at once
+   * When the soft limit is reached the backend worker thread will try to process a batch of cached
+   * transit events all at once
    *
    * The frontend queues are emptied on each iteration, so the actual popped messages
    * can be much greater than the transit_events_soft_limit.

include/quill/backend/BackendWorker.h

@@ -319,6 +319,7 @@ class BackendWorker
       {
         _cleanup_invalidated_thread_contexts();
         _cleanup_invalidated_loggers();
+        _try_shrink_empty_transit_event_buffers();
 
         // There is nothing left to do, and we can let this thread sleep for a while
         // buffer events are 0 here and also all the producer queues are empty
@@ -1162,6 +1163,13 @@ class BackendWorker
 
     if (read_result.allocation)
     {
+      if ((read_result.new_capacity < read_result.previous_capacity) && thread_context->_transit_event_buffer)
+      {
+        // The user explicitly requested to shrink the queue, indicating a preference for low memory
+        // usage. To align with this intent, we also request shrinking the backend buffer.
+        thread_context->_transit_event_buffer->request_shrink();
+      }
+
       // When allocation_info has a value it means that the queue has re-allocated
       if (_options.error_notifier)
       {
@@ -1428,6 +1436,21 @@ class BackendWorker
     }
   }
 
+  /**
+   * Shrinks empty TransitEvent buffers. This is triggered only when the user explicitly
+   * requests shrinking of the unbounded frontend queue to optimize memory usage.
+   */
+  QUILL_ATTRIBUTE_HOT void _try_shrink_empty_transit_event_buffers()
+  {
+    for (ThreadContext* thread_context : _active_thread_contexts_cache)
+    {
+      if (thread_context->_transit_event_buffer)
+      {
+        thread_context->_transit_event_buffer->try_shrink();
+      }
+    }
+  }
+
   /**
    * This function takes an `format_args_store` containing multiple arguments and formats them into
    * a single string using a generated format string. Due to limitations in the ability to

include/quill/backend/TransitEventBuffer.h

@@ -20,7 +20,8 @@ class TransitEventBuffer
 {
 public:
   explicit TransitEventBuffer(size_t initial_capacity)
-    : _capacity(next_power_of_two(initial_capacity)),
+    : _initial_capacity(next_power_of_two(initial_capacity)),
+      _capacity(_initial_capacity),
       _storage(std::make_unique<TransitEvent[]>(_capacity)),
       _mask(_capacity - 1u)
   {
@@ -97,6 +98,26 @@ class TransitEventBuffer
     return _reader_pos == _writer_pos;
   }
 
+  void request_shrink() noexcept { _shrink_requested = true; }
+
+  void try_shrink()
+  {
+    // we only shrink empty buffers
+    if (_shrink_requested && empty())
+    {
+      if (_capacity > _initial_capacity)
+      {
+        _storage = std::make_unique<TransitEvent[]>(_initial_capacity);
+        _capacity = _initial_capacity;
+        _mask = _capacity - 1;
+        _writer_pos = 0;
+        _reader_pos = 0;
+      }
+
+      _shrink_requested = false;
+    }
+  }
+
 private:
   void _expand()
   {
@@ -120,11 +141,13 @@ class TransitEventBuffer
     _reader_pos = 0;
   }
 
+  size_t _initial_capacity;
   size_t _capacity;
   std::unique_ptr<TransitEvent[]> _storage;
   size_t _mask;
   size_t _reader_pos{0};
   size_t _writer_pos{0};
+  bool _shrink_requested{false};
 };
 
 } // namespace detail

test/integration_tests/ShrinkThreadLocalQueueTest.cpp

@@ -29,11 +29,10 @@ 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 size_t iterations = 6;
   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
@@ -69,6 +68,13 @@ TEST_CASE("shrink_thread_local_queue")
 
     CustomFrontend::shrink_thread_local_queue(16 * 1024);
     REQUIRE_EQ(CustomFrontend::get_thread_local_queue_capacity(), 16 * 1024);
+
+    if (iter % 2 == 0)
+    {
+      // flush the log so that the backend goes idle and also backend shrink is tested
+      logger->flush_log();
+      std::this_thread::sleep_for(std::chrono::milliseconds{1});
+    }
   }
 
   logger->flush_log();