From a41ac571ab5a7a387683d29e793be3902f92aff5 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 17 May 2016 16:08:17 -0700 Subject: [PATCH] Windows port of error system --- include/grpc/impl/codegen/log.h | 4 + src/core/lib/iomgr/error.c | 19 ++ src/core/lib/iomgr/error.h | 5 + src/core/lib/iomgr/iocp_windows.c | 4 +- src/core/lib/iomgr/pollset_windows.c | 10 +- src/core/lib/iomgr/resolve_address_windows.c | 67 ++++--- src/core/lib/iomgr/tcp_client_windows.c | 47 +++-- src/core/lib/iomgr/tcp_server_windows.c | 190 +++++++++---------- src/core/lib/iomgr/tcp_windows.c | 82 ++++---- src/core/lib/iomgr/tcp_windows.h | 2 +- src/core/lib/iomgr/unix_sockets_posix_noop.c | 5 +- 11 files changed, 227 insertions(+), 208 deletions(-) diff --git a/include/grpc/impl/codegen/log.h b/include/grpc/impl/codegen/log.h index aa86fc4c179..c3f30a91479 100644 --- a/include/grpc/impl/codegen/log.h +++ b/include/grpc/impl/codegen/log.h @@ -43,6 +43,10 @@ extern "C" { #endif +#ifdef GPR_WIN32 +#include +#endif + /* GPR log API. Usage (within grpc): diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c index e75a643a542..1cb4f48e606 100644 --- a/src/core/lib/iomgr/error.c +++ b/src/core/lib/iomgr/error.c @@ -110,6 +110,8 @@ static const char *error_int_name(grpc_error_ints key) { return "security_status"; case GRPC_ERROR_INT_FD: return "fd"; + case GRPC_ERROR_INT_WSA_ERROR: + return "wsa_error"; } GPR_UNREACHABLE_CODE(return "unknown"); } @@ -492,6 +494,7 @@ const char *grpc_error_string(grpc_error *err) { return finish_kvs(&kvs); } +#ifdef GPR_POSIX_SOCKET grpc_error *grpc_os_error(const char *file, int line, int err, const char *call_name) { return grpc_error_set_str( @@ -501,6 +504,22 @@ grpc_error *grpc_os_error(const char *file, int line, int err, GRPC_ERROR_STR_OS_ERROR, strerror(err)), GRPC_ERROR_STR_SYSCALL, call_name); } +#endif + +#ifdef GPR_WIN32 +grpc_error *grpc_wsa_error(const char *file, int line, int err, + const char *call_name) { + char *utf8_message = gpr_format_message(err); + grpc_error *error = grpc_error_set_str( + grpc_error_set_str( + grpc_error_set_int(grpc_error_create(file, line, "OS Error", NULL, 0), + GRPC_ERROR_INT_WSA_ERROR, err), + GRPC_ERROR_STR_OS_ERROR, utf8_message), + GRPC_ERROR_STR_SYSCALL, call_name); + gpr_free(utf8_message); + return error; +} +#endif bool grpc_log_if_error(const char *what, grpc_error *error, const char *file, int line) { diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h index 2d736f7b0c0..64849ae0110 100644 --- a/src/core/lib/iomgr/error.h +++ b/src/core/lib/iomgr/error.h @@ -65,6 +65,7 @@ typedef enum { GRPC_ERROR_INT_HTTP2_ERROR, GRPC_ERROR_INT_TSI_CODE, GRPC_ERROR_INT_SECURITY_STATUS, + GRPC_ERROR_INT_WSA_ERROR, GRPC_ERROR_INT_FD, } grpc_error_ints; @@ -128,6 +129,10 @@ grpc_error *grpc_os_error(const char *file, int line, int err, const char *call_name); #define GRPC_OS_ERROR(err, call_name) \ grpc_os_error(__FILE__, __LINE__, err, call_name) +grpc_error *grpc_wsa_error(const char *file, int line, int err, + const char *call_name); +#define GRPC_WSA_ERROR(err, call_name) \ + grpc_wsa_error(__FILE__, __LINE__, err, call_name) bool grpc_log_if_error(const char *what, grpc_error *error, const char *file, int line); diff --git a/src/core/lib/iomgr/iocp_windows.c b/src/core/lib/iomgr/iocp_windows.c index d46558ab1b1..47bafd9e85a 100644 --- a/src/core/lib/iomgr/iocp_windows.c +++ b/src/core/lib/iomgr/iocp_windows.c @@ -121,7 +121,7 @@ grpc_iocp_work_status grpc_iocp_work(grpc_exec_ctx *exec_ctx, info->has_pending_iocp = 1; } gpr_mu_unlock(&socket->state_mu); - grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL); + grpc_exec_ctx_push(exec_ctx, closure, GRPC_ERROR_NONE, NULL); return GRPC_IOCP_WORK_WORK; } @@ -187,7 +187,7 @@ static void socket_notify_on_iocp(grpc_exec_ctx *exec_ctx, gpr_mu_lock(&socket->state_mu); if (info->has_pending_iocp) { info->has_pending_iocp = 0; - grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL); + grpc_exec_ctx_push(exec_ctx, closure, GRPC_ERROR_NONE, NULL); } else { info->closure = closure; } diff --git a/src/core/lib/iomgr/pollset_windows.c b/src/core/lib/iomgr/pollset_windows.c index bff5c586f85..eecb4ac3ad2 100644 --- a/src/core/lib/iomgr/pollset_windows.c +++ b/src/core/lib/iomgr/pollset_windows.c @@ -109,7 +109,7 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, pollset->shutting_down = 1; grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST); if (!pollset->is_iocp_worker) { - grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL); + grpc_exec_ctx_push(exec_ctx, closure, GRPC_ERROR_NONE, NULL); } else { pollset->on_shutdown = closure; } @@ -127,7 +127,7 @@ void grpc_pollset_reset(grpc_pollset *pollset) { pollset->on_shutdown = NULL; } -void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, +grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_pollset_worker **worker_hdl, gpr_timespec now, gpr_timespec deadline) { grpc_pollset_worker worker; @@ -167,7 +167,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, } if (pollset->shutting_down && pollset->on_shutdown != NULL) { - grpc_exec_ctx_enqueue(exec_ctx, pollset->on_shutdown, true, NULL); + grpc_exec_ctx_push(exec_ctx, pollset->on_shutdown, GRPC_ERROR_NONE, NULL); pollset->on_shutdown = NULL; } goto done; @@ -197,9 +197,10 @@ done: } gpr_cv_destroy(&worker.cv); *worker_hdl = NULL; + return GRPC_ERROR_NONE; } -void grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) { +grpc_error *grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) { if (specific_worker != NULL) { if (specific_worker == GRPC_POLLSET_KICK_BROADCAST) { for (specific_worker = @@ -233,6 +234,7 @@ void grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) { p->kicked_without_pollers = 1; } } + return GRPC_ERROR_NONE; } void grpc_kick_poller(void) { grpc_iocp_kick(); } diff --git a/src/core/lib/iomgr/resolve_address_windows.c b/src/core/lib/iomgr/resolve_address_windows.c index 914736234da..92bd30ca936 100644 --- a/src/core/lib/iomgr/resolve_address_windows.c +++ b/src/core/lib/iomgr/resolve_address_windows.c @@ -56,30 +56,36 @@ typedef struct { char *name; char *default_port; - grpc_resolve_cb cb; grpc_closure request_closure; - void *arg; + grpc_closure *on_done; + grpc_resolved_addresses **addresses; } request; -static grpc_resolved_addresses *blocking_resolve_address_impl( - const char *name, const char *default_port) { +static grpc_error *blocking_resolve_address_impl( + const char *name, const char *default_port, grpc_resolved_addresses **addresses) { struct addrinfo hints; struct addrinfo *result = NULL, *resp; char *host; char *port; int s; size_t i; - grpc_resolved_addresses *addrs = NULL; + grpc_error *error = GRPC_ERROR_NONE; /* parse name, splitting it into host and port parts */ gpr_split_host_port(name, &host, &port); if (host == NULL) { - gpr_log(GPR_ERROR, "unparseable host:port: '%s'", name); + char *msg; + gpr_asprintf(&msg, "unparseable host:port: '%s'", name); + error = GRPC_ERROR_CREATE(msg); + gpr_free(msg); goto done; } if (port == NULL) { if (default_port == NULL) { - gpr_log(GPR_ERROR, "no port in name '%s'", name); + char *msg; + gpr_asprintf(&msg, "no port in name '%s'", name); + error = GRPC_ERROR_CREATE(msg); + gpr_free(msg); goto done; } port = gpr_strdup(default_port); @@ -102,23 +108,23 @@ static grpc_resolved_addresses *blocking_resolve_address_impl( } /* Success path: set addrs non-NULL, fill it in */ - addrs = gpr_malloc(sizeof(grpc_resolved_addresses)); - addrs->naddrs = 0; + (*addresses) = gpr_malloc(sizeof(grpc_resolved_addresses)); + (*addresses)->naddrs = 0; for (resp = result; resp != NULL; resp = resp->ai_next) { - addrs->naddrs++; + (*addresses)->naddrs++; } - addrs->addrs = gpr_malloc(sizeof(grpc_resolved_address) * addrs->naddrs); + (*addresses)->addrs = gpr_malloc(sizeof(grpc_resolved_address) * (*addresses)->naddrs); i = 0; for (resp = result; resp != NULL; resp = resp->ai_next) { - memcpy(&addrs->addrs[i].addr, resp->ai_addr, resp->ai_addrlen); - addrs->addrs[i].len = resp->ai_addrlen; + memcpy(&(*addresses)->addrs[i].addr, resp->ai_addr, resp->ai_addrlen); + (*addresses)->addrs[i].len = resp->ai_addrlen; i++; } { - for (i = 0; i < addrs->naddrs; i++) { + for (i = 0; i < (*addresses)->naddrs; i++) { char *buf; - grpc_sockaddr_to_string(&buf, (struct sockaddr *)&addrs->addrs[i].addr, + grpc_sockaddr_to_string(&buf, (struct sockaddr *)&(*addresses)->addrs[i].addr, 0); gpr_free(buf); } @@ -130,23 +136,24 @@ done: if (result) { freeaddrinfo(result); } - return addrs; + return error; } -grpc_resolved_addresses *(*grpc_blocking_resolve_address)( - const char *name, const char *default_port) = blocking_resolve_address_impl; +grpc_error *(*grpc_blocking_resolve_address)( + const char *name, const char *default_port, grpc_resolved_addresses **addresses) = blocking_resolve_address_impl; /* Callback to be passed to grpc_executor to asynch-ify * grpc_blocking_resolve_address */ -static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, bool success) { +static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, grpc_error *error) { request *r = rp; - grpc_resolved_addresses *resolved = - grpc_blocking_resolve_address(r->name, r->default_port); - void *arg = r->arg; - grpc_resolve_cb cb = r->cb; + if (error == GRPC_ERROR_NONE) { + error = grpc_blocking_resolve_address(r->name, r->default_port, r->addresses); + } else { + GRPC_ERROR_REF(error); + } + grpc_exec_ctx_push(exec_ctx, r->on_done, error, NULL); gpr_free(r->name); gpr_free(r->default_port); - cb(exec_ctx, arg, resolved); gpr_free(r); } @@ -156,19 +163,17 @@ void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) { } static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name, - const char *default_port, grpc_resolve_cb cb, - void *arg) { + const char *default_port, grpc_closure *on_done, grpc_resolved_addresses **addresses) { request *r = gpr_malloc(sizeof(request)); grpc_closure_init(&r->request_closure, do_request_thread, r); r->name = gpr_strdup(name); r->default_port = gpr_strdup(default_port); - r->cb = cb; - r->arg = arg; - grpc_executor_enqueue(&r->request_closure, 1); + r->on_done = on_done; + r->addresses = addresses; + grpc_executor_push(&r->request_closure, GRPC_ERROR_NONE); } void (*grpc_resolve_address)(grpc_exec_ctx *exec_ctx, const char *name, - const char *default_port, grpc_resolve_cb cb, - void *arg) = resolve_address_impl; + const char *default_port, grpc_closure *on_done, grpc_resolved_addresses **addresses) = resolve_address_impl; #endif diff --git a/src/core/lib/iomgr/tcp_client_windows.c b/src/core/lib/iomgr/tcp_client_windows.c index 66f9ff7a465..2d28eee6f63 100644 --- a/src/core/lib/iomgr/tcp_client_windows.c +++ b/src/core/lib/iomgr/tcp_client_windows.c @@ -75,7 +75,7 @@ static void async_connect_unlock_and_cleanup(async_connect *ac, if (socket != NULL) grpc_winsocket_destroy(socket); } -static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, bool occured) { +static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) { async_connect *ac = acp; gpr_mu_lock(&ac->mu); if (ac->socket != NULL) { @@ -84,7 +84,7 @@ static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, bool occured) { async_connect_unlock_and_cleanup(ac, ac->socket); } -static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, bool from_iocp) { +static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) { async_connect *ac = acp; SOCKET sock = ac->socket->socket; grpc_endpoint **ep = ac->endpoint; @@ -92,6 +92,8 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, bool from_iocp) { grpc_winsocket_callback_info *info = &ac->socket->write_info; grpc_closure *on_done = ac->on_done; + GRPC_ERROR_REF(error); + gpr_mu_lock(&ac->mu); grpc_winsocket *socket = ac->socket; ac->socket = NULL; @@ -101,17 +103,14 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, bool from_iocp) { gpr_mu_lock(&ac->mu); - if (from_iocp && socket != NULL) { + if (error == GRPC_ERROR_NONE && socket != NULL) { DWORD transfered_bytes = 0; DWORD flags; BOOL wsa_success = WSAGetOverlappedResult(sock, &info->overlapped, &transfered_bytes, FALSE, &flags); GPR_ASSERT(transfered_bytes == 0); if (!wsa_success) { - char *utf8_message = gpr_format_message(WSAGetLastError()); - gpr_log(GPR_ERROR, "on_connect error connecting to '%s': %s", - ac->addr_name, utf8_message); - gpr_free(utf8_message); + error = GRPC_WSA_ERROR(WSAGetLastError(), "ConnectEx"); } else { *ep = grpc_tcp_create(socket, ac->addr_name); socket = NULL; @@ -121,7 +120,7 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, bool from_iocp) { async_connect_unlock_and_cleanup(ac, socket); /* If the connection was aborted, the callback was already called when the deadline was met. */ - on_done->cb(exec_ctx, on_done->cb_arg, *ep != NULL); + grpc_exec_ctx_push(exec_ctx, on_done, error, NULL); } /* Tries to issue one async connection, then schedules both an IOCP @@ -141,10 +140,8 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *on_done, LPFN_CONNECTEX ConnectEx; GUID guid = WSAID_CONNECTEX; DWORD ioctl_num_bytes; - const char *message = NULL; - char *utf8_message; grpc_winsocket_callback_info *info; - int last_error; + grpc_error *error = GRPC_ERROR_NONE; *endpoint = NULL; @@ -157,12 +154,12 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *on_done, sock = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED); if (sock == INVALID_SOCKET) { - message = "Unable to create socket: %s"; + error = GRPC_WSA_ERROR(WSAGetLastError(), "WSASocket"); goto failure; } - if (!grpc_tcp_prepare_socket(sock)) { - message = "Unable to set socket options: %s"; + error = grpc_tcp_prepare_socket(sock); + if (error != GRPC_ERROR_NONE) { goto failure; } @@ -173,7 +170,7 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *on_done, &ConnectEx, sizeof(ConnectEx), &ioctl_num_bytes, NULL, NULL); if (status != 0) { - message = "Unable to retrieve ConnectEx pointer: %s"; + error = GRPC_WSA_ERROR(WSAGetLastError(), "WSAIoctl(SIO_GET_EXTENSION_FUNCTION_POINTER)"); goto failure; } @@ -181,7 +178,7 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *on_done, status = bind(sock, (struct sockaddr *)&local_address, sizeof(local_address)); if (status != 0) { - message = "Unable to bind socket: %s"; + error = GRPC_WSA_ERROR(WSAGetLastError(), "bind"); goto failure; } @@ -193,9 +190,9 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *on_done, /* It wouldn't be unusual to get a success immediately. But we'll still get an IOCP notification, so let's ignore it. */ if (!success) { - int error = WSAGetLastError(); - if (error != ERROR_IO_PENDING) { - message = "ConnectEx failed: %s"; + int last_error = WSAGetLastError(); + if (last_error != ERROR_IO_PENDING) { + error = GRPC_WSA_ERROR(last_error, "ConnectEx"); goto failure; } } @@ -215,17 +212,17 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *on_done, return; failure: - last_error = WSAGetLastError(); - utf8_message = gpr_format_message(last_error); - gpr_log(GPR_ERROR, message, utf8_message); - gpr_log(GPR_ERROR, "last error = %d", last_error); - gpr_free(utf8_message); + GPR_ASSERT(error != GRPC_ERROR_NONE); + char *target_uri = grpc_sockaddr_to_uri(addr); + grpc_error *final_error = grpc_error_set_str(GRPC_ERROR_CREATE_REFERENCING("Failed to connect", &error, 1), + GRPC_ERROR_STR_TARGET_ADDRESS, target_uri); + GRPC_ERROR_UNREF(error); if (socket != NULL) { grpc_winsocket_destroy(socket); } else if (sock != INVALID_SOCKET) { closesocket(sock); } - grpc_exec_ctx_enqueue(exec_ctx, on_done, false, NULL); + grpc_exec_ctx_push(exec_ctx, on_done, final_error, NULL); } #endif /* GPR_WINSOCK_SOCKET */ diff --git a/src/core/lib/iomgr/tcp_server_windows.c b/src/core/lib/iomgr/tcp_server_windows.c index 125f521d87e..a98fedc8120 100644 --- a/src/core/lib/iomgr/tcp_server_windows.c +++ b/src/core/lib/iomgr/tcp_server_windows.c @@ -102,7 +102,7 @@ struct grpc_tcp_server { /* Public function. Allocates the proper data structures to hold a grpc_tcp_server. */ -grpc_tcp_server *grpc_tcp_server_create(grpc_closure *shutdown_complete) { +grpc_error *grpc_tcp_server_create(grpc_closure *shutdown_complete, grpc_tcp_server **server) { grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server)); gpr_ref_init(&s->refs, 1); gpr_mu_init(&s->mu); @@ -114,12 +114,13 @@ grpc_tcp_server *grpc_tcp_server_create(grpc_closure *shutdown_complete) { s->shutdown_starting.head = NULL; s->shutdown_starting.tail = NULL; s->shutdown_complete = shutdown_complete; - return s; + *server = s; + return GRPC_ERROR_NONE; } static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { if (s->shutdown_complete != NULL) { - grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, true, NULL); + grpc_exec_ctx_push(exec_ctx, s->shutdown_complete, GRPC_ERROR_NONE, NULL); } /* Now that the accepts have been aborted, we can destroy the sockets. @@ -143,7 +144,7 @@ grpc_tcp_server *grpc_tcp_server_ref(grpc_tcp_server *s) { void grpc_tcp_server_shutdown_starting_add(grpc_tcp_server *s, grpc_closure *shutdown_starting) { gpr_mu_lock(&s->mu); - grpc_closure_list_add(&s->shutdown_starting, shutdown_starting, 1); + grpc_closure_list_append(&s->shutdown_starting, shutdown_starting, GRPC_ERROR_NONE); gpr_mu_unlock(&s->mu); } @@ -187,51 +188,45 @@ void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { } /* Prepare (bind) a recently-created socket for listening. */ -static int prepare_socket(SOCKET sock, const struct sockaddr *addr, - size_t addr_len) { +static grpc_error *prepare_socket(SOCKET sock, const struct sockaddr *addr, + size_t addr_len, int *port) { struct sockaddr_storage sockname_temp; socklen_t sockname_len; + grpc_error *error = GRPC_ERROR_NONE; - if (sock == INVALID_SOCKET) goto error; - - if (!grpc_tcp_prepare_socket(sock)) { - char *utf8_message = gpr_format_message(WSAGetLastError()); - gpr_log(GPR_ERROR, "Unable to prepare socket: %s", utf8_message); - gpr_free(utf8_message); - goto error; + error = grpc_tcp_prepare_socket(sock); + if (error != GRPC_ERROR_NONE) { + goto failure; } if (bind(sock, addr, (int)addr_len) == SOCKET_ERROR) { - char *addr_str; - char *utf8_message = gpr_format_message(WSAGetLastError()); - grpc_sockaddr_to_string(&addr_str, addr, 0); - gpr_log(GPR_ERROR, "bind addr=%s: %s", addr_str, utf8_message); - gpr_free(utf8_message); - gpr_free(addr_str); - goto error; + error = GRPC_WSA_ERROR(WSAGetLastError(), "bind"); + goto failure; } if (listen(sock, SOMAXCONN) == SOCKET_ERROR) { - char *utf8_message = gpr_format_message(WSAGetLastError()); - gpr_log(GPR_ERROR, "listen: %s", utf8_message); - gpr_free(utf8_message); - goto error; + error = GRPC_WSA_ERROR(WSAGetLastError(), "listen"); + goto failure; } sockname_len = sizeof(sockname_temp); if (getsockname(sock, (struct sockaddr *)&sockname_temp, &sockname_len) == SOCKET_ERROR) { - char *utf8_message = gpr_format_message(WSAGetLastError()); - gpr_log(GPR_ERROR, "getsockname: %s", utf8_message); - gpr_free(utf8_message); - goto error; + error = GRPC_WSA_ERROR(WSAGetLastError(), "getsockname"); + goto failure; } - return grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); + *port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); + return GRPC_ERROR_NONE; -error: +failure: + GPR_ASSERT(error != GRPC_ERROR_NONE); + char *tgtaddr = grpc_sockaddr_to_uri(addr); + grpc_error *final_error = grpc_error_set_int( grpc_error_set_str(GRPC_ERROR_CREATE_REFERENCING("Failed to prepare server socket", &error, 1), GRPC_ERROR_STR_TARGET_ADDRESS, tgtaddr), GRPC_ERROR_INT_FD, (intptr_t)sock); + gpr_free(tgtaddr); + GRPC_ERROR_UNREF(error); if (sock != INVALID_SOCKET) closesocket(sock); - return -1; + return error; } static void decrement_active_ports_and_notify(grpc_exec_ctx *exec_ctx, @@ -251,26 +246,22 @@ static void decrement_active_ports_and_notify(grpc_exec_ctx *exec_ctx, /* In order to do an async accept, we need to create a socket first which will be the one assigned to the new incoming connection. */ -static void start_accept(grpc_exec_ctx *exec_ctx, grpc_tcp_listener *port) { +static grpc_error *start_accept(grpc_exec_ctx *exec_ctx, grpc_tcp_listener *port) { SOCKET sock = INVALID_SOCKET; - char *message; - char *utf8_message; BOOL success; DWORD addrlen = sizeof(struct sockaddr_in6) + 16; DWORD bytes_received = 0; + grpc_error *error = GRPC_ERROR_NONE; sock = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED); - if (sock == INVALID_SOCKET) { - message = "Unable to create socket: %s"; + error = GRPC_WSA_ERROR(WSAGetLastError(), "WSASocket"); goto failure; } - if (!grpc_tcp_prepare_socket(sock)) { - message = "Unable to prepare socket: %s"; - goto failure; - } + error = grpc_tcp_prepare_socket(sock); + if (error != GRPC_ERROR_NONE) goto failure; /* Start the "accept" asynchronously. */ success = port->AcceptEx(port->socket->socket, sock, port->addresses, 0, @@ -280,9 +271,9 @@ static void start_accept(grpc_exec_ctx *exec_ctx, grpc_tcp_listener *port) { /* It is possible to get an accept immediately without delay. However, we will still get an IOCP notification for it. So let's just ignore it. */ if (!success) { - int error = WSAGetLastError(); - if (error != ERROR_IO_PENDING) { - message = "AcceptEx failed: %s"; + int last_error = WSAGetLastError(); + if (last_error != ERROR_IO_PENDING) { + error = GRPC_WSA_ERROR(last_error, "AcceptEx"); goto failure; } } @@ -291,9 +282,10 @@ static void start_accept(grpc_exec_ctx *exec_ctx, grpc_tcp_listener *port) { immediately process an accept that happened in the meantime. */ port->new_socket = sock; grpc_socket_notify_on_read(exec_ctx, port->socket, &port->on_accept); - return; + return error; failure: + GPR_ASSERT(error != GRPC_ERROR_NONE); if (port->shutting_down) { /* We are abandoning the listener port, take that into account to prevent occasional hangs on shutdown. The hang happens when sp->shutting_down @@ -301,16 +293,15 @@ failure: but we fail there because the listening port has been closed in the meantime. */ decrement_active_ports_and_notify(exec_ctx, port); - return; + GRPC_ERROR_UNREF(error); + return GRPC_ERROR_NONE; } - utf8_message = gpr_format_message(WSAGetLastError()); - gpr_log(GPR_ERROR, message, utf8_message); - gpr_free(utf8_message); if (sock != INVALID_SOCKET) closesocket(sock); + return error; } /* Event manager callback when reads are ready. */ -static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, bool from_iocp) { +static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { grpc_tcp_listener *sp = arg; grpc_tcp_server_acceptor acceptor = {sp->server, sp->port_index, 0}; SOCKET sock = sp->new_socket; @@ -328,7 +319,10 @@ static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, bool from_iocp) { /* The general mechanism for shutting down is to queue abortion calls. While this is necessary in the read/write case, it's useless for the accept case. We only need to adjust the pending callback count */ - if (!from_iocp) { + if (error != GRPC_ERROR_NONE) { + const char *msg = grpc_error_string(error); + gpr_log(GPR_INFO, "Skipping on_accept due to error: %s", msg); + grpc_error_free_string(msg); return; } @@ -386,21 +380,20 @@ static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, bool from_iocp) { the former socked we created has now either been destroy or assigned to the new connection. We need to create a new one for the next connection. */ - start_accept(exec_ctx, sp); + GPR_ASSERT(GRPC_LOG_IF_ERROR("start_accept", start_accept(exec_ctx, sp))); } -static grpc_tcp_listener *add_socket_to_server(grpc_tcp_server *s, SOCKET sock, +static grpc_error *add_socket_to_server(grpc_tcp_server *s, SOCKET sock, const struct sockaddr *addr, size_t addr_len, - unsigned port_index) { + unsigned port_index, grpc_tcp_listener **listener) { grpc_tcp_listener *sp = NULL; int port; int status; GUID guid = WSAID_ACCEPTEX; DWORD ioctl_num_bytes; LPFN_ACCEPTEX AcceptEx; - - if (sock == INVALID_SOCKET) return NULL; + grpc_error *error = GRPC_ERROR_NONE; /* We need to grab the AcceptEx pointer for that port, as it may be interface-dependent. We'll cache it to avoid doing that again. */ @@ -416,35 +409,39 @@ static grpc_tcp_listener *add_socket_to_server(grpc_tcp_server *s, SOCKET sock, return NULL; } - port = prepare_socket(sock, addr, addr_len); - if (port >= 0) { - gpr_mu_lock(&s->mu); - GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server"); - sp = gpr_malloc(sizeof(grpc_tcp_listener)); - sp->next = NULL; - if (s->head == NULL) { - s->head = sp; - } else { - s->tail->next = sp; - } - s->tail = sp; - sp->server = s; - sp->socket = grpc_winsocket_create(sock, "listener"); - sp->shutting_down = 0; - sp->AcceptEx = AcceptEx; - sp->new_socket = INVALID_SOCKET; - sp->port = port; - sp->port_index = port_index; - grpc_closure_init(&sp->on_accept, on_accept, sp); - GPR_ASSERT(sp->socket); - gpr_mu_unlock(&s->mu); + error = prepare_socket(sock, addr, addr_len, &port); + if (error != GRPC_ERROR_NONE) { + return error; + } + + GPR_ASSERT(port >= 0); + gpr_mu_lock(&s->mu); + GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server"); + sp = gpr_malloc(sizeof(grpc_tcp_listener)); + sp->next = NULL; + if (s->head == NULL) { + s->head = sp; + } else { + s->tail->next = sp; } + s->tail = sp; + sp->server = s; + sp->socket = grpc_winsocket_create(sock, "listener"); + sp->shutting_down = 0; + sp->AcceptEx = AcceptEx; + sp->new_socket = INVALID_SOCKET; + sp->port = port; + sp->port_index = port_index; + grpc_closure_init(&sp->on_accept, on_accept, sp); + GPR_ASSERT(sp->socket); + gpr_mu_unlock(&s->mu); + *listener = sp; - return sp; + return GRPC_ERROR_NONE; } -int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, - size_t addr_len) { +grpc_error *grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, + size_t addr_len, int *port) { grpc_tcp_listener *sp; SOCKET sock; struct sockaddr_in6 addr6_v4mapped; @@ -452,8 +449,9 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, struct sockaddr *allocated_addr = NULL; struct sockaddr_storage sockname_temp; socklen_t sockname_len; - int port; unsigned port_index = 0; + grpc_error *error = GRPC_ERROR_NONE; + if (s->tail != NULL) { port_index = s->tail->port_index + 1; } @@ -465,11 +463,11 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, sockname_len = sizeof(sockname_temp); if (0 == getsockname(sp->socket->socket, (struct sockaddr *)&sockname_temp, &sockname_len)) { - port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); - if (port > 0) { + *port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); + if (*port > 0) { allocated_addr = gpr_malloc(addr_len); memcpy(allocated_addr, addr, addr_len); - grpc_sockaddr_set_port(allocated_addr, port); + grpc_sockaddr_set_port(allocated_addr, *port); addr = allocated_addr; break; } @@ -483,8 +481,8 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, } /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */ - if (grpc_sockaddr_is_wildcard(addr, &port)) { - grpc_sockaddr_make_wildcard6(port, &wildcard); + if (grpc_sockaddr_is_wildcard(addr, port)) { + grpc_sockaddr_make_wildcard6(*port, &wildcard); addr = (struct sockaddr *)&wildcard; addr_len = sizeof(wildcard); @@ -493,19 +491,21 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, sock = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED); if (sock == INVALID_SOCKET) { - char *utf8_message = gpr_format_message(WSAGetLastError()); - gpr_log(GPR_ERROR, "unable to create socket: %s", utf8_message); - gpr_free(utf8_message); + error = GRPC_WSA_ERROR(WSAGetLastError(), "WSASocket"); + goto done; } - sp = add_socket_to_server(s, sock, addr, addr_len, port_index); + error = add_socket_to_server(s, sock, addr, addr_len, port_index, &sp); + +done: gpr_free(allocated_addr); - if (sp) { - return sp->port; - } else { - return -1; + if (error != GRPC_ERROR_NONE) { + grpc_error *error_out = GRPC_ERROR_CREATE_REFERENCING("Failed to add port to server", &error, 1); + GRPC_ERROR_UNREF(error); + error = error_out; } + return error; } void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s, @@ -520,7 +520,7 @@ void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s, s->on_accept_cb = on_accept_cb; s->on_accept_cb_arg = on_accept_cb_arg; for (sp = s->head; sp; sp = sp->next) { - start_accept(exec_ctx, sp); + GPR_ASSERT(GRPC_LOG_IF_ERROR("start_accept", start_accept(exec_ctx, sp))); s->active_ports++; } gpr_mu_unlock(&s->mu); diff --git a/src/core/lib/iomgr/tcp_windows.c b/src/core/lib/iomgr/tcp_windows.c index 551149e1a62..f46b353f5a0 100644 --- a/src/core/lib/iomgr/tcp_windows.c +++ b/src/core/lib/iomgr/tcp_windows.c @@ -61,27 +61,30 @@ #define GRPC_FIONBIO FIONBIO #endif -static int set_non_block(SOCKET sock) { +static grpc_error * set_non_block(SOCKET sock) { int status; uint32_t param = 1; DWORD ret; status = WSAIoctl(sock, GRPC_FIONBIO, ¶m, sizeof(param), NULL, 0, &ret, NULL, NULL); - return status == 0; + return status == 0 ? GRPC_ERROR_NONE : GRPC_WSA_ERROR(WSAGetLastError(), "WSAIoctl(GRPC_FIONBIO)"); } -static int set_dualstack(SOCKET sock) { +static grpc_error * set_dualstack(SOCKET sock) { int status; unsigned long param = 0; status = setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char *)¶m, sizeof(param)); - return status == 0; + return status == 0 ? GRPC_ERROR_NONE : GRPC_WSA_ERROR(WSAGetLastError(), "setsockopt(IPV6_V6ONLY)"); } -int grpc_tcp_prepare_socket(SOCKET sock) { - if (!set_non_block(sock)) return 0; - if (!set_dualstack(sock)) return 0; - return 1; +grpc_error * grpc_tcp_prepare_socket(SOCKET sock) { + grpc_error * err; + err = set_non_block(sock); + if (err != GRPC_ERROR_NONE) return err; + err = set_dualstack(sock); + if (err != GRPC_ERROR_NONE) return err; + return GRPC_ERROR_NONE; } typedef struct grpc_tcp { @@ -148,39 +151,36 @@ static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); } #endif /* Asynchronous callback from the IOCP, or the background thread. */ -static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, bool success) { +static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) { grpc_tcp *tcp = tcpp; grpc_closure *cb = tcp->read_cb; grpc_winsocket *socket = tcp->socket; gpr_slice sub; grpc_winsocket_callback_info *info = &socket->read_info; - if (success) { + GRPC_ERROR_REF(error); + + if (error == GRPC_ERROR_NONE) { if (info->wsa_error != 0 && !tcp->shutting_down) { - if (info->wsa_error != WSAECONNRESET) { - char *utf8_message = gpr_format_message(info->wsa_error); - gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message); - gpr_free(utf8_message); - } - success = 0; + char *utf8_message = gpr_format_message(info->wsa_error); + gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message); + error = GRPC_ERROR_CREATE(utf8_message); + gpr_free(utf8_message); gpr_slice_unref(tcp->read_slice); } else { if (info->bytes_transfered != 0 && !tcp->shutting_down) { sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered); gpr_slice_buffer_add(tcp->read_slices, sub); - success = 1; } else { gpr_slice_unref(tcp->read_slice); - success = 0; + error = GRPC_ERROR_CREATE("End of TCP stream"); } } } tcp->read_cb = NULL; TCP_UNREF(tcp, "read"); - if (cb) { - cb->cb(exec_ctx, cb->cb_arg, success); - } + grpc_exec_ctx_push(exec_ctx, cb, error, NULL); } static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, @@ -194,7 +194,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, WSABUF buffer; if (tcp->shutting_down) { - grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL); + grpc_exec_ctx_push(exec_ctx, cb, GRPC_ERROR_CREATE("TCP socket is shutting down"), NULL); return; } @@ -218,7 +218,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, /* Did we get data immediately ? Yay. */ if (info->wsa_error != WSAEWOULDBLOCK) { info->bytes_transfered = bytes_read; - grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, true, NULL); + grpc_exec_ctx_push(exec_ctx, &tcp->on_read, GRPC_ERROR_NONE, NULL); return; } @@ -231,7 +231,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, int wsa_error = WSAGetLastError(); if (wsa_error != WSA_IO_PENDING) { info->wsa_error = wsa_error; - grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, false, NULL); + grpc_exec_ctx_push(exec_ctx, &tcp->on_read, GRPC_WSA_ERROR(info->wsa_error, "WSARecv"), NULL); return; } } @@ -240,32 +240,29 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, } /* Asynchronous callback from the IOCP, or the background thread. */ -static void on_write(grpc_exec_ctx *exec_ctx, void *tcpp, bool success) { +static void on_write(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) { grpc_tcp *tcp = (grpc_tcp *)tcpp; grpc_winsocket *handle = tcp->socket; grpc_winsocket_callback_info *info = &handle->write_info; grpc_closure *cb; + GRPC_ERROR_REF(error); + gpr_mu_lock(&tcp->mu); cb = tcp->write_cb; tcp->write_cb = NULL; gpr_mu_unlock(&tcp->mu); - if (success) { + if (error == GRPC_ERROR_NONE) { if (info->wsa_error != 0) { - if (info->wsa_error != WSAECONNRESET) { - char *utf8_message = gpr_format_message(info->wsa_error); - gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message); - gpr_free(utf8_message); - } - success = 0; + error = GRPC_WSA_ERROR(info->wsa_error, "WSASend"); } else { GPR_ASSERT(info->bytes_transfered == tcp->write_slices->length); } } TCP_UNREF(tcp, "write"); - cb->cb(exec_ctx, cb->cb_arg, success); + grpc_exec_ctx_push(exec_ctx, cb, error, NULL); } /* Initiates a write. */ @@ -283,7 +280,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, size_t len; if (tcp->shutting_down) { - grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL); + grpc_exec_ctx_push(exec_ctx, cb, GRPC_ERROR_CREATE("TCP socket is shutting down"), NULL); return; } @@ -311,19 +308,8 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, connection that has its send queue filled up. But if we don't, then we can avoid doing an async write operation at all. */ if (info->wsa_error != WSAEWOULDBLOCK) { - bool ok = false; - if (status == 0) { - ok = true; - GPR_ASSERT(bytes_sent == tcp->write_slices->length); - } else { - if (info->wsa_error != WSAECONNRESET) { - char *utf8_message = gpr_format_message(info->wsa_error); - gpr_log(GPR_ERROR, "WSASend error: %s", utf8_message); - gpr_free(utf8_message); - } - } - if (allocated) gpr_free(allocated); - grpc_exec_ctx_enqueue(exec_ctx, cb, ok, NULL); + grpc_error *error = status == 0 ? GRPC_ERROR_NONE : GRPC_WSA_ERROR(info->wsa_error, "WSASend"); + grpc_exec_ctx_push(exec_ctx, cb, error, NULL); return; } @@ -340,7 +326,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, int wsa_error = WSAGetLastError(); if (wsa_error != WSA_IO_PENDING) { TCP_UNREF(tcp, "write"); - grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL); + grpc_exec_ctx_push(exec_ctx, cb, GRPC_WSA_ERROR(wsa_error, "WSASend"), NULL); return; } } diff --git a/src/core/lib/iomgr/tcp_windows.h b/src/core/lib/iomgr/tcp_windows.h index a2f58eddd53..86d777235ef 100644 --- a/src/core/lib/iomgr/tcp_windows.h +++ b/src/core/lib/iomgr/tcp_windows.h @@ -52,6 +52,6 @@ */ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string); -int grpc_tcp_prepare_socket(SOCKET sock); +grpc_error *grpc_tcp_prepare_socket(SOCKET sock); #endif /* GRPC_CORE_LIB_IOMGR_TCP_WINDOWS_H */ diff --git a/src/core/lib/iomgr/unix_sockets_posix_noop.c b/src/core/lib/iomgr/unix_sockets_posix_noop.c index d30952789f5..250334b4af8 100644 --- a/src/core/lib/iomgr/unix_sockets_posix_noop.c +++ b/src/core/lib/iomgr/unix_sockets_posix_noop.c @@ -44,8 +44,9 @@ void grpc_create_socketpair_if_unix(int sv[2]) { GPR_ASSERT(0); } -grpc_resolved_addresses *grpc_resolve_unix_domain_address(const char *name) { - return NULL; +grpc_error *grpc_resolve_unix_domain_address(const char *name, grpc_resolved_addresses **addresses) { + *addresses = NULL; + return GRPC_ERROR_CREATE("Unix domain sockets are not supported on Windows"); } int grpc_is_unix_socket(const struct sockaddr *addr) { return false; }