mirror of https://github.com/grpc/grpc.git
Currently, if two threads call grpc_completion_queue_pluck on the same completion queue for different tags, there is a 50% chance that we deliver the completion wakeup to the wrong poller - forcing the correct poller to wait until its polling times out before it can return an event up to the application. This change tweaks our polling interfaces so that we can indeed wake a specific poller. Nothing has been performance tuned yet. It's definitely sub-optimal in a number of places. Wakeup file-descriptors should be recycled. We should have a path that avoids calling poll() followed by epoll(). We can probably live without it right at the second though. This code will fail on Windows at least (I'll do that port when I'm in the office and have a Windows machine).pull/2713/head
Craig Tiller
9 years ago
parent
4b6a0c73f3
commit
5ddbb9d405
37 changed files with 384 additions and 795 deletions
@ -1,168 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#ifdef GPR_POSIX_SOCKET |
||||
#include "src/core/iomgr/pollset_kick_posix.h" |
||||
|
||||
#include <errno.h> |
||||
#include <string.h> |
||||
#include <unistd.h> |
||||
|
||||
#include "src/core/iomgr/socket_utils_posix.h" |
||||
#include "src/core/iomgr/wakeup_fd_posix.h" |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
/* This implementation is based on a freelist of wakeup fds, with extra logic to
|
||||
* handle kicks while there is no attached fd. */ |
||||
|
||||
/* TODO(klempner): Autosize this, and consider providing a way to disable the
|
||||
* cap entirely on systems with large fd limits */ |
||||
#define GRPC_MAX_CACHED_WFDS 50 |
||||
|
||||
static grpc_kick_fd_info *fd_freelist = NULL; |
||||
static int fd_freelist_count = 0; |
||||
static gpr_mu fd_freelist_mu; |
||||
|
||||
static grpc_kick_fd_info *allocate_wfd(void) { |
||||
grpc_kick_fd_info *info = NULL; |
||||
gpr_mu_lock(&fd_freelist_mu); |
||||
if (fd_freelist != NULL) { |
||||
info = fd_freelist; |
||||
fd_freelist = fd_freelist->next; |
||||
--fd_freelist_count; |
||||
} |
||||
gpr_mu_unlock(&fd_freelist_mu); |
||||
if (info == NULL) { |
||||
info = gpr_malloc(sizeof(*info)); |
||||
grpc_wakeup_fd_create(&info->wakeup_fd); |
||||
info->next = NULL; |
||||
} |
||||
return info; |
||||
} |
||||
|
||||
static void destroy_wfd(grpc_kick_fd_info *wfd) { |
||||
grpc_wakeup_fd_destroy(&wfd->wakeup_fd); |
||||
gpr_free(wfd); |
||||
} |
||||
|
||||
static void free_wfd(grpc_kick_fd_info *fd_info) { |
||||
gpr_mu_lock(&fd_freelist_mu); |
||||
if (fd_freelist_count < GRPC_MAX_CACHED_WFDS) { |
||||
fd_info->next = fd_freelist; |
||||
fd_freelist = fd_info; |
||||
fd_freelist_count++; |
||||
fd_info = NULL; |
||||
} |
||||
gpr_mu_unlock(&fd_freelist_mu); |
||||
|
||||
if (fd_info) { |
||||
destroy_wfd(fd_info); |
||||
} |
||||
} |
||||
|
||||
void grpc_pollset_kick_init(grpc_pollset_kick_state *kick_state) { |
||||
gpr_mu_init(&kick_state->mu); |
||||
kick_state->kicked = 0; |
||||
kick_state->fd_list.next = kick_state->fd_list.prev = &kick_state->fd_list; |
||||
} |
||||
|
||||
void grpc_pollset_kick_destroy(grpc_pollset_kick_state *kick_state) { |
||||
gpr_mu_destroy(&kick_state->mu); |
||||
GPR_ASSERT(kick_state->fd_list.next == &kick_state->fd_list); |
||||
} |
||||
|
||||
grpc_kick_fd_info *grpc_pollset_kick_pre_poll( |
||||
grpc_pollset_kick_state *kick_state) { |
||||
grpc_kick_fd_info *fd_info; |
||||
gpr_mu_lock(&kick_state->mu); |
||||
if (kick_state->kicked) { |
||||
kick_state->kicked = 0; |
||||
gpr_mu_unlock(&kick_state->mu); |
||||
return NULL; |
||||
} |
||||
fd_info = allocate_wfd(); |
||||
fd_info->next = &kick_state->fd_list; |
||||
fd_info->prev = fd_info->next->prev; |
||||
fd_info->next->prev = fd_info->prev->next = fd_info; |
||||
gpr_mu_unlock(&kick_state->mu); |
||||
return fd_info; |
||||
} |
||||
|
||||
void grpc_pollset_kick_consume(grpc_pollset_kick_state *kick_state, |
||||
grpc_kick_fd_info *fd_info) { |
||||
grpc_wakeup_fd_consume_wakeup(&fd_info->wakeup_fd); |
||||
} |
||||
|
||||
void grpc_pollset_kick_post_poll(grpc_pollset_kick_state *kick_state, |
||||
grpc_kick_fd_info *fd_info) { |
||||
gpr_mu_lock(&kick_state->mu); |
||||
fd_info->next->prev = fd_info->prev; |
||||
fd_info->prev->next = fd_info->next; |
||||
free_wfd(fd_info); |
||||
gpr_mu_unlock(&kick_state->mu); |
||||
} |
||||
|
||||
void grpc_pollset_kick_kick(grpc_pollset_kick_state *kick_state) { |
||||
gpr_mu_lock(&kick_state->mu); |
||||
if (kick_state->fd_list.next != &kick_state->fd_list) { |
||||
grpc_wakeup_fd_wakeup(&kick_state->fd_list.next->wakeup_fd); |
||||
} else { |
||||
kick_state->kicked = 1; |
||||
} |
||||
gpr_mu_unlock(&kick_state->mu); |
||||
} |
||||
|
||||
void grpc_pollset_kick_global_init_fallback_fd(void) { |
||||
gpr_mu_init(&fd_freelist_mu); |
||||
grpc_wakeup_fd_global_init_force_fallback(); |
||||
} |
||||
|
||||
void grpc_pollset_kick_global_init(void) { |
||||
gpr_mu_init(&fd_freelist_mu); |
||||
grpc_wakeup_fd_global_init(); |
||||
} |
||||
|
||||
void grpc_pollset_kick_global_destroy(void) { |
||||
while (fd_freelist != NULL) { |
||||
grpc_kick_fd_info *current = fd_freelist; |
||||
fd_freelist = fd_freelist->next; |
||||
destroy_wfd(current); |
||||
} |
||||
grpc_wakeup_fd_global_destroy(); |
||||
gpr_mu_destroy(&fd_freelist_mu); |
||||
} |
||||
|
||||
#endif /* GPR_POSIX_SOCKET */ |
||||
@ -1,93 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_POSIX_H |
||||
#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_POSIX_H |
||||
|
||||
#include "src/core/iomgr/wakeup_fd_posix.h" |
||||
#include <grpc/support/sync.h> |
||||
|
||||
/* pollset kicking allows breaking a thread out of polling work for
|
||||
a given pollset. |
||||
writing a byte to a pipe is used as a posix-ly portable base |
||||
mechanism, and eventfds are utilized on Linux for better performance. */ |
||||
|
||||
typedef struct grpc_kick_fd_info { |
||||
grpc_wakeup_fd_info wakeup_fd; |
||||
/* used for polling list and free list */ |
||||
struct grpc_kick_fd_info *next; |
||||
/* only used when polling */ |
||||
struct grpc_kick_fd_info *prev; |
||||
} grpc_kick_fd_info; |
||||
|
||||
typedef struct grpc_pollset_kick_state { |
||||
gpr_mu mu; |
||||
int kicked; |
||||
struct grpc_kick_fd_info fd_list; |
||||
} grpc_pollset_kick_state; |
||||
|
||||
#define GRPC_POLLSET_KICK_GET_FD(kick_fd_info) \ |
||||
GRPC_WAKEUP_FD_GET_READ_FD(&(kick_fd_info)->wakeup_fd) |
||||
|
||||
/* This is an abstraction around the typical pipe mechanism for waking up a
|
||||
thread sitting in a poll() style call. */ |
||||
|
||||
void grpc_pollset_kick_global_init(void); |
||||
void grpc_pollset_kick_global_destroy(void); |
||||
|
||||
void grpc_pollset_kick_init(grpc_pollset_kick_state *kick_state); |
||||
void grpc_pollset_kick_destroy(grpc_pollset_kick_state *kick_state); |
||||
|
||||
/* Guarantees a pure posix implementation rather than a specialized one, if
|
||||
* applicable. Intended for testing. */ |
||||
void grpc_pollset_kick_global_init_fallback_fd(void); |
||||
|
||||
/* Must be called before entering poll(). If return value is NULL, this consumed
|
||||
an existing kick. Otherwise the return value is an FD to add to the poll set. |
||||
*/ |
||||
grpc_kick_fd_info *grpc_pollset_kick_pre_poll( |
||||
grpc_pollset_kick_state *kick_state); |
||||
|
||||
/* Consume an existing kick. Must be called after poll returns that the fd was
|
||||
readable, and before calling kick_post_poll. */ |
||||
void grpc_pollset_kick_consume(grpc_pollset_kick_state *kick_state, |
||||
grpc_kick_fd_info *fd_info); |
||||
|
||||
/* Must be called after pre_poll, and after consume if applicable */ |
||||
void grpc_pollset_kick_post_poll(grpc_pollset_kick_state *kick_state, |
||||
grpc_kick_fd_info *fd_info); |
||||
|
||||
/* Actually kick */ |
||||
void grpc_pollset_kick_kick(grpc_pollset_kick_state *kick_state); |
||||
|
||||
#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_POSIX_H */ |
||||
@ -1,130 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include "src/core/iomgr/pollset_kick_posix.h" |
||||
|
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
static void test_allocation(void) { |
||||
grpc_pollset_kick_state state; |
||||
grpc_pollset_kick_init(&state); |
||||
grpc_pollset_kick_destroy(&state); |
||||
} |
||||
|
||||
static void test_non_kick(void) { |
||||
grpc_pollset_kick_state state; |
||||
grpc_kick_fd_info *kfd; |
||||
|
||||
grpc_pollset_kick_init(&state); |
||||
kfd = grpc_pollset_kick_pre_poll(&state); |
||||
GPR_ASSERT(kfd != NULL); |
||||
|
||||
grpc_pollset_kick_post_poll(&state, kfd); |
||||
grpc_pollset_kick_destroy(&state); |
||||
} |
||||
|
||||
static void test_basic_kick(void) { |
||||
/* Kicked during poll */ |
||||
grpc_pollset_kick_state state; |
||||
grpc_kick_fd_info *kfd; |
||||
grpc_pollset_kick_init(&state); |
||||
|
||||
kfd = grpc_pollset_kick_pre_poll(&state); |
||||
GPR_ASSERT(kfd != NULL); |
||||
|
||||
grpc_pollset_kick_kick(&state); |
||||
|
||||
/* Now hypothetically we polled and found that we were kicked */ |
||||
grpc_pollset_kick_consume(&state, kfd); |
||||
|
||||
grpc_pollset_kick_post_poll(&state, kfd); |
||||
|
||||
grpc_pollset_kick_destroy(&state); |
||||
} |
||||
|
||||
static void test_non_poll_kick(void) { |
||||
/* Kick before entering poll */ |
||||
grpc_pollset_kick_state state; |
||||
grpc_kick_fd_info *kfd; |
||||
|
||||
grpc_pollset_kick_init(&state); |
||||
|
||||
grpc_pollset_kick_kick(&state); |
||||
kfd = grpc_pollset_kick_pre_poll(&state); |
||||
GPR_ASSERT(kfd == NULL); |
||||
grpc_pollset_kick_destroy(&state); |
||||
} |
||||
|
||||
#define GRPC_MAX_CACHED_PIPES 50 |
||||
|
||||
static void test_over_free(void) { |
||||
/* Check high watermark pipe free logic */ |
||||
int i; |
||||
grpc_kick_fd_info **kfds = |
||||
gpr_malloc(sizeof(grpc_kick_fd_info *) * GRPC_MAX_CACHED_PIPES); |
||||
grpc_pollset_kick_state state; |
||||
grpc_pollset_kick_init(&state); |
||||
for (i = 0; i < GRPC_MAX_CACHED_PIPES; ++i) { |
||||
kfds[i] = grpc_pollset_kick_pre_poll(&state); |
||||
GPR_ASSERT(kfds[i] != NULL); |
||||
} |
||||
|
||||
for (i = 0; i < GRPC_MAX_CACHED_PIPES; ++i) { |
||||
grpc_pollset_kick_post_poll(&state, kfds[i]); |
||||
} |
||||
grpc_pollset_kick_destroy(&state); |
||||
gpr_free(kfds); |
||||
} |
||||
|
||||
static void run_tests(void) { |
||||
test_allocation(); |
||||
test_basic_kick(); |
||||
test_non_poll_kick(); |
||||
test_non_kick(); |
||||
test_over_free(); |
||||
} |
||||
|
||||
int main(int argc, char **argv) { |
||||
grpc_test_init(argc, argv); |
||||
|
||||
grpc_pollset_kick_global_init(); |
||||
run_tests(); |
||||
grpc_pollset_kick_global_destroy(); |
||||
|
||||
grpc_pollset_kick_global_init_fallback_fd(); |
||||
run_tests(); |
||||
grpc_pollset_kick_global_destroy(); |
||||
return 0; |
||||
} |
||||
Loading…
Reference in new issue