ENVOY_LOG(debug, "completionThread running");

void* tag;

bool ok;

while (cq_.Next(&tag, &ok)) {

const auto& google_async_tag = *reinterpret_cast<GoogleAsyncTag*>(tag);

const GoogleAsyncTag::Operation op = google_async_tag.op_;

GoogleAsyncStreamImpl& stream = google_async_tag.stream_;

ENVOY_LOG(trace, "completionThread CQ event {} {}", static_cast<int>(op), ok);

Thread::LockGuard lock(stream.completed_ops_lock_);

// It's an invariant that there must only be one pending post for arbitrary

// length completed_ops_, otherwise we can race in stream destruction, where

// we process multiple events in onCompletedOps() but have only partially

// consumed the posts on the dispatcher.

// TODO(htuch): This may result in unbounded processing on the silo thread

// in onCompletedOps() in extreme cases, when we emplace_back() in

// completionThread() at a high rate, consider bounding the length of such

// sequences if this behavior becomes an issue.

if (stream.completed_ops_.empty()) {

stream.dispatcher_.post([&stream] { stream.onCompletedOps(); });

}

stream.completed_ops_.emplace_back(op, ok);

}

ENVOY_LOG(debug, "completionThread exiting");

GoogleAsyncClientThreadLocal& tls,

GoogleStubFactory& stub_factory,

Stats::ScopeSharedPtr scope,

const envoy::config::core::v3::GrpcService& config,

Server::Configuration::CommonFactoryContext& context,

const StatNames& stat_names)

: dispatcher_(dispatcher), tls_(tls), stat_prefix_(config.google_grpc().stat_prefix()),

target_uri_(config.google_grpc().target_uri()), scope_(scope),

per_stream_buffer_limit_bytes_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(

config.google_grpc(), per_stream_buffer_limit_bytes, DefaultBufferLimitBytes)),

metadata_parser_(THROW_OR_RETURN_VALUE(

Router::HeaderParser::configure(

config.initial_metadata(),

envoy::config::core::v3::HeaderValueOption::OVERWRITE_IF_EXISTS_OR_ADD),

Router::HeaderParserPtr)) {

// We rebuild the channel each time we construct the channel. It appears that the gRPC library is

// smart enough to do connection pooling and reuse with identical channel args, so this should

// have comparable overhead to what we are doing in Grpc::AsyncClientImpl, i.e. no expensive

// new connection implied.

std::shared_ptr<grpc::Channel> channel = GoogleGrpcUtils::createChannel(config, context);

// Get state with try_to_connect = true to try connection at channel creation.

// This is for initializing gRPC channel at channel creation. This GetState(true) is used to poke

// the gRPC lb at channel creation, it doesn't have any effect no matter it succeeds or fails. But

// it helps on initialization. Otherwise, the channel establishment still happens at the first

// request, no matter when we create the channel.

channel->GetState(true);

stub_ = stub_factory.createStub(channel);

scope_->counterFromStatName(stat_names.google_grpc_client_creation_).inc();

// Initialize client stats.

// TODO(jmarantz): Capture these names in async_client_manager_impl.cc and

// pass in a struct of StatName objects so we don't have to take locks here.

stats_.streams_total_ = &scope_->counterFromStatName(stat_names.streams_total_);

for (uint32_t i = 0; i <= Status::WellKnownGrpcStatus::MaximumKnown; ++i) {

stats_.streams_closed_[i] = &scope_->counterFromStatName(stat_names.streams_closed_[i]);

}

absl::string_view service_full_name,

absl::string_view method_name,

RawAsyncStreamCallbacks& callbacks,

const Http::AsyncClient::StreamOptions& options)

: parent_(parent), tls_(parent_.tls_), dispatcher_(parent_.dispatcher_), stub_(parent_.stub_),

service_full_name_(service_full_name), method_name_(method_name), callbacks_(callbacks),

options_(options), unused_stream_info_(Http::Protocol::Http2, dispatcher_.timeSource(),

Network::ConnectionInfoProviderSharedPtr{},

StreamInfo::FilterState::LifeSpan::FilterChain) {

// TODO(cainelli): add a common library for tracing tags between gRPC implementations.

if (options.parent_span_ != nullptr) {

const std::string child_span_name =

options.child_span_name_.empty()

? absl::StrCat("async ", service_full_name, ".", method_name, " egress")

: options.child_span_name_;

current_span_ = options.parent_span_->spawnChild(Tracing::EgressConfig::get(), child_span_name,

parent.timeSource().systemTime());

current_span_->setTag(Tracing::Tags::get().UpstreamCluster, parent.stat_prefix_);

current_span_->setTag(Tracing::Tags::get().UpstreamAddress, parent.target_uri_);

current_span_->setTag(Tracing::Tags::get().Component, Tracing::Tags::get().Proxy);

} else {

current_span_ = std::make_unique<Tracing::NullSpan>();

}

if (options.sampled_.has_value()) {

current_span_->setSampled(options.sampled_.value());

}

parent_.stats_.streams_total_->inc();

gpr_timespec abs_deadline =

options_.timeout

? gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),

