promise_based_filter.cc | Explore in Territory

// Copyright 2022 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/channel/promise_based_filter.h"

#include <algorithm>
#include <initializer_list>
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include "absl/base/attributes.h"
#include "absl/functional/function_ref.h"
#include "absl/status/status.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/str_format.h"
#include "absl/strings/str_join.h"

#include <grpc/status.h>

#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/debug/trace.h"
#include "src/core/lib/gprpp/crash.h"
#include "src/core/lib/gprpp/manual_constructor.h"
#include "src/core/lib/gprpp/status_helper.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/promise/detail/basic_seq.h"
#include "src/core/lib/slice/slice.h"

extern grpc_core::TraceFlag grpc_trace_channel;

namespace grpc_core {
namespace promise_filter_detail {

namespace {
class FakeActivity final : public Activity { … };

absl::Status StatusFromMetadata(const ServerMetadata& md) { … }
}  // namespace

///////////////////////////////////////////////////////////////////////////////
// BaseCallData

BaseCallData::BaseCallData(
    grpc_call_element* elem, const grpc_call_element_args* args, uint8_t flags,
    absl::FunctionRef<Interceptor*()> make_send_interceptor,
    absl::FunctionRef<Interceptor*()> make_recv_interceptor)
    : … { … }

BaseCallData::~BaseCallData() { … }

// We don't form ActivityPtr's to this type, and consequently don't need
// Orphan().
void BaseCallData::Orphan() { … }

// For now we don't care about owning/non-owning wakers, instead just share
// implementation.
Waker BaseCallData::MakeNonOwningWaker() { … }

Waker BaseCallData::MakeOwningWaker() { … }

void BaseCallData::Wakeup(void*) { … }

void BaseCallData::Drop(void*) { … }

std::string BaseCallData::LogTag() const { … }

///////////////////////////////////////////////////////////////////////////////
// BaseCallData::CapturedBatch

namespace {
uintptr_t* RefCountField(grpc_transport_stream_op_batch* b) { … }
}  // namespace

BaseCallData::CapturedBatch::CapturedBatch() : … { … }

BaseCallData::CapturedBatch::CapturedBatch(
    grpc_transport_stream_op_batch* batch) { … }

BaseCallData::CapturedBatch::~CapturedBatch() { … }

BaseCallData::CapturedBatch::CapturedBatch(const CapturedBatch& rhs)
    : … { … }

BaseCallData::CapturedBatch& BaseCallData::CapturedBatch::operator=(
    const CapturedBatch& b) { … }

BaseCallData::CapturedBatch::CapturedBatch(CapturedBatch&& rhs) noexcept
    : … { … }

BaseCallData::CapturedBatch& BaseCallData::CapturedBatch::operator=(
    CapturedBatch&& b) noexcept { … }

void BaseCallData::CapturedBatch::ResumeWith(Flusher* releaser) { … }

void BaseCallData::CapturedBatch::CompleteWith(Flusher* releaser) { … }

void BaseCallData::CapturedBatch::CancelWith(grpc_error_handle error,
                                             Flusher* releaser) { … }

///////////////////////////////////////////////////////////////////////////////
// BaseCallData::Flusher

BaseCallData::Flusher::Flusher(BaseCallData* call) : … { … }

BaseCallData::Flusher::~Flusher() { … }

///////////////////////////////////////////////////////////////////////////////
// BaseCallData::SendMessage

const char* BaseCallData::SendMessage::StateString(State state) { … }

void BaseCallData::SendMessage::StartOp(CapturedBatch batch) { … }

template <typename T>
void BaseCallData::SendMessage::GotPipe(T* pipe_end) { … }

bool BaseCallData::SendMessage::IsIdle() const { … }

void BaseCallData::SendMessage::OnComplete(absl::Status status) { … }

void BaseCallData::SendMessage::Done(const ServerMetadata& metadata,
                                     Flusher* flusher) { … }

void BaseCallData::SendMessage::WakeInsideCombiner(Flusher* flusher,
                                                   bool allow_push_to_pipe) { … }

///////////////////////////////////////////////////////////////////////////////
// BaseCallData::ReceiveMessage

const char* BaseCallData::ReceiveMessage::StateString(State state) { … }

void BaseCallData::ReceiveMessage::StartOp(CapturedBatch& batch) { … }

template <typename T>
void BaseCallData::ReceiveMessage::GotPipe(T* pipe_end) { … }

void BaseCallData::ReceiveMessage::OnComplete(absl::Status status) { … }

void BaseCallData::ReceiveMessage::Done(const ServerMetadata& metadata,
                                        Flusher* flusher) { … }

void BaseCallData::ReceiveMessage::WakeInsideCombiner(Flusher* flusher,
                                                      bool allow_push_to_pipe) { … }

///////////////////////////////////////////////////////////////////////////////
// ClientCallData

struct ClientCallData::RecvInitialMetadata final { … };

class ClientCallData::PollContext { … };

ClientCallData::ClientCallData(grpc_call_element* elem,
                               const grpc_call_element_args* args,
                               uint8_t flags)
    : … { … }

ClientCallData::~ClientCallData() { … }

// Activity implementation.
void ClientCallData::ForceImmediateRepoll() { … }

const char* ClientCallData::StateString(SendInitialState state) { … }

const char* ClientCallData::StateString(RecvTrailingState state) { … }

std::string ClientCallData::DebugString() const { … }

// Handle one grpc_transport_stream_op_batch
void ClientCallData::StartBatch(grpc_transport_stream_op_batch* b) { … }

// Handle cancellation.
void ClientCallData::Cancel(grpc_error_handle error, Flusher* flusher) { … }

// Begin running the promise - which will ultimately take some initial
// metadata and return some trailing metadata.
void ClientCallData::StartPromise(Flusher* flusher) { … }

void ClientCallData::RecvInitialMetadataReady(grpc_error_handle error) { … }

// Interject our callback into the op batch for recv trailing metadata ready.
// Stash a pointer to the trailing metadata that will be filled in, so we can
// manipulate it later.
void ClientCallData::HookRecvTrailingMetadata(CapturedBatch batch) { … }

// Construct a promise that will "call" the next filter.
// Effectively:
//   - put the modified initial metadata into the batch to be sent down.
//   - return a wrapper around PollTrailingMetadata as the promise.
ArenaPromise<ServerMetadataHandle> ClientCallData::MakeNextPromise(
    CallArgs call_args) { … }

// Wrapper to make it look like we're calling the next filter as a promise.
// First poll: send the send_initial_metadata op down the stack.
// All polls: await receiving the trailing metadata, then return it to the
// application.
Poll<ServerMetadataHandle> ClientCallData::PollTrailingMetadata() { … }

void ClientCallData::RecvTrailingMetadataReadyCallback(
    void* arg, grpc_error_handle error) { … }

void ClientCallData::RecvTrailingMetadataReady(grpc_error_handle error) { … }

// Given an error, fill in ServerMetadataHandle to represent that error.
void ClientCallData::SetStatusFromError(grpc_metadata_batch* metadata,
                                        grpc_error_handle error) { … }

// Wakeup and poll the promise if appropriate.
void ClientCallData::WakeInsideCombiner(Flusher* flusher) { … }

void ClientCallData::OnWakeup() { … }

///////////////////////////////////////////////////////////////////////////////
// ServerCallData

struct ServerCallData::SendInitialMetadata { … };

class ServerCallData::PollContext { … };

const char* ServerCallData::StateString(RecvInitialState state) { … }

const char* ServerCallData::StateString(SendTrailingState state) { … }

ServerCallData::ServerCallData(grpc_call_element* elem,
                               const grpc_call_element_args* args,
                               uint8_t flags)
    : … { … }

ServerCallData::~ServerCallData() { … }

// Activity implementation.
void ServerCallData::ForceImmediateRepoll() { … }

// Handle one grpc_transport_stream_op_batch
void ServerCallData::StartBatch(grpc_transport_stream_op_batch* b) { … }

// Handle cancellation.
void ServerCallData::Completed(grpc_error_handle error, Flusher* flusher) { … }

// Construct a promise that will "call" the next filter.
// Effectively:
//   - put the modified initial metadata into the batch being sent up.
//   - return a wrapper around PollTrailingMetadata as the promise.
ArenaPromise<ServerMetadataHandle> ServerCallData::MakeNextPromise(
    CallArgs call_args) { … }

// Wrapper to make it look like we're calling the next filter as a promise.
// All polls: await sending the trailing metadata, then foward it down the
// stack.
Poll<ServerMetadataHandle> ServerCallData::PollTrailingMetadata() { … }

void ServerCallData::RecvTrailingMetadataReadyCallback(
    void* arg, grpc_error_handle error) { … }

void ServerCallData::RecvTrailingMetadataReady(grpc_error_handle error) { … }

void ServerCallData::RecvInitialMetadataReadyCallback(void* arg,
                                                      grpc_error_handle error) { … }

void ServerCallData::RecvInitialMetadataReady(grpc_error_handle error) { … }

std::string ServerCallData::DebugString() const { … }

// Wakeup and poll the promise if appropriate.
void ServerCallData::WakeInsideCombiner(Flusher* flusher) { … }

void ServerCallData::OnWakeup() { … }

}  // namespace promise_filter_detail
}  // namespace grpc_core
chromium/third_party/grpc/src/src/core/lib/channel/promise_based_filter.cc
