reviewable/pr10712/r1
Craig Tiller 8 years ago
parent f840110de2
commit 79d24fb8eb
  1. 22
      src/core/lib/iomgr/ev_epollex_linux.c
  2. 6
      test/core/util/port_server_client.c
  3. 2
      test/cpp/microbenchmarks/bm_pollset.cc

@ -161,6 +161,7 @@ struct grpc_fd {
/* The fd is either closed or we relinquished control of it. In either /* The fd is either closed or we relinquished control of it. In either
cases, this indicates that the 'fd' on this structure is no longer cases, this indicates that the 'fd' on this structure is no longer
valid */ valid */
gpr_mu orphaned_mu;
bool orphaned; bool orphaned;
gpr_atm read_closure; gpr_atm read_closure;
@ -285,6 +286,7 @@ static void fd_destroy(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
/* Add the fd to the freelist */ /* Add the fd to the freelist */
grpc_iomgr_unregister_object(&fd->iomgr_object); grpc_iomgr_unregister_object(&fd->iomgr_object);
pollable_destroy(&fd->pollable); pollable_destroy(&fd->pollable);
gpr_mu_destroy(&fd->orphaned_mu);
gpr_mu_lock(&fd_freelist_mu); gpr_mu_lock(&fd_freelist_mu);
fd->freelist_next = fd_freelist; fd->freelist_next = fd_freelist;
fd_freelist = fd; fd_freelist = fd;
@ -347,6 +349,7 @@ static grpc_fd *fd_create(int fd, const char *name) {
gpr_atm_rel_store(&new_fd->refst, (gpr_atm)1); gpr_atm_rel_store(&new_fd->refst, (gpr_atm)1);
new_fd->fd = fd; new_fd->fd = fd;
gpr_mu_init(&new_fd->orphaned_mu);
new_fd->orphaned = false; new_fd->orphaned = false;
grpc_lfev_init(&new_fd->read_closure); grpc_lfev_init(&new_fd->read_closure);
grpc_lfev_init(&new_fd->write_closure); grpc_lfev_init(&new_fd->write_closure);
@ -374,11 +377,11 @@ static grpc_fd *fd_create(int fd, const char *name) {
static int fd_wrapped_fd(grpc_fd *fd) { static int fd_wrapped_fd(grpc_fd *fd) {
int ret_fd = -1; int ret_fd = -1;
gpr_mu_lock(&fd->pollable.po.mu); gpr_mu_lock(&fd->orphaned_mu);
if (!fd->orphaned) { if (!fd->orphaned) {
ret_fd = fd->fd; ret_fd = fd->fd;
} }
gpr_mu_unlock(&fd->pollable.po.mu); gpr_mu_unlock(&fd->orphaned_mu);
return ret_fd; return ret_fd;
} }
@ -390,6 +393,7 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
grpc_error *error = GRPC_ERROR_NONE; grpc_error *error = GRPC_ERROR_NONE;
gpr_mu_lock(&fd->pollable.po.mu); gpr_mu_lock(&fd->pollable.po.mu);
gpr_mu_lock(&fd->orphaned_mu);
fd->on_done_closure = on_done; fd->on_done_closure = on_done;
/* If release_fd is not NULL, we should be relinquishing control of the file /* If release_fd is not NULL, we should be relinquishing control of the file
@ -413,6 +417,7 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
grpc_closure_sched(exec_ctx, fd->on_done_closure, GRPC_ERROR_REF(error)); grpc_closure_sched(exec_ctx, fd->on_done_closure, GRPC_ERROR_REF(error));
gpr_mu_unlock(&fd->orphaned_mu);
gpr_mu_unlock(&fd->pollable.po.mu); gpr_mu_unlock(&fd->pollable.po.mu);
UNREF_BY(exec_ctx, fd, 2, reason); /* Drop the reference */ UNREF_BY(exec_ctx, fd, 2, reason); /* Drop the reference */
GRPC_LOG_IF_ERROR("fd_orphan", GRPC_ERROR_REF(error)); GRPC_LOG_IF_ERROR("fd_orphan", GRPC_ERROR_REF(error));
@ -545,9 +550,11 @@ static void pollable_init(pollable *p, polling_obj_type type) {
} }
static void pollable_destroy(pollable *p) { static void pollable_destroy(pollable *p) {
close(p->epfd);
grpc_wakeup_fd_destroy(&p->wakeup);
po_destroy(&p->po); po_destroy(&p->po);
if (p->epfd != -1) {
close(p->epfd);
grpc_wakeup_fd_destroy(&p->wakeup);
}
} }
/* ensure that p->epfd, p->wakeup are initialized; p->po.mu must be held */ /* ensure that p->epfd, p->wakeup are initialized; p->po.mu must be held */
@ -590,7 +597,13 @@ static grpc_error *pollable_add_fd(pollable *p, grpc_fd *fd) {
grpc_error *error = GRPC_ERROR_NONE; grpc_error *error = GRPC_ERROR_NONE;
static const char *err_desc = "pollable_add_fd"; static const char *err_desc = "pollable_add_fd";
const int epfd = p->epfd; const int epfd = p->epfd;
GPR_ASSERT(epfd != -1);
gpr_mu_lock(&fd->orphaned_mu);
if (fd->orphaned) {
gpr_mu_unlock(&fd->orphaned_mu);
return GRPC_ERROR_NONE;
}
struct epoll_event ev_fd = { struct epoll_event ev_fd = {
.events = EPOLLET | EPOLLIN | EPOLLOUT | EPOLLEXCLUSIVE, .data.ptr = fd}; .events = EPOLLET | EPOLLIN | EPOLLOUT | EPOLLEXCLUSIVE, .data.ptr = fd};
if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd->fd, &ev_fd) != 0) { if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd->fd, &ev_fd) != 0) {
@ -614,6 +627,7 @@ static grpc_error *pollable_add_fd(pollable *p, grpc_fd *fd) {
append_error(&error, GRPC_OS_ERROR(errno, "epoll_ctl"), err_desc); append_error(&error, GRPC_OS_ERROR(errno, "epoll_ctl"), err_desc);
} }
} }
gpr_mu_unlock(&fd->orphaned_mu);
return error; return error;
} }

