inproc_transport.cc | Explore in Territory

//
//
// Copyright 2017 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/ext/transport/inproc/inproc_transport.h"

#include <stdint.h>

#include <algorithm>
#include <memory>
#include <new>
#include <string>
#include <utility>

#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "absl/types/optional.h"

#include <grpc/grpc.h>
#include <grpc/impl/connectivity_state.h>
#include <grpc/status.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/sync.h>

#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/channel_args_preconditioning.h"
#include "src/core/lib/channel/channelz.h"
#include "src/core/lib/config/core_configuration.h"
#include "src/core/lib/gprpp/debug_location.h"
#include "src/core/lib/gprpp/ref_counted_ptr.h"
#include "src/core/lib/gprpp/status_helper.h"
#include "src/core/lib/gprpp/time.h"
#include "src/core/lib/iomgr/closure.h"
#include "src/core/lib/iomgr/endpoint.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/iomgr/iomgr_fwd.h"
#include "src/core/lib/resource_quota/arena.h"
#include "src/core/lib/slice/slice.h"
#include "src/core/lib/slice/slice_buffer.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/channel_stack_type.h"
#include "src/core/lib/surface/server.h"
#include "src/core/lib/transport/connectivity_state.h"
#include "src/core/lib/transport/metadata_batch.h"
#include "src/core/lib/transport/transport.h"
#include "src/core/lib/transport/transport_fwd.h"
#include "src/core/lib/transport/transport_impl.h"

#define INPROC_LOG(...) …

namespace {
struct inproc_stream;
bool cancel_stream_locked(inproc_stream* s, grpc_error_handle error);
void maybe_process_ops_locked(inproc_stream* s, grpc_error_handle error);
void op_state_machine_locked(inproc_stream* s, grpc_error_handle error);
void log_metadata(const grpc_metadata_batch* md_batch, bool is_client,
                  bool is_initial);
void fill_in_metadata(inproc_stream* s, const grpc_metadata_batch* metadata,
                      grpc_metadata_batch* out_md, bool* markfilled);

void ResetSendMessage(grpc_transport_stream_op_batch* batch) { … }

struct shared_mu { … };

struct inproc_transport { … };

struct inproc_stream { … };

void log_metadata(const grpc_metadata_batch* md_batch, bool is_client,
                  bool is_initial) { … }

namespace {

class CopySink { … };

}  // namespace

void fill_in_metadata(inproc_stream* s, const grpc_metadata_batch* metadata,
                      grpc_metadata_batch* out_md, bool* markfilled) { … }

int init_stream(grpc_transport* gt, grpc_stream* gs,
                grpc_stream_refcount* refcount, const void* server_data,
                grpc_core::Arena* arena) { … }

void close_stream_locked(inproc_stream* s) { … }

// This function means that we are done talking/listening to the other side
void close_other_side_locked(inproc_stream* s, const char* reason) { … }

// Call the on_complete closure associated with this stream_op_batch if
// this stream_op_batch is only one of the pending operations for this
// stream. This is called when one of the pending operations for the stream
// is done and about to be NULLed out
void complete_if_batch_end_locked(inproc_stream* s, grpc_error_handle error,
                                  grpc_transport_stream_op_batch* op,
                                  const char* msg) { … }

void maybe_process_ops_locked(inproc_stream* s, grpc_error_handle error) { … }

void fail_helper_locked(inproc_stream* s, grpc_error_handle error) { … }

// TODO(vjpai): It should not be necessary to drain the incoming byte
// stream and create a new one; instead, we should simply pass the byte
// stream from the sender directly to the receiver as-is.
//
// Note that fixing this will also avoid the assumption in this code
// that the incoming byte stream's next() call will always return
// synchronously.  That assumption is true today but may not always be
// true in the future.
void message_transfer_locked(inproc_stream* sender, inproc_stream* receiver) { … }

void op_state_machine_locked(inproc_stream* s, grpc_error_handle error) { … }

bool cancel_stream_locked(inproc_stream* s, grpc_error_handle error) { … }

void do_nothing(void* /*arg*/, grpc_error_handle /*error*/) { … }

void perform_stream_op(grpc_transport* gt, grpc_stream* gs,
                       grpc_transport_stream_op_batch* op) { … }

void close_transport_locked(inproc_transport* t) { … }

void perform_transport_op(grpc_transport* gt, grpc_transport_op* op) { … }

void destroy_stream(grpc_transport* gt, grpc_stream* gs,
                    grpc_closure* then_schedule_closure) { … }

void destroy_transport(grpc_transport* gt) { … }

//******************************************************************************
// INTEGRATION GLUE
//

void set_pollset(grpc_transport* /*gt*/, grpc_stream* /*gs*/,
                 grpc_pollset* /*pollset*/) { … }

void set_pollset_set(grpc_transport* /*gt*/, grpc_stream* /*gs*/,
                     grpc_pollset_set* /*pollset_set*/) { … }

grpc_endpoint* get_endpoint(grpc_transport* /*t*/) { … }

const grpc_transport_vtable inproc_vtable = …;

//******************************************************************************
// Main inproc transport functions
//
void inproc_transports_create(grpc_transport** server_transport,
                              grpc_transport** client_transport) { … }
}  // namespace

grpc_channel* grpc_inproc_channel_create(grpc_server* server,
                                         const grpc_channel_args* args,
                                         void* /*reserved*/) { … }
chromium/third_party/grpc/src/src/core/ext/transport/inproc/inproc_transport.cc
