diff --git a/include/grpc/impl/codegen/log.h b/include/grpc/impl/codegen/log.h
index aa86fc4c17938fb4b45a92ecb5ac4831f3b18a5d..c3f30a9147978fc3288d85731bd304a6f332e626 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 <grpc/support/log_win32.h>
+#endif
+
/* GPR log API.
Usage (within grpc):
diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c
index e75a643a542a8903bb33e2837807481c3308b9e3..1cb4f48e606d6105128617b60db5e41047d1cf05 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 2d736f7b0c02f8809c952d6f4491da0a6da6be61..64849ae01108b6074ffb07eab7b7063ec6c9ea20 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 d46558ab1b1c7255278337ba3427ec491d1c031f..47bafd9e85aa4f36d4a2af195d03127cb80724f6 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 bff5c586f8596704def8629ccc1dc19e009f8bb3..eecb4ac3ad221a9f173a781a7cf38647a5725c78 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 914736234da874cb6e11c6acbcdf00260ac95097..92bd30ca936b7b2e20f2c44c099e135e4aed25f7 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 66f9ff7a4657aa408980934dc341dfebce6179ec..2d28eee6f6336840409a734a64aed364a6d8594b 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 125f521d87e556b50a400d0cfdc943221d97242c..a98fedc8120e820eda64184a431fd44c13f54a49 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 551149e1a62cd4d6b324811aa6648c8f6d563634..f46b353f5a08885c7faabe6bc964abed4d43e3cc 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 a2f58eddd53b8fe92fbfac9ea57d64bb6a1855b6..86d777235ef0acbb824bc65730d0e0ea860b0c95 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 d30952789f55da0ffa33f337ac27fd99ebdf7ac8..250334b4af8298b8df5d4837257254f6e0a79393 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; }