Merge branch 'bm_trickle' into bm_closure

pull/9693/head
Craig Tiller 8 years ago
commit 2b085f9923
  1. 38
      .pylintrc
  2. 1
      BUILD
  3. 2
      CMakeLists.txt
  4. 2
      Makefile
  5. 3
      build.yaml
  6. 3
      gRPC-Core.podspec
  7. 4
      include/grpc/impl/codegen/grpc_types.h
  8. 2
      src/core/ext/transport/chttp2/transport/frame_settings.h
  9. 13
      src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
  10. 12
      src/core/ext/transport/cronet/transport/cronet_api_dummy.c
  11. 246
      src/core/ext/transport/cronet/transport/cronet_transport.c
  12. 43
      src/core/ext/transport/cronet/transport/cronet_transport.h
  13. 331
      src/objective-c/tests/CronetUnitTests/CronetUnitTests.m
  14. 1
      src/objective-c/tests/Tests.xcodeproj/project.pbxproj
  15. 19
      src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme
  16. 3
      templates/gRPC-Core.podspec.template
  17. 2
      test/core/end2end/gen_build_yaml.py
  18. 196
      test/core/util/trickle_endpoint.c
  19. 46
      test/core/util/trickle_endpoint.h
  20. 186
      test/cpp/microbenchmarks/bm_fullstack.cc
  21. 2
      tools/codegen/core/gen_nano_proto.sh
  22. 48
      tools/distrib/pylint_code.sh
  23. 11
      tools/profiling/microbenchmarks/bm2bq.py
  24. 11
      tools/run_tests/generated/sources_and_headers.json
  25. 38
      tools/run_tests/run_microbenchmark.py
  26. 1
      tools/run_tests/sanity/sanity_tests.yaml
  27. 3
      vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj
  28. 6
      vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters
  29. 3
      vsprojects/vcxproj/grpc_test_util_unsecure/grpc_test_util_unsecure.vcxproj
  30. 6
      vsprojects/vcxproj/grpc_test_util_unsecure/grpc_test_util_unsecure.vcxproj.filters

@ -0,0 +1,38 @@
[MESSAGES CONTROL]
#TODO: Enable missing-docstring
#TODO: Enable too-few-public-methods
#TODO: Enable too-many-arguments
#TODO: Enable no-init
#TODO: Enable duplicate-code
#TODO: Enable invalid-name
#TODO: Enable suppressed-message
#TODO: Enable locally-disabled
#TODO: Enable protected-access
#TODO: Enable no-name-in-module
#TODO: Enable unused-argument
#TODO: Enable fixme
#TODO: Enable wrong-import-order
#TODO: Enable no-value-for-parameter
#TODO: Enable cyclic-import
#TODO: Enable unused-variable
#TODO: Enable redefined-outer-name
#TODO: Enable unused-import
#TODO: Enable too-many-instance-attributes
#TODO: Enable broad-except
#TODO: Enable too-many-locals
#TODO: Enable too-many-lines
#TODO: Enable redefined-variable-type
#TODO: Enable next-method-called
#TODO: Enable import-error
#TODO: Enable useless-else-on-loop
#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

@ -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 = [

@ -1431,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
@ -1638,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)

@ -3267,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 \
@ -3467,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 += \

@ -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

@ -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

@ -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);

