Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Revert "Revert "better slice management for win_read""

Browse Source 
This reverts commit a050ae8ddc.
pull/17611/head
Jan Tattermusch 6 years ago
parent f0e960714e
commit 12aae4f7bb
1 changed files with 41 additions and 16 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 57
      src/core/lib/iomgr/tcp_windows.cc

57
src/core/lib/iomgr/tcp_windows.cc
Unescape View File

@ -113,7 +113,10 @@ typedef struct grpc_tcp {
grpc_closure* read_cb;
grpc_closure* write_cb;
grpc_slice read_slice;
/* garbage after the last read */
grpc_slice_buffer last_read_buffer;
grpc_slice_buffer* write_slices;
grpc_slice_buffer* read_slices;
@ -132,6 +135,7 @@ static void tcp_free(grpc_tcp* tcp) {
grpc_winsocket_destroy(tcp->socket);
gpr_mu_destroy(&tcp->mu);
gpr_free(tcp->peer_string);
grpc_slice_buffer_destroy_internal(&tcp->last_read_buffer);
grpc_resource_user_unref(tcp->resource_user);
if (tcp->shutting_down) GRPC_ERROR_UNREF(tcp->shutdown_error);
gpr_free(tcp);
@ -180,7 +184,6 @@ static void on_read(void* tcpp, grpc_error* error) {
grpc_tcp* tcp = (grpc_tcp*)tcpp;
grpc_closure* cb = tcp->read_cb;
grpc_winsocket* socket = tcp->socket;
grpc_slice sub;
grpc_winsocket_callback_info* info = &socket->read_info;
if (grpc_tcp_trace.enabled()) {
@ -194,11 +197,19 @@ static void on_read(void* tcpp, grpc_error* error) {
char* utf8_message = gpr_format_message(info->wsa_error);
error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(utf8_message);
gpr_free(utf8_message);
grpc_slice_unref_internal(tcp->read_slice);
grpc_slice_buffer_reset_and_unref_internal(tcp->read_slices);
} else {
if (info->bytes_transfered != 0 && !tcp->shutting_down) {
sub = grpc_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered);
grpc_slice_buffer_add(tcp->read_slices, sub);
GPR_ASSERT((size_t)info->bytes_transfered <= tcp->read_slices->length);
if (static_cast<size_t>(info->bytes_transfered) !=
tcp->read_slices->length) {
grpc_slice_buffer_trim_end(
tcp->read_slices,
tcp->read_slices->length -
static_cast<size_t>(info->bytes_transfered),
&tcp->last_read_buffer);
}
GPR_ASSERT((size_t)info->bytes_transfered == tcp->read_slices->length);
if (grpc_tcp_trace.enabled()) {
size_t i;
@ -214,7 +225,7 @@ static void on_read(void* tcpp, grpc_error* error) {
if (grpc_tcp_trace.enabled()) {
gpr_log(GPR_INFO, "TCP:%p unref read_slice", tcp);
}
grpc_slice_unref_internal(tcp->read_slice);
grpc_slice_buffer_reset_and_unref_internal(tcp->read_slices);
error = tcp->shutting_down
? GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"TCP stream shutting down", &tcp->shutdown_error, 1)
@ -228,6 +239,8 @@ static void on_read(void* tcpp, grpc_error* error) {
GRPC_CLOSURE_SCHED(cb, error);
}
#define DEFAULT_TARGET_READ_SIZE 8192
#define MAX_WSABUF_COUNT 16
static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices,
grpc_closure* cb) {
grpc_tcp* tcp = (grpc_tcp*)ep;
@ -236,7 +249,8 @@ static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices,
int status;
DWORD bytes_read = 0;
DWORD flags = 0;
WSABUF buffer;
WSABUF buffers[MAX_WSABUF_COUNT];
size_t i;
if (grpc_tcp_trace.enabled()) {
gpr_log(GPR_INFO, "TCP:%p win_read", tcp);
@ -252,18 +266,27 @@ static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices,
tcp->read_cb = cb;
tcp->read_slices = read_slices;
grpc_slice_buffer_reset_and_unref_internal(read_slices);
grpc_slice_buffer_swap(read_slices, &tcp->last_read_buffer);
tcp->read_slice = GRPC_SLICE_MALLOC(8192);
if (tcp->read_slices->length < DEFAULT_TARGET_READ_SIZE / 2 &&
tcp->read_slices->count < MAX_WSABUF_COUNT) {
// TODO(jtattermusch): slice should be allocated using resource quota
grpc_slice_buffer_add(tcp->read_slices,
GRPC_SLICE_MALLOC(DEFAULT_TARGET_READ_SIZE));
}
buffer.len = (ULONG)GRPC_SLICE_LENGTH(
tcp->read_slice); // we know slice size fits in 32bit.
buffer.buf = (char*)GRPC_SLICE_START_PTR(tcp->read_slice);
GPR_ASSERT(tcp->read_slices->count <= MAX_WSABUF_COUNT);
for (i = 0; i < tcp->read_slices->count; i++) {
buffers[i].len = (ULONG)GRPC_SLICE_LENGTH(
tcp->read_slices->slices[i]); // we know slice size fits in 32bit.
buffers[i].buf = (char*)GRPC_SLICE_START_PTR(tcp->read_slices->slices[i]);
}
TCP_REF(tcp, "read");
/* First let's try a synchronous, non-blocking read. */
status =
WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, NULL, NULL);
status = WSARecv(tcp->socket->socket, buffers, (DWORD)tcp->read_slices->count,
&bytes_read, &flags, NULL, NULL);
info->wsa_error = status == 0 ? 0 : WSAGetLastError();
/* Did we get data immediately ? Yay. */
@ -275,8 +298,8 @@ static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices,
/* Otherwise, let's retry, by queuing a read. */
memset(&tcp->socket->read_info.overlapped, 0, sizeof(OVERLAPPED));
status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags,
&info->overlapped, NULL);
status = WSARecv(tcp->socket->socket, buffers, (DWORD)tcp->read_slices->count,
&bytes_read, &flags, &info->overlapped, NULL);
if (status != 0) {
int wsa_error = WSAGetLastError();
@ -330,7 +353,7 @@ static void win_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
unsigned i;
DWORD bytes_sent;
int status;
WSABUF local_buffers[16];
WSABUF local_buffers[MAX_WSABUF_COUNT];
WSABUF* allocated = NULL;
WSABUF* buffers = local_buffers;
size_t len;
@ -449,6 +472,7 @@ static void win_shutdown(grpc_endpoint* ep, grpc_error* why) {
static void win_destroy(grpc_endpoint* ep) {
grpc_network_status_unregister_endpoint(ep);
grpc_tcp* tcp = (grpc_tcp*)ep;
grpc_slice_buffer_reset_and_unref_internal(&tcp->last_read_buffer);
TCP_UNREF(tcp, "destroy");
}
@ -500,6 +524,7 @@ grpc_endpoint* grpc_tcp_create(grpc_winsocket* socket,
GRPC_CLOSURE_INIT(&tcp->on_read, on_read, tcp, grpc_schedule_on_exec_ctx);
GRPC_CLOSURE_INIT(&tcp->on_write, on_write, tcp, grpc_schedule_on_exec_ctx);
tcp->peer_string = gpr_strdup(peer_string);
grpc_slice_buffer_init(&tcp->last_read_buffer);
tcp->resource_user = grpc_resource_user_create(resource_quota, peer_string);
/* Tell network status tracking code about the new endpoint */
grpc_network_status_register_endpoint(&tcp->base);

Write Preview
Loading…
Cancel
Save