From a050ae8ddc3a64151b344fd1a4d438db9dea2acb Mon Sep 17 00:00:00 2001 From: "Nicolas \"Pixel\" Noble" Date: Thu, 13 Dec 2018 00:29:10 +0100 Subject: [PATCH 1/2] Revert "better slice management for win_read" This reverts commit b0139e15425196be518b251dbdfa3b86648b4740. --- src/core/lib/iomgr/tcp_windows.cc | 57 +++++++++---------------------- 1 file changed, 16 insertions(+), 41 deletions(-) diff --git a/src/core/lib/iomgr/tcp_windows.cc b/src/core/lib/iomgr/tcp_windows.cc index 86ee1010cf7..aaf9fb4ea8a 100644 --- a/src/core/lib/iomgr/tcp_windows.cc +++ b/src/core/lib/iomgr/tcp_windows.cc @@ -113,10 +113,7 @@ typedef struct grpc_tcp { grpc_closure* read_cb; grpc_closure* write_cb; - - /* garbage after the last read */ - grpc_slice_buffer last_read_buffer; - + grpc_slice read_slice; grpc_slice_buffer* write_slices; grpc_slice_buffer* read_slices; @@ -135,7 +132,6 @@ 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); @@ -184,6 +180,7 @@ 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()) { @@ -197,19 +194,11 @@ 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_buffer_reset_and_unref_internal(tcp->read_slices); + grpc_slice_unref_internal(tcp->read_slice); } else { if (info->bytes_transfered != 0 && !tcp->shutting_down) { - GPR_ASSERT((size_t)info->bytes_transfered <= tcp->read_slices->length); - if (static_cast(info->bytes_transfered) != - tcp->read_slices->length) { - grpc_slice_buffer_trim_end( - tcp->read_slices, - tcp->read_slices->length - - static_cast(info->bytes_transfered), - &tcp->last_read_buffer); - } - GPR_ASSERT((size_t)info->bytes_transfered == tcp->read_slices->length); + sub = grpc_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered); + grpc_slice_buffer_add(tcp->read_slices, sub); if (grpc_tcp_trace.enabled()) { size_t i; @@ -225,7 +214,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_buffer_reset_and_unref_internal(tcp->read_slices); + grpc_slice_unref_internal(tcp->read_slice); error = tcp->shutting_down ? GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "TCP stream shutting down", &tcp->shutdown_error, 1) @@ -239,8 +228,6 @@ 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; @@ -249,8 +236,7 @@ static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices, int status; DWORD bytes_read = 0; DWORD flags = 0; - WSABUF buffers[MAX_WSABUF_COUNT]; - size_t i; + WSABUF buffer; if (grpc_tcp_trace.enabled()) { gpr_log(GPR_INFO, "TCP:%p win_read", tcp); @@ -266,27 +252,18 @@ 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); - 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)); - } + tcp->read_slice = GRPC_SLICE_MALLOC(8192); - 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]); - } + 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); TCP_REF(tcp, "read"); /* First let's try a synchronous, non-blocking read. */ - status = WSARecv(tcp->socket->socket, buffers, (DWORD)tcp->read_slices->count, - &bytes_read, &flags, NULL, NULL); + status = + WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, NULL, NULL); info->wsa_error = status == 0 ? 0 : WSAGetLastError(); /* Did we get data immediately ? Yay. */ @@ -298,8 +275,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, buffers, (DWORD)tcp->read_slices->count, - &bytes_read, &flags, &info->overlapped, NULL); + status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, + &info->overlapped, NULL); if (status != 0) { int wsa_error = WSAGetLastError(); @@ -353,7 +330,7 @@ static void win_write(grpc_endpoint* ep, grpc_slice_buffer* slices, unsigned i; DWORD bytes_sent; int status; - WSABUF local_buffers[MAX_WSABUF_COUNT]; + WSABUF local_buffers[16]; WSABUF* allocated = NULL; WSABUF* buffers = local_buffers; size_t len; @@ -472,7 +449,6 @@ 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"); } @@ -524,7 +500,6 @@ 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); From f438d72e6c5e5bd839a255322fb91c416822f629 Mon Sep 17 00:00:00 2001 From: "Nicolas \"Pixel\" Noble" Date: Thu, 13 Dec 2018 00:29:23 +0100 Subject: [PATCH 2/2] Revert "basic tcp_trace support for windows" This reverts commit 5861f082607344ed42215ac341e97e4b4bbf0abc. --- src/core/lib/iomgr/tcp_windows.cc | 37 ------------------------------- 1 file changed, 37 deletions(-) diff --git a/src/core/lib/iomgr/tcp_windows.cc b/src/core/lib/iomgr/tcp_windows.cc index aaf9fb4ea8a..4b5250803d1 100644 --- a/src/core/lib/iomgr/tcp_windows.cc +++ b/src/core/lib/iomgr/tcp_windows.cc @@ -42,7 +42,6 @@ #include "src/core/lib/iomgr/tcp_windows.h" #include "src/core/lib/iomgr/timer.h" #include "src/core/lib/slice/slice_internal.h" -#include "src/core/lib/slice/slice_string_helpers.h" #if defined(__MSYS__) && defined(GPR_ARCH_64) /* Nasty workaround for nasty bug when using the 64 bits msys compiler @@ -183,10 +182,6 @@ static void on_read(void* tcpp, grpc_error* error) { grpc_slice sub; grpc_winsocket_callback_info* info = &socket->read_info; - if (grpc_tcp_trace.enabled()) { - gpr_log(GPR_INFO, "TCP:%p on_read", tcp); - } - GRPC_ERROR_REF(error); if (error == GRPC_ERROR_NONE) { @@ -199,21 +194,7 @@ static void on_read(void* tcpp, grpc_error* error) { 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); - - if (grpc_tcp_trace.enabled()) { - size_t i; - for (i = 0; i < tcp->read_slices->count; i++) { - char* dump = grpc_dump_slice(tcp->read_slices->slices[i], - GPR_DUMP_HEX | GPR_DUMP_ASCII); - gpr_log(GPR_INFO, "READ %p (peer=%s): %s", tcp, tcp->peer_string, - dump); - gpr_free(dump); - } - } } else { - if (grpc_tcp_trace.enabled()) { - gpr_log(GPR_INFO, "TCP:%p unref read_slice", tcp); - } grpc_slice_unref_internal(tcp->read_slice); error = tcp->shutting_down ? GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( @@ -238,10 +219,6 @@ static void win_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices, DWORD flags = 0; WSABUF buffer; - if (grpc_tcp_trace.enabled()) { - gpr_log(GPR_INFO, "TCP:%p win_read", tcp); - } - if (tcp->shutting_down) { GRPC_CLOSURE_SCHED( cb, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( @@ -298,10 +275,6 @@ static void on_write(void* tcpp, grpc_error* error) { grpc_winsocket_callback_info* info = &handle->write_info; grpc_closure* cb; - if (grpc_tcp_trace.enabled()) { - gpr_log(GPR_INFO, "TCP:%p on_write", tcp); - } - GRPC_ERROR_REF(error); gpr_mu_lock(&tcp->mu); @@ -335,16 +308,6 @@ static void win_write(grpc_endpoint* ep, grpc_slice_buffer* slices, WSABUF* buffers = local_buffers; size_t len; - if (grpc_tcp_trace.enabled()) { - size_t i; - for (i = 0; i < slices->count; i++) { - char* data = - grpc_dump_slice(slices->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); - gpr_log(GPR_INFO, "WRITE %p (peer=%s): %s", tcp, tcp->peer_string, data); - gpr_free(data); - } - } - if (tcp->shutting_down) { GRPC_CLOSURE_SCHED( cb, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(