Add Concurrency option and MPSC support using farbot.

context: #6 #14

Author: atsushieno

CMakeLists.txt

@@ -36,6 +36,26 @@ if(NOT TARGET readerwriterqueue)
     FetchContent_MakeAvailable(ReaderWriterQueue)
 endif()
 
+if (NOT TARGET farbot)
+    include(FetchContent)
+
+    FetchContent_Declare(farbot
+        GIT_REPOSITORY https://github.com/hogliux/farbot
+        GIT_TAG 0416705394720c12f0d02e55c144e4f69bb06912
+    )
+    # Note we do not "MakeAvailable" here, because farbot does not fully work via FetchContent
+    if(NOT farbot_POPULATED)
+        FetchContent_Populate(farbot)
+    endif()
+    add_library(farbot INTERFACE)
+    add_library(farbot::farbot ALIAS farbot)
+
+    target_include_directories(farbot INTERFACE
+            $<BUILD_INTERFACE:${farbot_SOURCE_DIR}/include>
+            $<INSTALL_INTERFACE:include>
+    )
+endif()
+
 if(NOT TARGET stb::stb)
     # Avoid warning about DOWNLOAD_EXTRACT_TIMESTAMP in CMake 3.24:
     if (CMAKE_VERSION VERSION_GREATER_EQUAL "3.24.0")
@@ -73,6 +93,7 @@ target_link_libraries(rtlog
     INTERFACE 
         readerwriterqueue
         stb::stb
+        farbot::farbot
         $<$<BOOL:${RTLOG_USE_FMTLIB}>:fmt::fmt>
 )
 

README.md

