Add unbounded_queue_max_capacity to FrontEndOptions

Author: odygrd

CHANGELOG.md

@@ -93,6 +93,10 @@
   allowing huge page allocation to be attempted with a fallback to normal pages if unavailable. If you are using a
   custom `FrontendOptions` type, you will need to update it to use the new
   flag. ([#707](https://github.com/odygrd/quill/issues/707))
+- Previously, `QueueType::UnboundedDropping` and `QueueType::UnboundedBlocking` could grow up to 2 GB in size. This
+  limit is now configurable via `FrontendOptions::unbounded_queue_max_capacity`, which defaults to 2 GB.
+- `QueueType::UnboundedUnlimited` has been removed, as the same behavior can now be achieved by setting
+  `FrontendOptions::unbounded_queue_max_capacity` to the maximum value.
 - The `ConsoleSink` constructor now optionally accepts a `ConsoleSinkConfig`, similar to other sinks. If no
   `ConsoleSinkConfig` is provided, a default one is used, logging to `stdout` with `ColourMode::Automatic`. For example:
     ```c++

benchmarks/hot_path_latency/quill_hot_path_rdtsc_clock.cpp

@@ -13,8 +13,9 @@
 struct FrontendOptions
 {
   static constexpr quill::QueueType queue_type = quill::QueueType::UnboundedBlocking;
-  static constexpr uint32_t initial_queue_capacity = 131'072;
+  static constexpr size_t initial_queue_capacity = 131'072;
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
+  static constexpr size_t unbounded_queue_max_capacity = 2ull * 1024 * 1024 * 1024; // 2 GiB
   static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
 };
 

benchmarks/hot_path_latency/quill_hot_path_system_clock.cpp

@@ -13,8 +13,9 @@
 struct FrontendOptions
 {
   static constexpr quill::QueueType queue_type = quill::QueueType::UnboundedBlocking;
-  static constexpr uint32_t initial_queue_capacity = 131'072;
+  static constexpr size_t initial_queue_capacity = 131'072;
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
+  static constexpr size_t unbounded_queue_max_capacity = 2ull * 1024 * 1024 * 1024; // 2 GiB
   static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
 };
 

docs/frontend_options.rst

@@ -9,7 +9,6 @@ Each frontend thread operates with its own queue, which can be configured with o
 
 - **UnboundedBlocking**: Starts with a small initial capacity. The queue reallocates up to 2GB and then blocks further log messages.
 - **UnboundedDropping**: Starts with a small initial capacity. The queue reallocates up to 2GB and then discards log messages.
-- **UnboundedUnlimited**: Starts with a small initial capacity and reallocates without limit. This queue never blocks or drops log messages.
 - **BoundedBlocking**: Has a fixed capacity and never reallocates. It blocks log messages when the limit is reached.
 - **BoundedDropping**: Has a fixed capacity and never reallocates. It discards log messages when the limit is reached.
 

examples/bounded_dropping_queue_frontend.cpp

@@ -24,7 +24,7 @@ struct CustomFrontendOptions
   static constexpr quill::QueueType queue_type = quill::QueueType::BoundedDropping;
 
   // Set small capacity to demonstrate dropping messages in this example
-  static constexpr uint32_t initial_queue_capacity = 256;
+  static constexpr size_t initial_queue_capacity = 256;
 
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
   static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;

examples/custom_frontend_options.cpp

@@ -14,7 +14,7 @@
 struct CustomFrontendOptions
 {
   static constexpr quill::QueueType queue_type = quill::QueueType::BoundedDropping;
-  static constexpr uint32_t initial_queue_capacity = 131'072;
+  static constexpr size_t initial_queue_capacity = 131'072;
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
   static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
 };

include/quill/Logger.h

@@ -51,7 +51,6 @@ class LoggerImpl : public detail::LoggerBase
   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);
 
@@ -140,13 +139,8 @@ class LoggerImpl : public detail::LoggerBase
 
     std::byte* write_buffer = _prepare_write_buffer(total_size);
 
-    if constexpr (frontend_options_t::queue_type == QueueType::UnboundedUnlimited)
-    {
-      assert(write_buffer &&
-             "Unbounded unlimited queue will always allocate and have enough capacity");
-    }
-    else if constexpr ((frontend_options_t::queue_type == QueueType::BoundedDropping) ||
-                       (frontend_options_t::queue_type == QueueType::UnboundedDropping))
+    if constexpr ((frontend_options_t::queue_type == QueueType::BoundedDropping) ||
+                  (frontend_options_t::queue_type == QueueType::UnboundedDropping))
     {
       if (QUILL_UNLIKELY(write_buffer == nullptr))
       {
@@ -370,10 +364,10 @@ class LoggerImpl : public detail::LoggerBase
       // 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(
-        total_size, frontend_options_t::queue_type);
+        total_size, frontend_options_t::unbounded_queue_max_capacity);
 #else
       return thread_context->get_spsc_queue<frontend_options_t::queue_type>()
-        .template prepare_write<frontend_options_t::queue_type>(total_size);
+        .template prepare_write<frontend_options_t::unbounded_queue_max_capacity>(total_size);
 #endif
     }
     else

include/quill/core/BoundedSPSCQueue.h

@@ -267,7 +267,7 @@ class BoundedSPSCQueueImpl
       mem = ::mmap(nullptr, total_size, PROT_READ | PROT_WRITE, flags, -1, 0);
     }
   #endif
-    
+
     if (mem == MAP_FAILED)
     {
       QUILL_THROW(QuillError{std::string{"mmap failed. errno: "} + std::to_string(errno) +

include/quill/core/Common.h

@@ -50,7 +50,6 @@ enum class QueueType
 {
   UnboundedBlocking,
   UnboundedDropping,
-  UnboundedUnlimited,
   BoundedBlocking,
   BoundedDropping
 };

include/quill/core/FrontendOptions.h

@@ -17,9 +17,8 @@ struct FrontendOptions
 {
   /**
    * Each frontend thread has its own queue, which can be configured with various options:
-   * - UnboundedBlocking: Starts with initial_queue_capacity and reallocates up to 2GB, then blocks.
-   * - UnboundedDropping: Starts with initial_queue_capacity and reallocates up to 2GB, then drops log messages.
-   * - UnboundedUnlimited: Starts with initial_queue_capacity and reallocates without limit, subsequent queues are reallocated as needed. Never blocks or drops.
+   * - UnboundedBlocking: Starts with initial_queue_capacity and reallocates up to unbounded_queue_max_capacity, then blocks.
+   * - UnboundedDropping: Starts with initial_queue_capacity and reallocates up to unbounded_queue_max_capacity, then drops log messages.
    * - BoundedBlocking: Starts with initial_queue_capacity and never reallocates; blocks when the limit is reached.
    * - BoundedDropping: Starts with initial_queue_capacity and never reallocates; drops log messages when the limit is reached.
    *
@@ -28,21 +27,26 @@ struct FrontendOptions
   static constexpr QueueType queue_type = QueueType::UnboundedBlocking;
 
   /**
-   * Initial capacity of the queue. Used for UnboundedBlocking, UnboundedDropping, and
-   * UnboundedUnlimited. Also serves as the capacity for BoundedBlocking and BoundedDropping.
+   * Initial capacity of the queue.
    */
-  static constexpr uint32_t initial_queue_capacity = 128 * 1024; // 128 KiB
+  static constexpr size_t initial_queue_capacity = 128u * 1024u; // 128 KiB
 
   /**
    * Interval for retrying when using BoundedBlocking or UnboundedBlocking.
    * Applicable only when using BoundedBlocking or UnboundedBlocking.
    */
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
 
+  /**
+   * Maximum capacity for unbounded queues (UnboundedBlocking, UnboundedDropping).
+   * This defines the maximum size to which the queue can grow before blocking or dropping messages.
+   */
+  static constexpr size_t unbounded_queue_max_capacity = 2ull * 1024u * 1024u * 1024u; // 2 GiB
+
   /**
    * Enables huge pages on the frontend queues to reduce TLB misses. Available only for Linux.
    */
-  static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
+  static constexpr HugePagesPolicy huge_pages_policy = HugePagesPolicy::Never;
 };
 
 QUILL_END_NAMESPACE

include/quill/core/ThreadContextManager.h

@@ -59,18 +59,17 @@ class ThreadContext
 
 public:
   /***/
-  ThreadContext(QueueType queue_type, uint32_t initial_spsc_queue_capacity, HugePagesPolicy huge_pages_policy)
+  ThreadContext(QueueType queue_type, size_t initial_queue_capacity, HugePagesPolicy huge_pages_policy)
     : _queue_type(queue_type)
   {
     if (has_unbounded_queue_type())
     {
       new (&_spsc_queue_union.unbounded_spsc_queue)
-        UnboundedSPSCQueue{initial_spsc_queue_capacity, huge_pages_policy};
+        UnboundedSPSCQueue{initial_queue_capacity, huge_pages_policy};
     }
     else if (has_bounded_queue_type())
     {
-      new (&_spsc_queue_union.bounded_spsc_queue)
-        BoundedSPSCQueue{initial_spsc_queue_capacity, huge_pages_policy};
+      new (&_spsc_queue_union.bounded_spsc_queue) BoundedSPSCQueue{initial_queue_capacity, huge_pages_policy};
     }
   }
 
@@ -93,16 +92,11 @@ class ThreadContext
 
   /***/
   template <QueueType queue_type>
-  QUILL_NODISCARD QUILL_ATTRIBUTE_HOT
-    std::conditional_t<(queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedUnlimited) ||
-                         (queue_type == QueueType::UnboundedDropping),
-                       UnboundedSPSCQueue, BoundedSPSCQueue>&
-    get_spsc_queue() noexcept
+  QUILL_NODISCARD QUILL_ATTRIBUTE_HOT std::conditional_t<(queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedDropping), UnboundedSPSCQueue, BoundedSPSCQueue>& get_spsc_queue() noexcept
   {
     assert((_queue_type == queue_type) && "ThreadContext queue_type mismatch");
 
-    if constexpr ((queue_type == QueueType::UnboundedBlocking) ||
-                  (queue_type == QueueType::UnboundedUnlimited) || (queue_type == QueueType::UnboundedDropping))
+    if constexpr ((queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedDropping))
     {
       return _spsc_queue_union.unbounded_spsc_queue;
     }
@@ -114,16 +108,12 @@ class ThreadContext
 
   /***/
   template <QueueType queue_type>
-  QUILL_NODISCARD QUILL_ATTRIBUTE_HOT
-    std::conditional_t<(queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedUnlimited) ||
-                         (queue_type == QueueType::UnboundedDropping),
-                       UnboundedSPSCQueue, BoundedSPSCQueue> const&
-    get_spsc_queue() const noexcept
+  QUILL_NODISCARD QUILL_ATTRIBUTE_HOT std::conditional_t<(queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedDropping), UnboundedSPSCQueue, BoundedSPSCQueue> const& get_spsc_queue()
+    const noexcept
   {
     assert((_queue_type == queue_type) && "ThreadContext queue_type mismatch");
 
-    if constexpr ((queue_type == QueueType::UnboundedBlocking) ||
-                  (queue_type == QueueType::UnboundedUnlimited) || (queue_type == QueueType::UnboundedDropping))
+    if constexpr ((queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedDropping))
     {
       return _spsc_queue_union.unbounded_spsc_queue;
     }
@@ -148,8 +138,7 @@ class ThreadContext
   /***/
   QUILL_NODISCARD QUILL_ATTRIBUTE_HOT bool has_unbounded_queue_type() const noexcept
   {
-    return (_queue_type == QueueType::UnboundedBlocking) ||
-      (_queue_type == QueueType::UnboundedDropping) || (_queue_type == QueueType::UnboundedUnlimited);
+    return (_queue_type == QueueType::UnboundedBlocking) || (_queue_type == QueueType::UnboundedDropping);
   }
 
   /***/

include/quill/core/UnboundedSPSCQueue.h

@@ -107,9 +107,9 @@ class UnboundedSPSCQueue
    */
 #if defined(_MSC_VER)
   // MSVC doesn't like this as template <QueueType queue_type> when called from Logger, while it compiles on MSVC there will be false positives from clang-tidy
-  QUILL_NODISCARD QUILL_ATTRIBUTE_HOT std::byte* prepare_write(size_t nbytes, QueueType queue_type)
+  QUILL_NODISCARD QUILL_ATTRIBUTE_HOT std::byte* prepare_write(size_t nbytes, size_t unbounded_queue_max_capacity)
 #else
-  template <QueueType queue_type>
+  template <size_t unbounded_queue_max_capacity>
   QUILL_NODISCARD QUILL_ATTRIBUTE_HOT std::byte* prepare_write(size_t nbytes)
 #endif
   {
@@ -122,9 +122,9 @@ class UnboundedSPSCQueue
     }
 
 #if defined(_MSC_VER)
-    return _handle_full_queue(nbytes, queue_type);
+    return _handle_full_queue(nbytes, unbounded_queue_max_capacity);
 #else
-    return _handle_full_queue<queue_type>(nbytes);
+    return _handle_full_queue<unbounded_queue_max_capacity>(nbytes);
 #endif
   }
 
@@ -242,57 +242,45 @@ class UnboundedSPSCQueue
     return _consumer->bounded_queue.empty() && (_consumer->next.load(std::memory_order_relaxed) == nullptr);
   }
 
-
 private:
   /***/
 #if defined(_MSC_VER)
-  QUILL_NODISCARD std::byte* _handle_full_queue(size_t nbytes, QueueType queue_type)
+  QUILL_NODISCARD std::byte* _handle_full_queue(size_t nbytes, size_t unbounded_queue_max_capacity)
 #else
-  template <QueueType queue_type>
+  template <size_t unbounded_queue_max_capacity>
   QUILL_NODISCARD std::byte* _handle_full_queue(size_t nbytes)
 #endif
   {
     // Then it means the queue doesn't have enough size
     size_t capacity = _producer->bounded_queue.capacity() * 2ull;
-    while (capacity < (nbytes + 1))
+    while (capacity < nbytes)
     {
       capacity = capacity * 2ull;
     }
 
-#if defined(_MSC_VER)
-    if ((queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedDropping))
-#else
-    if constexpr ((queue_type == QueueType::UnboundedBlocking) || (queue_type == QueueType::UnboundedDropping))
-#endif
+    if (QUILL_UNLIKELY(capacity > unbounded_queue_max_capacity))
     {
-      size_t constexpr max_bounded_queue_size = 2ull * 1024 * 1024 * 1024; // 2 GB
-
-      if (QUILL_UNLIKELY(capacity > max_bounded_queue_size))
+      if (nbytes > unbounded_queue_max_capacity)
       {
-        if (nbytes > max_bounded_queue_size)
-        {
-          QUILL_THROW(QuillError{
-            "Logging single messages larger than 2 GB is not supported with the current queue "
-            "type. For UnboundedBlocking or UnboundedDropping queues, this limitation applies.\n"
-            "To log single messages larger than 2 GB, consider using the UnboundedUnlimited queue "
-            "type.\n"
-            "Message size: " +
-            std::to_string(nbytes) +
-            " bytes\n"
-            "Required queue capacity: " +
-            std::to_string(capacity) +
-            " bytes\n"
-            "Maximum allowed queue capacity: " +
-            std::to_string(max_bounded_queue_size) + " bytes"});
-        }
-
-        // we reached the max_bounded_queue_size we won't be allocating more
-        // instead return nullptr to block or drop
-        return nullptr;
+        QUILL_THROW(
+          QuillError{"Logging single messages larger than the configured maximum queue capacity "
+                     "is not possible.\n"
+                     "To log single messages exceeding this limit, consider increasing "
+                     "FrontendOptions::unbounded_queue_max_capacity.\n"
+                     "Message size: " +
+                     std::to_string(nbytes) +
+                     " bytes\n"
+                     "Required queue capacity: " +
+                     std::to_string(capacity) +
+                     " bytes\n"
+                     "Configured maximum queue capacity: " +
+                     std::to_string(unbounded_queue_max_capacity) + " bytes"});
       }
-    }
 
-    // else the UnboundedUnlimited queue has no limits
+      // we reached the unbounded_queue_max_capacity we won't be allocating more
+      // instead return nullptr to block or drop
+      return nullptr;
+    }
 
     // commit previous write to the old queue before switching
     _producer->bounded_queue.commit_write();

include/quill/sinks/JsonSink.h

@@ -141,7 +141,8 @@ class JsonFileSink : public detail::JsonSink<FileSink>
 public:
   JsonFileSink(fs::path const& filename, FileSinkConfig const& config,
                FileEventNotifier file_event_notifier = FileEventNotifier{}, bool do_fopen = true)
-    : detail::JsonSink<FileSink>(filename, static_cast<FileSinkConfig const&>(config), std::move(file_event_notifier), do_fopen)
+    : detail::JsonSink<FileSink>(filename, static_cast<FileSinkConfig const&>(config),
+                                 std::move(file_event_notifier), do_fopen)
   {
   }
 

include/quill/sinks/SyslogSink.h

@@ -154,12 +154,10 @@ class SyslogSink : public quill::Sink
   /**
    * @brief Writes a formatted log message to the stream
    */
-  void write_log(MacroMetadata const* /* log_metadata */,
-                 uint64_t /* log_timestamp */, std::string_view /* thread_id */,
-                 std::string_view /* thread_name */, std::string const& /* process_id */,
-                 std::string_view /* logger_name */, LogLevel log_level,
-                 std::string_view /* log_level_description */,
-                 std::string_view /* log_level_short_code */,
+  void write_log(MacroMetadata const* /* log_metadata */, uint64_t /* log_timestamp */,
+                 std::string_view /* thread_id */, std::string_view /* thread_name */,
+                 std::string const& /* process_id */, std::string_view /* logger_name */, LogLevel log_level,
+                 std::string_view /* log_level_description */, std::string_view /* log_level_short_code */,
                  std::vector<std::pair<std::string, std::string>> const* /* named_args */,
                  std::string_view log_message, std::string_view log_statement) override
   {

test/integration_tests/BoundedBlockingQueueTest.cpp

@@ -15,7 +15,7 @@ using namespace quill;
 struct CustomFrontendOptions
 {
   static constexpr quill::QueueType queue_type = quill::QueueType::BoundedBlocking;
-  static constexpr uint32_t initial_queue_capacity = 131'072;
+  static constexpr size_t initial_queue_capacity = 131'072;
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
   static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
 };

test/integration_tests/BoundedDroppingQueueDropMessagesTest.cpp

@@ -15,7 +15,7 @@ using namespace quill;
 struct CustomFrontendOptions
 {
   static constexpr quill::QueueType queue_type = quill::QueueType::BoundedDropping;
-  static constexpr uint32_t initial_queue_capacity = 1024;
+  static constexpr size_t initial_queue_capacity = 1024;
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
   static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
 };

test/integration_tests/BoundedDroppingQueueTest.cpp

@@ -15,7 +15,7 @@ using namespace quill;
 struct CustomFrontendOptions
 {
   static constexpr quill::QueueType queue_type = quill::QueueType::BoundedDropping;
-  static constexpr uint32_t initial_queue_capacity = 131'072;
+  static constexpr size_t initial_queue_capacity = 131'072;
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
   static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
 };

test/integration_tests/CsvWritingCustomFrontendTest.cpp

@@ -17,7 +17,7 @@ struct OrderCsvSchema
 struct CustomFrontendOptions
 {
   static constexpr quill::QueueType queue_type = quill::QueueType::BoundedBlocking;
-  static constexpr uint32_t initial_queue_capacity = 131'072;
+  static constexpr size_t initial_queue_capacity = 131'072;
   static constexpr uint32_t blocking_queue_retry_interval_ns = 800;
   static constexpr quill::HugePagesPolicy huge_pages_policy = quill::HugePagesPolicy::Never;
 };