Dynamically enable/disable packet coalecsing and test it

pull/9246/head
Muxi Yan 8 years ago
parent b0bd22dfc7
commit 60ab7ef00a
  1. 2
      include/grpc/grpc_cronet.h
  2. 6
      src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
  3. 72
      src/core/ext/transport/cronet/transport/cronet_transport.c
  4. 13
      src/objective-c/tests/CronetUnitTests/CronetUnitTests.m
  5. 1
      src/objective-c/tests/Podfile

@ -44,6 +44,8 @@ GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
void *engine, const char *target, const grpc_channel_args *args,
void *reserved);
GRPCAPI void grpc_cronet_use_packet_coalescing(bool use_coalescing);
#ifdef __cplusplus
}
#endif

@ -51,6 +51,8 @@ typedef struct cronet_transport {
extern grpc_transport_vtable grpc_cronet_vtable;
bool grpc_cronet_packet_coalescing_enabled = true;
GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
void *engine, const char *target, const grpc_channel_args *args,
void *reserved) {
@ -67,3 +69,7 @@ GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
return grpc_channel_create(&exec_ctx, target, args,
GRPC_CLIENT_DIRECT_CHANNEL, (grpc_transport *)ct);
}
GRPCAPI void grpc_cronet_use_packet_coalescing(bool use_coalescing) {
grpc_cronet_packet_coalescing_enabled = use_coalescing;
}

@ -61,6 +61,8 @@
/* TODO (makdharma): Hook up into the wider tracing mechanism */
int grpc_cronet_trace = 0;
extern bool grpc_cronet_packet_coalescing_enabled;
enum e_op_result {
ACTION_TAKEN_WITH_CALLBACK,
ACTION_TAKEN_NO_CALLBACK,
@ -150,12 +152,13 @@ struct op_state {
bool state_callback_received[OP_NUM_OPS];
bool fail_state;
bool flush_read;
#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
bool flush_cronet_when_ready;
bool pending_write_for_trailer;
#endif
bool unprocessed_send_message;
grpc_error *cancel_error;
/* Whether packet coalescing is enabled */
bool packet_coalescing_enabled;
/* data structure for storing data coming from server */
struct read_state rs;
/* data structure for storing data going to the server */
@ -425,12 +428,10 @@ static void on_stream_ready(bidirectional_stream *stream) {
}
/* Send the initial metadata on wire if there is no SEND_MESSAGE or
* SEND_TRAILING_METADATA ops pending */
#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
if (s->state.flush_cronet_when_ready) {
if (s->state.packet_coalescing_enabled && s->state.flush_cronet_when_ready) {
CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs);
bidirectional_stream_flush(stream);
}
#endif
gpr_mu_unlock(&s->mu);
execute_from_storage(s);
}
@ -568,10 +569,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);
#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
bidirectional_stream_flush(s->cbs);
#endif
if (s->state.packet_coalescing_enabled) {
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);
@ -768,11 +769,9 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op,
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]
#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
&& !stream_state->pending_write_for_trailer
#endif
)
!stream_state->state_callback_received[OP_SEND_MESSAGE] &&
!(stream_state->packet_coalescing_enabled &&
stream_state->pending_write_for_trailer))
result = false;
} else if (op_id == OP_CANCEL_ERROR) {
/* already executed */
@ -858,10 +857,10 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
s->cbs = bidirectional_stream_create(s->curr_ct.engine, s->curr_gs,
&cronet_callbacks);
CRONET_LOG(GPR_DEBUG, "%p = bidirectional_stream_create()", s->cbs);
#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
bidirectional_stream_disable_auto_flush(s->cbs, true);
bidirectional_stream_delay_request_headers_until_flush(s->cbs, true);
#endif
if (stream_state->packet_coalescing_enabled) {
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;
@ -872,11 +871,10 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_start(%p, %s)", s->cbs, url);
bidirectional_stream_start(s->cbs, url, 0, method, &s->header_array, false);
stream_state->state_op_done[OP_SEND_INITIAL_METADATA] = true;
#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
if (!stream_op->send_message && !stream_op->send_trailing_metadata) {
if (stream_state->packet_coalescing_enabled && !stream_op->send_message &&
!stream_op->send_trailing_metadata) {
s->state.flush_cronet_when_ready = true;
}
#endif
result = ACTION_TAKEN_WITH_CALLBACK;
} else if (stream_op->send_message &&
op_can_be_run(stream_op, stream_state, &oas->state,
@ -913,19 +911,18 @@ 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);
#ifdef GRPC_CRONET_WITH_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;
if (stream_state->packet_coalescing_enabled) {
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 {
stream_state->pending_write_for_trailer = true;
result = ACTION_TAKEN_NO_CALLBACK;
result = ACTION_TAKEN_WITH_CALLBACK;
}
#else
result = ACTION_TAKEN_WITH_CALLBACK;
#endif
} else {
result = NO_ACTION_POSSIBLE;
}
@ -944,10 +941,10 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
s->cbs);
stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
bidirectional_stream_write(s->cbs, "", 0, true);
#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
bidirectional_stream_flush(s->cbs);
#endif
if (stream_state->packet_coalescing_enabled) {
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;
@ -1176,10 +1173,9 @@ 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;
#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
s->state.flush_cronet_when_ready = s->state.pending_write_for_trailer = false;
#endif
s->state.unprocessed_send_message = false;
s->state.packet_coalescing_enabled = grpc_cronet_packet_coalescing_enabled;
gpr_mu_init(&s->mu);
return 0;
}

@ -269,7 +269,9 @@ unsigned int parse_h2_length(const char *field) {
grpc_completion_queue_destroy(cq);
}
- (void)testPacketCoalescing {
- (void)PacketCoalescing:(bool)use_coalescing {
grpc_cronet_use_packet_coalescing(use_coalescing);
grpc_call *c;
grpc_slice request_payload_slice =
grpc_slice_from_copied_string("hello world");
@ -379,7 +381,7 @@ unsigned int parse_h2_length(const char *field) {
long len;
bool coalesced = false;
while ((len = SSL_read(ssl, buf, sizeof(buf))) > 0) {
NSLog(@"Read len: %ld", len);
gpr_log(GPR_DEBUG, "Read len: %ld", len);
// Analyze the HTTP/2 frames in the same TLS PDU to identify if
// coalescing is successful
@ -404,7 +406,7 @@ unsigned int parse_h2_length(const char *field) {
}
}
XCTAssert(coalesced);
XCTAssert(coalesced == use_coalescing);
SSL_free(ssl);
SSL_CTX_free(ctx);
close(s);
@ -433,4 +435,9 @@ unsigned int parse_h2_length(const char *field) {
grpc_completion_queue_destroy(cq);
}
- (void)testPacketCoalescing {
[self PacketCoalescing:false];
[self PacketCoalescing:true];
}
@end

@ -97,7 +97,6 @@ post_install do |installer|
# GPR_UNREACHABLE_CODE causes "Control may reach end of non-void
# function" warning
config.build_settings['GCC_WARN_ABOUT_RETURN_TYPE'] = 'NO'
config.build_settings['GCC_PREPROCESSOR_DEFINITIONS'] = '$(inherited) COCOAPODS=1 GRPC_CRONET_WITH_PACKET_COALESCING=1'
end
end

Loading…
Cancel
Save