[3/3] kudu git commit: Synchronizer::WaitFor thread-safety

[da...@apache.org]

Synchronizer::WaitFor thread-safety

WaitFor's implementation previously was not thread safe in the case that
the waiter deallocated the Synchronizer after receiving a timeout.

Change-Id: I9565a50839ffd23b5bac6986a6fdee41ac21aa3a
Reviewed-on: http://gerrit.cloudera.org:8080/10783
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Kudu Jenkins


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

Branch: refs/heads/master
Commit: f6e8fe6c6bb6dd83e6485bc1b56364b11b250e94
Parents: f1e652b
Author: Dan Burkert <da...@apache.org>
Authored: Thu Jun 21 11:49:39 2018 -0700
Committer: Dan Burkert <da...@apache.org>
Committed: Fri Jun 22 22:15:55 2018 +0000

----------------------------------------------------------------------
 src/kudu/util/CMakeLists.txt     |   1 +
 src/kudu/util/async_util-test.cc | 129 ++++++++++++++++++++++++++++++++++
 src/kudu/util/async_util.h       |  71 +++++++++++--------
 3 files changed, 170 insertions(+), 31 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kudu/blob/f6e8fe6c/src/kudu/util/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/util/CMakeLists.txt b/src/kudu/util/CMakeLists.txt
