[call-v3] Direct channel implementation (#36734)
This change brings up the direct channel, and inproc promise based transports.
This work exposed a bug that was very difficult to fix with the current call_filters.cc implementation, so I've substantially revamped that - instead of having a pipe-like object per call element, we now have a big ol' combined state machine for the entire call. It's a touch more code, but substantially easier to reason about individual cases, so I much prefer this form (it's also a slight memory improvement: 12 bytes total to track call state, and 10 of those are wakeup bitmasks...).
Closes #36734
COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/36734 from ctiller:transport-refs-9 3e2a80b40d
PiperOrigin-RevId: 644034593
pull/36952/head
Craig Tiller
9 months ago
committed by
Copybara-Service
parent
426df93434
commit
e21467475f
59 changed files with 2190 additions and 1024 deletions
@ -0,0 +1,80 @@ |
||||
// Copyright 2024 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 "src/core/client_channel/direct_channel.h" |
||||
|
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "src/core/lib/gprpp/orphanable.h" |
||||
#include "src/core/lib/surface/channel_stack_type.h" |
||||
#include "src/core/lib/surface/client_call.h" |
||||
#include "src/core/lib/transport/interception_chain.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
absl::StatusOr<RefCountedPtr<DirectChannel>> DirectChannel::Create( |
||||
std::string target, const ChannelArgs& args) { |
||||
auto* transport = args.GetObject<Transport>(); |
||||
if (transport == nullptr) { |
||||
return absl::InvalidArgumentError("Transport not set in ChannelArgs"); |
||||
} |
||||
if (transport->client_transport() == nullptr) { |
||||
return absl::InvalidArgumentError("Transport is not a client transport"); |
||||
} |
||||
auto transport_call_destination = MakeRefCounted<TransportCallDestination>( |
||||
OrphanablePtr<ClientTransport>(transport->client_transport())); |
||||
auto event_engine = |
||||
args.GetObjectRef<grpc_event_engine::experimental::EventEngine>(); |
||||
if (event_engine == nullptr) { |
||||
return absl::InvalidArgumentError("EventEngine not set in ChannelArgs"); |
||||
} |
||||
InterceptionChainBuilder builder(args); |
||||
CoreConfiguration::Get().channel_init().AddToInterceptionChainBuilder( |
||||
GRPC_CLIENT_DIRECT_CHANNEL, builder); |
||||
auto interception_chain = builder.Build(transport_call_destination); |
||||
if (!interception_chain.ok()) return interception_chain.status(); |
||||
return MakeRefCounted<DirectChannel>( |
||||
std::move(target), args, std::move(event_engine), |
||||
std::move(transport_call_destination), std::move(*interception_chain)); |
||||
} |
||||
|
||||
void DirectChannel::Orphaned() { |
||||
transport_call_destination_.reset(); |
||||
interception_chain_.reset(); |
||||
} |
||||
|
||||
void DirectChannel::StartCall(UnstartedCallHandler unstarted_handler) { |
||||
unstarted_handler.SpawnInfallible( |
||||
"start", |
||||
[interception_chain = interception_chain_, unstarted_handler]() mutable { |
||||
interception_chain->StartCall(std::move(unstarted_handler)); |
||||
return []() { return Empty{}; }; |
||||
}); |
||||
} |
||||
|
||||
void DirectChannel::GetInfo(const grpc_channel_info*) { |
||||
// TODO(roth): Implement this.
|
||||
} |
||||
|
||||
grpc_call* DirectChannel::CreateCall( |
||||
grpc_call* parent_call, uint32_t propagation_mask, |
||||
grpc_completion_queue* cq, grpc_pollset_set* /*pollset_set_alternative*/, |
||||
Slice path, absl::optional<Slice> authority, Timestamp deadline, |
||||
bool /*registered_method*/) { |
||||
return MakeClientCall(parent_call, propagation_mask, cq, std::move(path), |
||||
std::move(authority), false, deadline, |
||||
compression_options(), event_engine_.get(), |
||||
call_arena_allocator()->MakeArena(), Ref()); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
@ -0,0 +1,101 @@ |
||||
// Copyright 2024 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.
|
||||
|
||||
#ifndef GRPC_SRC_CORE_CLIENT_CHANNEL_DIRECT_CHANNEL_H |
||||
#define GRPC_SRC_CORE_CLIENT_CHANNEL_DIRECT_CHANNEL_H |
||||
|
||||
#include <memory> |
||||
|
||||
#include "src/core/lib/surface/channel.h" |
||||
#include "src/core/lib/transport/transport.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
class DirectChannel final : public Channel { |
||||
public: |
||||
class TransportCallDestination final : public CallDestination { |
||||
public: |
||||
explicit TransportCallDestination(OrphanablePtr<ClientTransport> transport) |
||||
: transport_(std::move(transport)) {} |
||||
|
||||
ClientTransport* transport() { return transport_.get(); } |
||||
|
||||
void HandleCall(CallHandler handler) override { |
||||
transport_->StartCall(std::move(handler)); |
||||
} |
||||
|
||||
void Orphaned() override { transport_.reset(); } |
||||
|
||||
private: |
||||
OrphanablePtr<ClientTransport> transport_; |
||||
}; |
||||
|
||||
static absl::StatusOr<RefCountedPtr<DirectChannel>> Create( |
||||
std::string target, const ChannelArgs& args); |
||||
|
||||
DirectChannel( |
||||
std::string target, const ChannelArgs& args, |
||||
std::shared_ptr<grpc_event_engine::experimental::EventEngine> |
||||
event_engine, |
||||
RefCountedPtr<TransportCallDestination> transport_call_destination, |
||||
RefCountedPtr<UnstartedCallDestination> interception_chain) |
||||
: Channel(std::move(target), args), |
||||
transport_call_destination_(std::move(transport_call_destination)), |
||||
interception_chain_(std::move(interception_chain)), |
||||
event_engine_(std::move(event_engine)) {} |
||||
|
||||
void Orphaned() override; |
||||
void StartCall(UnstartedCallHandler unstarted_handler) override; |
||||
bool IsLame() const override { return false; } |
||||
grpc_call* CreateCall(grpc_call* parent_call, uint32_t propagation_mask, |
||||
grpc_completion_queue* cq, |
||||
grpc_pollset_set* pollset_set_alternative, Slice path, |
||||
absl::optional<Slice> authority, Timestamp deadline, |
||||
bool registered_method) override; |
||||
grpc_event_engine::experimental::EventEngine* event_engine() const override { |
||||
return event_engine_.get(); |
||||
} |
||||
bool SupportsConnectivityWatcher() const override { return false; } |
||||
grpc_connectivity_state CheckConnectivityState(bool) override { |
||||
Crash("CheckConnectivityState not supported"); |
||||
} |
||||
void WatchConnectivityState(grpc_connectivity_state, Timestamp, |
||||
grpc_completion_queue*, void*) override { |
||||
Crash("WatchConnectivityState not supported"); |
||||
} |
||||
void AddConnectivityWatcher( |
||||
grpc_connectivity_state, |
||||
OrphanablePtr<AsyncConnectivityStateWatcherInterface>) override { |
||||
Crash("AddConnectivityWatcher not supported"); |
||||
} |
||||
void RemoveConnectivityWatcher( |
||||
AsyncConnectivityStateWatcherInterface*) override { |
||||
Crash("RemoveConnectivityWatcher not supported"); |
||||
} |
||||
void GetInfo(const grpc_channel_info* channel_info) override; |
||||
void ResetConnectionBackoff() override {} |
||||
void Ping(grpc_completion_queue*, void*) override { |
||||
Crash("Ping not supported"); |
||||
} |
||||
|
||||
private: |
||||
RefCountedPtr<TransportCallDestination> transport_call_destination_; |
||||
RefCountedPtr<UnstartedCallDestination> interception_chain_; |
||||
const std::shared_ptr<grpc_event_engine::experimental::EventEngine> |
||||
event_engine_; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_SRC_CORE_CLIENT_CHANNEL_DIRECT_CHANNEL_H
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,189 @@ |
||||
// Copyright 2024 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 "src/core/lib/transport/call_spine.h" |
||||
|
||||
#include <atomic> |
||||
#include <memory> |
||||
#include <queue> |
||||
|
||||
#include "absl/strings/string_view.h" |
||||
#include "gmock/gmock.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
#include <grpc/grpc.h> |
||||
|
||||
#include "src/core/lib/resource_quota/arena.h" |
||||
#include "src/core/lib/transport/metadata.h" |
||||
#include "test/core/call/yodel/yodel_test.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
using EventEngine = grpc_event_engine::experimental::EventEngine; |
||||
|
||||
namespace { |
||||
const absl::string_view kTestPath = "/foo/bar"; |
||||
} // namespace
|
||||
|
||||
class CallSpineTest : public YodelTest { |
||||
protected: |
||||
using YodelTest::YodelTest; |
||||
|
||||
ClientMetadataHandle MakeClientInitialMetadata() { |
||||
auto client_initial_metadata = Arena::MakePooled<ClientMetadata>(); |
||||
client_initial_metadata->Set(HttpPathMetadata(), |
||||
Slice::FromCopiedString(kTestPath)); |
||||
return client_initial_metadata; |
||||
} |
||||
|
||||
CallInitiatorAndHandler MakeCall( |
||||
ClientMetadataHandle client_initial_metadata) { |
||||
return MakeCallPair(std::move(client_initial_metadata), |
||||
event_engine().get(), |
||||
SimpleArenaAllocator()->MakeArena()); |
||||
} |
||||
|
||||
void UnaryRequest(CallInitiator initiator, CallHandler handler); |
||||
|
||||
private: |
||||
void InitCoreConfiguration() override {} |
||||
|
||||
void Shutdown() override {} |
||||
}; |
||||
|
||||
#define CALL_SPINE_TEST(name) YODEL_TEST(CallSpineTest, name) |
||||
|
||||
CALL_SPINE_TEST(NoOp) {} |
||||
|
||||
CALL_SPINE_TEST(Create) { MakeCall(MakeClientInitialMetadata()); } |
||||
|
||||
void CallSpineTest::UnaryRequest(CallInitiator initiator, CallHandler handler) { |
||||
SpawnTestSeq( |
||||
initiator, "initiator", |
||||
[initiator]() mutable { |
||||
return initiator.PushMessage(Arena::MakePooled<Message>( |
||||
SliceBuffer(Slice::FromCopiedString("hello world")), 0)); |
||||
}, |
||||
[initiator](StatusFlag status) mutable { |
||||
EXPECT_TRUE(status.ok()); |
||||
initiator.FinishSends(); |
||||
return initiator.PullServerInitialMetadata(); |
||||
}, |
||||
[initiator]( |
||||
ValueOrFailure<absl::optional<ServerMetadataHandle>> md) mutable { |
||||
EXPECT_TRUE(md.ok()); |
||||
EXPECT_TRUE(md.value().has_value()); |
||||
EXPECT_EQ(*md.value().value()->get_pointer(ContentTypeMetadata()), |
||||
ContentTypeMetadata::kApplicationGrpc); |
||||
return initiator.PullMessage(); |
||||
}, |
||||
[initiator](ValueOrFailure<absl::optional<MessageHandle>> msg) mutable { |
||||
EXPECT_TRUE(msg.ok()); |
||||
EXPECT_TRUE(msg.value().has_value()); |
||||
EXPECT_EQ(msg.value().value()->payload()->JoinIntoString(), |
||||
"why hello neighbor"); |
||||
return initiator.PullMessage(); |
||||
}, |
||||
[initiator](ValueOrFailure<absl::optional<MessageHandle>> msg) mutable { |
||||
EXPECT_TRUE(msg.ok()); |
||||
EXPECT_FALSE(msg.value().has_value()); |
||||
return initiator.PullServerTrailingMetadata(); |
||||
}, |
||||
[initiator](ValueOrFailure<ServerMetadataHandle> md) mutable { |
||||
EXPECT_TRUE(md.ok()); |
||||
EXPECT_EQ(*md.value()->get_pointer(GrpcStatusMetadata()), |
||||
GRPC_STATUS_UNIMPLEMENTED); |
||||
return Empty{}; |
||||
}); |
||||
SpawnTestSeq( |
||||
handler, "handler", |
||||
[handler]() mutable { return handler.PullClientInitialMetadata(); }, |
||||
[handler](ValueOrFailure<ServerMetadataHandle> md) mutable { |
||||
EXPECT_TRUE(md.ok()); |
||||
EXPECT_EQ(md.value()->get_pointer(HttpPathMetadata())->as_string_view(), |
||||
kTestPath); |
||||
return handler.PullMessage(); |
||||
}, |
||||
[handler](ValueOrFailure<absl::optional<MessageHandle>> msg) mutable { |
||||
EXPECT_TRUE(msg.ok()); |
||||
EXPECT_TRUE(msg.value().has_value()); |
||||
EXPECT_EQ(msg.value().value()->payload()->JoinIntoString(), |
||||
"hello world"); |
||||
return handler.PullMessage(); |
||||
}, |
||||
[handler](ValueOrFailure<absl::optional<MessageHandle>> msg) mutable { |
||||
EXPECT_TRUE(msg.ok()); |
||||
EXPECT_FALSE(msg.value().has_value()); |
||||
auto md = Arena::MakePooled<ServerMetadata>(); |
||||
md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc); |
||||
return handler.PushServerInitialMetadata(std::move(md)); |
||||
}, |
||||
[handler](StatusFlag result) mutable { |
||||
EXPECT_TRUE(result.ok()); |
||||
return handler.PushMessage(Arena::MakePooled<Message>( |
||||
SliceBuffer(Slice::FromCopiedString("why hello neighbor")), 0)); |
||||
}, |
||||
[handler](StatusFlag result) mutable { |
||||
EXPECT_TRUE(result.ok()); |
||||
auto md = Arena::MakePooled<ServerMetadata>(); |
||||
md->Set(GrpcStatusMetadata(), GRPC_STATUS_UNIMPLEMENTED); |
||||
handler.PushServerTrailingMetadata(std::move(md)); |
||||
return Empty{}; |
||||
}); |
||||
} |
||||
|
||||
CALL_SPINE_TEST(UnaryRequest) { |
||||
auto call = MakeCall(MakeClientInitialMetadata()); |
||||
UnaryRequest(call.initiator, call.handler.StartWithEmptyFilterStack()); |
||||
WaitForAllPendingWork(); |
||||
} |
||||
|
||||
CALL_SPINE_TEST(UnaryRequestThroughForwardCall) { |
||||
auto call1 = MakeCall(MakeClientInitialMetadata()); |
||||
auto handler = call1.handler.StartWithEmptyFilterStack(); |
||||
SpawnTestSeq( |
||||
call1.initiator, "initiator", |
||||
[handler]() mutable { return handler.PullClientInitialMetadata(); }, |
||||
[this, handler, initiator = call1.initiator]( |
||||
ValueOrFailure<ClientMetadataHandle> md) mutable { |
||||
EXPECT_TRUE(md.ok()); |
||||
auto call2 = MakeCall(std::move(md.value())); |
||||
ForwardCall(handler, call2.initiator); |
||||
UnaryRequest(initiator, call2.handler.StartWithEmptyFilterStack()); |
||||
return Empty{}; |
||||
}); |
||||
WaitForAllPendingWork(); |
||||
} |
||||
|
||||
CALL_SPINE_TEST(UnaryRequestThroughForwardCallWithServerTrailingMetadataHook) { |
||||
auto call1 = MakeCall(MakeClientInitialMetadata()); |
||||
auto handler = call1.handler.StartWithEmptyFilterStack(); |
||||
bool got_md = false; |
||||
SpawnTestSeq( |
||||
call1.initiator, "initiator", |
||||
[handler]() mutable { return handler.PullClientInitialMetadata(); }, |
||||
[this, handler, initiator = call1.initiator, |
||||
&got_md](ValueOrFailure<ClientMetadataHandle> md) mutable { |
||||
EXPECT_TRUE(md.ok()); |
||||
auto call2 = MakeCall(std::move(md.value())); |
||||
ForwardCall(handler, call2.initiator, |
||||
[&got_md](ServerMetadata&) { got_md = true; }); |
||||
UnaryRequest(initiator, call2.handler.StartWithEmptyFilterStack()); |
||||
return Empty{}; |
||||
}); |
||||
WaitForAllPendingWork(); |
||||
EXPECT_TRUE(got_md); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
@ -0,0 +1 @@ |
||||
|
||||
Loading…
Reference in new issue