gpr_time_from_millis(options_.timeout.value().count(), GPR_TIMESPAN))

: gpr_inf_future(GPR_CLOCK_REALTIME);

ctxt_.set_deadline(abs_deadline);

// Fill service-wide initial metadata.

auto initial_metadata = Http::RequestHeaderMapImpl::create();

// TODO(cpakulski): Find a better way to access requestHeaders

// request headers should not be stored in stream_info.

// Maybe put it to parent_context?

parent_.metadata_parser_->evaluateHeaders(*initial_metadata, options_.parent_context.stream_info);

Tracing::HttpTraceContext trace_context(*initial_metadata);

Tracing::UpstreamContext upstream_context(nullptr, // host_

nullptr, // cluster_

Tracing::ServiceType::GoogleGrpc, // service_type_

true // async_client_span_

);

current_span_->injectContext(trace_context, upstream_context);

callbacks_.onCreateInitialMetadata(*initial_metadata);

initial_metadata->iterate([this](const Http::HeaderEntry& header) {

ctxt_.AddMetadata(std::string(header.key().getStringView()),

std::string(header.value().getStringView()));

return Http::HeaderMap::Iterate::Continue;

});

// Invoke stub call.

rw_ = parent_.stub_->PrepareCall(&ctxt_, "/" + service_full_name_ + "/" + method_name_,

&parent_.tls_.completionQueue());

if (rw_ == nullptr) {

notifyRemoteClose(Status::WellKnownGrpcStatus::Unavailable, nullptr, EMPTY_STRING);

call_failed_ = true;

return;

}

parent_.tls_.registerStream(this);

rw_->StartCall(&init_tag_);

++inflight_tags_;

Http::ResponseTrailerMapPtr trailing_metadata,