index 044660c..743e1af 100644
--- a/src/kudu/util/CMakeLists.txt
+++ b/src/kudu/util/CMakeLists.txt
@@ -331,6 +331,7 @@ target_link_libraries(protoc-gen-insertions gutil protobuf protoc ${KUDU_BASE_LI
 #######################################
 
 set(KUDU_TEST_LINK_LIBS kudu_util gutil ${KUDU_MIN_TEST_LIBS})
+ADD_KUDU_TEST(async_util-test)
 ADD_KUDU_TEST(atomic-test)
 ADD_KUDU_TEST(bit-util-test)
 ADD_KUDU_TEST(bitmap-test)

http://git-wip-us.apache.org/repos/asf/kudu/blob/f6e8fe6c/src/kudu/util/async_util-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/async_util-test.cc b/src/kudu/util/async_util-test.cc
new file mode 100644
index 0000000..5cb7a63
--- /dev/null
+++ b/src/kudu/util/async_util-test.cc
@@ -0,0 +1,129 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "kudu/util/async_util.h"
+
+#include <unistd.h>
+
+#include <functional>
+#include <thread>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "kudu/gutil/basictypes.h"
+#include "kudu/gutil/callback.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/status.h"
+#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
+
+using std::thread;
+using std::vector;
+
+namespace kudu {
+
+class AsyncUtilTest : public KuduTest {
+ public:
+  AsyncUtilTest() {
+    // Set up an alarm to fail the test in case of deadlock.
+    alarm(30);
+  }
+  ~AsyncUtilTest() {
+    // Disable the alarm on test exit.
+    alarm(0);
+  }
+};
+
+// Test completing the synchronizer through each of the APIs it exposes.
+TEST_F(AsyncUtilTest, TestSynchronizerCompletion) {
+  Synchronizer sync;
+
+  {
+    auto waiter = thread([sync] {
+        ignore_result(sync.Wait());
+    });
+    SleepFor(MonoDelta::FromMilliseconds(5));
+    sync.StatusCB(Status::OK());
+    waiter.join();
+  }
+  sync.Reset();
+  {
+    auto cb = sync.AsStatusCallback();
+    auto waiter = thread([sync] {
+        ignore_result(sync.Wait());
+    });
+    SleepFor(MonoDelta::FromMilliseconds(5));
+    cb.Run(Status::OK());
+    waiter.join();
+  }
+  sync.Reset();
+  {
+    auto cb = sync.AsStdStatusCallback();
+    auto waiter = thread([sync] {
+        ignore_result(sync.Wait());
+    });
+    SleepFor(MonoDelta::FromMilliseconds(5));
+    cb(Status::OK());
+    waiter.join();
+  }
+}
+
+TEST_F(AsyncUtilTest, TestSynchronizerMultiWait) {
+  Synchronizer sync;
+  vector<thread> waiters;
+  for (int i = 0; i < 5; i++) {
+    waiters.emplace_back([sync] {
+        ignore_result(sync.Wait());
+    });
+  }
+  SleepFor(MonoDelta::FromMilliseconds(5));
+  sync.StatusCB(Status::OK());
+
+  for (auto& waiter : waiters) {
+    waiter.join();
+  }
+}
+
+TEST_F(AsyncUtilTest, TestSynchronizerTimedWait) {
+  thread waiter;
+  {
+    Synchronizer sync;
+    auto cb = sync.AsStatusCallback();
+    waiter = thread([cb] {
+        SleepFor(MonoDelta::FromMilliseconds(5));
+        cb.Run(Status::OK());
+    });
+    ASSERT_OK(sync.WaitFor(MonoDelta::FromMilliseconds(1000)));
+  }
+  waiter.join();
+
+  {
+    Synchronizer sync;
+    auto cb = sync.AsStatusCallback();
+    waiter = thread([cb] {
+        SleepFor(MonoDelta::FromMilliseconds(1000));
+        cb.Run(Status::OK());
+    });
+    ASSERT_TRUE(sync.WaitFor(MonoDelta::FromMilliseconds(5)).IsTimedOut());
+  }
+
+  // Waiting on the thread gives TSAN to check that no thread safety issues
+  // occurred.
+  waiter.join();
+}
+} // namespace kudu

http://git-wip-us.apache.org/repos/asf/kudu/blob/f6e8fe6c/src/kudu/util/async_util.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/async_util.h b/src/kudu/util/async_util.h
index 727b7f7..338c6c2 100644
--- a/src/kudu/util/async_util.h
+++ b/src/kudu/util/async_util.h
@@ -16,10 +16,11 @@
 // under the License.
 //
 // Utility functions which are handy when doing async/callback-based programming.
-#ifndef KUDU_UTIL_ASYNC_UTIL_H
-#define KUDU_UTIL_ASYNC_UTIL_H
+
+#pragma once
 
 #include <functional>
+#include <memory>
 
 #include "kudu/gutil/bind.h"
 #include "kudu/gutil/macros.h"
@@ -32,59 +33,67 @@ namespace kudu {
 // Simple class which can be used to make async methods synchronous.
 // For example:
 //   Synchronizer s;
-//   SomeAsyncMethod(s.callback());
+//   SomeAsyncMethod(s.AsStatusCallback());
 //   CHECK_OK(s.Wait());
+//
+// The lifetime of the synchronizer is decoupled from the callback it produces.
+// If the callback outlives the synchronizer then executing it will be a no-op.
+// Callers must be careful not to allow the callback to be destructed without
+// completing it, otherwise the thread waiting on the synchronizer will block
+// indefinitely.
 class Synchronizer {
  public:
   Synchronizer()
-    : l_(1) {
+    : data_(std::make_shared<Data>()) {
   }
 
   void StatusCB(const Status& status) {
-    s_ = status;
-    l_.CountDown();
+    Data::Callback(std::weak_ptr<Data>(data_), status);
   }
 
   StatusCallback AsStatusCallback() {
-    // Synchronizers are often declared on the stack, so it doesn't make
-    // sense for a callback to take a reference to its synchronizer.
-    //
-    // Note: this means the returned callback _must_ go out of scope before
-    // its synchronizer.
-    return Bind(&Synchronizer::StatusCB, Unretained(this));
+    return Bind(Data::Callback, std::weak_ptr<Data>(data_));
   }
 
   StdStatusCallback AsStdStatusCallback() {
-    // Synchronizers are often declared on the stack, so it doesn't make
-    // sense for a callback to take a reference to its synchronizer.
-    //
-    // Note: this means the returned callback _must_ go out of scope before
-    // its synchronizer.
-    return std::bind(&Synchronizer::StatusCB, this, std::placeholders::_1);
+    return std::bind(Data::Callback, std::weak_ptr<Data>(data_), std::placeholders::_1);
   }
 
-  Status Wait() {
-    l_.Wait();
-    return s_;
+  Status Wait() const {
+    data_->latch.Wait();
+    return data_->status;
   }
 
-  Status WaitFor(const MonoDelta& delta) {
-    if (PREDICT_FALSE(!l_.WaitFor(delta))) {
-      return Status::TimedOut("Timed out while waiting for the callback to be called.");
+  Status WaitFor(const MonoDelta& delta) const {
+    if (PREDICT_FALSE(!data_->latch.WaitFor(delta))) {
+      return Status::TimedOut("timed out while waiting for the callback to be called");
     }
-    return s_;
+    return data_->status;
   }
 
   void Reset() {
-    l_.Reset(1);
+    data_->latch.Reset(1);
   }
 
  private:
-  Status s_;
-  CountDownLatch l_;
 
-  DISALLOW_COPY_AND_ASSIGN(Synchronizer);
-};
+  struct Data {
+    Data() : latch(1) {
+    }
 
+    static void Callback(std::weak_ptr<Data> weak, const Status& status) {
+      auto ptr = weak.lock();
+      if (ptr) {
+        ptr->status = status;
+        ptr->latch.CountDown();
+      }
+    }
+
+    Status status;
+    CountDownLatch latch;
+    DISALLOW_COPY_AND_ASSIGN(Data);
+  };
+
+  std::shared_ptr<Data> data_;
+};
 } // namespace kudu
-#endif /* KUDU_UTIL_ASYNC_UTIL_H */