Merge branch 'master' into fd_rw_atm_closure

pull/9683/head
Sree Kuchibhotla 8 years ago
commit 6e5cbc2095
  1. 1
      .gitignore
  2. 5
      .pylintrc
  3. 1
      BUILD
  4. 41
      CMakeLists.txt
  5. 50
      Makefile
  6. 20
      build.yaml
  7. 3
      gRPC-Core.podspec
  8. 4
      include/grpc/impl/codegen/grpc_types.h
  9. 20
      src/core/ext/transport/chttp2/transport/chttp2_transport.c
  10. 2
      src/core/ext/transport/chttp2/transport/frame_settings.h
  11. 54
      src/core/ext/transport/chttp2/transport/internal.h
  12. 18
      src/core/ext/transport/chttp2/transport/parsing.c
  13. 18
      src/core/ext/transport/chttp2/transport/writing.c
  14. 13
      src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
  15. 12
      src/core/ext/transport/cronet/transport/cronet_api_dummy.c
  16. 224
      src/core/ext/transport/cronet/transport/cronet_transport.c
  17. 43
      src/core/ext/transport/cronet/transport/cronet_transport.h
  18. 13
      src/core/lib/channel/http_server_filter.c
  19. 11
      src/core/lib/iomgr/ev_epoll_linux.c
  20. 11
      src/core/lib/iomgr/ev_poll_posix.c
  21. 4
      src/core/lib/iomgr/ev_posix.c
  22. 1
      src/core/lib/iomgr/ev_posix.h
  23. 3
      src/core/lib/iomgr/pollset.h
  24. 5
      src/core/lib/iomgr/pollset_uv.c
  25. 10
      src/core/lib/iomgr/pollset_windows.c
  26. 6
      src/core/lib/surface/channel.c
  27. 38
      src/core/lib/surface/completion_queue.c
  28. 3
      src/core/lib/surface/completion_queue.h
  29. 2
      src/core/lib/surface/init.c
  30. 291
      src/objective-c/tests/CronetUnitTests/CronetUnitTests.m
  31. 1
      src/objective-c/tests/Tests.xcodeproj/project.pbxproj
  32. 19
      src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme
  33. 7
      src/python/grpcio/grpc/_auth.py
  34. 28
      src/python/grpcio/grpc/_channel.py
  35. 23
      src/python/grpcio/grpc/_common.py
  36. 3
      templates/gRPC-Core.podspec.template
  37. 2
      test/core/end2end/gen_build_yaml.py
  38. 1
      test/core/internal_api_canaries/iomgr.c
  39. 196
      test/core/util/trickle_endpoint.c
  40. 46
      test/core/util/trickle_endpoint.h
  41. 356
      test/cpp/microbenchmarks/bm_closure.cc
  42. 187
      test/cpp/microbenchmarks/bm_fullstack.cc
  43. 2
      tools/codegen/core/gen_nano_proto.sh
  44. 2
      tools/distrib/yapf_code.sh
  45. 41
      tools/internal_ci/linux/grpc_interop.cfg
  46. 42
      tools/internal_ci/linux/grpc_interop.sh
  47. 21
      tools/profiling/microbenchmarks/bm2bq.py
  48. 29
      tools/run_tests/generated/sources_and_headers.json
  49. 22
      tools/run_tests/generated/tests.json
  50. 1
      tools/run_tests/python_utils/start_port_server.py
  51. 59
      tools/run_tests/run_microbenchmark.py
  52. 15
      tools/run_tests/run_tests.py
  53. 3
      vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj
  54. 6
      vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters
  55. 3
      vsprojects/vcxproj/grpc_test_util_unsecure/grpc_test_util_unsecure.vcxproj
  56. 6
      vsprojects/vcxproj/grpc_test_util_unsecure/grpc_test_util_unsecure.vcxproj.filters

1
.gitignore vendored

@ -9,6 +9,7 @@ objs
cython_debug/
python_build/
python_format_venv/
python_pylint_venv/
.coverage*
.eggs
htmlcov/

@ -29,10 +29,7 @@
#TODO: Enable too-many-return-statements
#TODO: Enable too-many-nested-blocks
#TODO: Enable super-init-not-called
#TODO: Enable simplifiable-if-statement
#TODO: Enable no-self-use
#TODO: Enable no-member
#TODO: Enable logging-format-interpolation
#TODO: Enable dangerous-default-value
disable=missing-docstring,too-few-public-methods,too-many-arguments,no-init,duplicate-code,invalid-name,suppressed-message,locally-disabled,protected-access,no-name-in-module,unused-argument,fixme,wrong-import-order,no-value-for-parameter,cyclic-import,unused-variable,redefined-outer-name,unused-import,too-many-instance-attributes,broad-except,too-many-locals,too-many-lines,redefined-variable-type,next-method-called,import-error,useless-else-on-loop,too-many-return-statements,too-many-nested-blocks,super-init-not-called,simplifiable-if-statement,no-self-use,no-member,logging-format-interpolation,dangerous-default-value
disable=missing-docstring,too-few-public-methods,too-many-arguments,no-init,duplicate-code,invalid-name,suppressed-message,locally-disabled,protected-access,no-name-in-module,unused-argument,fixme,wrong-import-order,no-value-for-parameter,cyclic-import,unused-variable,redefined-outer-name,unused-import,too-many-instance-attributes,broad-except,too-many-locals,too-many-lines,redefined-variable-type,next-method-called,import-error,useless-else-on-loop,too-many-return-statements,too-many-nested-blocks,super-init-not-called,no-self-use,no-member

@ -1076,6 +1076,7 @@ grpc_cc_library(
],
hdrs = [
"third_party/objective_c/Cronet/bidirectional_stream_c.h",
"src/core/ext/transport/cronet/transport/cronet_transport.h",
],
language = "c",
public_hdrs = [

@ -570,6 +570,9 @@ add_dependencies(buildtests_cxx alarm_cpp_test)
add_dependencies(buildtests_cxx async_end2end_test)
add_dependencies(buildtests_cxx auth_property_iterator_test)
if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
add_dependencies(buildtests_cxx bm_closure)
endif()
if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
add_dependencies(buildtests_cxx bm_fullstack)
endif()
add_dependencies(buildtests_cxx channel_arguments_test)
@ -1428,6 +1431,7 @@ add_library(grpc_test_util
test/core/util/port_uv.c
test/core/util/port_windows.c
test/core/util/slice_splitter.c
test/core/util/trickle_endpoint.c
src/core/lib/channel/channel_args.c
src/core/lib/channel/channel_stack.c
src/core/lib/channel/channel_stack_builder.c
@ -1635,6 +1639,7 @@ add_library(grpc_test_util_unsecure
test/core/util/port_uv.c
test/core/util/port_windows.c
test/core/util/slice_splitter.c
test/core/util/trickle_endpoint.c
)
if(WIN32 AND MSVC)
@ -7445,6 +7450,42 @@ endif (gRPC_BUILD_TESTS)
if (gRPC_BUILD_TESTS)
if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
add_executable(bm_closure
test/cpp/microbenchmarks/bm_closure.cc
third_party/googletest/src/gtest-all.cc
)
target_include_directories(bm_closure
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include
PRIVATE ${BORINGSSL_ROOT_DIR}/include
PRIVATE ${PROTOBUF_ROOT_DIR}/src
PRIVATE ${BENCHMARK_ROOT_DIR}/include
PRIVATE ${ZLIB_ROOT_DIR}
PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/third_party/zlib
PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/third_party/gflags/include
PRIVATE third_party/googletest/include
PRIVATE third_party/googletest
PRIVATE ${_gRPC_PROTO_GENS_DIR}
)
target_link_libraries(bm_closure
${_gRPC_PROTOBUF_LIBRARIES}
${_gRPC_ALLTARGETS_LIBRARIES}
benchmark
grpc_test_util
grpc
gpr_test_util
gpr
${_gRPC_GFLAGS_LIBRARIES}
)
endif()
endif (gRPC_BUILD_TESTS)
if (gRPC_BUILD_TESTS)
if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
add_executable(bm_fullstack
test/cpp/microbenchmarks/bm_fullstack.cc
third_party/googletest/src/gtest-all.cc

@ -1043,6 +1043,7 @@ wakeup_fd_cv_test: $(BINDIR)/$(CONFIG)/wakeup_fd_cv_test
alarm_cpp_test: $(BINDIR)/$(CONFIG)/alarm_cpp_test
async_end2end_test: $(BINDIR)/$(CONFIG)/async_end2end_test
auth_property_iterator_test: $(BINDIR)/$(CONFIG)/auth_property_iterator_test
bm_closure: $(BINDIR)/$(CONFIG)/bm_closure
bm_fullstack: $(BINDIR)/$(CONFIG)/bm_fullstack
channel_arguments_test: $(BINDIR)/$(CONFIG)/channel_arguments_test
channel_filter_test: $(BINDIR)/$(CONFIG)/channel_filter_test
@ -1449,6 +1450,7 @@ buildtests_cxx: privatelibs_cxx \
$(BINDIR)/$(CONFIG)/alarm_cpp_test \
$(BINDIR)/$(CONFIG)/async_end2end_test \
$(BINDIR)/$(CONFIG)/auth_property_iterator_test \
$(BINDIR)/$(CONFIG)/bm_closure \
$(BINDIR)/$(CONFIG)/bm_fullstack \
$(BINDIR)/$(CONFIG)/channel_arguments_test \
$(BINDIR)/$(CONFIG)/channel_filter_test \
@ -1553,6 +1555,7 @@ buildtests_cxx: privatelibs_cxx \
$(BINDIR)/$(CONFIG)/alarm_cpp_test \
$(BINDIR)/$(CONFIG)/async_end2end_test \
$(BINDIR)/$(CONFIG)/auth_property_iterator_test \
$(BINDIR)/$(CONFIG)/bm_closure \
$(BINDIR)/$(CONFIG)/bm_fullstack \
$(BINDIR)/$(CONFIG)/channel_arguments_test \
$(BINDIR)/$(CONFIG)/channel_filter_test \
@ -1870,6 +1873,8 @@ test_cxx: buildtests_cxx
$(Q) $(BINDIR)/$(CONFIG)/async_end2end_test || ( echo test async_end2end_test failed ; exit 1 )
$(E) "[RUN] Testing auth_property_iterator_test"
$(Q) $(BINDIR)/$(CONFIG)/auth_property_iterator_test || ( echo test auth_property_iterator_test failed ; exit 1 )
$(E) "[RUN] Testing bm_closure"
$(Q) $(BINDIR)/$(CONFIG)/bm_closure || ( echo test bm_closure failed ; exit 1 )
$(E) "[RUN] Testing bm_fullstack"
$(Q) $(BINDIR)/$(CONFIG)/bm_fullstack || ( echo test bm_fullstack failed ; exit 1 )
$(E) "[RUN] Testing channel_arguments_test"
@ -3262,6 +3267,7 @@ LIBGRPC_TEST_UTIL_SRC = \
test/core/util/port_uv.c \
test/core/util/port_windows.c \
test/core/util/slice_splitter.c \
test/core/util/trickle_endpoint.c \
src/core/lib/channel/channel_args.c \
src/core/lib/channel/channel_stack.c \
src/core/lib/channel/channel_stack_builder.c \
@ -3462,6 +3468,7 @@ LIBGRPC_TEST_UTIL_UNSECURE_SRC = \
test/core/util/port_uv.c \
test/core/util/port_windows.c \
test/core/util/slice_splitter.c \
test/core/util/trickle_endpoint.c \
PUBLIC_HEADERS_C += \
@ -12389,6 +12396,49 @@ endif
endif
BM_CLOSURE_SRC = \
test/cpp/microbenchmarks/bm_closure.cc \
BM_CLOSURE_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(BM_CLOSURE_SRC))))
ifeq ($(NO_SECURE),true)
# You can't build secure targets if you don't have OpenSSL.
$(BINDIR)/$(CONFIG)/bm_closure: openssl_dep_error
else
ifeq ($(NO_PROTOBUF),true)
# You can't build the protoc plugins or protobuf-enabled targets if you don't have protobuf 3.0.0+.
$(BINDIR)/$(CONFIG)/bm_closure: protobuf_dep_error
else
$(BINDIR)/$(CONFIG)/bm_closure: $(PROTOBUF_DEP) $(BM_CLOSURE_OBJS) $(LIBDIR)/$(CONFIG)/libbenchmark.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
$(E) "[LD] Linking $@"
$(Q) mkdir -p `dirname $@`
$(Q) $(LDXX) $(LDFLAGS) $(BM_CLOSURE_OBJS) $(LIBDIR)/$(CONFIG)/libbenchmark.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBSXX) $(LDLIBS_PROTOBUF) $(LDLIBS) $(LDLIBS_SECURE) $(GTEST_LIB) -o $(BINDIR)/$(CONFIG)/bm_closure
endif
endif
$(OBJDIR)/$(CONFIG)/test/cpp/microbenchmarks/bm_closure.o: $(LIBDIR)/$(CONFIG)/libbenchmark.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
deps_bm_closure: $(BM_CLOSURE_OBJS:.o=.dep)
ifneq ($(NO_SECURE),true)
ifneq ($(NO_DEPS),true)
-include $(BM_CLOSURE_OBJS:.o=.dep)
endif
endif
BM_FULLSTACK_SRC = \
test/cpp/microbenchmarks/bm_fullstack.cc \