const std::string& message) {

if (grpc_status > Status::WellKnownGrpcStatus::MaximumKnown || grpc_status < 0) {

ENVOY_LOG(error, "notifyRemoteClose invalid gRPC status code {}", grpc_status);

// Set the grpc_status as InvalidCode but increment the Unknown stream to avoid out-of-range

// crash..

grpc_status = Status::WellKnownGrpcStatus::InvalidCode;

parent_.stats_.streams_closed_[Status::WellKnownGrpcStatus::Unknown]->inc();

} else {

parent_.stats_.streams_closed_[grpc_status]->inc();

}

ENVOY_LOG(debug, "notifyRemoteClose {} {}", grpc_status, message);

current_span_->setTag(Tracing::Tags::get().GrpcStatusCode, std::to_string(grpc_status));

if (grpc_status != Status::WellKnownGrpcStatus::Ok) {

current_span_->setTag(Tracing::Tags::get().Error, Tracing::Tags::get().True);

}

current_span_->finishSpan();

if (!waiting_to_delete_on_remote_close_) {

callbacks_.onReceiveTrailingMetadata(

trailing_metadata ? std::move(trailing_metadata) : Http::ResponseTrailerMapImpl::create());

callbacks_.onRemoteClose(grpc_status, message);

}

write_pending_queue_.emplace(std::move(request), end_stream);

ENVOY_LOG(trace, "Queued message to write ({} bytes)",

write_pending_queue_.back().buf_.value().Length());

bytes_in_write_pending_queue_ += write_pending_queue_.back().buf_.value().Length();

writeQueued();

// Empty EOS write queued.

current_span_->setTag(Tracing::Tags::get().Status, Tracing::Tags::get().Canceled);

current_span_->finishSpan();

write_pending_queue_.emplace();

writeQueued();

ENVOY_LOG(debug, "resetStream");

// The gRPC API requires calling Finish() at the end of a stream, even

// if the stream is cancelled.

if (!finish_pending_) {

finish_pending_ = true;

rw_->Finish(&status_, &finish_tag_);

++inflight_tags_;

}

cleanup();

if (!waiting_to_delete_on_remote_close_) {

waiting_to_delete_on_remote_close_ = true;

remote_close_timer_ = dispatcher_.createTimer([this] { resetStream(); });

remote_close_timer_->enableTimer(options_.remote_close_timeout);

}

if (!call_initialized_ || finish_pending_ || write_pending_ || write_pending_queue_.empty() ||

draining_cq_) {

return;

}

write_pending_ = true;

const PendingMessage& msg = write_pending_queue_.front();

if (!msg.buf_) {

ASSERT(msg.end_stream_);

rw_->WritesDone(&write_last_tag_);

++inflight_tags_;

} else if (msg.end_stream_) {

grpc::WriteOptions write_options;

rw_->WriteLast(msg.buf_.value(), write_options, &write_last_tag_);

++inflight_tags_;

} else {

rw_->Write(msg.buf_.value(), &write_tag_);

++inflight_tags_;

}

ENVOY_LOG(trace, "Write op dispatched");

// The items in completed_ops_ execute in the order they were originally added to the queue since

// both the post callback scheduled by the completionThread and the deferred deletion of the

// GoogleAsyncClientThreadLocal happen on the dispatcher thread.

std::deque<std::pair<GoogleAsyncTag::Operation, bool>> completed_ops;

{

Thread::LockGuard lock(completed_ops_lock_);

completed_ops = std::move(completed_ops_);

// completed_ops_ should be empty after the move.

ASSERT(completed_ops_.empty());

}

while (!completed_ops.empty()) {

GoogleAsyncTag::Operation op;

bool ok;

std::tie(op, ok) = completed_ops.front();

completed_ops.pop_front();

handleOpCompletion(op, ok);

}

ENVOY_LOG(trace, "handleOpCompletion op={} ok={} inflight={}", static_cast<int>(op), ok,

inflight_tags_);

ASSERT(inflight_tags_ > 0);

--inflight_tags_;

if (draining_cq_) {

if (inflight_tags_ == 0) {

deferredDelete();

}

// Ignore op completions while draining CQ.

return;

}

// Consider failure cases first.

if (!ok) {

// Early fails can be just treated as Internal.

if (op == GoogleAsyncTag::Operation::Init ||

op == GoogleAsyncTag::Operation::ReadInitialMetadata) {

notifyRemoteClose(Status::WellKnownGrpcStatus::Internal, nullptr, EMPTY_STRING);

resetStream();

return;

}

// Remote server has closed, we can pick up some meaningful status.

// TODO(htuch): We're assuming here that a failed Write/WriteLast operation will result in

// stream termination, and pick up on the failed Read here. Confirm that this assumption is

// valid.

if (op == GoogleAsyncTag::Operation::Read) {

finish_pending_ = true;

rw_->Finish(&status_, &finish_tag_);

++inflight_tags_;

}

return;

}

switch (op) {

case GoogleAsyncTag::Operation::Init: {

ASSERT(ok);

ASSERT(!call_initialized_);

call_initialized_ = true;

rw_->ReadInitialMetadata(&read_initial_metadata_tag_);

++inflight_tags_;

writeQueued();

break;

}

case GoogleAsyncTag::Operation::ReadInitialMetadata: {

ASSERT(ok);

ASSERT(call_initialized_);

rw_->Read(&read_buf_, &read_tag_);

++inflight_tags_;

Http::ResponseHeaderMapPtr initial_metadata = Http::ResponseHeaderMapImpl::create();

metadataTranslate(ctxt_.GetServerInitialMetadata(), *initial_metadata);

if (!waiting_to_delete_on_remote_close_) {

callbacks_.onReceiveInitialMetadata(std::move(initial_metadata));

}

break;

}

case GoogleAsyncTag::Operation::Write: {

ASSERT(ok);

write_pending_ = false;

bytes_in_write_pending_queue_ -= write_pending_queue_.front().buf_.value().Length();

write_pending_queue_.pop();

writeQueued();

break;

}

case GoogleAsyncTag::Operation::WriteLast: {

ASSERT(ok);

write_pending_ = false;

break;

}

case GoogleAsyncTag::Operation::Read: {

ASSERT(ok);

auto buffer = GoogleGrpcUtils::makeBufferInstance(read_buf_);

if (!buffer || (!waiting_to_delete_on_remote_close_ &&

!callbacks_.onReceiveMessageRaw(std::move(buffer)))) {

// This is basically streamError in Grpc::AsyncClientImpl.

notifyRemoteClose(Status::WellKnownGrpcStatus::Internal, nullptr, EMPTY_STRING);

resetStream();

break;

}

rw_->Read(&read_buf_, &read_tag_);

++inflight_tags_;

break;

}

case GoogleAsyncTag::Operation::Finish: {

ASSERT(finish_pending_);

ENVOY_LOG(debug, "Finish with grpc-status code {}", static_cast<int>(status_.error_code()));

Http::ResponseTrailerMapPtr trailing_metadata = Http::ResponseTrailerMapImpl::create();

metadataTranslate(ctxt_.GetServerTrailingMetadata(), *trailing_metadata);

notifyRemoteClose(static_cast<Status::GrpcStatus>(status_.error_code()),

std::move(trailing_metadata), status_.error_message());

cleanup();

break;

}

}

const std::multimap<grpc::string_ref, grpc::string_ref>& grpc_metadata,

Http::HeaderMap& header_map) {

// More painful copying, this time due to the mismatch in header

// representation data structures in Envoy and Google gRPC.

for (const auto& it : grpc_metadata) {

auto key = Http::LowerCaseString(std::string(it.first.data(), it.first.size()));

if (absl::EndsWith(key.get(), "-bin")) {

auto value = Base64::encode(it.second.data(), it.second.size());

header_map.addCopy(key, value);

continue;

}

header_map.addCopy(key, std::string(it.second.data(), it.second.size()));

}

ENVOY_LOG(debug, "Deferred delete");

tls_.unregisterStream(this);

// We only get here following cleanup(), which has performed a

// remoteFromList(), resulting in self-ownership of the object's memory.

// Hence, it is safe here to create a unique_ptr to this and transfer

// ownership to dispatcher_.deferredDelete(). After this call, no further

// methods may be invoked on this object.

dispatcher_.deferredDelete(GoogleAsyncStreamImplPtr(this));

ENVOY_LOG(debug, "Stream cleanup with {} in-flight tags", inflight_tags_);

// We can get here if the client has already issued resetStream() and, while

// this is in progress, the destructor runs.

if (draining_cq_) {

ENVOY_LOG(debug, "Cleanup already in progress");

return;

}

draining_cq_ = true;

ctxt_.TryCancel();

if (LinkedObject<GoogleAsyncStreamImpl>::inserted()) {

// We take ownership of our own memory at this point.

LinkedObject<GoogleAsyncStreamImpl>::removeFromList(parent_.active_streams_).release();

if (inflight_tags_ == 0) {

deferredDelete();

}

}

remote_close_timer_ = nullptr;

GoogleAsyncClientImpl& parent, absl::string_view service_full_name,

absl::string_view method_name, Buffer::InstancePtr request, RawAsyncRequestCallbacks& callbacks,

Tracing::Span& parent_span, const Http::AsyncClient::RequestOptions& options)

