aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/iomgr/tcp_server_windows.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/iomgr/tcp_server_windows.c')
-rw-r--r--src/core/iomgr/tcp_server_windows.c121
1 files changed, 71 insertions, 50 deletions
diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c
index 3fea8b5b35..e4a1d7f498 100644
--- a/src/core/iomgr/tcp_server_windows.c
+++ b/src/core/iomgr/tcp_server_windows.c
@@ -35,7 +35,8 @@
#ifdef GPR_WINSOCK_SOCKET
-#define _GNU_SOURCE
+#include <io.h>
+
#include "src/core/iomgr/sockaddr_utils.h"
#include <grpc/support/alloc.h>
@@ -51,25 +52,29 @@
#include "src/core/iomgr/tcp_server.h"
#include "src/core/iomgr/tcp_windows.h"
-#define INIT_PORT_CAP 2
#define MIN_SAFE_ACCEPT_QUEUE_SIZE 100
/* one listening port */
-typedef struct server_port {
+struct grpc_tcp_listener {
/* This seemingly magic number comes from AcceptEx's documentation. each
address buffer needs to have at least 16 more bytes at their end. */
gpr_uint8 addresses[(sizeof(struct sockaddr_in6) + 16) * 2];
/* This will hold the socket for the next accept. */
SOCKET new_socket;
- /* The listener winsocked. */
+ /* The listener winsocket. */
grpc_winsocket *socket;
+ /* The actual TCP port number. */
+ int port;
grpc_tcp_server *server;
/* The cached AcceptEx for that port. */
LPFN_ACCEPTEX AcceptEx;
int shutting_down;
/* closure for socket notification of accept being ready */
grpc_closure on_accept;
-} server_port;
+ gpr_refcount refs;
+ /* linked list */
+ struct grpc_tcp_listener *next;
+};
/* the overall server */
struct grpc_tcp_server {
@@ -82,10 +87,8 @@ struct grpc_tcp_server {
/* active port count: how many ports are actually still listening */
int active_ports;
- /* all listening ports */
- server_port *ports;
- size_t nports;
- size_t port_capacity;
+ /* linked list of server ports */
+ grpc_tcp_listener *head;
/* shutdown callback */
grpc_closure *shutdown_complete;
@@ -99,9 +102,7 @@ grpc_tcp_server *grpc_tcp_server_create(void) {
s->active_ports = 0;
s->on_accept_cb = NULL;
s->on_accept_cb_arg = NULL;
- s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP);
- s->nports = 0;
- s->port_capacity = INIT_PORT_CAP;
+ s->head = NULL;
s->shutdown_complete = NULL;
return s;
}
@@ -109,26 +110,26 @@ grpc_tcp_server *grpc_tcp_server_create(void) {
static void dont_care_about_shutdown_completion(void *arg) {}
static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
- size_t i;
-
grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, 1);
/* Now that the accepts have been aborted, we can destroy the sockets.
The IOCP won't get notified on these, so we can flag them as already
closed by the system. */
- for (i = 0; i < s->nports; i++) {
- server_port *sp = &s->ports[i];
+ while (s->head) {
+ grpc_tcp_listener *sp = s->head;
+ s->head = sp->next;
+ sp->next = NULL;
grpc_winsocket_destroy(sp->socket);
+ grpc_tcp_listener_unref(sp);
}
- gpr_free(s->ports);
gpr_free(s);
}
/* Public function. Stops and destroys a grpc_tcp_server. */
void grpc_tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s,
grpc_closure *shutdown_complete) {
- size_t i;
int immediately_done = 0;
+ grpc_tcp_listener *sp;
gpr_mu_lock(&s->mu);
s->shutdown_complete = shutdown_complete;
@@ -138,8 +139,7 @@ void grpc_tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s,
if (s->active_ports == 0) {
immediately_done = 1;
}
- for (i = 0; i < s->nports; i++) {
- server_port *sp = &s->ports[i];
+ for (sp = s->head; sp; sp = sp->next) {
sp->shutting_down = 1;
grpc_winsocket_shutdown(sp->socket);
}
@@ -199,7 +199,7 @@ error:
}
static void decrement_active_ports_and_notify(grpc_exec_ctx *exec_ctx,
- server_port *sp) {
+ grpc_tcp_listener *sp) {
int notify = 0;
sp->shutting_down = 0;
gpr_mu_lock(&sp->server->mu);
@@ -216,7 +216,7 @@ 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, server_port *port) {
+static void start_accept(grpc_exec_ctx *exec_ctx, grpc_tcp_listener *port) {
SOCKET sock = INVALID_SOCKET;
char *message;
char *utf8_message;
@@ -276,7 +276,7 @@ failure:
/* Event manager callback when reads are ready. */
static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, int from_iocp) {
- server_port *sp = arg;
+ grpc_tcp_listener *sp = arg;
SOCKET sock = sp->new_socket;
grpc_winsocket_callback_info *info = &sp->socket->read_info;
grpc_endpoint *ep = NULL;
@@ -351,16 +351,17 @@ static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, int from_iocp) {
start_accept(exec_ctx, sp);
}
-static int add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
- const struct sockaddr *addr, size_t addr_len) {
- server_port *sp;
+static grpc_tcp_listener *add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
+ const struct sockaddr *addr,
+ size_t addr_len) {
+ 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 -1;
+ if (sock == INVALID_SOCKET) return NULL;
/* 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. */
@@ -373,37 +374,35 @@ static int add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
gpr_log(GPR_ERROR, "on_connect error: %s", utf8_message);
gpr_free(utf8_message);
closesocket(sock);
- return -1;
+ 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");
- /* append it to the list under a lock */
- if (s->nports == s->port_capacity) {
- /* too many ports, and we need to store their address in a closure */
- /* TODO(ctiller): make server_port a linked list */
- abort();
- }
- sp = &s->ports[s->nports++];
+ sp = gpr_malloc(sizeof(grpc_tcp_listener));
+ sp->next = s->head;
+ s->head = 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;
+ gpr_ref_init(&sp->refs, 1);
grpc_closure_init(&sp->on_accept, on_accept, sp);
GPR_ASSERT(sp->socket);
gpr_mu_unlock(&s->mu);
}
- return port;
+ return sp;
}
-int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr,
- size_t addr_len) {
- int allocated_port = -1;
- unsigned i;
+grpc_tcp_listener *grpc_tcp_server_add_port(grpc_tcp_server *s,
+ const void *addr,
+ size_t addr_len) {
+ grpc_tcp_listener *sp;
SOCKET sock;
struct sockaddr_in6 addr6_v4mapped;
struct sockaddr_in6 wildcard;
@@ -415,9 +414,9 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr,
/* Check if this is a wildcard port, and if so, try to keep the port the same
as some previously created listener. */
if (grpc_sockaddr_get_port(addr) == 0) {
- for (i = 0; i < s->nports; i++) {
+ for (sp = s->head; sp; sp = sp->next) {
sockname_len = sizeof(sockname_temp);
- if (0 == getsockname(s->ports[i].socket->socket,
+ if (0 == getsockname(sp->socket->socket,
(struct sockaddr *)&sockname_temp, &sockname_len)) {
port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp);
if (port > 0) {
@@ -452,33 +451,55 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr,
gpr_free(utf8_message);
}
- allocated_port = add_socket_to_server(s, sock, addr, addr_len);
+ sp = add_socket_to_server(s, sock, addr, addr_len);
gpr_free(allocated_addr);
- return allocated_port;
+ return sp;
}
-SOCKET
-grpc_tcp_server_get_socket(grpc_tcp_server *s, unsigned index) {
- return (index < s->nports) ? s->ports[index].socket->socket : INVALID_SOCKET;
+int grpc_tcp_server_get_fd(grpc_tcp_server *s, unsigned port_index) {
+ grpc_tcp_listener *sp;
+ for (sp = s->head; sp && port_index != 0; sp = sp->next, port_index--);
+ if (port_index == 0 && sp) {
+ return _open_osfhandle(sp->socket->socket, 0);
+ } else {
+ return -1;
+ }
}
void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s,
grpc_pollset **pollset, size_t pollset_count,
grpc_tcp_server_cb on_accept_cb,
void *on_accept_cb_arg) {
- size_t i;
+ grpc_tcp_listener *sp;
GPR_ASSERT(on_accept_cb);
gpr_mu_lock(&s->mu);
GPR_ASSERT(!s->on_accept_cb);
GPR_ASSERT(s->active_ports == 0);
s->on_accept_cb = on_accept_cb;
s->on_accept_cb_arg = on_accept_cb_arg;
- for (i = 0; i < s->nports; i++) {
- start_accept(exec_ctx, s->ports + i);
+ for (sp = s->head; sp; sp = sp->next) {
+ start_accept(exec_ctx, sp);
s->active_ports++;
}
gpr_mu_unlock(&s->mu);
}
+int grpc_tcp_listener_get_port(grpc_tcp_listener *listener) {
+ grpc_tcp_listener *sp = listener;
+ return sp->port;
+}
+
+void grpc_tcp_listener_ref(grpc_tcp_listener *listener) {
+ grpc_tcp_listener *sp = listener;
+ gpr_ref(&sp->refs);
+}
+
+void grpc_tcp_listener_unref(grpc_tcp_listener *listener) {
+ grpc_tcp_listener *sp = listener;
+ if (gpr_unref(&sp->refs)) {
+ gpr_free(listener);
+ }
+}
+
#endif /* GPR_WINSOCK_SOCKET */
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-06 08:07:50 +0000