@ -596,6 +596,7 @@ filegroups:
- test/core/util/port.h
- test/core/util/port_server_client.h
- test/core/util/slice_splitter.h
- test/core/util/trickle_endpoint.h
src:
- test/core/end2end/cq_verifier.c
- test/core/end2end/fake_resolver.c
@ -613,6 +614,7 @@ filegroups:
- test/core/util/port_uv.c
- test/core/util/port_windows.c
- test/core/util/slice_splitter.c
- test/core/util/trickle_endpoint.c
deps:
- grpc
- gpr_test_util
@ -724,6 +726,7 @@ filegroups:
- include/grpc/grpc_security.h
- include/grpc/grpc_security_constants.h
headers:
- src/core/ext/transport/cronet/transport/cronet_transport.h
- third_party/objective_c/Cronet/bidirectional_stream_c.h
src:
- src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
@ -2984,6 +2987,23 @@ targets:
- grpc
- gpr_test_util
- gpr
- name: bm_closure
build: test
language: c++
src:
- test/cpp/microbenchmarks/bm_closure.cc
deps:
- benchmark
- grpc_test_util
- grpc
- gpr_test_util
- gpr
args:
- --benchmark_min_time=0
platforms:
- mac
- linux
- posix
- name: bm_fullstack
build: test
language: c++

@ -884,7 +884,8 @@ Pod::Spec.new do |s|
s.subspec 'Cronet-Interface' do |ss|
ss.header_mappings_dir = 'include/grpc'
ss.source_files = 'include/grpc/grpc_cronet.h'
ss.source_files = 'include/grpc/grpc_cronet.h',
'src/core/ext/transport/cronet/transport/cronet_transport.h'
end
s.subspec 'Cronet-Implementation' do |ss|

@ -230,6 +230,10 @@ typedef struct {
#define GRPC_ARG_LB_POLICY_NAME "grpc.lb_policy_name"
/** The grpc_socket_mutator instance that set the socket options. A pointer. */
#define GRPC_ARG_SOCKET_MUTATOR "grpc.socket_mutator"
/** If non-zero, Cronet transport will coalesce packets to fewer frames when
* possible. */
#define GRPC_ARG_USE_CRONET_PACKET_COALESCING \
"grpc.use_cronet_packet_coalescing"
/** \} */
/** Result of a grpc call. If the caller satisfies the prerequisites of a

@ -265,7 +265,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
.gain_d = 0,
.initial_control_value = log2(DEFAULT_WINDOW),
.min_control_value = -1,
.max_control_value = 22,
.max_control_value = 25,
.integral_range = 10});
grpc_chttp2_goaway_parser_init(&t->goaway_parser);
@ -569,6 +569,14 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp,
GRPC_ERROR_UNREF(s->read_closed_error);
GRPC_ERROR_UNREF(s->write_closed_error);
if (s->incoming_window_delta > 0) {
GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA(
"destroy", t, s, s->incoming_window_delta);
} else if (s->incoming_window_delta < 0) {
GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA(
"destroy", t, s, -s->incoming_window_delta);
}
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "stream");
GPR_TIMER_END("destroy_stream", 0);
@ -1801,13 +1809,13 @@ static void update_bdp(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
if (delta == 0 || (bdp != 0 && delta > -1024 && delta < 1024)) {
return;
}
if (grpc_bdp_estimator_trace) {
gpr_log(GPR_DEBUG, "%s: update initial window size to %d", t->peer_string,
(int)bdp);
}
push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, bdp);
}
/*******************************************************************************
* INPUT PROCESSING - PARSING
*/
static grpc_error *try_http_parsing(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t) {
grpc_http_parser parser;
@ -2054,7 +2062,7 @@ static void incoming_byte_stream_update_flow_control(grpc_exec_ctx *exec_ctx,
(int64_t)have_already) {
write_type = GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED;
}
GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, incoming_window_delta,
GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA("op", t, s,
add_max_recv_bytes);
GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, announce_window,
add_max_recv_bytes);

@ -87,7 +87,7 @@ extern const grpc_chttp2_setting_parameters
grpc_chttp2_settings_parameters[GRPC_CHTTP2_NUM_SETTINGS];
/* Create a settings frame by diffing old & new, and updating old to be new */
grpc_slice grpc_chttp2_settings_create(uint32_t *old, const uint32_t *new,
grpc_slice grpc_chttp2_settings_create(uint32_t *old, const uint32_t *newval,
uint32_t force_mask, size_t count);
/* Create an ack settings frame */
grpc_slice grpc_chttp2_settings_ack_create(void);

@ -271,10 +271,6 @@ struct grpc_chttp2_transport {
/** data to write next write */
grpc_slice_buffer qbuf;
/** window available to announce to peer */
int64_t announce_incoming_window;
/** how much window would we like to have for incoming_window */
uint32_t connection_window_target;
/** how much data are we willing to buffer when the WRITE_BUFFER_HINT is set?
*/
uint32_t write_buffer_size;
@ -328,6 +324,16 @@ struct grpc_chttp2_transport {
/** window available for peer to send to us */
int64_t incoming_window;
/** calculating what we should give for incoming window:
we track the total amount of flow control over initial window size
across all streams: this is data that we want to receive right now (it
has an outstanding read)
and the total amount of flow control under initial window size across all
streams: this is data we've read early
we want to adjust incoming_window such that:
incoming_window = total_over - max(bdp - total_under, 0) */
int64_t stream_total_over_incoming_window;
int64_t stream_total_under_incoming_window;
/* deframing */
grpc_chttp2_deframe_transport_state deframe_state;
@ -634,6 +640,44 @@ typedef enum {
GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, dst_context, 0, dst_context, dst_var, \
amount)
#define GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE( \
phase, transport, dst_context) \
if (dst_context->incoming_window_delta < 0) { \
transport->stream_total_under_incoming_window += \
dst_context->incoming_window_delta; \
} else if (dst_context->incoming_window_delta > 0) { \
transport->stream_total_over_incoming_window -= \
dst_context->incoming_window_delta; \
}
#define GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE( \
phase, transport, dst_context) \
if (dst_context->incoming_window_delta < 0) { \
transport->stream_total_under_incoming_window -= \
dst_context->incoming_window_delta; \
} else if (dst_context->incoming_window_delta > 0) { \
transport->stream_total_over_incoming_window += \
dst_context->incoming_window_delta; \
}
#define GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA( \
phase, transport, dst_context, amount) \
GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE(phase, transport, \
dst_context); \
GRPC_CHTTP2_FLOW_DEBIT_STREAM(phase, transport, dst_context, \
incoming_window_delta, amount); \
GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE(phase, transport, \
dst_context);
#define GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA( \
phase, transport, dst_context, amount) \
GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE(phase, transport, \
dst_context); \
GRPC_CHTTP2_FLOW_CREDIT_STREAM(phase, transport, dst_context, \
incoming_window_delta, amount); \
GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE(phase, transport, \
dst_context);
#define GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, transport, id, dst_context, \
dst_var, amount) \
do { \
@ -752,4 +796,6 @@ void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t,
grpc_chttp2_stream *s, grpc_error *error);
uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */

@ -376,15 +376,6 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
return err;
}
uint32_t target_incoming_window = GPR_MAX(
t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
1024);
GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", t, incoming_window,
incoming_frame_size);
if (t->incoming_window <= target_incoming_window / 2) {
grpc_chttp2_initiate_write(exec_ctx, t, false, "flow_control");
}
if (s != NULL) {
if (incoming_frame_size >
s->incoming_window_delta +
@ -402,7 +393,7 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
return err;
}
GRPC_CHTTP2_FLOW_DEBIT_STREAM("parse", t, s, incoming_window_delta,
GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA("parse", t, s,
incoming_frame_size);
if ((int64_t)t->settings[GRPC_SENT_SETTINGS]
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] +
@ -417,6 +408,13 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
s->received_bytes += incoming_frame_size;
}
uint32_t target_incoming_window = grpc_chttp2_target_incoming_window(t);
GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", t, incoming_window,
incoming_frame_size);
if (t->incoming_window <= target_incoming_window / 2) {
grpc_chttp2_initiate_write(exec_ctx, t, false, "flow_control");
}
return GRPC_ERROR_NONE;
}

