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

Change udp_server to use linked list of ports

Browse Source
pull/7615/head
Robbie Shade 8 years ago
parent 2507fef737
commit 128d3ff361
1 changed files with 52 additions and 38 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. 90
      src/core/lib/iomgr/udp_server.c

90
src/core/lib/iomgr/udp_server.c
Unescape View File

@ -67,10 +67,9 @@
#include "src/core/lib/iomgr/socket_utils_posix.h"
#include "src/core/lib/support/string.h"
#define INIT_PORT_CAP 2
/* one listening port */
typedef struct {
typedef struct grpc_udp_listener grpc_udp_listener;
struct grpc_udp_listener {
int fd;
grpc_fd *emfd;
grpc_udp_server *server;
@ -83,7 +82,9 @@ typedef struct {
grpc_closure destroyed_closure;
grpc_udp_server_read_cb read_cb;
grpc_udp_server_orphan_cb orphan_cb;
} server_port;
struct grpc_udp_listener *next;
};
/* the overall server */
struct grpc_udp_server {
@ -98,10 +99,10 @@ struct grpc_udp_server {
/* is this server shutting down? (boolean) */
int shutdown;
/* all listening ports */
server_port *ports;
size_t nports;
size_t port_capacity;
/* linked list of server ports */
grpc_udp_listener *head;
grpc_udp_listener *tail;
unsigned nports;
/* shutdown callback */
grpc_closure *shutdown_complete;
@ -121,9 +122,9 @@ grpc_udp_server *grpc_udp_server_create(void) {
s->active_ports = 0;
s->destroyed_ports = 0;
s->shutdown = 0;
s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP);
s->head = NULL;
s->tail = NULL;
s->nports = 0;
s->port_capacity = INIT_PORT_CAP;
return s;
}
@ -131,10 +132,16 @@ grpc_udp_server *grpc_udp_server_create(void) {
static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
grpc_exec_ctx_sched(exec_ctx, s->shutdown_complete, GRPC_ERROR_NONE, NULL);
gpr_mu_destroy(&s->mu);
gpr_cv_destroy(&s->cv);
gpr_mu_destroy(&s->mu);
while (s->head) {
grpc_udp_listener *sp = s->head;
s->head = sp->next;
gpr_free(sp);
}
gpr_free(s->ports);
gpr_free(s);
}
@ -165,9 +172,9 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
return;
}
if (s->nports) {
for (i = 0; i < s->nports; i++) {
server_port *sp = &s->ports[i];
if (s->head) {
grpc_udp_listener *sp;
for (sp = s->head; sp; sp = sp->next) {
sp->destroyed_closure.cb = destroyed_port;
sp->destroyed_closure.cb_arg = s;
@ -187,6 +194,7 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
grpc_closure *on_done) {
size_t i;
grpc_udp_listener* sp;
gpr_mu_lock(&s->mu);
GPR_ASSERT(!s->shutdown);
@ -196,8 +204,10 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
/* shutdown all fd's */
if (s->active_ports) {
for (i = 0; i < s->nports; i++) {
grpc_fd_shutdown(exec_ctx, s->ports[i].emfd);
for (sp = s->head; sp; sp = sp->next) {
GPR_ASSERT(sp->orphan_cb);
sp->orphan_cb(sp->emfd);
grpc_fd_shutdown(exec_ctx, sp->emfd);
}
gpr_mu_unlock(&s->mu);
} else {
@ -274,10 +284,9 @@ error:
/* event manager callback when reads are ready */
static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
server_port *sp = arg;
grpc_udp_listener *sp = arg;
if (error != GRPC_ERROR_NONE) {
gpr_mu_lock(&sp->server->mu);
if (0 == --sp->server->active_ports) {
gpr_mu_unlock(&sp->server->mu);
deactivated_all_ports(exec_ctx, sp->server);
@ -299,7 +308,7 @@ static int add_socket_to_server(grpc_udp_server *s, int fd,
const struct sockaddr *addr, size_t addr_len,
grpc_udp_server_read_cb read_cb,
grpc_udp_server_orphan_cb orphan_cb) {
server_port *sp;
grpc_udp_listener *sp;
int port;
char *addr_str;
char *name;
@ -310,12 +319,15 @@ static int add_socket_to_server(grpc_udp_server *s, int fd,
gpr_asprintf(&name, "udp-server-listener:%s", addr_str);
gpr_free(addr_str);
gpr_mu_lock(&s->mu);
/* append it to the list under a lock */
if (s->nports == s->port_capacity) {
s->port_capacity *= 2;
s->ports = gpr_realloc(s->ports, sizeof(server_port) * s->port_capacity);
s->nports++;
sp = gpr_malloc(sizeof(grpc_udp_listener));
sp->next = NULL;
if (s->head == NULL) {
s->head = sp;
} else {
s->tail->next = sp;
}
sp = &s->ports[s->nports++];
s->tail = sp;
sp->server = s;
sp->fd = fd;
sp->emfd = grpc_fd_create(fd, name);
@ -334,6 +346,7 @@ static int add_socket_to_server(grpc_udp_server *s, int fd,
int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr,
size_t addr_len, grpc_udp_server_read_cb read_cb,
grpc_udp_server_orphan_cb orphan_cb) {
grpc_udp_listener* sp;
int allocated_port1 = -1;
int allocated_port2 = -1;
unsigned i;
@ -351,9 +364,9 @@ int grpc_udp_server_add_port(grpc_udp_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].fd, (struct sockaddr *)&sockname_temp,
if (0 == getsockname(sp->fd, (struct sockaddr *)&sockname_temp,
&sockname_len)) {
port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp);
if (port > 0) {
@ -413,28 +426,29 @@ done:
return allocated_port1 >= 0 ? allocated_port1 : allocated_port2;
}
int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index) {
return (port_index < s->nports) ? s->ports[port_index].fd : -1;
}
void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
grpc_pollset **pollsets, size_t pollset_count,
grpc_server *server) {
size_t i, j;
gpr_mu_lock(&s->mu);
grpc_udp_listener *sp;
GPR_ASSERT(s->active_ports == 0);
s->pollsets = pollsets;
s->grpc_server = server;
for (i = 0; i < s->nports; i++) {
for (j = 0; j < pollset_count; j++) {
grpc_pollset_add_fd(exec_ctx, pollsets[j], s->ports[i].emfd);
sp = s->head;
while (sp != NULL) {
for (i = 0; i < pollset_count; i++) {
grpc_pollset_add_fd(exec_ctx, pollsets[i], sp->emfd);
}
s->ports[i].read_closure.cb = on_read;
s->ports[i].read_closure.cb_arg = &s->ports[i];
grpc_fd_notify_on_read(exec_ctx, s->ports[i].emfd,
&s->ports[i].read_closure);
sp->read_closure.cb = on_read;
sp->read_closure.cb_arg = sp;
grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure);
s->active_ports++;
sp = sp->next;
}
gpr_mu_unlock(&s->mu);
}

Write Preview
Loading…
Cancel
Save