@ -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);
result = ACTION_TAKEN_WITH_CALLBACK;
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,14 +1257,58 @@ 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),
"cronet_http",
init_stream,
set_pollset_do_nothing,
set_pollset_set_do_nothing,
perform_stream_op,
perform_op,
destroy_stream,
destroy_transport,
get_peer,
get_endpoint};
static const grpc_transport_vtable grpc_cronet_vtable = {
sizeof(stream_obj),
"cronet_http",
init_stream,
set_pollset_do_nothing,
set_pollset_set_do_nothing,
perform_stream_op,
perform_op,
destroy_stream,
destroy_transport,
get_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 */

@ -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,47 +71,99 @@ 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];
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 =
grpc_slice_from_copied_string("hello world");
grpc_slice_from_copied_string("hello world");
grpc_byte_buffer *request_payload =
grpc_raw_byte_buffer_create(&request_payload_slice, 1);
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,35 +218,40 @@ 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);
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), ^{
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);
bind(sl, (struct sockaddr*)&s_addr, sizeof(s_addr));
listen(sl, 5);
int s = accept(sl, NULL, NULL);
sleep(1);
close(s);
close(sl);
});
CQ_EXPECT_COMPLETION(cqv, (void*)1, 1);
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);
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);
});
CQ_EXPECT_COMPLETION(cqv, (void *)1, 1);
cq_verify(cqv);
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);
@ -204,11 +263,195 @@ static void drain_cq(grpc_completion_queue *cq) {
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);
}
- (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">

@ -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,
}

@ -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

@ -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
*/
@ -777,6 +855,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 +1019,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

@ -0,0 +1,48 @@
#!/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.
set -ex
# change to root directory
cd $(dirname $0)/../..
DIRS=src/python/grpcio/grpc
VIRTUALENV=python_pylint_venv
virtualenv $VIRTUALENV
PYTHON=`realpath $VIRTUALENV/bin/python`
$PYTHON -m pip install pylint==1.6.5
for dir in $DIRS; do
$PYTHON -m pylint --rcfile=.pylintrc -rn $dir || exit $?
done
exit 0

@ -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):

@ -7797,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",
@ -7832,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"
@ -8047,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,
@ -8058,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"

@ -128,31 +128,35 @@ def collect_perf(bm_name, args):
'CONFIG=mutrace', '-j', '%d' % multiprocessing.cpu_count()])
for line in subprocess.check_output(['bins/mutrace/%s' % bm_name,
'--benchmark_list_tests']).splitlines():
subprocess.check_call(['sudo', 'perf', 'record', '-o', 'perf.data',
subprocess.check_call(['perf', 'record', '-o', '%s-perf.data' % fnize(line),
'-g', '-c', '1000',
'bins/mutrace/%s' % bm_name,
'--benchmark_filter=^%s$' % line,
'--benchmark_min_time=20'])
with open('bm.perf', 'w') as f:
f.write(subprocess.check_output(['sudo', 'perf', 'script']))
with open('bm.folded', 'w') as f:
f.write(subprocess.check_output([
'%s/stackcollapse-perf.pl' % flamegraph_dir, 'bm.perf']))
link(line, '%s.svg' % fnize(line))
with open('reports/%s.svg' % fnize(line), 'w') as f:
f.write(subprocess.check_output([
'%s/flamegraph.pl' % flamegraph_dir, 'bm.folded']))
'--benchmark_min_time=10'])
env = os.environ.copy()
env.update({
'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)
subprocess.check_call(['rm', '%s-perf.data' % fnize(line)])
subprocess.check_call(['rm', '%s-out.perf' % fnize(line)])
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,
'--benchmark_out=out.json',
'--benchmark_out_format=json']))
cmd = ['bins/counters/%s' % bm_name,
'--benchmark_out=out.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('/tmp/out.csv', 'w') as f:
with open('out.csv', 'w') as f:
f.write(subprocess.check_output(['tools/profiling/microbenchmarks/bm2bq.py', 'out.json']))
subprocess.check_call(['bq', 'load', 'microbenchmarks.microbenchmarks', 'out.csv'])
@ -178,6 +182,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:

@ -12,6 +12,7 @@
- script: tools/distrib/check_trailing_newlines.sh
- script: tools/distrib/check_nanopb_output.sh
- script: tools/distrib/check_include_guards.py
- script: tools/distrib/pylint_code.sh
- script: tools/distrib/yapf_code.sh
- script: tools/distrib/python/check_grpcio_tools.py

@ -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