@ -152,6 +152,17 @@ static bool stream_ref_if_not_destroyed(gpr_refcount *r) {
return true;
}
uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t) {
return (uint32_t)GPR_MAX(
(int64_t)((1u << 31) - 1),
t->stream_total_over_incoming_window +
(int64_t)GPR_MAX(
t->settings[GRPC_SENT_SETTINGS]
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] -
t->stream_total_under_incoming_window,
0));
}
bool grpc_chttp2_begin_write(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t) {
grpc_chttp2_stream *s;
@ -310,13 +321,12 @@ bool grpc_chttp2_begin_write(grpc_exec_ctx *exec_ctx,
/* if the grpc_chttp2_transport is ready to send a window update, do so here
also; 3/4 is a magic number that will likely get tuned soon */
uint32_t target_incoming_window = GPR_MAX(
t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
1024);
uint32_t target_incoming_window = grpc_chttp2_target_incoming_window(t);
uint32_t threshold_to_send_transport_window_update =
t->outbuf.count > 0 ? 3 * target_incoming_window / 4
: target_incoming_window / 2;
if (t->incoming_window <= threshold_to_send_transport_window_update) {
if (t->incoming_window <= threshold_to_send_transport_window_update &&
t->incoming_window != target_incoming_window) {
maybe_initiate_ping(exec_ctx, t,
GRPC_CHTTP2_PING_BEFORE_TRANSPORT_WINDOW_UPDATE);
uint32_t announced = (uint32_t)GPR_CLAMP(

@ -39,6 +39,7 @@
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include "src/core/ext/transport/cronet/transport/cronet_transport.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/transport/transport_impl.h"
@ -54,16 +55,14 @@ extern grpc_transport_vtable grpc_cronet_vtable;
GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
void *engine, const char *target, const grpc_channel_args *args,
void *reserved) {
cronet_transport *ct = gpr_malloc(sizeof(cronet_transport));
ct->base.vtable = &grpc_cronet_vtable;
ct->engine = engine;
ct->host = gpr_malloc(strlen(target) + 1);
strcpy(ct->host, target);
gpr_log(GPR_DEBUG,
"grpc_create_cronet_transport: stream_engine = %p, target=%s", engine,
ct->host);
target);
grpc_transport *ct =
grpc_create_cronet_transport(engine, target, args, reserved);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
return grpc_channel_create(&exec_ctx, target, args,
GRPC_CLIENT_DIRECT_CHANNEL, (grpc_transport *)ct);
GRPC_CLIENT_DIRECT_CHANNEL, ct);
}

@ -80,4 +80,16 @@ void bidirectional_stream_cancel(bidirectional_stream* stream) {
GPR_ASSERT(0);
}
void bidirectional_stream_disable_auto_flush(bidirectional_stream* stream,
bool disable_auto_flush) {
GPR_ASSERT(0);
}
void bidirectional_stream_delay_request_headers_until_flush(
bidirectional_stream* stream, bool delay_headers_until_flush) {
GPR_ASSERT(0);
}
void bidirectional_stream_flush(bidirectional_stream* stream) { GPR_ASSERT(0); }
#endif /* GRPC_COMPILE_WITH_CRONET */

@ -88,7 +88,7 @@ enum e_op_id {
/* Cronet callbacks. See cronet_c_for_grpc.h for documentation for each. */
static void on_request_headers_sent(bidirectional_stream *);
static void on_stream_ready(bidirectional_stream *);
static void on_response_headers_received(
bidirectional_stream *, const bidirectional_stream_header_array *,
const char *);
@ -100,7 +100,7 @@ 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_request_headers_sent,
on_stream_ready,
on_response_headers_received,
on_read_completed,
on_write_completed,
@ -114,6 +114,7 @@ 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;
@ -152,6 +153,9 @@ struct op_state {
bool state_callback_received[OP_NUM_OPS];
bool fail_state;
bool flush_read;
bool flush_cronet_when_ready;
bool pending_write_for_trailer;
bool unprocessed_send_message;
grpc_error *cancel_error;
/* data structure for storing data coming from server */
struct read_state rs;
@ -175,7 +179,7 @@ struct op_storage {
struct stream_obj {
struct op_and_state *oas;
grpc_transport_stream_op *curr_op;
grpc_cronet_transport curr_ct;
grpc_cronet_transport *curr_ct;
grpc_stream *curr_gs;
bidirectional_stream *cbs;
bidirectional_stream_header_array header_array;
@ -274,6 +278,9 @@ static void add_to_storage(struct stream_obj *s, grpc_transport_stream_op *op) {
new_op->next = storage->head;
storage->head = new_op;
storage->num_pending_ops++;
if (op->send_message) {
s->state.unprocessed_send_message = true;
}
CRONET_LOG(GPR_DEBUG, "adding new op %p. %d in the queue.", new_op,
storage->num_pending_ops);
gpr_mu_unlock(&s->mu);
@ -406,9 +413,10 @@ static void on_succeeded(bidirectional_stream *stream) {
/*
Cronet callback
*/
static void on_request_headers_sent(bidirectional_stream *stream) {
CRONET_LOG(GPR_DEBUG, "W: on_request_headers_sent(%p)", stream);
static void on_stream_ready(bidirectional_stream *stream) {
CRONET_LOG(GPR_DEBUG, "W: on_stream_ready(%p)", stream);
stream_obj *s = (stream_obj *)stream->annotation;
grpc_cronet_transport *t = (grpc_cronet_transport *)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;
@ -417,6 +425,14 @@ static void on_request_headers_sent(bidirectional_stream *stream) {
gpr_free(s->header_array.headers);
s->header_array.headers = NULL;
}
/* 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);
}
@ -528,6 +544,7 @@ static void on_response_trailers_received(
CRONET_LOG(GPR_DEBUG, "R: on_response_trailers_received(%p,%p)", stream,
trailers);
stream_obj *s = (stream_obj *)stream->annotation;
grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
gpr_mu_lock(&s->mu);
memset(&s->state.rs.trailing_metadata, 0,
sizeof(s->state.rs.trailing_metadata));
@ -558,6 +575,10 @@ static void on_response_trailers_received(
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);
@ -607,7 +628,7 @@ static void convert_metadata_to_cronet_headers(
curr = curr->next;
num_headers_available++;
}
/* Allocate enough memory. It is freed in the on_request_headers_sent callback
/* Allocate enough memory. It is freed in the on_stream_ready callback
*/
bidirectional_stream_header *headers =
(bidirectional_stream_header *)gpr_malloc(
@ -687,8 +708,10 @@ static bool header_has_authority(grpc_linked_mdelem *head) {
executed. This is the heart of the state machine.
*/
static bool op_can_be_run(grpc_transport_stream_op *curr_op,
struct op_state *stream_state,
struct op_state *op_state, enum e_op_id op_id) {
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
@ -755,12 +778,14 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op,
else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
result = false;
/* we haven't sent message yet */
else if (curr_op->send_message &&
else if (stream_state->unprocessed_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])
!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 */
@ -833,24 +858,28 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
struct op_and_state *oas) {
grpc_transport_stream_op *stream_op = &oas->op;
struct stream_obj *s = oas->s;
grpc_cronet_transport *t = (grpc_cronet_transport *)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, stream_state, &oas->state,
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 == NULL);
GPR_ASSERT(!stream_state->state_op_done[OP_SEND_INITIAL_METADATA]);
s->cbs = bidirectional_stream_create(s->curr_ct.engine, s->curr_gs,
&cronet_callbacks);
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 = NULL;
const char *method = "POST";
s->header_array.headers = NULL;
convert_metadata_to_cronet_headers(
stream_op->send_initial_metadata->list.head, s->curr_ct.host, &url,
stream_op->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);
@ -862,30 +891,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
gpr_free((void *)s->header_array.headers[header_index].value);
}
stream_state->state_op_done[OP_SEND_INITIAL_METADATA] = true;
result = ACTION_TAKEN_WITH_CALLBACK;
} else if (stream_op->recv_initial_metadata &&
op_can_be_run(stream_op, stream_state, &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(exec_ctx, stream_op->recv_initial_metadata_ready,
GRPC_ERROR_NONE);
} else if (stream_state->state_callback_received[OP_FAILED]) {
grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
GRPC_ERROR_NONE);
} else {
grpc_chttp2_incoming_metadata_buffer_publish(
exec_ctx, &oas->s->state.rs.initial_metadata,
stream_op->recv_initial_metadata);
grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
GRPC_ERROR_NONE);
if (t->use_packet_coalescing) {
if (!stream_op->send_message && !stream_op->send_trailing_metadata) {
s->state.flush_cronet_when_ready = true;
}
stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true;
result = ACTION_TAKEN_NO_CALLBACK;
}
result = ACTION_TAKEN_WITH_CALLBACK;
} else if (stream_op->send_message &&
op_can_be_run(stream_op, stream_state, &oas->state,
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->unprocessed_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.");
@ -916,16 +931,63 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
bidirectional_stream_write(s->cbs, stream_state->ws.write_buffer,
(int)write_buffer_size, false);
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;
} 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(exec_ctx, stream_op->recv_initial_metadata_ready,
GRPC_ERROR_NONE);
} else if (stream_state->state_callback_received[OP_FAILED]) {
grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
GRPC_ERROR_NONE);
} else {
grpc_chttp2_incoming_metadata_buffer_publish(
exec_ctx, &oas->s->state.rs.initial_metadata,
stream_op->recv_initial_metadata);
grpc_closure_sched(exec_ctx, stream_op->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, stream_state, &oas->state,
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.");
@ -980,6 +1042,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
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;
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);
result = ACTION_TAKEN_NO_CALLBACK;
}
} else if (stream_state->rs.remaining_bytes == 0) {
@ -1027,7 +1099,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
result = ACTION_TAKEN_NO_CALLBACK;
}
} else if (stream_op->recv_trailing_metadata &&
op_can_be_run(stream_op, stream_state, &oas->state,
op_can_be_run(stream_op, s, &oas->state,
OP_RECV_TRAILING_METADATA)) {
CRONET_LOG(GPR_DEBUG, "running: %p OP_RECV_TRAILING_METADATA", oas);
if (oas->s->state.rs.trailing_metadata_valid) {
@ -1038,23 +1110,8 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
}
stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true;
result = ACTION_TAKEN_NO_CALLBACK;
} else if (stream_op->send_trailing_metadata &&
op_can_be_run(stream_op, stream_state, &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);
result = ACTION_TAKEN_WITH_CALLBACK;
}
stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true;
} else if (stream_op->cancel_error &&
op_can_be_run(stream_op, stream_state, &oas->state,
OP_CANCEL_ERROR)) {
op_can_be_run(stream_op, s, &oas->state, OP_CANCEL_ERROR)) {
CRONET_LOG(GPR_DEBUG, "running: %p OP_CANCEL_ERROR", oas);
CRONET_LOG(GPR_DEBUG, "W: bidirectional_stream_cancel(%p)", s->cbs);
if (s->cbs) {
@ -1068,8 +1125,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
stream_state->cancel_error = GRPC_ERROR_REF(stream_op->cancel_error);
}
} else if (stream_op->on_complete &&
op_can_be_run(stream_op, stream_state, &oas->state,
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(exec_ctx, stream_op->on_complete,
@ -1133,6 +1189,12 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
sizeof(s->state.state_callback_received));
s->state.fail_state = s->state.flush_read = false;
s->state.cancel_error = NULL;
s->state.flush_cronet_when_ready = s->state.pending_write_for_trailer = false;
s->state.unprocessed_send_message = false;
s->curr_gs = gs;
s->curr_ct = (grpc_cronet_transport *)gt;
gpr_mu_init(&s->mu);
return 0;
}
@ -1148,8 +1210,6 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_stream *gs, grpc_transport_stream_op *op) {
CRONET_LOG(GPR_DEBUG, "perform_stream_op");
stream_obj *s = (stream_obj *)gs;
s->curr_gs = gs;
memcpy(&s->curr_ct, gt, sizeof(grpc_cronet_transport));
add_to_storage(s, op);
if (op->send_initial_metadata &&
header_has_authority(op->send_initial_metadata->list.head)) {
@ -1197,7 +1257,8 @@ static grpc_endpoint *get_endpoint(grpc_exec_ctx *exec_ctx,
static void perform_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_transport_op *op) {}
const grpc_transport_vtable grpc_cronet_vtable = {sizeof(stream_obj),
static const grpc_transport_vtable grpc_cronet_vtable = {
sizeof(stream_obj),
"cronet_http",
init_stream,
set_pollset_do_nothing,
@ -1208,3 +1269,46 @@ const grpc_transport_vtable grpc_cronet_vtable = {sizeof(stream_obj),
destroy_transport,
get_peer,
get_endpoint};
grpc_transport *grpc_create_cronet_transport(void *engine, const char *target,
const grpc_channel_args *args,
void *reserved) {
grpc_cronet_transport *ct = gpr_malloc(sizeof(grpc_cronet_transport));
if (!ct) {
goto error;
}
ct->base.vtable = &grpc_cronet_vtable;
ct->engine = engine;
ct->host = 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 (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 NULL;
}

@ -0,0 +1,43 @@
/*
*
* Copyright 2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#ifndef GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H
#define GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H
#include "src/core/lib/transport/transport.h"
grpc_transport *grpc_create_cronet_transport(void *engine, const char *target,
const grpc_channel_args *args,
void *reserved);
#endif /* GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H */

@ -198,14 +198,17 @@ static grpc_error *server_filter_incoming_metadata(grpc_exec_ctx *exec_ctx,
GRPC_ERROR_STR_KEY, ":path"));
}
if (b->idx.named.host != NULL) {
if (b->idx.named.host != NULL && b->idx.named.authority == NULL) {
grpc_linked_mdelem *el = b->idx.named.host;
grpc_mdelem md = GRPC_MDELEM_REF(el->md);
grpc_metadata_batch_remove(exec_ctx, b, el);
add_error(
error_name, &error,
grpc_metadata_batch_substitute(
exec_ctx, b, b->idx.named.host,
grpc_mdelem_from_slices(
grpc_metadata_batch_add_head(
exec_ctx, b, el, grpc_mdelem_from_slices(
exec_ctx, GRPC_MDSTR_AUTHORITY,
grpc_slice_ref_internal(GRPC_MDVALUE(b->idx.named.host->md)))));
grpc_slice_ref_internal(GRPC_MDVALUE(md)))));
GRPC_MDELEM_UNREF(exec_ctx, md);
}
if (b->idx.named.authority == NULL) {

@ -1531,16 +1531,6 @@ static void pollset_destroy(grpc_pollset *pollset) {
gpr_mu_destroy(&pollset->po.mu);
}
static void pollset_reset(grpc_pollset *pollset) {
GPR_ASSERT(pollset->shutting_down);
GPR_ASSERT(!pollset_has_workers(pollset));
pollset->shutting_down = false;
pollset->finish_shutdown_called = false;
pollset->kicked_without_pollers = false;
pollset->shutdown_done = NULL;
GPR_ASSERT(pollset->po.pi == NULL);
}
static bool maybe_do_workqueue_work(grpc_exec_ctx *exec_ctx,
polling_island *pi) {
if (gpr_mu_trylock(&pi->workqueue_read_mu)) {
@ -2084,7 +2074,6 @@ static const grpc_event_engine_vtable vtable = {
.pollset_init = pollset_init,
.pollset_shutdown = pollset_shutdown,
.pollset_reset = pollset_reset,
.pollset_destroy = pollset_destroy,
.pollset_work = pollset_work,
.pollset_kick = pollset_kick,

@ -815,16 +815,6 @@ static void pollset_destroy(grpc_pollset *pollset) {
gpr_mu_destroy(&pollset->mu);
}
static void pollset_reset(grpc_pollset *pollset) {
GPR_ASSERT(pollset->shutting_down);
GPR_ASSERT(!pollset_has_workers(pollset));
GPR_ASSERT(pollset->idle_jobs.head == pollset->idle_jobs.tail);
GPR_ASSERT(pollset->fd_count == 0);
pollset->shutting_down = 0;
pollset->called_shutdown = 0;
pollset->kicked_without_pollers = 0;
}
static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_fd *fd) {
gpr_mu_lock(&pollset->mu);
@ -1514,7 +1504,6 @@ static const grpc_event_engine_vtable vtable = {
.pollset_init = pollset_init,
.pollset_shutdown = pollset_shutdown,
.pollset_reset = pollset_reset,
.pollset_destroy = pollset_destroy,
.pollset_work = pollset_work,
.pollset_kick = pollset_kick,

@ -191,10 +191,6 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
g_event_engine->pollset_shutdown(exec_ctx, pollset, closure);
}
void grpc_pollset_reset(grpc_pollset *pollset) {
g_event_engine->pollset_reset(pollset);
}
void grpc_pollset_destroy(grpc_pollset *pollset) {
g_event_engine->pollset_destroy(pollset);
}

@ -64,7 +64,6 @@ typedef struct grpc_event_engine_vtable {
void (*pollset_init)(grpc_pollset *pollset, gpr_mu **mu);
void (*pollset_shutdown)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_closure *closure);
void (*pollset_reset)(grpc_pollset *pollset);
void (*pollset_destroy)(grpc_pollset *pollset);
grpc_error *(*pollset_work)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_pollset_worker **worker, gpr_timespec now,

@ -58,9 +58,6 @@ void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu);
* pollset's mutex must be held */
void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_closure *closure);
/** Reset the pollset to its initial state (perhaps with some cached objects);
* must have been previously shutdown */
void grpc_pollset_reset(grpc_pollset *pollset);
void grpc_pollset_destroy(grpc_pollset *pollset);
/* Do some work on a pollset.

@ -97,11 +97,6 @@ void grpc_pollset_destroy(grpc_pollset *pollset) {
}
}
void grpc_pollset_reset(grpc_pollset *pollset) {
GPR_ASSERT(pollset->shutting_down);
pollset->shutting_down = 0;
}
static void timer_run_cb(uv_timer_t *timer) {}
grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,

@ -117,16 +117,6 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
void grpc_pollset_destroy(grpc_pollset *pollset) {}
void grpc_pollset_reset(grpc_pollset *pollset) {
GPR_ASSERT(pollset->shutting_down);
GPR_ASSERT(
!has_workers(&pollset->root_worker, GRPC_POLLSET_WORKER_LINK_POLLSET));
pollset->shutting_down = 0;
pollset->is_iocp_worker = 0;
pollset->kicked_without_pollers = 0;
pollset->on_shutdown = NULL;
}
grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_pollset_worker **worker_hdl,
gpr_timespec now, gpr_timespec deadline) {

@ -128,7 +128,8 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
}
channel->default_authority = grpc_mdelem_from_slices(
exec_ctx, GRPC_MDSTR_AUTHORITY,
grpc_slice_from_copied_string(args->args[i].value.string));
grpc_slice_intern(
grpc_slice_from_static_string(args->args[i].value.string)));
}
} else if (0 ==
strcmp(args->args[i].key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)) {
@ -144,7 +145,8 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
} else {
channel->default_authority = grpc_mdelem_from_slices(
exec_ctx, GRPC_MDSTR_AUTHORITY,
grpc_slice_from_copied_string(args->args[i].value.string));
grpc_slice_intern(
grpc_slice_from_static_string(args->args[i].value.string)));
}
}
} else if (0 == strcmp(args->args[i].key,

@ -96,9 +96,6 @@ struct grpc_completion_queue {
#define POLLSET_FROM_CQ(cq) ((grpc_pollset *)(cq + 1))
#define CQ_FROM_POLLSET(ps) (((grpc_completion_queue *)ps) - 1)
static gpr_mu g_freelist_mu;
static grpc_completion_queue *g_freelist;
int grpc_cq_pluck_trace;
int grpc_cq_event_timeout_trace;
@ -113,21 +110,6 @@ int grpc_cq_event_timeout_trace;
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc,
grpc_error *error);
void grpc_cq_global_init(void) { gpr_mu_init(&g_freelist_mu); }
void grpc_cq_global_shutdown(void) {
gpr_mu_destroy(&g_freelist_mu);
while (g_freelist) {
grpc_completion_queue *next = g_freelist->next_free;
grpc_pollset_destroy(POLLSET_FROM_CQ(g_freelist));
#ifndef NDEBUG
gpr_free(g_freelist->outstanding_tags);
#endif
gpr_free(g_freelist);
g_freelist = next;
}
}
grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
grpc_completion_queue *cc;
GPR_ASSERT(!reserved);
@ -136,22 +118,12 @@ grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved));
gpr_mu_lock(&g_freelist_mu);
if (g_freelist == NULL) {
gpr_mu_unlock(&g_freelist_mu);
cc = gpr_malloc(sizeof(grpc_completion_queue) + grpc_pollset_size());
grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu);
#ifndef NDEBUG
cc->outstanding_tags = NULL;
cc->outstanding_tag_capacity = 0;
#endif
} else {
cc = g_freelist;
g_freelist = g_freelist->next_free;
gpr_mu_unlock(&g_freelist_mu);
/* pollset already initialized */
}
/* Initial ref is dropped by grpc_completion_queue_shutdown */
gpr_ref_init(&cc->pending_events, 1);
@ -203,11 +175,11 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) {
#endif
if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->completed_head.next == (uintptr_t)&cc->completed_head);
grpc_pollset_reset(POLLSET_FROM_CQ(cc));
gpr_mu_lock(&g_freelist_mu);
cc->next_free = g_freelist;
g_freelist = cc;
gpr_mu_unlock(&g_freelist_mu);
grpc_pollset_destroy(POLLSET_FROM_CQ(cc));
#ifndef NDEBUG
gpr_free(cc->outstanding_tags);
#endif
gpr_free(cc);
}
}

@ -99,7 +99,4 @@ bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc);
void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
int grpc_cq_is_server_cq(grpc_completion_queue *cc);
void grpc_cq_global_init(void);
void grpc_cq_global_shutdown(void);
#endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */

@ -209,7 +209,6 @@ void grpc_init(void) {
grpc_iomgr_init();
grpc_executor_init();
gpr_timers_global_init();
grpc_cq_global_init();
grpc_handshaker_factory_registry_init();
grpc_security_init();
for (i = 0; i < g_number_of_plugins; i++) {
@ -236,7 +235,6 @@ void grpc_shutdown(void) {
gpr_mu_lock(&g_init_mu);
if (--g_initializations == 0) {
grpc_executor_shutdown(&exec_ctx);
grpc_cq_global_shutdown();
grpc_iomgr_shutdown(&exec_ctx);
gpr_timers_global_destroy();
grpc_tracer_shutdown();

@ -32,13 +32,13 @@
*/
#import <XCTest/XCTest.h>
#import <sys/socket.h>
#import <netinet/in.h>
#import <sys/socket.h>
#import <Cronet/Cronet.h>
#import <grpc/support/host_port.h>
#import <grpc/grpc_cronet.h>
#import <grpc/grpc.h>
#import <grpc/grpc_cronet.h>
#import <grpc/support/host_port.h>
#import "test/core/end2end/cq_verifier.h"
#import "test/core/util/port.h"
@ -49,16 +49,19 @@
#import "src/core/lib/support/env.h"
#import "src/core/lib/support/string.h"
#import "src/core/lib/support/tmpfile.h"
#import "test/core/end2end/data/ssl_test_data.h"
#import "test/core/util/test_config.h"
#import <BoringSSL/openssl/ssl.h>
static void drain_cq(grpc_completion_queue *cq) {
grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5), NULL);
ev = grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5),
NULL);
} while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
@interface CronetUnitTests : XCTestCase
@end
@ -68,29 +71,76 @@ static void drain_cq(grpc_completion_queue *cq) {
+ (void)setUp {
[super setUp];
/*** FILE *roots_file;
size_t roots_size = strlen(test_root_cert);*/
char *argv[] = {"CoreCronetEnd2EndTests"};
grpc_test_init(1, argv);
grpc_init();
[Cronet setHttp2Enabled:YES];
[Cronet setSslKeyLogFileName:@"Documents/key"];
[Cronet enableTestCertVerifierForTesting];
NSURL *url = [[[NSFileManager defaultManager]
URLsForDirectory:NSDocumentDirectory
inDomains:NSUserDomainMask] lastObject];
NSLog(@"Documents directory: %@", url);
[Cronet start];
[Cronet startNetLogToFile:@"Documents/cronet_netlog.json" logBytes:YES];
init_ssl();
}
+ (void)tearDown {
grpc_shutdown();
cleanup_ssl();
[super tearDown];
}
void init_ssl(void) {
SSL_load_error_strings();
OpenSSL_add_ssl_algorithms();
}
void cleanup_ssl(void) { EVP_cleanup(); }
int alpn_cb(SSL *ssl, const unsigned char **out, unsigned char *outlen,
const unsigned char *in, unsigned int inlen, void *arg) {
// Always select "h2" as the ALPN protocol to be used
*out = (const unsigned char *)"h2";
*outlen = 2;
return SSL_TLSEXT_ERR_OK;
}
void init_ctx(SSL_CTX *ctx) {
// Install server certificate
BIO *pem = BIO_new_mem_buf((void *)test_server1_cert,
(int)strlen(test_server1_cert));
X509 *cert = PEM_read_bio_X509_AUX(pem, NULL, NULL, "");
SSL_CTX_use_certificate(ctx, cert);
X509_free(cert);
BIO_free(pem);
// Install server private key
pem =
BIO_new_mem_buf((void *)test_server1_key, (int)strlen(test_server1_key));
EVP_PKEY *key = PEM_read_bio_PrivateKey(pem, NULL, NULL, "");
SSL_CTX_use_PrivateKey(ctx, key);
EVP_PKEY_free(key);
BIO_free(pem);
// Select cipher suite
SSL_CTX_set_cipher_list(ctx, "ECDHE-RSA-AES128-GCM-SHA256");
// Select ALPN protocol
SSL_CTX_set_alpn_select_cb(ctx, alpn_cb, NULL);
}
unsigned int parse_h2_length(const char *field) {
return ((unsigned int)(unsigned char)(field[0])) * 65536 +
((unsigned int)(unsigned char)(field[1])) * 256 +
((unsigned int)(unsigned char)(field[2]));
}
- (void)testInternalError {
grpc_call *c;
grpc_slice request_payload_slice =
@ -98,17 +148,22 @@ static void drain_cq(grpc_completion_queue *cq) {
grpc_byte_buffer *request_payload =
grpc_raw_byte_buffer_create(&request_payload_slice, 1);
gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
grpc_metadata meta_c[2] = {
{"key1", "val1", 4, 0, {{NULL, NULL, NULL, NULL}}},
{"key2", "val2", 4, 0, {{NULL, NULL, NULL, NULL}}}};
grpc_metadata meta_c[2] = {{grpc_slice_from_static_string("key1"),
grpc_slice_from_static_string("val1"),
0,
{{NULL, NULL, NULL, NULL}}},
{grpc_slice_from_static_string("key2"),
grpc_slice_from_static_string("val2"),
0,
{{NULL, NULL, NULL, NULL}}}};
int port = grpc_pick_unused_port_or_die();
char *addr;
gpr_join_host_port(&addr, "127.0.0.1", port);
grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
cronet_engine *cronetEngine = [Cronet getGlobalEngine];
grpc_channel *client = grpc_cronet_secure_channel_create(cronetEngine, addr,
NULL, NULL);
stream_engine *cronetEngine = [Cronet getGlobalEngine];
grpc_channel *client =
grpc_cronet_secure_channel_create(cronetEngine, addr, NULL, NULL);
cq_verifier *cqv = cq_verifier_create(cq);
grpc_op ops[6];
@ -120,12 +175,11 @@ static void drain_cq(grpc_completion_queue *cq) {
grpc_call_details call_details;
grpc_status_code status;
grpc_call_error error;
char *details = NULL;
size_t details_capacity = 0;
grpc_slice details;
c = grpc_channel_create_call(
client, NULL, GRPC_PROPAGATE_DEFAULTS, cq, "/foo",
NULL, deadline, NULL);
c = grpc_channel_create_call(client, NULL, GRPC_PROPAGATE_DEFAULTS, cq,
grpc_slice_from_static_string("/foo"), NULL,
deadline, NULL);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
@ -164,24 +218,29 @@ static void drain_cq(grpc_completion_queue *cq) {
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->data.recv_status_on_client.status_details_capacity = &details_capacity;
op->flags = 0;
op->reserved = NULL;
op++;
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), (void *)1, NULL);
GPR_ASSERT(GRPC_CALL_OK == error);
dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
dispatch_async(
dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
int sl = socket(AF_INET, SOCK_STREAM, 0);
GPR_ASSERT(sl >= 0);
// Make and TCP endpoint to accept the connection
struct sockaddr_in s_addr;
memset(&s_addr, 0, sizeof(s_addr));
s_addr.sin_family = AF_INET;
s_addr.sin_addr.s_addr = htonl(INADDR_ANY);
s_addr.sin_port = htons(port);
bind(sl, (struct sockaddr*)&s_addr, sizeof(s_addr));
listen(sl, 5);
GPR_ASSERT(0 == bind(sl, (struct sockaddr *)&s_addr, sizeof(s_addr)));
GPR_ASSERT(0 == listen(sl, 5));
int s = accept(sl, NULL, NULL);
GPR_ASSERT(s >= 0);
// Close the connection after 1 second to trigger Cronet's on_failed()
sleep(1);
close(s);
close(sl);
@ -192,7 +251,7 @@ static void drain_cq(grpc_completion_queue *cq) {
GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
gpr_free(details);
grpc_slice_unref(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
@ -211,4 +270,188 @@ static void drain_cq(grpc_completion_queue *cq) {
grpc_completion_queue_destroy(cq);
}
- (void)packetCoalescing:(BOOL)useCoalescing {
grpc_arg arg;
arg.key = GRPC_ARG_USE_CRONET_PACKET_COALESCING;
arg.type = GRPC_ARG_INTEGER;
arg.value.integer = useCoalescing ? 1 : 0;
grpc_channel_args *args = grpc_channel_args_copy_and_add(NULL, &arg, 1);
grpc_call *c;
grpc_slice request_payload_slice =
grpc_slice_from_copied_string("hello world");
grpc_byte_buffer *request_payload =
grpc_raw_byte_buffer_create(&request_payload_slice, 1);
gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
grpc_metadata meta_c[2] = {{grpc_slice_from_static_string("key1"),
grpc_slice_from_static_string("val1"),
0,
{{NULL, NULL, NULL, NULL}}},
{grpc_slice_from_static_string("key2"),
grpc_slice_from_static_string("val2"),
0,
{{NULL, NULL, NULL, NULL}}}};
int port = grpc_pick_unused_port_or_die();
char *addr;
gpr_join_host_port(&addr, "127.0.0.1", port);
grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
stream_engine *cronetEngine = [Cronet getGlobalEngine];
grpc_channel *client =
grpc_cronet_secure_channel_create(cronetEngine, addr, args, NULL);
cq_verifier *cqv = cq_verifier_create(cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
grpc_byte_buffer *response_payload_recv = NULL;
grpc_call_details call_details;
grpc_status_code status;
grpc_call_error error;
grpc_slice details;
c = grpc_channel_create_call(client, NULL, GRPC_PROPAGATE_DEFAULTS, cq,
grpc_slice_from_static_string("/foo"), NULL,
deadline, NULL);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
memset(ops, 0, sizeof(ops));
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 2;
op->data.send_initial_metadata.metadata = meta_c;
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message.send_message = request_payload;
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message.recv_message = &response_payload_recv;
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->flags = 0;
op->reserved = NULL;
op++;
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), (void *)1, NULL);
GPR_ASSERT(GRPC_CALL_OK == error);
__weak XCTestExpectation *expectation = [self expectationWithDescription:@"Coalescing"];
dispatch_async(
dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
int sl = socket(AF_INET, SOCK_STREAM, 0);
GPR_ASSERT(sl >= 0);
struct sockaddr_in s_addr;
memset(&s_addr, 0, sizeof(s_addr));
s_addr.sin_family = AF_INET;
s_addr.sin_addr.s_addr = htonl(INADDR_ANY);
s_addr.sin_port = htons(port);
GPR_ASSERT(0 == bind(sl, (struct sockaddr *)&s_addr, sizeof(s_addr)));
GPR_ASSERT(0 == listen(sl, 5));
int s = accept(sl, NULL, NULL);
GPR_ASSERT(s >= 0);
struct timeval tv;
tv.tv_sec = 2;
tv.tv_usec = 0;
setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
// Make an TLS endpoint to receive Cronet's transmission
SSL_CTX *ctx = SSL_CTX_new(TLSv1_2_server_method());
init_ctx(ctx);
SSL *ssl = SSL_new(ctx);
SSL_set_fd(ssl, s);
SSL_accept(ssl);
const char magic[] = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
char buf[4096];
long len;
BOOL coalesced = NO;
while ((len = SSL_read(ssl, buf, sizeof(buf))) > 0) {
gpr_log(GPR_DEBUG, "Read len: %ld", len);
// Analyze the HTTP/2 frames in the same TLS PDU to identify if
// coalescing is successful
unsigned int p = 0;
while (p < len) {
if (len - p >= 24 && 0 == memcmp(&buf[p], magic, 24)) {
p += 24;
continue;
}
if (buf[p + 3] == 0 && // Type is DATA
parse_h2_length(&buf[p]) == 0x10 && // Length is correct
(buf[p + 4] & 1) != 0 && // EOS bit is set
0 == memcmp("hello world", &buf[p + 14],
11)) { // Message is correct
coalesced = YES;
break;
}
p += (parse_h2_length(&buf[p]) + 9);
}
if (coalesced) {
break;
}
}
XCTAssert(coalesced == useCoalescing);
SSL_free(ssl);
SSL_CTX_free(ctx);
close(s);
close(sl);
[expectation fulfill];
});
CQ_EXPECT_COMPLETION(cqv, (void *)1, 1);
cq_verify(cqv);
grpc_slice_unref(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
grpc_call_destroy(c);
cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload_recv);
grpc_channel_destroy(client);
grpc_completion_queue_shutdown(cq);
drain_cq(cq);
grpc_completion_queue_destroy(cq);
[self waitForExpectationsWithTimeout:4 handler:nil];
}
- (void)testPacketCoalescing {
[self packetCoalescing:YES];
[self packetCoalescing:NO];
}
@end

@ -1474,6 +1474,7 @@
"$(inherited)",
"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
"GRPC_COMPILE_WITH_CRONET=1",
"GRPC_CRONET_WITH_PACKET_COALESCING=1",
);
INFOPLIST_FILE = InteropTestsRemoteWithCronet/Info.plist;
IPHONEOS_DEPLOYMENT_TARGET = 9.3;

@ -59,6 +59,16 @@
ReferencedContainer = "container:Tests.xcodeproj">
</BuildableReference>
</TestableReference>
<TestableReference
skipped = "NO">
<BuildableReference
BuildableIdentifier = "primary"
BlueprintIdentifier = "5EAD6D231E27047400002378"
BuildableName = "CronetUnitTests.xctest"
BlueprintName = "CronetUnitTests"
ReferencedContainer = "container:Tests.xcodeproj">
</BuildableReference>
</TestableReference>
</Testables>
<MacroExpansion>
<BuildableReference
@ -100,6 +110,15 @@
savedToolIdentifier = ""
useCustomWorkingDirectory = "NO"
debugDocumentVersioning = "YES">
<MacroExpansion>
<BuildableReference
BuildableIdentifier = "primary"
BlueprintIdentifier = "63423F431B150A5F006CF63C"
BuildableName = "AllTests.xctest"
BlueprintName = "AllTests"
ReferencedContainer = "container:Tests.xcodeproj">
</BuildableReference>
</MacroExpansion>
</ProfileAction>
<AnalyzeAction
buildConfiguration = "Debug">

@ -48,11 +48,8 @@ class GoogleCallCredentials(grpc.AuthMetadataPlugin):
# Hack to determine if these are JWT creds and we need to pass
# additional_claims when getting a token
if 'additional_claims' in inspect.getargspec(
credentials.get_access_token).args:
self._is_jwt = True
else:
self._is_jwt = False
self._is_jwt = 'additional_claims' in inspect.getargspec(
credentials.get_access_token).args
def __call__(self, context, callback):
# MetadataPlugins cannot block (see grpc.beta.interfaces.py)

@ -444,10 +444,10 @@ def _start_unary_request(request, timeout, request_serializer):
return deadline, deadline_timespec, serialized_request, None
def _end_unary_response_blocking(state, with_call, deadline):
def _end_unary_response_blocking(state, call, with_call, deadline):
if state.code is grpc.StatusCode.OK:
if with_call:
rendezvous = _Rendezvous(state, None, None, deadline)
rendezvous = _Rendezvous(state, call, None, deadline)
return state.response, rendezvous
else:
return state.response
@ -499,17 +499,17 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable):
_check_call_error(call_error, metadata)
_handle_event(completion_queue.poll(), state,
self._response_deserializer)
return state, deadline
return state, call, deadline
def __call__(self, request, timeout=None, metadata=None, credentials=None):
state, deadline, = self._blocking(request, timeout, metadata,
state, call, deadline = self._blocking(request, timeout, metadata,
credentials)
return _end_unary_response_blocking(state, False, deadline)
return _end_unary_response_blocking(state, call, False, deadline)
def with_call(self, request, timeout=None, metadata=None, credentials=None):
state, deadline, = self._blocking(request, timeout, metadata,
state, call, deadline = self._blocking(request, timeout, metadata,
credentials)
return _end_unary_response_blocking(state, True, deadline)
return _end_unary_response_blocking(state, call, True, deadline)
def future(self, request, timeout=None, metadata=None, credentials=None):
state, operations, deadline, deadline_timespec, rendezvous = self._prepare(
@ -619,25 +619,25 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable):
state.condition.notify_all()
if not state.due:
break
return state, deadline
return state, call, deadline
def __call__(self,
request_iterator,
timeout=None,
metadata=None,
credentials=None):
state, deadline, = self._blocking(request_iterator, timeout, metadata,
credentials)
return _end_unary_response_blocking(state, False, deadline)
state, call, deadline = self._blocking(request_iterator, timeout,
metadata, credentials)
return _end_unary_response_blocking(state, call, False, deadline)
def with_call(self,
request_iterator,
timeout=None,
metadata=None,
credentials=None):
state, deadline, = self._blocking(request_iterator, timeout, metadata,
credentials)
return _end_unary_response_blocking(state, True, deadline)
state, call, deadline = self._blocking(request_iterator, timeout,
metadata, credentials)
return _end_unary_response_blocking(state, call, True, deadline)
def future(self,
request_iterator,

@ -92,7 +92,7 @@ def decode(b):
try:
return b.decode('utf8')
except UnicodeDecodeError:
logging.exception('Invalid encoding on {}'.format(b))
logging.exception('Invalid encoding on %s', b)
return b.decode('latin1')
@ -153,13 +153,7 @@ class CleanupThread(threading.Thread):
we accomplish this by overriding the join() method.
"""
def __init__(self,
behavior,
group=None,
target=None,
name=None,
args=(),
kwargs={}):
def __init__(self, behavior, *args, **kwargs):
"""Constructor.
Args:
@ -167,17 +161,10 @@ class CleanupThread(threading.Thread):
argument, timeout, indicating the maximum duration of
`behavior`, or None indicating `behavior` has no deadline.
`behavior` must be idempotent.
group (None): should be None. Reseved for future extensions
when ThreadGroup is implemented.
target (function): The function to invoke when this thread is
run. Defaults to None.
name (str): The name of this thread. Defaults to None.
args (tuple[object]): A tuple of arguments to pass to `target`.
kwargs (dict[str,object]): A dictionary of keyword arguments to
pass to `target`.
args: Positional arguments passed to threading.Thread constructor.
kwargs: Keyword arguments passed to threading.Thread constructor.
"""
super(CleanupThread, self).__init__(
group=group, target=target, name=name, args=args, kwargs=kwargs)
super(CleanupThread, self).__init__(*args, **kwargs)
self._behavior = behavior
def join(self, timeout=None):

@ -161,7 +161,8 @@
s.subspec 'Cronet-Interface' do |ss|
ss.header_mappings_dir = 'include/grpc'
ss.source_files = 'include/grpc/grpc_cronet.h'
ss.source_files = 'include/grpc/grpc_cronet.h',
'src/core/ext/transport/cronet/transport/cronet_transport.h'
end
s.subspec 'Cronet-Implementation' do |ss|

@ -91,6 +91,7 @@ LOWCPU = 0.1
# maps test names to options
END2END_TESTS = {
'authority_not_supported': default_test_options,
'bad_hostname': default_test_options,
'binary_metadata': default_test_options,
'resource_quota_server': default_test_options._replace(large_writes=True,
@ -142,7 +143,6 @@ END2END_TESTS = {
'simple_request': default_test_options,
'streaming_error_response': default_test_options,
'trailing_metadata': default_test_options,
'authority_not_supported': default_test_options,
'write_buffering': default_test_options,
'write_buffering_at_end': default_test_options,
}

@ -105,7 +105,6 @@ static void test_code(void) {
grpc_pollset_size();
grpc_pollset_init(NULL, NULL);
grpc_pollset_shutdown(NULL, NULL, NULL);
grpc_pollset_reset(NULL);
grpc_pollset_destroy(NULL);
GRPC_ERROR_UNREF(grpc_pollset_work(NULL, NULL, NULL,
gpr_now(GPR_CLOCK_REALTIME),

@ -0,0 +1,196 @@
/*
*
* Copyright 2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#include "test/core/util/passthru_endpoint.h"
#include <inttypes.h>
#include <string.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include <grpc/support/useful.h>
#include "src/core/lib/iomgr/sockaddr.h"
#include "src/core/lib/slice/slice_internal.h"
typedef struct {
grpc_endpoint base;
double bytes_per_second;
grpc_endpoint *wrapped;
gpr_timespec last_write;
gpr_mu mu;
grpc_slice_buffer write_buffer;
grpc_slice_buffer writing_buffer;
grpc_error *error;
bool writing;
} trickle_endpoint;
static void te_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_slice_buffer *slices, grpc_closure *cb) {
trickle_endpoint *te = (trickle_endpoint *)ep;
grpc_endpoint_read(exec_ctx, te->wrapped, slices, cb);
}
static void te_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_slice_buffer *slices, grpc_closure *cb) {
trickle_endpoint *te = (trickle_endpoint *)ep;
for (size_t i = 0; i < slices->count; i++) {
grpc_slice_ref_internal(slices->slices[i]);
}
gpr_mu_lock(&te->mu);
if (te->write_buffer.length == 0) {
te->last_write = gpr_now(GPR_CLOCK_MONOTONIC);
}
grpc_slice_buffer_addn(&te->write_buffer, slices->slices, slices->count);
grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_REF(te->error));
gpr_mu_unlock(&te->mu);
}
static grpc_workqueue *te_get_workqueue(grpc_endpoint *ep) {
trickle_endpoint *te = (trickle_endpoint *)ep;
return grpc_endpoint_get_workqueue(te->wrapped);
}
static void te_add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_pollset *pollset) {
trickle_endpoint *te = (trickle_endpoint *)ep;
grpc_endpoint_add_to_pollset(exec_ctx, te->wrapped, pollset);
}
static void te_add_to_pollset_set(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_pollset_set *pollset_set) {
trickle_endpoint *te = (trickle_endpoint *)ep;
grpc_endpoint_add_to_pollset_set(exec_ctx, te->wrapped, pollset_set);
}
static void te_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_error *why) {
trickle_endpoint *te = (trickle_endpoint *)ep;
gpr_mu_lock(&te->mu);
if (te->error == GRPC_ERROR_NONE) {
te->error = GRPC_ERROR_REF(why);
}
gpr_mu_unlock(&te->mu);
grpc_endpoint_shutdown(exec_ctx, te->wrapped, why);
}
static void te_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
trickle_endpoint *te = (trickle_endpoint *)ep;
grpc_endpoint_destroy(exec_ctx, te->wrapped);
gpr_mu_destroy(&te->mu);
grpc_slice_buffer_destroy_internal(exec_ctx, &te->write_buffer);
grpc_slice_buffer_destroy_internal(exec_ctx, &te->writing_buffer);
GRPC_ERROR_UNREF(te->error);
gpr_free(te);
}
static grpc_resource_user *te_get_resource_user(grpc_endpoint *ep) {
trickle_endpoint *te = (trickle_endpoint *)ep;
return grpc_endpoint_get_resource_user(te->wrapped);
}
static char *te_get_peer(grpc_endpoint *ep) {
trickle_endpoint *te = (trickle_endpoint *)ep;
return grpc_endpoint_get_peer(te->wrapped);
}
static int te_get_fd(grpc_endpoint *ep) {
trickle_endpoint *te = (trickle_endpoint *)ep;
return grpc_endpoint_get_fd(te->wrapped);
}
static void te_finish_write(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) {
trickle_endpoint *te = arg;
gpr_mu_lock(&te->mu);
te->writing = false;
grpc_slice_buffer_reset_and_unref(&te->writing_buffer);
gpr_mu_unlock(&te->mu);
}
static const grpc_endpoint_vtable vtable = {te_read,
te_write,
te_get_workqueue,
te_add_to_pollset,
te_add_to_pollset_set,
te_shutdown,
te_destroy,
te_get_resource_user,
te_get_peer,
te_get_fd};
grpc_endpoint *grpc_trickle_endpoint_create(grpc_endpoint *wrap,
double bytes_per_second) {
trickle_endpoint *te = gpr_malloc(sizeof(*te));
te->base.vtable = &vtable;
te->wrapped = wrap;
te->bytes_per_second = bytes_per_second;
gpr_mu_init(&te->mu);
grpc_slice_buffer_init(&te->write_buffer);
grpc_slice_buffer_init(&te->writing_buffer);
te->error = GRPC_ERROR_NONE;
te->writing = false;
return &te->base;
}
static double ts2dbl(gpr_timespec s) {
return (double)s.tv_sec + 1e-9 * (double)s.tv_nsec;
}
size_t grpc_trickle_endpoint_trickle(grpc_exec_ctx *exec_ctx,
grpc_endpoint *ep) {
trickle_endpoint *te = (trickle_endpoint *)ep;
gpr_mu_lock(&te->mu);
if (!te->writing && te->write_buffer.length > 0) {
gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
double elapsed = ts2dbl(gpr_time_sub(now, te->last_write));
size_t bytes = (size_t)(te->bytes_per_second * elapsed);
// gpr_log(GPR_DEBUG, "%lf elapsed --> %" PRIdPTR " bytes", elapsed, bytes);
if (bytes > 0) {
grpc_slice_buffer_move_first(&te->write_buffer,
GPR_MIN(bytes, te->write_buffer.length),
&te->writing_buffer);
te->writing = true;
te->last_write = now;
grpc_endpoint_write(
exec_ctx, te->wrapped, &te->writing_buffer,
grpc_closure_create(te_finish_write, te, grpc_schedule_on_exec_ctx));
}
}
size_t backlog = te->write_buffer.length;
gpr_mu_unlock(&te->mu);
return backlog;
}

@ -0,0 +1,46 @@
/*
*
* Copyright 2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#ifndef TRICKLE_ENDPOINT_H
#define TRICKLE_ENDPOINT_H
#include "src/core/lib/iomgr/endpoint.h"
grpc_endpoint *grpc_trickle_endpoint_create(grpc_endpoint *wrap,
double bytes_per_second);
/* Allow up to \a bytes through the endpoint. Returns the new backlog. */
size_t grpc_trickle_endpoint_trickle(grpc_exec_ctx *exec_ctx,
grpc_endpoint *endpoint);
#endif

@ -0,0 +1,356 @@
/*
*
* Copyright 2015, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
/* Test various closure related operations */
#include <grpc/grpc.h>
extern "C" {
#include "src/core/lib/iomgr/closure.h"
#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/iomgr/exec_ctx.h"
}
#include "third_party/benchmark/include/benchmark/benchmark.h"
static class InitializeStuff {
public:
InitializeStuff() { grpc_init(); }
~InitializeStuff() { grpc_shutdown(); }
} initialize_stuff;
static void BM_NoOpExecCtx(benchmark::State& state) {
while (state.KeepRunning()) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_exec_ctx_finish(&exec_ctx);
}
}
BENCHMARK(BM_NoOpExecCtx);
static void BM_WellFlushed(benchmark::State& state) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_exec_ctx_flush(&exec_ctx);
}
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_WellFlushed);
static void DoNothing(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {}
static void BM_ClosureInitAgainstExecCtx(benchmark::State& state) {
grpc_closure c;
while (state.KeepRunning()) {
benchmark::DoNotOptimize(
grpc_closure_init(&c, DoNothing, NULL, grpc_schedule_on_exec_ctx));
}
}
BENCHMARK(BM_ClosureInitAgainstExecCtx);
static void BM_ClosureInitAgainstCombiner(benchmark::State& state) {
grpc_combiner* combiner = grpc_combiner_create(NULL);
grpc_closure c;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
benchmark::DoNotOptimize(grpc_closure_init(
&c, DoNothing, NULL, grpc_combiner_scheduler(combiner, false)));
}
GRPC_COMBINER_UNREF(&exec_ctx, combiner, "finished");
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureInitAgainstCombiner);
static void BM_ClosureRunOnExecCtx(benchmark::State& state) {
grpc_closure c;
grpc_closure_init(&c, DoNothing, NULL, grpc_schedule_on_exec_ctx);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_run(&exec_ctx, &c, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureRunOnExecCtx);
static void BM_ClosureCreateAndRun(benchmark::State& state) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_run(&exec_ctx, grpc_closure_create(DoNothing, NULL,
grpc_schedule_on_exec_ctx),
GRPC_ERROR_NONE);
}
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureCreateAndRun);
static void BM_ClosureInitAndRun(benchmark::State& state) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_closure c;
while (state.KeepRunning()) {
grpc_closure_run(&exec_ctx, grpc_closure_init(&c, DoNothing, NULL,
grpc_schedule_on_exec_ctx),
GRPC_ERROR_NONE);
}
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureInitAndRun);
static void BM_ClosureSchedOnExecCtx(benchmark::State& state) {
grpc_closure c;
grpc_closure_init(&c, DoNothing, NULL, grpc_schedule_on_exec_ctx);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_sched(&exec_ctx, &c, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureSchedOnExecCtx);
static void BM_ClosureSched2OnExecCtx(benchmark::State& state) {
grpc_closure c1;
grpc_closure c2;
grpc_closure_init(&c1, DoNothing, NULL, grpc_schedule_on_exec_ctx);
grpc_closure_init(&c2, DoNothing, NULL, grpc_schedule_on_exec_ctx);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_sched(&exec_ctx, &c1, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c2, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureSched2OnExecCtx);
static void BM_ClosureSched3OnExecCtx(benchmark::State& state) {
grpc_closure c1;
grpc_closure c2;
grpc_closure c3;
grpc_closure_init(&c1, DoNothing, NULL, grpc_schedule_on_exec_ctx);
grpc_closure_init(&c2, DoNothing, NULL, grpc_schedule_on_exec_ctx);
grpc_closure_init(&c3, DoNothing, NULL, grpc_schedule_on_exec_ctx);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_sched(&exec_ctx, &c1, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c2, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c3, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureSched3OnExecCtx);
static void BM_AcquireMutex(benchmark::State& state) {
// for comparison with the combiner stuff below
gpr_mu mu;
gpr_mu_init(&mu);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
gpr_mu_lock(&mu);
DoNothing(&exec_ctx, NULL, GRPC_ERROR_NONE);
gpr_mu_unlock(&mu);
}
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_AcquireMutex);
static void BM_ClosureSchedOnCombiner(benchmark::State& state) {
grpc_combiner* combiner = grpc_combiner_create(NULL);
grpc_closure c;
grpc_closure_init(&c, DoNothing, NULL,
grpc_combiner_scheduler(combiner, false));
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_sched(&exec_ctx, &c, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
GRPC_COMBINER_UNREF(&exec_ctx, combiner, "finished");
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureSchedOnCombiner);
static void BM_ClosureSched2OnCombiner(benchmark::State& state) {
grpc_combiner* combiner = grpc_combiner_create(NULL);
grpc_closure c1;
grpc_closure c2;
grpc_closure_init(&c1, DoNothing, NULL,
grpc_combiner_scheduler(combiner, false));
grpc_closure_init(&c2, DoNothing, NULL,
grpc_combiner_scheduler(combiner, false));
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_sched(&exec_ctx, &c1, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c2, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
GRPC_COMBINER_UNREF(&exec_ctx, combiner, "finished");
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureSched2OnCombiner);
static void BM_ClosureSched3OnCombiner(benchmark::State& state) {
grpc_combiner* combiner = grpc_combiner_create(NULL);
grpc_closure c1;
grpc_closure c2;
grpc_closure c3;
grpc_closure_init(&c1, DoNothing, NULL,
grpc_combiner_scheduler(combiner, false));
grpc_closure_init(&c2, DoNothing, NULL,
grpc_combiner_scheduler(combiner, false));
grpc_closure_init(&c3, DoNothing, NULL,
grpc_combiner_scheduler(combiner, false));
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_sched(&exec_ctx, &c1, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c2, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c3, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
GRPC_COMBINER_UNREF(&exec_ctx, combiner, "finished");
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureSched3OnCombiner);
static void BM_ClosureSched2OnTwoCombiners(benchmark::State& state) {
grpc_combiner* combiner1 = grpc_combiner_create(NULL);
grpc_combiner* combiner2 = grpc_combiner_create(NULL);
grpc_closure c1;
grpc_closure c2;
grpc_closure_init(&c1, DoNothing, NULL,
grpc_combiner_scheduler(combiner1, false));
grpc_closure_init(&c2, DoNothing, NULL,
grpc_combiner_scheduler(combiner2, false));
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_sched(&exec_ctx, &c1, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c2, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
GRPC_COMBINER_UNREF(&exec_ctx, combiner1, "finished");
GRPC_COMBINER_UNREF(&exec_ctx, combiner2, "finished");
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureSched2OnTwoCombiners);
static void BM_ClosureSched4OnTwoCombiners(benchmark::State& state) {
grpc_combiner* combiner1 = grpc_combiner_create(NULL);
grpc_combiner* combiner2 = grpc_combiner_create(NULL);
grpc_closure c1;
grpc_closure c2;
grpc_closure c3;
grpc_closure c4;
grpc_closure_init(&c1, DoNothing, NULL,
grpc_combiner_scheduler(combiner1, false));
grpc_closure_init(&c2, DoNothing, NULL,
grpc_combiner_scheduler(combiner2, false));
grpc_closure_init(&c3, DoNothing, NULL,
grpc_combiner_scheduler(combiner1, false));
grpc_closure_init(&c4, DoNothing, NULL,
grpc_combiner_scheduler(combiner2, false));
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
while (state.KeepRunning()) {
grpc_closure_sched(&exec_ctx, &c1, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c2, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c3, GRPC_ERROR_NONE);
grpc_closure_sched(&exec_ctx, &c4, GRPC_ERROR_NONE);
grpc_exec_ctx_flush(&exec_ctx);
}
GRPC_COMBINER_UNREF(&exec_ctx, combiner1, "finished");
GRPC_COMBINER_UNREF(&exec_ctx, combiner2, "finished");
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureSched4OnTwoCombiners);
// Helper that continuously reschedules the same closure against something until
// the benchmark is complete
class Rescheduler {
public:
Rescheduler(benchmark::State& state, grpc_closure_scheduler* scheduler)
: state_(state) {
grpc_closure_init(&closure_, Step, this, scheduler);
}
void ScheduleFirst(grpc_exec_ctx* exec_ctx) {
grpc_closure_sched(exec_ctx, &closure_, GRPC_ERROR_NONE);
}
void ScheduleFirstAgainstDifferentScheduler(
grpc_exec_ctx* exec_ctx, grpc_closure_scheduler* scheduler) {
grpc_closure_sched(exec_ctx, grpc_closure_create(Step, this, scheduler),
GRPC_ERROR_NONE);
}
private:
benchmark::State& state_;
grpc_closure closure_;
static void Step(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
Rescheduler* self = static_cast<Rescheduler*>(arg);
if (self->state_.KeepRunning()) {
grpc_closure_sched(exec_ctx, &self->closure_, GRPC_ERROR_NONE);
}
}
};
static void BM_ClosureReschedOnExecCtx(benchmark::State& state) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
Rescheduler r(state, grpc_schedule_on_exec_ctx);
r.ScheduleFirst(&exec_ctx);
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureReschedOnExecCtx);
static void BM_ClosureReschedOnCombiner(benchmark::State& state) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_combiner* combiner = grpc_combiner_create(NULL);
Rescheduler r(state, grpc_combiner_scheduler(combiner, false));
r.ScheduleFirst(&exec_ctx);
grpc_exec_ctx_flush(&exec_ctx);
GRPC_COMBINER_UNREF(&exec_ctx, combiner, "finished");
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureReschedOnCombiner);
static void BM_ClosureReschedOnCombinerFinally(benchmark::State& state) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_combiner* combiner = grpc_combiner_create(NULL);
Rescheduler r(state, grpc_combiner_finally_scheduler(combiner, false));
r.ScheduleFirstAgainstDifferentScheduler(
&exec_ctx, grpc_combiner_scheduler(combiner, false));
grpc_exec_ctx_flush(&exec_ctx);
GRPC_COMBINER_UNREF(&exec_ctx, combiner, "finished");
grpc_exec_ctx_finish(&exec_ctx);
}
BENCHMARK(BM_ClosureReschedOnCombinerFinally);
BENCHMARK_MAIN();

@ -46,6 +46,7 @@
extern "C" {
#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
#include "src/core/ext/transport/chttp2/transport/internal.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/iomgr/endpoint.h"
#include "src/core/lib/iomgr/endpoint_pair.h"
@ -57,6 +58,7 @@ extern "C" {
#include "test/core/util/memory_counters.h"
#include "test/core/util/passthru_endpoint.h"
#include "test/core/util/port.h"
#include "test/core/util/trickle_endpoint.h"
}
#include "src/core/lib/profiling/timers.h"
#include "src/cpp/client/create_channel_internal.h"
@ -197,7 +199,8 @@ class UDS : public FullstackFixture {
class EndpointPairFixture : public BaseFixture {
public:
EndpointPairFixture(Service* service, grpc_endpoint_pair endpoints) {
EndpointPairFixture(Service* service, grpc_endpoint_pair endpoints)
: endpoint_pair_(endpoints) {
ServerBuilder b;
cq_ = b.AddCompletionQueue(true);
b.RegisterService(service);
@ -210,7 +213,7 @@ class EndpointPairFixture : public BaseFixture {
{
const grpc_channel_args* server_args =
grpc_server_get_channel_args(server_->c_server());
grpc_transport* transport = grpc_create_chttp2_transport(
server_transport_ = grpc_create_chttp2_transport(
&exec_ctx, server_args, endpoints.server, 0 /* is_client */);
grpc_pollset** pollsets;
@ -221,9 +224,9 @@ class EndpointPairFixture : public BaseFixture {
grpc_endpoint_add_to_pollset(&exec_ctx, endpoints.server, pollsets[i]);
}
grpc_server_setup_transport(&exec_ctx, server_->c_server(), transport,
NULL, server_args);
grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);
grpc_server_setup_transport(&exec_ctx, server_->c_server(),
server_transport_, NULL, server_args);
grpc_chttp2_transport_start_reading(&exec_ctx, server_transport_, NULL);
}
/* create channel */
@ -233,12 +236,13 @@ class EndpointPairFixture : public BaseFixture {
ApplyCommonChannelArguments(&args);
grpc_channel_args c_args = args.c_channel_args();
grpc_transport* transport =
client_transport_ =
grpc_create_chttp2_transport(&exec_ctx, &c_args, endpoints.client, 1);
GPR_ASSERT(transport);
grpc_channel* channel = grpc_channel_create(
&exec_ctx, "target", &c_args, GRPC_CLIENT_DIRECT_CHANNEL, transport);
grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);
GPR_ASSERT(client_transport_);
grpc_channel* channel =
grpc_channel_create(&exec_ctx, "target", &c_args,
GRPC_CLIENT_DIRECT_CHANNEL, client_transport_);
grpc_chttp2_transport_start_reading(&exec_ctx, client_transport_, NULL);
channel_ = CreateChannelInternal("", channel);
}
@ -258,6 +262,11 @@ class EndpointPairFixture : public BaseFixture {
ServerCompletionQueue* cq() { return cq_.get(); }
std::shared_ptr<Channel> channel() { return channel_; }
protected:
grpc_endpoint_pair endpoint_pair_;
grpc_transport* client_transport_;
grpc_transport* server_transport_;
private:
std::unique_ptr<Server> server_;
std::unique_ptr<ServerCompletionQueue> cq_;
@ -295,6 +304,75 @@ class InProcessCHTTP2 : public EndpointPairFixture {
}
};
class TrickledCHTTP2 : public EndpointPairFixture {
public:
TrickledCHTTP2(Service* service, size_t megabits_per_second)
: EndpointPairFixture(service, MakeEndpoints(megabits_per_second)) {}
void AddToLabel(std::ostream& out, benchmark::State& state) {
out << " writes/iter:"
<< ((double)stats_.num_writes / (double)state.iterations())
<< " cli_transport_stalls/iter:"
<< ((double)
client_stats_.streams_stalled_due_to_transport_flow_control /
(double)state.iterations())
<< " cli_stream_stalls/iter:"
<< ((double)client_stats_.streams_stalled_due_to_stream_flow_control /
(double)state.iterations())
<< " svr_transport_stalls/iter:"
<< ((double)
server_stats_.streams_stalled_due_to_transport_flow_control /
(double)state.iterations())
<< " svr_stream_stalls/iter:"
<< ((double)server_stats_.streams_stalled_due_to_stream_flow_control /
(double)state.iterations());
}
void Step() {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
size_t client_backlog =
grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.client);
size_t server_backlog =
grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.server);
grpc_exec_ctx_finish(&exec_ctx);
UpdateStats((grpc_chttp2_transport*)client_transport_, &client_stats_,
client_backlog);
UpdateStats((grpc_chttp2_transport*)server_transport_, &server_stats_,
server_backlog);
}
private:
grpc_passthru_endpoint_stats stats_;
struct Stats {
int streams_stalled_due_to_stream_flow_control = 0;
int streams_stalled_due_to_transport_flow_control = 0;
};
Stats client_stats_;
Stats server_stats_;
grpc_endpoint_pair MakeEndpoints(size_t kilobits) {
grpc_endpoint_pair p;
grpc_passthru_endpoint_create(&p.client, &p.server, initialize_stuff.rq(),
&stats_);
double bytes_per_second = 125.0 * kilobits;
p.client = grpc_trickle_endpoint_create(p.client, bytes_per_second);
p.server = grpc_trickle_endpoint_create(p.server, bytes_per_second);
return p;
}
void UpdateStats(grpc_chttp2_transport* t, Stats* s, size_t backlog) {
if (backlog == 0) {
if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != NULL) {
s->streams_stalled_due_to_stream_flow_control++;
}
if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != NULL) {
s->streams_stalled_due_to_transport_flow_control++;
}
}
}
};
/*******************************************************************************
* CONTEXT MUTATORS
*/
@ -620,6 +698,7 @@ static void BM_StreamingPingPongMsgs(benchmark::State& state) {
}
while (state.KeepRunning()) {
GPR_TIMER_SCOPE("BenchmarkCycle", 0);
request_rw->Write(send_request, tag(0)); // Start client send
response_rw.Read(&recv_request, tag(1)); // Start server recv
request_rw->Read(&recv_response, tag(2)); // Start client recv
@ -777,6 +856,81 @@ static void BM_PumpStreamServerToClient(benchmark::State& state) {
state.SetBytesProcessed(state.range(0) * state.iterations());
}
static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok) {
while (true) {
switch (fixture->cq()->AsyncNext(
t, ok, gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
gpr_time_from_micros(100, GPR_TIMESPAN)))) {
case CompletionQueue::TIMEOUT:
fixture->Step();
break;
case CompletionQueue::SHUTDOWN:
GPR_ASSERT(false);
break;
case CompletionQueue::GOT_EVENT:
return;
}
}
}
static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
EchoTestService::AsyncService service;
std::unique_ptr<TrickledCHTTP2> fixture(
new TrickledCHTTP2(&service, state.range(1)));
{
EchoResponse send_response;
EchoResponse recv_response;
if (state.range(0) > 0) {
send_response.set_message(std::string(state.range(0), 'a'));
}
Status recv_status;
ServerContext svr_ctx;
ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx);
service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(),
fixture->cq(), tag(0));
std::unique_ptr<EchoTestService::Stub> stub(
EchoTestService::NewStub(fixture->channel()));
ClientContext cli_ctx;
auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1));
int need_tags = (1 << 0) | (1 << 1);
void* t;
bool ok;
while (need_tags) {
TrickleCQNext(fixture.get(), &t, &ok);
GPR_ASSERT(ok);
int i = (int)(intptr_t)t;
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
}
request_rw->Read(&recv_response, tag(0));
while (state.KeepRunning()) {
GPR_TIMER_SCOPE("BenchmarkCycle", 0);
response_rw.Write(send_response, tag(1));
while (true) {
TrickleCQNext(fixture.get(), &t, &ok);
if (t == tag(0)) {
request_rw->Read(&recv_response, tag(0));
} else if (t == tag(1)) {
break;
} else {
GPR_ASSERT(false);
}
}
}
response_rw.Finish(Status::OK, tag(1));
need_tags = (1 << 0) | (1 << 1);
while (need_tags) {
TrickleCQNext(fixture.get(), &t, &ok);
int i = (int)(intptr_t)t;
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
}
}
fixture->Finish(state);
fixture.reset();
state.SetBytesProcessed(state.range(0) * state.iterations());
}
/*******************************************************************************
* CONFIGURATIONS
*/
@ -866,6 +1020,19 @@ BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, SockPair)
BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2)
->Range(0, 128 * 1024 * 1024);
static void TrickleArgs(benchmark::internal::Benchmark* b) {
for (int i = 1; i <= 128 * 1024 * 1024; i *= 8) {
for (int j = 1; j <= 128 * 1024 * 1024; j *= 8) {
double expected_time =
static_cast<double>(14 + i) / (125.0 * static_cast<double>(j));
if (expected_time > 0.01) continue;
b->Args({i, j});
}
}
}
BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(TrickleArgs);
// Generate Args for StreamingPingPong benchmarks. Currently generates args for
// only "small streams" (i.e streams with 0, 1 or 2 messages)
static void StreamingPingPongArgs(benchmark::internal::Benchmark* b) {

@ -83,7 +83,7 @@ popd
# this should be the same version as the submodule we compile against
# ideally we'd update this as a template to ensure that
pip install protobuf==3.0.0
pip install protobuf==3.2.0
pushd "$(dirname $INPUT_PROTO)" > /dev/null

@ -53,7 +53,7 @@ for dir in $DIRS; do
tempdir=`mktemp -d`
cp -RT $dir $tempdir
$PYTHON -m yapf -i -r -p $exclusion_args $dir
if ! diff -rq $dir $tempdir; then
if ! diff -r $dir $tempdir; then
script_result=1
fi
rm -rf $tempdir

@ -0,0 +1,41 @@
#!/bin/bash
# Copyright 2017, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
# * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
# Config file for the internal CI (in protobuf text format)
# Location of the continuous shell script in repository.
build_file: "grpc/tools/internal_ci/linux/grpc_interop.sh"
# grpc_interop tests can take 6+ hours to complete.
timeout_mins: 480
action {
define_artifacts {
regex: "**/sponge_log.xml"
}
}

@ -0,0 +1,42 @@
#!/usr/bin/env bash
# Copyright 2017, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
# * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
set -ex
export LANG=en_US.UTF-8
# Enter the gRPC repo root
cd $(dirname $0)/../../..
git submodule update --init
tools/run_tests/run_interop_tests.py -l all -s all --cloud_to_prod --cloud_to_prod_auth --use_docker --http2_interop -t -j 12 $@ || FAILED="true"
tools/run_tests/run_interop_tests.py -l java --use_docker --http2_badserver_interop $@ || FAILED="true"
tools/run_tests/run_interop_tests.py -l python --use_docker --http2_badserver_interop $@ || FAILED="true"

@ -61,6 +61,11 @@ columns = [
('allocs_per_iteration', 'float'),
('locks_per_iteration', 'float'),
('writes_per_iteration', 'float'),
('bandwidth_kilobits', 'integer'),
('cli_transport_stalls_per_iteration', 'float'),
('cli_stream_stalls_per_iteration', 'float'),
('svr_transport_stalls_per_iteration', 'float'),
('svr_stream_stalls_per_iteration', 'float'),
]
if sys.argv[1] == '--schema':
@ -92,7 +97,11 @@ bm_specs = {
'BM_StreamingPingPongMsgs': {
'tpl': ['fixture', 'client_mutator', 'server_mutator'],
'dyn': ['request_size'],
}
},
'BM_PumpStreamServerToClient_Trickle': {
'tpl': [],
'dyn': ['request_size', 'bandwidth_kilobits'],
},
}
def numericalize(s):
@ -106,6 +115,8 @@ def numericalize(s):
assert 'not a number: %s' % s
def parse_name(name):
if '<' not in name and '/' not in name and name not in bm_specs:
return {'name': name}
rest = name
out = {}
tpl_args = []
@ -136,7 +147,7 @@ def parse_name(name):
rest = s[0]
dyn_args = s[1:]
name = rest
assert name in bm_specs
assert name in bm_specs, 'bm_specs needs to be expanded for %s' % name
assert len(dyn_args) == len(bm_specs[name]['dyn'])
assert len(tpl_args) == len(bm_specs[name]['tpl'])
out['name'] = name
@ -146,10 +157,13 @@ def parse_name(name):
for bm in js['benchmarks']:
context = js['context']
labels_list = [s.split(':') for s in bm.get('label', '').split(' ')]
if 'label' in bm:
labels_list = [s.split(':') for s in bm['label'].split(' ')]
for el in labels_list:
el[0] = el[0].replace('/iter', '_per_iteration')
labels = dict(labels_list)
else:
labels = {}
row = {
'jenkins_build': os.environ.get('BUILD_NUMBER', ''),
'jenkins_job': os.environ.get('JOB_NAME', ''),
@ -158,5 +172,6 @@ for bm in js['benchmarks']:
row.update(bm)
row.update(parse_name(row['name']))
row.update(labels)
if 'label' in row:
del row['label']
writer.writerow(row)

@ -2363,6 +2363,24 @@
"third_party": false,
"type": "target"
},
{
"deps": [
"benchmark",
"gpr",
"gpr_test_util",
"grpc",
"grpc_test_util"
],
"headers": [],
"is_filegroup": false,
"language": "c++",
"name": "bm_closure",
"src": [
"test/cpp/microbenchmarks/bm_closure.cc"
],
"third_party": false,
"type": "target"
},
{
"deps": [
"benchmark",
@ -7779,7 +7797,8 @@
"test/core/util/passthru_endpoint.h",
"test/core/util/port.h",
"test/core/util/port_server_client.h",
"test/core/util/slice_splitter.h"
"test/core/util/slice_splitter.h",
"test/core/util/trickle_endpoint.h"
],
"is_filegroup": true,
"language": "c",
@ -7814,7 +7833,9 @@
"test/core/util/port_uv.c",
"test/core/util/port_windows.c",
"test/core/util/slice_splitter.c",
"test/core/util/slice_splitter.h"
"test/core/util/slice_splitter.h",
"test/core/util/trickle_endpoint.c",
"test/core/util/trickle_endpoint.h"
],
"third_party": false,
"type": "filegroup"
@ -8029,6 +8050,7 @@
"include/grpc/grpc_cronet.h",
"include/grpc/grpc_security.h",
"include/grpc/grpc_security_constants.h",
"src/core/ext/transport/cronet/transport/cronet_transport.h",
"third_party/objective_c/Cronet/bidirectional_stream_c.h"
],
"is_filegroup": true,
@ -8040,7 +8062,8 @@
"include/grpc/grpc_security_constants.h",
"src/core/ext/transport/cronet/client/secure/cronet_channel_create.c",
"src/core/ext/transport/cronet/transport/cronet_api_dummy.c",
"src/core/ext/transport/cronet/transport/cronet_transport.c"
"src/core/ext/transport/cronet/transport/cronet_transport.c",
"src/core/ext/transport/cronet/transport/cronet_transport.h"
],
"third_party": false,
"type": "filegroup"

@ -2469,6 +2469,28 @@
"windows"
]
},
{
"args": [
"--benchmark_min_time=0"
],
"ci_platforms": [
"linux",
"mac",
"posix"
],
"cpu_cost": 1.0,
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
"gtest": false,
"language": "c++",
"name": "bm_closure",
"platforms": [
"linux",
"mac",
"posix"
]
},
{
"args": [
"--benchmark_min_time=0"

@ -36,6 +36,7 @@ import tempfile
import sys
import time
import jobset
import socket
def start_port_server(port_server_port):
# check if a compatible port server is running

@ -28,6 +28,7 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import cgi
import multiprocessing
import os
import subprocess
@ -71,11 +72,12 @@ def heading(name):
def link(txt, tgt):
global index_html
index_html += "<p><a href=\"%s\">%s</a></p>\n" % (tgt, txt)
index_html += "<p><a href=\"%s\">%s</a></p>\n" % (
cgi.escape(tgt, quote=True), cgi.escape(txt))
def text(txt):
global index_html
index_html += "<p><pre>%s</pre></p>\n" % txt
index_html += "<p><pre>%s</pre></p>\n" % cgi.escape(txt)
def collect_latency(bm_name, args):
"""generate latency profiles"""
@ -126,30 +128,57 @@ def collect_perf(bm_name, args):
subprocess.check_call(
['make', bm_name,
'CONFIG=mutrace', '-j', '%d' % multiprocessing.cpu_count()])
benchmarks = []
profile_analysis = []
cleanup = []
for line in subprocess.check_output(['bins/mutrace/%s' % bm_name,
'--benchmark_list_tests']).splitlines():
subprocess.check_call(['perf', 'record', '-o', '%s-perf.data' % fnize(line),
'-g', '-c', '1000',
link(line, '%s.svg' % fnize(line))
benchmarks.append(
jobset.JobSpec(['perf', 'record', '-o', '%s-perf.data' % fnize(line),
'-g', '-F', '997',
'bins/mutrace/%s' % bm_name,
'--benchmark_filter=^%s$' % line,
'--benchmark_min_time=20'])
env = os.environ.copy()
env.update({
'--benchmark_min_time=10']))
profile_analysis.append(
jobset.JobSpec(['tools/run_tests/performance/process_local_perf_flamegraphs.sh'],
environ = {
'PERF_BASE_NAME': fnize(line),
'OUTPUT_DIR': 'reports',
'OUTPUT_FILENAME': fnize(line),
})
subprocess.check_call(['tools/run_tests/performance/process_local_perf_flamegraphs.sh'],
env=env)
}))
cleanup.append(jobset.JobSpec(['rm', '%s-perf.data' % fnize(line)]))
cleanup.append(jobset.JobSpec(['rm', '%s-out.perf' % fnize(line)]))
# periodically flush out the list of jobs: temporary space required for this
# processing is large
if len(benchmarks) >= 20:
# run up to half the cpu count: each benchmark can use up to two cores
# (one for the microbenchmark, one for the data flush)
jobset.run(benchmarks, maxjobs=1,
add_env={'GRPC_TEST_PORT_SERVER': 'localhost:%d' % port_server_port})
jobset.run(profile_analysis, maxjobs=multiprocessing.cpu_count())
jobset.run(cleanup, maxjobs=multiprocessing.cpu_count())
benchmarks = []
profile_analysis = []
cleanup = []
# run the remaining benchmarks that weren't flushed
if len(benchmarks):
jobset.run(benchmarks, maxjobs=1,
add_env={'GRPC_TEST_PORT_SERVER': 'localhost:%d' % port_server_port})
jobset.run(profile_analysis, maxjobs=multiprocessing.cpu_count())
jobset.run(cleanup, maxjobs=multiprocessing.cpu_count())
def collect_summary(bm_name, args):
heading('Summary: %s' % bm_name)
subprocess.check_call(
['make', bm_name,
'CONFIG=counters', '-j', '%d' % multiprocessing.cpu_count()])
text(subprocess.check_output(['bins/counters/%s' % bm_name,
cmd = ['bins/counters/%s' % bm_name,
'--benchmark_out=out.json',
'--benchmark_out_format=json']))
'--benchmark_out_format=json']
if args.summary_time is not None:
cmd += ['--benchmark_min_time=%d' % args.summary_time]
text(subprocess.check_output(cmd))
if args.bigquery_upload:
with open('out.csv', 'w') as f:
f.write(subprocess.check_output(['tools/profiling/microbenchmarks/bm2bq.py', 'out.json']))
@ -168,7 +197,7 @@ argp.add_argument('-c', '--collect',
default=sorted(collectors.keys()),
help='Which collectors should be run against each benchmark')
argp.add_argument('-b', '--benchmarks',
default=['bm_fullstack'],
default=['bm_fullstack', 'bm_closure'],
nargs='+',
type=str,
help='Which microbenchmarks should be run')
@ -177,6 +206,10 @@ argp.add_argument('--bigquery_upload',
action='store_const',
const=True,
help='Upload results from summary collection to bigquery')
argp.add_argument('--summary_time',
default=None,
type=int,
help='Minimum time to run benchmarks for the summary collection')
args = argp.parse_args()
for bm_name in args.benchmarks:

@ -80,6 +80,13 @@ def platform_string():
_DEFAULT_TIMEOUT_SECONDS = 5 * 60
def run_shell_command(cmd, env=None, cwd=None):
try:
subprocess.check_output(cmd, shell=True, env=env, cwd=cwd)
except subprocess.CalledProcessError as e:
print("Error while running command '%s'. Exit status %d. Output:\n%s",
e.cmd, e.returncode, e.output)
raise
# SimpleConfig: just compile with CONFIG=config, and run the binary to test
class Config(object):
@ -1199,7 +1206,7 @@ for spec in args.update_submodules:
cwd = 'third_party/%s' % submodule
def git(cmd, cwd=cwd):
print('in %s: git %s' % (cwd, cmd))
subprocess.check_call('git %s' % cmd, cwd=cwd, shell=True)
run_shell_command('git %s' % cmd, cwd=cwd)
git('fetch')
git('checkout %s' % branch)
git('pull origin %s' % branch)
@ -1207,7 +1214,7 @@ for spec in args.update_submodules:
need_to_regenerate_projects = True
if need_to_regenerate_projects:
if jobset.platform_string() == 'linux':
subprocess.check_call('tools/buildgen/generate_projects.sh', shell=True)
run_shell_command('tools/buildgen/generate_projects.sh')
else:
print('WARNING: may need to regenerate projects, but since we are not on')
print(' Linux this step is being skipped. Compilation MAY fail.')
@ -1276,9 +1283,7 @@ if args.use_docker:
if not args.travis:
env['TTY_FLAG'] = '-t' # enables Ctrl-C when not on Jenkins.
subprocess.check_call(['tools/run_tests/dockerize/build_docker_and_run_tests.sh'],
shell=True,
env=env)
run_shell_command('tools/run_tests/dockerize/build_docker_and_run_tests.sh', env=env)
sys.exit(0)
_check_arch_option(args.arch)

@ -193,6 +193,7 @@
<ClInclude Include="$(SolutionDir)\..\test\core\util\port.h" />
<ClInclude Include="$(SolutionDir)\..\test\core\util\port_server_client.h" />
<ClInclude Include="$(SolutionDir)\..\test\core\util\slice_splitter.h" />
<ClInclude Include="$(SolutionDir)\..\test\core\util\trickle_endpoint.h" />
<ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\channel_args.h" />
<ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\channel_stack.h" />
<ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\channel_stack_builder.h" />
@ -343,6 +344,8 @@
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\test\core\util\slice_splitter.c">
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\test\core\util\trickle_endpoint.c">
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\channel_args.c">
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\channel_stack.c">

@ -64,6 +64,9 @@
<ClCompile Include="$(SolutionDir)\..\test\core\util\slice_splitter.c">
<Filter>test\core\util</Filter>
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\test\core\util\trickle_endpoint.c">
<Filter>test\core\util</Filter>
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\channel_args.c">
<Filter>src\core\lib\channel</Filter>
</ClCompile>
@ -554,6 +557,9 @@
<ClInclude Include="$(SolutionDir)\..\test\core\util\slice_splitter.h">
<Filter>test\core\util</Filter>
</ClInclude>
<ClInclude Include="$(SolutionDir)\..\test\core\util\trickle_endpoint.h">
<Filter>test\core\util</Filter>
</ClInclude>
<ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\channel_args.h">
<Filter>src\core\lib\channel</Filter>
</ClInclude>

@ -161,6 +161,7 @@
<ClInclude Include="$(SolutionDir)\..\test\core\util\port.h" />
<ClInclude Include="$(SolutionDir)\..\test\core\util\port_server_client.h" />
<ClInclude Include="$(SolutionDir)\..\test\core\util\slice_splitter.h" />
<ClInclude Include="$(SolutionDir)\..\test\core\util\trickle_endpoint.h" />
</ItemGroup>
<ItemGroup>
<ClCompile Include="$(SolutionDir)\..\test\core\end2end\cq_verifier.c">
@ -195,6 +196,8 @@
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\test\core\util\slice_splitter.c">
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\test\core\util\trickle_endpoint.c">
</ClCompile>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="$(SolutionDir)\..\vsprojects\vcxproj\.\gpr\gpr.vcxproj">

@ -49,6 +49,9 @@
<ClCompile Include="$(SolutionDir)\..\test\core\util\slice_splitter.c">
<Filter>test\core\util</Filter>
</ClCompile>
<ClCompile Include="$(SolutionDir)\..\test\core\util\trickle_endpoint.c">
<Filter>test\core\util</Filter>
</ClCompile>
</ItemGroup>
<ItemGroup>
<ClInclude Include="$(SolutionDir)\..\test\core\end2end\cq_verifier.h">
@ -93,6 +96,9 @@
<ClInclude Include="$(SolutionDir)\..\test\core\util\slice_splitter.h">
<Filter>test\core\util</Filter>
</ClInclude>
<ClInclude Include="$(SolutionDir)\..\test\core\util\trickle_endpoint.h">
<Filter>test\core\util</Filter>
</ClInclude>
</ItemGroup>
<ItemGroup>

Loading…
Cancel
Save