| /* |
| * |
| * Copyright 2016 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 <string.h> |
| |
| #include <grpc/slice_buffer.h> |
| #include <grpc/support/alloc.h> |
| #include <grpc/support/log.h> |
| #include <grpc/support/string_util.h> |
| |
| #include "src/core/ext/transport/chttp2/transport/bin_decoder.h" |
| #include "src/core/ext/transport/chttp2/transport/bin_encoder.h" |
| #include "src/core/ext/transport/chttp2/transport/incoming_metadata.h" |
| #include "src/core/ext/transport/cronet/transport/cronet_transport.h" |
| #include "src/core/lib/gpr/host_port.h" |
| #include "src/core/lib/gpr/string.h" |
| #include "src/core/lib/gprpp/manual_constructor.h" |
| #include "src/core/lib/iomgr/endpoint.h" |
| #include "src/core/lib/iomgr/exec_ctx.h" |
| #include "src/core/lib/slice/slice_internal.h" |
| #include "src/core/lib/slice/slice_string_helpers.h" |
| #include "src/core/lib/surface/channel.h" |
| #include "src/core/lib/transport/metadata_batch.h" |
| #include "src/core/lib/transport/static_metadata.h" |
| #include "src/core/lib/transport/transport_impl.h" |
| #include "third_party/objective_c/Cronet/bidirectional_stream_c.h" |
| |
| #define GRPC_HEADER_SIZE_IN_BYTES 5 |
| #define GRPC_FLUSH_READ_SIZE 4096 |
| |
| #define CRONET_LOG(...) \ |
| do { \ |
| if (grpc_cronet_trace) gpr_log(__VA_ARGS__); \ |
| } while (0) |
| |
| /* TODO (makdharma): Hook up into the wider tracing mechanism */ |
| int grpc_cronet_trace = 0; |
| |
| enum e_op_result { |
| ACTION_TAKEN_WITH_CALLBACK, |
| ACTION_TAKEN_NO_CALLBACK, |
| NO_ACTION_POSSIBLE |
| }; |
| |
| enum e_op_id { |
| OP_SEND_INITIAL_METADATA = 0, |
| OP_SEND_MESSAGE, |
| OP_SEND_TRAILING_METADATA, |
| OP_RECV_MESSAGE, |
| OP_RECV_INITIAL_METADATA, |
| OP_RECV_TRAILING_METADATA, |
| OP_CANCEL_ERROR, |
| OP_ON_COMPLETE, |
| OP_FAILED, |
| OP_SUCCEEDED, |
| OP_CANCELED, |
| OP_RECV_MESSAGE_AND_ON_COMPLETE, |
| OP_READ_REQ_MADE, |
| OP_NUM_OPS |
| }; |
| |
| /* Cronet callbacks. See cronet_c_for_grpc.h for documentation for each. */ |
| |
| static void on_stream_ready(bidirectional_stream*); |
| static void on_response_headers_received( |
| bidirectional_stream*, const bidirectional_stream_header_array*, |
| const char*); |
| static void on_write_completed(bidirectional_stream*, const char*); |
| static void on_read_completed(bidirectional_stream*, char*, int); |
| static void on_response_trailers_received( |
| bidirectional_stream*, const bidirectional_stream_header_array*); |
| static void on_succeeded(bidirectional_stream*); |
| static void on_failed(bidirectional_stream*, int); |
| static void on_canceled(bidirectional_stream*); |
| static bidirectional_stream_callback cronet_callbacks = { |
| on_stream_ready, |
| on_response_headers_received, |
| on_read_completed, |
| on_write_completed, |
| on_response_trailers_received, |
| on_succeeded, |
| on_failed, |
| on_canceled}; |
| |
| /* Cronet transport object */ |
| struct grpc_cronet_transport { |
| grpc_transport base; /* must be first element in this structure */ |
| stream_engine* engine; |
| char* host; |
| bool use_packet_coalescing; |
| }; |
| typedef struct grpc_cronet_transport grpc_cronet_transport; |
| |
| /* TODO (makdharma): reorder structure for memory efficiency per |
| http://www.catb.org/esr/structure-packing/#_structure_reordering: */ |
| struct read_state { |
| read_state(gpr_arena* arena) |
| : trailing_metadata(arena), initial_metadata(arena) { |
| grpc_slice_buffer_init(&read_slice_buffer); |
| } |
| |
| /* vars to store data coming from server */ |
| char* read_buffer = nullptr; |
| bool length_field_received = false; |
| int received_bytes = 0; |
| int remaining_bytes = 0; |
| int length_field = 0; |
| bool compressed = 0; |
| char grpc_header_bytes[GRPC_HEADER_SIZE_IN_BYTES] = {}; |
| char* payload_field = nullptr; |
| bool read_stream_closed = 0; |
| |
| /* vars for holding data destined for the application */ |
| grpc_core::ManualConstructor<grpc_core::SliceBufferByteStream> sbs; |
| grpc_slice_buffer read_slice_buffer; |
| |
| /* vars for trailing metadata */ |
| grpc_chttp2_incoming_metadata_buffer trailing_metadata; |
| bool trailing_metadata_valid = false; |
| |
| /* vars for initial metadata */ |
| grpc_chttp2_incoming_metadata_buffer initial_metadata; |
| }; |
| |
| struct write_state { |
| char* write_buffer = nullptr; |
| }; |
| |
| /* track state of one stream op */ |
| struct op_state { |
| op_state(gpr_arena* arena) : rs(arena) {} |
| |
| bool state_op_done[OP_NUM_OPS] = {}; |
| bool state_callback_received[OP_NUM_OPS] = {}; |
| /* A non-zero gRPC status code has been seen */ |
| bool fail_state = false; |
| /* Transport is discarding all buffered messages */ |
| bool flush_read = false; |
| bool flush_cronet_when_ready = false; |
| bool pending_write_for_trailer = false; |
| bool pending_send_message = false; |
| /* User requested RECV_TRAILING_METADATA */ |
| bool pending_recv_trailing_metadata = false; |
| /* Cronet has not issued a callback of a bidirectional read */ |
| bool pending_read_from_cronet = false; |
| grpc_error* cancel_error = GRPC_ERROR_NONE; |
| /* data structure for storing data coming from server */ |
| struct read_state rs; |
| /* data structure for storing data going to the server */ |
| struct write_state ws; |
| }; |
| |
| struct stream_obj; |
| |
| struct op_and_state { |
| op_and_state(stream_obj* s, const grpc_transport_stream_op_batch& op); |
| |
| grpc_transport_stream_op_batch op; |
| struct op_state state; |
| bool done = false; |
| struct stream_obj* s; /* Pointer back to the stream object */ |
| /* next op_and_state in the linked list */ |
| struct op_and_state* next; |
| }; |
| |
| struct op_storage { |
| int num_pending_ops = 0; |
| struct op_and_state* head = nullptr; |
| }; |
| |
| struct stream_obj { |
| stream_obj(grpc_transport* gt, grpc_stream* gs, |
| grpc_stream_refcount* refcount, gpr_arena* arena); |
| ~stream_obj(); |
| |
| gpr_arena* arena; |
| struct op_and_state* oas = nullptr; |
| grpc_transport_stream_op_batch* curr_op = nullptr; |
| grpc_cronet_transport* curr_ct; |
| grpc_stream* curr_gs; |
| bidirectional_stream* cbs = nullptr; |
| bidirectional_stream_header_array header_array = |
| bidirectional_stream_header_array(); // Zero-initialize the structure. |
| |
| /* Stream level state. Some state will be tracked both at stream and stream_op |
| * level */ |
| struct op_state state; |
| |
| /* OP storage */ |
| struct op_storage storage; |
| |
| /* Mutex to protect storage */ |
| gpr_mu mu; |
| |
| /* Refcount object of the stream */ |
| grpc_stream_refcount* refcount; |
| }; |
| |
| #ifndef NDEBUG |
| #define GRPC_CRONET_STREAM_REF(stream, reason) \ |
| grpc_cronet_stream_ref((stream), (reason)) |
| #define GRPC_CRONET_STREAM_UNREF(stream, reason) \ |
| grpc_cronet_stream_unref((stream), (reason)) |
| void grpc_cronet_stream_ref(stream_obj* s, const char* reason) { |
| grpc_stream_ref(s->refcount, reason); |
| } |
| void grpc_cronet_stream_unref(stream_obj* s, const char* reason) { |
| grpc_stream_unref(s->refcount, reason); |
| } |
| #else |
| #define GRPC_CRONET_STREAM_REF(stream, reason) grpc_cronet_stream_ref((stream)) |
| #define GRPC_CRONET_STREAM_UNREF(stream, reason) \ |
| grpc_cronet_stream_unref((stream)) |
| void grpc_cronet_stream_ref(stream_obj* s) { grpc_stream_ref(s->refcount); } |
| void grpc_cronet_stream_unref(stream_obj* s) { grpc_stream_unref(s->refcount); } |
| #endif |
| |
| static enum e_op_result execute_stream_op(struct op_and_state* oas); |
| |
| /* |
| Utility function to translate enum into string for printing |
| */ |
| static const char* op_result_string(enum e_op_result i) { |
| switch (i) { |
| case ACTION_TAKEN_WITH_CALLBACK: |
| return "ACTION_TAKEN_WITH_CALLBACK"; |
| case ACTION_TAKEN_NO_CALLBACK: |
| return "ACTION_TAKEN_NO_CALLBACK"; |
| case NO_ACTION_POSSIBLE: |
| return "NO_ACTION_POSSIBLE"; |
| } |
| GPR_UNREACHABLE_CODE(return "UNKNOWN"); |
| } |
| |
| static const char* op_id_string(enum e_op_id i) { |
| switch (i) { |
| case OP_SEND_INITIAL_METADATA: |
| return "OP_SEND_INITIAL_METADATA"; |
| case OP_SEND_MESSAGE: |
| return "OP_SEND_MESSAGE"; |
| case OP_SEND_TRAILING_METADATA: |
| return "OP_SEND_TRAILING_METADATA"; |
| case OP_RECV_MESSAGE: |
| return "OP_RECV_MESSAGE"; |
| case OP_RECV_INITIAL_METADATA: |
| return "OP_RECV_INITIAL_METADATA"; |
| case OP_RECV_TRAILING_METADATA: |
| return "OP_RECV_TRAILING_METADATA"; |
| case OP_CANCEL_ERROR: |
| return "OP_CANCEL_ERROR"; |
| case OP_ON_COMPLETE: |
| return "OP_ON_COMPLETE"; |
| case OP_FAILED: |
| return "OP_FAILED"; |
| case OP_SUCCEEDED: |
| return "OP_SUCCEEDED"; |
| case OP_CANCELED: |
| return "OP_CANCELED"; |
| case OP_RECV_MESSAGE_AND_ON_COMPLETE: |
| return "OP_RECV_MESSAGE_AND_ON_COMPLETE"; |
| case OP_READ_REQ_MADE: |
| return "OP_READ_REQ_MADE"; |
| case OP_NUM_OPS: |
| return "OP_NUM_OPS"; |
| } |
| return "UNKNOWN"; |
| } |
| |
| static void null_and_maybe_free_read_buffer(stream_obj* s) { |
| if (s->state.rs.read_buffer && |
| s->state.rs.read_buffer != s->state.rs.grpc_header_bytes) { |
| gpr_free(s->state.rs.read_buffer); |
| } |
| s->state.rs.read_buffer = nullptr; |
| } |
| |
| static void maybe_flush_read(stream_obj* s) { |
| /* To enter flush read state (discarding all the buffered messages in |
| * transport layer), two conditions must be satisfied: 1) non-zero grpc status |
| * has been received, and 2) an op requesting the status code |
| * (RECV_TRAILING_METADATA) is issued by the user. (See |
| * doc/status_ordering.md) */ |
| /* Whenever the evaluation of any of the two condition is changed, we check |
| * whether we should enter the flush read state. */ |
| if (s->state.pending_recv_trailing_metadata && s->state.fail_state) { |
| if (!s->state.flush_read && !s->state.rs.read_stream_closed) { |
| CRONET_LOG(GPR_DEBUG, "%p: Flush read", s); |
| s->state.flush_read = true; |
| null_and_maybe_free_read_buffer(s); |
| s->state.rs.read_buffer = |
| static_cast<char*>(gpr_malloc(GRPC_FLUSH_READ_SIZE)); |
| if (!s->state.pending_read_from_cronet) { |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); |
| bidirectional_stream_read(s->cbs, s->state.rs.read_buffer, |
| GRPC_FLUSH_READ_SIZE); |
| s->state.pending_read_from_cronet = true; |
| } |
| } |
| } |
| } |
| |
| static grpc_error* make_error_with_desc(int error_code, const char* desc) { |
| grpc_error* error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(desc); |
| error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, error_code); |
| return error; |
| } |
| |
| inline op_and_state::op_and_state(stream_obj* s, |
| const grpc_transport_stream_op_batch& op) |
| : op(op), state(s->arena), s(s), next(s->storage.head) {} |
| |
| /* |
| Add a new stream op to op storage. |
| */ |
| static void add_to_storage(struct stream_obj* s, |
| grpc_transport_stream_op_batch* op) { |
| struct op_storage* storage = &s->storage; |
| /* add new op at the beginning of the linked list. The memory is freed |
| in remove_from_storage */ |
| op_and_state* new_op = grpc_core::New<op_and_state>(s, *op); |
| gpr_mu_lock(&s->mu); |
| storage->head = new_op; |
| storage->num_pending_ops++; |
| if (op->send_message) { |
| s->state.pending_send_message = true; |
| } |
| if (op->recv_trailing_metadata) { |
| s->state.pending_recv_trailing_metadata = true; |
| maybe_flush_read(s); |
| } |
| CRONET_LOG(GPR_DEBUG, "adding new op %p. %d in the queue.", new_op, |
| storage->num_pending_ops); |
| gpr_mu_unlock(&s->mu); |
| } |
| |
| /* |
| Traverse the linked list and delete op and free memory |
| */ |
| static void remove_from_storage(struct stream_obj* s, |
| struct op_and_state* oas) { |
| struct op_and_state* curr; |
| if (s->storage.head == nullptr || oas == nullptr) { |
| return; |
| } |
| if (s->storage.head == oas) { |
| s->storage.head = oas->next; |
| grpc_core::Delete(oas); |
| s->storage.num_pending_ops--; |
| CRONET_LOG(GPR_DEBUG, "Freed %p. Now %d in the queue", oas, |
| s->storage.num_pending_ops); |
| } else { |
| for (curr = s->storage.head; curr != nullptr; curr = curr->next) { |
| if (curr->next == oas) { |
| curr->next = oas->next; |
| s->storage.num_pending_ops--; |
| CRONET_LOG(GPR_DEBUG, "Freed %p. Now %d in the queue", oas, |
| s->storage.num_pending_ops); |
| grpc_core::Delete(oas); |
| break; |
| } else if (GPR_UNLIKELY(curr->next == nullptr)) { |
| CRONET_LOG(GPR_ERROR, "Reached end of LL and did not find op to free"); |
| } |
| } |
| } |
| } |
| |
| /* |
| Cycle through ops and try to take next action. Break when either |
| an action with callback is taken, or no action is possible. |
| This can get executed from the Cronet network thread via cronet callback |
| or on the application supplied thread via the perform_stream_op function. |
| */ |
| static void execute_from_storage(stream_obj* s) { |
| gpr_mu_lock(&s->mu); |
| for (struct op_and_state* curr = s->storage.head; curr != nullptr;) { |
| CRONET_LOG(GPR_DEBUG, "calling op at %p. done = %d", curr, curr->done); |
| GPR_ASSERT(curr->done == 0); |
| enum e_op_result result = execute_stream_op(curr); |
| CRONET_LOG(GPR_DEBUG, "execute_stream_op[%p] returns %s", curr, |
| op_result_string(result)); |
| /* if this op is done, then remove it and free memory */ |
| if (curr->done) { |
| struct op_and_state* next = curr->next; |
| remove_from_storage(s, curr); |
| curr = next; |
| } |
| /* continue processing the same op if ACTION_TAKEN_WITHOUT_CALLBACK */ |
| if (result == NO_ACTION_POSSIBLE) { |
| curr = curr->next; |
| } else if (result == ACTION_TAKEN_WITH_CALLBACK) { |
| break; |
| } |
| } |
| gpr_mu_unlock(&s->mu); |
| } |
| |
| static void convert_cronet_array_to_metadata( |
| const bidirectional_stream_header_array* header_array, |
| grpc_chttp2_incoming_metadata_buffer* mds) { |
| for (size_t i = 0; i < header_array->count; i++) { |
| CRONET_LOG(GPR_DEBUG, "header key=%s, value=%s", |
| header_array->headers[i].key, header_array->headers[i].value); |
| grpc_slice key = grpc_slice_intern( |
| grpc_slice_from_static_string(header_array->headers[i].key)); |
| grpc_slice value; |
| if (grpc_is_binary_header(key)) { |
| value = grpc_slice_from_static_string(header_array->headers[i].value); |
| value = grpc_slice_intern(grpc_chttp2_base64_decode_with_length( |
| value, grpc_chttp2_base64_infer_length_after_decode(value))); |
| } else { |
| value = grpc_slice_intern( |
| grpc_slice_from_static_string(header_array->headers[i].value)); |
| } |
| GRPC_LOG_IF_ERROR("convert_cronet_array_to_metadata", |
| grpc_chttp2_incoming_metadata_buffer_add( |
| mds, grpc_mdelem_from_slices(key, value))); |
| } |
| } |
| |
| /* |
| Cronet callback |
| */ |
| static void on_failed(bidirectional_stream* stream, int net_error) { |
| gpr_log(GPR_ERROR, "on_failed(%p, %d)", stream, net_error); |
| grpc_core::ExecCtx exec_ctx; |
| |
| stream_obj* s = static_cast<stream_obj*>(stream->annotation); |
| gpr_mu_lock(&s->mu); |
| bidirectional_stream_destroy(s->cbs); |
| s->state.state_callback_received[OP_FAILED] = true; |
| s->cbs = nullptr; |
| if (s->header_array.headers) { |
| gpr_free(s->header_array.headers); |
| s->header_array.headers = nullptr; |
| } |
| if (s->state.ws.write_buffer) { |
| gpr_free(s->state.ws.write_buffer); |
| s->state.ws.write_buffer = nullptr; |
| } |
| null_and_maybe_free_read_buffer(s); |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| GRPC_CRONET_STREAM_UNREF(s, "cronet transport"); |
| } |
| |
| /* |
| Cronet callback |
| */ |
| static void on_canceled(bidirectional_stream* stream) { |
| CRONET_LOG(GPR_DEBUG, "on_canceled(%p)", stream); |
| grpc_core::ExecCtx exec_ctx; |
| |
| stream_obj* s = static_cast<stream_obj*>(stream->annotation); |
| gpr_mu_lock(&s->mu); |
| bidirectional_stream_destroy(s->cbs); |
| s->state.state_callback_received[OP_CANCELED] = true; |
| s->cbs = nullptr; |
| if (s->header_array.headers) { |
| gpr_free(s->header_array.headers); |
| s->header_array.headers = nullptr; |
| } |
| if (s->state.ws.write_buffer) { |
| gpr_free(s->state.ws.write_buffer); |
| s->state.ws.write_buffer = nullptr; |
| } |
| null_and_maybe_free_read_buffer(s); |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| GRPC_CRONET_STREAM_UNREF(s, "cronet transport"); |
| } |
| |
| /* |
| Cronet callback |
| */ |
| static void on_succeeded(bidirectional_stream* stream) { |
| CRONET_LOG(GPR_DEBUG, "on_succeeded(%p)", stream); |
| grpc_core::ExecCtx exec_ctx; |
| |
| stream_obj* s = static_cast<stream_obj*>(stream->annotation); |
| gpr_mu_lock(&s->mu); |
| bidirectional_stream_destroy(s->cbs); |
| s->state.state_callback_received[OP_SUCCEEDED] = true; |
| s->cbs = nullptr; |
| null_and_maybe_free_read_buffer(s); |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| GRPC_CRONET_STREAM_UNREF(s, "cronet transport"); |
| } |
| |
| /* |
| Cronet callback |
| */ |
| static void on_stream_ready(bidirectional_stream* stream) { |
| CRONET_LOG(GPR_DEBUG, "W: on_stream_ready(%p)", stream); |
| grpc_core::ExecCtx exec_ctx; |
| stream_obj* s = static_cast<stream_obj*>(stream->annotation); |
| grpc_cronet_transport* t = s->curr_ct; |
| gpr_mu_lock(&s->mu); |
| s->state.state_op_done[OP_SEND_INITIAL_METADATA] = true; |
| s->state.state_callback_received[OP_SEND_INITIAL_METADATA] = true; |
| /* Free the memory allocated for headers */ |
| if (s->header_array.headers) { |
| gpr_free(s->header_array.headers); |
| s->header_array.headers = nullptr; |
| } |
| /* Send the initial metadata on wire if there is no SEND_MESSAGE or |
| * SEND_TRAILING_METADATA ops pending */ |
| if (t->use_packet_coalescing) { |
| if (s->state.flush_cronet_when_ready) { |
| CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs); |
| bidirectional_stream_flush(stream); |
| } |
| } |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| } |
| |
| /* |
| Cronet callback |
| */ |
| static void on_response_headers_received( |
| bidirectional_stream* stream, |
| const bidirectional_stream_header_array* headers, |
| const char* negotiated_protocol) { |
| grpc_core::ExecCtx exec_ctx; |
| CRONET_LOG(GPR_DEBUG, "R: on_response_headers_received(%p, %p, %s)", stream, |
| headers, negotiated_protocol); |
| stream_obj* s = static_cast<stream_obj*>(stream->annotation); |
| |
| /* Identify if this is a header or a trailer (in a trailer-only response case) |
| */ |
| for (size_t i = 0; i < headers->count; i++) { |
| if (0 == strcmp("grpc-status", headers->headers[i].key)) { |
| on_response_trailers_received(stream, headers); |
| return; |
| } |
| } |
| |
| gpr_mu_lock(&s->mu); |
| convert_cronet_array_to_metadata(headers, &s->state.rs.initial_metadata); |
| s->state.state_callback_received[OP_RECV_INITIAL_METADATA] = true; |
| if (!(s->state.state_op_done[OP_CANCEL_ERROR] || |
| s->state.state_callback_received[OP_FAILED])) { |
| /* Do an extra read to trigger on_succeeded() callback in case connection |
| is closed */ |
| GPR_ASSERT(s->state.rs.length_field_received == false); |
| s->state.rs.read_buffer = s->state.rs.grpc_header_bytes; |
| s->state.rs.compressed = false; |
| s->state.rs.received_bytes = 0; |
| s->state.rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES; |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); |
| bidirectional_stream_read(s->cbs, s->state.rs.read_buffer, |
| s->state.rs.remaining_bytes); |
| s->state.pending_read_from_cronet = true; |
| } |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| } |
| |
| /* |
| Cronet callback |
| */ |
| static void on_write_completed(bidirectional_stream* stream, const char* data) { |
| grpc_core::ExecCtx exec_ctx; |
| stream_obj* s = static_cast<stream_obj*>(stream->annotation); |
| CRONET_LOG(GPR_DEBUG, "W: on_write_completed(%p, %s)", stream, data); |
| gpr_mu_lock(&s->mu); |
| if (s->state.ws.write_buffer) { |
| gpr_free(s->state.ws.write_buffer); |
| s->state.ws.write_buffer = nullptr; |
| } |
| s->state.state_callback_received[OP_SEND_MESSAGE] = true; |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| } |
| |
| /* |
| Cronet callback |
| */ |
| static void on_read_completed(bidirectional_stream* stream, char* data, |
| int count) { |
| grpc_core::ExecCtx exec_ctx; |
| stream_obj* s = static_cast<stream_obj*>(stream->annotation); |
| CRONET_LOG(GPR_DEBUG, "R: on_read_completed(%p, %p, %d)", stream, data, |
| count); |
| gpr_mu_lock(&s->mu); |
| s->state.pending_read_from_cronet = false; |
| s->state.state_callback_received[OP_RECV_MESSAGE] = true; |
| if (count > 0 && s->state.flush_read) { |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); |
| bidirectional_stream_read(s->cbs, s->state.rs.read_buffer, |
| GRPC_FLUSH_READ_SIZE); |
| s->state.pending_read_from_cronet = true; |
| gpr_mu_unlock(&s->mu); |
| } else if (count > 0) { |
| s->state.rs.received_bytes += count; |
| s->state.rs.remaining_bytes -= count; |
| if (s->state.rs.remaining_bytes > 0) { |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); |
| s->state.state_op_done[OP_READ_REQ_MADE] = true; |
| bidirectional_stream_read( |
| s->cbs, s->state.rs.read_buffer + s->state.rs.received_bytes, |
| s->state.rs.remaining_bytes); |
| s->state.pending_read_from_cronet = true; |
| gpr_mu_unlock(&s->mu); |
| } else { |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| } |
| } else { |
| null_and_maybe_free_read_buffer(s); |
| s->state.rs.read_stream_closed = true; |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| } |
| } |
| |
| /* |
| Cronet callback |
| */ |
| static void on_response_trailers_received( |
| bidirectional_stream* stream, |
| const bidirectional_stream_header_array* trailers) { |
| grpc_core::ExecCtx exec_ctx; |
| CRONET_LOG(GPR_DEBUG, "R: on_response_trailers_received(%p,%p)", stream, |
| trailers); |
| stream_obj* s = static_cast<stream_obj*>(stream->annotation); |
| grpc_cronet_transport* t = s->curr_ct; |
| gpr_mu_lock(&s->mu); |
| s->state.rs.trailing_metadata_valid = false; |
| convert_cronet_array_to_metadata(trailers, &s->state.rs.trailing_metadata); |
| if (trailers->count > 0) { |
| s->state.rs.trailing_metadata_valid = true; |
| } |
| for (size_t i = 0; i < trailers->count; i++) { |
| if (0 == strcmp(trailers->headers[i].key, "grpc-status") && |
| 0 != strcmp(trailers->headers[i].value, "0")) { |
| s->state.fail_state = true; |
| maybe_flush_read(s); |
| } |
| } |
| s->state.state_callback_received[OP_RECV_TRAILING_METADATA] = true; |
| /* Send a EOS when server terminates the stream (testServerFinishesRequest) to |
| * trigger on_succeeded */ |
| if (!s->state.state_op_done[OP_SEND_TRAILING_METADATA] && |
| !(s->state.state_op_done[OP_CANCEL_ERROR] || |
| s->state.state_callback_received[OP_FAILED])) { |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs); |
| s->state.state_callback_received[OP_SEND_MESSAGE] = false; |
| bidirectional_stream_write(s->cbs, "", 0, true); |
| if (t->use_packet_coalescing) { |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs); |
| bidirectional_stream_flush(s->cbs); |
| } |
| s->state.state_op_done[OP_SEND_TRAILING_METADATA] = true; |
| |
| gpr_mu_unlock(&s->mu); |
| } else { |
| gpr_mu_unlock(&s->mu); |
| execute_from_storage(s); |
| } |
| } |
| |
| /* |
| Utility function that takes the data from s->write_slice_buffer and assembles |
| into a contiguous byte stream with 5 byte gRPC header prepended. |
| */ |
| static void create_grpc_frame(grpc_slice_buffer* write_slice_buffer, |
| char** pp_write_buffer, |
| size_t* p_write_buffer_size, uint32_t flags) { |
| grpc_slice slice = grpc_slice_buffer_take_first(write_slice_buffer); |
| size_t length = GRPC_SLICE_LENGTH(slice); |
| *p_write_buffer_size = length + GRPC_HEADER_SIZE_IN_BYTES; |
| /* This is freed in the on_write_completed callback */ |
| char* write_buffer = |
| static_cast<char*>(gpr_malloc(length + GRPC_HEADER_SIZE_IN_BYTES)); |
| *pp_write_buffer = write_buffer; |
| uint8_t* p = reinterpret_cast<uint8_t*>(write_buffer); |
| /* Append 5 byte header */ |
| /* Compressed flag */ |
| *p++ = static_cast<uint8_t>((flags & GRPC_WRITE_INTERNAL_COMPRESS) ? 1 : 0); |
| /* Message length */ |
| *p++ = static_cast<uint8_t>(length >> 24); |
| *p++ = static_cast<uint8_t>(length >> 16); |
| *p++ = static_cast<uint8_t>(length >> 8); |
| *p++ = static_cast<uint8_t>(length); |
| /* append actual data */ |
| memcpy(p, GRPC_SLICE_START_PTR(slice), length); |
| grpc_slice_unref_internal(slice); |
| } |
| |
| /* |
| Convert metadata in a format that Cronet can consume |
| */ |
| static void convert_metadata_to_cronet_headers( |
| grpc_linked_mdelem* head, const char* host, char** pp_url, |
| bidirectional_stream_header** pp_headers, size_t* p_num_headers, |
| const char** method) { |
| grpc_linked_mdelem* curr = head; |
| /* Walk the linked list and get number of header fields */ |
| size_t num_headers_available = 0; |
| while (curr != nullptr) { |
| curr = curr->next; |
| num_headers_available++; |
| } |
| /* Allocate enough memory. It is freed in the on_stream_ready callback |
| */ |
| bidirectional_stream_header* headers = |
| static_cast<bidirectional_stream_header*>(gpr_malloc( |
| sizeof(bidirectional_stream_header) * num_headers_available)); |
| *pp_headers = headers; |
| |
| /* Walk the linked list again, this time copying the header fields. |
| s->num_headers can be less than num_headers_available, as some headers |
| are not used for cronet. |
| TODO (makdharma): Eliminate need to traverse the LL second time for perf. |
| */ |
| curr = head; |
| size_t num_headers = 0; |
| while (num_headers < num_headers_available) { |
| grpc_mdelem mdelem = curr->md; |
| curr = curr->next; |
| char* key = grpc_slice_to_c_string(GRPC_MDKEY(mdelem)); |
| char* value; |
| if (grpc_is_binary_header(GRPC_MDKEY(mdelem))) { |
| grpc_slice wire_value = grpc_chttp2_base64_encode(GRPC_MDVALUE(mdelem)); |
| value = grpc_slice_to_c_string(wire_value); |
| grpc_slice_unref_internal(wire_value); |
| } else { |
| value = grpc_slice_to_c_string(GRPC_MDVALUE(mdelem)); |
| } |
| if (grpc_slice_eq(GRPC_MDKEY(mdelem), GRPC_MDSTR_SCHEME) || |
| grpc_slice_eq(GRPC_MDKEY(mdelem), GRPC_MDSTR_AUTHORITY)) { |
| /* Cronet populates these fields on its own */ |
| gpr_free(key); |
| gpr_free(value); |
| continue; |
| } |
| if (grpc_slice_eq(GRPC_MDKEY(mdelem), GRPC_MDSTR_METHOD)) { |
| if (grpc_slice_eq(GRPC_MDVALUE(mdelem), GRPC_MDSTR_PUT)) { |
| *method = "PUT"; |
| } else { |
| /* POST method in default*/ |
| *method = "POST"; |
| } |
| gpr_free(key); |
| gpr_free(value); |
| continue; |
| } |
| if (grpc_slice_eq(GRPC_MDKEY(mdelem), GRPC_MDSTR_PATH)) { |
| /* Create URL by appending :path value to the hostname */ |
| gpr_asprintf(pp_url, "https://%s%s", host, value); |
| gpr_free(key); |
| gpr_free(value); |
| continue; |
| } |
| CRONET_LOG(GPR_DEBUG, "header %s = %s", key, value); |
| headers[num_headers].key = key; |
| headers[num_headers].value = value; |
| num_headers++; |
| if (curr == nullptr) { |
| break; |
| } |
| } |
| *p_num_headers = num_headers; |
| } |
| |
| static void parse_grpc_header(const uint8_t* data, int* length, |
| bool* compressed) { |
| const uint8_t c = *data; |
| const uint8_t* p = data + 1; |
| *compressed = ((c & 0x01) == 0x01); |
| *length = 0; |
| *length |= (*p++) << 24; |
| *length |= (*p++) << 16; |
| *length |= (*p++) << 8; |
| *length |= (*p++); |
| } |
| |
| static bool header_has_authority(grpc_linked_mdelem* head) { |
| while (head != nullptr) { |
| if (grpc_slice_eq(GRPC_MDKEY(head->md), GRPC_MDSTR_AUTHORITY)) { |
| return true; |
| } |
| head = head->next; |
| } |
| return false; |
| } |
| |
| /* |
| Op Execution: Decide if one of the actions contained in the stream op can be |
| executed. This is the heart of the state machine. |
| */ |
| static bool op_can_be_run(grpc_transport_stream_op_batch* curr_op, |
| struct stream_obj* s, struct op_state* op_state, |
| enum e_op_id op_id) { |
| struct op_state* stream_state = &s->state; |
| grpc_cronet_transport* t = s->curr_ct; |
| bool result = true; |
| /* When call is canceled, every op can be run, except under following |
| conditions |
| */ |
| bool is_canceled_or_failed = stream_state->state_op_done[OP_CANCEL_ERROR] || |
| stream_state->state_callback_received[OP_FAILED]; |
| if (is_canceled_or_failed) { |
| if (op_id == OP_SEND_INITIAL_METADATA) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| if (op_id == OP_SEND_MESSAGE) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| if (op_id == OP_SEND_TRAILING_METADATA) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| if (op_id == OP_CANCEL_ERROR) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| /* already executed */ |
| if (op_id == OP_RECV_INITIAL_METADATA && |
| stream_state->state_op_done[OP_RECV_INITIAL_METADATA]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| if (op_id == OP_RECV_MESSAGE && op_state->state_op_done[OP_RECV_MESSAGE]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| if (op_id == OP_RECV_TRAILING_METADATA && |
| stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| /* ON_COMPLETE can be processed if one of the following conditions is met: |
| * 1. the stream failed |
| * 2. the stream is cancelled, and the callback is received |
| * 3. the stream succeeded before cancel is effective |
| * 4. the stream is cancelled, and the stream is never started */ |
| if (op_id == OP_ON_COMPLETE && |
| !(stream_state->state_callback_received[OP_FAILED] || |
| stream_state->state_callback_received[OP_CANCELED] || |
| stream_state->state_callback_received[OP_SUCCEEDED] || |
| !stream_state->state_op_done[OP_SEND_INITIAL_METADATA])) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| } else if (op_id == OP_SEND_INITIAL_METADATA) { |
| /* already executed */ |
| if (stream_state->state_op_done[OP_SEND_INITIAL_METADATA]) result = false; |
| } else if (op_id == OP_RECV_INITIAL_METADATA) { |
| /* already executed */ |
| if (stream_state->state_op_done[OP_RECV_INITIAL_METADATA]) result = false; |
| /* we haven't sent headers yet. */ |
| else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) |
| result = false; |
| /* we haven't received headers yet. */ |
| else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA] && |
| !stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) |
| result = false; |
| } else if (op_id == OP_SEND_MESSAGE) { |
| /* already executed (note we're checking op specific state, not stream |
| state) */ |
| if (op_state->state_op_done[OP_SEND_MESSAGE]) result = false; |
| /* we haven't sent headers yet. */ |
| else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) |
| result = false; |
| } else if (op_id == OP_RECV_MESSAGE) { |
| /* already executed */ |
| if (op_state->state_op_done[OP_RECV_MESSAGE]) result = false; |
| /* we haven't received headers yet. */ |
| else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA] && |
| !stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) |
| result = false; |
| } else if (op_id == OP_RECV_TRAILING_METADATA) { |
| /* already executed */ |
| if (stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) result = false; |
| /* we have asked for but haven't received message yet. */ |
| else if (stream_state->state_op_done[OP_READ_REQ_MADE] && |
| !stream_state->state_op_done[OP_RECV_MESSAGE]) |
| result = false; |
| /* we haven't received trailers yet. */ |
| else if (!stream_state->state_callback_received[OP_RECV_TRAILING_METADATA]) |
| result = false; |
| /* we haven't received on_succeeded yet. */ |
| else if (!stream_state->state_callback_received[OP_SUCCEEDED]) |
| result = false; |
| } else if (op_id == OP_SEND_TRAILING_METADATA) { |
| /* already executed */ |
| if (stream_state->state_op_done[OP_SEND_TRAILING_METADATA]) result = false; |
| /* we haven't sent initial metadata yet */ |
| else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) |
| result = false; |
| /* we haven't sent message yet */ |
| else if (stream_state->pending_send_message && |
| !stream_state->state_op_done[OP_SEND_MESSAGE]) |
| result = false; |
| /* we haven't got on_write_completed for the send yet */ |
| else if (stream_state->state_op_done[OP_SEND_MESSAGE] && |
| !stream_state->state_callback_received[OP_SEND_MESSAGE] && |
| !(t->use_packet_coalescing && |
| stream_state->pending_write_for_trailer)) |
| result = false; |
| } else if (op_id == OP_CANCEL_ERROR) { |
| /* already executed */ |
| if (stream_state->state_op_done[OP_CANCEL_ERROR]) result = false; |
| } else if (op_id == OP_ON_COMPLETE) { |
| /* already executed (note we're checking op specific state, not stream |
| state) */ |
| if (op_state->state_op_done[OP_ON_COMPLETE]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| /* Check if every op that was asked for is done. */ |
| /* TODO(muxi): We should not consider the recv ops here, since they |
| * have their own callbacks. We should invoke a batch's on_complete |
| * as soon as all of the batch's send ops are complete, even if |
| * there are still recv ops pending. */ |
| else if (curr_op->send_initial_metadata && |
| !stream_state->state_callback_received[OP_SEND_INITIAL_METADATA]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } else if (curr_op->send_message && |
| !op_state->state_op_done[OP_SEND_MESSAGE]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } else if (curr_op->send_message && |
| !stream_state->state_callback_received[OP_SEND_MESSAGE]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } else if (curr_op->send_trailing_metadata && |
| !stream_state->state_op_done[OP_SEND_TRAILING_METADATA]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } else if (curr_op->recv_initial_metadata && |
| !stream_state->state_op_done[OP_RECV_INITIAL_METADATA]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } else if (curr_op->recv_message && |
| !op_state->state_op_done[OP_RECV_MESSAGE]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } else if (curr_op->cancel_stream && |
| !stream_state->state_callback_received[OP_CANCELED]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } else if (curr_op->recv_trailing_metadata) { |
| /* We aren't done with trailing metadata yet */ |
| if (!stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| /* We've asked for actual message in an earlier op, and it hasn't been |
| delivered yet. */ |
| else if (stream_state->state_op_done[OP_READ_REQ_MADE]) { |
| /* If this op is not the one asking for read, (which means some earlier |
| op has asked), and the read hasn't been delivered. */ |
| if (!curr_op->recv_message && |
| !stream_state->state_callback_received[OP_SUCCEEDED]) { |
| CRONET_LOG(GPR_DEBUG, "Because"); |
| result = false; |
| } |
| } |
| } |
| /* We should see at least one on_write_completed for the trailers that we |
| sent */ |
| else if (curr_op->send_trailing_metadata && |
| !stream_state->state_callback_received[OP_SEND_MESSAGE]) |
| result = false; |
| } |
| CRONET_LOG(GPR_DEBUG, "op_can_be_run %s : %s", op_id_string(op_id), |
| result ? "YES" : "NO"); |
| return result; |
| } |
| |
| /* |
| TODO (makdharma): Break down this function in smaller chunks for readability. |
| */ |
| static enum e_op_result execute_stream_op(struct op_and_state* oas) { |
| grpc_transport_stream_op_batch* stream_op = &oas->op; |
| struct stream_obj* s = oas->s; |
| grpc_cronet_transport* t = s->curr_ct; |
| struct op_state* stream_state = &s->state; |
| enum e_op_result result = NO_ACTION_POSSIBLE; |
| if (stream_op->send_initial_metadata && |
| op_can_be_run(stream_op, s, &oas->state, OP_SEND_INITIAL_METADATA)) { |
| CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_INITIAL_METADATA", oas); |
| /* Start new cronet stream. It is destroyed in on_succeeded, on_canceled, |
| * on_failed */ |
| GPR_ASSERT(s->cbs == nullptr); |
| GPR_ASSERT(!stream_state->state_op_done[OP_SEND_INITIAL_METADATA]); |
| s->cbs = |
| bidirectional_stream_create(t->engine, s->curr_gs, &cronet_callbacks); |
| CRONET_LOG(GPR_DEBUG, "%p = bidirectional_stream_create()", s->cbs); |
| if (t->use_packet_coalescing) { |
| bidirectional_stream_disable_auto_flush(s->cbs, true); |
| bidirectional_stream_delay_request_headers_until_flush(s->cbs, true); |
| } |
| char* url = nullptr; |
| const char* method = "POST"; |
| s->header_array.headers = nullptr; |
| convert_metadata_to_cronet_headers(stream_op->payload->send_initial_metadata |
| .send_initial_metadata->list.head, |
| t->host, &url, &s->header_array.headers, |
| &s->header_array.count, &method); |
| s->header_array.capacity = s->header_array.count; |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_start(%p, %s)", s->cbs, url); |
| bidirectional_stream_start(s->cbs, url, 0, method, &s->header_array, false); |
| if (url) { |
| gpr_free(url); |
| } |
| unsigned int header_index; |
| for (header_index = 0; header_index < s->header_array.count; |
| header_index++) { |
| gpr_free((void*)s->header_array.headers[header_index].key); |
| gpr_free((void*)s->header_array.headers[header_index].value); |
| } |
| stream_state->state_op_done[OP_SEND_INITIAL_METADATA] = true; |
| if (t->use_packet_coalescing) { |
| if (!stream_op->send_message && !stream_op->send_trailing_metadata) { |
| s->state.flush_cronet_when_ready = true; |
| } |
| } |
| result = ACTION_TAKEN_WITH_CALLBACK; |
| } else if (stream_op->send_message && |
| op_can_be_run(stream_op, s, &oas->state, OP_SEND_MESSAGE)) { |
| CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_MESSAGE", oas); |
| stream_state->pending_send_message = false; |
| if (stream_state->state_callback_received[OP_FAILED]) { |
| result = NO_ACTION_POSSIBLE; |
| CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed."); |
| } else { |
| grpc_slice_buffer write_slice_buffer; |
| grpc_slice slice; |
| grpc_slice_buffer_init(&write_slice_buffer); |
| if (1 != stream_op->payload->send_message.send_message->Next( |
| stream_op->payload->send_message.send_message->length(), |
| nullptr)) { |
| /* Should never reach here */ |
| GPR_ASSERT(false); |
| } |
| if (GRPC_ERROR_NONE != |
| stream_op->payload->send_message.send_message->Pull(&slice)) { |
| /* Should never reach here */ |
| GPR_ASSERT(false); |
| } |
| grpc_slice_buffer_add(&write_slice_buffer, slice); |
| if (GPR_UNLIKELY(write_slice_buffer.count != 1)) { |
| /* Empty request not handled yet */ |
| gpr_log(GPR_ERROR, "Empty request is not supported"); |
| GPR_ASSERT(write_slice_buffer.count == 1); |
| } |
| if (write_slice_buffer.count > 0) { |
| size_t write_buffer_size; |
| create_grpc_frame( |
| &write_slice_buffer, &stream_state->ws.write_buffer, |
| &write_buffer_size, |
| stream_op->payload->send_message.send_message->flags()); |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, %p)", s->cbs, |
| stream_state->ws.write_buffer); |
| stream_state->state_callback_received[OP_SEND_MESSAGE] = false; |
| bidirectional_stream_write(s->cbs, stream_state->ws.write_buffer, |
| static_cast<int>(write_buffer_size), false); |
| grpc_slice_buffer_destroy_internal(&write_slice_buffer); |
| if (t->use_packet_coalescing) { |
| if (!stream_op->send_trailing_metadata) { |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs); |
| bidirectional_stream_flush(s->cbs); |
| result = ACTION_TAKEN_WITH_CALLBACK; |
| } else { |
| stream_state->pending_write_for_trailer = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } |
| } else { |
| result = ACTION_TAKEN_WITH_CALLBACK; |
| } |
| } else { |
| result = NO_ACTION_POSSIBLE; |
| } |
| } |
| stream_state->state_op_done[OP_SEND_MESSAGE] = true; |
| oas->state.state_op_done[OP_SEND_MESSAGE] = true; |
| stream_op->payload->send_message.send_message.reset(); |
| } else if (stream_op->send_trailing_metadata && |
| op_can_be_run(stream_op, s, &oas->state, |
| OP_SEND_TRAILING_METADATA)) { |
| CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_TRAILING_METADATA", oas); |
| if (stream_state->state_callback_received[OP_FAILED]) { |
| result = NO_ACTION_POSSIBLE; |
| CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed."); |
| } else { |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs); |
| stream_state->state_callback_received[OP_SEND_MESSAGE] = false; |
| bidirectional_stream_write(s->cbs, "", 0, true); |
| if (t->use_packet_coalescing) { |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs); |
| bidirectional_stream_flush(s->cbs); |
| } |
| result = ACTION_TAKEN_WITH_CALLBACK; |
| } |
| stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true; |
| } else if (stream_op->recv_initial_metadata && |
| op_can_be_run(stream_op, s, &oas->state, |
| OP_RECV_INITIAL_METADATA)) { |
| CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_INITIAL_METADATA", oas); |
| if (stream_state->state_op_done[OP_CANCEL_ERROR]) { |
| GRPC_CLOSURE_SCHED( |
| stream_op->payload->recv_initial_metadata.recv_initial_metadata_ready, |
| GRPC_ERROR_NONE); |
| } else if (stream_state->state_callback_received[OP_FAILED]) { |
| GRPC_CLOSURE_SCHED( |
| stream_op->payload->recv_initial_metadata.recv_initial_metadata_ready, |
| GRPC_ERROR_NONE); |
| } else if (stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) { |
| GRPC_CLOSURE_SCHED( |
| stream_op->payload->recv_initial_metadata.recv_initial_metadata_ready, |
| GRPC_ERROR_NONE); |
| } else { |
| grpc_chttp2_incoming_metadata_buffer_publish( |
| &oas->s->state.rs.initial_metadata, |
| stream_op->payload->recv_initial_metadata.recv_initial_metadata); |
| GRPC_CLOSURE_SCHED( |
| stream_op->payload->recv_initial_metadata.recv_initial_metadata_ready, |
| GRPC_ERROR_NONE); |
| } |
| stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } else if (stream_op->recv_message && |
| op_can_be_run(stream_op, s, &oas->state, OP_RECV_MESSAGE)) { |
| CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_MESSAGE", oas); |
| if (stream_state->state_op_done[OP_CANCEL_ERROR]) { |
| CRONET_LOG(GPR_DEBUG, "Stream is cancelled."); |
| GRPC_CLOSURE_SCHED(stream_op->payload->recv_message.recv_message_ready, |
| GRPC_ERROR_NONE); |
| stream_state->state_op_done[OP_RECV_MESSAGE] = true; |
| oas->state.state_op_done[OP_RECV_MESSAGE] = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } else if (stream_state->state_callback_received[OP_FAILED]) { |
| CRONET_LOG(GPR_DEBUG, "Stream failed."); |
| GRPC_CLOSURE_SCHED(stream_op->payload->recv_message.recv_message_ready, |
| GRPC_ERROR_NONE); |
| stream_state->state_op_done[OP_RECV_MESSAGE] = true; |
| oas->state.state_op_done[OP_RECV_MESSAGE] = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } else if (stream_state->rs.read_stream_closed == true) { |
| /* No more data will be received */ |
| CRONET_LOG(GPR_DEBUG, "read stream closed"); |
| GRPC_CLOSURE_SCHED(stream_op->payload->recv_message.recv_message_ready, |
| GRPC_ERROR_NONE); |
| stream_state->state_op_done[OP_RECV_MESSAGE] = true; |
| oas->state.state_op_done[OP_RECV_MESSAGE] = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } else if (stream_state->flush_read) { |
| CRONET_LOG(GPR_DEBUG, "flush read"); |
| GRPC_CLOSURE_SCHED(stream_op->payload->recv_message.recv_message_ready, |
| GRPC_ERROR_NONE); |
| stream_state->state_op_done[OP_RECV_MESSAGE] = true; |
| oas->state.state_op_done[OP_RECV_MESSAGE] = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } else if (stream_state->rs.length_field_received == false) { |
| if (stream_state->rs.received_bytes == GRPC_HEADER_SIZE_IN_BYTES && |
| stream_state->rs.remaining_bytes == 0) { |
| /* Start a read operation for data */ |
| stream_state->rs.length_field_received = true; |
| parse_grpc_header( |
| reinterpret_cast<const uint8_t*>(stream_state->rs.read_buffer), |
| &stream_state->rs.length_field, &stream_state->rs.compressed); |
| CRONET_LOG(GPR_DEBUG, "length field = %d", |
| stream_state->rs.length_field); |
| if (stream_state->rs.length_field > 0) { |
| stream_state->rs.read_buffer = static_cast<char*>( |
| gpr_malloc(static_cast<size_t>(stream_state->rs.length_field))); |
| GPR_ASSERT(stream_state->rs.read_buffer); |
| stream_state->rs.remaining_bytes = stream_state->rs.length_field; |
| stream_state->rs.received_bytes = 0; |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); |
| stream_state->state_op_done[OP_READ_REQ_MADE] = |
| true; /* Indicates that at least one read request has been made */ |
| bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer, |
| stream_state->rs.remaining_bytes); |
| stream_state->pending_read_from_cronet = true; |
| result = ACTION_TAKEN_WITH_CALLBACK; |
| } else { |
| stream_state->rs.remaining_bytes = 0; |
| CRONET_LOG(GPR_DEBUG, "read operation complete. Empty response."); |
| /* Clean up read_slice_buffer in case there is unread data. */ |
| grpc_slice_buffer_destroy_internal( |
| &stream_state->rs.read_slice_buffer); |
| grpc_slice_buffer_init(&stream_state->rs.read_slice_buffer); |
| uint32_t flags = 0; |
| if (stream_state->rs.compressed) { |
| flags |= GRPC_WRITE_INTERNAL_COMPRESS; |
| } |
| stream_state->rs.sbs.Init(&stream_state->rs.read_slice_buffer, flags); |
| stream_op->payload->recv_message.recv_message->reset( |
| stream_state->rs.sbs.get()); |
| GRPC_CLOSURE_SCHED( |
| stream_op->payload->recv_message.recv_message_ready, |
| GRPC_ERROR_NONE); |
| stream_state->state_op_done[OP_RECV_MESSAGE] = true; |
| oas->state.state_op_done[OP_RECV_MESSAGE] = true; |
| |
| /* Extra read to trigger on_succeed */ |
| stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes; |
| stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES; |
| stream_state->rs.received_bytes = 0; |
| stream_state->rs.compressed = false; |
| stream_state->rs.length_field_received = false; |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); |
| stream_state->state_op_done[OP_READ_REQ_MADE] = |
| true; /* Indicates that at least one read request has been made */ |
| bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer, |
| stream_state->rs.remaining_bytes); |
| stream_state->pending_read_from_cronet = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } |
| } else if (stream_state->rs.remaining_bytes == 0) { |
| /* Start a read operation for first 5 bytes (GRPC header) */ |
| stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes; |
| stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES; |
| stream_state->rs.received_bytes = 0; |
| stream_state->rs.compressed = false; |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); |
| stream_state->state_op_done[OP_READ_REQ_MADE] = |
| true; /* Indicates that at least one read request has been made */ |
| bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer, |
| stream_state->rs.remaining_bytes); |
| stream_state->pending_read_from_cronet = true; |
| result = ACTION_TAKEN_WITH_CALLBACK; |
| } else { |
| result = NO_ACTION_POSSIBLE; |
| } |
| } else if (stream_state->rs.remaining_bytes == 0) { |
| CRONET_LOG(GPR_DEBUG, "read operation complete"); |
| grpc_slice read_data_slice = |
| GRPC_SLICE_MALLOC((uint32_t)stream_state->rs.length_field); |
| uint8_t* dst_p = GRPC_SLICE_START_PTR(read_data_slice); |
| memcpy(dst_p, stream_state->rs.read_buffer, |
| static_cast<size_t>(stream_state->rs.length_field)); |
| null_and_maybe_free_read_buffer(s); |
| /* Clean up read_slice_buffer in case there is unread data. */ |
| grpc_slice_buffer_destroy_internal(&stream_state->rs.read_slice_buffer); |
| grpc_slice_buffer_init(&stream_state->rs.read_slice_buffer); |
| grpc_slice_buffer_add(&stream_state->rs.read_slice_buffer, |
| read_data_slice); |
| uint32_t flags = 0; |
| if (stream_state->rs.compressed) { |
| flags = GRPC_WRITE_INTERNAL_COMPRESS; |
| } |
| stream_state->rs.sbs.Init(&stream_state->rs.read_slice_buffer, flags); |
| stream_op->payload->recv_message.recv_message->reset( |
| stream_state->rs.sbs.get()); |
| GRPC_CLOSURE_SCHED(stream_op->payload->recv_message.recv_message_ready, |
| GRPC_ERROR_NONE); |
| stream_state->state_op_done[OP_RECV_MESSAGE] = true; |
| oas->state.state_op_done[OP_RECV_MESSAGE] = true; |
| /* Do an extra read to trigger on_succeeded() callback in case connection |
| is closed */ |
| stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes; |
| stream_state->rs.compressed = false; |
| stream_state->rs.received_bytes = 0; |
| stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES; |
| stream_state->rs.length_field_received = false; |
| CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs); |
| bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer, |
| stream_state->rs.remaining_bytes); |
| stream_state->pending_read_from_cronet = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } |
| } else if (stream_op->recv_trailing_metadata && |
| op_can_be_run(stream_op, s, &oas->state, |
| OP_RECV_TRAILING_METADATA)) { |
| CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_TRAILING_METADATA", oas); |
| grpc_error* error = GRPC_ERROR_NONE; |
| if (stream_state->state_op_done[OP_CANCEL_ERROR]) { |
| error = GRPC_ERROR_REF(stream_state->cancel_error); |
| } else if (stream_state->state_callback_received[OP_FAILED]) { |
| error = make_error_with_desc(GRPC_STATUS_UNAVAILABLE, "Unavailable."); |
| } else if (oas->s->state.rs.trailing_metadata_valid) { |
| grpc_chttp2_incoming_metadata_buffer_publish( |
| &oas->s->state.rs.trailing_metadata, |
| stream_op->payload->recv_trailing_metadata.recv_trailing_metadata); |
| stream_state->rs.trailing_metadata_valid = false; |
| } |
| GRPC_CLOSURE_SCHED( |
| stream_op->payload->recv_trailing_metadata.recv_trailing_metadata_ready, |
| error); |
| stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true; |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } else if (stream_op->cancel_stream && |
| op_can_be_run(stream_op, s, &oas->state, OP_CANCEL_ERROR)) { |
| CRONET_LOG(GPR_DEBUG, "running: %p OP_CANCEL_ERROR", oas); |
| if (s->cbs) { |
| CRONET_LOG(GPR_DEBUG, "W: bidirectional_stream_cancel(%p)", s->cbs); |
| bidirectional_stream_cancel(s->cbs); |
| result = ACTION_TAKEN_WITH_CALLBACK; |
| } else { |
| result = ACTION_TAKEN_NO_CALLBACK; |
| } |
| stream_state->state_op_done[OP_CANCEL_ERROR] = true; |
| if (!stream_state->cancel_error) { |
| stream_state->cancel_error = |
| GRPC_ERROR_REF(stream_op->payload->cancel_stream.cancel_error); |
| } |
| } else if (stream_op->on_complete && |
| op_can_be_run(stream_op, s, &oas->state, OP_ON_COMPLETE)) { |
| CRONET_LOG(GPR_DEBUG, "running: %p OP_ON_COMPLETE", oas); |
| if (stream_state->state_op_done[OP_CANCEL_ERROR]) { |
| GRPC_CLOSURE_SCHED(stream_op->on_complete, |
| GRPC_ERROR_REF(stream_state->cancel_error)); |
| } else if (stream_state->state_callback_received[OP_FAILED]) { |
| GRPC_CLOSURE_SCHED( |
| stream_op->on_complete, |
| make_error_with_desc(GRPC_STATUS_UNAVAILABLE, "Unavailable.")); |
| } else { |
| /* All actions in this stream_op are complete. Call the on_complete |
| * callback |
| */ |
| GRPC_CLOSURE_SCHED(stream_op->on_complete, GRPC_ERROR_NONE); |
| } |
| oas->state.state_op_done[OP_ON_COMPLETE] = true; |
| oas->done = true; |
| /* reset any send message state, only if this ON_COMPLETE is about a send. |
| */ |
| if (stream_op->send_message) { |
| stream_state->state_callback_received[OP_SEND_MESSAGE] = false; |
| stream_state->state_op_done[OP_SEND_MESSAGE] = false; |
| } |
| result = ACTION_TAKEN_NO_CALLBACK; |
| /* If this is the on_complete callback being called for a received message - |
| make a note */ |
| if (stream_op->recv_message) |
| stream_state->state_op_done[OP_RECV_MESSAGE_AND_ON_COMPLETE] = true; |
| } else { |
| result = NO_ACTION_POSSIBLE; |
| } |
| return result; |
| } |
| |
| /* |
| Functions used by upper layers to access transport functionality. |
| */ |
| |
| inline stream_obj::stream_obj(grpc_transport* gt, grpc_stream* gs, |
| grpc_stream_refcount* refcount, gpr_arena* arena) |
| : arena(arena), |
| curr_ct(reinterpret_cast<grpc_cronet_transport*>(gt)), |
| curr_gs(gs), |
| state(arena), |
| refcount(refcount) { |
| GRPC_CRONET_STREAM_REF(this, "cronet transport"); |
| gpr_mu_init(&mu); |
| } |
| |
| inline stream_obj::~stream_obj() { |
| null_and_maybe_free_read_buffer(this); |
| /* Clean up read_slice_buffer in case there is unread data. */ |
| grpc_slice_buffer_destroy_internal(&state.rs.read_slice_buffer); |
| GRPC_ERROR_UNREF(state.cancel_error); |
| } |
| |
| static int init_stream(grpc_transport* gt, grpc_stream* gs, |
| grpc_stream_refcount* refcount, const void* server_data, |
| gpr_arena* arena) { |
| new (gs) stream_obj(gt, gs, refcount, arena); |
| return 0; |
| } |
| |
| static void set_pollset_do_nothing(grpc_transport* gt, grpc_stream* gs, |
| grpc_pollset* pollset) {} |
| |
| static void set_pollset_set_do_nothing(grpc_transport* gt, grpc_stream* gs, |
| grpc_pollset_set* pollset_set) {} |
| |
| static void perform_stream_op(grpc_transport* gt, grpc_stream* gs, |
| grpc_transport_stream_op_batch* op) { |
| CRONET_LOG(GPR_DEBUG, "perform_stream_op"); |
| if (op->send_initial_metadata && |
| header_has_authority(op->payload->send_initial_metadata |
| .send_initial_metadata->list.head)) { |
| /* Cronet does not support :authority header field. We cancel the call when |
| this field is present in metadata */ |
| if (op->recv_initial_metadata) { |
| GRPC_CLOSURE_SCHED( |
| op->payload->recv_initial_metadata.recv_initial_metadata_ready, |
| GRPC_ERROR_CANCELLED); |
| } |
| if (op->recv_message) { |
| GRPC_CLOSURE_SCHED(op->payload->recv_message.recv_message_ready, |
| GRPC_ERROR_CANCELLED); |
| } |
| if (op->recv_trailing_metadata) { |
| GRPC_CLOSURE_SCHED( |
| op->payload->recv_trailing_metadata.recv_trailing_metadata_ready, |
| GRPC_ERROR_CANCELLED); |
| } |
| GRPC_CLOSURE_SCHED(op->on_complete, GRPC_ERROR_CANCELLED); |
| return; |
| } |
| stream_obj* s = reinterpret_cast<stream_obj*>(gs); |
| add_to_storage(s, op); |
| execute_from_storage(s); |
| } |
| |
| static void destroy_stream(grpc_transport* gt, grpc_stream* gs, |
| grpc_closure* then_schedule_closure) { |
| stream_obj* s = reinterpret_cast<stream_obj*>(gs); |
| s->~stream_obj(); |
| GRPC_CLOSURE_SCHED(then_schedule_closure, GRPC_ERROR_NONE); |
| } |
| |
| static void destroy_transport(grpc_transport* gt) {} |
| |
| static grpc_endpoint* get_endpoint(grpc_transport* gt) { return nullptr; } |
| |
| static void perform_op(grpc_transport* gt, grpc_transport_op* op) {} |
| |
| static const grpc_transport_vtable grpc_cronet_vtable = { |
| sizeof(stream_obj), |
| "cronet_http", |
| init_stream, |
| set_pollset_do_nothing, |
| set_pollset_set_do_nothing, |
| perform_stream_op, |
| perform_op, |
| destroy_stream, |
| destroy_transport, |
| get_endpoint}; |
| |
| grpc_transport* grpc_create_cronet_transport(void* engine, const char* target, |
| const grpc_channel_args* args, |
| void* reserved) { |
| grpc_cronet_transport* ct = static_cast<grpc_cronet_transport*>( |
| gpr_malloc(sizeof(grpc_cronet_transport))); |
| if (!ct) { |
| goto error; |
| } |
| ct->base.vtable = &grpc_cronet_vtable; |
| ct->engine = static_cast<stream_engine*>(engine); |
| ct->host = static_cast<char*>(gpr_malloc(strlen(target) + 1)); |
| if (!ct->host) { |
| goto error; |
| } |
| strcpy(ct->host, target); |
| |
| ct->use_packet_coalescing = true; |
| if (args) { |
| for (size_t i = 0; i < args->num_args; i++) { |
| if (0 == |
| strcmp(args->args[i].key, GRPC_ARG_USE_CRONET_PACKET_COALESCING)) { |
| if (GPR_UNLIKELY(args->args[i].type != GRPC_ARG_INTEGER)) { |
| gpr_log(GPR_ERROR, "%s ignored: it must be an integer", |
| GRPC_ARG_USE_CRONET_PACKET_COALESCING); |
| } else { |
| ct->use_packet_coalescing = (args->args[i].value.integer != 0); |
| } |
| } |
| } |
| } |
| |
| return &ct->base; |
| |
| error: |
| if (ct) { |
| if (ct->host) { |
| gpr_free(ct->host); |
| } |
| gpr_free(ct); |
| } |
| |
| return nullptr; |
| } |