You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by to...@apache.org on 2016/05/25 05:36:17 UTC
[1/3] incubator-kudu git commit: Add request id and sequence number
to the rpc request header
Repository: incubator-kudu
Updated Branches:
refs/heads/master a25401254 -> 57c4b47f8
Add request id and sequence number to the rpc request header
This adds a new (optional) field to RpcHeader: RequestIdPB.
This new field contains the information necessary to implement the "replay cache"
component, including the unique id of the client and the rpc's sequence number.
RequestIdPB also includes two additional "helper" fields, one to keep track of
number of attempts of a particular rpc, so that we can skip the replay cache
altogether in certain cases and one to let the server know that it can garbage
collect completion records.
Change-Id: I30b07f070f8e80aaad982ccfe71ffb7e22a6a703
Reviewed-on: http://gerrit.cloudera.org:8080/3079
Tested-by: Kudu Jenkins
Reviewed-by: David Ribeiro Alves <da...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/ab500fc2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/ab500fc2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/ab500fc2
Branch: refs/heads/master
Commit: ab500fc223866ad49a9f1ec50e61482a36896312
Parents: a254012
Author: David Alves <da...@cloudera.com>
Authored: Sat May 14 14:23:10 2016 -0700
Committer: David Ribeiro Alves <da...@cloudera.com>
Committed: Wed May 25 03:07:41 2016 +0000
----------------------------------------------------------------------
src/kudu/rpc/rpc_header.proto | 22 ++++++++++++++++++++++
1 file changed, 22 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ab500fc2/src/kudu/rpc/rpc_header.proto
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc_header.proto b/src/kudu/rpc/rpc_header.proto
index 5c22a37..685460e 100644
--- a/src/kudu/rpc/rpc_header.proto
+++ b/src/kudu/rpc/rpc_header.proto
@@ -110,6 +110,23 @@ message RemoteMethodPB {
required string method_name = 2;
};
+// The Id of a retriable RPC, whose results should be tracked on the server (see result_tracker.h).
+// This also includes some information that is useful for execution/garbage collection.
+message RequestIdPB {
+ // The (globally unique) id of the client performing this RPC.
+ required string client_id = 1;
+
+ // The (per-client unique) sequence number of this RPC.
+ required int64 seq_no = 2;
+
+ // The sequence number of the first RPC that has not been marked as completed by the client.
+ // Unset if there isn't an incomplete RPC.
+ optional int64 first_incomplete_seq_no = 3;
+
+ // The number of times this RPC has been tried, only set after the first attempt.
+ optional int32 attempt_no = 4;
+}
+
// The header for the RPC request frame.
message RequestHeader {
// A sequence number that is sent back in the Response. Hadoop specifies a uint32 and
@@ -142,6 +159,11 @@ message RequestHeader {
// NOTE: the server will only interpret this field if it supports the
// APPLICATION_FEATURE_FLAGS flag.
repeated uint32 required_feature_flags = 11;
+
+ // The unique id of this request, if it's retriable and if the results are to be tracked.
+ // Optional for requests that are naturally idempotent or to maintain compatibility with
+ // older clients for requests that are not.
+ optional RequestIdPB request_id = 15;
}
message ResponseHeader {
[3/3] incubator-kudu git commit: Move the PbTracer in rpc_context.cc
to pb_util
Posted by to...@apache.org.
Move the PbTracer in rpc_context.cc to pb_util
We'll need to use this outside of RpcContext so this moves
it to a place where it makes more sense and can be used
elsewhere.
Change-Id: Ib601deea9fe0dcc6e2428416a6993c7567ff7335
Reviewed-on: http://gerrit.cloudera.org:8080/3194
Tested-by: David Ribeiro Alves <da...@cloudera.com>
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/57c4b47f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/57c4b47f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/57c4b47f
Branch: refs/heads/master
Commit: 57c4b47f8b09fee4dac682b70bd923ed6b4960f8
Parents: 9d5ce00
Author: David Alves <da...@cloudera.com>
Authored: Tue May 24 10:15:35 2016 -0700
Committer: David Ribeiro Alves <da...@cloudera.com>
Committed: Wed May 25 03:08:52 2016 +0000
----------------------------------------------------------------------
src/kudu/rpc/rpc_context.cc | 39 +++------------------------------------
src/kudu/util/pb_util.cc | 18 ++++++++++++++++++
src/kudu/util/pb_util.h | 33 +++++++++++++++++++++++++++++++++
3 files changed, 54 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/57c4b47f/src/kudu/rpc/rpc_context.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc_context.cc b/src/kudu/rpc/rpc_context.cc
index 6019607..083b969 100644
--- a/src/kudu/rpc/rpc_context.cc
+++ b/src/kudu/rpc/rpc_context.cc
@@ -27,8 +27,6 @@
#include "kudu/util/hdr_histogram.h"
#include "kudu/util/metrics.h"
#include "kudu/util/trace.h"
-#include "kudu/util/debug/trace_event.h"
-#include "kudu/util/jsonwriter.h"
#include "kudu/util/pb_util.h"
using google::protobuf::Message;
@@ -36,37 +34,6 @@ using google::protobuf::Message;
namespace kudu {
namespace rpc {
-namespace {
-
-// Wrapper for a protobuf message which lazily converts to JSON when
-// the trace buffer is dumped. This pushes the work of stringification
-// to the trace dumping process.
-class PbTracer : public debug::ConvertableToTraceFormat {
- public:
- enum {
- kMaxFieldLengthToTrace = 100
- };
-
- explicit PbTracer(const Message& msg) : msg_(msg.New()) {
- msg_->CopyFrom(msg);
- }
-
- virtual void AppendAsTraceFormat(std::string* out) const OVERRIDE {
- pb_util::TruncateFields(msg_.get(), kMaxFieldLengthToTrace);
- std::stringstream ss;
- JsonWriter jw(&ss, JsonWriter::COMPACT);
- jw.Protobuf(*msg_);
- out->append(ss.str());
- }
- private:
- const gscoped_ptr<Message> msg_;
-};
-
-scoped_refptr<debug::ConvertableToTraceFormat> TracePb(const Message& msg) {
- return make_scoped_refptr(new PbTracer(msg));
-}
-} // anonymous namespace
-
RpcContext::RpcContext(InboundCall *call,
const google::protobuf::Message *request_pb,
google::protobuf::Message *response_pb)
@@ -77,7 +44,7 @@ RpcContext::RpcContext(InboundCall *call,
<< call_->ToString() << ":" << std::endl << request_pb_->DebugString();
TRACE_EVENT_ASYNC_BEGIN2("rpc_call", "RPC", this,
"call", call_->ToString(),
- "request", TracePb(*request_pb_));
+ "request", pb_util::PbTracer::TracePb(*request_pb_));
}
RpcContext::~RpcContext() {
@@ -87,7 +54,7 @@ void RpcContext::RespondSuccess() {
VLOG(4) << call_->remote_method().service_name() << ": Sending RPC success response for "
<< call_->ToString() << ":" << std::endl << response_pb_->DebugString();
TRACE_EVENT_ASYNC_END2("rpc_call", "RPC", this,
- "response", TracePb(*response_pb_),
+ "response", pb_util::PbTracer::TracePb(*response_pb_),
"trace", trace()->DumpToString());
call_->RespondSuccess(*response_pb_);
delete this;
@@ -122,7 +89,7 @@ void RpcContext::RespondApplicationError(int error_ext_id, const std::string& me
VLOG(4) << call_->remote_method().service_name() << ": Sending application error response for "
<< call_->ToString() << ":" << std::endl << err.DebugString();
TRACE_EVENT_ASYNC_END2("rpc_call", "RPC", this,
- "response", TracePb(app_error_pb),
+ "response", pb_util::PbTracer::TracePb(app_error_pb),
"trace", trace()->DumpToString());
}
call_->RespondApplicationError(error_ext_id, message, app_error_pb);
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/57c4b47f/src/kudu/util/pb_util.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/pb_util.cc b/src/kudu/util/pb_util.cc
index b506666..0ee7794 100644
--- a/src/kudu/util/pb_util.cc
+++ b/src/kudu/util/pb_util.cc
@@ -55,6 +55,7 @@
#include "kudu/util/debug/trace_event.h"
#include "kudu/util/env.h"
#include "kudu/util/env_util.h"
+#include "kudu/util/jsonwriter.h"
#include "kudu/util/path_util.h"
#include "kudu/util/pb_util-internal.h"
#include "kudu/util/pb_util.pb.h"
@@ -874,5 +875,22 @@ Status WritePBContainerToPath(Env* env, const std::string& path,
return Status::OK();
}
+
+scoped_refptr<debug::ConvertableToTraceFormat> PbTracer::TracePb(const Message& msg) {
+ return make_scoped_refptr(new PbTracer(msg));
+}
+
+PbTracer::PbTracer(const Message& msg) : msg_(msg.New()) {
+ msg_->CopyFrom(msg);
+}
+
+void PbTracer::AppendAsTraceFormat(std::string* out) const {
+ pb_util::TruncateFields(msg_.get(), kMaxFieldLengthToTrace);
+ std::stringstream ss;
+ JsonWriter jw(&ss, JsonWriter::COMPACT);
+ jw.Protobuf(*msg_);
+ out->append(ss.str());
+}
+
} // namespace pb_util
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/57c4b47f/src/kudu/util/pb_util.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/pb_util.h b/src/kudu/util/pb_util.h
index 0410f13..64b8a86 100644
--- a/src/kudu/util/pb_util.h
+++ b/src/kudu/util/pb_util.h
@@ -26,6 +26,7 @@
#include <gtest/gtest_prod.h>
#include "kudu/gutil/gscoped_ptr.h"
+#include "kudu/util/debug/trace_event.h"
#include "kudu/util/faststring.h"
namespace google {
@@ -424,6 +425,38 @@ Status WritePBContainerToPath(Env* env, const std::string& path,
CreateMode create,
SyncMode sync);
+// Wrapper for a protobuf message which lazily converts to JSON when
+// the trace buffer is dumped.
+//
+// When tracing, an instance of this class can be associated with
+// a given trace, instead of a stringified PB, thus avoiding doing
+// stringification inline and moving that work to the tracing process.
+//
+// Example usage:
+// TRACE_EVENT_ASYNC_END2("rpc_call", "RPC", this,
+// "response", pb_util::PbTracer::TracePb(*response_pb_),
+// ...);
+//
+class PbTracer : public debug::ConvertableToTraceFormat {
+ public:
+ enum {
+ kMaxFieldLengthToTrace = 100
+ };
+
+ // Static helper to be called when adding a stringified PB to a trace.
+ // This does not actually stringify 'msg', that will be done later
+ // when/if AppendAsTraceFormat() is called on the returned object.
+ static scoped_refptr<debug::ConvertableToTraceFormat> TracePb(
+ const google::protobuf::Message& msg);
+
+ explicit PbTracer(const google::protobuf::Message& msg);
+
+ // Actually stringifies the PB and appends the string to 'out'.
+ void AppendAsTraceFormat(std::string* out) const override;
+ private:
+ const gscoped_ptr<google::protobuf::Message> msg_;
+};
+
} // namespace pb_util
} // namespace kudu
#endif
[2/3] incubator-kudu git commit: Add a request tracker to track
client rpc sequence numbers
Posted by to...@apache.org.
Add a request tracker to track client rpc sequence numbers
This adds a new component to the rpc subsystem: the RequestTracker. This is inspired
by the RequestTracker in "Implementing Linearizability at Large Scale and Low Latency"
by Colin Lee et al. and is responsible for generating new sequence numbers for rpcs
and to keep track of the incomplete ones so that we're able to implement exactly-once
semantics for certain rpcs.
Change-Id: I23201625ca02f244dc94205d88dabc01608de471
Reviewed-on: http://gerrit.cloudera.org:8080/3078
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/9d5ce002
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/9d5ce002
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/9d5ce002
Branch: refs/heads/master
Commit: 9d5ce0027eeede5ef906545d9abc258041309d7d
Parents: ab500fc
Author: David Alves <da...@cloudera.com>
Authored: Tue May 3 13:23:08 2016 -0700
Committer: David Ribeiro Alves <da...@cloudera.com>
Committed: Wed May 25 03:08:02 2016 +0000
----------------------------------------------------------------------
src/kudu/rpc/CMakeLists.txt | 2 +
src/kudu/rpc/request_tracker-test.cc | 89 +++++++++++++++++++++++++++++++
src/kudu/rpc/request_tracker.cc | 49 +++++++++++++++++
src/kudu/rpc/request_tracker.h | 84 +++++++++++++++++++++++++++++
4 files changed, 224 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/9d5ce002/src/kudu/rpc/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/CMakeLists.txt b/src/kudu/rpc/CMakeLists.txt
index 6dae57e..8d2037e 100644
--- a/src/kudu/rpc/CMakeLists.txt
+++ b/src/kudu/rpc/CMakeLists.txt
@@ -53,6 +53,7 @@ set(KRPC_SRCS
proxy.cc
reactor.cc
remote_method.cc
+ request_tracker.cc
rpc.cc
rpc_context.cc
rpc_controller.cc
@@ -113,6 +114,7 @@ target_link_libraries(rtest_krpc
set(KUDU_TEST_LINK_LIBS rtest_krpc krpc ${KUDU_MIN_TEST_LIBS})
ADD_KUDU_TEST(mt-rpc-test RUN_SERIAL true)
ADD_KUDU_TEST(reactor-test)
+ADD_KUDU_TEST(request_tracker-test)
ADD_KUDU_TEST(rpc-bench RUN_SERIAL true)
ADD_KUDU_TEST(rpc-test)
ADD_KUDU_TEST(rpc_stub-test)
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/9d5ce002/src/kudu/rpc/request_tracker-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/request_tracker-test.cc b/src/kudu/rpc/request_tracker-test.cc
new file mode 100644
index 0000000..19efef6
--- /dev/null
+++ b/src/kudu/rpc/request_tracker-test.cc
@@ -0,0 +1,89 @@
+// 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 <gtest/gtest.h>
+#include <vector>
+
+#include "kudu/rpc/request_tracker.h"
+#include "kudu/util/test_util.h"
+
+using std::vector;
+
+namespace kudu {
+namespace rpc {
+
+TEST(RequestTrackerTest, TestSequenceNumberGeneration) {
+ const int MAX = 10;
+
+ scoped_refptr<RequestTracker> tracker_(new RequestTracker("test_client"));
+
+ // A new tracker should have no incomplete RPCs
+ ASSERT_TRUE(tracker_->FirstIncomplete(nullptr).IsNotFound());
+
+ vector<RequestTracker::SequenceNumber> generated_seq_nos;
+
+ // Generate MAX in flight RPCs, making sure they are correctly returned.
+ for (int i = 0; i < MAX; i++) {
+ RequestTracker::SequenceNumber seq_no;
+ ASSERT_OK(tracker_->NewSeqNo(&seq_no));
+ generated_seq_nos.push_back(seq_no);
+ }
+
+ // Now we should get a first incomplete.
+ RequestTracker::SequenceNumber first_incomplete;
+ ASSERT_OK(tracker_->FirstIncomplete(&first_incomplete));
+ ASSERT_EQ(generated_seq_nos[0], first_incomplete);
+
+ // Marking 'first_incomplete' as done, should advance the first incomplete.
+ tracker_->RpcCompleted(first_incomplete);
+
+ ASSERT_OK(tracker_->FirstIncomplete(&first_incomplete));
+ ASSERT_EQ(generated_seq_nos[1], first_incomplete);
+
+ // Marking a 'middle' rpc, should not advance 'first_incomplete'.
+ tracker_->RpcCompleted(generated_seq_nos[5]);
+ ASSERT_OK(tracker_->FirstIncomplete(&first_incomplete));
+ ASSERT_EQ(generated_seq_nos[1], first_incomplete);
+
+ // Marking half the rpc as complete should advance FirstIncomplete.
+ // Note that this also tests that RequestTracker::RpcCompleted() is idempotent, i.e. that
+ // marking the same sequence number as complete twice is a no-op.
+ for (int i = 0; i < MAX / 2; i++) {
+ tracker_->RpcCompleted(generated_seq_nos[i]);
+ }
+
+ ASSERT_OK(tracker_->FirstIncomplete(&first_incomplete));
+ ASSERT_EQ(generated_seq_nos[6], first_incomplete);
+
+ for (int i = MAX / 2; i <= MAX; i++) {
+ RequestTracker::SequenceNumber seq_no;
+ ASSERT_OK(tracker_->NewSeqNo(&seq_no));
+ generated_seq_nos.push_back(seq_no);
+ }
+
+ // Marking them all as completed should cause RequestTracker::FirstIncomplete() to return
+ // Status::NotFound() again.
+ for (auto seq_no : generated_seq_nos) {
+ tracker_->RpcCompleted(seq_no);
+ }
+
+ ASSERT_TRUE(tracker_->FirstIncomplete(nullptr).IsNotFound());
+}
+
+} // namespace rpc
+} // namespace kudu
+
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/9d5ce002/src/kudu/rpc/request_tracker.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/request_tracker.cc b/src/kudu/rpc/request_tracker.cc
new file mode 100644
index 0000000..4774c04
--- /dev/null
+++ b/src/kudu/rpc/request_tracker.cc
@@ -0,0 +1,49 @@
+// 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/request_tracker.h"
+#include "kudu/gutil/map-util.h"
+
+namespace kudu {
+namespace rpc {
+
+RequestTracker::RequestTracker(const string& client_id)
+ : client_id_(client_id),
+ next_(0) {}
+
+Status RequestTracker::NewSeqNo(SequenceNumber* seq_no) {
+ lock_guard<simple_spinlock> l(&lock_);
+ *seq_no = next_;
+ InsertOrDie(&incomplete_rpcs_, *seq_no);
+ next_++;
+ return Status::OK();
+}
+
+Status RequestTracker::FirstIncomplete(SequenceNumber* seq_no) {
+ lock_guard<simple_spinlock> l(&lock_);
+ if (incomplete_rpcs_.empty()) return Status::NotFound("There are no incomplete RPCs");
+ *seq_no = *incomplete_rpcs_.begin();
+ return Status::OK();
+}
+
+void RequestTracker::RpcCompleted(const SequenceNumber& seq_no) {
+ lock_guard<simple_spinlock> l(&lock_);
+ incomplete_rpcs_.erase(seq_no);
+}
+
+} // namespace rpc
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/9d5ce002/src/kudu/rpc/request_tracker.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/request_tracker.h b/src/kudu/rpc/request_tracker.h
new file mode 100644
index 0000000..8147e21
--- /dev/null
+++ b/src/kudu/rpc/request_tracker.h
@@ -0,0 +1,84 @@
+// 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 <set>
+#include <string>
+
+#include "kudu/util/locks.h"
+#include "kudu/util/status.h"
+
+namespace kudu {
+namespace rpc {
+
+// RequestTracker implementation, inspired by:
+// "Implementing Linearizability at Large Scale and Low Latency" by Colin Lee et al.
+//
+// This generates sequence numbers for retriable RPCs and tracks the ongoing ones.
+// The main point of this is to enable exactly-once semantics, i.e. making sure that
+// an RPC is only executed once, by uniquely identifying each RPC that is sent to
+// the server.
+//
+// Note that the sequence numbers here are differet from RPC 'call ids'. A call id
+// uniquely identifies a call _to a server_. All calls have a call id that is
+// assigned incrementally. Sequence numbers, on the other hand, uniquely identify
+// the RPC operation itself. That is, if an RPC is retried on another server it will
+// have a different call id, but the same sequence number.
+//
+// By keeping track of the RPCs that are in-flight and which ones are completed
+// we can determine the first incomplete RPC. When this information is sent
+// to the server it can use it to garbage collect RPC results that it might be
+// saving for future retries, since it now knows there won't be any.
+//
+// This class is thread safe.
+class RequestTracker : public RefCountedThreadSafe<RequestTracker> {
+ public:
+ typedef int64_t SequenceNumber;
+ explicit RequestTracker(const std::string& client_id);
+
+ // Creates a new, unique, sequence number.
+ // Sequence numbers are assigned in increasing integer order.
+ // Returns Status::OK() and sets 'seq_no' if it was able to generate a sequence number
+ // or returns Status::ServiceUnavailable() if too many RPCs are in-flight, in which case
+ // the caller should try again later.
+ Status NewSeqNo(SequenceNumber* seq_no);
+
+ // Returns the sequence number of the first incomplete RPC.
+ // If there is no incomplete RPC returns Status::NotFound. 'seq_no' is not set.
+ Status FirstIncomplete(SequenceNumber* seq_no);
+
+ // Marks the rpc with 'seq_no' as completed.
+ void RpcCompleted(const SequenceNumber& seq_no);
+
+ // Returns the client id for this request tracker.
+ const std::string& client_id() { return client_id_; }
+ private:
+ // The client id for this request tracker.
+ const std::string client_id_;
+
+ // Lock that protects all non-const fields.
+ simple_spinlock lock_;
+
+ // The next sequence number.
+ SequenceNumber next_;
+
+ // The (ordered) set of incomplete RPCs.
+ std::set<SequenceNumber> incomplete_rpcs_;
+};
+
+} // namespace rpc
+} // namespace kudu