kudu git commit: KUDU-1192 Periodically flush glog buffers from a thread

[to...@apache.org]

Repository: kudu
Updated Branches:
  refs/heads/master fd1f1699f -> 30d759274


KUDU-1192 Periodically flush glog buffers from a thread

Added a wait timeout to AsyncLogger RunThread based on FLAGS_logbufsec
When the wait times out, it will flush even if there is nothing enqueued.
Also sets the logbufsecs default to 5 seconds, instead of the 30s.

Change-Id: Id4c6d440e9259efcf222530f13137f7de5bf00fc
Reviewed-on: http://gerrit.cloudera.org:8080/6853
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Reviewed-by: Todd Lipcon <to...@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/30d75927
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/30d75927
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/30d75927

Branch: refs/heads/master
Commit: 30d7592747a40feb8319d57f002583e80654855e
Parents: fd1f169
Author: William Li <wi...@inspur.com>
Authored: Wed May 10 16:48:53 2017 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Wed May 17 21:12:19 2017 +0000

----------------------------------------------------------------------
 src/kudu/util/async_logger.cc |  6 +++++-
 src/kudu/util/flags.cc        |  4 ++++
 src/kudu/util/logging-test.cc | 28 ++++++++++++++++++++++++++--
 3 files changed, 35 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kudu/blob/30d75927/src/kudu/util/async_logger.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/async_logger.cc b/src/kudu/util/async_logger.cc
index b6456de..8e6a4a9 100644
--- a/src/kudu/util/async_logger.cc
+++ b/src/kudu/util/async_logger.cc
@@ -22,6 +22,7 @@
 #include <thread>
 
 #include "kudu/util/locks.h"
+#include "kudu/util/monotime.h"
 
 using std::string;
 
@@ -115,7 +116,10 @@ void AsyncLogger::RunThread() {
   MutexLock l(lock_);
   while (state_ == RUNNING || active_buf_->needs_flush_or_write()) {
     while (!active_buf_->needs_flush_or_write() && state_ == RUNNING) {
-      wake_flusher_cond_.Wait();
+      if (!wake_flusher_cond_.TimedWait(MonoDelta::FromSeconds(FLAGS_logbufsecs))) {
+        // In case of wait timeout, force it to flush regardless whether there is anything enqueued.
+        active_buf_->flush = true;
+      }
     }
 
     active_buf_.swap(flushing_buf_);

http://git-wip-us.apache.org/repos/asf/kudu/blob/30d75927/src/kudu/util/flags.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/flags.cc b/src/kudu/util/flags.cc
index 00885a0..6b57f54 100644
--- a/src/kudu/util/flags.cc
+++ b/src/kudu/util/flags.cc
@@ -454,6 +454,10 @@ void SetUmask() {
 } // anonymous namespace
 
 int ParseCommandLineFlags(int* argc, char*** argv, bool remove_flags) {
+  // The logbufsecs default is 30 seconds which is a bit too long.
+  google::SetCommandLineOptionWithMode("logbufsecs", "5",
+                                       google::FlagSettingMode::SET_FLAGS_DEFAULT);
+
   int ret = google::ParseCommandLineNonHelpFlags(argc, argv, remove_flags);
   HandleCommonFlags();
   return ret;

http://git-wip-us.apache.org/repos/asf/kudu/blob/30d75927/src/kudu/util/logging-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/logging-test.cc b/src/kudu/util/logging-test.cc
index b3eb458..e3a0771 100644
--- a/src/kudu/util/logging-test.cc
+++ b/src/kudu/util/logging-test.cc
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <atomic>
 #include <glog/logging.h>
 #include <gmock/gmock.h>
 #include <string>
@@ -29,6 +30,7 @@
 #include "kudu/util/logging_test_util.h"
 #include "kudu/util/monotime.h"
 #include "kudu/util/stopwatch.h"
+#include "kudu/util/test_util.h"
 
 using std::string;
 using std::vector;
@@ -115,8 +117,8 @@ class CountingLogger : public google::base::Logger {
     return 0;
   }
 
-  int flush_count_ = 0;
-  int message_count_ = 0;
+  std::atomic<int> flush_count_ = {0};
+  std::atomic<int> message_count_ = {0};
 };
 
 TEST(LoggingTest, TestAsyncLogger) {
@@ -160,6 +162,28 @@ TEST(LoggingTest, TestAsyncLogger) {
   ASSERT_GT(async.app_threads_blocked_count_for_tests(), 0);
 }
 
+TEST(LoggingTest, TestAsyncLoggerAutoFlush) {
+  const int kBuffer = 10000;
+  CountingLogger base;
+  AsyncLogger async(&base, kBuffer);
+
+  FLAGS_logbufsecs = 1;
+  async.Start();
+
+  // Write some log messages with non-force_flush types.
+  async.Write(false, 0, "test-x", 1);
+  async.Write(false, 1, "test-y", 1);
+
+  // The flush wait timeout might take a little bit of time to run.
+  ASSERT_EVENTUALLY([&]() {
+    ASSERT_EQ(base.message_count_, 2);
+    // The AsyncLogger should have flushed at least once by the timer automatically
+    // so there should be no more messages in the buffer.
+    ASSERT_GT(base.flush_count_, 0);
+  });
+  async.Stop();
+}
+
 // Basic test that the redaction utilities work as expected.
 TEST(LoggingTest, TestRedactionBasic) {
   ASSERT_STREQ("<redacted>", KUDU_REDACT("hello"));