@ -60,7 +60,6 @@ static void destroy_pops_and_shutdown(grpc_exec_ctx *exec_ctx, void *p,
grpc_pollset *pollset = grpc_polling_entity_pollset(p); grpc_pollset *pollset = grpc_polling_entity_pollset(p);
grpc_pollset_destroy(exec_ctx, pollset); grpc_pollset_destroy(exec_ctx, pollset);
gpr_free(pollset); gpr_free(pollset);
grpc_shutdown();
} }
static void freed_port_from_server(grpc_exec_ctx *exec_ctx, void *arg, static void freed_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,
@ -122,12 +121,13 @@ void grpc_free_port_using_server(int port) {
gpr_mu_unlock(pr.mu); gpr_mu_unlock(pr.mu);
grpc_httpcli_context_destroy(&exec_ctx, &context); grpc_httpcli_context_destroy(&exec_ctx, &context);
grpc_exec_ctx_finish(&exec_ctx);
grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops), grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops),
shutdown_closure); shutdown_closure);
grpc_exec_ctx_finish(&exec_ctx); grpc_exec_ctx_finish(&exec_ctx);
gpr_free(path); gpr_free(path);
grpc_http_response_destroy(&rsp); grpc_http_response_destroy(&rsp);
grpc_shutdown();
} }
typedef struct portreq { typedef struct portreq {
@ -239,7 +239,6 @@ int grpc_pick_port_using_server(void) {
grpc_closure_create(got_port_from_server, &pr, grpc_schedule_on_exec_ctx), grpc_closure_create(got_port_from_server, &pr, grpc_schedule_on_exec_ctx),
&pr.response); &pr.response);
grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_resource_quota_unref_internal(&exec_ctx, resource_quota);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(pr.mu); gpr_mu_lock(pr.mu);
while (pr.port == -1) { while (pr.port == -1) {
grpc_pollset_worker *worker = NULL; grpc_pollset_worker *worker = NULL;
@ -258,6 +257,7 @@ int grpc_pick_port_using_server(void) {
grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops), grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops),
shutdown_closure); shutdown_closure);
grpc_exec_ctx_finish(&exec_ctx); grpc_exec_ctx_finish(&exec_ctx);
grpc_shutdown();
return pr.port; return pr.port;
} }

@ -59,7 +59,7 @@ extern "C" {
auto& force_library_initialization = Library::get(); auto& force_library_initialization = Library::get();
static void shutdown_ps(grpc_exec_ctx* exec_ctx, void* ps, grpc_error* error) { static void shutdown_ps(grpc_exec_ctx* exec_ctx, void* ps, grpc_error* error) {
grpc_pollset_destroy(static_cast<grpc_pollset*>(ps)); grpc_pollset_destroy(exec_ctx, static_cast<grpc_pollset*>(ps));
} }
static void BM_CreateDestroyPollset(benchmark::State& state) { static void BM_CreateDestroyPollset(benchmark::State& state) {

Loading…
Cancel
Save