diff options
Diffstat (limited to 'src/core/lib')
-rw-r--r-- | src/core/lib/iomgr/buffer_list.cc | 15 | ||||
-rw-r--r-- | src/core/lib/iomgr/buffer_list.h | 2 | ||||
-rw-r--r-- | src/core/lib/iomgr/endpoint.cc | 7 | ||||
-rw-r--r-- | src/core/lib/iomgr/endpoint.h | 3 | ||||
-rw-r--r-- | src/core/lib/iomgr/endpoint_pair_posix.cc | 4 | ||||
-rw-r--r-- | src/core/lib/iomgr/tcp_custom.cc | 5 | ||||
-rw-r--r-- | src/core/lib/iomgr/tcp_posix.cc | 68 | ||||
-rw-r--r-- | src/core/lib/security/transport/secure_endpoint.cc | 8 |
8 files changed, 99 insertions, 13 deletions
diff --git a/src/core/lib/iomgr/buffer_list.cc b/src/core/lib/iomgr/buffer_list.cc index 6ada23db1c..d7884a5965 100644 --- a/src/core/lib/iomgr/buffer_list.cc +++ b/src/core/lib/iomgr/buffer_list.cc @@ -31,6 +31,7 @@ namespace grpc_core { void TracedBuffer::AddNewEntry(TracedBuffer** head, uint32_t seq_no, void* arg) { + gpr_log(GPR_INFO, "new entry %u", seq_no); GPR_DEBUG_ASSERT(head != nullptr); TracedBuffer* new_elem = New<TracedBuffer>(seq_no, arg); /* Store the current time as the sendmsg time. */ @@ -64,6 +65,7 @@ void (*timestamps_callback)(void*, grpc_core::Timestamps*, void TracedBuffer::ProcessTimestamp(TracedBuffer** head, struct sock_extended_err* serr, struct scm_timestamping* tss) { + gpr_log(GPR_INFO, "process timestamp %u", serr->ee_data); GPR_DEBUG_ASSERT(head != nullptr); TracedBuffer* elem = *head; TracedBuffer* next = nullptr; @@ -85,6 +87,7 @@ void TracedBuffer::ProcessTimestamp(TracedBuffer** head, /* Got all timestamps. Do the callback and free this TracedBuffer. * The thing below can be passed by value if we don't want the * restriction on the lifetime. */ + gpr_log(GPR_INFO, "calling"); timestamps_callback(elem->arg_, &(elem->ts_), GRPC_ERROR_NONE); next = elem->next_; Delete<TracedBuffer>(elem); @@ -99,18 +102,22 @@ void TracedBuffer::ProcessTimestamp(TracedBuffer** head, } } -void TracedBuffer::Shutdown(TracedBuffer** head, grpc_error* shutdown_err) { +void TracedBuffer::Shutdown(TracedBuffer** head, void* remaining, + grpc_error* shutdown_err) { GPR_DEBUG_ASSERT(head != nullptr); TracedBuffer* elem = *head; + gpr_log(GPR_INFO, "shutdown"); while (elem != nullptr) { - if (timestamps_callback) { - timestamps_callback(elem->arg_, &(elem->ts_), shutdown_err); - } + timestamps_callback(elem->arg_, &(elem->ts_), shutdown_err); + gpr_log(GPR_INFO, "iter"); auto* next = elem->next_; Delete<TracedBuffer>(elem); elem = next; } *head = nullptr; + if (remaining != nullptr) { + timestamps_callback(remaining, nullptr, shutdown_err); + } GRPC_ERROR_UNREF(shutdown_err); } diff --git a/src/core/lib/iomgr/buffer_list.h b/src/core/lib/iomgr/buffer_list.h index cbbf50a657..f7d2f6c370 100644 --- a/src/core/lib/iomgr/buffer_list.h +++ b/src/core/lib/iomgr/buffer_list.h @@ -67,7 +67,7 @@ class TracedBuffer { /** Cleans the list by calling the callback for each traced buffer in the list * with timestamps that it has. */ - static void Shutdown(grpc_core::TracedBuffer** head, + static void Shutdown(grpc_core::TracedBuffer** head, void* remaining, grpc_error* shutdown_err); private: diff --git a/src/core/lib/iomgr/endpoint.cc b/src/core/lib/iomgr/endpoint.cc index 44fb47e19d..5e5effb2f1 100644 --- a/src/core/lib/iomgr/endpoint.cc +++ b/src/core/lib/iomgr/endpoint.cc @@ -61,3 +61,10 @@ int grpc_endpoint_get_fd(grpc_endpoint* ep) { return ep->vtable->get_fd(ep); } grpc_resource_user* grpc_endpoint_get_resource_user(grpc_endpoint* ep) { return ep->vtable->get_resource_user(ep); } + +bool grpc_endpoint_can_track_err(grpc_endpoint* ep) { + if (ep->vtable->can_track_err != nullptr) { + return ep->vtable->can_track_err(ep); + } + return false; +} diff --git a/src/core/lib/iomgr/endpoint.h b/src/core/lib/iomgr/endpoint.h index 1f590a80ca..79c8ece263 100644 --- a/src/core/lib/iomgr/endpoint.h +++ b/src/core/lib/iomgr/endpoint.h @@ -47,6 +47,7 @@ struct grpc_endpoint_vtable { grpc_resource_user* (*get_resource_user)(grpc_endpoint* ep); char* (*get_peer)(grpc_endpoint* ep); int (*get_fd)(grpc_endpoint* ep); + bool (*can_track_err)(grpc_endpoint* ep); }; /* When data is available on the connection, calls the callback with slices. @@ -95,6 +96,8 @@ void grpc_endpoint_delete_from_pollset_set(grpc_endpoint* ep, grpc_resource_user* grpc_endpoint_get_resource_user(grpc_endpoint* endpoint); +bool grpc_endpoint_can_track_err(grpc_endpoint* ep); + struct grpc_endpoint { const grpc_endpoint_vtable* vtable; }; diff --git a/src/core/lib/iomgr/endpoint_pair_posix.cc b/src/core/lib/iomgr/endpoint_pair_posix.cc index 3afbfd7254..5c5c246f99 100644 --- a/src/core/lib/iomgr/endpoint_pair_posix.cc +++ b/src/core/lib/iomgr/endpoint_pair_posix.cc @@ -59,11 +59,11 @@ grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char* name, grpc_core::ExecCtx exec_ctx; gpr_asprintf(&final_name, "%s:client", name); - p.client = grpc_tcp_create(grpc_fd_create(sv[1], final_name, true), args, + p.client = grpc_tcp_create(grpc_fd_create(sv[1], final_name, false), args, "socketpair-server"); gpr_free(final_name); gpr_asprintf(&final_name, "%s:server", name); - p.server = grpc_tcp_create(grpc_fd_create(sv[0], final_name, true), args, + p.server = grpc_tcp_create(grpc_fd_create(sv[0], final_name, false), args, "socketpair-client"); gpr_free(final_name); diff --git a/src/core/lib/iomgr/tcp_custom.cc b/src/core/lib/iomgr/tcp_custom.cc index e02a1898f2..f7a5f36cdc 100644 --- a/src/core/lib/iomgr/tcp_custom.cc +++ b/src/core/lib/iomgr/tcp_custom.cc @@ -326,6 +326,8 @@ static grpc_resource_user* endpoint_get_resource_user(grpc_endpoint* ep) { static int endpoint_get_fd(grpc_endpoint* ep) { return -1; } +static bool endpoint_can_track_err(grpc_endpoint* ep) { return false; } + static grpc_endpoint_vtable vtable = {endpoint_read, endpoint_write, endpoint_add_to_pollset, @@ -335,7 +337,8 @@ static grpc_endpoint_vtable vtable = {endpoint_read, endpoint_destroy, endpoint_get_resource_user, endpoint_get_peer, - endpoint_get_fd}; + endpoint_get_fd, + endpoint_can_track_err}; grpc_endpoint* custom_tcp_endpoint_create(grpc_custom_socket* socket, grpc_resource_quota* resource_quota, diff --git a/src/core/lib/iomgr/tcp_posix.cc b/src/core/lib/iomgr/tcp_posix.cc index 1db2790265..23cbc20910 100644 --- a/src/core/lib/iomgr/tcp_posix.cc +++ b/src/core/lib/iomgr/tcp_posix.cc @@ -375,8 +375,15 @@ static void tcp_ref(grpc_tcp* tcp) { gpr_ref(&tcp->refcount); } static void tcp_destroy(grpc_endpoint* ep) { grpc_network_status_unregister_endpoint(ep); grpc_tcp* tcp = reinterpret_cast<grpc_tcp*>(ep); + gpr_log(GPR_INFO, "tcp destroy %p %p", ep, tcp->outgoing_buffer_arg); grpc_slice_buffer_reset_and_unref_internal(&tcp->last_read_buffer); if (grpc_event_engine_can_track_errors()) { + gpr_mu_lock(&tcp->tb_mu); + grpc_core::TracedBuffer::Shutdown( + &tcp->tb_head, tcp->outgoing_buffer_arg, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("endpoint destroyed")); + gpr_mu_unlock(&tcp->tb_mu); + tcp->outgoing_buffer_arg = nullptr; gpr_atm_no_barrier_store(&tcp->stop_error_notification, true); grpc_fd_set_error(tcp->em_fd); } @@ -578,6 +585,7 @@ static bool tcp_write_with_timestamps(grpc_tcp* tcp, struct msghdr* msg, ssize_t* sent_length, grpc_error** error) { if (!tcp->socket_ts_enabled) { + gpr_log(GPR_INFO, "set timestamps"); uint32_t opt = grpc_core::kTimestampingSocketOptions; if (setsockopt(tcp->fd, SOL_SOCKET, SO_TIMESTAMPING, static_cast<void*>(&opt), sizeof(opt)) != 0) { @@ -610,6 +618,7 @@ static bool tcp_write_with_timestamps(grpc_tcp* tcp, struct msghdr* msg, *sent_length = length; /* Only save timestamps if all the bytes were taken by sendmsg. */ if (sending_length == static_cast<size_t>(length)) { + gpr_log(GPR_INFO, "tcp new entry %p %p", tcp, tcp->outgoing_buffer_arg); gpr_mu_lock(&tcp->tb_mu); grpc_core::TracedBuffer::AddNewEntry( &tcp->tb_head, static_cast<int>(tcp->bytes_counter + length), @@ -668,6 +677,7 @@ struct cmsghdr* process_timestamp(grpc_tcp* tcp, msghdr* msg, * non-linux platforms, error processing is not used/enabled currently. */ static bool process_errors(grpc_tcp* tcp) { + gpr_log(GPR_INFO, "process errors"); while (true) { struct iovec iov; iov.iov_base = nullptr; @@ -730,6 +740,8 @@ static bool process_errors(grpc_tcp* tcp) { } static void tcp_handle_error(void* arg /* grpc_tcp */, grpc_error* error) { + gpr_log(GPR_INFO, "handle error %p", arg); + GRPC_LOG_IF_ERROR("handle error", GRPC_ERROR_REF(error)); grpc_tcp* tcp = static_cast<grpc_tcp*>(arg); if (grpc_tcp_trace.enabled()) { gpr_log(GPR_INFO, "TCP:%p got_error: %s", tcp, grpc_error_string(error)); @@ -739,7 +751,8 @@ static void tcp_handle_error(void* arg /* grpc_tcp */, grpc_error* error) { static_cast<bool>(gpr_atm_acq_load(&tcp->stop_error_notification))) { /* We aren't going to register to hear on error anymore, so it is safe to * unref. */ - grpc_core::TracedBuffer::Shutdown(&tcp->tb_head, GRPC_ERROR_REF(error)); + gpr_log(GPR_INFO, "%p %d early return", error, + static_cast<bool>(gpr_atm_acq_load(&tcp->stop_error_notification))); TCP_UNREF(tcp, "error-tracking"); return; } @@ -774,6 +787,17 @@ static void tcp_handle_error(void* arg /* grpc_tcp */, grpc_error* error) { } #endif /* GRPC_LINUX_ERRQUEUE */ +void tcp_shutdown_buffer_list(grpc_tcp* tcp) { + if (tcp->outgoing_buffer_arg) { + gpr_mu_lock(&tcp->tb_mu); + grpc_core::TracedBuffer::Shutdown( + &tcp->tb_head, tcp->outgoing_buffer_arg, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("endpoint destroyed")); + gpr_mu_unlock(&tcp->tb_mu); + tcp->outgoing_buffer_arg = nullptr; + } +} + /* returns true if done, false if pending; if returning true, *error is set */ #if defined(IOV_MAX) && IOV_MAX < 1000 #define MAX_WRITE_IOVEC IOV_MAX @@ -821,8 +845,11 @@ static bool tcp_flush(grpc_tcp* tcp, grpc_error** error) { msg.msg_flags = 0; if (tcp->outgoing_buffer_arg != nullptr) { if (!tcp_write_with_timestamps(tcp, &msg, sending_length, &sent_length, - error)) + error)) { + gpr_log(GPR_INFO, "something went wrong"); + tcp_shutdown_buffer_list(tcp); return true; /* something went wrong with timestamps */ + } } else { msg.msg_control = nullptr; msg.msg_controllen = 0; @@ -844,12 +871,16 @@ static bool tcp_flush(grpc_tcp* tcp, grpc_error** error) { } return false; } else if (errno == EPIPE) { + gpr_log(GPR_INFO, "something went wrong"); *error = tcp_annotate_error(GRPC_OS_ERROR(errno, "sendmsg"), tcp); grpc_slice_buffer_reset_and_unref_internal(tcp->outgoing_buffer); + tcp_shutdown_buffer_list(tcp); return true; } else { + gpr_log(GPR_INFO, "something went wrong"); *error = tcp_annotate_error(GRPC_OS_ERROR(errno, "sendmsg"), tcp); grpc_slice_buffer_reset_and_unref_internal(tcp->outgoing_buffer); + tcp_shutdown_buffer_list(tcp); return true; } } @@ -910,6 +941,7 @@ static void tcp_handle_write(void* arg /* grpc_tcp */, grpc_error* error) { static void tcp_write(grpc_endpoint* ep, grpc_slice_buffer* buf, grpc_closure* cb, void* arg) { GPR_TIMER_SCOPE("tcp_write", 0); + gpr_log(GPR_INFO, "tcp_write %p %p", ep, arg); grpc_tcp* tcp = reinterpret_cast<grpc_tcp*>(ep); grpc_error* error = GRPC_ERROR_NONE; @@ -926,17 +958,18 @@ static void tcp_write(grpc_endpoint* ep, grpc_slice_buffer* buf, GPR_ASSERT(tcp->write_cb == nullptr); + tcp->outgoing_buffer_arg = arg; if (buf->length == 0) { GRPC_CLOSURE_SCHED( cb, grpc_fd_is_shutdown(tcp->em_fd) ? tcp_annotate_error( GRPC_ERROR_CREATE_FROM_STATIC_STRING("EOF"), tcp) : GRPC_ERROR_NONE); + tcp_shutdown_buffer_list(tcp); return; } tcp->outgoing_buffer = buf; tcp->outgoing_byte_idx = 0; - tcp->outgoing_buffer_arg = arg; if (arg) { GPR_ASSERT(grpc_event_engine_can_track_errors()); } @@ -949,6 +982,7 @@ static void tcp_write(grpc_endpoint* ep, grpc_slice_buffer* buf, } notify_on_write(tcp); } else { + gpr_log(GPR_INFO, "imm sched"); if (grpc_tcp_trace.enabled()) { const char* str = grpc_error_string(error); gpr_log(GPR_INFO, "write: %s", str); @@ -989,6 +1023,23 @@ static grpc_resource_user* tcp_get_resource_user(grpc_endpoint* ep) { return tcp->resource_user; } +static bool tcp_can_track_err(grpc_endpoint* ep) { + grpc_tcp* tcp = reinterpret_cast<grpc_tcp*>(ep); + if (!grpc_event_engine_can_track_errors()) { + return false; + } + struct sockaddr addr; + socklen_t len = sizeof(addr); + if (getsockname(tcp->fd, &addr, &len) < 0) { + gpr_log(GPR_ERROR, "getsockname"); + return false; + } + if (addr.sa_family == AF_INET || addr.sa_family == AF_INET6) { + return true; + } + return false; +} + static const grpc_endpoint_vtable vtable = {tcp_read, tcp_write, tcp_add_to_pollset, @@ -998,7 +1049,8 @@ static const grpc_endpoint_vtable vtable = {tcp_read, tcp_destroy, tcp_get_resource_user, tcp_get_peer, - tcp_get_fd}; + tcp_get_fd, + tcp_can_track_err}; #define MAX_CHUNK_SIZE 32 * 1024 * 1024 @@ -1059,6 +1111,7 @@ grpc_endpoint* grpc_tcp_create(grpc_fd* em_fd, tcp->is_first_read = true; tcp->bytes_counter = -1; tcp->socket_ts_enabled = false; + tcp->outgoing_buffer_arg = nullptr; /* paired with unref in grpc_tcp_destroy */ gpr_ref_init(&tcp->refcount, 1); gpr_atm_no_barrier_store(&tcp->shutdown_count, 0); @@ -1097,12 +1150,19 @@ void grpc_tcp_destroy_and_release_fd(grpc_endpoint* ep, int* fd, grpc_closure* done) { grpc_network_status_unregister_endpoint(ep); grpc_tcp* tcp = reinterpret_cast<grpc_tcp*>(ep); + gpr_log(GPR_INFO, "destroy and release %p %p", ep, tcp->outgoing_buffer_arg); GPR_ASSERT(ep->vtable == &vtable); tcp->release_fd = fd; tcp->release_fd_cb = done; grpc_slice_buffer_reset_and_unref_internal(&tcp->last_read_buffer); if (grpc_event_engine_can_track_errors()) { /* Stop errors notification. */ + gpr_mu_lock(&tcp->tb_mu); + grpc_core::TracedBuffer::Shutdown( + &tcp->tb_head, tcp->outgoing_buffer_arg, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("endpoint destroyed")); + gpr_mu_unlock(&tcp->tb_mu); + tcp->outgoing_buffer_arg = nullptr; gpr_atm_no_barrier_store(&tcp->stop_error_notification, true); grpc_fd_set_error(tcp->em_fd); } diff --git a/src/core/lib/security/transport/secure_endpoint.cc b/src/core/lib/security/transport/secure_endpoint.cc index f40f969bb7..9b103834e1 100644 --- a/src/core/lib/security/transport/secure_endpoint.cc +++ b/src/core/lib/security/transport/secure_endpoint.cc @@ -389,6 +389,11 @@ static grpc_resource_user* endpoint_get_resource_user( return grpc_endpoint_get_resource_user(ep->wrapped_ep); } +static bool endpoint_can_track_err(grpc_endpoint* secure_ep) { + secure_endpoint* ep = reinterpret_cast<secure_endpoint*>(secure_ep); + return grpc_endpoint_can_track_err(ep->wrapped_ep); +} + static const grpc_endpoint_vtable vtable = {endpoint_read, endpoint_write, endpoint_add_to_pollset, @@ -398,7 +403,8 @@ static const grpc_endpoint_vtable vtable = {endpoint_read, endpoint_destroy, endpoint_get_resource_user, endpoint_get_peer, - endpoint_get_fd}; + endpoint_get_fd, + endpoint_can_track_err}; grpc_endpoint* grpc_secure_endpoint_create( struct tsi_frame_protector* protector, |