diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc
index 8800a87345085f520390086ebc0b24f89b6fc7d7..0deb2eff95bbb39786c2f562c7e0c3b5def91b3e 100644
--- a/test/cpp/end2end/async_end2end_test.cc
+++ b/test/cpp/end2end/async_end2end_test.cc
@@ -775,6 +775,8 @@ TEST_P(AsyncEnd2endTest, UnimplementedRpc) {
EXPECT_EQ("", recv_status.error_message());
}
+// This class is for testing scenarios where RPCs are cancelled on the server
+// by calling ServerContext::TryCancel()
class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
protected:
typedef enum {
@@ -791,6 +793,18 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
EXPECT_TRUE(context->IsCancelled());
}
+ // Helper for testing client-streaming RPCs which are cancelled on the server.
+ // Depending on the value of server_try_cancel parameter, this will test one
+ // of the following three scenarios:
+ // CANCEL_BEFORE_PROCESSING: Rpc is cancelled by the server before reading
+ // any messages from the client
+ //
+ // CANCEL_DURING_PROCESSING: Rpc is cancelled by the server while reading
+ // messages from the client
+ //
+ // CANCEL_AFTER PROCESSING: Rpc is cancelled by server after reading all
+ // messages from the client (but before sending any status back to the
+ // client)
void TestClientStreamingServerCancel(
ServerTryCancelRequestPhase server_try_cancel) {
ResetStub();
@@ -864,22 +878,37 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
ServerTryCancel(&srv_ctx);
}
- // Note: The RPC has been cancelled at this point for sure. So, from this
- // point forward, we know that cq results are supposed to return false on
- // server.
+ // The RPC has been cancelled at this point for sure (i.e irrespective of
+ // the value of `server_try_cancel` is). So, from this point forward, we
+ // know that cq results are supposed to return false on server.
+ // Server sends the final message and cancelled status (but the RPC is
+ // already cancelled at this point. So we expect the operation to fail)
send_response.set_message("Pong");
srv_stream.Finish(send_response, Status::CANCELLED, tag(9));
Verifier(GetParam()).Expect(9, false).Verify(cq_.get());
+ // Client will see the cancellation
cli_stream->Finish(&recv_status, tag(10));
- // TODO: sreek: The expectation here should be true. This seems like a bug.
- // Investigating
+ // TODO: sreek: The expectation here should be true. This is a bug (github
+ // issue #4972)
Verifier(GetParam()).Expect(10, false).Verify(cq_.get());
EXPECT_FALSE(recv_status.ok());
EXPECT_EQ(::grpc::StatusCode::CANCELLED, recv_status.error_code());
}
+ // Helper for testing server-streaming RPCs which are cancelled on the server.
+ // Depending on the value of server_try_cancel parameter, this will test one
+ // of the following three scenarios:
+ // CANCEL_BEFORE_PROCESSING: Rpc is cancelled by the server before sending
+ // any messages to the client
+ //
+ // CANCEL_DURING_PROCESSING: Rpc is cancelled by the server while sending
+ // messages to the client
+ //
+ // CANCEL_AFTER PROCESSING: Rpc is cancelled by server after sending all
+ // messages to the client (but before sending any status back to the
+ // client)
void TestServerStreamingServerCancel(
ServerTryCancelRequestPhase server_try_cancel) {
ResetStub();
@@ -898,7 +927,6 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
std::unique_ptr<ClientAsyncReader<EchoResponse>> cli_stream(
stub_->AsyncResponseStream(&cli_ctx, send_request, cq_.get(), tag(1)));
Verifier(GetParam()).Expect(1, true).Verify(cq_.get());
-
// On the server, request to be notified of 'ResponseStream' calls and
// receive the call just made by the client
service_.RequestResponseStream(&srv_ctx, &recv_request, &srv_stream,
@@ -924,8 +952,8 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
// Server will cancel the RPC in a parallel thread while writing responses
// to the client. Since the cancellation can happen at anytime, some of
- // the cq results (i.e those until cancellation) might be true but
- // its non deterministic. So better to ignore the cq results
+ // the cq results (i.e those until cancellation) might be true but it is
+ // non deterministic. So better to ignore the cq results
ignore_cq_result = true;
}
@@ -938,11 +966,16 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
.Verify(cq_.get(), ignore_cq_result);
}
+ if (server_try_cancel_thd != NULL) {
+ server_try_cancel_thd->join();
+ delete server_try_cancel_thd;
+ }
+
if (server_try_cancel == CANCEL_AFTER_PROCESSING) {
ServerTryCancel(&srv_ctx);
}
- // Client attemts to read the three messages
+ // Client attemts to read the three messages from the server
for (int tag_idx = 6; tag_idx <= 8; tag_idx++) {
cli_stream->Read(&recv_response, tag(tag_idx));
Verifier(GetParam())
@@ -950,25 +983,35 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
.Verify(cq_.get(), ignore_cq_result);
}
- if (server_try_cancel_thd != NULL) {
- server_try_cancel_thd->join();
- delete server_try_cancel_thd;
- }
-
- // Note: At this point, we know that server has cancelled the request for
- // sure.
+ // The RPC has been cancelled at this point for sure (i.e irrespective of
+ // the value of `server_try_cancel` is). So, from this point forward, we
+ // know that cq results are supposed to return false on server.
- // Server finishes the stream
+ // Server finishes the stream (but the RPC is already cancelled)
srv_stream.Finish(Status::CANCELLED, tag(9));
Verifier(GetParam()).Expect(9, false).Verify(cq_.get());
- // Client receives the cancellation
+ // Client will see the cancellation
cli_stream->Finish(&recv_status, tag(10));
Verifier(GetParam()).Expect(10, true).Verify(cq_.get());
EXPECT_FALSE(recv_status.ok());
EXPECT_EQ(::grpc::StatusCode::CANCELLED, recv_status.error_code());
}
+ // Helper for testing bidirectinal-streaming RPCs which are cancelled on the
+ // server.
+ //
+ // Depending on the value of server_try_cancel parameter, this will
+ // test one of the following three scenarios:
+ // CANCEL_BEFORE_PROCESSING: Rpc is cancelled by the server before reading/
+ // writing any messages from/to the client
+ //
+ // CANCEL_DURING_PROCESSING: Rpc is cancelled by the server while reading
+ // messages from the client
+ //
+ // CANCEL_AFTER PROCESSING: Rpc is cancelled by server after reading all
+ // messages from the client (but before sending any status back to the
+ // client)
void TestBidiStreamingServerCancel(
ServerTryCancelRequestPhase server_try_cancel) {
ResetStub();
@@ -993,12 +1036,14 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
tag(2));
Verifier(GetParam()).Expect(2, true).Verify(cq_.get());
+ // Client sends the first and the only message
send_request.set_message("Ping");
cli_stream->Write(send_request, tag(3));
Verifier(GetParam()).Expect(3, true).Verify(cq_.get());
bool expected_cq_result = true;
bool ignore_cq_result = false;
+
if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
ServerTryCancel(&srv_ctx);
@@ -1038,8 +1083,10 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
cli_stream->WritesDone(tag(7));
Verifier(GetParam()).Expect(7, true).Verify(cq_.get());
- // This is expected to fail in all cases (Either there are no more msgs from
- // the client or the RPC is cancelled on the server)
+ // This is expected to fail in all cases i.e for all values of
+ // server_try_cancel. This is becasue at this point, either there are no
+ // more msgs from the client (because client called WritesDone) or the RPC
+ // is cancelled on the server
srv_stream.Read(&recv_request, tag(8));
Verifier(GetParam()).Expect(8, false).Verify(cq_.get());
@@ -1052,7 +1099,9 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
ServerTryCancel(&srv_ctx);
}
- // At this point, we know that the server cancelled the request for sure
+ // The RPC has been cancelled at this point for sure (i.e irrespective of
+ // the value of `server_try_cancel` is). So, from this point forward, we
+ // know that cq results are supposed to return false on server.
srv_stream.Finish(Status::CANCELLED, tag(9));
Verifier(GetParam()).Expect(9, false).Verify(cq_.get());
diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc
index 5fae056b63b1c364fc8aef4aa3d5fef4015aabf4..107e46f4381e79310bf820c5897926049487ad27 100644
--- a/test/cpp/end2end/end2end_test.cc
+++ b/test/cpp/end2end/end2end_test.cc
@@ -251,8 +251,10 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
int server_try_cancel = GetIntValueFromMetadata(
kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
if (server_try_cancel > DO_NOT_CANCEL) {
- // For unary RPC, the actual value of server_try_cancel does not matter
- // (as long as it is greater than DO_NOT_CANCEL)
+ // Since this is a unary RPC, by the time this server handler is called,
+ // the 'request' message is already read from the client. So the scenarios
+ // in server_try_cancel don't make much sense. Just cancel the RPC as long
+ // as server_try_cancel is not DO_NOT_CANCEL
ServerTryCancel(context);
return Status::CANCELLED;
}
@@ -318,13 +320,27 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
Status RequestStream(ServerContext* context,
ServerReader<EchoRequest>* reader,
EchoResponse* response) GRPC_OVERRIDE {
- EchoRequest request;
- response->set_message("");
- int cancel_after_reads = GetIntValueFromMetadata(
- kServerCancelAfterReads, context->client_metadata(), 0);
+ // If 'server_try_cancel' is set in the metadata, the RPC is cancelled by
+ // the server by calling ServerContext::TryCancel() depending on the value:
+ // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server reads
+ // any message from the client
+ // CANCEL_DURING_PROCESSING: The RPC is cancelled while the server is
+ // reading messages from the client
+ // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server reads
+ // all the messages from the client
int server_try_cancel = GetIntValueFromMetadata(
kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
+ // If 'cancel_after_reads' is set in the metadata AND non-zero, the server
+ // will cancel the RPC (by just returning Status::CANCELLED - doesn't call
+ // ServerContext::TryCancel()) after reading the number of records specified
+ // by the 'cancel_after_reads' value set in the metadata.
+ int cancel_after_reads = GetIntValueFromMetadata(
+ kServerCancelAfterReads, context->client_metadata(), 0);
+
+ EchoRequest request;
+ response->set_message("");
+
if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
ServerTryCancel(context);
return Status::CANCELLED;
@@ -367,6 +383,14 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
// TODO(yangg) make it generic by adding a parameter into EchoRequest
Status ResponseStream(ServerContext* context, const EchoRequest* request,
ServerWriter<EchoResponse>* writer) GRPC_OVERRIDE {
+ // If server_try_cancel is set in the metadata, the RPC is cancelled by the
+ // server by calling ServerContext::TryCancel() depending on the value:
+ // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server writes
+ // any messages to the client
+ // CANCEL_DURING_PROCESSING: The RPC is cancelled while the server is
+ // writing messages to the client
+ // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server writes
+ // all the messages to the client
int server_try_cancel = GetIntValueFromMetadata(
kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
@@ -404,12 +428,20 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
Status BidiStream(ServerContext* context,
ServerReaderWriter<EchoResponse, EchoRequest>* stream)
GRPC_OVERRIDE {
- EchoRequest request;
- EchoResponse response;
-
+ // If server_try_cancel is set in the metadata, the RPC is cancelled by the
+ // server by calling ServerContext::TryCancel() depending on the value:
+ // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server reads/
+ // writes any messages from/to the client
+ // CANCEL_DURING_PROCESSING: The RPC is cancelled while the server is
+ // reading/writing messages from/to the client
+ // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server
+ // reads/writes all messages from/to the client
int server_try_cancel = GetIntValueFromMetadata(
kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL);
+ EchoRequest request;
+ EchoResponse response;
+
if (server_try_cancel == CANCEL_BEFORE_PROCESSING) {
ServerTryCancel(context);
return Status::CANCELLED;
@@ -574,11 +606,23 @@ static void SendRpc(grpc::testing::EchoTestService::Stub* stub, int num_rpcs) {
}
}
-// == Tests for cancelling RPC from server side ==
-
+// This class is for testing scenarios where RPCs are cancelled on the server
+// by calling ServerContext::TryCancel()
class End2endServerTryCancelTest : public End2endTest {
protected:
- // Tests for Client streaming
+ // Helper for testing client-streaming RPCs which are cancelled on the server.
+ // Depending on the value of server_try_cancel parameter, this will test one
+ // of the following three scenarios:
+ // CANCEL_BEFORE_PROCESSING: Rpc is cancelled by the server before reading
+ // any messages from the client
+ //
+ // CANCEL_DURING_PROCESSING: Rpc is cancelled by the server while reading
+ // messages from the client
+ //
+ // CANCEL_AFTER PROCESSING: Rpc is cancelled by server after reading all
+ // the messages from the client
+ //
+ // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL.
void TestRequestStreamServerCancel(
ServerTryCancelRequestPhase server_try_cancel, int num_msgs_to_send) {
ResetStub();
@@ -586,10 +630,12 @@ class End2endServerTryCancelTest : public End2endTest {
EchoResponse response;
ClientContext context;
+ // Send server_try_cancel value in the client metadata
context.AddMetadata(kServerTryCancelRequest,
std::to_string(server_try_cancel));
auto stream = stub_->RequestStream(&context, &response);
+
int num_msgs_sent = 0;
while (num_msgs_sent < num_msgs_to_send) {
request.set_message("hello");
@@ -599,17 +645,32 @@ class End2endServerTryCancelTest : public End2endTest {
num_msgs_sent++;
}
gpr_log(GPR_INFO, "Sent %d messages", num_msgs_sent);
+
stream->WritesDone();
Status s = stream->Finish();
+ // At this point, we know for sure that RPC was cancelled by the server
+ // since we passed server_try_cancel value in the metadata. Depending on the
+ // value of server_try_cancel, the RPC might have been cancelled by the
+ // server at different stages. The following validates our expectations of
+ // number of messages sent in various cancellation scenarios:
+
switch (server_try_cancel) {
case CANCEL_BEFORE_PROCESSING:
case CANCEL_DURING_PROCESSING:
+ // If the RPC is cancelled by server before / during messages from the
+ // client, it means that the client most likely did not get a chance to
+ // send all the messages it wanted to send. i.e num_msgs_sent <=
+ // num_msgs_to_send
EXPECT_LE(num_msgs_sent, num_msgs_to_send);
break;
+
case CANCEL_AFTER_PROCESSING:
+ // If the RPC was cancelled after all messages were read by the server,
+ // the client did get a chance to send all its messages
EXPECT_EQ(num_msgs_sent, num_msgs_to_send);
break;
+
default:
gpr_log(GPR_ERROR, "Invalid server_try_cancel value: %d",
server_try_cancel);
@@ -622,7 +683,19 @@ class End2endServerTryCancelTest : public End2endTest {
EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
}
- // Test for server streaming
+ // Helper for testing server-streaming RPCs which are cancelled on the server.
+ // Depending on the value of server_try_cancel parameter, this will test one
+ // of the following three scenarios:
+ // CANCEL_BEFORE_PROCESSING: Rpc is cancelled by the server before writing
+ // any messages to the client
+ //
+ // CANCEL_DURING_PROCESSING: Rpc is cancelled by the server while writing
+ // messages to the client
+ //
+ // CANCEL_AFTER PROCESSING: Rpc is cancelled by server after writing all
+ // the messages to the client
+ //
+ // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL.
void TestResponseStreamServerCancel(
ServerTryCancelRequestPhase server_try_cancel) {
ResetStub();
@@ -630,8 +703,10 @@ class End2endServerTryCancelTest : public End2endTest {
EchoResponse response;
ClientContext context;
+ // Send server_try_cancel in the client metadata
context.AddMetadata(kServerTryCancelRequest,
std::to_string(server_try_cancel));
+
request.set_message("hello");
auto stream = stub_->ResponseStream(&context, request);
@@ -648,18 +723,29 @@ class End2endServerTryCancelTest : public End2endTest {
Status s = stream->Finish();
+ // Depending on the value of server_try_cancel, the RPC might have been
+ // cancelled by the server at different stages. The following validates our
+ // expectations of number of messages read in various cancellation
+ // scenarios:
switch (server_try_cancel) {
- case CANCEL_BEFORE_PROCESSING: {
+ case CANCEL_BEFORE_PROCESSING:
+ // Server cancelled before sending any messages. Which means the client
+ // wouldn't have read any
EXPECT_EQ(num_msgs_read, 0);
break;
- }
- case CANCEL_DURING_PROCESSING: {
+
+ case CANCEL_DURING_PROCESSING:
+ // Server cancelled while writing messages. Client must have read less
+ // than or equal to the expected number of messages
EXPECT_LE(num_msgs_read, kNumResponseStreamsMsgs);
break;
- }
- case CANCEL_AFTER_PROCESSING: {
+
+ case CANCEL_AFTER_PROCESSING:
+ // Server cancelled after writing all messages. Client must have read
+ // all messages
EXPECT_EQ(num_msgs_read, kNumResponseStreamsMsgs);
- }
+ break;
+
default: {
gpr_log(GPR_ERROR, "Invalid server_try_cancel value: %d",
server_try_cancel);
@@ -673,6 +759,19 @@ class End2endServerTryCancelTest : public End2endTest {
EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
}
+ // Helper for testing bidirectional-streaming RPCs which are cancelled on the
+ // server. Depending on the value of server_try_cancel parameter, this will
+ // test one of the following three scenarios:
+ // CANCEL_BEFORE_PROCESSING: Rpc is cancelled by the server before reading/
+ // writing any messages from/to the client
+ //
+ // CANCEL_DURING_PROCESSING: Rpc is cancelled by the server while reading/
+ // writing messages from/to the client
+ //
+ // CANCEL_AFTER PROCESSING: Rpc is cancelled by server after reading/writing
+ // all the messages from/to the client
+ //
+ // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL.
void TestBidiStreamServerCancel(ServerTryCancelRequestPhase server_try_cancel,
int num_messages) {
ResetStub();
@@ -680,6 +779,7 @@ class End2endServerTryCancelTest : public End2endTest {
EchoResponse response;
ClientContext context;
+ // Send server_try_cancel in the client metadata
context.AddMetadata(kServerTryCancelRequest,
std::to_string(server_try_cancel));
@@ -707,27 +807,31 @@ class End2endServerTryCancelTest : public End2endTest {
stream->WritesDone();
Status s = stream->Finish();
+ // Depending on the value of server_try_cancel, the RPC might have been
+ // cancelled by the server at different stages. The following validates our
+ // expectations of number of messages read in various cancellation
+ // scenarios:
switch (server_try_cancel) {
- case CANCEL_BEFORE_PROCESSING: {
+ case CANCEL_BEFORE_PROCESSING:
EXPECT_EQ(num_msgs_read, 0);
break;
- }
- case CANCEL_DURING_PROCESSING: {
+
+ case CANCEL_DURING_PROCESSING:
EXPECT_LE(num_msgs_sent, num_messages);
EXPECT_LE(num_msgs_read, num_msgs_sent);
break;
- }
- case CANCEL_AFTER_PROCESSING: {
+
+ case CANCEL_AFTER_PROCESSING:
EXPECT_EQ(num_msgs_sent, num_messages);
EXPECT_EQ(num_msgs_read, num_msgs_sent);
- }
- default: {
+ break;
+
+ default:
gpr_log(GPR_ERROR, "Invalid server_try_cancel value: %d",
server_try_cancel);
EXPECT_TRUE(server_try_cancel > DO_NOT_CANCEL &&
server_try_cancel <= CANCEL_AFTER_PROCESSING);
break;
- }
}
EXPECT_FALSE(s.ok());