: GoogleAsyncStreamImpl(parent, service_full_name, method_name, *this, options),

request_(std::move(request)), callbacks_(callbacks) {

current_span_ =

parent_span.spawnChild(Tracing::EgressConfig::get(),

absl::StrCat("async ", service_full_name, ".", method_name, " egress"),

parent.timeSource().systemTime());

current_span_->setTag(Tracing::Tags::get().UpstreamCluster, parent.stat_prefix_);

current_span_->setTag(Tracing::Tags::get().UpstreamAddress, parent.target_uri_);

current_span_->setTag(Tracing::Tags::get().Component, Tracing::Tags::get().Proxy);

GoogleAsyncStreamImpl::initialize(buffer_body_for_retry);

if (callFailed()) {

return;

}

sendMessageRaw(std::move(request_), true);

current_span_->setTag(Tracing::Tags::get().Status, Tracing::Tags::get().Canceled);

current_span_->finishSpan();

resetStream();

// TODO(wbpcode): In most tracers the span will hold a reference to the tracer self

// and it's possible that become a dangling reference for long time async request.

// This require further PR to resolve.

options_.sidestream_watermark_callbacks = nullptr;

options_.parent_span_ = nullptr;

options_.parent_context.stream_info = nullptr;

streamInfo().clearParentStreamInfo();

Tracing::HttpTraceContext trace_context(metadata);

Tracing::UpstreamContext upstream_context(nullptr, // host_

nullptr, // cluster_

Tracing::ServiceType::GoogleGrpc, // service_type_

true // async_client_span_

);

current_span_->injectContext(trace_context, upstream_context);

callbacks_.onCreateInitialMetadata(metadata);

const std::string& message) {

current_span_->setTag(Tracing::Tags::get().GrpcStatusCode, std::to_string(status));

if (status != Grpc::Status::WellKnownGrpcStatus::Ok) {

current_span_->setTag(Tracing::Tags::get().Error, Tracing::Tags::get().True);

callbacks_.onFailure(status, message, *current_span_);

} else if (response_ == nullptr) {

current_span_->setTag(Tracing::Tags::get().Error, Tracing::Tags::get().True);

callbacks_.onFailure(Status::Internal, EMPTY_STRING, *current_span_);

} else {

callbacks_.onSuccessRaw(std::move(response_), *current_span_);

}

current_span_->finishSpan();