mirror of https://github.com/grpc/grpc.git
[chaotic-good] Add client transport error handling. (#34611)
This is a follow-up PR of #34191, which handles the error condition of endpoints failed to write/read in chaotic-good client transport. This PR needs to be merged after #34191. <!-- If you know who should review your pull request, please assign it to that person, otherwise the pull request would get assigned randomly. If your pull request is for a specific language, please add the appropriate lang label. -->pull/34905/head
nanahpang
1 year ago
committed by
GitHub
parent
e58268525a
commit
1e15d00ec4
10 changed files with 669 additions and 41 deletions
@ -0,0 +1,441 @@ |
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed 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 "absl/status/status.h" |
||||
|
||||
#include "src/core/ext/transport/chaotic_good/client_transport.h" |
||||
#include "src/core/lib/transport/promise_endpoint.h" |
||||
#include "src/core/lib/transport/transport.h" |
||||
|
||||
// IWYU pragma: no_include <sys/socket.h>
|
||||
|
||||
#include <stddef.h> |
||||
|
||||
#include <algorithm> // IWYU pragma: keep |
||||
#include <memory> |
||||
#include <string> // IWYU pragma: keep |
||||
#include <tuple> |
||||
#include <utility> |
||||
#include <vector> // IWYU pragma: keep |
||||
|
||||
#include "absl/functional/any_invocable.h" |
||||
#include "absl/status/statusor.h" // IWYU pragma: keep |
||||
#include "absl/strings/str_format.h" // IWYU pragma: keep |
||||
#include "absl/types/optional.h" // IWYU pragma: keep |
||||
#include "gmock/gmock.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
#include <grpc/event_engine/event_engine.h> |
||||
#include <grpc/event_engine/memory_allocator.h> |
||||
#include <grpc/event_engine/slice.h> // IWYU pragma: keep |
||||
#include <grpc/event_engine/slice_buffer.h> |
||||
#include <grpc/grpc.h> |
||||
#include <grpc/status.h> // IWYU pragma: keep |
||||
|
||||
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
||||
#include "src/core/lib/iomgr/timer_manager.h" |
||||
#include "src/core/lib/promise/activity.h" |
||||
#include "src/core/lib/promise/event_engine_wakeup_scheduler.h" |
||||
#include "src/core/lib/promise/if.h" |
||||
#include "src/core/lib/promise/join.h" |
||||
#include "src/core/lib/promise/loop.h" |
||||
#include "src/core/lib/promise/pipe.h" |
||||
#include "src/core/lib/promise/seq.h" |
||||
#include "src/core/lib/resource_quota/arena.h" |
||||
#include "src/core/lib/resource_quota/memory_quota.h" |
||||
#include "src/core/lib/resource_quota/resource_quota.h" |
||||
#include "src/core/lib/slice/slice_buffer.h" |
||||
#include "src/core/lib/slice/slice_internal.h" // IWYU pragma: keep |
||||
#include "src/core/lib/transport/metadata_batch.h" // IWYU pragma: keep |
||||
#include "test/core/event_engine/fuzzing_event_engine/fuzzing_event_engine.h" |
||||
#include "test/core/event_engine/fuzzing_event_engine/fuzzing_event_engine.pb.h" |
||||
|
||||
using testing::MockFunction; |
||||
using testing::Return; |
||||
using testing::Sequence; |
||||
using testing::StrictMock; |
||||
using testing::WithArgs; |
||||
|
||||
namespace grpc_core { |
||||
namespace chaotic_good { |
||||
namespace testing { |
||||
|
||||
class MockEndpoint |
||||
: public grpc_event_engine::experimental::EventEngine::Endpoint { |
||||
public: |
||||
MOCK_METHOD( |
||||
bool, Read, |
||||
(absl::AnyInvocable<void(absl::Status)> on_read, |
||||
grpc_event_engine::experimental::SliceBuffer* buffer, |
||||
const grpc_event_engine::experimental::EventEngine::Endpoint::ReadArgs* |
||||
args), |
||||
(override)); |
||||
|
||||
MOCK_METHOD( |
||||
bool, Write, |
||||
(absl::AnyInvocable<void(absl::Status)> on_writable, |
||||
grpc_event_engine::experimental::SliceBuffer* data, |
||||
const grpc_event_engine::experimental::EventEngine::Endpoint::WriteArgs* |
||||
args), |
||||
(override)); |
||||
|
||||
MOCK_METHOD( |
||||
const grpc_event_engine::experimental::EventEngine::ResolvedAddress&, |
||||
GetPeerAddress, (), (const, override)); |
||||
MOCK_METHOD( |
||||
const grpc_event_engine::experimental::EventEngine::ResolvedAddress&, |
||||
GetLocalAddress, (), (const, override)); |
||||
}; |
||||
|
||||
class ClientTransportTest : public ::testing::Test { |
||||
public: |
||||
ClientTransportTest() |
||||
: control_endpoint_ptr_(new StrictMock<MockEndpoint>()), |
||||
data_endpoint_ptr_(new StrictMock<MockEndpoint>()), |
||||
memory_allocator_( |
||||
ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator( |
||||
"test")), |
||||
control_endpoint_(*control_endpoint_ptr_), |
||||
data_endpoint_(*data_endpoint_ptr_), |
||||
event_engine_(std::make_shared< |
||||
grpc_event_engine::experimental::FuzzingEventEngine>( |
||||
[]() { |
||||
grpc_timer_manager_set_threading(false); |
||||
grpc_event_engine::experimental::FuzzingEventEngine::Options |
||||
options; |
||||
return options; |
||||
}(), |
||||
fuzzing_event_engine::Actions())), |
||||
arena_(MakeScopedArena(initial_arena_size, &memory_allocator_)), |
||||
pipe_client_to_server_messages_(arena_.get()), |
||||
pipe_server_to_client_messages_(arena_.get()), |
||||
pipe_server_intial_metadata_(arena_.get()), |
||||
pipe_client_to_server_messages_second_(arena_.get()), |
||||
pipe_server_to_client_messages_second_(arena_.get()), |
||||
pipe_server_intial_metadata_second_(arena_.get()) {} |
||||
// Initial ClientTransport with read expecations
|
||||
void InitialClientTransport() { |
||||
client_transport_ = std::make_unique<ClientTransport>( |
||||
std::make_unique<PromiseEndpoint>( |
||||
std::unique_ptr<MockEndpoint>(control_endpoint_ptr_), |
||||
SliceBuffer()), |
||||
std::make_unique<PromiseEndpoint>( |
||||
std::unique_ptr<MockEndpoint>(data_endpoint_ptr_), SliceBuffer()), |
||||
event_engine_); |
||||
} |
||||
// Send messages from client to server.
|
||||
auto SendClientToServerMessages( |
||||
Pipe<MessageHandle>& pipe_client_to_server_messages, |
||||
int num_of_messages) { |
||||
return Loop([&pipe_client_to_server_messages, num_of_messages, |
||||
this]() mutable { |
||||
bool has_message = (num_of_messages > 0); |
||||
return If( |
||||
has_message, |
||||
Seq(pipe_client_to_server_messages.sender.Push( |
||||
arena_->MakePooled<Message>()), |
||||
[&num_of_messages]() -> LoopCtl<absl::Status> { |
||||
num_of_messages--; |
||||
return Continue(); |
||||
}), |
||||
[&pipe_client_to_server_messages]() mutable -> LoopCtl<absl::Status> { |
||||
pipe_client_to_server_messages.sender.Close(); |
||||
return absl::OkStatus(); |
||||
}); |
||||
}); |
||||
} |
||||
// Add stream into client transport, and expect return trailers of
|
||||
// "grpc-status:code".
|
||||
auto AddStream(CallArgs args) { |
||||
return client_transport_->AddStream(std::move(args)); |
||||
} |
||||
|
||||
private: |
||||
MockEndpoint* control_endpoint_ptr_; |
||||
MockEndpoint* data_endpoint_ptr_; |
||||
size_t initial_arena_size = 1024; |
||||
MemoryAllocator memory_allocator_; |
||||
|
||||
protected: |
||||
MockEndpoint& control_endpoint_; |
||||
MockEndpoint& data_endpoint_; |
||||
std::shared_ptr<grpc_event_engine::experimental::FuzzingEventEngine> |
||||
event_engine_; |
||||
std::unique_ptr<ClientTransport> client_transport_; |
||||
ScopedArenaPtr arena_; |
||||
Pipe<MessageHandle> pipe_client_to_server_messages_; |
||||
Pipe<MessageHandle> pipe_server_to_client_messages_; |
||||
Pipe<ServerMetadataHandle> pipe_server_intial_metadata_; |
||||
// Added for mutliple streams tests.
|
||||
Pipe<MessageHandle> pipe_client_to_server_messages_second_; |
||||
Pipe<MessageHandle> pipe_server_to_client_messages_second_; |
||||
Pipe<ServerMetadataHandle> pipe_server_intial_metadata_second_; |
||||
absl::AnyInvocable<void(absl::Status)> read_callback_; |
||||
Sequence control_endpoint_sequence_; |
||||
Sequence data_endpoint_sequence_; |
||||
// Added to verify received message payload.
|
||||
const std::string message_ = {0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}; |
||||
}; |
||||
|
||||
TEST_F(ClientTransportTest, AddOneStreamWithWriteFailed) { |
||||
// Mock write failed and read is pending.
|
||||
EXPECT_CALL(control_endpoint_, Write) |
||||
.WillOnce( |
||||
WithArgs<0>([](absl::AnyInvocable<void(absl::Status)> on_write) { |
||||
on_write(absl::InternalError("control endpoint write failed.")); |
||||
return false; |
||||
})); |
||||
EXPECT_CALL(data_endpoint_, Write) |
||||
.WillOnce( |
||||
WithArgs<0>([](absl::AnyInvocable<void(absl::Status)> on_write) { |
||||
on_write(absl::InternalError("data endpoint write failed.")); |
||||
return false; |
||||
})); |
||||
EXPECT_CALL(control_endpoint_, Read) |
||||
.InSequence(control_endpoint_sequence_) |
||||
.WillOnce(Return(false)); |
||||
InitialClientTransport(); |
||||
ClientMetadataHandle md; |
||||
auto args = CallArgs{std::move(md), |
||||
ClientInitialMetadataOutstandingToken::Empty(), |
||||
nullptr, |
||||
&pipe_server_intial_metadata_.sender, |
||||
&pipe_client_to_server_messages_.receiver, |
||||
&pipe_server_to_client_messages_.sender}; |
||||
StrictMock<MockFunction<void(absl::Status)>> on_done; |
||||
EXPECT_CALL(on_done, Call(absl::OkStatus())); |
||||
auto activity = MakeActivity( |
||||
Seq( |
||||
// Concurrently: write and read messages in client transport.
|
||||
Join( |
||||
// Add first stream with call_args into client transport.
|
||||
// Expect return trailers "grpc-status:unavailable".
|
||||
AddStream(std::move(args)), |
||||
// Send messages to call_args.client_to_server_messages pipe,
|
||||
// which will be eventually sent to control/data endpoints.
|
||||
SendClientToServerMessages(pipe_client_to_server_messages_, 1)), |
||||
// Once complete, verify successful sending and the received value.
|
||||
[](const std::tuple<ServerMetadataHandle, absl::Status>& ret) { |
||||
EXPECT_EQ(std::get<0>(ret)->get(GrpcStatusMetadata()).value(), |
||||
GRPC_STATUS_UNAVAILABLE); |
||||
EXPECT_TRUE(std::get<1>(ret).ok()); |
||||
return absl::OkStatus(); |
||||
}), |
||||
EventEngineWakeupScheduler(event_engine_), |
||||
[&on_done](absl::Status status) { on_done.Call(std::move(status)); }); |
||||
// Wait until ClientTransport's internal activities to finish.
|
||||
event_engine_->TickUntilIdle(); |
||||
event_engine_->UnsetGlobalHooks(); |
||||
} |
||||
|
||||
TEST_F(ClientTransportTest, AddOneStreamWithReadFailed) { |
||||
// Mock read failed.
|
||||
EXPECT_CALL(control_endpoint_, Read) |
||||
.InSequence(control_endpoint_sequence_) |
||||
.WillOnce(WithArgs<0>( |
||||
[](absl::AnyInvocable<void(absl::Status)> on_read) mutable { |
||||
on_read(absl::InternalError("control endpoint read failed.")); |
||||
// Return false to mock EventEngine read not finish.
|
||||
return false; |
||||
})); |
||||
InitialClientTransport(); |
||||
ClientMetadataHandle md; |
||||
auto args = CallArgs{std::move(md), |
||||
ClientInitialMetadataOutstandingToken::Empty(), |
||||
nullptr, |
||||
&pipe_server_intial_metadata_.sender, |
||||
&pipe_client_to_server_messages_.receiver, |
||||
&pipe_server_to_client_messages_.sender}; |
||||
StrictMock<MockFunction<void(absl::Status)>> on_done; |
||||
EXPECT_CALL(on_done, Call(absl::OkStatus())); |
||||
auto activity = MakeActivity( |
||||
Seq( |
||||
// Concurrently: write and read messages in client transport.
|
||||
Join( |
||||
// Add first stream with call_args into client transport.
|
||||
// Expect return trailers "grpc-status:unavailable".
|
||||
AddStream(std::move(args)), |
||||
// Send messages to call_args.client_to_server_messages pipe.
|
||||
SendClientToServerMessages(pipe_client_to_server_messages_, 1)), |
||||
// Once complete, verify successful sending and the received value.
|
||||
[](const std::tuple<ServerMetadataHandle, absl::Status>& ret) { |
||||
EXPECT_EQ(std::get<0>(ret)->get(GrpcStatusMetadata()).value(), |
||||
GRPC_STATUS_UNAVAILABLE); |
||||
EXPECT_TRUE(std::get<1>(ret).ok()); |
||||
return absl::OkStatus(); |
||||
}), |
||||
EventEngineWakeupScheduler(event_engine_), |
||||
[&on_done](absl::Status status) { on_done.Call(std::move(status)); }); |
||||
// Wait until ClientTransport's internal activities to finish.
|
||||
event_engine_->TickUntilIdle(); |
||||
event_engine_->UnsetGlobalHooks(); |
||||
} |
||||
|
||||
TEST_F(ClientTransportTest, AddMultipleStreamWithWriteFailed) { |
||||
// Mock write failed at first stream and second stream's write will fail too.
|
||||
EXPECT_CALL(control_endpoint_, Write) |
||||
.Times(1) |
||||
.WillRepeatedly( |
||||
WithArgs<0>([](absl::AnyInvocable<void(absl::Status)> on_write) { |
||||
on_write(absl::InternalError("control endpoint write failed.")); |
||||
return false; |
||||
})); |
||||
EXPECT_CALL(data_endpoint_, Write) |
||||
.Times(1) |
||||
.WillRepeatedly( |
||||
WithArgs<0>([](absl::AnyInvocable<void(absl::Status)> on_write) { |
||||
on_write(absl::InternalError("data endpoint write failed.")); |
||||
return false; |
||||
})); |
||||
EXPECT_CALL(control_endpoint_, Read) |
||||
.InSequence(control_endpoint_sequence_) |
||||
.WillOnce(Return(false)); |
||||
InitialClientTransport(); |
||||
ClientMetadataHandle first_stream_md; |
||||
ClientMetadataHandle second_stream_md; |
||||
auto first_stream_args = |
||||
CallArgs{std::move(first_stream_md), |
||||
ClientInitialMetadataOutstandingToken::Empty(), |
||||
nullptr, |
||||
&pipe_server_intial_metadata_.sender, |
||||
&pipe_client_to_server_messages_.receiver, |
||||
&pipe_server_to_client_messages_.sender}; |
||||
auto second_stream_args = |
||||
CallArgs{std::move(second_stream_md), |
||||
ClientInitialMetadataOutstandingToken::Empty(), |
||||
nullptr, |
||||
&pipe_server_intial_metadata_second_.sender, |
||||
&pipe_client_to_server_messages_second_.receiver, |
||||
&pipe_server_to_client_messages_second_.sender}; |
||||
StrictMock<MockFunction<void(absl::Status)>> on_done; |
||||
EXPECT_CALL(on_done, Call(absl::OkStatus())); |
||||
auto activity = MakeActivity( |
||||
Seq( |
||||
// Concurrently: write and read messages from client transport.
|
||||
Join( |
||||
// Add first stream with call_args into client transport.
|
||||
// Expect return trailers "grpc-status:unavailable".
|
||||
AddStream(std::move(first_stream_args)), |
||||
// Send messages to first stream's
|
||||
// call_args.client_to_server_messages pipe.
|
||||
SendClientToServerMessages(pipe_client_to_server_messages_, 1)), |
||||
// Once complete, verify successful sending and the received value.
|
||||
[](const std::tuple<ServerMetadataHandle, absl::Status>& ret) { |
||||
EXPECT_EQ(std::get<0>(ret)->get(GrpcStatusMetadata()).value(), |
||||
GRPC_STATUS_UNAVAILABLE); |
||||
EXPECT_TRUE(std::get<1>(ret).ok()); |
||||
return absl::OkStatus(); |
||||
}, |
||||
Join( |
||||
// Add second stream with call_args into client transport.
|
||||
// Expect return trailers "grpc-status:unavailable".
|
||||
AddStream(std::move(second_stream_args)), |
||||
// Send messages to second stream's
|
||||
// call_args.client_to_server_messages pipe.
|
||||
SendClientToServerMessages(pipe_client_to_server_messages_second_, |
||||
1)), |
||||
// Once complete, verify successful sending and the received value.
|
||||
[](const std::tuple<ServerMetadataHandle, absl::Status>& ret) { |
||||
EXPECT_EQ(std::get<0>(ret)->get(GrpcStatusMetadata()).value(), |
||||
GRPC_STATUS_UNAVAILABLE); |
||||
EXPECT_TRUE(std::get<1>(ret).ok()); |
||||
return absl::OkStatus(); |
||||
}), |
||||
EventEngineWakeupScheduler(event_engine_), |
||||
[&on_done](absl::Status status) { on_done.Call(std::move(status)); }); |
||||
// Wait until ClientTransport's internal activities to finish.
|
||||
event_engine_->TickUntilIdle(); |
||||
event_engine_->UnsetGlobalHooks(); |
||||
} |
||||
|
||||
TEST_F(ClientTransportTest, AddMultipleStreamWithReadFailed) { |
||||
// Mock read failed at first stream, and second stream's write will fail too.
|
||||
EXPECT_CALL(control_endpoint_, Read) |
||||
.InSequence(control_endpoint_sequence_) |
||||
.WillOnce(WithArgs<0>( |
||||
[](absl::AnyInvocable<void(absl::Status)> on_read) mutable { |
||||
on_read(absl::InternalError("control endpoint read failed.")); |
||||
// Return false to mock EventEngine read not finish.
|
||||
return false; |
||||
})); |
||||
InitialClientTransport(); |
||||
ClientMetadataHandle first_stream_md; |
||||
ClientMetadataHandle second_stream_md; |
||||
auto first_stream_args = |
||||
CallArgs{std::move(first_stream_md), |
||||
ClientInitialMetadataOutstandingToken::Empty(), |
||||
nullptr, |
||||
&pipe_server_intial_metadata_.sender, |
||||
&pipe_client_to_server_messages_.receiver, |
||||
&pipe_server_to_client_messages_.sender}; |
||||
auto second_stream_args = |
||||
CallArgs{std::move(second_stream_md), |
||||
ClientInitialMetadataOutstandingToken::Empty(), |
||||
nullptr, |
||||
&pipe_server_intial_metadata_second_.sender, |
||||
&pipe_client_to_server_messages_second_.receiver, |
||||
&pipe_server_to_client_messages_second_.sender}; |
||||
StrictMock<MockFunction<void(absl::Status)>> on_done; |
||||
EXPECT_CALL(on_done, Call(absl::OkStatus())); |
||||
auto activity = MakeActivity( |
||||
Seq( |
||||
// Concurrently: write and read messages from client transport.
|
||||
Join( |
||||
// Add first stream with call_args into client transport.
|
||||
AddStream(std::move(first_stream_args)), |
||||
// Send messages to first stream's
|
||||
// call_args.client_to_server_messages pipe, which will be
|
||||
// eventually sent to control/data endpoints.
|
||||
SendClientToServerMessages(pipe_client_to_server_messages_, 1)), |
||||
// Once complete, verify successful sending and the received value.
|
||||
[](const std::tuple<ServerMetadataHandle, absl::Status>& ret) { |
||||
EXPECT_EQ(std::get<0>(ret)->get(GrpcStatusMetadata()).value(), |
||||
GRPC_STATUS_UNAVAILABLE); |
||||
EXPECT_TRUE(std::get<1>(ret).ok()); |
||||
return absl::OkStatus(); |
||||
}, |
||||
Join( |
||||
// Add second stream with call_args into client transport.
|
||||
AddStream(std::move(second_stream_args)), |
||||
// Send messages to second stream's
|
||||
// call_args.client_to_server_messages pipe, which will be
|
||||
// eventually sent to control/data endpoints.
|
||||
SendClientToServerMessages(pipe_client_to_server_messages_second_, |
||||
1)), |
||||
// Once complete, verify successful sending and the received value.
|
||||
[](const std::tuple<ServerMetadataHandle, absl::Status>& ret) { |
||||
EXPECT_EQ(std::get<0>(ret)->get(GrpcStatusMetadata()).value(), |
||||
GRPC_STATUS_UNAVAILABLE); |
||||
EXPECT_TRUE(std::get<1>(ret).ok()); |
||||
return absl::OkStatus(); |
||||
}), |
||||
EventEngineWakeupScheduler(event_engine_), |
||||
[&on_done](absl::Status status) { on_done.Call(std::move(status)); }); |
||||
// Wait until ClientTransport's internal activities to finish.
|
||||
event_engine_->TickUntilIdle(); |
||||
event_engine_->UnsetGlobalHooks(); |
||||
} |
||||
|
||||
} // namespace testing
|
||||
} // namespace chaotic_good
|
||||
} // namespace grpc_core
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
// Must call to create default EventEngine.
|
||||
grpc_init(); |
||||
int ret = RUN_ALL_TESTS(); |
||||
grpc_shutdown(); |
||||
return ret; |
||||
} |
||||
Loading…
Reference in new issue