mirror of https://github.com/grpc/grpc.git
[call-v3] Filter executor (#35533)
A call execution environment for the V3 runtime.
The `CallFilters` class will ultimately be a (private) member of `CallSpine`, and the `StackBuilder` component will be used by a channel when all of the filters it needs are known to allow the call spine to start processing a call.
This is accompanied by a reasonably extensive test suite.
I expect to fine tune semantics, implementation, and tests over the coming weeks/months as we iterate to bring up the rest of the pieces.
Closes #35533
COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/35533 from ctiller:filters 689c7b527b
PiperOrigin-RevId: 599220150
pull/35549/head
Craig Tiller
1 year ago
committed by
Copybara-Service
parent
c6755cb4b9
commit
584c0c0c98
29 changed files with 3691 additions and 124 deletions
@ -0,0 +1,343 @@ |
|||||||
|
// 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 <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/lib/transport/call_filters.h" |
||||||
|
|
||||||
|
#include "src/core/lib/gprpp/crash.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
namespace { |
||||||
|
void* Offset(void* base, size_t amt) { return static_cast<char*>(base) + amt; } |
||||||
|
} // namespace
|
||||||
|
|
||||||
|
namespace filters_detail { |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
OperationExecutor<T>::~OperationExecutor() { |
||||||
|
if (promise_data_ != nullptr) { |
||||||
|
ops_->early_destroy(promise_data_); |
||||||
|
gpr_free_aligned(promise_data_); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
Poll<ResultOr<T>> OperationExecutor<T>::Start( |
||||||
|
const Layout<FallibleOperator<T>>* layout, T input, void* call_data) { |
||||||
|
ops_ = layout->ops.data(); |
||||||
|
end_ops_ = ops_ + layout->ops.size(); |
||||||
|
if (layout->promise_size == 0) { |
||||||
|
// No call state ==> instantaneously ready
|
||||||
|
auto r = InitStep(std::move(input), call_data); |
||||||
|
GPR_ASSERT(r.ready()); |
||||||
|
return r; |
||||||
|
} |
||||||
|
promise_data_ = |
||||||
|
gpr_malloc_aligned(layout->promise_size, layout->promise_alignment); |
||||||
|
return InitStep(std::move(input), call_data); |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
Poll<ResultOr<T>> OperationExecutor<T>::InitStep(T input, void* call_data) { |
||||||
|
while (true) { |
||||||
|
if (ops_ == end_ops_) { |
||||||
|
return ResultOr<T>{std::move(input), nullptr}; |
||||||
|
} |
||||||
|
auto p = |
||||||
|
ops_->promise_init(promise_data_, Offset(call_data, ops_->call_offset), |
||||||
|
ops_->channel_data, std::move(input)); |
||||||
|
if (auto* r = p.value_if_ready()) { |
||||||
|
if (r->ok == nullptr) return std::move(*r); |
||||||
|
input = std::move(r->ok); |
||||||
|
++ops_; |
||||||
|
continue; |
||||||
|
} |
||||||
|
return Pending{}; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
Poll<ResultOr<T>> OperationExecutor<T>::Step(void* call_data) { |
||||||
|
GPR_DEBUG_ASSERT(promise_data_ != nullptr); |
||||||
|
auto p = ContinueStep(call_data); |
||||||
|
if (p.ready()) { |
||||||
|
gpr_free_aligned(promise_data_); |
||||||
|
promise_data_ = nullptr; |
||||||
|
} |
||||||
|
return p; |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
Poll<ResultOr<T>> OperationExecutor<T>::ContinueStep(void* call_data) { |
||||||
|
auto p = ops_->poll(promise_data_); |
||||||
|
if (auto* r = p.value_if_ready()) { |
||||||
|
if (r->ok == nullptr) return std::move(*r); |
||||||
|
++ops_; |
||||||
|
return InitStep(std::move(r->ok), call_data); |
||||||
|
} |
||||||
|
return Pending{}; |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
InfallibleOperationExecutor<T>::~InfallibleOperationExecutor() { |
||||||
|
if (promise_data_ != nullptr) { |
||||||
|
ops_->early_destroy(promise_data_); |
||||||
|
gpr_free_aligned(promise_data_); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
Poll<T> InfallibleOperationExecutor<T>::Start( |
||||||
|
const Layout<InfallibleOperator<T>>* layout, T input, void* call_data) { |
||||||
|
ops_ = layout->ops.data(); |
||||||
|
end_ops_ = ops_ + layout->ops.size(); |
||||||
|
if (layout->promise_size == 0) { |
||||||
|
// No call state ==> instantaneously ready
|
||||||
|
auto r = InitStep(std::move(input), call_data); |
||||||
|
GPR_ASSERT(r.ready()); |
||||||
|
return r; |
||||||
|
} |
||||||
|
promise_data_ = |
||||||
|
gpr_malloc_aligned(layout->promise_size, layout->promise_alignment); |
||||||
|
return InitStep(std::move(input), call_data); |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
Poll<T> InfallibleOperationExecutor<T>::InitStep(T input, void* call_data) { |
||||||
|
while (true) { |
||||||
|
if (ops_ == end_ops_) { |
||||||
|
return input; |
||||||
|
} |
||||||
|
auto p = |
||||||
|
ops_->promise_init(promise_data_, Offset(call_data, ops_->call_offset), |
||||||
|
ops_->channel_data, std::move(input)); |
||||||
|
if (auto* r = p.value_if_ready()) { |
||||||
|
input = std::move(*r); |
||||||
|
++ops_; |
||||||
|
continue; |
||||||
|
} |
||||||
|
return Pending{}; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
Poll<T> InfallibleOperationExecutor<T>::Step(void* call_data) { |
||||||
|
GPR_DEBUG_ASSERT(promise_data_ != nullptr); |
||||||
|
auto p = ContinueStep(call_data); |
||||||
|
if (p.ready()) { |
||||||
|
gpr_free_aligned(promise_data_); |
||||||
|
promise_data_ = nullptr; |
||||||
|
} |
||||||
|
return p; |
||||||
|
} |
||||||
|
|
||||||
|
template <typename T> |
||||||
|
Poll<T> InfallibleOperationExecutor<T>::ContinueStep(void* call_data) { |
||||||
|
auto p = ops_->poll(promise_data_); |
||||||
|
if (auto* r = p.value_if_ready()) { |
||||||
|
++ops_; |
||||||
|
return InitStep(std::move(*r), call_data); |
||||||
|
} |
||||||
|
return Pending{}; |
||||||
|
} |
||||||
|
|
||||||
|
// Explicit instantiations of some types used in filters.h
|
||||||
|
// We'll need to add ServerMetadataHandle to this when it becomes different
|
||||||
|
// to ClientMetadataHandle
|
||||||
|
template class OperationExecutor<ClientMetadataHandle>; |
||||||
|
template class OperationExecutor<MessageHandle>; |
||||||
|
template class InfallibleOperationExecutor<ServerMetadataHandle>; |
||||||
|
} // namespace filters_detail
|
||||||
|
|
||||||
|
///////////////////////////////////////////////////////////////////////////////
|
||||||
|
// CallFilters
|
||||||
|
|
||||||
|
CallFilters::CallFilters() : stack_(nullptr), call_data_(nullptr) {} |
||||||
|
|
||||||
|
CallFilters::CallFilters(RefCountedPtr<Stack> stack) |
||||||
|
: stack_(std::move(stack)), |
||||||
|
call_data_(gpr_malloc_aligned(stack_->data_.call_data_size, |
||||||
|
stack_->data_.call_data_alignment)) { |
||||||
|
client_initial_metadata_state_.Start(); |
||||||
|
client_to_server_message_state_.Start(); |
||||||
|
server_initial_metadata_state_.Start(); |
||||||
|
server_to_client_message_state_.Start(); |
||||||
|
} |
||||||
|
|
||||||
|
CallFilters::~CallFilters() { |
||||||
|
if (call_data_ != nullptr) gpr_free_aligned(call_data_); |
||||||
|
} |
||||||
|
|
||||||
|
void CallFilters::SetStack(RefCountedPtr<Stack> stack) { |
||||||
|
if (call_data_ != nullptr) gpr_free_aligned(call_data_); |
||||||
|
stack_ = std::move(stack); |
||||||
|
call_data_ = gpr_malloc_aligned(stack_->data_.call_data_size, |
||||||
|
stack_->data_.call_data_alignment); |
||||||
|
client_initial_metadata_state_.Start(); |
||||||
|
client_to_server_message_state_.Start(); |
||||||
|
server_initial_metadata_state_.Start(); |
||||||
|
server_to_client_message_state_.Start(); |
||||||
|
} |
||||||
|
|
||||||
|
void CallFilters::Finalize(const grpc_call_final_info* final_info) { |
||||||
|
for (auto& finalizer : stack_->data_.finalizers) { |
||||||
|
finalizer.final(Offset(call_data_, finalizer.call_offset), |
||||||
|
finalizer.channel_data, final_info); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
void CallFilters::CancelDueToFailedPipeOperation() { |
||||||
|
// We expect something cancelled before now
|
||||||
|
if (server_trailing_metadata_ == nullptr) return; |
||||||
|
gpr_log(GPR_DEBUG, "Cancelling due to failed pipe operation"); |
||||||
|
server_trailing_metadata_ = |
||||||
|
ServerMetadataFromStatus(absl::CancelledError("Failed pipe operation")); |
||||||
|
server_trailing_metadata_waiter_.Wake(); |
||||||
|
} |
||||||
|
|
||||||
|
///////////////////////////////////////////////////////////////////////////////
|
||||||
|
// CallFilters::StackBuilder
|
||||||
|
|
||||||
|
RefCountedPtr<CallFilters::Stack> CallFilters::StackBuilder::Build() { |
||||||
|
if (data_.call_data_size % data_.call_data_alignment != 0) { |
||||||
|
data_.call_data_size += data_.call_data_alignment - |
||||||
|
data_.call_data_size % data_.call_data_alignment; |
||||||
|
} |
||||||
|
// server -> client needs to be reversed so that we can iterate all stacks
|
||||||
|
// in the same order
|
||||||
|
data_.server_initial_metadata.Reverse(); |
||||||
|
data_.server_to_client_messages.Reverse(); |
||||||
|
data_.server_trailing_metadata.Reverse(); |
||||||
|
return RefCountedPtr<Stack>(new Stack(std::move(data_))); |
||||||
|
} |
||||||
|
|
||||||
|
///////////////////////////////////////////////////////////////////////////////
|
||||||
|
// CallFilters::PipeState
|
||||||
|
|
||||||
|
void filters_detail::PipeState::Start() { |
||||||
|
GPR_DEBUG_ASSERT(!started_); |
||||||
|
started_ = true; |
||||||
|
wait_recv_.Wake(); |
||||||
|
} |
||||||
|
|
||||||
|
void filters_detail::PipeState::BeginPush() { |
||||||
|
switch (state_) { |
||||||
|
case ValueState::kIdle: |
||||||
|
state_ = ValueState::kQueued; |
||||||
|
break; |
||||||
|
case ValueState::kWaiting: |
||||||
|
state_ = ValueState::kReady; |
||||||
|
wait_recv_.Wake(); |
||||||
|
break; |
||||||
|
case ValueState::kClosed: |
||||||
|
case ValueState::kError: |
||||||
|
break; |
||||||
|
case ValueState::kQueued: |
||||||
|
case ValueState::kReady: |
||||||
|
case ValueState::kProcessing: |
||||||
|
Crash("Only one push allowed to be outstanding"); |
||||||
|
break; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
void filters_detail::PipeState::DropPush() { |
||||||
|
switch (state_) { |
||||||
|
case ValueState::kQueued: |
||||||
|
case ValueState::kReady: |
||||||
|
case ValueState::kProcessing: |
||||||
|
case ValueState::kWaiting: |
||||||
|
state_ = ValueState::kError; |
||||||
|
wait_recv_.Wake(); |
||||||
|
break; |
||||||
|
case ValueState::kIdle: |
||||||
|
case ValueState::kClosed: |
||||||
|
case ValueState::kError: |
||||||
|
break; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
void filters_detail::PipeState::DropPull() { |
||||||
|
switch (state_) { |
||||||
|
case ValueState::kQueued: |
||||||
|
case ValueState::kReady: |
||||||
|
case ValueState::kProcessing: |
||||||
|
case ValueState::kWaiting: |
||||||
|
state_ = ValueState::kError; |
||||||
|
wait_send_.Wake(); |
||||||
|
break; |
||||||
|
case ValueState::kIdle: |
||||||
|
case ValueState::kClosed: |
||||||
|
case ValueState::kError: |
||||||
|
break; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
Poll<StatusFlag> filters_detail::PipeState::PollPush() { |
||||||
|
switch (state_) { |
||||||
|
case ValueState::kIdle: |
||||||
|
// Read completed and new read started => we see waiting here
|
||||||
|
case ValueState::kWaiting: |
||||||
|
case ValueState::kClosed: |
||||||
|
return Success{}; |
||||||
|
case ValueState::kQueued: |
||||||
|
case ValueState::kReady: |
||||||
|
case ValueState::kProcessing: |
||||||
|
return wait_send_.pending(); |
||||||
|
case ValueState::kError: |
||||||
|
return Failure{}; |
||||||
|
} |
||||||
|
GPR_UNREACHABLE_CODE(return Pending{}); |
||||||
|
} |
||||||
|
|
||||||
|
Poll<StatusFlag> filters_detail::PipeState::PollPull() { |
||||||
|
switch (state_) { |
||||||
|
case ValueState::kWaiting: |
||||||
|
return wait_recv_.pending(); |
||||||
|
case ValueState::kIdle: |
||||||
|
state_ = ValueState::kWaiting; |
||||||
|
return wait_recv_.pending(); |
||||||
|
case ValueState::kReady: |
||||||
|
case ValueState::kQueued: |
||||||
|
if (!started_) return wait_recv_.pending(); |
||||||
|
state_ = ValueState::kProcessing; |
||||||
|
return Success{}; |
||||||
|
case ValueState::kProcessing: |
||||||
|
Crash("Only one pull allowed to be outstanding"); |
||||||
|
case ValueState::kClosed: |
||||||
|
case ValueState::kError: |
||||||
|
return Failure{}; |
||||||
|
} |
||||||
|
GPR_UNREACHABLE_CODE(return Pending{}); |
||||||
|
} |
||||||
|
|
||||||
|
void filters_detail::PipeState::AckPull() { |
||||||
|
switch (state_) { |
||||||
|
case ValueState::kProcessing: |
||||||
|
state_ = ValueState::kIdle; |
||||||
|
wait_send_.Wake(); |
||||||
|
break; |
||||||
|
case ValueState::kWaiting: |
||||||
|
case ValueState::kIdle: |
||||||
|
case ValueState::kQueued: |
||||||
|
case ValueState::kReady: |
||||||
|
case ValueState::kClosed: |
||||||
|
Crash("AckPullValue called in invalid state"); |
||||||
|
case ValueState::kError: |
||||||
|
break; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,43 @@ |
|||||||
|
// 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 <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/lib/transport/message.h" |
||||||
|
|
||||||
|
#include <grpc/impl/grpc_types.h> |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
std::string Message::DebugString() const { |
||||||
|
std::string out = absl::StrCat(payload_.Length(), "b"); |
||||||
|
auto flags = flags_; |
||||||
|
auto explain = [&flags, &out](uint32_t flag, absl::string_view name) { |
||||||
|
if (flags & flag) { |
||||||
|
flags &= ~flag; |
||||||
|
absl::StrAppend(&out, ":", name); |
||||||
|
} |
||||||
|
}; |
||||||
|
explain(GRPC_WRITE_BUFFER_HINT, "write_buffer"); |
||||||
|
explain(GRPC_WRITE_NO_COMPRESS, "no_compress"); |
||||||
|
explain(GRPC_WRITE_THROUGH, "write_through"); |
||||||
|
explain(GRPC_WRITE_INTERNAL_COMPRESS, "compress"); |
||||||
|
explain(GRPC_WRITE_INTERNAL_TEST_ONLY_WAS_COMPRESSED, "was_compressed"); |
||||||
|
if (flags != 0) { |
||||||
|
absl::StrAppend(&out, ":huh=0x", absl::Hex(flags)); |
||||||
|
} |
||||||
|
return out; |
||||||
|
} |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
@ -0,0 +1,61 @@ |
|||||||
|
// 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_LIB_TRANSPORT_MESSAGE_H |
||||||
|
#define GRPC_SRC_CORE_LIB_TRANSPORT_MESSAGE_H |
||||||
|
|
||||||
|
#include <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/lib/resource_quota/arena.h" |
||||||
|
#include "src/core/lib/slice/slice_buffer.h" |
||||||
|
|
||||||
|
/// Internal bit flag for grpc_begin_message's \a flags signaling the use of
|
||||||
|
/// compression for the message. (Does not apply for stream compression.)
|
||||||
|
#define GRPC_WRITE_INTERNAL_COMPRESS (0x80000000u) |
||||||
|
/// Internal bit flag for determining whether the message was compressed and had
|
||||||
|
/// to be decompressed by the message_decompress filter. (Does not apply for
|
||||||
|
/// stream compression.)
|
||||||
|
#define GRPC_WRITE_INTERNAL_TEST_ONLY_WAS_COMPRESSED (0x40000000u) |
||||||
|
/// Mask of all valid internal flags.
|
||||||
|
#define GRPC_WRITE_INTERNAL_USED_MASK \ |
||||||
|
(GRPC_WRITE_INTERNAL_COMPRESS | GRPC_WRITE_INTERNAL_TEST_ONLY_WAS_COMPRESSED) |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
class Message { |
||||||
|
public: |
||||||
|
Message() = default; |
||||||
|
~Message() = default; |
||||||
|
Message(SliceBuffer payload, uint32_t flags) |
||||||
|
: payload_(std::move(payload)), flags_(flags) {} |
||||||
|
Message(const Message&) = delete; |
||||||
|
Message& operator=(const Message&) = delete; |
||||||
|
|
||||||
|
uint32_t flags() const { return flags_; } |
||||||
|
uint32_t& mutable_flags() { return flags_; } |
||||||
|
SliceBuffer* payload() { return &payload_; } |
||||||
|
const SliceBuffer* payload() const { return &payload_; } |
||||||
|
|
||||||
|
std::string DebugString() const; |
||||||
|
|
||||||
|
private: |
||||||
|
SliceBuffer payload_; |
||||||
|
uint32_t flags_ = 0; |
||||||
|
}; |
||||||
|
|
||||||
|
using MessageHandle = Arena::PoolPtr<Message>; |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
|
|
||||||
|
#endif // GRPC_SRC_CORE_LIB_TRANSPORT_MESSAGE_H
|
||||||
@ -0,0 +1,37 @@ |
|||||||
|
// 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 <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/lib/transport/metadata.h" |
||||||
|
|
||||||
|
#include "src/core/lib/transport/error_utils.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
ServerMetadataHandle ServerMetadataFromStatus(const absl::Status& status, |
||||||
|
Arena* arena) { |
||||||
|
auto hdl = arena->MakePooled<ServerMetadata>(arena); |
||||||
|
grpc_status_code code; |
||||||
|
std::string message; |
||||||
|
grpc_error_get_status(status, Timestamp::InfFuture(), &code, &message, |
||||||
|
nullptr, nullptr); |
||||||
|
hdl->Set(GrpcStatusMetadata(), code); |
||||||
|
if (!status.ok()) { |
||||||
|
hdl->Set(GrpcMessageMetadata(), Slice::FromCopiedString(message)); |
||||||
|
} |
||||||
|
return hdl; |
||||||
|
} |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
@ -0,0 +1,78 @@ |
|||||||
|
// 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_LIB_TRANSPORT_METADATA_H |
||||||
|
#define GRPC_SRC_CORE_LIB_TRANSPORT_METADATA_H |
||||||
|
|
||||||
|
#include <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/lib/transport/metadata_batch.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
// Server metadata type
|
||||||
|
// TODO(ctiller): This should be a bespoke instance of MetadataMap<>
|
||||||
|
using ServerMetadata = grpc_metadata_batch; |
||||||
|
using ServerMetadataHandle = Arena::PoolPtr<ServerMetadata>; |
||||||
|
|
||||||
|
// Client initial metadata type
|
||||||
|
// TODO(ctiller): This should be a bespoke instance of MetadataMap<>
|
||||||
|
using ClientMetadata = grpc_metadata_batch; |
||||||
|
using ClientMetadataHandle = Arena::PoolPtr<ClientMetadata>; |
||||||
|
|
||||||
|
// Ok/not-ok check for trailing metadata, so that it can be used as result types
|
||||||
|
// for TrySeq.
|
||||||
|
inline bool IsStatusOk(const ServerMetadataHandle& m) { |
||||||
|
return m->get(GrpcStatusMetadata()).value_or(GRPC_STATUS_UNKNOWN) == |
||||||
|
GRPC_STATUS_OK; |
||||||
|
} |
||||||
|
|
||||||
|
ServerMetadataHandle ServerMetadataFromStatus( |
||||||
|
const absl::Status& status, Arena* arena = GetContext<Arena>()); |
||||||
|
|
||||||
|
template <> |
||||||
|
struct StatusCastImpl<ServerMetadataHandle, absl::Status> { |
||||||
|
static ServerMetadataHandle Cast(const absl::Status& m) { |
||||||
|
return ServerMetadataFromStatus(m); |
||||||
|
} |
||||||
|
}; |
||||||
|
|
||||||
|
template <> |
||||||
|
struct StatusCastImpl<ServerMetadataHandle, const absl::Status&> { |
||||||
|
static ServerMetadataHandle Cast(const absl::Status& m) { |
||||||
|
return ServerMetadataFromStatus(m); |
||||||
|
} |
||||||
|
}; |
||||||
|
|
||||||
|
template <> |
||||||
|
struct StatusCastImpl<ServerMetadataHandle, absl::Status&> { |
||||||
|
static ServerMetadataHandle Cast(const absl::Status& m) { |
||||||
|
return ServerMetadataFromStatus(m); |
||||||
|
} |
||||||
|
}; |
||||||
|
|
||||||
|
// Anything that can be first cast to absl::Status can then be cast to
|
||||||
|
// ServerMetadataHandle.
|
||||||
|
template <typename T> |
||||||
|
struct StatusCastImpl< |
||||||
|
ServerMetadataHandle, T, |
||||||
|
absl::void_t<decltype(&StatusCastImpl<absl::Status, T>::Cast)>> { |
||||||
|
static ServerMetadataHandle Cast(const T& m) { |
||||||
|
return ServerMetadataFromStatus(StatusCast<absl::Status>(m)); |
||||||
|
} |
||||||
|
}; |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
|
|
||||||
|
#endif // GRPC_SRC_CORE_LIB_TRANSPORT_METADATA_H
|
||||||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in new issue