diff options
Diffstat (limited to 'src/core/ext/filters')
16 files changed, 117 insertions, 105 deletions
diff --git a/src/core/ext/filters/client_channel/backup_poller.h b/src/core/ext/filters/client_channel/backup_poller.h index 7285b9b93e..8f132f968c 100644 --- a/src/core/ext/filters/client_channel/backup_poller.h +++ b/src/core/ext/filters/client_channel/backup_poller.h @@ -23,7 +23,6 @@ #include <grpc/grpc.h> #include "src/core/lib/channel/channel_stack.h" -#include "src/core/lib/iomgr/exec_ctx.h" /* Start polling \a interested_parties periodically in the timer thread */ void grpc_client_channel_start_backup_polling( diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index 09cc06e169..bf3911e5ee 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -798,7 +798,8 @@ typedef struct { grpc_linked_mdelem* send_initial_metadata_storage; grpc_metadata_batch send_initial_metadata; // For send_message. - grpc_caching_byte_stream send_message; + grpc_core::ManualConstructor<grpc_core::ByteStreamCache::CachingByteStream> + send_message; // For send_trailing_metadata. grpc_linked_mdelem* send_trailing_metadata_storage; grpc_metadata_batch send_trailing_metadata; @@ -808,7 +809,7 @@ typedef struct { bool trailing_metadata_available; // For intercepting recv_message. grpc_closure recv_message_ready; - grpc_byte_stream* recv_message; + grpc_core::OrphanablePtr<grpc_core::ByteStream> recv_message; // For intercepting recv_trailing_metadata. grpc_metadata_batch recv_trailing_metadata; grpc_transport_stream_stats collect_stats; @@ -914,12 +915,12 @@ typedef struct client_channel_call_data { gpr_atm* peer_string; // send_message // When we get a send_message op, we replace the original byte stream - // with a grpc_caching_byte_stream that caches the slices to a - // local buffer for use in retries. + // with a CachingByteStream that caches the slices to a local buffer for + // use in retries. // Note: We inline the cache for the first 3 send_message ops and use // dynamic allocation after that. This number was essentially picked // at random; it could be changed in the future to tune performance. - grpc_core::InlinedVector<grpc_byte_stream_cache*, 3> send_messages; + grpc_core::InlinedVector<grpc_core::ByteStreamCache*, 3> send_messages; // send_trailing_metadata bool seen_send_trailing_metadata; grpc_linked_mdelem* send_trailing_metadata_storage; @@ -964,10 +965,11 @@ static void maybe_cache_send_ops_for_batch(call_data* calld, } // Set up cache for send_message ops. if (batch->send_message) { - grpc_byte_stream_cache* cache = (grpc_byte_stream_cache*)gpr_arena_alloc( - calld->arena, sizeof(grpc_byte_stream_cache)); - grpc_byte_stream_cache_init(cache, - batch->payload->send_message.send_message); + grpc_core::ByteStreamCache* cache = + static_cast<grpc_core::ByteStreamCache*>( + gpr_arena_alloc(calld->arena, sizeof(grpc_core::ByteStreamCache))); + new (cache) grpc_core::ByteStreamCache( + std::move(batch->payload->send_message.send_message)); calld->send_messages.push_back(cache); } // Save metadata batch for send_trailing_metadata ops. @@ -1002,7 +1004,7 @@ static void free_cached_send_op_data_after_commit( "]", chand, calld, i); } - grpc_byte_stream_cache_destroy(calld->send_messages[i]); + calld->send_messages[i]->Destroy(); } if (retry_state->completed_send_trailing_metadata) { grpc_metadata_batch_destroy(&calld->send_trailing_metadata); @@ -1026,8 +1028,8 @@ static void free_cached_send_op_data_for_completed_batch( "]", chand, calld, retry_state->completed_send_message_count - 1); } - grpc_byte_stream_cache_destroy( - calld->send_messages[retry_state->completed_send_message_count - 1]); + calld->send_messages[retry_state->completed_send_message_count - 1] + ->Destroy(); } if (batch_data->batch.send_trailing_metadata) { grpc_metadata_batch_destroy(&calld->send_trailing_metadata); @@ -1079,7 +1081,7 @@ static void pending_batches_add(grpc_call_element* elem, if (batch->send_message) { calld->pending_send_message = true; calld->bytes_buffered_for_retry += - batch->payload->send_message.send_message->length; + batch->payload->send_message.send_message->length(); } if (batch->send_trailing_metadata) { calld->pending_send_trailing_metadata = true; @@ -1680,7 +1682,7 @@ static void invoke_recv_message_callback(void* arg, grpc_error* error) { GPR_ASSERT(pending != nullptr); // Return payload. *pending->batch->payload->recv_message.recv_message = - batch_data->recv_message; + std::move(batch_data->recv_message); // Update bookkeeping. // Note: Need to do this before invoking the callback, since invoking // the callback will result in yielding the call combiner. @@ -2124,13 +2126,13 @@ static void add_retriable_send_message_op( "chand=%p calld=%p: starting calld->send_messages[%" PRIuPTR "]", chand, calld, retry_state->started_send_message_count); } - grpc_byte_stream_cache* cache = + grpc_core::ByteStreamCache* cache = calld->send_messages[retry_state->started_send_message_count]; ++retry_state->started_send_message_count; - grpc_caching_byte_stream_init(&batch_data->send_message, cache); + batch_data->send_message.Init(cache); batch_data->batch.send_message = true; - batch_data->batch.payload->send_message.send_message = - &batch_data->send_message.base; + batch_data->batch.payload->send_message.send_message.reset( + batch_data->send_message.get()); } // Adds retriable send_trailing_metadata op to batch_data. diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc index cb39e4224e..47e1deef12 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc @@ -61,6 +61,7 @@ #include <grpc/support/port_platform.h> #include "src/core/lib/iomgr/sockaddr.h" +#include "src/core/lib/iomgr/socket_utils.h" #include <inttypes.h> #include <limits.h> @@ -417,20 +418,20 @@ void ParseServer(const grpc_grpclb_server* server, grpc_resolved_address* addr) { memset(addr, 0, sizeof(*addr)); if (server->drop) return; - const uint16_t netorder_port = htons((uint16_t)server->port); + const uint16_t netorder_port = grpc_htons((uint16_t)server->port); /* the addresses are given in binary format (a in(6)_addr struct) in * server->ip_address.bytes. */ const grpc_grpclb_ip_address* ip = &server->ip_address; if (ip->size == 4) { - addr->len = sizeof(struct sockaddr_in); - struct sockaddr_in* addr4 = (struct sockaddr_in*)&addr->addr; - addr4->sin_family = AF_INET; + addr->len = sizeof(grpc_sockaddr_in); + grpc_sockaddr_in* addr4 = reinterpret_cast<grpc_sockaddr_in*>(&addr->addr); + addr4->sin_family = GRPC_AF_INET; memcpy(&addr4->sin_addr, ip->bytes, ip->size); addr4->sin_port = netorder_port; } else if (ip->size == 16) { - addr->len = sizeof(struct sockaddr_in6); - struct sockaddr_in6* addr6 = (struct sockaddr_in6*)&addr->addr; - addr6->sin6_family = AF_INET6; + addr->len = sizeof(grpc_sockaddr_in6); + grpc_sockaddr_in6* addr6 = (grpc_sockaddr_in6*)&addr->addr; + addr6->sin6_family = GRPC_AF_INET6; memcpy(&addr6->sin6_addr, ip->bytes, ip->size); addr6->sin6_port = netorder_port; } diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.h b/src/core/ext/filters/client_channel/lb_policy_factory.h index b8bbd32072..6440258158 100644 --- a/src/core/ext/filters/client_channel/lb_policy_factory.h +++ b/src/core/ext/filters/client_channel/lb_policy_factory.h @@ -21,7 +21,6 @@ #include <grpc/support/port_platform.h> -#include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/iomgr/resolve_address.h" #include "src/core/ext/filters/client_channel/client_channel_factory.h" diff --git a/src/core/ext/filters/client_channel/lb_policy_registry.h b/src/core/ext/filters/client_channel/lb_policy_registry.h index 2283d848bd..2e9bb061ed 100644 --- a/src/core/ext/filters/client_channel/lb_policy_registry.h +++ b/src/core/ext/filters/client_channel/lb_policy_registry.h @@ -24,7 +24,6 @@ #include "src/core/ext/filters/client_channel/lb_policy_factory.h" #include "src/core/lib/gprpp/memory.h" #include "src/core/lib/gprpp/orphanable.h" -#include "src/core/lib/iomgr/exec_ctx.h" namespace grpc_core { diff --git a/src/core/ext/filters/client_channel/parse_address.cc b/src/core/ext/filters/client_channel/parse_address.cc index e78dc99e0b..92ea259cf0 100644 --- a/src/core/ext/filters/client_channel/parse_address.cc +++ b/src/core/ext/filters/client_channel/parse_address.cc @@ -20,6 +20,7 @@ #include "src/core/ext/filters/client_channel/parse_address.h" #include "src/core/lib/iomgr/sockaddr.h" +#include "src/core/lib/iomgr/socket_utils.h" #include <stdio.h> #include <string.h> @@ -71,10 +72,10 @@ bool grpc_parse_ipv4_hostport(const char* hostport, grpc_resolved_address* addr, if (!gpr_split_host_port(hostport, &host, &port)) return false; // Parse IP address. memset(addr, 0, sizeof(*addr)); - addr->len = sizeof(struct sockaddr_in); - struct sockaddr_in* in = reinterpret_cast<struct sockaddr_in*>(addr->addr); - in->sin_family = AF_INET; - if (inet_pton(AF_INET, host, &in->sin_addr) == 0) { + addr->len = sizeof(grpc_sockaddr_in); + grpc_sockaddr_in* in = reinterpret_cast<grpc_sockaddr_in*>(addr->addr); + in->sin_family = GRPC_AF_INET; + if (grpc_inet_pton(GRPC_AF_INET, host, &in->sin_addr) == 0) { if (log_errors) gpr_log(GPR_ERROR, "invalid ipv4 address: '%s'", host); goto done; } @@ -88,7 +89,7 @@ bool grpc_parse_ipv4_hostport(const char* hostport, grpc_resolved_address* addr, if (log_errors) gpr_log(GPR_ERROR, "invalid ipv4 port: '%s'", port); goto done; } - in->sin_port = htons(static_cast<uint16_t>(port_num)); + in->sin_port = grpc_htons(static_cast<uint16_t>(port_num)); success = true; done: gpr_free(host); @@ -117,19 +118,20 @@ bool grpc_parse_ipv6_hostport(const char* hostport, grpc_resolved_address* addr, if (!gpr_split_host_port(hostport, &host, &port)) return false; // Parse IP address. memset(addr, 0, sizeof(*addr)); - addr->len = sizeof(struct sockaddr_in6); - struct sockaddr_in6* in6 = reinterpret_cast<struct sockaddr_in6*>(addr->addr); - in6->sin6_family = AF_INET6; + addr->len = sizeof(grpc_sockaddr_in6); + grpc_sockaddr_in6* in6 = reinterpret_cast<grpc_sockaddr_in6*>(addr->addr); + in6->sin6_family = GRPC_AF_INET6; // Handle the RFC6874 syntax for IPv6 zone identifiers. char* host_end = static_cast<char*>(gpr_memrchr(host, '%', strlen(host))); if (host_end != nullptr) { GPR_ASSERT(host_end >= host); - char host_without_scope[INET6_ADDRSTRLEN]; + char host_without_scope[GRPC_INET6_ADDRSTRLEN]; size_t host_without_scope_len = static_cast<size_t>(host_end - host); uint32_t sin6_scope_id = 0; strncpy(host_without_scope, host, host_without_scope_len); host_without_scope[host_without_scope_len] = '\0'; - if (inet_pton(AF_INET6, host_without_scope, &in6->sin6_addr) == 0) { + if (grpc_inet_pton(GRPC_AF_INET6, host_without_scope, &in6->sin6_addr) == + 0) { gpr_log(GPR_ERROR, "invalid ipv6 address: '%s'", host_without_scope); goto done; } @@ -142,7 +144,7 @@ bool grpc_parse_ipv6_hostport(const char* hostport, grpc_resolved_address* addr, // Handle "sin6_scope_id" being type "u_long". See grpc issue #10027. in6->sin6_scope_id = sin6_scope_id; } else { - if (inet_pton(AF_INET6, host, &in6->sin6_addr) == 0) { + if (grpc_inet_pton(GRPC_AF_INET6, host, &in6->sin6_addr) == 0) { gpr_log(GPR_ERROR, "invalid ipv6 address: '%s'", host); goto done; } @@ -157,7 +159,7 @@ bool grpc_parse_ipv6_hostport(const char* hostport, grpc_resolved_address* addr, if (log_errors) gpr_log(GPR_ERROR, "invalid ipv6 port: '%s'", port); goto done; } - in6->sin6_port = htons(static_cast<uint16_t>(port_num)); + in6->sin6_port = grpc_htons(static_cast<uint16_t>(port_num)); success = true; done: gpr_free(host); diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc index aa93e5d8de..c63de3c509 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc @@ -440,6 +440,19 @@ class AresDnsResolverFactory : public ResolverFactory { } // namespace grpc_core +extern grpc_address_resolver_vtable* grpc_resolve_address_impl; +static grpc_address_resolver_vtable* default_resolver; + +static grpc_error* blocking_resolve_address_ares( + const char* name, const char* default_port, + grpc_resolved_addresses** addresses) { + return default_resolver->blocking_resolve_address(name, default_port, + addresses); +} + +static grpc_address_resolver_vtable ares_resolver = { + grpc_resolve_address_ares, blocking_resolve_address_ares}; + void grpc_resolver_dns_ares_init() { char* resolver_env = gpr_getenv("GRPC_DNS_RESOLVER"); /* TODO(zyc): Turn on c-ares based resolver by default after the address @@ -450,7 +463,8 @@ void grpc_resolver_dns_ares_init() { GRPC_LOG_IF_ERROR("ares_library_init() failed", error); return; } - grpc_resolve_address = grpc_resolve_address_ares; + default_resolver = grpc_resolve_address_impl; + grpc_set_resolver_impl(&ares_resolver); grpc_core::ResolverRegistry::Builder::RegisterResolverFactory( grpc_core::UniquePtr<grpc_core::ResolverFactory>( grpc_core::New<grpc_core::AresDnsResolverFactory>())); diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h index 0bc13e35f4..6239549534 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h @@ -22,7 +22,6 @@ #include <grpc/support/port_platform.h> #include <ares.h> -#include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/iomgr/pollset_set.h" typedef struct grpc_ares_ev_driver grpc_ares_ev_driver; diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h index bda9cd1729..3e8ea01e12 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h @@ -22,7 +22,6 @@ #include <grpc/support/port_platform.h> #include "src/core/ext/filters/client_channel/lb_policy_factory.h" -#include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/iomgr/resolve_address.h" diff --git a/src/core/ext/filters/client_channel/uri_parser.h b/src/core/ext/filters/client_channel/uri_parser.h index 1966da932b..d749f23308 100644 --- a/src/core/ext/filters/client_channel/uri_parser.h +++ b/src/core/ext/filters/client_channel/uri_parser.h @@ -22,7 +22,6 @@ #include <grpc/support/port_platform.h> #include <stddef.h> -#include "src/core/lib/iomgr/exec_ctx.h" typedef struct { char* scheme; diff --git a/src/core/ext/filters/deadline/deadline_filter.cc b/src/core/ext/filters/deadline/deadline_filter.cc index dda3b61108..27d3eac8d6 100644 --- a/src/core/ext/filters/deadline/deadline_filter.cc +++ b/src/core/ext/filters/deadline/deadline_filter.cc @@ -27,7 +27,6 @@ #include <grpc/support/time.h> #include "src/core/lib/channel/channel_stack_builder.h" -#include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/iomgr/timer.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/surface/channel_init.h" diff --git a/src/core/ext/filters/http/client/http_client_filter.cc b/src/core/ext/filters/http/client/http_client_filter.cc index 58aefd17c7..ae94ce47b9 100644 --- a/src/core/ext/filters/http/client/http_client_filter.cc +++ b/src/core/ext/filters/http/client/http_client_filter.cc @@ -20,9 +20,11 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/string_util.h> +#include <stdint.h> #include <string.h> #include "src/core/ext/filters/http/client/http_client_filter.h" #include "src/core/lib/gpr/string.h" +#include "src/core/lib/gprpp/manual_constructor.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/slice/b64.h" #include "src/core/lib/slice/percent_encoding.h" @@ -58,8 +60,9 @@ struct call_data { // State for handling send_message ops. grpc_transport_stream_op_batch* send_message_batch; size_t send_message_bytes_read; - grpc_byte_stream_cache send_message_cache; - grpc_caching_byte_stream send_message_caching_stream; + grpc_core::ManualConstructor<grpc_core::ByteStreamCache> send_message_cache; + grpc_core::ManualConstructor<grpc_core::ByteStreamCache::CachingByteStream> + send_message_caching_stream; grpc_closure on_send_message_next_done; grpc_closure* original_send_message_on_complete; grpc_closure send_message_on_complete; @@ -166,7 +169,7 @@ static void recv_trailing_metadata_on_complete(void* user_data, static void send_message_on_complete(void* arg, grpc_error* error) { grpc_call_element* elem = static_cast<grpc_call_element*>(arg); call_data* calld = static_cast<call_data*>(elem->call_data); - grpc_byte_stream_cache_destroy(&calld->send_message_cache); + calld->send_message_cache.Destroy(); GRPC_CLOSURE_RUN(calld->original_send_message_on_complete, GRPC_ERROR_REF(error)); } @@ -175,8 +178,7 @@ static void send_message_on_complete(void* arg, grpc_error* error) { // calld->send_message_bytes_read. static grpc_error* pull_slice_from_send_message(call_data* calld) { grpc_slice incoming_slice; - grpc_error* error = grpc_byte_stream_pull( - &calld->send_message_caching_stream.base, &incoming_slice); + grpc_error* error = calld->send_message_caching_stream->Pull(&incoming_slice); if (error == GRPC_ERROR_NONE) { calld->send_message_bytes_read += GRPC_SLICE_LENGTH(incoming_slice); grpc_slice_unref_internal(incoming_slice); @@ -186,24 +188,23 @@ static grpc_error* pull_slice_from_send_message(call_data* calld) { // Reads as many slices as possible from the send_message byte stream. // Upon successful return, if calld->send_message_bytes_read == -// calld->send_message_caching_stream.base.length, then we have completed +// calld->send_message_caching_stream->length(), then we have completed // reading from the byte stream; otherwise, an async read has been dispatched // and on_send_message_next_done() will be invoked when it is complete. static grpc_error* read_all_available_send_message_data(call_data* calld) { - while (grpc_byte_stream_next(&calld->send_message_caching_stream.base, - ~static_cast<size_t>(0), - &calld->on_send_message_next_done)) { + while (calld->send_message_caching_stream->Next( + SIZE_MAX, &calld->on_send_message_next_done)) { grpc_error* error = pull_slice_from_send_message(calld); if (error != GRPC_ERROR_NONE) return error; if (calld->send_message_bytes_read == - calld->send_message_caching_stream.base.length) { + calld->send_message_caching_stream->length()) { break; } } return GRPC_ERROR_NONE; } -// Async callback for grpc_byte_stream_next(). +// Async callback for ByteStream::Next(). static void on_send_message_next_done(void* arg, grpc_error* error) { grpc_call_element* elem = static_cast<grpc_call_element*>(arg); call_data* calld = static_cast<call_data*>(elem->call_data); @@ -222,7 +223,7 @@ static void on_send_message_next_done(void* arg, grpc_error* error) { // here, then we know that all of the data was not available // synchronously, so we were not able to do a cached call. Instead, // we just reset the byte stream and then send down the batch as-is. - grpc_caching_byte_stream_reset(&calld->send_message_caching_stream); + calld->send_message_caching_stream->Reset(); grpc_call_next_op(elem, calld->send_message_batch); } @@ -253,7 +254,7 @@ static grpc_error* update_path_for_get(grpc_call_element* elem, size_t estimated_len = GRPC_SLICE_LENGTH(path_slice); estimated_len++; /* for the '?' */ estimated_len += grpc_base64_estimate_encoded_size( - batch->payload->send_message.send_message->length, true /* url_safe */, + batch->payload->send_message.send_message->length(), true /* url_safe */, false /* multi_line */); grpc_slice path_with_query_slice = GRPC_SLICE_MALLOC(estimated_len); /* memcopy individual pieces into this slice */ @@ -265,9 +266,9 @@ static grpc_error* update_path_for_get(grpc_call_element* elem, write_ptr += GRPC_SLICE_LENGTH(path_slice); *write_ptr++ = '?'; char* payload_bytes = - slice_buffer_to_string(&calld->send_message_cache.cache_buffer); + slice_buffer_to_string(calld->send_message_cache->cache_buffer()); grpc_base64_encode_core(write_ptr, payload_bytes, - batch->payload->send_message.send_message->length, + batch->payload->send_message.send_message->length(), true /* url_safe */, false /* multi_line */); gpr_free(payload_bytes); /* remove trailing unused memory and add trailing 0 to terminate string */ @@ -326,15 +327,14 @@ static void hc_start_transport_stream_op_batch( if (batch->send_message && (batch->payload->send_initial_metadata.send_initial_metadata_flags & GRPC_INITIAL_METADATA_CACHEABLE_REQUEST) && - batch->payload->send_message.send_message->length < + batch->payload->send_message.send_message->length() < channeld->max_payload_size_for_get) { calld->send_message_bytes_read = 0; - grpc_byte_stream_cache_init(&calld->send_message_cache, - batch->payload->send_message.send_message); - grpc_caching_byte_stream_init(&calld->send_message_caching_stream, - &calld->send_message_cache); - batch->payload->send_message.send_message = - &calld->send_message_caching_stream.base; + calld->send_message_cache.Init( + std::move(batch->payload->send_message.send_message)); + calld->send_message_caching_stream.Init(calld->send_message_cache.get()); + batch->payload->send_message.send_message.reset( + calld->send_message_caching_stream.get()); calld->original_send_message_on_complete = batch->on_complete; batch->on_complete = &calld->send_message_on_complete; calld->send_message_batch = batch; @@ -342,12 +342,12 @@ static void hc_start_transport_stream_op_batch( if (error != GRPC_ERROR_NONE) goto done; // If all the data has been read, then we can use GET. if (calld->send_message_bytes_read == - calld->send_message_caching_stream.base.length) { + calld->send_message_caching_stream->length()) { method = GRPC_MDELEM_METHOD_GET; error = update_path_for_get(elem, batch); if (error != GRPC_ERROR_NONE) goto done; batch->send_message = false; - grpc_byte_stream_destroy(&calld->send_message_caching_stream.base); + calld->send_message_caching_stream->Orphan(); } else { // Not all data is available. The batch will be sent down // asynchronously in on_send_message_next_done(). diff --git a/src/core/ext/filters/http/message_compress/message_compress_filter.cc b/src/core/ext/filters/http/message_compress/message_compress_filter.cc index efe0085c5b..e7d9949386 100644 --- a/src/core/ext/filters/http/message_compress/message_compress_filter.cc +++ b/src/core/ext/filters/http/message_compress/message_compress_filter.cc @@ -32,6 +32,7 @@ #include "src/core/lib/compression/compression_internal.h" #include "src/core/lib/compression/message_compress.h" #include "src/core/lib/gpr/string.h" +#include "src/core/lib/gprpp/manual_constructor.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" @@ -62,7 +63,8 @@ struct call_data { grpc_closure start_send_message_batch_in_call_combiner; grpc_transport_stream_op_batch* send_message_batch; grpc_slice_buffer slices; /**< Buffers up input slices to be compressed */ - grpc_slice_buffer_stream replacement_stream; + grpc_core::ManualConstructor<grpc_core::SliceBufferByteStream> + replacement_stream; grpc_closure* original_send_message_on_complete; grpc_closure send_message_on_complete; grpc_closure on_send_message_next_done; @@ -220,7 +222,7 @@ static void finish_send_message(grpc_call_element* elem) { grpc_slice_buffer tmp; grpc_slice_buffer_init(&tmp); uint32_t send_flags = - calld->send_message_batch->payload->send_message.send_message->flags; + calld->send_message_batch->payload->send_message.send_message->flags(); bool did_compress = grpc_msg_compress(calld->message_compression_algorithm, &calld->slices, &tmp); if (did_compress) { @@ -253,12 +255,9 @@ static void finish_send_message(grpc_call_element* elem) { grpc_slice_buffer_destroy_internal(&tmp); // Swap out the original byte stream with our new one and send the // batch down. - grpc_byte_stream_destroy( - calld->send_message_batch->payload->send_message.send_message); - grpc_slice_buffer_stream_init(&calld->replacement_stream, &calld->slices, - send_flags); - calld->send_message_batch->payload->send_message.send_message = - &calld->replacement_stream.base; + calld->replacement_stream.Init(&calld->slices, send_flags); + calld->send_message_batch->payload->send_message.send_message.reset( + calld->replacement_stream.get()); calld->original_send_message_on_complete = calld->send_message_batch->on_complete; calld->send_message_batch->on_complete = &calld->send_message_on_complete; @@ -278,9 +277,9 @@ static void fail_send_message_batch_in_call_combiner(void* arg, // Pulls a slice from the send_message byte stream and adds it to calld->slices. static grpc_error* pull_slice_from_send_message(call_data* calld) { grpc_slice incoming_slice; - grpc_error* error = grpc_byte_stream_pull( - calld->send_message_batch->payload->send_message.send_message, - &incoming_slice); + grpc_error* error = + calld->send_message_batch->payload->send_message.send_message->Pull( + &incoming_slice); if (error == GRPC_ERROR_NONE) { grpc_slice_buffer_add(&calld->slices, incoming_slice); } @@ -289,12 +288,11 @@ static grpc_error* pull_slice_from_send_message(call_data* calld) { // Reads as many slices as possible from the send_message byte stream. // If all data has been read, invokes finish_send_message(). Otherwise, -// an async call to grpc_byte_stream_next() has been started, which will +// an async call to ByteStream::Next() has been started, which will // eventually result in calling on_send_message_next_done(). static void continue_reading_send_message(grpc_call_element* elem) { call_data* calld = static_cast<call_data*>(elem->call_data); - while (grpc_byte_stream_next( - calld->send_message_batch->payload->send_message.send_message, + while (calld->send_message_batch->payload->send_message.send_message->Next( ~static_cast<size_t>(0), &calld->on_send_message_next_done)) { grpc_error* error = pull_slice_from_send_message(calld); if (error != GRPC_ERROR_NONE) { @@ -303,15 +301,15 @@ static void continue_reading_send_message(grpc_call_element* elem) { GRPC_ERROR_UNREF(error); return; } - if (calld->slices.length == - calld->send_message_batch->payload->send_message.send_message->length) { + if (calld->slices.length == calld->send_message_batch->payload->send_message + .send_message->length()) { finish_send_message(elem); break; } } } -// Async callback for grpc_byte_stream_next(). +// Async callback for ByteStream::Next(). static void on_send_message_next_done(void* arg, grpc_error* error) { grpc_call_element* elem = static_cast<grpc_call_element*>(arg); call_data* calld = static_cast<call_data*>(elem->call_data); @@ -328,7 +326,7 @@ static void on_send_message_next_done(void* arg, grpc_error* error) { return; } if (calld->slices.length == - calld->send_message_batch->payload->send_message.send_message->length) { + calld->send_message_batch->payload->send_message.send_message->length()) { finish_send_message(elem); } else { continue_reading_send_message(elem); @@ -340,7 +338,8 @@ static void start_send_message_batch(void* arg, grpc_error* unused) { call_data* calld = static_cast<call_data*>(elem->call_data); if (skip_compression( elem, - calld->send_message_batch->payload->send_message.send_message->flags, + calld->send_message_batch->payload->send_message.send_message + ->flags(), calld->send_initial_metadata_state == HAS_COMPRESSION_ALGORITHM)) { send_message_batch_continue(elem); } else { @@ -365,9 +364,7 @@ static void compress_start_transport_stream_op_batch( grpc_schedule_on_exec_ctx), GRPC_ERROR_REF(calld->cancel_error), "failing send_message op"); } else { - grpc_byte_stream_shutdown( - - calld->send_message_batch->payload->send_message.send_message, + calld->send_message_batch->payload->send_message.send_message->Shutdown( GRPC_ERROR_REF(calld->cancel_error)); } } diff --git a/src/core/ext/filters/http/server/http_server_filter.cc b/src/core/ext/filters/http/server/http_server_filter.cc index 57ec8dce34..c202015875 100644 --- a/src/core/ext/filters/http/server/http_server_filter.cc +++ b/src/core/ext/filters/http/server/http_server_filter.cc @@ -23,6 +23,7 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <string.h> +#include "src/core/lib/gprpp/manual_constructor.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/slice/b64.h" #include "src/core/lib/slice/percent_encoding.h" @@ -53,8 +54,8 @@ struct call_data { */ grpc_closure* recv_message_ready; grpc_closure* on_complete; - grpc_byte_stream** pp_recv_message; - grpc_slice_buffer_stream read_stream; + grpc_core::OrphanablePtr<grpc_core::ByteStream>* pp_recv_message; + grpc_core::ManualConstructor<grpc_core::SliceBufferByteStream> read_stream; /** Receive closures are chained: we inject this closure as the on_done_recv up-call on transport_op, and remember to call our on_done_recv member @@ -232,7 +233,7 @@ static grpc_error* server_filter_incoming_metadata(grpc_call_element* elem, grpc_base64_decode_with_len( reinterpret_cast<const char*> GRPC_SLICE_START_PTR(query_slice), GRPC_SLICE_LENGTH(query_slice), k_url_safe)); - grpc_slice_buffer_stream_init(&calld->read_stream, &read_slice_buffer, 0); + calld->read_stream.Init(&read_slice_buffer, 0); grpc_slice_buffer_destroy_internal(&read_slice_buffer); calld->seen_path_with_query = true; grpc_slice_unref_internal(query_slice); @@ -281,10 +282,10 @@ static void hs_on_complete(void* user_data, grpc_error* err) { call_data* calld = static_cast<call_data*>(elem->call_data); /* Call recv_message_ready if we got the payload via the path field */ if (calld->seen_path_with_query && calld->recv_message_ready != nullptr) { - *calld->pp_recv_message = - calld->payload_bin_delivered - ? nullptr - : reinterpret_cast<grpc_byte_stream*>(&calld->read_stream); + calld->pp_recv_message->reset( + calld->payload_bin_delivered ? nullptr + : reinterpret_cast<grpc_core::ByteStream*>( + calld->read_stream.get())); // Re-enter call combiner for recv_message_ready, since the surface // code will release the call combiner for each callback it receives. GRPC_CALL_COMBINER_START(calld->call_combiner, calld->recv_message_ready, @@ -405,7 +406,7 @@ static void destroy_call_elem(grpc_call_element* elem, grpc_closure* ignored) { call_data* calld = static_cast<call_data*>(elem->call_data); if (calld->seen_path_with_query && !calld->payload_bin_delivered) { - grpc_byte_stream_destroy(&calld->read_stream.base); + calld->read_stream->Orphan(); } } diff --git a/src/core/ext/filters/message_size/message_size_filter.cc b/src/core/ext/filters/message_size/message_size_filter.cc index b1b14dde02..c7fc3f2e62 100644 --- a/src/core/ext/filters/message_size/message_size_filter.cc +++ b/src/core/ext/filters/message_size/message_size_filter.cc @@ -100,7 +100,7 @@ struct call_data { // call our next_recv_message_ready member after handling it. grpc_closure recv_message_ready; // Used by recv_message_ready. - grpc_byte_stream** recv_message; + grpc_core::OrphanablePtr<grpc_core::ByteStream>* recv_message; // Original recv_message_ready callback, invoked after our own. grpc_closure* next_recv_message_ready; }; @@ -121,12 +121,12 @@ static void recv_message_ready(void* user_data, grpc_error* error) { grpc_call_element* elem = static_cast<grpc_call_element*>(user_data); call_data* calld = static_cast<call_data*>(elem->call_data); if (*calld->recv_message != nullptr && calld->limits.max_recv_size >= 0 && - (*calld->recv_message)->length > + (*calld->recv_message)->length() > static_cast<size_t>(calld->limits.max_recv_size)) { char* message_string; gpr_asprintf(&message_string, "Received message larger than max (%u vs. %d)", - (*calld->recv_message)->length, calld->limits.max_recv_size); + (*calld->recv_message)->length(), calld->limits.max_recv_size); grpc_error* new_error = grpc_error_set_int( GRPC_ERROR_CREATE_FROM_COPIED_STRING(message_string), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED); @@ -150,11 +150,11 @@ static void start_transport_stream_op_batch( call_data* calld = static_cast<call_data*>(elem->call_data); // Check max send message size. if (op->send_message && calld->limits.max_send_size >= 0 && - op->payload->send_message.send_message->length > + op->payload->send_message.send_message->length() > static_cast<size_t>(calld->limits.max_send_size)) { char* message_string; gpr_asprintf(&message_string, "Sent message larger than max (%u vs. %d)", - op->payload->send_message.send_message->length, + op->payload->send_message.send_message->length(), calld->limits.max_send_size); grpc_transport_stream_op_batch_finish_with_failure( op, diff --git a/src/core/ext/filters/workarounds/workaround_cronet_compression_filter.cc b/src/core/ext/filters/workarounds/workaround_cronet_compression_filter.cc index bed1004c57..c7070d4d9b 100644 --- a/src/core/ext/filters/workarounds/workaround_cronet_compression_filter.cc +++ b/src/core/ext/filters/workarounds/workaround_cronet_compression_filter.cc @@ -93,7 +93,9 @@ static void start_transport_stream_op_batch( /* Send message happens after client's user-agent (initial metadata) is * received, so workaround_active must be set already */ if (calld->workaround_active) { - op->payload->send_message.send_message->flags |= GRPC_WRITE_NO_COMPRESS; + op->payload->send_message.send_message->set_flags( + op->payload->send_message.send_message->flags() | + GRPC_WRITE_NO_COMPRESS); } } |