From 2d89d30d665c91fe1d05c28a86a8a16a5f8301d5 Mon Sep 17 00:00:00 2001 From: Carlos O'Ryan Date: Tue, 30 Jun 2020 09:38:29 -0400 Subject: [PATCH 1/2] fix: CompletionQueue::RunAsync is always async Prior to this change we sometimes ran the callable passed to `CompletionQueue::RunAsync()` in the calling thread and not in the thread pool, which was hard to reason about. As part of these changes I also fixed the code to support move-only callables, and added a test for it. The code is "future proofed" to support a `void()` callable. Which is interesting for a number of reasons (think "executors"). --- google/cloud/completion_queue.cc | 39 +++++++++++++++++++ google/cloud/completion_queue.h | 41 ++++++++++++++------ google/cloud/completion_queue_test.cc | 56 +++++++++++++++++++++++++++ 3 files changed, 125 insertions(+), 11 deletions(-) diff --git a/google/cloud/completion_queue.cc b/google/cloud/completion_queue.cc index 3153bcf84bf9a..4041edd32ed41 100644 --- a/google/cloud/completion_queue.cc +++ b/google/cloud/completion_queue.cc @@ -76,6 +76,38 @@ class AsyncTimerFuture : public internal::AsyncGrpcOperation { std::unique_ptr alarm_; }; +class AsyncFunction : public internal::AsyncGrpcOperation { + public: + AsyncFunction(std::unique_ptr fun, + std::unique_ptr alarm) + : fun_(std::move(fun)), alarm_(std::move(alarm)) {} + + void Set(grpc::CompletionQueue& cq, + std::chrono::system_clock::time_point deadline, void* tag) { + if (alarm_) { + alarm_->Set(&cq, deadline, tag); + } + } + + void Cancel() override { + if (alarm_) { + alarm_->Cancel(); + } + } + + private: + bool Notify(bool) override { + fun_->exec(); + fun_.reset(); + return true; + } + + std::shared_ptr cq_; + std::unique_ptr fun_; + // Holds the underlying handle, it might be a nullpotr in tests. + std::unique_ptr alarm_; +}; + } // namespace CompletionQueue::CompletionQueue() : impl_(new internal::CompletionQueueImpl) {} @@ -95,6 +127,13 @@ CompletionQueue::MakeDeadlineTimer( return op->GetFuture(); } +void CompletionQueue::RunAsyncImpl(std::unique_ptr f) { + auto deadline = std::chrono::system_clock::now(); + auto op = std::make_shared(std::move(f), impl_->CreateAlarm()); + impl_->StartOperation( + op, [&](void* tag) { op->Set(impl_->cq(), deadline, tag); }); +} + } // namespace GOOGLE_CLOUD_CPP_NS } // namespace cloud } // namespace google diff --git a/google/cloud/completion_queue.h b/google/cloud/completion_queue.h index 22606916a1eff..fc00834e4ac6a 100644 --- a/google/cloud/completion_queue.h +++ b/google/cloud/completion_queue.h @@ -20,10 +20,20 @@ #include "google/cloud/internal/completion_queue_impl.h" #include "google/cloud/status_or.h" #include "google/cloud/version.h" +#include "absl/memory/memory.h" +#include "absl/meta/type_traits.h" namespace google { namespace cloud { inline namespace GOOGLE_CLOUD_CPP_NS { +namespace internal { +// Type erase the callables in RunAsync() +struct RunAsyncBase { + virtual ~RunAsyncBase() = default; + virtual void exec() = 0; +}; +} // namespace internal + /** * Call the functor associated with asynchronous operations when they complete. */ @@ -178,20 +188,29 @@ class CompletionQueue { typename std::enable_if< internal::CheckRunAsyncCallback::value, int>::type = 0> void RunAsync(Functor&& functor) { - auto impl = impl_; - MakeRelativeTimer(std::chrono::seconds(0)) - .then( - [impl, functor]( - future< - StatusOr>) mutable { - // We intentionally ignore the status here; the functor is always - // called, even after a call to `CancelAll`. - CompletionQueue cq(impl); - functor(cq); - }); + class Wrapper : public internal::RunAsyncBase { + public: + Wrapper(std::weak_ptr impl, Functor&& f) + : impl_(std::move(impl)), fun_(std::forward(f)) {} + ~Wrapper() override = default; + void exec() override { + auto impl = impl_.lock(); + if (!impl) return; + CompletionQueue cq(std::move(impl)); + fun_(cq); + } + + private: + std::weak_ptr impl_; + absl::decay_t fun_; + }; + RunAsyncImpl( + absl::make_unique(impl_, std::forward(functor))); } private: + void RunAsyncImpl(std::unique_ptr); + std::shared_ptr impl_; }; diff --git a/google/cloud/completion_queue_test.cc b/google/cloud/completion_queue_test.cc index 815c8f7bc3b50..1f1c120906ccb 100644 --- a/google/cloud/completion_queue_test.cc +++ b/google/cloud/completion_queue_test.cc @@ -32,6 +32,7 @@ namespace btadmin = ::google::bigtable::admin::v2; namespace btproto = ::google::bigtable::v2; using ::google::cloud::testing_util::MockCompletionQueue; using ::testing::_; +using ::testing::Contains; using ::testing::StrictMock; class MockClient { @@ -341,6 +342,61 @@ TEST(CompletionQueueTest, RunAsyncCompletionQueueDestroyed) { done_promise.get_future().get(); } +TEST(CompletionQueueTest, RunAsyncMoveOnly) { + struct MoveOnly { + promise p; + void operator()(CompletionQueue&) { p.set_value(); } + }; + static_assert(!std::is_copy_assignable::value, + "MoveOnly test type should not copy-assignable"); + + promise p; + auto done = p.get_future(); + CompletionQueue cq; + std::thread t{[&cq] { cq.Run(); }}; + cq.RunAsync(MoveOnly{std::move(p)}); + done.get(); + cq.Shutdown(); + t.join(); +} + +TEST(CompletionQueueTest, RunAsyncThread) { + CompletionQueue cq; + + std::set runner_ids; + auto constexpr kRunners = 8; + std::vector runners(kRunners); + for (auto& t : runners) { + promise started; + auto f = started.get_future(); + t = std::thread( + [&cq](promise p) { + p.set_value(std::this_thread::get_id()); + cq.Run(); + }, + std::move(started)); + runner_ids.insert(f.get()); + } + + auto constexpr kIterations = 10000; + std::vector> pending(kIterations); + std::vector> actual; + for (int i = 0; i != kIterations; ++i) { + auto& p = pending[i]; + actual.push_back(p.get_future()); + cq.RunAsync( + [&p](CompletionQueue&) { p.set_value(std::this_thread::get_id()); }); + } + + for (auto& done : actual) { + auto id = done.get(); + EXPECT_THAT(runner_ids, Contains(id)); + } + + cq.Shutdown(); + for (auto& t : runners) t.join(); +} + // Sets up a timer that reschedules itself and verifies we can shut down // cleanly whether we call `CancelAll()` on the queue first or not. namespace { From aa371516f06f31b2965540b6406e584fc61887da Mon Sep 17 00:00:00 2001 From: Carlos O'Ryan Date: Wed, 1 Jul 2020 08:06:37 -0400 Subject: [PATCH 2/2] Address review comments --- google/cloud/completion_queue.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/google/cloud/completion_queue.cc b/google/cloud/completion_queue.cc index 4041edd32ed41..4ef40dcbcc70e 100644 --- a/google/cloud/completion_queue.cc +++ b/google/cloud/completion_queue.cc @@ -104,7 +104,7 @@ class AsyncFunction : public internal::AsyncGrpcOperation { std::shared_ptr cq_; std::unique_ptr fun_; - // Holds the underlying handle, it might be a nullpotr in tests. + // Holds the underlying handle, it might be a nullptr in tests. std::unique_ptr alarm_; };