Call list progress
diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c
index 213b2e1..0c5e005 100644
--- a/src/core/iomgr/tcp_server_posix.c
+++ b/src/core/iomgr/tcp_server_posix.c
@@ -117,16 +117,12 @@
size_t port_capacity;
/* shutdown callback */
- void (*shutdown_complete)(void *);
- void *shutdown_complete_arg;
+ grpc_closure *shutdown_complete;
/* all pollsets interested in new connections */
grpc_pollset **pollsets;
/* number of pollsets in the pollsets array */
size_t pollset_count;
-
- /** workqueue for interally created async work */
- grpc_workqueue *workqueue;
};
grpc_tcp_server *grpc_tcp_server_create(void) {
@@ -140,40 +136,37 @@
s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP);
s->nports = 0;
s->port_capacity = INIT_PORT_CAP;
- s->workqueue = grpc_workqueue_create();
return s;
}
-static void finish_shutdown(grpc_tcp_server *s) {
- s->shutdown_complete(s->shutdown_complete_arg);
- s->shutdown_complete = NULL;
+static void finish_shutdown(grpc_tcp_server *s, grpc_call_list *call_list) {
+ grpc_call_list_add(call_list, s->shutdown_complete, 1);
gpr_mu_destroy(&s->mu);
gpr_free(s->ports);
- GRPC_WORKQUEUE_UNREF(s->workqueue, "destroy");
gpr_free(s);
}
-static void destroyed_port(void *server, int success) {
+static void destroyed_port(void *server, int success,
+ grpc_call_list *call_list) {
grpc_tcp_server *s = server;
gpr_mu_lock(&s->mu);
s->destroyed_ports++;
if (s->destroyed_ports == s->nports) {
gpr_mu_unlock(&s->mu);
- finish_shutdown(s);
+ finish_shutdown(s, call_list);
} else {
GPR_ASSERT(s->destroyed_ports < s->nports);
gpr_mu_unlock(&s->mu);
}
}
-static void dont_care_about_shutdown_completion(void *ignored) {}
-
/* called when all listening endpoints have been shutdown, so no further
events will be received on them - at this point it's safe to destroy
things */
-static void deactivated_all_ports(grpc_tcp_server *s) {
+static void deactivated_all_ports(grpc_tcp_server *s,
+ grpc_call_list *call_list) {
size_t i;
/* delete ALL the things */
@@ -192,38 +185,35 @@
}
sp->destroyed_closure.cb = destroyed_port;
sp->destroyed_closure.cb_arg = s;
- grpc_fd_orphan(sp->emfd, &sp->destroyed_closure, "tcp_listener_shutdown");
+ grpc_fd_orphan(sp->emfd, &sp->destroyed_closure, "tcp_listener_shutdown",
+ call_list);
}
gpr_mu_unlock(&s->mu);
} else {
gpr_mu_unlock(&s->mu);
- finish_shutdown(s);
+ finish_shutdown(s, call_list);
}
}
-void grpc_tcp_server_destroy(
- grpc_tcp_server *s, void (*shutdown_complete)(void *shutdown_complete_arg),
- void *shutdown_complete_arg) {
+void grpc_tcp_server_destroy(grpc_tcp_server *s, grpc_closure *closure,
+ grpc_call_list *call_list) {
size_t i;
gpr_mu_lock(&s->mu);
GPR_ASSERT(!s->shutdown);
s->shutdown = 1;
- s->shutdown_complete = shutdown_complete
- ? shutdown_complete
- : dont_care_about_shutdown_completion;
- s->shutdown_complete_arg = shutdown_complete_arg;
+ s->shutdown_complete = closure;
/* shutdown all fd's */
if (s->active_ports) {
for (i = 0; i < s->nports; i++) {
- grpc_fd_shutdown(s->ports[i].emfd);
+ grpc_fd_shutdown(s->ports[i].emfd, call_list);
}
gpr_mu_unlock(&s->mu);
} else {
gpr_mu_unlock(&s->mu);
- deactivated_all_ports(s);
+ deactivated_all_ports(s, call_list);
}
}
@@ -308,7 +298,7 @@
}
/* event manager callback when reads are ready */
-static void on_read(void *arg, int success) {
+static void on_read(void *arg, int success, grpc_call_list *call_list) {
server_port *sp = arg;
grpc_fd *fdobj;
size_t i;
@@ -331,7 +321,7 @@
case EINTR:
continue;
case EAGAIN:
- grpc_fd_notify_on_read(sp->emfd, &sp->read_closure);
+ grpc_fd_notify_on_read(sp->emfd, &sp->read_closure, call_list);
return;
default:
gpr_log(GPR_ERROR, "Failed accept4: %s", strerror(errno));
@@ -348,16 +338,17 @@
gpr_log(GPR_DEBUG, "SERVER_CONNECT: incoming connection: %s", addr_str);
}
- fdobj = grpc_fd_create(fd, sp->server->workqueue, name);
+ fdobj = grpc_fd_create(fd, name);
/* TODO(ctiller): revise this when we have server-side sharding
of channels -- we certainly should not be automatically adding every
incoming channel to every pollset owned by the server */
for (i = 0; i < sp->server->pollset_count; i++) {
- grpc_pollset_add_fd(sp->server->pollsets[i], fdobj);
+ grpc_pollset_add_fd(sp->server->pollsets[i], fdobj, call_list);
}
sp->server->cb(
sp->server->cb_arg,
- grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str));
+ grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str),
+ call_list);
gpr_free(name);
gpr_free(addr_str);
@@ -369,7 +360,7 @@
gpr_mu_lock(&sp->server->mu);
if (0 == --sp->server->active_ports) {
gpr_mu_unlock(&sp->server->mu);
- deactivated_all_ports(sp->server);
+ deactivated_all_ports(sp->server, call_list);
} else {
gpr_mu_unlock(&sp->server->mu);
}
@@ -396,7 +387,7 @@
sp = &s->ports[s->nports++];
sp->server = s;
sp->fd = fd;
- sp->emfd = grpc_fd_create(fd, s->workqueue, name);
+ sp->emfd = grpc_fd_create(fd, name);
memcpy(sp->addr.untyped, addr, addr_len);
sp->addr_len = addr_len;
GPR_ASSERT(sp->emfd);
@@ -495,7 +486,7 @@
void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollsets,
size_t pollset_count, grpc_tcp_server_cb cb,
- void *cb_arg) {
+ void *cb_arg, grpc_call_list *call_list) {
size_t i, j;
GPR_ASSERT(cb);
gpr_mu_lock(&s->mu);
@@ -507,11 +498,12 @@
s->pollset_count = pollset_count;
for (i = 0; i < s->nports; i++) {
for (j = 0; j < pollset_count; j++) {
- grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd);
+ grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd, call_list);
}
s->ports[i].read_closure.cb = on_read;
s->ports[i].read_closure.cb_arg = &s->ports[i];
- grpc_fd_notify_on_read(s->ports[i].emfd, &s->ports[i].read_closure);
+ grpc_fd_notify_on_read(s->ports[i].emfd, &s->ports[i].read_closure,
+ call_list);
s->active_ports++;
}
gpr_mu_unlock(&s->mu);