@@ -11,13 +11,14 @@ The design behind this logger was presented at ADCx 2023. Presentation [video](h
 - Ability to log messages of any type and size from the real-time thread
 - Statically allocated memory at compile time, no allocations in the real-time thread
 - Support for printf-style format specifiers (using [a version of the printf family](https://github.com/nothings/stb/blob/master/stb_sprintf.h) that doesn't hit the `localeconv` lock)
-- Efficient thread-safe logging using a [lock free queue](https://github.com/cameron314/readerwriterqueue).
+- Efficient thread-safe logging using lock free queues (either [single consumer](https://github.com/cameron314/readerwriterqueue) or [multi consumer](https://github.com/hogliux/farbot)).
 
 ## Requirements
 
 - A C++17 compatible compiler
 - The C++17 standard library
 - moodycamel::ReaderWriterQueue (will be downloaded via cmake if not provided)
+- farbot's fifo (will be downloaded via cmake if not provided)
 - stb's vsnprintf (will be downloaded via cmake if not provided)
 
 ## Installation via CMake

examples/everlog/everlogmain.cpp

@@ -4,7 +4,7 @@
 
 namespace evr {
 constexpr auto MAX_LOG_MESSAGE_LENGTH = 256;
-constexpr auto MAX_NUM_LOG_MESSAGES = 100;
+constexpr auto MAX_NUM_LOG_MESSAGES = 128;
 
 enum class LogLevel { Debug, Info, Warning, Critical };
 

include/rtlog/rtlog.h

@@ -10,6 +10,7 @@
 #include <fmt/format.h>
 #endif // RTLOG_USE_FMTLIB
 
+#include <farbot/fifo.hpp>
 #include <readerwriterqueue.h>
 
 #ifndef STB_SPRINTF_IMPLEMENTATION
@@ -31,14 +32,20 @@ enum class Status {
   Error_MessageTruncated = 2,
 };
 
+enum class QueueConcurrency {
+  Single_Producer_Single_Consumer = 0,
+  Multi_Producer_Single_Consumer = 1,
+};
+
 /**
  * @brief A logger class for logging messages.
  * This class allows you to log messages of type LogData.
  * This type is user defined, and is often the additional data outside the
  * format string you want to log. For instance: The log level, the log region,
  * the file name, the line number, etc. See examples or tests for some ideas.
  *
- * TODO: Currently is built on a single input/single output queue. Do not call
+ * NOTE: by default, it is built on a single input/single output queue. You have
+ * to specify QueueConcurrency for other types of queues. Otherwise, do not call
  * Log or PrintAndClearLogQueue from multiple threads.
  *
  * @tparam LogData The type of the data to be logged.
@@ -49,9 +56,16 @@ enum class Status {
  * @tparam SequenceNumber This number is incremented when the message is
  * enqueued. It is assumed that your non-realtime logger increments and logs it
  * on Log.
+ * @tparam QueueConcurrency The concurrency type of the internal queue.
+ * The default Single_Producer_Single_Consumer is for the simplest queue that
+ * works in single-producer thread model.
+ * Multi_Producer_Single_Consumer is for such an application that needs to
+ * handle multiple logging clients.
  */
 template <typename LogData, size_t MaxNumMessages, size_t MaxMessageLength,
-          std::atomic<std::size_t> &SequenceNumber>
+          std::atomic<std::size_t> &SequenceNumber,
+          QueueConcurrency Concurrency =
+              QueueConcurrency::Single_Producer_Single_Consumer>
 class Logger {
 public:
   /*
@@ -97,7 +111,7 @@ class Logger {
 
     // Even if the message was truncated, we still try to enqueue it to minimize
     // data loss
-    const bool dataWasEnqueued = mQueue.try_enqueue(dataToQueue);
+    const bool dataWasEnqueued = mQueue->tryEnqueue(std::move(dataToQueue));
 
     if (!dataWasEnqueued)
       retVal = Status::Error_QueueFull;
@@ -196,7 +210,7 @@ class Logger {
 
     // Even if the message was truncated, we still try to enqueue it to minimize
     // data loss
-    const bool dataWasEnqueued = mQueue.try_enqueue(dataToQueue);
+    const bool dataWasEnqueued = mQueue->tryEnqueue(std::move(dataToQueue));
 
     if (!dataWasEnqueued)
       retVal = Status::Error_QueueFull;
@@ -227,7 +241,7 @@ class Logger {
     int numProcessed = 0;
 
     InternalLogData value;
-    while (mQueue.try_dequeue(value)) {
+    while (mQueue->tryDequeue(value)) {
       printLogFn(value.mLogData, value.mSequenceNumber, "%s",
                  value.mMessage.data());
       numProcessed++;
@@ -243,7 +257,39 @@ class Logger {
     std::array<char, MaxMessageLength> mMessage{};
   };
 
-  moodycamel::ReaderWriterQueue<InternalLogData> mQueue{MaxNumMessages};
+  class InternalQueue {
+  public:
+    virtual bool tryEnqueue(InternalLogData&& value) = 0;
+    virtual bool tryDequeue(InternalLogData& value) = 0;
+  };
+  class InternalQueueSPSC : public InternalQueue {
+    moodycamel::ReaderWriterQueue<InternalLogData> mQueue{MaxNumMessages};
+  public:
+    bool tryEnqueue(InternalLogData&& value) override {
+      return mQueue.try_enqueue(std::move(value));
+    }
+    bool tryDequeue(InternalLogData& value) override {
+      return mQueue.try_dequeue(value);
+    }
+  };
+  class InternalQueueMPSC : public InternalQueue {
+    farbot::fifo<InternalLogData, farbot::fifo_options::concurrency::single,
+                 farbot::fifo_options::concurrency::multiple>
+        mQueue{ MaxNumMessages };
+  public:
+    bool tryEnqueue(InternalLogData&& value) override {
+      return mQueue.push(std::move(value));
+    }
+    bool tryDequeue(InternalLogData& value) override {
+      return mQueue.pop(value);
+    }
+  };
+
+  std::unique_ptr<InternalQueue> mQueue{
+      Concurrency == QueueConcurrency::Single_Producer_Single_Consumer
+          ? (std::unique_ptr<InternalQueue>)
+                std::make_unique<InternalQueueSPSC>()
+          : std::make_unique<InternalQueueMPSC>()};
 };
 
 /**

test/test_rtlog.cpp

@@ -6,7 +6,7 @@ namespace rtlog::test {
 std::atomic<std::size_t> gSequenceNumber{0};
 
 constexpr auto MAX_LOG_MESSAGE_LENGTH = 256;
-constexpr auto MAX_NUM_LOG_MESSAGES = 100;
+constexpr auto MAX_NUM_LOG_MESSAGES = 128;
 
 enum class ExampleLogLevel { Debug, Info, Warning, Critical };
 
@@ -81,6 +81,22 @@ TEST_CASE("Test rtlog basic construction") {
   CHECK(logger.PrintAndClearLogQueue(PrintMessage) == 4);
 }
 
+TEST_CASE("Test rtlog MPSC basic construction") {
+  rtlog::Logger<ExampleLogData, MAX_NUM_LOG_MESSAGES, MAX_LOG_MESSAGE_LENGTH,
+                gSequenceNumber,
+                rtlog::QueueConcurrency::Multi_Producer_Single_Consumer>
+      logger;
+  logger.Log({ExampleLogLevel::Debug, ExampleLogRegion::Engine},
+             "Hello, world!");
+  logger.Log({ExampleLogLevel::Info, ExampleLogRegion::Game}, "Hello, world!");
+  logger.Log({ExampleLogLevel::Warning, ExampleLogRegion::Network},
+             "Hello, world!");
+  logger.Log({ExampleLogLevel::Critical, ExampleLogRegion::Audio},
+             "Hello, world!");
+
+  CHECK(logger.PrintAndClearLogQueue(PrintMessage) == 4);
+}
+
 TEST_CASE("va_args works as intended") {
   rtlog::Logger<ExampleLogData, MAX_NUM_LOG_MESSAGES, MAX_LOG_MESSAGE_LENGTH,
                 gSequenceNumber>
@@ -199,7 +215,7 @@ TEST_CASE("Errors are returned from Log") {
   }
 
   SUBCASE("Enqueue more than capacity and get an error") {
-    const auto maxNumMessages = 10;
+    const auto maxNumMessages = 16;
     rtlog::Logger<ExampleLogData, maxNumMessages, MAX_LOG_MESSAGE_LENGTH,
                   gSequenceNumber>
         logger;