flush refactor/fix

include/journal.hpp

@@ -68,6 +68,9 @@ using detail::LogVerbosity;
 auto init_std_out(const LogVerbosity &min = LogVerbosity::Info) -> void;
 auto init_std_out_err(const LogVerbosity &min = LogVerbosity::Info) -> void;
 auto init() -> void;
+auto init_no_threads() -> void;
+auto add_std_out() -> void;
+auto add_std_err() -> void;
 auto add_file(std::string_view file_path,
               std::ios_base::openmode = std::ios_base::ate,
               const LogVerbosity &min = LogVerbosity::Trace,

src/journal.cc

@@ -186,9 +186,6 @@ struct BaseSink {
   Verbosity min_verbosity;
   Verbosity max_verbosity;
 
-  std::optional<std::chrono::system_clock::time_point> last_flushed;
-  static constexpr auto TIME_BETWEEN_FLUSH = 2s;
-
   BaseSink()
       : min_verbosity(LogVerbosity::Trace), max_verbosity(LogVerbosity::Error) {
   }
@@ -208,16 +205,6 @@ struct BaseSink {
     return std::exchange(min_verbosity, new_min);
   }
 
-  auto should_flush() -> bool {
-    const auto now = std::chrono::system_clock::now();
-    if (!last_flushed || (now - last_flushed.value()) >= TIME_BETWEEN_FLUSH) {
-      last_flushed = now;
-      return true;
-    } else {
-      return false;
-    }
-  }
-
   virtual auto flush() -> void {}
   virtual auto operator<<(std::string_view str) -> void {}
 };
@@ -264,34 +251,60 @@ static auto try_flush_all_loop(std::stop_token) -> void;
 
 class Journal {
   std::vector<std::unique_ptr<sinks::BaseSink>> sinks;
-  std::jthread flush_thread;
+  std::optional<std::jthread> flush_thread;
+  std::chrono::system_clock::duration flush_interval;
 
 public:
-  Journal() { flush_thread = std::jthread(&detail::try_flush_all_loop); }
+  Journal() : flush_interval(250ms) {}
   ~Journal() {
-    log(LogVerbosity::Trace, source_location::current(),
-        "Requesting stop for flush thread");
-    flush_thread.request_stop();
-    log(LogVerbosity::Trace, source_location::current(),
-        "Waiting for flush thread to join..");
-    flush_thread.join();
-    log(LogVerbosity::Trace, source_location::current(), "Flush thread joined");
-    try_flush_all();
+    if (flush_thread) {
+      log(LogVerbosity::Trace, source_location::current(),
+          "Requesting stop for flush thread");
+      flush_thread->request_stop();
+      log(LogVerbosity::Trace, source_location::current(),
+          "Waiting for flush thread to join..");
+      flush_thread->join();
+      log(LogVerbosity::Trace, source_location::current(),
+          "Flush thread joined");
+    }
+    flush_all();
   }
 
-  auto try_flush_all() -> void {
+  auto
+  set_flush_interval(const std::chrono::system_clock::duration &new_interval)
+      -> void {
+    flush_interval = new_interval;
+  }
+  auto get_flush_interval() const -> std::chrono::system_clock::duration {
+    return flush_interval;
+  }
+
+  auto init_flush_thread() -> void {
+    flush_thread = std::jthread(&detail::try_flush_all_loop);
+  }
+
+  auto should_flush() const -> bool {
+    return !flush_thread.has_value() || flush_interval < 1ms;
+  }
+
+  auto flush_all() -> void { force_flush_all(); }
+
+  auto force_flush_all() -> void {
     for (auto &&sink : sinks) {
-      if (sink->should_flush()) {
-        sink->flush();
-      }
+      sink->flush();
     }
   }
 
   auto log(const Verbosity &verbosity, source_location source,
            std::string_view msg) -> void {
+    const auto do_flush = should_flush();
     for (auto &&sink : sinks) {
       if (sink->should_sink_verbosity(verbosity)) {
         (*sink) << format(msg, verbosity, source);
+
+        if (do_flush) {
+          sink->flush();
+        }
       }
     }
   }
@@ -332,12 +345,14 @@ auto add_std_err() { add_sink(std::make_unique<sinks::StdErrSink>()); }
 static mutex::Mutex<std::unique_ptr<Journal>> global_journal;
 
 static auto try_flush_all_loop(std::stop_token stoken) -> void {
+  std::chrono::system_clock::duration interval = 250ms;
   while (!stoken.stop_requested()) {
     {
       auto &&[lock, ref] = global_journal.mut();
-      ref->try_flush_all();
+      interval = ref->get_flush_interval();
+      ref->flush_all();
     }
-    std::this_thread::sleep_for(5s);
+    std::this_thread::sleep_for(interval);
   }
 }
 
@@ -350,6 +365,14 @@ auto log_internal(const LogVerbosity &verbosity, detail::source_location source,
 } // namespace detail
 
 auto init() -> void {
+  auto &&[lock, ref] = detail::global_journal.mut();
+  if (!ref.get()) {
+    ref = std::make_unique<detail::Journal>();
+    ref->init_flush_thread();
+  }
+}
+
+auto init_no_threads() -> void {
   auto &&[lock, ref] = detail::global_journal.mut();
   if (!ref.get()) {
     ref = std::make_unique<detail::Journal>();
@@ -369,6 +392,15 @@ auto init_std_out_err(const LogVerbosity &min) -> void {
   ref->add_std_err();
 }
 
+auto add_std_out() -> void {
+  auto &&[lock, ref] = detail::global_journal.mut();
+  ref->add_std_out();
+}
+auto add_std_err() -> void {
+  auto &&[lock, ref] = detail::global_journal.mut();
+  ref->add_std_err();
+}
+
 auto add_sink(std::shared_ptr<std::ostream> stream, const LogVerbosity &min,
               const LogVerbosity &max) -> void {
   auto &&[lock, ref] = detail::global_journal.mut();

src/main.cc

@@ -6,7 +6,9 @@
 int main(int argc, const char* argv[]) {
   auto stream = std::make_shared<std::stringstream>();
 
-  journal::init_std_out_err();
+  journal::init_no_threads();
+  journal::add_std_err();
+  journal::add_std_out();
   journal::add_sink(stream, journal::LogVerbosity::Warn);
   journal::add_file("test.log");
   journal::info("some info!");