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

Windows port of error system

Browse Source
pull/6897/head
Craig Tiller 9 years ago
parent 012f8f4c53
commit a41ac571ab
11 changed files with 227 additions and 208 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      include/grpc/impl/codegen/log.h
  2. 19
      src/core/lib/iomgr/error.c
  3. 5
      src/core/lib/iomgr/error.h
  4. 4
      src/core/lib/iomgr/iocp_windows.c
  5. 10
      src/core/lib/iomgr/pollset_windows.c
  6. 67
      src/core/lib/iomgr/resolve_address_windows.c
  7. 47
      src/core/lib/iomgr/tcp_client_windows.c
  8. 152
      src/core/lib/iomgr/tcp_server_windows.c
  9. 76
      src/core/lib/iomgr/tcp_windows.c
  10. 2
      src/core/lib/iomgr/tcp_windows.h
  11. 5
      src/core/lib/iomgr/unix_sockets_posix_noop.c

4
include/grpc/impl/codegen/log.h
Unescape View File

@ -43,6 +43,10 @@
extern "C" {
#endif
#ifdef GPR_WIN32
#include <grpc/support/log_win32.h>
#endif
/* GPR log API.
Usage (within grpc):

19
src/core/lib/iomgr/error.c
Unescape View File

@ -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) {

5
src/core/lib/iomgr/error.h
Unescape View File

@ -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);

4
src/core/lib/iomgr/iocp_windows.c
Unescape View File

@ -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;
}

10
src/core/lib/iomgr/pollset_windows.c
Unescape View File

@ -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(); }

67
src/core/lib/iomgr/resolve_address_windows.c
Unescape View File

@ -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

47
src/core/lib/iomgr/tcp_client_windows.c
Unescape View File

@ -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 */

152
src/core/lib/iomgr/tcp_server_windows.c
Unescape View File

@ -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,8 +409,12 @@ 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) {
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));
@ -438,13 +435,13 @@ static grpc_tcp_listener *add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
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);

76
src/core/lib/iomgr/tcp_windows.c
Unescape View File

@ -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, &param, 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 *)&param,
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);
error = GRPC_ERROR_CREATE(utf8_message);
gpr_free(utf8_message);
}
success = 0;
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;
}
}

2
src/core/lib/iomgr/tcp_windows.h
Unescape View File

@ -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 */

5
src/core/lib/iomgr/unix_sockets_posix_noop.c
Unescape View File

@ -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; }

Write Preview
Loading…
Cancel
Save