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 2017/08/24 01:05:32 UTC
[1/4] kudu git commit: remove control flow warnings
Repository: kudu
Updated Branches:
refs/heads/master a9d17c00a -> 55c40fc72
remove control flow warnings
This patch eliminates two -Wreturn-type warnings by adding default switch
statements that crash.
../../src/kudu/consensus/consensus_meta.cc: In member function ‘const kudu::consensus::RaftConfigPB& kudu::consensus::ConsensusMetadata::config_unlocked(kudu::consensus::RaftConfigState) const’:
../../src/kudu/consensus/consensus_meta.cc:136:1: warning: control reaches end of non-void function [-Wreturn-type]
}
../../src/kudu/integration-tests/fuzz-itest.cc: In function ‘TestOpType kudu::tablet::PickOpAtRandom(kudu::tablet::TestOpSets)’:
../../src/kudu/integration-tests/fuzz-itest.cc:477:1: warning: control reaches end of non-void function [-Wreturn-type]
}
Change-Id: I45ca805bc7db2e234f55d2c59efcf01a28593e12
Reviewed-on: http://gerrit.cloudera.org:8080/7794
Reviewed-by: Mike Percy <mp...@apache.org>
Tested-by: Adar Dembo <ad...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/145c01da
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/145c01da
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/145c01da
Branch: refs/heads/master
Commit: 145c01da49037e9999d057898891dee0734afd93
Parents: a9d17c0
Author: Adar Dembo <ad...@cloudera.com>
Authored: Wed Aug 23 16:42:54 2017 -0700
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Thu Aug 24 00:47:45 2017 +0000
----------------------------------------------------------------------
src/kudu/consensus/consensus_meta.cc | 1 +
src/kudu/integration-tests/fuzz-itest.cc | 2 ++
2 files changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/145c01da/src/kudu/consensus/consensus_meta.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus_meta.cc b/src/kudu/consensus/consensus_meta.cc
index 11a27a0..b2d53b0 100644
--- a/src/kudu/consensus/consensus_meta.cc
+++ b/src/kudu/consensus/consensus_meta.cc
@@ -132,6 +132,7 @@ const RaftConfigPB& ConsensusMetadata::config_unlocked(RaftConfigState type) con
case ACTIVE_CONFIG: return active_config_unlocked();
case COMMITTED_CONFIG: return committed_config_unlocked();
case PENDING_CONFIG: return pending_config_unlocked();
+ default: LOG(FATAL) << "Unknown RaftConfigState type: " << type;
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/145c01da/src/kudu/integration-tests/fuzz-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/fuzz-itest.cc b/src/kudu/integration-tests/fuzz-itest.cc
index 17bee5c..7ba9bb9 100644
--- a/src/kudu/integration-tests/fuzz-itest.cc
+++ b/src/kudu/integration-tests/fuzz-itest.cc
@@ -473,6 +473,8 @@ TestOpType PickOpAtRandom(TestOpSets sets) {
return kAllOps[rand() % kAllOps.size()];
case PK_ONLY:
return kPkOnlyOps[rand() % kPkOnlyOps.size()];
+ default:
+ LOG(FATAL) << "Unknown TestOpSets type: " << sets;
}
}
[2/4] kudu git commit: Run the gradle build as a part of the gerrit
tests
Posted by to...@apache.org.
Run the gradle build as a part of the gerrit tests
Change-Id: I1430cbd05ff78a69f2439e3a8f90e1ddde83a8d7
Reviewed-on: http://gerrit.cloudera.org:8080/7651
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Kudu Jenkins
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/b2f5837f
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/b2f5837f
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/b2f5837f
Branch: refs/heads/master
Commit: b2f5837f0e88f6bcddfae9bd3ca1ab668093c147
Parents: 145c01d
Author: Grant Henke <gr...@gmail.com>
Authored: Thu Aug 10 16:53:24 2017 -0500
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Thu Aug 24 00:47:59 2017 +0000
----------------------------------------------------------------------
build-support/jenkins/build-and-test.sh | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/b2f5837f/build-support/jenkins/build-and-test.sh
----------------------------------------------------------------------
diff --git a/build-support/jenkins/build-and-test.sh b/build-support/jenkins/build-and-test.sh
index 0a6f567..5291acb 100755
--- a/build-support/jenkins/build-and-test.sh
+++ b/build-support/jenkins/build-and-test.sh
@@ -56,6 +56,10 @@
# BUILD_JAVA Default: 1
# Build and test java code if this is set to 1.
#
+# BUILD_GRADLE Default: 1
+# When building java code, also build the gradle build if this is set
+# to 1.
+#
# BUILD_PYTHON Default: 1
# Build and test the Python wrapper of the client API.
#
@@ -68,6 +72,9 @@
# Extra flags which are passed to 'mvn' when building and running Java
# tests. This can be useful, for example, to choose a different maven
# repository location.
+#
+# GRADLE_FLAGS Default: ""
+# Extra flags which are passed to 'gradle' when running Gradle commands.
# If a commit messages contains a line that says 'DONT_BUILD', exit
# immediately.
@@ -103,6 +110,7 @@ export KUDU_ALLOW_SLOW_TESTS=${KUDU_ALLOW_SLOW_TESTS:-$DEFAULT_ALLOW_SLOW_TESTS}
export KUDU_COMPRESS_TEST_OUTPUT=${KUDU_COMPRESS_TEST_OUTPUT:-1}
export TEST_TMPDIR=${TEST_TMPDIR:-/tmp/kudutest-$UID}
BUILD_JAVA=${BUILD_JAVA:-1}
+BUILD_GRADLE=${BUILD_GRADLE:-1}
BUILD_PYTHON=${BUILD_PYTHON:-1}
BUILD_PYTHON3=${BUILD_PYTHON3:-1}
@@ -125,6 +133,7 @@ mkdir -p $BUILD_ROOT
# Same for the Java tests, which aren't inside BUILD_ROOT
rm -rf $SOURCE_ROOT/java/*/target
+rm -rf $SOURCE_ROOT/java/*/build
list_flaky_tests() {
local url="http://$TEST_RESULT_SERVER/list_failed_tests?num_days=3&build_pattern=%25kudu-test%25"
@@ -192,6 +201,7 @@ elif [ "$BUILD_TYPE" = "COVERAGE" ]; then
# We currently dont capture coverage for Java or Python.
BUILD_JAVA=0
+ BUILD_GRADLE=0
BUILD_PYTHON=0
BUILD_PYTHON3=0
elif [ "$BUILD_TYPE" = "LINT" ]; then
@@ -311,6 +321,7 @@ if [ "$RUN_FLAKY_ONLY" == "1" ] ; then
BUILD_PYTHON=0
BUILD_PYTHON3=0
BUILD_JAVA=0
+ BUILD_GRADLE=0
fi
EXIT_STATUS=0
@@ -380,6 +391,16 @@ if [ "$BUILD_JAVA" == "1" ]; then
FAILURES="$FAILURES"$'Spark 1.x build/test failed\n'
fi
+ # Rerun the build using the Gradle build.
+ # Note: We just ensure we can assemble and don't rerun the tests.
+ if [ "$BUILD_GRADLE" == "1" ]; then
+ GRADLE_FLAGS="$GRADLE_FLAGS --console=plain"
+ if ! ./gradlew $GRADLE_FLAGS clean assemble; then
+ EXIT_STATUS=1
+ FAILURES="$FAILURES"$'Java Gradle build failed\n'
+ fi
+ fi
+
set +x
popd
fi
[3/4] kudu git commit: [java] Gradle test task improvements
Posted by to...@apache.org.
[java] Gradle test task improvements
Previously some tests would be extremely flaky when
running via Gradle. Especially secure tests with token
renewal. This fixes that issue by forking a new jvm for
each test class to isolate the tests.
I also add a few system properties to match the
Maven build and support filtering tests from the root
directory with the --tests argument.
Change-Id: Ifb660b15423b68c299518b7ba54a15135972a007
Reviewed-on: http://gerrit.cloudera.org:8080/7782
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Adar Dembo <ad...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/df2fa4f7
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/df2fa4f7
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/df2fa4f7
Branch: refs/heads/master
Commit: df2fa4f7cf88b44b1dbf07ba5e8859f73af0b617
Parents: b2f5837
Author: Grant Henke <gr...@gmail.com>
Authored: Wed Aug 23 12:54:34 2017 -0500
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Thu Aug 24 00:48:31 2017 +0000
----------------------------------------------------------------------
java/gradle/tests.gradle | 21 ++++++++++++++++++++-
java/pom.xml | 2 +-
2 files changed, 21 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/df2fa4f7/java/gradle/tests.gradle
----------------------------------------------------------------------
diff --git a/java/gradle/tests.gradle b/java/gradle/tests.gradle
index 01e3c0d..6963d2e 100644
--- a/java/gradle/tests.gradle
+++ b/java/gradle/tests.gradle
@@ -22,12 +22,31 @@ test {
maxParallelForks = propertyWithDefault("maxParallelForks", "1").toInteger()
}
-// Log all test events.
tasks.withType(Test) {
+ // Log all test events.
testLogging {
events "passed", "skipped", "failed"
exceptionFormat = "full"
}
+
+ // Fork the jvm for every test class.
+ // This takes more time and resources but ensures isolation.
+ forkEvery 1
+
+ // Enable assertions during tests.
+ jvmArgs "-enableassertions"
+
+ // Set a few system properties.
+ systemProperty "java.awt.headless", true
+ systemProperty "java.net.preferIPv4Stack", true
+ systemProperty "java.security.egd", "file:/dev/urandom" // Improve RNG generation speed.
+
+ // Don't fail the build if a --tests filter doesn't match any tests.
+ // This is useful for filtering tests from the top of the project where some
+ // subprojects won't match the filter. Without this, those subprojects would fail.
+ filter {
+ failOnNoMatchingTests = false
+ }
}
// Adds pattern based integration test support.
http://git-wip-us.apache.org/repos/asf/kudu/blob/df2fa4f7/java/pom.xml
----------------------------------------------------------------------
diff --git a/java/pom.xml b/java/pom.xml
index 20b9eff..6567883 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -98,7 +98,7 @@
<!-- Misc variables -->
<testdata.dir>target/testdata</testdata.dir>
<testArgLine>-enableassertions -Xmx1900m
- -Djava.security.egd=file:/dev/./urandom -Djava.net.preferIPv4Stack=true
+ -Djava.security.egd=file:/dev/urandom -Djava.net.preferIPv4Stack=true
-Djava.awt.headless=true
</testArgLine>
</properties>
[4/4] kudu git commit: Revert "KUDU-2065,
KUDU-2011: Release sidecars on cancellation or timeout"
Posted by to...@apache.org.
Revert "KUDU-2065, KUDU-2011: Release sidecars on cancellation or timeout"
This reverts commit 351337ee2d92361e2c87699d1c87a49774a3ab4d.
This patch seemed to cause a problem in test cluster that runs YCSB
stress tests. An RPC client and server got "out of sync" somehow on a
connection so that the footer of one call was interpreted as the length
prefix of the next call by the server. It then expected a 40MB+ RPC and
just sat there consuming RPC calls for many hours assuming it was all
part of one large RPC. The other end just kept timing out since it was
sending calls and not getting any responses.
Michael and I looked into it for a while and found one issue but still
couldn't fully explain how this happened, so we'll revert for now.
Change-Id: I1430cabfc54e858c1ad073574404821ae71c169f
Reviewed-on: http://gerrit.cloudera.org:8080/7788
Tested-by: Todd Lipcon <to...@apache.org>
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/55c40fc7
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/55c40fc7
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/55c40fc7
Branch: refs/heads/master
Commit: 55c40fc72241a0569d311bdb2f0268389512a807
Parents: df2fa4f
Author: Todd Lipcon <to...@apache.org>
Authored: Wed Aug 23 15:02:04 2017 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Thu Aug 24 01:03:35 2017 +0000
----------------------------------------------------------------------
docs/design-docs/rpc.md | 41 ++------
src/kudu/rpc/blocking_ops.cc | 3 +-
src/kudu/rpc/connection.cc | 169 +++++++++++---------------------
src/kudu/rpc/connection.h | 49 +--------
src/kudu/rpc/constants.cc | 6 +-
src/kudu/rpc/inbound_call.cc | 9 +-
src/kudu/rpc/inbound_call.h | 11 +--
src/kudu/rpc/outbound_call.cc | 71 +++-----------
src/kudu/rpc/outbound_call.h | 63 ++++--------
src/kudu/rpc/reactor.cc | 4 +-
src/kudu/rpc/rpc-test-base.h | 20 ++--
src/kudu/rpc/rpc-test.cc | 28 ++----
src/kudu/rpc/rpc_header.proto | 14 ---
src/kudu/rpc/rtest.proto | 3 +-
src/kudu/rpc/serialization.cc | 52 ++--------
src/kudu/rpc/serialization.h | 15 ---
src/kudu/rpc/server_negotiation.cc | 3 -
src/kudu/rpc/transfer.cc | 107 ++++----------------
src/kudu/rpc/transfer.h | 21 +---
19 files changed, 158 insertions(+), 531 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/docs/design-docs/rpc.md
----------------------------------------------------------------------
diff --git a/docs/design-docs/rpc.md b/docs/design-docs/rpc.md
index 6a168d4..474962d 100644
--- a/docs/design-docs/rpc.md
+++ b/docs/design-docs/rpc.md
@@ -271,10 +271,6 @@ a typical message looks like without sidecars:
+------------------------------------------------+
| Main message protobuf |
+------------------------------------------------+
-| RPC Footer protobuf length (variable encoding) | --- Exists only in client requests.
-+------------------------------------------------+
-| RPC Footer protobuf | --- Exists only in client requests.
-+------------------------------------------------+
```
In this case, the main message length is equal to the protobuf's byte size.
@@ -303,10 +299,6 @@ Here's what it looks like with the sidecars:
+------------------------------------------------+ --- ...
| ... |
+------------------------------------------------+
-| RPC Footer protobuf length (variable encoding) | --- Exists only in client requests.
-+------------------------------------------------+
-| RPC Footer protobuf | --- Exists only in client requests.
-+------------------------------------------------+
```
When there are sidecars, the sidecar_offsets member in the header will be a
@@ -359,11 +351,6 @@ serialized as follows:
- for typical RPC calls, this is the user-specified request or response protobuf
- for RPC calls which caused an error, the response is an ErrorStatusPB
- during negotiation, this is a NegotiatePB
-
- footer: varint-prefixed footer protobuf
- - the last part of a RPC message
- - exists only in client->server message if both client and server support HAS_FOOTER feature flag
- - contains a flag which is set when the client->server message is cancelled mid-transmission.
```
### Example packet capture
@@ -371,29 +358,19 @@ serialized as follows:
An example call (captured with strace on rpc-test.cc) follows:
```
- "\x00\x00\x00\x40" (total_size: 64 bytes to follow)
- "\x2f" RequestHeader varint: 47 bytes
- "\x18\x09\x32\x28\x0a\x21\x6b\x75\x64\x75\x2e\x72\x70\x63\x2e\x47\x65\x6e\x65"
- "\x72\x69\x63\x43\x61\x6c\x63\x75\x6c\x61\x74\x6f\x72\x53\x65\x72\x76\x69\x63"
- "\x65\x12\x03\x41\x64\x64\x50\x90\x4e" (RequestHeader protobuf)
- Decoded with printf "\x08\x0a..." | protoc --decode=kudu.rpc.RequestHeader rpc_header.proto:
- call_id: 9
- remote_method {
- service_name: "kudu.rpc.GenericCalculatorService"
- method_name: "Add"
- }
- timeout_millis: 10000
+ "\x00\x00\x00\x17" (total_size: 23 bytes to follow)
+ "\x09" RequestHeader varint: 9 bytes
+ "\x08\x0a\x1a\x03\x41\x64\x64\x20\x01" (RequestHeader protobuf)
+ Decoded with protoc --decode=RequestHeader rpc_header.proto:
+ callId: 10
+ methodName: "Add"
+ requestParam: true
"\x0c" Request parameter varint: 12 bytes
- "\x08\xd4\x90\x80\x91\x01\x10\xf8\xcf\xc4\xed\x04" (Request parameter protobuf)
- Decoded with printf "\x08\xd4..." | protoc --decode=kudu.rpc_test.AddRequestPB rtest.proto
+ "\x08\xd4\x90\x80\x91\x01\x10\xf8\xcf\xc4\xed\x04" Request parameter
+ Decoded with protoc --decode=kudu.rpc_test.AddRequestPB rpc/rtest.proto
x: 304089172
y: 1303455736
-
- "\x02" RequestFooter varint: 2 bytes
- "\x08\x00" (RequestFooter protobuf)
- Decoded with printf "\x08\x00" | protoc --decode=kudu.rpc_test.RequestFooterPB rpc_header.proto
- aborted: false
```
## Negotiation
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/blocking_ops.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/blocking_ops.cc b/src/kudu/rpc/blocking_ops.cc
index bd66ce3..e676e94 100644
--- a/src/kudu/rpc/blocking_ops.cc
+++ b/src/kudu/rpc/blocking_ops.cc
@@ -120,8 +120,7 @@ Status ReceiveFramedMessageBlocking(Socket* sock, faststring* recv_buf,
recv_buf->resize(payload_len + kMsgLengthPrefixLength);
RETURN_NOT_OK(sock->BlockingRecv(recv_buf->data() + kMsgLengthPrefixLength,
payload_len, &recvd, deadline));
- RETURN_NOT_OK(
- serialization::ParseMessage(Slice(*recv_buf), header, nullptr, param_buf));
+ RETURN_NOT_OK(serialization::ParseMessage(Slice(*recv_buf), header, param_buf));
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/connection.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/connection.cc b/src/kudu/rpc/connection.cc
index f8588db..d9ac03b 100644
--- a/src/kudu/rpc/connection.cc
+++ b/src/kudu/rpc/connection.cc
@@ -234,45 +234,12 @@ void Connection::CallAwaitingResponse::HandleTimeout(ev::timer &watcher, int rev
conn->HandleOutboundCallTimeout(this);
}
-void Connection::CancelOutboundTransfer(CallAwaitingResponse *car, const Status &status) {
- DCHECK(reactor_thread_->IsCurrentThread());
- DCHECK(car->call);
- const shared_ptr<OutboundCall>& call = car->call;
- if (call->IsInTransmission()) {
- DCHECK(call->call_id_assigned());
- DCHECK(car->transfer != nullptr);
- DCHECK(car->transfer->is_for_outbound_call());
- if (call->IsOnOutboundQueue()) {
- // If the transfer hasn't started yet, it should be safe to just call Abort().
- car->transfer->Abort(status);
- } else if (RemoteSupportsFooter()) {
- car->transfer->Cancel(car->call, status);
- } else {
- // If the remote server doesn't support parsing footer, there is no way to safely
- // cancel a mid-transmission trasfer with sidecars. It's fatal to have non-empty
- // sidecars attached to the call as it may result in use-after-free of the sidecars.
- // Please see the header comment for details.
- CHECK(!call->HasNonEmptySidecars());
- }
- }
-}
-
void Connection::HandleOutboundCallTimeout(CallAwaitingResponse *car) {
DCHECK(reactor_thread_->IsCurrentThread());
-
- if (!car->call) {
- // If cancellation callback is scheduled before the timeout callback is
- // invoked, car->call may be nullptr.
- return;
- }
-
+ DCHECK(car->call);
// The timeout timer is stopped by the car destructor exiting Connection::HandleCallResponse()
DCHECK(!car->call->IsFinished());
- // Cancel or abort any outbound transfer for this call so references to the sidecars
- // can be relinquished.
- CancelOutboundTransfer(car, Status::Aborted("timed out"));
-
// Mark the call object as failed.
car->call->SetTimedOut(negotiation_complete_ ? Phase::REMOTE_CALL
: Phase::CONNECTION_NEGOTIATION);
@@ -295,10 +262,7 @@ void Connection::HandleOutboundCallTimeout(CallAwaitingResponse *car) {
void Connection::CancelOutboundCall(const shared_ptr<OutboundCall> &call) {
CallAwaitingResponse* car = FindPtrOrNull(awaiting_response_, call->call_id());
if (car != nullptr) {
- car->timeout_timer.stop();
- // Cancel any outbound transfer for this call so references to the sidecars can be
- // relinquished.
- CancelOutboundTransfer(car, Status::Aborted("cancelled"));
+ // car->call may be NULL if the call has timed out already.
DCHECK(!car->call || car->call.get() == call.get());
car->call.reset();
}
@@ -362,8 +326,8 @@ void Connection::QueueOutboundCall(const shared_ptr<OutboundCall> &call) {
// yet assigned a call ID.
DCHECK(!call->call_id_assigned());
- // We shouldn't reach this point if 'call' was cancelled.
- DCHECK(!call->IsCancelled());
+ // We shouldn't reach this point if 'call' was requested to be cancelled.
+ DCHECK(!call->cancellation_requested());
// Assign the call ID.
int32_t call_id = GetNextCallId();
@@ -427,11 +391,9 @@ void Connection::QueueOutboundCall(const shared_ptr<OutboundCall> &call) {
}
TransferCallbacks *cb = new CallTransferCallbacks(call, this);
- gscoped_ptr<OutboundTransfer>
- transfer(OutboundTransfer::CreateForCallRequest(call_id, tmp_slices, n_slices, cb));
- car->transfer = transfer.get();
awaiting_response_[call_id] = car.release();
- QueueOutbound(std::move(transfer));
+ QueueOutbound(gscoped_ptr<OutboundTransfer>(
+ OutboundTransfer::CreateForCallRequest(call_id, tmp_slices, n_slices, cb)));
}
// Callbacks for sending an RPC call response from the server.
@@ -448,7 +410,7 @@ struct ResponseTransferCallbacks : public TransferCallbacks {
~ResponseTransferCallbacks() {
// Remove the call from the map.
InboundCall *call_from_map = EraseKeyReturnValuePtr(
- &conn_->calls_being_handled_, call_->call_id());
+ &conn_->calls_being_handled_, call_->call_id());
DCHECK_EQ(call_from_map, call_.get());
}
@@ -457,8 +419,8 @@ struct ResponseTransferCallbacks : public TransferCallbacks {
}
virtual void NotifyTransferAborted(const Status &status) OVERRIDE {
- LOG(WARNING) << "Connection torn down before " << call_->ToString()
- << " could send its response: " << status.ToString();
+ LOG(WARNING) << "Connection torn down before " <<
+ call_->ToString() << " could send its response";
delete this;
}
@@ -581,11 +543,7 @@ void Connection::HandleIncomingCall(gscoped_ptr<InboundTransfer> transfer) {
gscoped_ptr<InboundCall> call(new InboundCall(this));
Status s = call->ParseFrom(std::move(transfer));
- if (PREDICT_FALSE(s.IsAborted())) {
- VLOG(2) << "Incoming call was aborted due to: "<< s.ToString();
- return;
- }
- if (PREDICT_FALSE(!s.ok())) {
+ if (!s.ok()) {
LOG(WARNING) << ToString() << ": received bad data: " << s.ToString();
// TODO: shutdown? probably, since any future stuff on this socket will be
// "unsynchronized"
@@ -610,7 +568,7 @@ void Connection::HandleCallResponse(gscoped_ptr<InboundTransfer> transfer) {
CHECK_OK(resp->ParseFrom(std::move(transfer)));
CallAwaitingResponse *car_ptr =
- EraseKeyReturnValuePtr(&awaiting_response_, resp->call_id());
+ EraseKeyReturnValuePtr(&awaiting_response_, resp->call_id());
if (PREDICT_FALSE(car_ptr == nullptr)) {
LOG(WARNING) << ToString() << ": Got a response for call id " << resp->call_id() << " which "
<< "was not pending! Ignoring.";
@@ -633,58 +591,6 @@ void Connection::HandleCallResponse(gscoped_ptr<InboundTransfer> transfer) {
MaybeInjectCancellation(car->call);
}
-bool Connection::StartCallTransfer(OutboundTransfer *transfer) {
- DCHECK(!transfer->TransferStarted());
- CallAwaitingResponse* car = FindOrDie(awaiting_response_, transfer->call_id());
-
- // If the call has already timed out or has already been cancelled, the 'call'
- // field would be set to NULL. In that case, don't bother sending it.
- if (!car->call) {
- DCHECK(transfer->TransferAborted());
- return false;
- }
-
- DCHECK(!transfer->TransferAborted());
- // If this is the start of the transfer, then check if the server has the
- // required RPC flags. We have to wait until just before the transfer in
- // order to ensure that the negotiation has taken place, so that the flags
- // are available.
- DCHECK(negotiation_complete_);
- const set<RpcFeatureFlag>& required_features = car->call->required_rpc_features();
- if (!includes(remote_features_.begin(), remote_features_.end(),
- required_features.begin(), required_features.end())) {
- Status s = Status::NotSupported("server does not support the required RPC features");
- transfer->Abort(s);
- car->call->SetFailed(s, Phase::REMOTE_CALL);
- // Test cancellation when 'call_' is in 'FINISHED_ERROR' state.
- MaybeInjectCancellation(car->call);
- car->call.reset();
- return false;
- }
-
- // The transfer is ready to be sent. Append a footer if the remote system supports it.
- // It has to be done here because remote features cannot be determined until negotiation
- // completes. We know for sure that negotiation has completed when we reach this point.
- if (RemoteSupportsFooter()) {
- transfer->AppendFooter(car->call);
- }
- car->call->SetSending();
- // Test cancellation when 'call_' is in 'SENDING' state.
- MaybeInjectCancellation(car->call);
- return true;
-}
-
-void Connection::OutboundQueuePopFront() {
- OutboundTransfer *transfer = &(outbound_transfers_.front());
- // Remove all references to 'transfer' before deleting it.
- if (transfer->is_for_outbound_call()) {
- CallAwaitingResponse* car = FindOrDie(awaiting_response_, transfer->call_id());
- car->transfer = nullptr;
- }
- outbound_transfers_.pop_front();
- delete transfer;
-}
-
void Connection::WriteHandler(ev::io &watcher, int revents) {
DCHECK(reactor_thread_->IsCurrentThread());
@@ -695,6 +601,7 @@ void Connection::WriteHandler(ev::io &watcher, int revents) {
}
DVLOG(3) << ToString() << ": writeHandler: revents = " << revents;
+ OutboundTransfer *transfer;
if (outbound_transfers_.empty()) {
LOG(WARNING) << ToString() << " got a ready-to-write callback, but there is "
"nothing to write.";
@@ -703,14 +610,45 @@ void Connection::WriteHandler(ev::io &watcher, int revents) {
}
while (!outbound_transfers_.empty()) {
- OutboundTransfer *transfer = &(outbound_transfers_.front());
- // Transfer for outbound call must call StartCallTransfer() before transmission can
- // begin to append footer to the payload if the remote supports it.
- if (!transfer->TransferStarted() &&
- transfer->is_for_outbound_call() &&
- !StartCallTransfer(transfer)) {
- OutboundQueuePopFront();
- continue;
+ transfer = &(outbound_transfers_.front());
+
+ if (!transfer->TransferStarted()) {
+
+ if (transfer->is_for_outbound_call()) {
+ CallAwaitingResponse* car = FindOrDie(awaiting_response_, transfer->call_id());
+ if (!car->call) {
+ // If the call has already timed out or has already been cancelled, the 'call'
+ // field would be set to NULL. In that case, don't bother sending it.
+ outbound_transfers_.pop_front();
+ transfer->Abort(Status::Aborted("already timed out or cancelled"));
+ delete transfer;
+ continue;
+ }
+
+ // If this is the start of the transfer, then check if the server has the
+ // required RPC flags. We have to wait until just before the transfer in
+ // order to ensure that the negotiation has taken place, so that the flags
+ // are available.
+ const set<RpcFeatureFlag>& required_features = car->call->required_rpc_features();
+ if (!includes(remote_features_.begin(), remote_features_.end(),
+ required_features.begin(), required_features.end())) {
+ outbound_transfers_.pop_front();
+ Status s = Status::NotSupported("server does not support the required RPC features");
+ transfer->Abort(s);
+ car->call->SetFailed(s, negotiation_complete_ ? Phase::REMOTE_CALL
+ : Phase::CONNECTION_NEGOTIATION);
+ // Test cancellation when 'call_' is in 'FINISHED_ERROR' state.
+ MaybeInjectCancellation(car->call);
+ car->call.reset();
+ delete transfer;
+ continue;
+ }
+
+ car->call->SetSending();
+
+ // Test cancellation when 'call_' is in 'SENDING' state.
+ MaybeInjectCancellation(car->call);
+ }
}
last_activity_time_ = reactor_thread_->cur_time();
@@ -720,11 +658,14 @@ void Connection::WriteHandler(ev::io &watcher, int revents) {
reactor_thread_->DestroyConnection(this, status);
return;
}
+
if (!transfer->TransferFinished()) {
DVLOG(3) << ToString() << ": writeHandler: xfer not finished.";
return;
}
- OutboundQueuePopFront();
+
+ outbound_transfers_.pop_front();
+ delete transfer;
}
// If we were able to write all of our outbound transfers,
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/connection.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/connection.h b/src/kudu/rpc/connection.h
index 92ad56a..44c85f7 100644
--- a/src/kudu/rpc/connection.h
+++ b/src/kudu/rpc/connection.h
@@ -130,7 +130,7 @@ class Connection : public RefCountedThreadSafe<Connection> {
void QueueResponseForCall(gscoped_ptr<InboundCall> call);
// Cancel an outbound call by removing any reference to it by CallAwaitingResponse
- // in 'awaiting_response_'.
+ // in 'awaiting_responses_'.
void CancelOutboundCall(const std::shared_ptr<OutboundCall> &call);
// The address of the remote end of the connection.
@@ -203,16 +203,6 @@ class Connection : public RefCountedThreadSafe<Connection> {
remote_features_ = std::move(remote_features);
}
- // Returns true if the remote end has feature flag "REQUEST_FOOTERS".
- // Always returns false before negotiation completes.
- bool RemoteSupportsFooter() const {
- // Note the negotiation thread may be writing to 'remote_features_' before
- // 'negotiation_complete_' is set to true. It's not safe to read 'remote_features_'
- // from the reactor thread when 'negotiation_complete_' is false.
- return negotiation_complete_ &&
- ContainsKey(remote_features_, RpcFeatureFlag::REQUEST_FOOTERS);
- }
-
void set_remote_user(RemoteUser user) {
DCHECK_EQ(direction_, SERVER);
remote_user_ = std::move(user);
@@ -250,14 +240,10 @@ class Connection : public RefCountedThreadSafe<Connection> {
// Notification from libev that the call has timed out.
void HandleTimeout(ev::timer &watcher, int revents);
- Connection *conn = nullptr;
+ Connection *conn;
std::shared_ptr<OutboundCall> call;
ev::timer timeout_timer;
- // The pending outbound transfer for this call. Set when an outbound call is
- // enqueued. Set back to NULL once the transfer has been completely sent.
- OutboundTransfer* transfer = nullptr;
-
// We time out RPC calls in two stages. This is set to the amount of timeout
// remaining after the next timeout fires. See Connection::QueueOutboundCall().
double remaining_timeout;
@@ -292,42 +278,11 @@ class Connection : public RefCountedThreadSafe<Connection> {
// Set it to Failed.
void HandleOutboundCallTimeout(CallAwaitingResponse *car);
- // Perform some checks on 'transfer' for an outbound call before beginning
- // transmission. The check could fail if:
- // 1. the outbound call has already timed out or has been cancelled.
- // 2. the remote server doesn't have all the required feature flags.
- //
- // If the checks pass, the outbound call is set to 'SENDING' state.
- // Appends a footer to the outbound call if remote supports "REQUEST_FOOTERS".
- //
- // If the checks fail due to (1), the transfer is already aborted and callback
- // has already been invoked. If it fails due to (2), the transfer is aborted
- // and the outbound call's callback is called with the failure status.
- //
- // Returns true if all checks pass; returns false otherwise.
- bool StartCallTransfer(OutboundTransfer *transfer);
-
- // Cancel any outbound transfer associated with 'car' and abort it with 'status'.
- // Called when an outbound call is cancelled or timed out. If the remote server
- // doesn't support footer (e.g. older version of Kudu server), it's fatal for the
- // call to have any non-empty sidecars. The reasoning is as follows:
- //
- // Up till the point footer is introduced, Kudu doesn't have any outbound call which
- // uses client sidecars. If any future RPCs make use of client sidecars, the Kudu
- // server code has to be updated to process the sidecars too, in which case the newer
- // version of Kudu server should already be able to process footers.
- void CancelOutboundTransfer(CallAwaitingResponse *car, const Status &status);
-
// Queue a transfer for sending on this connection.
// We will take ownership of the transfer.
// This must be called from the reactor thread.
void QueueOutbound(gscoped_ptr<OutboundTransfer> transfer);
- // Remove the front transfer from the outbound transfers queue. If the transfer is
- // associated with an outbound call, also clears the transfer reference from the
- // associated CallAwaitingResponse.
- void OutboundQueuePopFront();
-
// Internal test function for injecting cancellation request when 'call'
// reaches state specified in 'FLAGS_rpc_inject_cancellation_state'.
void MaybeInjectCancellation(const std::shared_ptr<OutboundCall> &call);
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/constants.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/constants.cc b/src/kudu/rpc/constants.cc
index 51945ca..bcf9985 100644
--- a/src/kudu/rpc/constants.cc
+++ b/src/kudu/rpc/constants.cc
@@ -31,12 +31,8 @@ const char* const kSaslProtoName = "kudu";
//
// NOTE: the TLS_AUTHENTICATION_ONLY flag is dynamically added on both
// sides based on the remote peer's address.
-//
-// NOTE: the REQUEST_FOOTERS is always set on the client side. The server side
-// which supports parsing footer sets REQUEST_FOOTERS if client side has it set.
set<RpcFeatureFlag> kSupportedServerRpcFeatureFlags = { APPLICATION_FEATURE_FLAGS };
-set<RpcFeatureFlag> kSupportedClientRpcFeatureFlags = { APPLICATION_FEATURE_FLAGS,
- REQUEST_FOOTERS };
+set<RpcFeatureFlag> kSupportedClientRpcFeatureFlags = { APPLICATION_FEATURE_FLAGS };
} // namespace rpc
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/inbound_call.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/inbound_call.cc b/src/kudu/rpc/inbound_call.cc
index a360e0b..5271a5a 100644
--- a/src/kudu/rpc/inbound_call.cc
+++ b/src/kudu/rpc/inbound_call.cc
@@ -69,14 +69,7 @@ InboundCall::~InboundCall() {}
Status InboundCall::ParseFrom(gscoped_ptr<InboundTransfer> transfer) {
TRACE_EVENT_FLOW_BEGIN0("rpc", "InboundCall", this);
TRACE_EVENT0("rpc", "InboundCall::ParseFrom");
- bool has_footer = conn_->RemoteSupportsFooter();
- RETURN_NOT_OK(serialization::ParseMessage(transfer->data(), &header_,
- has_footer ? &footer_ : nullptr, &serialized_request_));
-
- // If the transfer was already aborted, return.
- if (PREDICT_FALSE(has_footer && footer_.has_aborted() && footer_.aborted())) {
- return Status::Aborted("request footer has aborted flag set");
- }
+ RETURN_NOT_OK(serialization::ParseMessage(transfer->data(), &header_, &serialized_request_));
// Adopt the service/method info from the header as soon as it's available.
if (PREDICT_FALSE(!header_.has_remote_method())) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/inbound_call.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/inbound_call.h b/src/kudu/rpc/inbound_call.h
index b6281e7..ac903c6 100644
--- a/src/kudu/rpc/inbound_call.h
+++ b/src/kudu/rpc/inbound_call.h
@@ -77,9 +77,9 @@ class InboundCall {
// Parse an inbound call message.
//
- // This only deserializes the call header and footer, populating the 'header_',
- // 'footer_' and 'serialized_request_' member variables. The actual call parameter
- // is not deserialized, as this may be CPU-expensive, and this is called
+ // This only deserializes the call header, populating the 'header_' and
+ // 'serialized_request_' member variables. The actual call parameter is
+ // not deserialized, as this may be CPU-expensive, and this is called
// from the reactor thread.
Status ParseFrom(gscoped_ptr<InboundTransfer> transfer);
@@ -228,12 +228,9 @@ class InboundCall {
// The connection on which this inbound call arrived.
scoped_refptr<Connection> conn_;
- // The header of the incoming call. Set by ParseFrom().
+ // The header of the incoming call. Set by ParseFrom()
RequestHeader header_;
- // The footer of the incoming call. Set by ParseFrom().
- RequestFooterPB footer_;
-
// The serialized bytes of the request param protobuf. Set by ParseFrom().
// This references memory held by 'transfer_'.
Slice serialized_request_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/outbound_call.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/outbound_call.cc b/src/kudu/rpc/outbound_call.cc
index 2dad6e7..e300fe4 100644
--- a/src/kudu/rpc/outbound_call.cc
+++ b/src/kudu/rpc/outbound_call.cc
@@ -34,7 +34,6 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/sysinfo.h"
#include "kudu/gutil/walltime.h"
-#include "kudu/rpc/connection.h"
#include "kudu/rpc/constants.h"
#include "kudu/rpc/outbound_call.h"
#include "kudu/rpc/rpc_controller.h"
@@ -88,7 +87,8 @@ OutboundCall::OutboundCall(const ConnectionId& conn_id,
conn_id_(conn_id),
callback_(std::move(callback)),
controller_(DCHECK_NOTNULL(controller)),
- response_(DCHECK_NOTNULL(response_storage)) {
+ response_(DCHECK_NOTNULL(response_storage)),
+ cancellation_requested_(false) {
DVLOG(4) << "OutboundCall " << this << " constructed with state_: " << StateName(state_)
<< " and RPC timeout: "
<< (controller->timeout().Initialized() ? controller->timeout().ToString() : "none");
@@ -127,7 +127,6 @@ size_t OutboundCall::SerializeTo(TransferPayload* slices) {
serialization::SerializeHeader(
header_, sidecar_byte_size_ + request_buf_.size(), &header_buf_);
- // Header and Msg in addition to sidecars.
size_t n_slices = 2 + sidecars_.size();
DCHECK_LE(n_slices, slices->size());
auto slice_iter = slices->begin();
@@ -140,28 +139,6 @@ size_t OutboundCall::SerializeTo(TransferPayload* slices) {
return n_slices;
}
-void OutboundCall::AppendFooter(Slice* slice) {
- DCHECK(!footer_.has_aborted());
- footer_.set_aborted(false);
- serialization::SerializeFooter(footer_, &footer_buf_);
- *slice = Slice(footer_buf_);
- serialization::IncrementMsgLength(footer_buf_.size(), &header_buf_);
-}
-
-void OutboundCall::MarkPayloadCancelled(TransferPayload* slices,
- const uint8_t* relocated_dst) {
- DCHECK(footer_.has_aborted());
- auto slice_iter = slices->begin() + 2;
- for (auto& sidecar : sidecars_) {
- *slice_iter++ = Slice(relocated_dst, sidecar->AsSlice().size());
- }
- size_t old_size = footer_buf_.size();
- footer_.set_aborted(true);
- serialization::SerializeFooter(footer_, &footer_buf_);
- DCHECK_EQ(footer_buf_.size(), old_size);
- *slice_iter++ = Slice(footer_buf_);
-}
-
void OutboundCall::SetRequestPayload(const Message& req,
vector<unique_ptr<RpcSidecar>>&& sidecars) {
DCHECK_EQ(-1, sidecar_byte_size_);
@@ -252,8 +229,7 @@ void OutboundCall::set_state_unlocked(State new_state) {
DCHECK(state_ == SENT || state_ == ON_OUTBOUND_QUEUE || state_ == SENDING);
break;
case CANCELLED:
- DCHECK(state_ == READY || state_ == ON_OUTBOUND_QUEUE || state_ == SENDING ||
- state_ == SENT);
+ DCHECK(state_ == READY || state_ == ON_OUTBOUND_QUEUE || state_ == SENT);
break;
case FINISHED_SUCCESS:
DCHECK_EQ(state_, SENT);
@@ -266,24 +242,17 @@ void OutboundCall::set_state_unlocked(State new_state) {
state_ = new_state;
}
-void OutboundCall::Cancel(const Connection* conn) {
+void OutboundCall::Cancel() {
+ cancellation_requested_ = true;
// No lock needed as it's called from reactor thread
switch (state_) {
- case SENDING:
- // 'conn' can be NULL if the connection was somehow shut down due to errors.
- // If the remote server doesn't support parsing footer and the call is being sent
- // already, ignore the cancellation request as if it were completed already.
- if (PREDICT_FALSE(conn == nullptr || !conn->RemoteSupportsFooter())) {
- break;
- }
- // fall-through if remote supports parsing footer.
- FALLTHROUGH_INTENDED;
case READY:
case ON_OUTBOUND_QUEUE:
case SENT: {
SetCancelled();
break;
}
+ case SENDING:
case NEGOTIATION_TIMED_OUT:
case TIMED_OUT:
case CANCELLED:
@@ -358,19 +327,21 @@ void OutboundCall::SetSending() {
void OutboundCall::SetSent() {
set_state(SENT);
- // This method is called in the reactor thread, so free the header and footer buf,
- // which were also allocated from this thread. tcmalloc's thread caching
+ // This method is called in the reactor thread, so free the header buf,
+ // which was also allocated from this thread. tcmalloc's thread caching
// behavior is a lot more efficient if memory is freed from the same thread
// which allocated it -- this lets it keep to thread-local operations instead
// of taking a mutex to put memory back on the global freelist.
- header_buf_.clear();
- header_buf_.shrink_to_fit();
- footer_buf_.clear();
- footer_buf_.shrink_to_fit();
+ delete [] header_buf_.release();
// request_buf_ is also done being used here, but since it was allocated by
// the caller thread, we would rather let that thread free it whenever it
// deletes the RpcController.
+
+ // If cancellation was requested, it's now a good time to do the actual cancellation.
+ if (cancellation_requested()) {
+ SetCancelled();
+ }
}
void OutboundCall::SetFailed(const Status &status,
@@ -454,16 +425,6 @@ bool OutboundCall::IsNegotiationError() const {
}
}
-bool OutboundCall::IsOnOutboundQueue() const {
- std::lock_guard<simple_spinlock> l(lock_);
- return state_ == ON_OUTBOUND_QUEUE;
-}
-
-bool OutboundCall::IsInTransmission() const {
- std::lock_guard<simple_spinlock> l(lock_);
- return state_ == ON_OUTBOUND_QUEUE || state_ == SENDING;
-}
-
bool OutboundCall::IsFinished() const {
std::lock_guard<simple_spinlock> l(lock_);
switch (state_) {
@@ -550,8 +511,8 @@ Status CallResponse::GetSidecar(int idx, Slice* sidecar) const {
Status CallResponse::ParseFrom(gscoped_ptr<InboundTransfer> transfer) {
CHECK(!parsed_);
- RETURN_NOT_OK(serialization::ParseMessage(
- transfer->data(), &header_, nullptr, &serialized_response_));
+ RETURN_NOT_OK(serialization::ParseMessage(transfer->data(), &header_,
+ &serialized_response_));
// Use information from header to extract the payload slices.
RETURN_NOT_OK(RpcSidecar::ParseSidecars(header_.sidecar_offsets(),
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/outbound_call.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/outbound_call.h b/src/kudu/rpc/outbound_call.h
index a55fe97..52788de 100644
--- a/src/kudu/rpc/outbound_call.h
+++ b/src/kudu/rpc/outbound_call.h
@@ -105,32 +105,17 @@ class OutboundCall {
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. The serialized payload are stored as slices
- // in "slices". Returns the number of slices in the serialized call.
+ // 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);
- // Called right before the outbound call begins transmission to add a footer to the
- // message. This function sets aborted flag in 'footer' to false, serializes it to PB
- // and stores it in 'footer_buf_' and updates 'slice' to point to it. The total message
- // length (stored in the first 4 bytes of "header_buf_") is incremented to account for
- // the footer.
- // REQUIRES: must be called from the reactor thread.
- void AppendFooter(Slice* slice);
-
- // Called when the call is cancelled. Updates all sidecar slices to point to
- // 'relocated_dst' while keeping their sizes unchanged. Also update the footer
- // to indicate that the call has been aborted. Can only be called if AppendFooter()
- // was called before.
- // REQUIRES: must be called from the reactor thread.
- void MarkPayloadCancelled(TransferPayload* slices, const uint8_t* relocated_dst);
-
// 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. If the RPC is being sent, cancel it only
- // if the remote supports "REQUEST_FOOTERS". If the call is finished, it's a no-op.
+ // 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(const Connection* conn);
+ void Cancel();
// Callback after the call has been put on the outbound connection queue.
void SetQueued();
@@ -153,27 +138,15 @@ class OutboundCall {
// 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;
- bool IsOnOutboundQueue() const;
-
- // True if the call is scheduled to be sent or in the process of being sent.
- // There is an entry for the call in the OutboundTransfer queue.
- bool IsInTransmission() const;
-
// Is the call finished?
bool IsFinished() const;
- // True if the outbound call has any non-empty sidecars attached to it.
- bool HasNonEmptySidecars() const {
- return sidecar_byte_size_ > 0;
- }
-
// Fill in the call response.
void SetResponse(gscoped_ptr<CallResponse> resp);
@@ -205,6 +178,12 @@ class OutboundCall {
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 {
@@ -254,11 +233,6 @@ class OutboundCall {
// This will only be non-NULL if status().IsRemoteError().
const ErrorStatusPB* error_pb() const;
- // 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();
-
// Lock for state_ status_, error_pb_ fields, since they
// may be mutated by the reactor thread while the client thread
// reads them.
@@ -267,15 +241,16 @@ class OutboundCall {
Status status_;
gscoped_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 RPC footer.
- // This is sent last as part of a RPC request.
- RequestFooterPB footer_;
-
// The remote method being called.
RemoteMethod remote_method_;
@@ -292,7 +267,6 @@ class OutboundCall {
// Buffers for storing segments of the wire-format request.
faststring header_buf_;
faststring request_buf_;
- faststring footer_buf_;
// Once a response has been received for this call, contains that response.
// Otherwise NULL.
@@ -304,6 +278,9 @@ class OutboundCall {
// Total size in bytes of all sidecars in 'sidecars_'. Set in SetRequestPayload().
int64_t sidecar_byte_size_ = -1;
+ // True if cancellation was requested on this call.
+ bool cancellation_requested_;
+
DISALLOW_COPY_AND_ASSIGN(OutboundCall);
};
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/reactor.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/reactor.cc b/src/kudu/rpc/reactor.cc
index b9ecb67..f46e8db 100644
--- a/src/kudu/rpc/reactor.cc
+++ b/src/kudu/rpc/reactor.cc
@@ -296,7 +296,7 @@ void ReactorThread::CancelOutboundCall(const shared_ptr<OutboundCall>& call) {
&conn)) {
conn->CancelOutboundCall(call);
}
- call->Cancel(conn.get());
+ call->Cancel();
}
//
@@ -501,7 +501,7 @@ Status ReactorThread::StartConnectionNegotiation(const scoped_refptr<Connection>
ThreadPool* negotiation_pool =
reactor()->messenger()->negotiation_pool(conn->direction());
RETURN_NOT_OK(negotiation_pool->SubmitClosure(
- Bind(&Negotiation::RunNegotiation, conn, authentication, encryption, deadline)));
+ Bind(&Negotiation::RunNegotiation, conn, authentication, encryption, deadline)));
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/rpc-test-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc-test-base.h b/src/kudu/rpc/rpc-test-base.h
index 25d3845..757f1fe 100644
--- a/src/kudu/rpc/rpc-test-base.h
+++ b/src/kudu/rpc/rpc-test-base.h
@@ -226,19 +226,15 @@ class GenericCalculatorService : public ServiceIf {
LOG(INFO) << "got call: " << pb_util::SecureShortDebugString(req);
SleepFor(MonoDelta::FromMicroseconds(req.sleep_micros()));
- uint32_t pattern = req.pattern();
- for (int i = 0; i < req.sidecar_idx_size(); ++i) {
- Slice sidecar;
- CHECK_OK(incoming->GetInboundSidecar(req.sidecar_idx(i), &sidecar));
- CHECK_EQ(sidecar.size() % sizeof(uint32_t), 0);
- const uint32_t *data = reinterpret_cast<const uint32_t*>(sidecar.data());
- int num_repetitions = sidecar.size() / sizeof(uint32_t);
- for (int j = 0; j < num_repetitions; ++j) {
- CHECK_EQ(data[j], pattern);
- }
- }
+ uint32 pattern = req.pattern();
+ uint32 num_repetitions = req.num_repetitions();
+ Slice sidecar;
+ CHECK_OK(incoming->GetInboundSidecar(req.sidecar_idx(), &sidecar));
+ CHECK_EQ(sidecar.size(), sizeof(uint32) * num_repetitions);
+ const uint32_t *data = reinterpret_cast<const uint32_t*>(sidecar.data());
+ for (int i = 0; i < num_repetitions; ++i) CHECK_EQ(data[i], pattern);
- SleepWithSidecarResponsePB resp;
+ SleepResponsePB resp;
incoming->RespondSuccess(resp);
}
};
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/rpc-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc-test.cc b/src/kudu/rpc/rpc-test.cc
index a367976..f07dc8d 100644
--- a/src/kudu/rpc/rpc-test.cc
+++ b/src/kudu/rpc/rpc-test.cc
@@ -1011,7 +1011,7 @@ TEST_P(TestRpc, TestCancellation) {
client_messenger->Shutdown();
}
-#define TEST_PAYLOAD_SIZE (1 << 26)
+#define TEST_PAYLOAD_SIZE (1 << 23)
#define TEST_SLEEP_TIME_MS (500)
static void SleepCallback(uint8_t* payload, CountDownLatch* latch) {
@@ -1042,38 +1042,30 @@ TEST_P(TestRpc, TestCancellationAsync) {
// Used to generate sleep time between invoking RPC and requesting cancellation.
Random rand(SeedRandom());
- // Set the max allowed payload size to 1GB.
- FLAGS_rpc_max_message_size = 1 << 30;
-
- for (int i = 0; i < 100; ++i) {
+ for (int i = 0; i < 10; ++i) {
SleepWithSidecarRequestPB req;
SleepWithSidecarResponsePB resp;
// Initialize the payload with non-zero pattern.
+ memset(payload.get(), 0xff, TEST_PAYLOAD_SIZE);
req.set_sleep_micros(TEST_SLEEP_TIME_MS);
req.set_pattern(0xffffffff);
- memset(payload.get(), 0xff, TEST_PAYLOAD_SIZE);
+ req.set_num_repetitions(TEST_PAYLOAD_SIZE / sizeof(uint32_t));
- uint32 num_sidecars = rand.Uniform32(TransferLimits::kMaxSidecars);
- for (int j = 0; j < num_sidecars; ++j) {
- uint32 num_repetitions = rand.Uniform32(TEST_PAYLOAD_SIZE) / sizeof(uint32_t);
- Slice s(payload.get(), num_repetitions * sizeof(uint32_t));
- int idx;
- CHECK_OK(controller.AddOutboundSidecar(RpcSidecar::FromSlice(s), &idx));
- req.add_sidecar_idx(idx);
- }
+ int idx;
+ Slice s(payload.get(), TEST_PAYLOAD_SIZE);
+ CHECK_OK(controller.AddOutboundSidecar(RpcSidecar::FromSlice(s), &idx));
+ req.set_sidecar_idx(idx);
CountDownLatch latch(1);
- controller.set_timeout(MonoDelta::FromMicroseconds(rand.Uniform64(i * 500L + 1)));
p.AsyncRequest(GenericCalculatorService::kSleepWithSidecarMethodName,
req, &resp, &controller,
boost::bind(SleepCallback, payload.get(), &latch));
// Sleep for a while before cancelling the RPC.
- SleepFor(MonoDelta::FromMicroseconds(rand.Uniform64(i * 500L + 1)));
+ if (i > 0) SleepFor(MonoDelta::FromMicroseconds(rand.Uniform64(i * 30)));
controller.Cancel();
latch.Wait();
- const Status &status = controller.status();
- ASSERT_TRUE(status.IsTimedOut() || status.IsAborted() || status.ok());
+ ASSERT_TRUE(controller.status().IsAborted() || controller.status().ok());
controller.Reset();
}
client_messenger->Shutdown();
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/rpc_header.proto
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc_header.proto b/src/kudu/rpc/rpc_header.proto
index a9f7396..1d55b6a 100644
--- a/src/kudu/rpc/rpc_header.proto
+++ b/src/kudu/rpc/rpc_header.proto
@@ -91,10 +91,6 @@ enum RpcFeatureFlag {
// This is currently used for loopback connections only, so that compute
// frameworks which schedule for locality don't pay encryption overhead.
TLS_AUTHENTICATION_ONLY = 3;
-
- // The RPC system is required to support parsing a footer in an incoming request.
- // See RequestFooterPB below.
- REQUEST_FOOTERS = 4;
};
// An authentication type. This is modeled as a oneof in case any of these
@@ -269,16 +265,6 @@ message RequestHeader {
repeated uint32 sidecar_offsets = 16;
}
-message RequestFooterPB {
- // This flag indicates whether the incoming request has been aborted by the client
- // and the request should be ignored.
- optional bool aborted = 1 [ default = false ];
-
- // Note that the footer is designed to be modified after the initial serialization
- // and it will be re-serialized after modification. To avoid unexpected change in
- // the total message size, keep to using fixed sized fields only in the footer.
-}
-
message ResponseHeader {
required int32 call_id = 1;
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/rtest.proto
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rtest.proto b/src/kudu/rpc/rtest.proto
index 257ca29..d212cef 100644
--- a/src/kudu/rpc/rtest.proto
+++ b/src/kudu/rpc/rtest.proto
@@ -58,7 +58,8 @@ message SleepResponsePB {
message SleepWithSidecarRequestPB {
required uint32 sleep_micros = 1;
required uint32 pattern = 2;
- repeated uint32 sidecar_idx = 3;
+ required uint32 num_repetitions = 3;
+ required uint32 sidecar_idx = 4;
}
message SleepWithSidecarResponsePB {
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/serialization.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/serialization.cc b/src/kudu/rpc/serialization.cc
index ac494d9..4103989 100644
--- a/src/kudu/rpc/serialization.cc
+++ b/src/kudu/rpc/serialization.cc
@@ -53,7 +53,7 @@ enum {
};
void SerializeMessage(const MessageLite& message, faststring* param_buf,
- int additional_size, bool use_cached_size) {
+ int additional_size, bool use_cached_size) {
int pb_size = use_cached_size ? message.GetCachedSize() : message.ByteSize();
DCHECK_EQ(message.ByteSize(), pb_size);
int recorded_size = pb_size + additional_size;
@@ -71,7 +71,7 @@ void SerializeMessage(const MessageLite& message, faststring* param_buf,
uint8_t* dst = param_buf->data();
dst = CodedOutputStream::WriteVarint32ToArray(recorded_size, dst);
dst = message.SerializeWithCachedSizesToArray(dst);
- DCHECK_EQ(dst, param_buf->data() + size_with_delim);
+ CHECK_EQ(dst, param_buf->data() + size_with_delim);
}
void SerializeHeader(const MessageLite& header,
@@ -101,32 +101,11 @@ void SerializeHeader(const MessageLite& header,
dst = header.SerializeWithCachedSizesToArray(dst);
// We should have used the whole buffer we allocated.
- DCHECK_EQ(dst, header_buf->data() + header_tot_len);
-}
-
-void SerializeFooter(const MessageLite& footer,
- faststring* footer_buf) {
- size_t footer_pb_len = footer.ByteSize();
- size_t footer_tot_len = footer_pb_len + CodedOutputStream::VarintSize32(footer_pb_len);
-
- footer_buf->resize(footer_tot_len);
- uint8_t* dst = footer_buf->data();
-
- dst = CodedOutputStream::WriteVarint32ToArray(footer_pb_len, dst);
- dst = footer.SerializeWithCachedSizesToArray(dst);
-
- DCHECK_EQ(dst, footer_buf->data() + footer_tot_len);
-}
-
-void IncrementMsgLength(size_t inc_len, faststring* header_buf) {
- uint8_t* dst = header_buf->data();
- uint32_t tot_len = NetworkByteOrder::Load32(dst);
- NetworkByteOrder::Store32(dst, tot_len + inc_len);
+ CHECK_EQ(dst, header_buf->data() + header_tot_len);
}
Status ParseMessage(const Slice& buf,
MessageLite* parsed_header,
- MessageLite* parsed_footer,
Slice* parsed_main_message) {
// First grab the total length
@@ -162,37 +141,20 @@ Status ParseMessage(const Slice& buf,
KUDU_REDACT(buf.ToDebugString()));
}
- int main_msg_offset = in.CurrentPosition();
- DCHECK_GT(main_msg_offset, 0);
if (PREDICT_FALSE(!in.Skip(main_msg_len))) {
return Status::Corruption(
StringPrintf("Invalid packet: data too short, expected %d byte main_msg", main_msg_len),
KUDU_REDACT(buf.ToDebugString()));
}
- if (parsed_footer != nullptr) {
- uint32_t footer_len = 0;
- if (PREDICT_FALSE(!in.ReadVarint32(&footer_len))) {
- return Status::Corruption("Invalid packet: missing footer delimiter",
- KUDU_REDACT(buf.ToDebugString()));
- }
-
- l = in.PushLimit(footer_len);
- if (PREDICT_FALSE(!parsed_footer->ParseFromCodedStream(&in))) {
- return Status::Corruption("Invalid packet: footer too short",
- KUDU_REDACT(buf.ToDebugString()));
- }
- in.PopLimit(l);
- }
-
- *parsed_main_message = Slice(buf.data() + main_msg_offset, main_msg_len);
-
if (PREDICT_FALSE(in.BytesUntilLimit() > 0)) {
return Status::Corruption(
- StringPrintf("Invalid packet: %d extra bytes at end of packet",
- in.BytesUntilLimit()), KUDU_REDACT(buf.ToDebugString()));
+ StringPrintf("Invalid packet: %d extra bytes at end of packet", in.BytesUntilLimit()),
+ KUDU_REDACT(buf.ToDebugString()));
}
+ *parsed_main_message = Slice(buf.data() + buf.size() - main_msg_len,
+ main_msg_len);
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/serialization.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/serialization.h b/src/kudu/rpc/serialization.h
index d82e378..8406a1f 100644
--- a/src/kudu/rpc/serialization.h
+++ b/src/kudu/rpc/serialization.h
@@ -65,28 +65,13 @@ void SerializeHeader(const google::protobuf::MessageLite& header,
size_t param_len,
faststring* header_buf);
-// Serialize the request footer into a buffer which is allocated by this function.
-// In: 'footer' Protobuf footer to serialize,
-// Out: 'footer_buf' faststring to be populated with the serialized bytes.
-void SerializeFooter(const google::protobuf::MessageLite& footer,
- faststring* footer_buf);
-
-// Increment the total message length in a header after it was serialized initially.
-// In: 'inc_len' bytes to increment the total message length by.
-// 'header_buf' is the serialized PB buffer returned from SerializeHeader().
-// Assumes that the first 4 bytes encode the total msg length in big endian order.
-void IncrementMsgLength(size_t inc_len,
- faststring* header_buf);
-
// Deserialize the request.
// In: data buffer Slice.
// Out: parsed_header PB initialized,
-// parsed_footer PB initialized if it's not NULL,
// parsed_main_message pointing to offset in original buffer containing
// the main payload.
Status ParseMessage(const Slice& buf,
google::protobuf::MessageLite* parsed_header,
- google::protobuf::MessageLite* parsed_footer,
Slice* parsed_main_message);
// Serialize the RPC connection header (magic number + flags).
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/server_negotiation.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/server_negotiation.cc b/src/kudu/rpc/server_negotiation.cc
index ca39ced..7ac21e0 100644
--- a/src/kudu/rpc/server_negotiation.cc
+++ b/src/kudu/rpc/server_negotiation.cc
@@ -504,9 +504,6 @@ Status ServerNegotiation::HandleNegotiate(const NegotiatePB& request) {
server_features_.insert(TLS_AUTHENTICATION_ONLY);
}
}
- if (ContainsKey(client_features_, REQUEST_FOOTERS)) {
- server_features_.insert(REQUEST_FOOTERS);
- }
for (RpcFeatureFlag feature : server_features_) {
response.add_supported_features(feature);
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/transfer.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/transfer.cc b/src/kudu/rpc/transfer.cc
index 84d8bc9..f9fd746 100644
--- a/src/kudu/rpc/transfer.cc
+++ b/src/kudu/rpc/transfer.cc
@@ -32,7 +32,6 @@
#include "kudu/rpc/outbound_call.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/logging.h"
-#include "kudu/util/memory/overwrite.h"
#include "kudu/util/net/socket.h"
DEFINE_int32(rpc_max_message_size, (50 * 1024 * 1024),
@@ -55,19 +54,10 @@ namespace kudu {
namespace rpc {
using std::ostringstream;
-using std::shared_ptr;
+using std::set;
using std::string;
using strings::Substitute;
-// The outbound transfer for an outbound call can be aborted after it has started already.
-// The system has to continue sending up to number of bytes specified in the header or the
-// receiving end will never make progress, leading to resource leak. In order to release
-// the sidecars early, we call MarkPayloadCancelled() on the outbound call to relocate the
-// sidecars to point to 'g_dummy_buffer'. SendBuffer() below recognizes slices pointing to
-// this dummy buffer and handles them differently.
-#define DUMMY_BUFFER_SIZE (1 << 20)
-uint8_t g_dummy_buffer[DUMMY_BUFFER_SIZE];
-
#define RETURN_ON_ERROR_OR_SOCKET_NOT_READY(status) \
if (PREDICT_FALSE(!status.ok())) { \
if (Socket::IsTemporarySocketError(status.posix_code())) { \
@@ -76,13 +66,6 @@ uint8_t g_dummy_buffer[DUMMY_BUFFER_SIZE];
return status; \
}
-// Initialize the dummy buffer with a known pattern for easier debugging.
-__attribute__((constructor))
-static void InitializeDummyBuffer() {
- OverwriteWithPattern(reinterpret_cast<char*>(g_dummy_buffer),
- DUMMY_BUFFER_SIZE, "ABORTED-TRANSFER");
-}
-
TransferCallbacks::~TransferCallbacks()
{}
@@ -170,30 +153,23 @@ OutboundTransfer::OutboundTransfer(int32_t call_id,
const TransferPayload &payload,
size_t n_payload_slices,
TransferCallbacks *callbacks)
- : n_payload_slices_(n_payload_slices),
- cur_slice_idx_(0),
+ : cur_slice_idx_(0),
cur_offset_in_slice_(0),
callbacks_(callbacks),
call_id_(call_id),
aborted_(false) {
- // We should leave the last entry for the footer.
- CHECK_LE(n_payload_slices_, payload_slices_.size() - 1);
+
+ n_payload_slices_ = n_payload_slices;
+ CHECK_LE(n_payload_slices_, payload_slices_.size());
for (int i = 0; i < n_payload_slices; i++) {
payload_slices_[i] = payload[i];
}
}
-void OutboundTransfer::AppendFooter(const shared_ptr<OutboundCall> &call) {
- DCHECK(!TransferStarted());
- DCHECK(is_for_outbound_call());
- DCHECK_LE(n_payload_slices_, payload_slices_.size() - 1);
- call->AppendFooter(&payload_slices_[n_payload_slices_++]);
-}
-
OutboundTransfer::~OutboundTransfer() {
if (!TransferFinished() && !aborted_) {
callbacks_->NotifyTransferAborted(
- Status::RuntimeError("RPC transfer destroyed before it finished sending"));
+ Status::RuntimeError("RPC transfer destroyed before it finished sending"));
}
}
@@ -204,57 +180,20 @@ void OutboundTransfer::Abort(const Status &status) {
aborted_ = true;
}
-void OutboundTransfer::Cancel(const shared_ptr<OutboundCall> &call,
- const Status &status) {
- // Only transfer for outbound calls support cancellation.
- DCHECK(is_for_outbound_call());
- // If transfer has finished already, it's a no-op.
- if (TransferFinished()) {
- return;
- }
- // If transfer hasn't started yet, simply abort it.
- if (!TransferStarted()) {
- Abort(status);
- return;
- }
- // Transfer has started already. Update the payload to give up the sidecars
- // if they haven't all been sent yet. We only reach this point if remote also
- // supports cancellation so the last slice is always the footer. Don't cancel
- // the transfer if only the footer is left.
- if (cur_slice_idx_ < n_payload_slices_ - 1) {
- call->MarkPayloadCancelled(&payload_slices_, g_dummy_buffer);
- status_ = status;
- }
-}
-
-#define IO_VEC_SIZE (16)
-
Status OutboundTransfer::SendBuffer(Socket &socket) {
- DCHECK_LT(cur_slice_idx_, n_payload_slices_);
-
- int idx = cur_slice_idx_;
- int offset_in_slice = cur_offset_in_slice_;
- struct iovec iovec[IO_VEC_SIZE];
- int n_iovecs;
- for (n_iovecs = 0; n_iovecs < IO_VEC_SIZE && idx < n_payload_slices_; ++n_iovecs) {
- Slice &slice = payload_slices_[idx];
- uint8_t* ptr = slice.mutable_data();
- if (PREDICT_TRUE(ptr != g_dummy_buffer)) {
- iovec[n_iovecs].iov_base = ptr + offset_in_slice;
- iovec[n_iovecs].iov_len = slice.size() - offset_in_slice;
- } else {
- iovec[n_iovecs].iov_base = g_dummy_buffer;
- iovec[n_iovecs].iov_len = slice.size() - offset_in_slice;
- // The dummy buffer is of limited size. Split a slice into multiple slices
- // if it's larger than DUMMY_BUFFER_SIZE.
- if (iovec[n_iovecs].iov_len > DUMMY_BUFFER_SIZE) {
- iovec[n_iovecs].iov_len = DUMMY_BUFFER_SIZE;
- offset_in_slice += DUMMY_BUFFER_SIZE;
- continue;
- }
+ CHECK_LT(cur_slice_idx_, n_payload_slices_);
+
+ int n_iovecs = n_payload_slices_ - cur_slice_idx_;
+ struct iovec iovec[n_iovecs];
+ {
+ int offset_in_slice = cur_offset_in_slice_;
+ for (int i = 0; i < n_iovecs; i++) {
+ Slice &slice = payload_slices_[cur_slice_idx_ + i];
+ iovec[i].iov_base = slice.mutable_data() + offset_in_slice;
+ iovec[i].iov_len = slice.size() - offset_in_slice;
+
+ offset_in_slice = 0;
}
- offset_in_slice = 0;
- ++idx;
}
int32_t written;
@@ -280,11 +219,7 @@ Status OutboundTransfer::SendBuffer(Socket &socket) {
}
if (cur_slice_idx_ == n_payload_slices_) {
- if (PREDICT_FALSE(!status_.ok())) {
- callbacks_->NotifyTransferAborted(status_);
- } else {
- callbacks_->NotifyTransferFinished();
- }
+ callbacks_->NotifyTransferFinished();
DCHECK_EQ(0, cur_offset_in_slice_);
} else {
DCHECK_LT(cur_slice_idx_, n_payload_slices_);
@@ -306,10 +241,6 @@ bool OutboundTransfer::TransferFinished() const {
return false;
}
-bool OutboundTransfer::TransferAborted() const {
- return aborted_;
-}
-
string OutboundTransfer::HexDump() const {
if (KUDU_SHOULD_REDACT()) {
return kRedactionMessage;
http://git-wip-us.apache.org/repos/asf/kudu/blob/55c40fc7/src/kudu/rpc/transfer.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/transfer.h b/src/kudu/rpc/transfer.h
index 0bbbc85..a9ecb6c 100644
--- a/src/kudu/rpc/transfer.h
+++ b/src/kudu/rpc/transfer.h
@@ -42,14 +42,13 @@ class Socket;
namespace rpc {
-class OutboundCall;
struct TransferCallbacks;
class TransferLimits {
public:
enum {
kMaxSidecars = 10,
- kMaxPayloadSlices = kMaxSidecars + 3 // (header + msg + footer)
+ kMaxPayloadSlices = kMaxSidecars + 2 // (header + msg)
};
DISALLOW_IMPLICIT_CONSTRUCTORS(TransferLimits);
@@ -138,17 +137,6 @@ class OutboundTransfer : public boost::intrusive::list_base_hook<> {
// This triggers TransferCallbacks::NotifyTransferAborted.
void Abort(const Status &status);
- // Cancel the current transfer. If the transfer hasn't started yet, call Abort()
- // with 'status'. If the transfer has started already, relinquish the sidecars
- // associated with 'call' and send the remaining bytes as dummy bytes. When the
- // transfer eventually finishes, NotifyTransferAborted() wll be called with 'status'.
- void Cancel(const std::shared_ptr<OutboundCall> &call, const Status &status);
-
- // Called right before the outbound transfer is sent for the first time to append
- // a footer to the entire message. Only valid if the transfer is for an outbound
- // call and the remote server has RPC feature flag "REQUEST_FOOTERS".
- void AppendFooter(const std::shared_ptr<OutboundCall> &call);
-
// send from our buffers into the sock
Status SendBuffer(Socket &socket);
@@ -158,9 +146,6 @@ class OutboundTransfer : public boost::intrusive::list_base_hook<> {
// Return true if the entire transfer has been sent.
bool TransferFinished() const;
- // Return true if the transfer has been aborted before completion.
- bool TransferAborted() const;
-
// Return the total number of bytes to be sent (including those already sent)
int32_t TotalLength() const;
@@ -199,12 +184,8 @@ class OutboundTransfer : public boost::intrusive::list_base_hook<> {
// In the case of call responses, kInvalidCallId
int32_t call_id_;
- // Set to true when the outbound transfer is aborted (e.g. time out, cancellation).
bool aborted_;
- // Status passed to NotifyTransferAborted() when the transfer is aborted.
- Status status_;
-
DISALLOW_COPY_AND_ASSIGN(OutboundTransfer);
};