async write for easylog

include/easylog/easylog.h

@@ -20,6 +20,7 @@
 #include <vector>
 
 #include "appender.hpp"
+#include "thread_pool.h"
 
 namespace easylog {
 
@@ -31,8 +32,17 @@ class logger {
     return instance;
   }
 
+  void async_write(std::shared_ptr<record_t> record) {
+    thread_pool::instance().async([this, record] {
+      write(*record);
+      return true;
+    });
+  }
+
   void operator+=(const record_t &record) { write(record); }
 
+  void operator+=(std::shared_ptr<record_t> record) { async_write(record); }
+
   void write(const record_t &record) {
     append_format(record);
 
@@ -42,24 +52,22 @@ class logger {
     }
   }
 
-  void flush() {
-    if (appender_) {
-      appender_->flush();
-    }
-  }
+  void flush() { appender_->flush(); }
 
   void init(Severity min_severity, bool enable_console,
             const std::string &filename, size_t max_file_size, size_t max_files,
-            bool flush_every_time) {
-    static appender appender(filename, max_file_size, max_files,
-                             flush_every_time);
-    appender_ = &appender;
+            bool flush_every_time, bool enable_async_write) {
+    appender_ = std::make_unique<appender>(filename, max_file_size, max_files,
+                                           flush_every_time);
     min_severity_ = min_severity;
     enable_console_ = enable_console;
+    enbale_async_write_ = enable_async_write;
   }
 
   bool check_severity(Severity severity) { return severity >= min_severity_; }
 
+  bool check_async_write() { return enbale_async_write_; }
+
   void add_appender(std::function<void(std::string_view)> fn) {
     appenders_.emplace_back(std::move(fn));
   }
@@ -68,6 +76,8 @@ class logger {
   logger() = default;
   logger(const logger &) = default;
 
+  ~logger() { thread_pool::instance().quit(); }
+
   template <size_t N>
   size_t get_time_str(char (&buf)[N], const auto &now) {
     const auto nowAsTimeT = std::chrono::system_clock::to_time_t(now);
@@ -120,16 +130,23 @@ class logger {
 
   Severity min_severity_;
   bool enable_console_ = true;
-  appender *appender_ = nullptr;
+  bool enbale_async_write_ = false;
+  std::unique_ptr<appender> appender_ = nullptr;
   std::vector<std::function<void(std::string_view)>> appenders_;
 };
 
 template <size_t Id = 0>
 inline void init_log(Severity min_severity, const std::string &filename = "",
                      bool enable_console = true, size_t max_file_size = 0,
-                     size_t max_files = 0, bool flush_every_time = false) {
+                     size_t max_files = 0, bool flush_every_time = false,
+                     bool enable_async_write = false) {
+  if (enable_async_write) {
+    thread_pool::instance().init(1);
+  }
+
   logger<Id>::instance().init(min_severity, enable_console, filename,
-                              max_file_size, max_files, flush_every_time);
+                              max_file_size, max_files, flush_every_time,
+                              enable_async_write);
 }
 
 template <size_t Id = 0>
@@ -155,19 +172,28 @@ inline void add_appender(std::function<void(std::string_view)> fn) {
 
 #define ELOG(severity, ...) ELOG_IMPL(Severity::severity, __VA_ARGS__, 0)
 
-#define ELOGV_IMPL(severity, Id, fmt, ...)                            \
-  if (!easylog::logger<Id>::instance().check_severity(severity)) {    \
-    ;                                                                 \
-  }                                                                   \
-  else {                                                              \
-    easylog::logger<Id>::instance() +=                                \
-        easylog::record_t(std::chrono::system_clock::now(), severity, \
-                          GET_STRING(__FILE__, __LINE__))             \
-            .sprintf(fmt, __VA_ARGS__);                               \
-    if (severity == Severity::CRITICAL) {                             \
-      easylog::flush<Id>();                                           \
-      std::exit(EXIT_FAILURE);                                        \
-    }                                                                 \
+#define ELOGV_IMPL(severity, Id, fmt, ...)                              \
+  if (!easylog::logger<Id>::instance().check_severity(severity)) {      \
+    ;                                                                   \
+  }                                                                     \
+  else {                                                                \
+    if (!easylog::logger<Id>::instance().check_async_write()) {         \
+      easylog::logger<Id>::instance() +=                                \
+          easylog::record_t(std::chrono::system_clock::now(), severity, \
+                            GET_STRING(__FILE__, __LINE__))             \
+              .sprintf(fmt, __VA_ARGS__);                               \
+    }                                                                   \
+    else {                                                              \
+      auto record = std::make_shared<easylog::record_t>(                \
+          std::chrono::system_clock::now(), severity,                   \
+          GET_STRING(__FILE__, __LINE__));                              \
+      record->sprintf(fmt, __VA_ARGS__);                                \
+      easylog::logger<Id>::instance() += record;                        \
+    }                                                                   \
+    if (severity == Severity::CRITICAL) {                               \
+      easylog::flush<Id>();                                             \
+      std::exit(EXIT_FAILURE);                                          \
+    }                                                                   \
   }
 
 #define ELOGV(severity, ...) \
@@ -195,4 +221,4 @@ inline void add_appender(std::function<void(std::string_view)> fn) {
 #define ELOGI ELOG_INFO
 #define ELOGW ELOG_WARN
 #define ELOGE ELOG_ERROR
-#define ELOGC ELOG_CRITICAL
+#define ELOGC ELOG_CRITICAL

include/easylog/thread_pool.h

@@ -0,0 +1,150 @@
+/*
+ * Copyright (c) 2023, Alibaba Group Holding Limited;
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <thread>
+
+namespace easylog {
+
+template <typename T>
+class blocking_queue {
+ public:
+  blocking_queue() = default;
+
+  void enqueue(T &&item) {
+    {
+      std::unique_lock<std::mutex> lock(queue_mutex_);
+      queue_.push_back(std::move(item));
+    }
+    push_cv_.notify_one();
+  }
+
+  bool dequeue_for(T &popped_item, std::chrono::milliseconds wait_duration) {
+    {
+      std::unique_lock<std::mutex> lock(queue_mutex_);
+      if (!push_cv_.wait_for(lock, wait_duration, [this] {
+            return !queue_.empty();
+          })) {
+        return false;
+      }
+
+      popped_item = std::move(queue_.front());
+      queue_.pop_front();
+    }
+    return true;
+  }
+
+  size_t size() {
+    std::unique_lock<std::mutex> lock(queue_mutex_);
+    return queue_.size();
+  }
+
+  void clear() {
+    std::unique_lock<std::mutex> lock(queue_mutex_);
+    queue_.clear();
+  }
+
+ private:
+  std::mutex queue_mutex_;
+  std::condition_variable push_cv_;
+  std::list<T> queue_;
+};
+
+class thread_pool {
+ public:
+  static thread_pool &instance() {
+    static thread_pool thread_instance;
+    return thread_instance;
+  }
+
+ public:
+  typedef std::function<bool()> Task;
+  using item_type = std::unique_ptr<Task>;
+
+  void init(size_t threads_n) {
+    std::lock_guard<std::mutex> lock(mutex_);
+
+    if (!threads_.empty()) {
+      return;
+    }
+
+    for (size_t i = 0; i < threads_n; i++) {
+      threads_.emplace_back([this] {
+        thread_pool::worker_loop();
+      });
+    }
+  }
+
+  void quit() {
+    std::lock_guard<std::mutex> lock(mutex_);
+
+    if (threads_.empty()) {
+      return;
+    }
+
+    queue_.clear();
+
+    for (size_t i = 0; i < threads_.size(); i++) {
+      async([]() {
+        return false;
+      });
+    }
+
+    for (auto &t : threads_) {
+      t.join();
+    }
+
+    threads_.clear();
+  }
+
+  void async(Task &&task) {
+    queue_.enqueue(std::make_unique<Task>(std::move(task)));
+  }
+
+ private:
+  thread_pool() = default;
+
+  blocking_queue<item_type> queue_;
+
+  std::vector<std::thread> threads_;
+
+  std::mutex mutex_;
+
+ private:
+  void worker_loop() {
+    while (process_next_msg()) {
+    }
+  }
+
+  bool process_next_msg() {
+    std::unique_ptr<Task> record_task;
+    bool dequeued = queue_.dequeue_for(record_task, std::chrono::seconds(10));
+    if (!dequeued) {
+      return true;
+    }
+
+    return (*record_task)();
+  }
+};
+
+}  // namespace easylog

src/easylog/benchmark/main.cpp

@@ -66,15 +66,28 @@ void test_glog() {
 
 void test_easylog() {
   std::filesystem::remove("easylog.txt");
-  easylog::init_log(Severity::DEBUG, "easylog.txt", false, 1024 * 1024, 1);
+  easylog::init_log(Severity::DEBUG, "easylog.txt", false, 1024 * 1024, 1,
+                    false);
   {
     ScopedTimer timer("easylog");
     for (int i = 0; i < 5000; i++)
-      ELOG(INFO) << "Hello, it is a long string test! " << 42 << 21 << 2.5;
+      MELOGV(INFO, 0, "Hello, it is a long string test! Hello World");
+  }
+}
+
+void test_async_easylog() {
+  std::filesystem::remove("async_easylog.txt");
+  easylog::init_log<1>(Severity::DEBUG, "async_easylog.txt", false, 1024 * 1024,
+                       1, false, true);
+  {
+    ScopedTimer timer("async_easylog");
+    for (int i = 0; i < 5000; i++)
+      MELOGV(INFO, 1, "Hello, it is a long string test!  Hello World");
   }
 }
 
 int main() {
   test_glog();
   test_easylog();
+  test_async_easylog();
 }

src/easylog/tests/test_easylog.cpp

@@ -154,3 +154,14 @@ TEST_CASE("file_roll") {
             "he string that should be saved in the file.") !=
         std::string::npos);
 }
+
+TEST_CASE("async_write") {
+  std::string async_write_file = "async_write.txt";
+  std::filesystem::remove(async_write_file);
+  constexpr size_t SeventhId = 7;
+  easylog::init_log<SeventhId>(Severity::DEBUG, async_write_file, true,
+                               1024 * 1024, 1, false, true);
+  for (int i = 0; i < 50; i++)
+    ELOG(INFO, SeventhId) << "Hello, it is a long string test! " << 42 << 21
+                          << 2.5;
+}