You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2018/07/13 06:03:54 UTC
[42/51] [abbrv] impala git commit: IMPALA-7006: Add KRPC folders from
kudu@334ecafd
http://git-wip-us.apache.org/repos/asf/impala/blob/fcf190c4/be/src/kudu/rpc/outbound_call.h
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/outbound_call.h b/be/src/kudu/rpc/outbound_call.h
new file mode 100644
index 0000000..8d43891
--- /dev/null
+++ b/be/src/kudu/rpc/outbound_call.h
@@ -0,0 +1,348 @@
+// 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.
+#ifndef KUDU_RPC_CLIENT_CALL_H
+#define KUDU_RPC_CLIENT_CALL_H
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <ostream>
+#include <set>
+#include <string>
+#include <vector>
+
+#include <gflags/gflags_declare.h>
+#include <glog/logging.h>
+#include <gtest/gtest_prod.h>
+
+#include "kudu/gutil/gscoped_ptr.h"
+#include "kudu/gutil/macros.h"
+#include "kudu/rpc/connection_id.h"
+#include "kudu/rpc/constants.h"
+#include "kudu/rpc/remote_method.h"
+#include "kudu/rpc/response_callback.h"
+#include "kudu/rpc/rpc_header.pb.h"
+#include "kudu/rpc/transfer.h"
+#include "kudu/util/faststring.h"
+#include "kudu/util/locks.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/slice.h"
+#include "kudu/util/status.h"
+
+DECLARE_int32(rpc_inject_cancellation_state);
+
+namespace google {
+namespace protobuf {
+class Message;
+} // namespace protobuf
+} // namespace google
+
+namespace kudu {
+namespace rpc {
+
+class CallResponse;
+class DumpRunningRpcsRequestPB;
+class RpcCallInProgressPB;
+class RpcController;
+class RpcSidecar;
+
+// Tracks the status of a call on the client side.
+//
+// This is an internal-facing class -- clients interact with the
+// RpcController class.
+//
+// This is allocated by the Proxy when a call is first created,
+// then passed to the reactor thread to send on the wire. It's typically
+// kept using a shared_ptr because a call may terminate in any number
+// of different threads, making it tricky to enforce single ownership.
+class OutboundCall {
+ public:
+
+ // Phases of an outbound RPC. Making an outbound RPC might involve establishing
+ // a connection to the remote server first, and the actual call is made only
+ // once the connection to the server is established.
+ enum class Phase {
+ // The phase of connection negotiation between the caller and the callee.
+ CONNECTION_NEGOTIATION,
+
+ // The phase of sending a call over already established connection.
+ REMOTE_CALL,
+ };
+
+ OutboundCall(const ConnectionId& conn_id, const RemoteMethod& remote_method,
+ google::protobuf::Message* response_storage,
+ RpcController* controller, ResponseCallback callback);
+
+ ~OutboundCall();
+
+ // Serialize the given request PB into this call's internal storage, and assume
+ // ownership of any sidecars that should accompany this request.
+ //
+ // Because the request data is fully serialized by this call, 'req' may be subsequently
+ // mutated with no ill effects.
+ void SetRequestPayload(const google::protobuf::Message& req,
+ std::vector<std::unique_ptr<RpcSidecar>>&& sidecars);
+
+ // Assign the call ID for this call. This is called from the reactor
+ // thread once a connection has been assigned. Must only be called once.
+ void set_call_id(int32_t call_id) {
+ DCHECK_EQ(header_.call_id(), kInvalidCallId) << "Already has a call ID";
+ header_.set_call_id(call_id);
+ }
+
+ // Serialize the call for the wire. Requires that SetRequestPayload()
+ // is called first. This is called from the Reactor thread.
+ // Returns the number of slices in the serialized call.
+ size_t SerializeTo(TransferPayload* slices);
+
+ // Mark in the call that cancellation has been requested. If the call hasn't yet
+ // started sending or has finished sending the RPC request but is waiting for a
+ // response, cancel the RPC right away. Otherwise, wait until the RPC has finished
+ // sending before cancelling it. If the call is finished, it's a no-op.
+ // REQUIRES: must be called from the reactor thread.
+ void Cancel();
+
+ // Callback after the call has been put on the outbound connection queue.
+ void SetQueued();
+
+ // Update the call state to show that the request has started being sent
+ // on the socket.
+ void SetSending();
+
+ // Update the call state to show that the request has been sent.
+ void SetSent();
+
+ // Mark the call as failed. This also triggers the callback to notify
+ // the caller. If the call failed due to a remote error, then err_pb
+ // should be set to the error returned by the remote server.
+ void SetFailed(Status status,
+ Phase phase = Phase::REMOTE_CALL,
+ std::unique_ptr<ErrorStatusPB> err_pb = nullptr);
+
+ // Mark the call as timed out. This also triggers the callback to notify
+ // the caller.
+ void SetTimedOut(Phase phase);
+ bool IsTimedOut() const;
+
+ bool IsNegotiationError() const;
+
+ bool IsCancelled() const;
+
+ // Is the call finished?
+ bool IsFinished() const;
+
+ // Fill in the call response.
+ void SetResponse(gscoped_ptr<CallResponse> resp);
+
+ const std::set<RpcFeatureFlag>& required_rpc_features() const {
+ return required_rpc_features_;
+ }
+
+ std::string ToString() const;
+
+ void DumpPB(const DumpRunningRpcsRequestPB& req, RpcCallInProgressPB* resp);
+
+ ////////////////////////////////////////////////////////////
+ // Getters
+ ////////////////////////////////////////////////////////////
+
+ const ConnectionId& conn_id() const { return conn_id_; }
+ const RemoteMethod& remote_method() const { return remote_method_; }
+ const ResponseCallback &callback() const { return callback_; }
+ RpcController* controller() { return controller_; }
+ const RpcController* controller() const { return controller_; }
+
+ // Return true if a call ID has been assigned to this call.
+ bool call_id_assigned() const {
+ return header_.call_id() != kInvalidCallId;
+ }
+
+ int32_t call_id() const {
+ DCHECK(call_id_assigned());
+ return header_.call_id();
+ }
+
+ // Returns true if cancellation has been requested. Must be called from
+ // reactor thread.
+ bool cancellation_requested() const {
+ return cancellation_requested_;
+ }
+
+ // Test function which returns true if a cancellation request should be injected
+ // at the current state.
+ bool ShouldInjectCancellation() const {
+ return FLAGS_rpc_inject_cancellation_state != -1 &&
+ FLAGS_rpc_inject_cancellation_state == state();
+ }
+
+ private:
+ friend class RpcController;
+ FRIEND_TEST(TestRpc, TestCancellation);
+
+ // Various states the call propagates through.
+ // NB: if adding another state, be sure to update OutboundCall::IsFinished()
+ // and OutboundCall::StateName(State state) as well.
+ enum State {
+ READY = 0,
+ ON_OUTBOUND_QUEUE,
+ SENDING,
+ SENT,
+ NEGOTIATION_TIMED_OUT,
+ TIMED_OUT,
+ CANCELLED,
+ FINISHED_NEGOTIATION_ERROR,
+ FINISHED_ERROR,
+ FINISHED_SUCCESS
+ };
+
+ static std::string StateName(State state);
+
+ // Mark the call as cancelled. This also invokes the callback to notify the caller.
+ void SetCancelled();
+
+ void set_state(State new_state);
+ State state() const;
+
+ // Same as set_state, but requires that the caller already holds
+ // lock_
+ void set_state_unlocked(State new_state);
+
+ // return current status
+ Status status() const;
+
+ // Time when the call was first initiatied.
+ MonoTime start_time_;
+
+ // Return the error protobuf, if a remote error occurred.
+ // This will only be non-NULL if status().IsRemoteError().
+ const ErrorStatusPB* error_pb() const;
+
+ // Lock for state_ status_, error_pb_ fields, since they
+ // may be mutated by the reactor thread while the client thread
+ // reads them.
+ mutable simple_spinlock lock_;
+ State state_;
+ Status status_;
+ std::unique_ptr<ErrorStatusPB> error_pb_;
+
+ // Call the user-provided callback. Note that entries in 'sidecars_' are cleared
+ // prior to invoking the callback so the client can assume that the call doesn't
+ // hold references to outbound sidecars.
+ void CallCallback();
+
+ // The RPC header.
+ // Parts of this (eg the call ID) are only assigned once this call has been
+ // passed to the reactor thread and assigned a connection.
+ RequestHeader header_;
+
+ // The remote method being called.
+ RemoteMethod remote_method_;
+
+ // RPC-system features required to send this call.
+ std::set<RpcFeatureFlag> required_rpc_features_;
+
+ const ConnectionId conn_id_;
+ ResponseCallback callback_;
+ RpcController* controller_;
+
+ // Pointer for the protobuf where the response should be written.
+ google::protobuf::Message* response_;
+
+ // Buffers for storing segments of the wire-format request.
+ faststring header_buf_;
+ faststring request_buf_;
+
+ // Once a response has been received for this call, contains that response.
+ // Otherwise NULL.
+ gscoped_ptr<CallResponse> call_response_;
+
+ // All sidecars to be sent with this call.
+ std::vector<std::unique_ptr<RpcSidecar>> sidecars_;
+
+ // Total size in bytes of all sidecars in 'sidecars_'. Set in SetRequestPayload().
+ // This cannot exceed TransferLimits::kMaxTotalSidecarBytes.
+ int32_t sidecar_byte_size_ = -1;
+
+ // True if cancellation was requested on this call.
+ bool cancellation_requested_;
+
+ DISALLOW_COPY_AND_ASSIGN(OutboundCall);
+};
+
+// A response to a call, on the client side.
+// Upon receiving a response, this is allocated in the reactor thread and filled
+// into the OutboundCall instance via OutboundCall::SetResponse.
+//
+// This may either be a success or error response.
+//
+// This class takes care of separating out the distinct payload slices sent
+// over.
+class CallResponse {
+ public:
+ CallResponse();
+
+ // Parse the response received from a call. This must be called before any
+ // other methods on this object.
+ Status ParseFrom(gscoped_ptr<InboundTransfer> transfer);
+
+ // Return true if the call succeeded.
+ bool is_success() const {
+ DCHECK(parsed_);
+ return !header_.is_error();
+ }
+
+ // Return the call ID that this response is related to.
+ int32_t call_id() const {
+ DCHECK(parsed_);
+ return header_.call_id();
+ }
+
+ // Return the serialized response data. This is just the response "body" --
+ // either a serialized ErrorStatusPB, or the serialized user response protobuf.
+ const Slice &serialized_response() const {
+ DCHECK(parsed_);
+ return serialized_response_;
+ }
+
+ // See RpcController::GetSidecar()
+ Status GetSidecar(int idx, Slice* sidecar) const;
+
+ private:
+ // True once ParseFrom() is called.
+ bool parsed_;
+
+ // The parsed header.
+ ResponseHeader header_;
+
+ // The slice of data for the encoded protobuf response.
+ // This slice refers to memory allocated by transfer_
+ Slice serialized_response_;
+
+ // Slices of data for rpc sidecars. They point into memory owned by transfer_.
+ Slice sidecar_slices_[TransferLimits::kMaxSidecars];
+
+ // The incoming transfer data - retained because serialized_response_
+ // and sidecar_slices_ refer into its data.
+ gscoped_ptr<InboundTransfer> transfer_;
+
+ DISALLOW_COPY_AND_ASSIGN(CallResponse);
+};
+
+} // namespace rpc
+} // namespace kudu
+
+#endif
http://git-wip-us.apache.org/repos/asf/impala/blob/fcf190c4/be/src/kudu/rpc/periodic-test.cc
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/periodic-test.cc b/be/src/kudu/rpc/periodic-test.cc
new file mode 100644
index 0000000..cd793bd
--- /dev/null
+++ b/be/src/kudu/rpc/periodic-test.cc
@@ -0,0 +1,295 @@
+// 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 <atomic>
+#include <cstdint>
+#include <memory>
+#include <ostream>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+
+#include "kudu/rpc/messenger.h"
+#include "kudu/rpc/periodic.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/scoped_cleanup.h"
+#include "kudu/util/stopwatch.h"
+#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
+
+using std::atomic;
+using std::shared_ptr;
+using std::vector;
+
+namespace kudu {
+namespace rpc {
+
+class PeriodicTimerTest : public KuduTest {
+ public:
+ PeriodicTimerTest()
+ : period_ms_(200) {}
+
+ protected:
+ const int64_t period_ms_;
+};
+
+class JitteredPeriodicTimerTest : public PeriodicTimerTest,
+ public ::testing::WithParamInterface<double> {
+ public:
+ // In TSAN builds it takes a long time to de-schedule a thread. Also,
+ // the actual time that thread spends sleeping in SleepFor() scenarios
+ // might be much longer than requested. Setting the task period to be long
+ // enough allows for more stable behavior of the test, so no flakiness
+ // is observed even under substantial load. Otherwise it would be necessary
+ // to introduce additional logic to verify that the actual timings satisfy
+ // the implicit constraints of the test scenarios below.
+ JitteredPeriodicTimerTest()
+ : counter_(0) {
+ }
+
+ virtual void SetUp() override {
+ PeriodicTimerTest::SetUp();
+
+ MessengerBuilder builder("test");
+ ASSERT_OK(builder.Build(&messenger_));
+
+ timer_ = PeriodicTimer::Create(messenger_,
+ [&] { counter_++; },
+ MonoDelta::FromMilliseconds(period_ms_),
+ GetOptions());
+ }
+
+ virtual void TearDown() override {
+ // Ensure that the reactor threads are fully quiesced (and thus no timer
+ // callbacks are running) by the time 'counter_' is destroyed.
+ messenger_->Shutdown();
+
+ KuduTest::TearDown();
+ }
+
+ protected:
+
+ virtual PeriodicTimer::Options GetOptions() {
+ PeriodicTimer::Options opts;
+ opts.jitter_pct = GetParam();
+ return opts;
+ }
+
+ atomic<int64_t> counter_;
+ shared_ptr<Messenger> messenger_;
+ shared_ptr<PeriodicTimer> timer_;
+};
+
+INSTANTIATE_TEST_CASE_P(AllJitterModes,
+ JitteredPeriodicTimerTest,
+ ::testing::Values(0.0, 0.25));
+
+TEST_P(JitteredPeriodicTimerTest, TestStartStop) {
+ // Before the timer starts, the counter's value should not change.
+ SleepFor(MonoDelta::FromMilliseconds(period_ms_ * 2));
+ ASSERT_EQ(0, counter_);
+
+ // Once started, it should increase (exactly how much depends on load and the
+ // underlying OS scheduler).
+ timer_->Start();
+ SleepFor(MonoDelta::FromMilliseconds(period_ms_ * 2));
+ ASSERT_EVENTUALLY([&]{
+ ASSERT_GT(counter_, 0);
+ });
+
+ // After stopping the timer, the value should either remain the same or
+ // increment once (if Stop() raced with a scheduled task).
+ timer_->Stop();
+ int64_t v = counter_;
+ messenger_->Shutdown();
+ ASSERT_TRUE(counter_ == v ||
+ counter_ == v + 1);
+}
+
+TEST_P(JitteredPeriodicTimerTest, TestReset) {
+ timer_->Start();
+ MonoTime start_time = MonoTime::Now();
+
+ // Loop for a little while, resetting the timer's period over and over. As a
+ // result, the timer should never fire.
+ while (true) {
+ MonoTime now = MonoTime::Now();
+ if (now - start_time > MonoDelta::FromMilliseconds(period_ms_ * 5)) {
+ break;
+ }
+ timer_->Snooze();
+ ASSERT_EQ(0, counter_);
+ SleepFor(MonoDelta::FromMilliseconds(1));
+ }
+}
+
+TEST_P(JitteredPeriodicTimerTest, TestResetWithDelta) {
+ timer_->Start();
+ timer_->Snooze(MonoDelta::FromMilliseconds(period_ms_ * 5));
+
+ // One period later, the counter still hasn't incremented...
+ SleepFor(MonoDelta::FromMilliseconds(period_ms_));
+ ASSERT_EQ(0, counter_);
+
+ // ...but it will increment eventually.
+ ASSERT_EVENTUALLY([&](){
+ ASSERT_GT(counter_, 0);
+ });
+}
+
+TEST_P(JitteredPeriodicTimerTest, TestStartWithDelta) {
+ timer_->Start(MonoDelta::FromMilliseconds(period_ms_ * 5));
+
+ // One period later, the counter still hasn't incremented...
+ SleepFor(MonoDelta::FromMilliseconds(period_ms_));
+ ASSERT_EQ(0, counter_);
+
+ // ...but it will increment eventually.
+ ASSERT_EVENTUALLY([&](){
+ ASSERT_GT(counter_, 0);
+ });
+}
+
+TEST_F(PeriodicTimerTest, TestCallbackRestartsTimer) {
+ const int64_t kPeriods = 10;
+
+ shared_ptr<Messenger> messenger;
+ ASSERT_OK(MessengerBuilder("test").Build(&messenger));
+
+ // Create a timer that restarts itself from within its functor.
+ PeriodicTimer::Options opts;
+ opts.jitter_pct = 0.0; // don't need jittering
+ shared_ptr<PeriodicTimer> timer = PeriodicTimer::Create(
+ messenger,
+ [&] {
+ timer->Stop();
+ timer->Start();
+ },
+ MonoDelta::FromMilliseconds(period_ms_),
+ std::move(opts));
+
+ // Run the timer for a fixed amount of time.
+ timer->Start();
+ SleepFor(MonoDelta::FromMilliseconds(period_ms_ * kPeriods));
+ timer->Stop();
+
+ // Although the timer is restarted by its functor, its overall period should
+ // remain more or less the same (since the period expired just as the functor
+ // ran). As such, we should see no more than three callbacks per period:
+ // one to start scheduling the callback loop, one when it fires, and one more
+ // after it has been replaced by a new callback loop.
+ ASSERT_LE(timer->NumCallbacksForTests(), kPeriods * 3);
+}
+
+class JitteredOneShotPeriodicTimerTest : public JitteredPeriodicTimerTest {
+ protected:
+ virtual PeriodicTimer::Options GetOptions() override {
+ PeriodicTimer::Options opts;
+ opts.jitter_pct = GetParam();
+ opts.one_shot = true;
+ return opts;
+ }
+};
+
+INSTANTIATE_TEST_CASE_P(AllJitterModes,
+ JitteredOneShotPeriodicTimerTest,
+ ::testing::Values(0.0, 0.25));
+
+TEST_P(JitteredOneShotPeriodicTimerTest, TestBasics) {
+ // Kick off the one-shot timer a few times.
+ for (int i = 0; i < 3; i++) {
+ ASSERT_EQ(i, counter_);
+
+ // Eventually the task will run.
+ timer_->Start();
+ ASSERT_EVENTUALLY([&](){
+ ASSERT_EQ(i + 1, counter_);
+ });
+
+ // Even if we explicitly wait another few periods, the counter value
+ // shouldn't change.
+ SleepFor(MonoDelta::FromMilliseconds(period_ms_ * 2));
+ ASSERT_EQ(i + 1, counter_);
+ }
+}
+
+TEST_F(PeriodicTimerTest, TestCallbackRestartsOneShotTimer) {
+ atomic<int64_t> counter(0);
+ shared_ptr<Messenger> messenger;
+ ASSERT_OK(MessengerBuilder("test")
+ .Build(&messenger));
+
+ // Create a timer that restarts itself from within its functor.
+ PeriodicTimer::Options opts;
+ opts.jitter_pct = 0.0; // don't need jittering
+ opts.one_shot = true;
+ shared_ptr<PeriodicTimer> timer = PeriodicTimer::Create(
+ messenger,
+ [&] {
+ counter++;
+ timer->Start();
+ },
+ MonoDelta::FromMilliseconds(period_ms_),
+ std::move(opts));
+
+ // Because the timer restarts itself every time the functor runs, we
+ // should see the counter value increase with each period.
+ timer->Start();
+ ASSERT_EVENTUALLY([&](){
+ ASSERT_GE(counter, 5);
+ });
+
+ // Ensure that the reactor threads are fully quiesced (and thus no timer
+ // callbacks are running) by the time 'counter' is destroyed.
+ messenger->Shutdown();
+}
+
+TEST_F(PeriodicTimerTest, TestPerformance) {
+ const int kNumTimers = 1000;
+ shared_ptr<Messenger> messenger;
+ ASSERT_OK(MessengerBuilder("test")
+ .set_num_reactors(1)
+ .Build(&messenger));
+ SCOPED_CLEANUP({ messenger->Shutdown(); });
+
+ vector<shared_ptr<PeriodicTimer>> timers;
+ for (int i = 0; i < kNumTimers; i++) {
+ timers.emplace_back(PeriodicTimer::Create(
+ messenger,
+ [&] {}, // No-op.
+ MonoDelta::FromMilliseconds(10)));
+ timers.back()->Start();
+ }
+
+ Stopwatch sw(Stopwatch::ALL_THREADS);
+ sw.start();
+ SleepFor(MonoDelta::FromSeconds(1));
+ sw.stop();
+ LOG(INFO) << "User CPU for running " << kNumTimers << " timers for 1 second: "
+ << sw.elapsed().user_cpu_seconds() << "s";
+
+ for (auto& t : timers) {
+ t->Stop();
+ }
+
+}
+
+} // namespace rpc
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/impala/blob/fcf190c4/be/src/kudu/rpc/periodic.cc
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/periodic.cc b/be/src/kudu/rpc/periodic.cc
new file mode 100644
index 0000000..9ab9ce9
--- /dev/null
+++ b/be/src/kudu/rpc/periodic.cc
@@ -0,0 +1,219 @@
+// 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/rpc/periodic.h"
+
+#include <algorithm>
+#include <memory>
+#include <mutex>
+
+#include <boost/function.hpp>
+#include <glog/logging.h>
+
+#include "kudu/rpc/messenger.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/random.h"
+#include "kudu/util/random_util.h"
+#include "kudu/util/status.h"
+
+using std::shared_ptr;
+using std::weak_ptr;
+
+namespace kudu {
+namespace rpc {
+
+PeriodicTimer::Options::Options()
+ : jitter_pct(0.25),
+ one_shot(false) {
+}
+
+shared_ptr<PeriodicTimer> PeriodicTimer::Create(
+ shared_ptr<Messenger> messenger,
+ RunTaskFunctor functor,
+ MonoDelta period,
+ Options options) {
+ return PeriodicTimer::make_shared(
+ std::move(messenger), std::move(functor), period, options);
+}
+
+PeriodicTimer::PeriodicTimer(
+ shared_ptr<Messenger> messenger,
+ RunTaskFunctor functor,
+ MonoDelta period,
+ Options options)
+ : messenger_(std::move(messenger)),
+ functor_(std::move(functor)),
+ period_(period),
+ options_(options),
+ rng_(GetRandomSeed32()),
+ current_callback_generation_(0),
+ num_callbacks_for_tests_(0),
+ started_(false) {
+ DCHECK_GE(options_.jitter_pct, 0);
+ DCHECK_LE(options_.jitter_pct, 1);
+}
+
+PeriodicTimer::~PeriodicTimer() {
+ Stop();
+}
+
+void PeriodicTimer::Start(boost::optional<MonoDelta> next_task_delta) {
+ std::unique_lock<simple_spinlock> l(lock_);
+ if (!started_) {
+ started_ = true;
+ SnoozeUnlocked(std::move(next_task_delta));
+ int new_callback_generation = ++current_callback_generation_;
+
+ // Invoke Callback() with the lock released.
+ l.unlock();
+ Callback(new_callback_generation);
+ }
+}
+
+void PeriodicTimer::Stop() {
+ std::lock_guard<simple_spinlock> l(lock_);
+ StopUnlocked();
+}
+
+void PeriodicTimer::StopUnlocked() {
+ DCHECK(lock_.is_locked());
+ started_ = false;
+}
+
+void PeriodicTimer::Snooze(boost::optional<MonoDelta> next_task_delta) {
+ std::lock_guard<simple_spinlock> l(lock_);
+ SnoozeUnlocked(std::move(next_task_delta));
+}
+
+void PeriodicTimer::SnoozeUnlocked(boost::optional<MonoDelta> next_task_delta) {
+ DCHECK(lock_.is_locked());
+ if (!started_) {
+ return;
+ }
+
+ if (!next_task_delta) {
+ // Given jitter percentage J and period P, this yields a delay somewhere
+ // between (1-J)*P and (1+J)*P.
+ next_task_delta = MonoDelta::FromMilliseconds(
+ GetMinimumPeriod().ToMilliseconds() +
+ rng_.NextDoubleFraction() *
+ options_.jitter_pct *
+ (2 * period_.ToMilliseconds()));
+ }
+ next_task_time_ = MonoTime::Now() + *next_task_delta;
+}
+
+bool PeriodicTimer::started() const {
+ std::lock_guard<simple_spinlock> l(lock_);
+ return started_;
+}
+
+MonoDelta PeriodicTimer::GetMinimumPeriod() {
+ // Given jitter percentage J and period P, this returns (1-J)*P, which is
+ // the lowest possible jittered value.
+ return MonoDelta::FromMilliseconds((1.0 - options_.jitter_pct) *
+ period_.ToMilliseconds());
+}
+
+int64_t PeriodicTimer::NumCallbacksForTests() const {
+ std::lock_guard<simple_spinlock> l(lock_);
+ return num_callbacks_for_tests_;
+}
+
+void PeriodicTimer::Callback(int64_t my_callback_generation) {
+ // To simplify the implementation, a timer may have only one outstanding
+ // callback scheduled at a time. This means that once the callback is
+ // scheduled, the timer's task cannot run any earlier than whenever the
+ // callback runs. Thus, the delay used when scheduling the callback dictates
+ // the lowest possible value of 'next_task_delta' that Snooze() can honor.
+ //
+ // If the callback's delay is very low, Snooze() can honor a low
+ // 'next_task_delta', but the callback will run often and burn more CPU
+ // cycles. If the delay is very high, the timer will be more efficient but
+ // the granularity for 'next_task_delta' will rise accordingly.
+ //
+ // As a "happy medium" we use GetMinimumPeriod() as the delay. This ensures
+ // that a no-arg Snooze() on a jittered timer will always be honored, and as
+ // long as the caller passes a value of at least GetMinimumPeriod() to
+ // Snooze(), that too will be honored.
+ MonoDelta delay = GetMinimumPeriod();
+ bool run_task = false;
+ {
+ std::lock_guard<simple_spinlock> l(lock_);
+ num_callbacks_for_tests_++;
+
+ // If the timer was stopped, exit.
+ if (!started_) {
+ return;
+ }
+
+ // If there's a new callback loop in town, exit.
+ //
+ // We could check again just before calling Messenger::ScheduleOnReactor()
+ // (in case someone else restarted the timer while the functor ran, or in
+ // case the functor itself restarted the timer), but there's no real reason
+ // to do so: the very next iteration of this callback loop will wind up here
+ // and exit.
+ if (current_callback_generation_ > my_callback_generation) {
+ return;
+ }
+
+ MonoTime now = MonoTime::Now();
+ if (now < next_task_time_) {
+ // It's not yet time to run the task. Reduce the scheduled delay if
+ // enough time has elapsed, but don't increase it.
+ delay = std::min(delay, next_task_time_ - now);
+ } else {
+ // It's time to run the task. Although the next task time is reset now,
+ // it may be reset again by virtue of running the task itself.
+ run_task = true;
+
+ if (options_.one_shot) {
+ // Stop the timer first, in case the task wants to restart it.
+ StopUnlocked();
+ }
+ }
+ }
+
+ if (run_task) {
+ functor_();
+
+ if (options_.one_shot) {
+ // The task was run; exit the loop. Even if the task restarted the timer,
+ // that will have started a new callback loop, so exiting here is always
+ // the correct thing to do.
+ return;
+ }
+ Snooze();
+ }
+
+ // Capture a weak_ptr reference into the submitted functor so that we can
+ // safely handle the functor outliving its timer.
+ weak_ptr<PeriodicTimer> w = shared_from_this();
+ messenger_->ScheduleOnReactor([w, my_callback_generation](const Status& s) {
+ if (!s.ok()) {
+ // The reactor was shut down.
+ return;
+ }
+ if (auto timer = w.lock()) {
+ timer->Callback(my_callback_generation);
+ }
+ }, delay);
+}
+
+} // namespace rpc
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/impala/blob/fcf190c4/be/src/kudu/rpc/periodic.h
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/periodic.h b/be/src/kudu/rpc/periodic.h
new file mode 100644
index 0000000..592cbd5
--- /dev/null
+++ b/be/src/kudu/rpc/periodic.h
@@ -0,0 +1,215 @@
+// 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.
+
+#pragma once
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+
+#include <boost/optional/optional.hpp>
+#include <gtest/gtest_prod.h>
+
+#include "kudu/gutil/macros.h"
+#include "kudu/util/locks.h"
+#include "kudu/util/make_shared.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/random.h"
+
+namespace kudu {
+namespace rpc {
+
+class Messenger;
+
+// Repeatedly runs a task on a fixed period.
+//
+// PeriodicTimer's periodicity is maintained via tail recursive calls to
+// Messenger::ScheduleOnReactor(). Every time the scheduled callback is
+// invoked, it checks the current time, updates some internal bookkeeping,
+// runs the user's task if the time is right, and makes another call to
+// Messenger::ScheduleOnReactor() to run itself again in the future. This
+// looping behavior is called a "callback loop".
+//
+// Every time Stop() and then Start() (or just Start(), if this is the first
+// such call) are invoked, PeriodicTimer will kick off a new callback loop. If
+// there was an old loop, it remains intact until its scheduled callback runs,
+// at which point it will detect that a new loop was created and exit.
+//
+// The use of Messenger::ScheduleOnReactor() is merely for convenience;
+// PeriodicTimer could also be built on libev, a hashed wheel timer, o
+// something equivalent.
+//
+// PeriodicTimers have shared ownership, but that's largely an implementation
+// detail to support asynchronous stopping. Users can treat them as exclusively
+// owned (though care must be taken when writing the task functor; see Stop()
+// for more details).
+//
+// TODO(adar): eventually we should build directly on libev as it supports
+// timer cancelation, which would allow us to implement synchronous Stop(), use
+// exclusive ownership, and remove the restriction that the delta passed
+// into Snooze() be greater than GetMinimumPeriod().
+class PeriodicTimer : public std::enable_shared_from_this<PeriodicTimer>,
+ public enable_make_shared<PeriodicTimer> {
+ public:
+ typedef std::function<void(void)> RunTaskFunctor;
+
+ struct Options {
+ Options();
+
+ // Defines the percentage of the period that will be jittered up or down
+ // randomly. Together with the period, the periodicity of the timer will
+ // vary between (1-J)*P and (1+J)*P.
+ //
+ // Must be between 0 and 1.
+ //
+ // If not set, defaults to 0.25.
+ double jitter_pct;
+
+ // The timer will automatically stop after running the user's task.
+ //
+ // Just as with a normal timer, Snooze() will postpone the running of the
+ // task, and Stop() will cancel the task outright. Unlike a normal timer,
+ // both operations will no-op if the timer has already fired.
+ //
+ // If not set, defaults to false.
+ bool one_shot;
+ };
+
+ // Creates a new PeriodicTimer.
+ //
+ // A ref is taken on 'messenger', which is used for scheduling callbacks.
+ //
+ // 'functor' defines the user's task and is owned for the lifetime of the
+ // PeriodicTimer. The task will run on the messenger's reactor threads so it
+ // should do very little work (i.e. no I/O).
+ //
+ // 'period' defines the period between tasks.
+ //
+ // 'options' allows additional (optional) customization of the timer.
+ static std::shared_ptr<PeriodicTimer> Create(
+ std::shared_ptr<Messenger> messenger,
+ RunTaskFunctor functor,
+ MonoDelta period,
+ Options options = {});
+
+ ~PeriodicTimer();
+
+ // Starts the timer.
+ //
+ // The timer's task will run in accordance with the period and jitter mode
+ // provided during timer construction.
+ //
+ // If 'next_task_delta' is set, it is used verbatim as the delay for the very
+ // first task, with the configured period and jitter mode only applying to
+ // subsequent tasks.
+ //
+ // Does nothing if the timer was already started.
+ void Start(boost::optional<MonoDelta> next_task_delta = boost::none);
+
+ // Snoozes the timer for one period.
+ //
+ // If 'next_task_delta' is set, it is used verbatim as the delay for the next
+ // task. Subsequent tasks will revert to the timer's regular period. The
+ // value of 'next_task_delta' must be greater than GetMinimumPeriod();
+ // otherwise the task is not guaranteed to run in a timely manner.
+ //
+ // Note: Snooze() is not additive. That is, if called at time X and again at
+ // time X + P/2, the timer is snoozed until X+P/2+P, not X+2P.
+ //
+ // Does nothing if the timer is stopped.
+ void Snooze(boost::optional<MonoDelta> next_task_delta = boost::none);
+
+ // Stops the timer.
+ //
+ // Stopping is asynchronous; that is, it is still possible for the task to
+ // run after Stop() returns. Because of this, the task's functor should be
+ // written to do nothing if objects it depends on have been destroyed.
+ //
+ // Does nothing if the timer is already stopped.
+ void Stop();
+
+ // Returns true iff the timer has been started.
+ bool started() const;
+
+ protected:
+ PeriodicTimer(std::shared_ptr<Messenger> messenger,
+ RunTaskFunctor functor,
+ MonoDelta period,
+ Options options);
+
+ private:
+ FRIEND_TEST(PeriodicTimerTest, TestCallbackRestartsTimer);
+ // Calculate the minimum period for the timer, which varies depending on
+ // 'jitter_pct_' and the output of the PRNG.
+ MonoDelta GetMinimumPeriod();
+
+ // Called by Messenger::ScheduleOnReactor when the timer fires.
+ // 'my_callback_generation' is the callback generation assigned to this loop
+ // when it was constructed.
+ void Callback(int64_t my_callback_generation);
+
+ // Like Stop() but must be called with 'lock_' held.
+ void StopUnlocked();
+
+ // Like Snooze() but must be called with 'lock_' held.
+ void SnoozeUnlocked(boost::optional<MonoDelta> next_task_delta = boost::none);
+
+ // Returns the number of times that Callback() has been called by this timer.
+ //
+ // Should only be used for tests!
+ int64_t NumCallbacksForTests() const;
+
+ // Schedules invocations of Callback() in the future.
+ std::shared_ptr<Messenger> messenger_;
+
+ // User-defined task functor.
+ RunTaskFunctor functor_;
+
+ // User-specified task period.
+ const MonoDelta period_;
+
+ // User-specified options.
+ const Options options_;
+
+ // Protects all mutable state below.
+ mutable simple_spinlock lock_;
+
+ // PRNG used when generating jitter.
+ Random rng_;
+
+ // The next time at which the task's functor should be run.
+ MonoTime next_task_time_;
+
+ // The most recent callback generation.
+ //
+ // When started, a callback loop is assigned a generation, which it remembers
+ // for its entire lifespan. If 'current_callback_generation_' exceeds the
+ // loop's assigned generation, that means another loop has been created and
+ // the (now old) loop should exit.
+ int64_t current_callback_generation_;
+
+ // The number of times that Callback() has been invoked.
+ int64_t num_callbacks_for_tests_;
+
+ // Whether the timer is running or not.
+ bool started_;
+
+ DISALLOW_COPY_AND_ASSIGN(PeriodicTimer);
+};
+
+} // namespace rpc
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/impala/blob/fcf190c4/be/src/kudu/rpc/protoc-gen-krpc.cc
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/protoc-gen-krpc.cc b/be/src/kudu/rpc/protoc-gen-krpc.cc
new file mode 100644
index 0000000..cc25f67
--- /dev/null
+++ b/be/src/kudu/rpc/protoc-gen-krpc.cc
@@ -0,0 +1,691 @@
+// 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.
+
+////////////////////////////////////////////////////////////////////////////////
+// Example usage:
+// protoc --plugin=protoc-gen-krpc --krpc_out . --proto_path . <file>.proto
+////////////////////////////////////////////////////////////////////////////////
+
+#include <cstddef>
+#include <map>
+#include <memory>
+#include <set>
+#include <string>
+#include <vector>
+
+#include <boost/optional/optional.hpp>
+#include <glog/logging.h>
+#include <google/protobuf/compiler/code_generator.h>
+#include <google/protobuf/compiler/plugin.h>
+#include <google/protobuf/descriptor.h>
+#include <google/protobuf/descriptor.pb.h>
+#include <google/protobuf/io/printer.h>
+#include <google/protobuf/io/zero_copy_stream.h>
+
+#include "kudu/gutil/gscoped_ptr.h"
+#include "kudu/gutil/port.h"
+#include "kudu/gutil/strings/join.h"
+#include "kudu/gutil/strings/numbers.h"
+#include "kudu/gutil/strings/split.h"
+#include "kudu/gutil/strings/stringpiece.h"
+#include "kudu/gutil/strings/strip.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/gutil/strings/util.h"
+#include "kudu/rpc/rpc_header.pb.h"
+#include "kudu/util/status.h"
+#include "kudu/util/string_case.h"
+
+using boost::optional;
+using google::protobuf::FileDescriptor;
+using google::protobuf::io::Printer;
+using google::protobuf::MethodDescriptor;
+using google::protobuf::ServiceDescriptor;
+using std::map;
+using std::shared_ptr;
+using std::set;
+using std::string;
+using std::vector;
+
+namespace kudu {
+namespace rpc {
+
+namespace {
+
+// Return the name of the authorization method specified for this
+// RPC method, or boost::none if none is specified.
+//
+// This handles fallback to the service-wide default.
+optional<string> GetAuthzMethod(const MethodDescriptor& method) {
+ if (method.options().HasExtension(authz_method)) {
+ return method.options().GetExtension(authz_method);
+ }
+ if (method.service()->options().HasExtension(default_authz_method)) {
+ return method.service()->options().GetExtension(default_authz_method);
+ }
+ return boost::none;
+}
+
+} // anonymous namespace
+
+class Substituter {
+ public:
+ virtual ~Substituter() {}
+ virtual void InitSubstitutionMap(map<string, string> *map) const = 0;
+};
+
+// NameInfo contains information about the output names.
+class FileSubstitutions : public Substituter {
+ public:
+ static const std::string kProtoExtension;
+
+ Status Init(const FileDescriptor *file) {
+ const string& path = file->name();
+ map_["path"] = path;
+
+ // Initialize path_
+ // If path = /foo/bar/baz_stuff.proto, path_ = /foo/bar/baz_stuff
+ if (!TryStripSuffixString(path, kProtoExtension, &path_no_extension_)) {
+ return Status::InvalidArgument("file name " + path +
+ " did not end in " + kProtoExtension);
+ }
+ map_["path_no_extension"] = path_no_extension_;
+
+ // If path = /foo/bar/baz_stuff.proto, base_ = baz_stuff
+ string base;
+ GetBaseName(path_no_extension_, &base);
+ map_["base"] = base;
+
+ // If path = /foo/bar/baz_stuff.proto, camel_case_ = BazStuff
+ string camel_case;
+ SnakeToCamelCase(base, &camel_case);
+ map_["camel_case"] = camel_case;
+
+ // If path = /foo/bar/baz_stuff.proto, upper_case_ = BAZ_STUFF
+ string upper_case;
+ ToUpperCase(base, &upper_case);
+ map_["upper_case"] = upper_case;
+
+ map_["open_namespace"] = GenerateOpenNamespace(file->package());
+ map_["close_namespace"] = GenerateCloseNamespace(file->package());
+
+ return Status::OK();
+ }
+
+ virtual void InitSubstitutionMap(map<string, string> *map) const OVERRIDE {
+ typedef std::map<string, string>::value_type kv_pair;
+ for (const kv_pair &pair : map_) {
+ (*map)[pair.first] = pair.second;
+ }
+ }
+
+ std::string service_header() const {
+ return path_no_extension_ + ".service.h";
+ }
+
+ std::string service() const {
+ return path_no_extension_ + ".service.cc";
+ }
+
+ std::string proxy_header() const {
+ return path_no_extension_ + ".proxy.h";
+ }
+
+ std::string proxy() const {
+ return path_no_extension_ + ".proxy.cc";
+ }
+
+ private:
+ // Extract the last filename component.
+ static void GetBaseName(const string &path,
+ string *base) {
+ size_t last_slash = path.find_last_of('/');
+ if (last_slash != string::npos) {
+ *base = path.substr(last_slash + 1);
+ } else {
+ *base = path;
+ }
+ }
+
+ static string GenerateOpenNamespace(const string &str) {
+ vector<string> components = strings::Split(str, ".");
+ string out;
+ for (const string &c : components) {
+ out.append("namespace ").append(c).append(" {\n");
+ }
+ return out;
+ }
+
+ static string GenerateCloseNamespace(const string &str) {
+ vector<string> components = strings::Split(str, ".");
+ string out;
+ for (auto c = components.crbegin(); c != components.crend(); c++) {
+ out.append("} // namespace ").append(*c).append("\n");
+ }
+ return out;
+ }
+
+ std::string path_no_extension_;
+ map<string, string> map_;
+};
+
+const std::string FileSubstitutions::kProtoExtension(".proto");
+
+class MethodSubstitutions : public Substituter {
+ public:
+ explicit MethodSubstitutions(const MethodDescriptor *method)
+ : method_(method) {
+ }
+
+ virtual void InitSubstitutionMap(map<string, string> *map) const OVERRIDE {
+
+ (*map)["rpc_name"] = method_->name();
+ (*map)["rpc_full_name"] = method_->full_name();
+ (*map)["rpc_full_name_plainchars"] =
+ StringReplace(method_->full_name(), ".", "_", true);
+ (*map)["request"] =
+ ReplaceNamespaceDelimiters(
+ StripNamespaceIfPossible(method_->service()->full_name(),
+ method_->input_type()->full_name()));
+ (*map)["response"] =
+ ReplaceNamespaceDelimiters(
+ StripNamespaceIfPossible(method_->service()->full_name(),
+ method_->output_type()->full_name()));
+ (*map)["metric_enum_key"] = strings::Substitute("kMetricIndex$0", method_->name());
+ bool track_result = static_cast<bool>(method_->options().GetExtension(track_rpc_result));
+ (*map)["track_result"] = track_result ? " true" : "false";
+ (*map)["authz_method"] = GetAuthzMethod(*method_).get_value_or("AuthorizeAllowAll");
+ }
+
+ // Strips the package from method arguments if they are in the same package as
+ // the service, otherwise leaves them so that we can have fully qualified
+ // namespaces for method arguments.
+ static std::string StripNamespaceIfPossible(const std::string& service_full_name,
+ const std::string& arg_full_name) {
+ StringPiece service_package(service_full_name);
+ if (!service_package.contains(".")) {
+ return arg_full_name;
+ }
+ // remove the service name so that we are left with only the package, including
+ // the last '.' so that we account for different packages with the same prefix.
+ service_package.remove_suffix(service_package.length() -
+ service_package.find_last_of(".") - 1);
+
+ StringPiece argfqn(arg_full_name);
+ if (argfqn.starts_with(service_package)) {
+ argfqn.remove_prefix(argfqn.find_last_of(".") + 1);
+ }
+ return argfqn.ToString();
+ }
+
+ static std::string ReplaceNamespaceDelimiters(const std::string& arg_full_name) {
+ return JoinStrings(strings::Split(arg_full_name, "."), "::");
+ }
+
+ private:
+ const MethodDescriptor *method_;
+};
+
+class ServiceSubstitutions : public Substituter {
+ public:
+ explicit ServiceSubstitutions(const ServiceDescriptor *service)
+ : service_(service)
+ {}
+
+ virtual void InitSubstitutionMap(map<string, string> *map) const OVERRIDE {
+ (*map)["service_name"] = service_->name();
+ (*map)["full_service_name"] = service_->full_name();
+ (*map)["service_method_count"] = SimpleItoa(service_->method_count());
+
+ // TODO: upgrade to protobuf 2.5.x and attach service comments
+ // to the generated service classes using the SourceLocation API.
+ }
+
+ private:
+ const ServiceDescriptor *service_;
+};
+
+
+class SubstitutionContext {
+ public:
+ // Takes ownership of the substituter
+ void Push(const Substituter *sub) {
+ subs_.push_back(shared_ptr<const Substituter>(sub));
+ }
+
+ void PushMethod(const MethodDescriptor *method) {
+ Push(new MethodSubstitutions(method));
+ }
+
+ void PushService(const ServiceDescriptor *service) {
+ Push(new ServiceSubstitutions(service));
+ }
+
+ void Pop() {
+ CHECK(!subs_.empty());
+ subs_.pop_back();
+ }
+
+ void InitSubstitutionMap(map<string, string> *subs) const {
+ for (const shared_ptr<const Substituter> &sub : subs_) {
+ sub->InitSubstitutionMap(subs);
+ }
+ }
+
+ private:
+ vector<shared_ptr<const Substituter> > subs_;
+};
+
+
+
+class CodeGenerator : public ::google::protobuf::compiler::CodeGenerator {
+ public:
+ CodeGenerator() { }
+
+ ~CodeGenerator() { }
+
+ bool Generate(const google::protobuf::FileDescriptor *file,
+ const std::string &/* parameter */,
+ google::protobuf::compiler::GeneratorContext *gen_context,
+ std::string *error) const OVERRIDE {
+ auto name_info = new FileSubstitutions();
+ Status ret = name_info->Init(file);
+ if (!ret.ok()) {
+ *error = "name_info.Init failed: " + ret.ToString();
+ return false;
+ }
+
+ SubstitutionContext subs;
+ subs.Push(name_info);
+
+ gscoped_ptr<google::protobuf::io::ZeroCopyOutputStream> ih_output(
+ gen_context->Open(name_info->service_header()));
+ Printer ih_printer(ih_output.get(), '$');
+ GenerateServiceIfHeader(&ih_printer, &subs, file);
+
+ gscoped_ptr<google::protobuf::io::ZeroCopyOutputStream> i_output(
+ gen_context->Open(name_info->service()));
+ Printer i_printer(i_output.get(), '$');
+ GenerateServiceIf(&i_printer, &subs, file);
+
+ gscoped_ptr<google::protobuf::io::ZeroCopyOutputStream> ph_output(
+ gen_context->Open(name_info->proxy_header()));
+ Printer ph_printer(ph_output.get(), '$');
+ GenerateProxyHeader(&ph_printer, &subs, file);
+
+ gscoped_ptr<google::protobuf::io::ZeroCopyOutputStream> p_output(
+ gen_context->Open(name_info->proxy()));
+ Printer p_printer(p_output.get(), '$');
+ GenerateProxy(&p_printer, &subs, file);
+
+ return true;
+ }
+
+ private:
+ void Print(Printer *printer,
+ const SubstitutionContext &sub,
+ const char *text) const {
+ map<string, string> subs;
+ sub.InitSubstitutionMap(&subs);
+ printer->Print(subs, text);
+ }
+
+ void GenerateServiceIfHeader(Printer *printer,
+ SubstitutionContext *subs,
+ const FileDescriptor *file) const {
+ Print(printer, *subs,
+ "// THIS FILE IS AUTOGENERATED FROM $path$\n"
+ "\n"
+ "#ifndef KUDU_RPC_$upper_case$_SERVICE_IF_DOT_H\n"
+ "#define KUDU_RPC_$upper_case$_SERVICE_IF_DOT_H\n"
+ "\n"
+ "#include <string>\n"
+ "\n"
+ "#include \"kudu/gutil/ref_counted.h\"\n"
+ "#include \"kudu/rpc/service_if.h\"\n"
+ "\n"
+ "namespace google {\n"
+ "namespace protobuf {\n"
+ "class Message;\n"
+ "} // namespace protobuf\n"
+ "} // namespace google\n"
+ "\n"
+ "namespace kudu {\n"
+ "class MetricEntity;\n"
+ "namespace rpc {\n"
+ "class ResultTracker;\n"
+ "class RpcContext;\n"
+ "} // namespace rpc\n"
+ "} // namespace kudu\n"
+ "\n"
+ "$open_namespace$"
+ "\n");
+
+ for (int service_idx = 0; service_idx < file->service_count();
+ ++service_idx) {
+ const ServiceDescriptor *service = file->service(service_idx);
+ subs->PushService(service);
+
+ Print(printer, *subs,
+ "class $service_name$If : public ::kudu::rpc::GeneratedServiceIf {\n"
+ " public:\n"
+ " explicit $service_name$If(const scoped_refptr<::kudu::MetricEntity>& entity,"
+ " const scoped_refptr<::kudu::rpc::ResultTracker>& result_tracker);\n"
+ " virtual ~$service_name$If();\n"
+ " std::string service_name() const override;\n"
+ " static std::string static_service_name();\n"
+ "\n"
+ );
+
+ set<string> authz_methods;
+ for (int method_idx = 0; method_idx < service->method_count();
+ ++method_idx) {
+ const MethodDescriptor *method = service->method(method_idx);
+ subs->PushMethod(method);
+
+ Print(printer, *subs,
+ " virtual void $rpc_name$(const class $request$ *req,\n"
+ " class $response$ *resp, ::kudu::rpc::RpcContext *context) = 0;\n"
+ );
+ subs->Pop();
+ if (auto m = GetAuthzMethod(*method)) {
+ authz_methods.insert(m.get());
+ }
+ }
+
+ if (!authz_methods.empty()) {
+ printer->Print(
+ "\n\n"
+ " // Authorization methods\n"
+ " // ---------------------\n\n");
+ }
+ for (const string& m : authz_methods) {
+ printer->Print({ {"m", m} },
+ " virtual bool $m$(const google::protobuf::Message* req,\n"
+ " google::protobuf::Message* resp, ::kudu::rpc::RpcContext *context) = 0;\n");
+ }
+
+ Print(printer, *subs,
+ "\n"
+ "};\n"
+ );
+
+ subs->Pop(); // Service
+ }
+
+ Print(printer, *subs,
+ "\n"
+ "$close_namespace$\n"
+ "#endif\n");
+ }
+
+ void GenerateServiceIf(Printer *printer,
+ SubstitutionContext *subs,
+ const FileDescriptor *file) const {
+ Print(printer, *subs,
+ "// THIS FILE IS AUTOGENERATED FROM $path$\n"
+ "\n"
+ "#include <functional>\n"
+ "#include <memory>\n"
+ "#include <unordered_map>\n"
+ "#include <utility>\n"
+ "\n"
+ "#include <google/protobuf/message.h>\n"
+ "\n"
+ "#include \"$path_no_extension$.pb.h\"\n"
+ "#include \"$path_no_extension$.service.h\"\n"
+ "\n"
+ "#include \"kudu/rpc/result_tracker.h\"\n"
+ "#include \"kudu/rpc/service_if.h\"\n"
+ "#include \"kudu/util/metrics.h\"\n"
+ "\n");
+
+ // Define metric prototypes for each method in the service.
+ for (int service_idx = 0; service_idx < file->service_count();
+ ++service_idx) {
+ const ServiceDescriptor *service = file->service(service_idx);
+ subs->PushService(service);
+
+ for (int method_idx = 0; method_idx < service->method_count();
+ ++method_idx) {
+ const MethodDescriptor *method = service->method(method_idx);
+ subs->PushMethod(method);
+ Print(printer, *subs,
+ "METRIC_DEFINE_histogram(server, handler_latency_$rpc_full_name_plainchars$,\n"
+ " \"$rpc_full_name$ RPC Time\",\n"
+ " kudu::MetricUnit::kMicroseconds,\n"
+ " \"Microseconds spent handling $rpc_full_name$() RPC requests\",\n"
+ " 60000000LU, 2);\n"
+ "\n");
+ subs->Pop();
+ }
+
+ subs->Pop();
+ }
+
+ Print(printer, *subs,
+ "using google::protobuf::Message;\n"
+ "using kudu::MetricEntity;\n"
+ "using kudu::rpc::ResultTracker;\n"
+ "using kudu::rpc::RpcContext;\n"
+ "using kudu::rpc::RpcMethodInfo;\n"
+ "using std::unique_ptr;\n"
+ "\n"
+ "$open_namespace$"
+ "\n");
+
+ for (int service_idx = 0; service_idx < file->service_count();
+ ++service_idx) {
+ const ServiceDescriptor *service = file->service(service_idx);
+ subs->PushService(service);
+
+ Print(printer, *subs,
+ "$service_name$If::$service_name$If(const scoped_refptr<MetricEntity>& entity,"
+ " const scoped_refptr<ResultTracker>& result_tracker) {\n"
+ "result_tracker_ = result_tracker;\n"
+ );
+ for (int method_idx = 0; method_idx < service->method_count();
+ ++method_idx) {
+ const MethodDescriptor *method = service->method(method_idx);
+ subs->PushMethod(method);
+
+ Print(printer, *subs,
+ " {\n"
+ " scoped_refptr<RpcMethodInfo> mi(new RpcMethodInfo());\n"
+ " mi->req_prototype.reset(new $request$());\n"
+ " mi->resp_prototype.reset(new $response$());\n"
+ " mi->authz_method = [this](const Message* req, Message* resp,\n"
+ " RpcContext* ctx) {\n"
+ " return this->$authz_method$(static_cast<const $request$*>(req),\n"
+ " static_cast<$response$*>(resp),\n"
+ " ctx);\n"
+ " };\n"
+ " mi->track_result = $track_result$;\n"
+ " mi->handler_latency_histogram =\n"
+ " METRIC_handler_latency_$rpc_full_name_plainchars$.Instantiate(entity);\n"
+ " mi->func = [this](const Message* req, Message* resp, RpcContext* ctx) {\n"
+ " this->$rpc_name$(static_cast<const $request$*>(req),\n"
+ " static_cast<$response$*>(resp),\n"
+ " ctx);\n"
+ " };\n"
+ " methods_by_name_[\"$rpc_name$\"] = std::move(mi);\n"
+ " }\n");
+ subs->Pop();
+ }
+
+ Print(printer, *subs,
+ "}\n"
+ "\n"
+ "$service_name$If::~$service_name$If() {\n"
+ "}\n"
+ "\n"
+ "std::string $service_name$If::service_name() const {\n"
+ " return \"$full_service_name$\";\n"
+ "}\n"
+ "std::string $service_name$If::static_service_name() {\n"
+ " return \"$full_service_name$\";\n"
+ "}\n"
+ "\n"
+ );
+
+ subs->Pop();
+ }
+
+ Print(printer, *subs,
+ "$close_namespace$"
+ );
+ }
+
+ void GenerateProxyHeader(Printer *printer,
+ SubstitutionContext *subs,
+ const FileDescriptor *file) const {
+ Print(printer, *subs,
+ "// THIS FILE IS AUTOGENERATED FROM $path$\n"
+ "\n"
+ "#ifndef KUDU_RPC_$upper_case$_PROXY_DOT_H\n"
+ "#define KUDU_RPC_$upper_case$_PROXY_DOT_H\n"
+ "\n"
+ "#include <memory>\n"
+ "#include <string>\n"
+ "\n"
+ "#include \"kudu/rpc/proxy.h\"\n"
+ "#include \"kudu/rpc/response_callback.h\"\n"
+ "#include \"kudu/util/status.h\"\n"
+ "\n"
+ "namespace kudu { class Sockaddr; }\n"
+ "namespace kudu {\n"
+ "namespace rpc {\n"
+ "class Messenger;\n"
+ "class RpcController;\n"
+ "} // namespace rpc\n"
+ "} // namespace kudu\n"
+ "\n"
+ "$open_namespace$"
+ "\n");
+
+ for (int service_idx = 0; service_idx < file->service_count();
+ ++service_idx) {
+ const ServiceDescriptor *service = file->service(service_idx);
+ subs->PushService(service);
+
+ Print(printer, *subs,
+ "class $service_name$Proxy : public ::kudu::rpc::Proxy {\n"
+ " public:\n"
+ " $service_name$Proxy(std::shared_ptr<::kudu::rpc::Messenger>\n"
+ " messenger, const ::kudu::Sockaddr &sockaddr,"
+ " std::string hostname);\n"
+ " ~$service_name$Proxy();\n"
+ "\n"
+ );
+
+ for (int method_idx = 0; method_idx < service->method_count();
+ ++method_idx) {
+ const MethodDescriptor *method = service->method(method_idx);
+ subs->PushMethod(method);
+
+ Print(printer, *subs,
+ "\n"
+ " ::kudu::Status $rpc_name$(const class $request$ &req,\n"
+ " class $response$ *resp,\n"
+ " ::kudu::rpc::RpcController *controller);\n"
+ " void $rpc_name$Async(const class $request$ &req,\n"
+ " class $response$ *response,\n"
+ " ::kudu::rpc::RpcController *controller,\n"
+ " const ::kudu::rpc::ResponseCallback &callback);\n"
+ );
+ subs->Pop();
+ }
+ Print(printer, *subs,
+ "};\n");
+ subs->Pop();
+ }
+ Print(printer, *subs,
+ "\n"
+ "$close_namespace$"
+ "\n"
+ "#endif\n"
+ );
+ }
+
+ void GenerateProxy(Printer *printer,
+ SubstitutionContext *subs,
+ const FileDescriptor *file) const {
+ Print(printer, *subs,
+ "// THIS FILE IS AUTOGENERATED FROM $path$\n"
+ "\n"
+ "#include <string>\n"
+ "#include <utility>\n"
+ "\n"
+ "#include \"$path_no_extension$.pb.h\"\n"
+ "#include \"$path_no_extension$.proxy.h\"\n"
+ "\n"
+ "namespace kudu {\n"
+ "namespace rpc {\n"
+ "class Messenger;\n"
+ "class RpcController;\n"
+ "} // namespace rpc\n"
+ "} // namespace kudu\n"
+ "\n"
+ "$open_namespace$"
+ "\n");
+
+ for (int service_idx = 0; service_idx < file->service_count();
+ ++service_idx) {
+ const ServiceDescriptor *service = file->service(service_idx);
+ subs->PushService(service);
+ Print(printer, *subs,
+ "$service_name$Proxy::$service_name$Proxy(\n"
+ " std::shared_ptr< ::kudu::rpc::Messenger> messenger,\n"
+ " const ::kudu::Sockaddr &remote, std::string hostname)\n"
+ " : Proxy(std::move(messenger), remote, std::move(hostname), \"$full_service_name$\") {\n"
+ "}\n"
+ "\n"
+ "$service_name$Proxy::~$service_name$Proxy() {\n"
+ "}\n"
+ "\n"
+ "\n");
+ for (int method_idx = 0; method_idx < service->method_count();
+ ++method_idx) {
+ const MethodDescriptor *method = service->method(method_idx);
+ subs->PushMethod(method);
+ Print(printer, *subs,
+ "::kudu::Status $service_name$Proxy::$rpc_name$(const $request$ &req, $response$ *resp,\n"
+ " ::kudu::rpc::RpcController *controller) {\n"
+ " return SyncRequest(\"$rpc_name$\", req, resp, controller);\n"
+ "}\n"
+ "\n"
+ "void $service_name$Proxy::$rpc_name$Async(const $request$ &req,\n"
+ " $response$ *resp, ::kudu::rpc::RpcController *controller,\n"
+ " const ::kudu::rpc::ResponseCallback &callback) {\n"
+ " AsyncRequest(\"$rpc_name$\", req, resp, controller, callback);\n"
+ "}\n"
+ "\n");
+ subs->Pop();
+ }
+
+ subs->Pop();
+ }
+ Print(printer, *subs,
+ "$close_namespace$");
+ }
+};
+} // namespace rpc
+} // namespace kudu
+
+int main(int argc, char *argv[]) {
+ kudu::rpc::CodeGenerator generator;
+ return google::protobuf::compiler::PluginMain(argc, argv, &generator);
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/fcf190c4/be/src/kudu/rpc/proxy.cc
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/proxy.cc b/be/src/kudu/rpc/proxy.cc
new file mode 100644
index 0000000..54b8085
--- /dev/null
+++ b/be/src/kudu/rpc/proxy.cc
@@ -0,0 +1,116 @@
+// 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/rpc/proxy.h"
+
+#include <iostream>
+#include <memory>
+#include <utility>
+
+#include <boost/bind.hpp> // IWYU pragma: keep
+#include <boost/core/ref.hpp>
+#include <glog/logging.h>
+
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/rpc/outbound_call.h"
+#include "kudu/rpc/messenger.h"
+#include "kudu/rpc/remote_method.h"
+#include "kudu/rpc/response_callback.h"
+#include "kudu/rpc/rpc_controller.h"
+#include "kudu/rpc/user_credentials.h"
+#include "kudu/util/net/sockaddr.h"
+#include "kudu/util/countdown_latch.h"
+#include "kudu/util/status.h"
+#include "kudu/util/user.h"
+
+using google::protobuf::Message;
+using std::string;
+using std::shared_ptr;
+
+namespace kudu {
+namespace rpc {
+
+Proxy::Proxy(std::shared_ptr<Messenger> messenger,
+ const Sockaddr& remote,
+ string hostname,
+ string service_name)
+ : service_name_(std::move(service_name)),
+ messenger_(std::move(messenger)),
+ is_started_(false) {
+ CHECK(messenger_ != nullptr);
+ DCHECK(!service_name_.empty()) << "Proxy service name must not be blank";
+
+ // By default, we set the real user to the currently logged-in user.
+ // Effective user and password remain blank.
+ string real_user;
+ Status s = GetLoggedInUser(&real_user);
+ if (!s.ok()) {
+ LOG(WARNING) << "Proxy for " << service_name_ << ": Unable to get logged-in user name: "
+ << s.ToString() << " before connecting to remote: " << remote.ToString();
+ }
+
+ UserCredentials creds;
+ creds.set_real_user(std::move(real_user));
+ conn_id_ = ConnectionId(remote, std::move(hostname), std::move(creds));
+}
+
+Proxy::~Proxy() {
+}
+
+void Proxy::AsyncRequest(const string& method,
+ const google::protobuf::Message& req,
+ google::protobuf::Message* response,
+ RpcController* controller,
+ const ResponseCallback& callback) const {
+ CHECK(!controller->call_) << "Controller should be reset";
+ base::subtle::NoBarrier_Store(&is_started_, true);
+ RemoteMethod remote_method(service_name_, method);
+ controller->call_.reset(
+ new OutboundCall(conn_id_, remote_method, response, controller, callback));
+ controller->SetRequestParam(req);
+ controller->SetMessenger(messenger_.get());
+
+ // If this fails to queue, the callback will get called immediately
+ // and the controller will be in an ERROR state.
+ messenger_->QueueOutboundCall(controller->call_);
+}
+
+
+Status Proxy::SyncRequest(const string& method,
+ const google::protobuf::Message& req,
+ google::protobuf::Message* resp,
+ RpcController* controller) const {
+ CountDownLatch latch(1);
+ AsyncRequest(method, req, DCHECK_NOTNULL(resp), controller,
+ boost::bind(&CountDownLatch::CountDown, boost::ref(latch)));
+
+ latch.Wait();
+ return controller->status();
+}
+
+void Proxy::set_user_credentials(const UserCredentials& user_credentials) {
+ CHECK(base::subtle::NoBarrier_Load(&is_started_) == false)
+ << "It is illegal to call set_user_credentials() after request processing has started";
+ conn_id_.set_user_credentials(user_credentials);
+}
+
+std::string Proxy::ToString() const {
+ return strings::Substitute("$0@$1", service_name_, conn_id_.ToString());
+}
+
+} // namespace rpc
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/impala/blob/fcf190c4/be/src/kudu/rpc/proxy.h
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/proxy.h b/be/src/kudu/rpc/proxy.h
new file mode 100644
index 0000000..92b3079
--- /dev/null
+++ b/be/src/kudu/rpc/proxy.h
@@ -0,0 +1,126 @@
+// 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.
+
+#ifndef KUDU_RPC_PROXY_H
+#define KUDU_RPC_PROXY_H
+
+#include <memory>
+#include <string>
+
+#include "kudu/gutil/atomicops.h"
+#include "kudu/gutil/macros.h"
+#include "kudu/rpc/connection_id.h"
+#include "kudu/rpc/response_callback.h"
+#include "kudu/util/status.h"
+
+namespace google {
+namespace protobuf {
+class Message;
+} // namespace protobuf
+} // namespace google
+
+namespace kudu {
+
+class Sockaddr;
+
+namespace rpc {
+
+class Messenger;
+class RpcController;
+class UserCredentials;
+
+// Interface to send calls to a remote service.
+//
+// Proxy objects do not map one-to-one with TCP connections. The underlying TCP
+// connection is not established until the first call, and may be torn down and
+// re-established as necessary by the messenger. Additionally, the messenger is
+// likely to multiplex many Proxy objects on the same connection.
+//
+// Proxy objects are thread-safe after initialization only.
+// Setters on the Proxy are not thread-safe, and calling a setter after any RPC
+// request has started will cause a fatal error.
+//
+// After initialization, multiple threads may make calls using the same proxy object.
+class Proxy {
+ public:
+ Proxy(std::shared_ptr<Messenger> messenger,
+ const Sockaddr& remote,
+ std::string hostname,
+ std::string service_name);
+
+ ~Proxy();
+
+ // Call a remote method asynchronously.
+ //
+ // Typically, users will not call this directly, but rather through
+ // a generated Proxy subclass.
+ //
+ // method: the method name to invoke on the remote server.
+ //
+ // req: the request protobuf. This will be serialized immediately,
+ // so the caller may free or otherwise mutate 'req' safely.
+ //
+ // resp: the response protobuf. This protobuf will be mutated upon
+ // completion of the call. The RPC system does not take ownership
+ // of this storage.
+ //
+ // NOTE: 'req' and 'resp' should be the appropriate protocol buffer implementation
+ // class corresponding to the parameter and result types of the service method
+ // defined in the service's '.proto' file.
+ //
+ // controller: the RpcController to associate with this call. Each call
+ // must use a unique controller object. Does not take ownership.
+ //
+ // callback: the callback to invoke upon call completion. This callback may
+ // be invoked before AsyncRequest() itself returns, or any time
+ // thereafter. It may be invoked either on the caller's thread
+ // or by an RPC IO thread, and thus should take care to not
+ // block or perform any heavy CPU work.
+ void AsyncRequest(const std::string& method,
+ const google::protobuf::Message& req,
+ google::protobuf::Message* resp,
+ RpcController* controller,
+ const ResponseCallback& callback) const;
+
+ // The same as AsyncRequest(), except that the call blocks until the call
+ // finishes. If the call fails, returns a non-OK result.
+ Status SyncRequest(const std::string& method,
+ const google::protobuf::Message& req,
+ google::protobuf::Message* resp,
+ RpcController* controller) const;
+
+ // Set the user credentials which should be used to log in.
+ void set_user_credentials(const UserCredentials& user_credentials);
+
+ // Get the user credentials which should be used to log in.
+ const UserCredentials& user_credentials() const { return conn_id_.user_credentials(); }
+
+ std::string ToString() const;
+
+ private:
+ const std::string service_name_;
+ std::shared_ptr<Messenger> messenger_;
+ ConnectionId conn_id_;
+ mutable Atomic32 is_started_;
+
+ DISALLOW_COPY_AND_ASSIGN(Proxy);
+};
+
+} // namespace rpc
+} // namespace kudu
+
+#endif
http://git-wip-us.apache.org/repos/asf/impala/blob/fcf190c4/be/src/kudu/rpc/reactor-test.cc
----------------------------------------------------------------------
diff --git a/be/src/kudu/rpc/reactor-test.cc b/be/src/kudu/rpc/reactor-test.cc
new file mode 100644
index 0000000..2de5f58
--- /dev/null
+++ b/be/src/kudu/rpc/reactor-test.cc
@@ -0,0 +1,112 @@
+// 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 <memory>
+
+#include <boost/bind.hpp> // IWYU pragma: keep
+#include <boost/function.hpp>
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+
+#include "kudu/rpc/messenger.h"
+#include "kudu/rpc/rpc-test-base.h"
+#include "kudu/util/countdown_latch.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/status.h"
+#include "kudu/util/test_macros.h"
+#include "kudu/util/thread.h"
+
+using std::shared_ptr;
+
+namespace kudu {
+namespace rpc {
+
+class ReactorTest : public RpcTestBase {
+ public:
+ ReactorTest()
+ : latch_(1) {
+ }
+
+ void SetUp() override {
+ RpcTestBase::SetUp();
+ ASSERT_OK(CreateMessenger("my_messenger", &messenger_, 4));
+ }
+
+ void ScheduledTask(const Status& status, const Status& expected_status) {
+ CHECK_EQ(expected_status.CodeAsString(), status.CodeAsString());
+ latch_.CountDown();
+ }
+
+ void ScheduledTaskCheckThread(const Status& status, const Thread* thread) {
+ CHECK_OK(status);
+ CHECK_EQ(thread, Thread::current_thread());
+ latch_.CountDown();
+ }
+
+ void ScheduledTaskScheduleAgain(const Status& status) {
+ messenger_->ScheduleOnReactor(
+ boost::bind(&ReactorTest::ScheduledTaskCheckThread, this, _1,
+ Thread::current_thread()),
+ MonoDelta::FromMilliseconds(0));
+ latch_.CountDown();
+ }
+
+ protected:
+ shared_ptr<Messenger> messenger_;
+ CountDownLatch latch_;
+};
+
+TEST_F(ReactorTest, TestFunctionIsCalled) {
+ messenger_->ScheduleOnReactor(
+ boost::bind(&ReactorTest::ScheduledTask, this, _1, Status::OK()),
+ MonoDelta::FromSeconds(0));
+ latch_.Wait();
+}
+
+TEST_F(ReactorTest, TestFunctionIsCalledAtTheRightTime) {
+ MonoTime before = MonoTime::Now();
+ messenger_->ScheduleOnReactor(
+ boost::bind(&ReactorTest::ScheduledTask, this, _1, Status::OK()),
+ MonoDelta::FromMilliseconds(100));
+ latch_.Wait();
+ MonoTime after = MonoTime::Now();
+ MonoDelta delta = after - before;
+ CHECK_GE(delta.ToMilliseconds(), 100);
+}
+
+TEST_F(ReactorTest, TestFunctionIsCalledIfReactorShutdown) {
+ messenger_->ScheduleOnReactor(
+ boost::bind(&ReactorTest::ScheduledTask, this, _1,
+ Status::Aborted("doesn't matter")),
+ MonoDelta::FromSeconds(60));
+ messenger_->Shutdown();
+ latch_.Wait();
+}
+
+TEST_F(ReactorTest, TestReschedulesOnSameReactorThread) {
+ // Our scheduled task will schedule yet another task.
+ latch_.Reset(2);
+
+ messenger_->ScheduleOnReactor(
+ boost::bind(&ReactorTest::ScheduledTaskScheduleAgain, this, _1),
+ MonoDelta::FromSeconds(0));
+ latch_.Wait();
+ latch_.Wait();
+}
+
+} // namespace rpc
+} // namespace kudu