From 42b6c93c36d818a75344d16e353144a0e609e01e Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 30 Jul 2015 09:59:25 -0700 Subject: [PATCH 1/3] Make TSAN happy with our lock free stack --- include/grpc/support/atm_gcc_atomic.h | 2 ++ include/grpc/support/atm_gcc_sync.h | 5 +++++ include/grpc/support/atm_win32.h | 5 +++++ src/core/support/stack_lockfree.c | 12 ++++++++++-- 4 files changed, 22 insertions(+), 2 deletions(-) diff --git a/include/grpc/support/atm_gcc_atomic.h b/include/grpc/support/atm_gcc_atomic.h index 65d3d0c60f8..a2c83860289 100644 --- a/include/grpc/support/atm_gcc_atomic.h +++ b/include/grpc/support/atm_gcc_atomic.h @@ -46,6 +46,8 @@ typedef gpr_intptr gpr_atm; #define gpr_atm_no_barrier_load(p) (__atomic_load_n((p), __ATOMIC_RELAXED)) #define gpr_atm_rel_store(p, value) \ (__atomic_store_n((p), (gpr_intptr)(value), __ATOMIC_RELEASE)) +#define gpr_atm_no_barrier_store(p, value) \ + (__atomic_store_n((p), (gpr_intptr)(value), __ATOMIC_RELAXED)) #define gpr_atm_no_barrier_fetch_add(p, delta) \ (__atomic_fetch_add((p), (gpr_intptr)(delta), __ATOMIC_RELAXED)) diff --git a/include/grpc/support/atm_gcc_sync.h b/include/grpc/support/atm_gcc_sync.h index 4955e4436f4..38b5a9eec25 100644 --- a/include/grpc/support/atm_gcc_sync.h +++ b/include/grpc/support/atm_gcc_sync.h @@ -68,6 +68,11 @@ static __inline void gpr_atm_rel_store(gpr_atm *p, gpr_atm value) { *p = value; } +static __inline void gpr_atm_no_barrier_store(gpr_atm *p, gpr_atm value) { + GPR_ATM_COMPILE_BARRIER_(); + *p = value; +} + #undef GPR_ATM_LS_BARRIER_ #undef GPR_ATM_COMPILE_BARRIER_ diff --git a/include/grpc/support/atm_win32.h b/include/grpc/support/atm_win32.h index da99021c24b..694528a9ba9 100644 --- a/include/grpc/support/atm_win32.h +++ b/include/grpc/support/atm_win32.h @@ -57,6 +57,11 @@ static __inline void gpr_atm_rel_store(gpr_atm *p, gpr_atm value) { *p = value; } +static __inline void gpr_atm_no_barrier_store(gpr_atm *p, gpr_atm value) { + /* TODO(ctiller): Can we implement something better here? */ + gpr_atm_rel_store(p, value); +} + static __inline int gpr_atm_no_barrier_cas(gpr_atm *p, gpr_atm o, gpr_atm n) { /* InterlockedCompareExchangePointerNoFence() not available on vista or windows7 */ diff --git a/src/core/support/stack_lockfree.c b/src/core/support/stack_lockfree.c index f24e272207f..bc741f8c703 100644 --- a/src/core/support/stack_lockfree.c +++ b/src/core/support/stack_lockfree.c @@ -95,6 +95,8 @@ gpr_stack_lockfree *gpr_stack_lockfree_create(int entries) { memset(&stack->pushed, 0, sizeof(stack->pushed)); #endif + GPR_ASSERT(sizeof(stack->entries->atm) == sizeof(stack->entries->contents)); + /* Point the head at reserved dummy entry */ stack->head.contents.index = INVALID_ENTRY_INDEX; return stack; @@ -108,11 +110,15 @@ void gpr_stack_lockfree_destroy(gpr_stack_lockfree *stack) { int gpr_stack_lockfree_push(gpr_stack_lockfree *stack, int entry) { lockfree_node head; lockfree_node newhead; + lockfree_node curent; + lockfree_node newent; /* First fill in the entry's index and aba ctr for new head */ newhead.contents.index = (gpr_uint16)entry; /* Also post-increment the aba_ctr */ - newhead.contents.aba_ctr = stack->entries[entry].contents.aba_ctr++; + curent.atm = gpr_atm_no_barrier_load(&stack->entries[entry].atm); + newhead.contents.aba_ctr = ++curent.contents.aba_ctr; + gpr_atm_no_barrier_store(&stack->entries[entry].atm, curent.atm); #ifndef NDEBUG /* Check for double push */ @@ -131,7 +137,9 @@ int gpr_stack_lockfree_push(gpr_stack_lockfree *stack, int entry) { /* Atomically get the existing head value for use */ head.atm = gpr_atm_no_barrier_load(&(stack->head.atm)); /* Point to it */ - stack->entries[entry].contents.index = head.contents.index; + newent.atm = gpr_atm_no_barrier_load(&stack->entries[entry].atm); + newent.contents.index = head.contents.index; + gpr_atm_no_barrier_store(&stack->entries[entry].atm, newent.atm); } while (!gpr_atm_rel_cas(&(stack->head.atm), head.atm, newhead.atm)); /* Use rel_cas above to make sure that entry index is set properly */ return head.contents.index == INVALID_ENTRY_INDEX; From e2b3bfa4e4669611305798fdec31648a457ff935 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 30 Jul 2015 10:40:22 -0700 Subject: [PATCH 2/3] Ensure server is alive when we free requested calls --- src/core/surface/server.c | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/core/surface/server.c b/src/core/surface/server.c index f19bcbd090b..7031e639166 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -1271,6 +1271,8 @@ static void done_request_event(void *req, grpc_cq_completion *c) { } else { gpr_free(req); } + + server_unref(server); } static void fail_call(grpc_server *server, requested_call *rc) { @@ -1283,6 +1285,7 @@ static void fail_call(grpc_server *server, requested_call *rc) { rc->data.registered.initial_metadata->count = 0; break; } + server_ref(server); grpc_cq_end_op(rc->cq_for_notification, rc->tag, 0, done_request_event, rc, &rc->completion); } @@ -1293,6 +1296,8 @@ static void publish_registered_or_batch(grpc_call *call, int success, grpc_call_stack_element(grpc_call_get_call_stack(call), 0); requested_call *rc = prc; call_data *calld = elem->call_data; + channel_data *chand = elem->channel_data; + server_ref(chand->server); grpc_cq_end_op(calld->cq_new, rc->tag, success, done_request_event, rc, &rc->completion); GRPC_CALL_INTERNAL_UNREF(call, "server", 0); From 0613e5835b30367d7e60c934044a9a074df109d8 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 30 Jul 2015 11:55:43 -0700 Subject: [PATCH 3/3] Fix a race that could cause a double delete --- src/core/iomgr/fd_posix.c | 6 +++++- src/core/iomgr/fd_posix.h | 1 + src/core/iomgr/tcp_server_posix.c | 2 ++ 3 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c index 6ad377ce1c2..a2df838d4a0 100644 --- a/src/core/iomgr/fd_posix.c +++ b/src/core/iomgr/fd_posix.c @@ -102,6 +102,7 @@ static grpc_fd *alloc_fd(int fd) { r->freelist_next = NULL; r->read_watcher = r->write_watcher = NULL; r->on_done_closure = NULL; + r->closed = 0; return r; } @@ -209,6 +210,8 @@ void grpc_fd_orphan(grpc_fd *fd, grpc_iomgr_closure *on_done, REF_BY(fd, 1, reason); /* remove active status, but keep referenced */ gpr_mu_lock(&fd->watcher_mu); if (!has_watchers(fd)) { + GPR_ASSERT(!fd->closed); + fd->closed = 1; close(fd->fd); if (fd->on_done_closure) { grpc_iomgr_add_callback(fd->on_done_closure); @@ -426,7 +429,8 @@ void grpc_fd_end_poll(grpc_fd_watcher *watcher, int got_read, int got_write) { if (kick) { maybe_wake_one_watcher_locked(fd); } - if (grpc_fd_is_orphaned(fd) && !has_watchers(fd)) { + if (grpc_fd_is_orphaned(fd) && !has_watchers(fd) && !fd->closed) { + fd->closed = 1; close(fd->fd); if (fd->on_done_closure != NULL) { grpc_iomgr_add_callback(fd->on_done_closure); diff --git a/src/core/iomgr/fd_posix.h b/src/core/iomgr/fd_posix.h index 94d0019fa4b..4e8e267ffd1 100644 --- a/src/core/iomgr/fd_posix.h +++ b/src/core/iomgr/fd_posix.h @@ -60,6 +60,7 @@ struct grpc_fd { gpr_mu set_state_mu; gpr_atm shutdown; + int closed; /* The watcher list. diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c index 85386001123..6399aaadb9c 100644 --- a/src/core/iomgr/tcp_server_posix.c +++ b/src/core/iomgr/tcp_server_posix.c @@ -142,6 +142,7 @@ grpc_tcp_server *grpc_tcp_server_create(void) { static void finish_shutdown(grpc_tcp_server *s) { s->shutdown_complete(s->shutdown_complete_arg); + s->shutdown_complete = NULL; gpr_mu_destroy(&s->mu); @@ -157,6 +158,7 @@ static void destroyed_port(void *server, int success) { gpr_mu_unlock(&s->mu); finish_shutdown(s); } else { + GPR_ASSERT(s->destroyed_ports < s->nports); gpr_mu_unlock(&s->mu); } }