src/core/lib/transport/transport.cc

   1 /*
   2  *
   3  * Copyright 2015 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <grpc/support/port_platform.h>
  20
  21 #include "src/core/lib/transport/transport.h"
  22
  23 #include <string.h>
  24
  25 #include <grpc/support/alloc.h>
  26 #include <grpc/support/atm.h>
  27 #include <grpc/support/log.h>
  28 #include <grpc/support/sync.h>
  29
  30 #include "src/core/lib/gpr/alloc.h"
  31 #include "src/core/lib/gpr/string.h"
  32 #include "src/core/lib/gprpp/memory.h"
  33 #include "src/core/lib/iomgr/executor.h"
  34 #include "src/core/lib/iomgr/iomgr.h"
  35 #include "src/core/lib/slice/slice_internal.h"
  36 #include "src/core/lib/slice/slice_string_helpers.h"
  37 #include "src/core/lib/transport/transport_impl.h"
  38
  39 grpc_core::DebugOnlyTraceFlag grpc_trace_stream_refcount(false,
  40                                                          "stream_refcount");
  41
  42 void grpc_stream_destroy(grpc_stream_refcount* refcount) {
  43   if (!grpc_iomgr_is_any_background_poller_thread() &&
  44       (grpc_core::ExecCtx::Get()->flags() &
  45        GRPC_EXEC_CTX_FLAG_THREAD_RESOURCE_LOOP)) {
  46     /* Ick.
  47        The thread we're running on MAY be owned (indirectly) by a call-stack.
  48        If that's the case, destroying the call-stack MAY try to destroy the
  49        thread, which is a tangled mess that we just don't want to ever have to
  50        cope with.
  51        Throw this over to the executor (on a core-owned thread) and process it
  52        there. */
  53     grpc_core::Executor::Run(&refcount->destroy, GRPC_ERROR_NONE);
  54   } else {
  55     grpc_core::ExecCtx::Run(DEBUG_LOCATION, &refcount->destroy,
  56                             GRPC_ERROR_NONE);
  57   }
  58 }
  59
  60 void slice_stream_destroy(void* arg) {
  61   grpc_stream_destroy(static_cast<grpc_stream_refcount*>(arg));
  62 }
  63
  64 #define STREAM_REF_FROM_SLICE_REF(p)       \
  65   ((grpc_stream_refcount*)(((uint8_t*)p) - \
  66                            offsetof(grpc_stream_refcount, slice_refcount)))
  67
  68 grpc_slice grpc_slice_from_stream_owned_buffer(grpc_stream_refcount* refcount,
  69                                                void* buffer, size_t length) {
  70 #ifndef NDEBUG
  71   grpc_stream_ref(STREAM_REF_FROM_SLICE_REF(&refcount->slice_refcount),
  72                   "slice");
  73 #else
  74   grpc_stream_ref(STREAM_REF_FROM_SLICE_REF(&refcount->slice_refcount));
  75 #endif
  76   grpc_slice res;
  77   res.refcount = &refcount->slice_refcount;
  78   res.data.refcounted.bytes = static_cast<uint8_t*>(buffer);
  79   res.data.refcounted.length = length;
  80   return res;
  81 }
  82
  83 #ifndef NDEBUG
  84 void grpc_stream_ref_init(grpc_stream_refcount* refcount, int /*initial_refs*/,
  85                           grpc_iomgr_cb_func cb, void* cb_arg,
  86                           const char* object_type) {
  87   refcount->object_type = object_type;
  88 #else
  89 void grpc_stream_ref_init(grpc_stream_refcount* refcount, int /*initial_refs*/,
  90                           grpc_iomgr_cb_func cb, void* cb_arg) {
  91 #endif
  92   GRPC_CLOSURE_INIT(&refcount->destroy, cb, cb_arg, grpc_schedule_on_exec_ctx);
  93
  94   new (&refcount->refs) grpc_core::RefCount(1, &grpc_trace_stream_refcount);
  95   new (&refcount->slice_refcount) grpc_slice_refcount(
  96       grpc_slice_refcount::Type::REGULAR, &refcount->refs, slice_stream_destroy,
  97       refcount, &refcount->slice_refcount);
  98 }
  99
 100 static void move64(uint64_t* from, uint64_t* to) {
 101   *to += *from;
 102   *from = 0;
 103 }
 104
 105 void grpc_transport_move_one_way_stats(grpc_transport_one_way_stats* from,
 106                                        grpc_transport_one_way_stats* to) {
 107   move64(&from->framing_bytes, &to->framing_bytes);
 108   move64(&from->data_bytes, &to->data_bytes);
 109   move64(&from->header_bytes, &to->header_bytes);
 110 }
 111
 112 void grpc_transport_move_stats(grpc_transport_stream_stats* from,
 113                                grpc_transport_stream_stats* to) {
 114   grpc_transport_move_one_way_stats(&from->incoming, &to->incoming);
 115   grpc_transport_move_one_way_stats(&from->outgoing, &to->outgoing);
 116 }
 117
 118 size_t grpc_transport_stream_size(grpc_transport* transport) {
 119   return GPR_ROUND_UP_TO_ALIGNMENT_SIZE(transport->vtable->sizeof_stream);
 120 }
 121
 122 void grpc_transport_destroy(grpc_transport* transport) {
 123   transport->vtable->destroy(transport);
 124 }
 125
 126 int grpc_transport_init_stream(grpc_transport* transport, grpc_stream* stream,
 127                                grpc_stream_refcount* refcount,
 128                                const void* server_data,
 129                                grpc_core::Arena* arena) {
 130   return transport->vtable->init_stream(transport, stream, refcount,
 131                                         server_data, arena);
 132 }
 133
 134 void grpc_transport_perform_stream_op(grpc_transport* transport,
 135                                       grpc_stream* stream,
 136                                       grpc_transport_stream_op_batch* op) {
 137   transport->vtable->perform_stream_op(transport, stream, op);
 138 }
 139
 140 void grpc_transport_perform_op(grpc_transport* transport,
 141                                grpc_transport_op* op) {
 142   transport->vtable->perform_op(transport, op);
 143 }
 144
 145 void grpc_transport_set_pops(grpc_transport* transport, grpc_stream* stream,
 146                              grpc_polling_entity* pollent) {
 147   grpc_pollset* pollset;
 148   grpc_pollset_set* pollset_set;
 149   if ((pollset = grpc_polling_entity_pollset(pollent)) != nullptr) {
 150     transport->vtable->set_pollset(transport, stream, pollset);
 151   } else if ((pollset_set = grpc_polling_entity_pollset_set(pollent)) !=
 152              nullptr) {
 153     transport->vtable->set_pollset_set(transport, stream, pollset_set);
 154   } else {
 155     // No-op for empty pollset. Empty pollset is possible when using
 156     // non-fd-based event engines such as CFStream.
 157   }
 158 }
 159
 160 void grpc_transport_destroy_stream(grpc_transport* transport,
 161                                    grpc_stream* stream,
 162                                    grpc_closure* then_schedule_closure) {
 163   transport->vtable->destroy_stream(transport, stream, then_schedule_closure);
 164 }
 165
 166 grpc_endpoint* grpc_transport_get_endpoint(grpc_transport* transport) {
 167   return transport->vtable->get_endpoint(transport);
 168 }
 169
 170 // This comment should be sung to the tune of
 171 // "Supercalifragilisticexpialidocious":
 172 //
 173 // grpc_transport_stream_op_batch_finish_with_failure
 174 // is a function that must always unref cancel_error
 175 // though it lives in lib, it handles transport stream ops sure
 176 // it's grpc_transport_stream_op_batch_finish_with_failure
 177 void grpc_transport_stream_op_batch_finish_with_failure(
 178     grpc_transport_stream_op_batch* batch, grpc_error* error,
 179     grpc_core::CallCombiner* call_combiner) {
 180   if (batch->send_message) {
 181     batch->payload->send_message.send_message.reset();
 182   }
 183   if (batch->cancel_stream) {
 184     GRPC_ERROR_UNREF(batch->payload->cancel_stream.cancel_error);
 185   }
 186   // Construct a list of closures to execute.
 187   grpc_core::CallCombinerClosureList closures;
 188   if (batch->recv_initial_metadata) {
 189     closures.Add(
 190         batch->payload->recv_initial_metadata.recv_initial_metadata_ready,
 191         GRPC_ERROR_REF(error), "failing recv_initial_metadata_ready");
 192   }
 193   if (batch->recv_message) {
 194     closures.Add(batch->payload->recv_message.recv_message_ready,
 195                  GRPC_ERROR_REF(error), "failing recv_message_ready");
 196   }
 197   if (batch->recv_trailing_metadata) {
 198     closures.Add(
 199         batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready,
 200         GRPC_ERROR_REF(error), "failing recv_trailing_metadata_ready");
 201   }
 202   if (batch->on_complete != nullptr) {
 203     closures.Add(batch->on_complete, GRPC_ERROR_REF(error),
 204                  "failing on_complete");
 205   }
 206   // Execute closures.
 207   closures.RunClosures(call_combiner);
 208   GRPC_ERROR_UNREF(error);
 209 }
 210
 211 struct made_transport_op {
 212   grpc_closure outer_on_complete;
 213   grpc_closure* inner_on_complete = nullptr;
 214   grpc_transport_op op;
 215   made_transport_op() {
 216     memset(&outer_on_complete, 0, sizeof(outer_on_complete));
 217   }
 218 };
 219
 220 static void destroy_made_transport_op(void* arg, grpc_error* error) {
 221   made_transport_op* op = static_cast<made_transport_op*>(arg);
 222   grpc_core::ExecCtx::Run(DEBUG_LOCATION, op->inner_on_complete,
 223                           GRPC_ERROR_REF(error));
 224   delete op;
 225 }
 226
 227 grpc_transport_op* grpc_make_transport_op(grpc_closure* on_complete) {
 228   made_transport_op* op = new made_transport_op();
 229   GRPC_CLOSURE_INIT(&op->outer_on_complete, destroy_made_transport_op, op,
 230                     grpc_schedule_on_exec_ctx);
 231   op->inner_on_complete = on_complete;
 232   op->op.on_consumed = &op->outer_on_complete;
 233   return &op->op;
 234 }
 235
 236 struct made_transport_stream_op {
 237   grpc_closure outer_on_complete;
 238   grpc_closure* inner_on_complete;
 239   grpc_transport_stream_op_batch op;
 240   grpc_transport_stream_op_batch_payload payload;
 241 };
 242 static void destroy_made_transport_stream_op(void* arg, grpc_error* error) {
 243   made_transport_stream_op* op = static_cast<made_transport_stream_op*>(arg);
 244   grpc_closure* c = op->inner_on_complete;
 245   gpr_free(op);
 246   grpc_core::Closure::Run(DEBUG_LOCATION, c, GRPC_ERROR_REF(error));
 247 }
 248
 249 grpc_transport_stream_op_batch* grpc_make_transport_stream_op(
 250     grpc_closure* on_complete) {
 251   made_transport_stream_op* op =
 252       static_cast<made_transport_stream_op*>(gpr_zalloc(sizeof(*op)));
 253   op->op.payload = &op->payload;
 254   GRPC_CLOSURE_INIT(&op->outer_on_complete, destroy_made_transport_stream_op,
 255                     op, grpc_schedule_on_exec_ctx);
 256   op->inner_on_complete = on_complete;
 257   op->op.on_complete = &op->outer_on_complete;
 258   return &op->op;
 259 }