diff options
61 files changed, 1825 insertions, 72 deletions
@@ -474,6 +474,7 @@ grpc_cc_library( "src/core/lib/iomgr/endpoint_pair_windows.c", "src/core/lib/iomgr/error.c", "src/core/lib/iomgr/ev_epoll_linux.c", + "src/core/lib/iomgr/ev_epollex_linux.c", "src/core/lib/iomgr/ev_poll_posix.c", "src/core/lib/iomgr/ev_posix.c", "src/core/lib/iomgr/exec_ctx.c", @@ -596,6 +597,7 @@ grpc_cc_library( "src/core/lib/iomgr/error.h", "src/core/lib/iomgr/error_internal.h", "src/core/lib/iomgr/ev_epoll_linux.h", + "src/core/lib/iomgr/ev_epollex_linux.h", "src/core/lib/iomgr/ev_poll_posix.h", "src/core/lib/iomgr/ev_posix.h", "src/core/lib/iomgr/exec_ctx.h", diff --git a/CMakeLists.txt b/CMakeLists.txt index 0283810a21..d02c99f245 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -935,6 +935,7 @@ add_library(grpc src/core/lib/iomgr/endpoint_pair_windows.c src/core/lib/iomgr/error.c src/core/lib/iomgr/ev_epoll_linux.c + src/core/lib/iomgr/ev_epollex_linux.c src/core/lib/iomgr/ev_poll_posix.c src/core/lib/iomgr/ev_posix.c src/core/lib/iomgr/exec_ctx.c @@ -1260,6 +1261,7 @@ add_library(grpc_cronet src/core/lib/iomgr/endpoint_pair_windows.c src/core/lib/iomgr/error.c src/core/lib/iomgr/ev_epoll_linux.c + src/core/lib/iomgr/ev_epollex_linux.c src/core/lib/iomgr/ev_poll_posix.c src/core/lib/iomgr/ev_posix.c src/core/lib/iomgr/exec_ctx.c @@ -1570,6 +1572,7 @@ add_library(grpc_test_util src/core/lib/iomgr/endpoint_pair_windows.c src/core/lib/iomgr/error.c src/core/lib/iomgr/ev_epoll_linux.c + src/core/lib/iomgr/ev_epollex_linux.c src/core/lib/iomgr/ev_poll_posix.c src/core/lib/iomgr/ev_posix.c src/core/lib/iomgr/exec_ctx.c @@ -1826,6 +1829,7 @@ add_library(grpc_unsecure src/core/lib/iomgr/endpoint_pair_windows.c src/core/lib/iomgr/error.c src/core/lib/iomgr/ev_epoll_linux.c + src/core/lib/iomgr/ev_epollex_linux.c src/core/lib/iomgr/ev_poll_posix.c src/core/lib/iomgr/ev_posix.c src/core/lib/iomgr/exec_ctx.c @@ -2245,6 +2249,7 @@ add_library(grpc++ src/core/lib/iomgr/endpoint_pair_windows.c src/core/lib/iomgr/error.c src/core/lib/iomgr/ev_epoll_linux.c + src/core/lib/iomgr/ev_epollex_linux.c src/core/lib/iomgr/ev_poll_posix.c src/core/lib/iomgr/ev_posix.c src/core/lib/iomgr/exec_ctx.c @@ -2570,6 +2575,7 @@ add_library(grpc++_cronet src/core/lib/iomgr/endpoint_pair_windows.c src/core/lib/iomgr/error.c src/core/lib/iomgr/ev_epoll_linux.c + src/core/lib/iomgr/ev_epollex_linux.c src/core/lib/iomgr/ev_poll_posix.c src/core/lib/iomgr/ev_posix.c src/core/lib/iomgr/exec_ctx.c @@ -3334,6 +3340,7 @@ add_library(grpc++_unsecure src/core/lib/iomgr/endpoint_pair_windows.c src/core/lib/iomgr/error.c src/core/lib/iomgr/ev_epoll_linux.c + src/core/lib/iomgr/ev_epollex_linux.c src/core/lib/iomgr/ev_poll_posix.c src/core/lib/iomgr/ev_posix.c src/core/lib/iomgr/exec_ctx.c @@ -2888,6 +2888,7 @@ LIBGRPC_SRC = \ src/core/lib/iomgr/endpoint_pair_windows.c \ src/core/lib/iomgr/error.c \ src/core/lib/iomgr/ev_epoll_linux.c \ + src/core/lib/iomgr/ev_epollex_linux.c \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_posix.c \ src/core/lib/iomgr/exec_ctx.c \ @@ -3211,6 +3212,7 @@ LIBGRPC_CRONET_SRC = \ src/core/lib/iomgr/endpoint_pair_windows.c \ src/core/lib/iomgr/error.c \ src/core/lib/iomgr/ev_epoll_linux.c \ + src/core/lib/iomgr/ev_epollex_linux.c \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_posix.c \ src/core/lib/iomgr/exec_ctx.c \ @@ -3520,6 +3522,7 @@ LIBGRPC_TEST_UTIL_SRC = \ src/core/lib/iomgr/endpoint_pair_windows.c \ src/core/lib/iomgr/error.c \ src/core/lib/iomgr/ev_epoll_linux.c \ + src/core/lib/iomgr/ev_epollex_linux.c \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_posix.c \ src/core/lib/iomgr/exec_ctx.c \ @@ -3748,6 +3751,7 @@ LIBGRPC_UNSECURE_SRC = \ src/core/lib/iomgr/endpoint_pair_windows.c \ src/core/lib/iomgr/error.c \ src/core/lib/iomgr/ev_epoll_linux.c \ + src/core/lib/iomgr/ev_epollex_linux.c \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_posix.c \ src/core/lib/iomgr/exec_ctx.c \ @@ -4144,6 +4148,7 @@ LIBGRPC++_SRC = \ src/core/lib/iomgr/endpoint_pair_windows.c \ src/core/lib/iomgr/error.c \ src/core/lib/iomgr/ev_epoll_linux.c \ + src/core/lib/iomgr/ev_epollex_linux.c \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_posix.c \ src/core/lib/iomgr/exec_ctx.c \ @@ -4477,6 +4482,7 @@ LIBGRPC++_CRONET_SRC = \ src/core/lib/iomgr/endpoint_pair_windows.c \ src/core/lib/iomgr/error.c \ src/core/lib/iomgr/ev_epoll_linux.c \ + src/core/lib/iomgr/ev_epollex_linux.c \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_posix.c \ src/core/lib/iomgr/exec_ctx.c \ @@ -5238,6 +5244,7 @@ LIBGRPC++_UNSECURE_SRC = \ src/core/lib/iomgr/endpoint_pair_windows.c \ src/core/lib/iomgr/error.c \ src/core/lib/iomgr/ev_epoll_linux.c \ + src/core/lib/iomgr/ev_epollex_linux.c \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_posix.c \ src/core/lib/iomgr/exec_ctx.c \ diff --git a/binding.gyp b/binding.gyp index a08bf78263..82242c057c 100644 --- a/binding.gyp +++ b/binding.gyp @@ -673,6 +673,7 @@ 'src/core/lib/iomgr/endpoint_pair_windows.c', 'src/core/lib/iomgr/error.c', 'src/core/lib/iomgr/ev_epoll_linux.c', + 'src/core/lib/iomgr/ev_epollex_linux.c', 'src/core/lib/iomgr/ev_poll_posix.c', 'src/core/lib/iomgr/ev_posix.c', 'src/core/lib/iomgr/exec_ctx.c', diff --git a/build.yaml b/build.yaml index ed7f4cce64..fb817af012 100644 --- a/build.yaml +++ b/build.yaml @@ -194,6 +194,7 @@ filegroups: - src/core/lib/iomgr/error.h - src/core/lib/iomgr/error_internal.h - src/core/lib/iomgr/ev_epoll_linux.h + - src/core/lib/iomgr/ev_epollex_linux.h - src/core/lib/iomgr/ev_poll_posix.h - src/core/lib/iomgr/ev_posix.h - src/core/lib/iomgr/exec_ctx.h @@ -301,6 +302,7 @@ filegroups: - src/core/lib/iomgr/endpoint_pair_windows.c - src/core/lib/iomgr/error.c - src/core/lib/iomgr/ev_epoll_linux.c + - src/core/lib/iomgr/ev_epollex_linux.c - src/core/lib/iomgr/ev_poll_posix.c - src/core/lib/iomgr/ev_posix.c - src/core/lib/iomgr/exec_ctx.c @@ -109,6 +109,7 @@ if test "$PHP_GRPC" != "no"; then src/core/lib/iomgr/endpoint_pair_windows.c \ src/core/lib/iomgr/error.c \ src/core/lib/iomgr/ev_epoll_linux.c \ + src/core/lib/iomgr/ev_epollex_linux.c \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_posix.c \ src/core/lib/iomgr/exec_ctx.c \ diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 4bd9cd9363..c5ff33ef87 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -276,6 +276,7 @@ Pod::Spec.new do |s| 'src/core/lib/iomgr/error.h', 'src/core/lib/iomgr/error_internal.h', 'src/core/lib/iomgr/ev_epoll_linux.h', + 'src/core/lib/iomgr/ev_epollex_linux.h', 'src/core/lib/iomgr/ev_poll_posix.h', 'src/core/lib/iomgr/ev_posix.h', 'src/core/lib/iomgr/exec_ctx.h', @@ -481,6 +482,7 @@ Pod::Spec.new do |s| 'src/core/lib/iomgr/endpoint_pair_windows.c', 'src/core/lib/iomgr/error.c', 'src/core/lib/iomgr/ev_epoll_linux.c', + 'src/core/lib/iomgr/ev_epollex_linux.c', 'src/core/lib/iomgr/ev_poll_posix.c', 'src/core/lib/iomgr/ev_posix.c', 'src/core/lib/iomgr/exec_ctx.c', @@ -731,6 +733,7 @@ Pod::Spec.new do |s| 'src/core/lib/iomgr/error.h', 'src/core/lib/iomgr/error_internal.h', 'src/core/lib/iomgr/ev_epoll_linux.h', + 'src/core/lib/iomgr/ev_epollex_linux.h', 'src/core/lib/iomgr/ev_poll_posix.h', 'src/core/lib/iomgr/ev_posix.h', 'src/core/lib/iomgr/exec_ctx.h', diff --git a/grpc.gemspec b/grpc.gemspec index e53bd29cd4..33822cc649 100755 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -192,6 +192,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/iomgr/error.h ) s.files += %w( src/core/lib/iomgr/error_internal.h ) s.files += %w( src/core/lib/iomgr/ev_epoll_linux.h ) + s.files += %w( src/core/lib/iomgr/ev_epollex_linux.h ) s.files += %w( src/core/lib/iomgr/ev_poll_posix.h ) s.files += %w( src/core/lib/iomgr/ev_posix.h ) s.files += %w( src/core/lib/iomgr/exec_ctx.h ) @@ -397,6 +398,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/iomgr/endpoint_pair_windows.c ) s.files += %w( src/core/lib/iomgr/error.c ) s.files += %w( src/core/lib/iomgr/ev_epoll_linux.c ) + s.files += %w( src/core/lib/iomgr/ev_epollex_linux.c ) s.files += %w( src/core/lib/iomgr/ev_poll_posix.c ) s.files += %w( src/core/lib/iomgr/ev_posix.c ) s.files += %w( src/core/lib/iomgr/exec_ctx.c ) diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h index 7ac23349c8..a56f81dc4b 100644 --- a/include/grpc++/server_builder.h +++ b/include/grpc++/server_builder.h @@ -195,7 +195,10 @@ class ServerBuilder { struct SyncServerSettings { SyncServerSettings() - : num_cqs(1), min_pollers(1), max_pollers(2), cq_timeout_msec(10000) {} + : num_cqs(gpr_cpu_num_cores()), + min_pollers(1), + max_pollers(2), + cq_timeout_msec(10000) {} // Number of server completion queues to create to listen to incoming RPCs. int num_cqs; diff --git a/package.xml b/package.xml index 8e6b30e07e..3f438a51ce 100644 --- a/package.xml +++ b/package.xml @@ -201,6 +201,7 @@ <file baseinstalldir="/" name="src/core/lib/iomgr/error.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/error_internal.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/ev_epoll_linux.h" role="src" /> + <file baseinstalldir="/" name="src/core/lib/iomgr/ev_epollex_linux.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/ev_poll_posix.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/ev_posix.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/exec_ctx.h" role="src" /> @@ -406,6 +407,7 @@ <file baseinstalldir="/" name="src/core/lib/iomgr/endpoint_pair_windows.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/error.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/ev_epoll_linux.c" role="src" /> + <file baseinstalldir="/" name="src/core/lib/iomgr/ev_epollex_linux.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/ev_poll_posix.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/ev_posix.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/iomgr/exec_ctx.c" role="src" /> diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c index e603a75593..0f1bd833f7 100644 --- a/src/core/lib/iomgr/ev_epoll_linux.c +++ b/src/core/lib/iomgr/ev_epoll_linux.c @@ -63,8 +63,9 @@ #include "src/core/lib/profiling/timers.h" #include "src/core/lib/support/block_annotate.h" +#define GRPC_POLLSET_KICK_BROADCAST ((grpc_pollset_worker *)1) + /* TODO: sreek - Move this to init.c and initialize this like other tracers. */ -static int grpc_polling_trace = 0; /* Disabled by default */ #define GRPC_POLLING_TRACE(fmt, ...) \ if (grpc_polling_trace) { \ gpr_log(GPR_INFO, (fmt), __VA_ARGS__); \ @@ -1332,7 +1333,7 @@ static void pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, /* pollset_shutdown is guaranteed to be called before pollset_destroy. So other * than destroying the mutexes, there is nothing special that needs to be done * here */ -static void pollset_destroy(grpc_pollset *pollset) { +static void pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) { GPR_ASSERT(!pollset_has_workers(pollset)); gpr_mu_destroy(&pollset->po.mu); } diff --git a/src/core/lib/iomgr/ev_epollex_linux.c b/src/core/lib/iomgr/ev_epollex_linux.c new file mode 100644 index 0000000000..adbdaf6e5b --- /dev/null +++ b/src/core/lib/iomgr/ev_epollex_linux.c @@ -0,0 +1,1579 @@ +/* + * + * Copyright 2016, 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/lib/iomgr/port.h" + +/* This polling engine is only relevant on linux kernels supporting epoll() */ +#ifdef GRPC_LINUX_EPOLL + +#include "src/core/lib/iomgr/ev_epoll_linux.h" + +#include <assert.h> +#include <errno.h> +#include <poll.h> +#include <pthread.h> +#include <string.h> +#include <sys/epoll.h> +#include <sys/socket.h> +#include <unistd.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include <grpc/support/tls.h> +#include <grpc/support/useful.h> + +#include "src/core/lib/iomgr/ev_posix.h" +#include "src/core/lib/iomgr/iomgr_internal.h" +#include "src/core/lib/iomgr/lockfree_event.h" +#include "src/core/lib/iomgr/timer.h" +#include "src/core/lib/iomgr/wakeup_fd_posix.h" +#include "src/core/lib/iomgr/workqueue.h" +#include "src/core/lib/profiling/timers.h" +#include "src/core/lib/support/block_annotate.h" +#include "src/core/lib/support/spinlock.h" + +#ifndef EPOLLEXCLUSIVE +#define EPOLLEXCLUSIVE (1u << 28) +#endif + +/* TODO: sreek: Right now, this wakes up all pollers. In future we should make + * sure to wake up one polling thread (which can wake up other threads if + * needed) */ +static grpc_wakeup_fd global_wakeup_fd; + +/******************************************************************************* + * Pollset-set sibling link + */ + +typedef enum { + PO_POLLING_GROUP, + PO_POLLSET_SET, + PO_POLLSET, + PO_FD, /* ordering is important: we always want to lock pollsets before fds: + this guarantees that using an fd as a pollable is safe */ + PO_EMPTY_POLLABLE, + PO_COUNT +} polling_obj_type; + +typedef struct polling_obj polling_obj; +typedef struct polling_group polling_group; + +struct polling_obj { + gpr_mu mu; + polling_obj_type type; + polling_group *group; + struct polling_obj *next; + struct polling_obj *prev; +}; + +struct polling_group { + polling_obj po; + gpr_refcount refs; +}; + +static void po_init(polling_obj *po, polling_obj_type type); +static void po_destroy(polling_obj *po); +static void po_join(grpc_exec_ctx *exec_ctx, polling_obj *a, polling_obj *b); +static int po_cmp(polling_obj *a, polling_obj *b); + +static void pg_create(grpc_exec_ctx *exec_ctx, polling_obj **initial_po, + size_t initial_po_count); +static polling_group *pg_ref(polling_group *pg); +static void pg_unref(polling_group *pg); +static void pg_merge(grpc_exec_ctx *exec_ctx, polling_group *a, + polling_group *b); +static void pg_join(grpc_exec_ctx *exec_ctx, polling_group *pg, + polling_obj *po); + +/******************************************************************************* + * pollable Declarations + */ + +typedef struct pollable { + polling_obj po; + int epfd; + grpc_wakeup_fd wakeup; + grpc_pollset_worker *root_worker; +} pollable; + +static pollable g_empty_pollable; + +static void pollable_init(pollable *p, polling_obj_type type); +static void pollable_destroy(pollable *p); +/* ensure that p->epfd, p->wakeup are initialized; p->po.mu must be held */ +static grpc_error *pollable_materialize(pollable *p); + +/******************************************************************************* + * Fd Declarations + */ + +struct grpc_fd { + pollable pollable; + int fd; + /* refst format: + bit 0 : 1=Active / 0=Orphaned + bits 1-n : refcount + Ref/Unref by two to avoid altering the orphaned bit */ + gpr_atm refst; + + /* Wakeup fd used to wake pollers to check the contents of workqueue_items */ + grpc_wakeup_fd workqueue_wakeup_fd; + grpc_closure_scheduler workqueue_scheduler; + /* Spinlock guarding the read end of the workqueue (must be held to pop from + * workqueue_items) */ + gpr_spinlock workqueue_read_mu; + /* Queue of closures to be executed */ + gpr_mpscq workqueue_items; + /* Count of items in workqueue_items */ + gpr_atm workqueue_item_count; + + /* 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 + valid */ + gpr_mu orphaned_mu; + bool orphaned; + + gpr_atm read_closure; + gpr_atm write_closure; + + struct grpc_fd *freelist_next; + grpc_closure *on_done_closure; + + /* The pollset that last noticed that the fd is readable. The actual type + * stored in this is (grpc_pollset *) */ + gpr_atm read_notifier_pollset; + + grpc_iomgr_object iomgr_object; +}; + +static void fd_global_init(void); +static void fd_global_shutdown(void); + +static void workqueue_enqueue(grpc_exec_ctx *exec_ctx, grpc_closure *closure, + grpc_error *error); + +static const grpc_closure_scheduler_vtable workqueue_scheduler_vtable = { + workqueue_enqueue, workqueue_enqueue, "workqueue"}; + +/******************************************************************************* + * Pollset Declarations + */ + +typedef struct pollset_worker_link { + grpc_pollset_worker *next; + grpc_pollset_worker *prev; +} pollset_worker_link; + +typedef enum { + PWL_POLLSET, + PWL_POLLABLE, + POLLSET_WORKER_LINK_COUNT +} pollset_worker_links; + +struct grpc_pollset_worker { + bool kicked; + bool initialized_cv; + pollset_worker_link links[POLLSET_WORKER_LINK_COUNT]; + gpr_cv cv; + grpc_pollset *pollset; + pollable *pollable; +}; + +struct grpc_pollset { + pollable pollable; + pollable *current_pollable; + bool kicked_without_poller; + grpc_closure *shutdown_closure; + grpc_pollset_worker *root_worker; +}; + +/******************************************************************************* + * Pollset-set Declarations + */ +struct grpc_pollset_set { + polling_obj po; +}; + +/******************************************************************************* + * Common helpers + */ + +static bool append_error(grpc_error **composite, grpc_error *error, + const char *desc) { + if (error == GRPC_ERROR_NONE) return true; + if (*composite == GRPC_ERROR_NONE) { + *composite = GRPC_ERROR_CREATE_FROM_COPIED_STRING(desc); + } + *composite = grpc_error_add_child(*composite, error); + return false; +} + +/******************************************************************************* + * Fd Definitions + */ + +/* We need to keep a freelist not because of any concerns of malloc performance + * but instead so that implementations with multiple threads in (for example) + * epoll_wait deal with the race between pollset removal and incoming poll + * notifications. + * + * The problem is that the poller ultimately holds a reference to this + * object, so it is very difficult to know when is safe to free it, at least + * without some expensive synchronization. + * + * If we keep the object freelisted, in the worst case losing this race just + * becomes a spurious read notification on a reused fd. + */ + +/* The alarm system needs to be able to wakeup 'some poller' sometimes + * (specifically when a new alarm needs to be triggered earlier than the next + * alarm 'epoch'). This wakeup_fd gives us something to alert on when such a + * case occurs. */ + +static grpc_fd *fd_freelist = NULL; +static gpr_mu fd_freelist_mu; + +#ifdef GRPC_FD_REF_COUNT_DEBUG +#define REF_BY(fd, n, reason) ref_by(fd, n, reason, __FILE__, __LINE__) +#define UNREF_BY(ec, fd, n, reason) \ + unref_by(ec, fd, n, reason, __FILE__, __LINE__) +static void ref_by(grpc_fd *fd, int n, const char *reason, const char *file, + int line) { + gpr_log(GPR_DEBUG, "FD %d %p ref %d %ld -> %ld [%s; %s:%d]", fd->fd, + (void *)fd, n, gpr_atm_no_barrier_load(&fd->refst), + gpr_atm_no_barrier_load(&fd->refst) + n, reason, file, line); +#else +#define REF_BY(fd, n, reason) ref_by(fd, n) +#define UNREF_BY(ec, fd, n, reason) unref_by(ec, fd, n) +static void ref_by(grpc_fd *fd, int n) { +#endif + GPR_ASSERT(gpr_atm_no_barrier_fetch_add(&fd->refst, n) > 0); +} + +static void fd_destroy(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + grpc_fd *fd = arg; + /* Add the fd to the freelist */ + grpc_iomgr_unregister_object(&fd->iomgr_object); + pollable_destroy(&fd->pollable); + gpr_mu_destroy(&fd->orphaned_mu); + gpr_mu_lock(&fd_freelist_mu); + fd->freelist_next = fd_freelist; + fd_freelist = fd; + + grpc_lfev_destroy(&fd->read_closure); + grpc_lfev_destroy(&fd->write_closure); + + gpr_mu_unlock(&fd_freelist_mu); +} + +#ifdef GRPC_FD_REF_COUNT_DEBUG +static void unref_by(grpc_exec_ctx *exec_ctx, grpc_fd *fd, int n, + const char *reason, const char *file, int line) { + gpr_atm old; + gpr_log(GPR_DEBUG, "FD %d %p unref %d %ld -> %ld [%s; %s:%d]", fd->fd, + (void *)fd, n, gpr_atm_no_barrier_load(&fd->refst), + gpr_atm_no_barrier_load(&fd->refst) - n, reason, file, line); +#else +static void unref_by(grpc_exec_ctx *exec_ctx, grpc_fd *fd, int n) { + gpr_atm old; +#endif + old = gpr_atm_full_fetch_add(&fd->refst, -n); + if (old == n) { + grpc_closure_sched(exec_ctx, grpc_closure_create(fd_destroy, fd, + grpc_schedule_on_exec_ctx), + GRPC_ERROR_NONE); + } else { + GPR_ASSERT(old > n); + } +} + +static void fd_global_init(void) { gpr_mu_init(&fd_freelist_mu); } + +static void fd_global_shutdown(void) { + gpr_mu_lock(&fd_freelist_mu); + gpr_mu_unlock(&fd_freelist_mu); + while (fd_freelist != NULL) { + grpc_fd *fd = fd_freelist; + fd_freelist = fd_freelist->freelist_next; + gpr_free(fd); + } + gpr_mu_destroy(&fd_freelist_mu); +} + +static grpc_fd *fd_create(int fd, const char *name) { + grpc_fd *new_fd = NULL; + + gpr_mu_lock(&fd_freelist_mu); + if (fd_freelist != NULL) { + new_fd = fd_freelist; + fd_freelist = fd_freelist->freelist_next; + } + gpr_mu_unlock(&fd_freelist_mu); + + if (new_fd == NULL) { + new_fd = gpr_malloc(sizeof(grpc_fd)); + } + + pollable_init(&new_fd->pollable, PO_FD); + + gpr_atm_rel_store(&new_fd->refst, (gpr_atm)1); + new_fd->fd = fd; + gpr_mu_init(&new_fd->orphaned_mu); + new_fd->orphaned = false; + grpc_lfev_init(&new_fd->read_closure); + grpc_lfev_init(&new_fd->write_closure); + gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm)NULL); + + GRPC_LOG_IF_ERROR("fd_create", + grpc_wakeup_fd_init(&new_fd->workqueue_wakeup_fd)); + new_fd->workqueue_scheduler.vtable = &workqueue_scheduler_vtable; + new_fd->workqueue_read_mu = GPR_SPINLOCK_INITIALIZER; + gpr_mpscq_init(&new_fd->workqueue_items); + gpr_atm_no_barrier_store(&new_fd->workqueue_item_count, 0); + + new_fd->freelist_next = NULL; + new_fd->on_done_closure = NULL; + + char *fd_name; + gpr_asprintf(&fd_name, "%s fd=%d", name, fd); + grpc_iomgr_register_object(&new_fd->iomgr_object, fd_name); +#ifdef GRPC_FD_REF_COUNT_DEBUG + gpr_log(GPR_DEBUG, "FD %d %p create %s", fd, (void *)new_fd, fd_name); +#endif + gpr_free(fd_name); + return new_fd; +} + +static int fd_wrapped_fd(grpc_fd *fd) { + int ret_fd = -1; + gpr_mu_lock(&fd->orphaned_mu); + if (!fd->orphaned) { + ret_fd = fd->fd; + } + gpr_mu_unlock(&fd->orphaned_mu); + + return ret_fd; +} + +static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, + grpc_closure *on_done, int *release_fd, + const char *reason) { + bool is_fd_closed = false; + grpc_error *error = GRPC_ERROR_NONE; + + gpr_mu_lock(&fd->pollable.po.mu); + gpr_mu_lock(&fd->orphaned_mu); + fd->on_done_closure = on_done; + + /* If release_fd is not NULL, we should be relinquishing control of the file + descriptor fd->fd (but we still own the grpc_fd structure). */ + if (release_fd != NULL) { + *release_fd = fd->fd; + } else { + close(fd->fd); + is_fd_closed = true; + } + + fd->orphaned = true; + + if (!is_fd_closed) { + gpr_log(GPR_DEBUG, "TODO: handle fd removal?"); + } + + /* Remove the active status but keep referenced. We want this grpc_fd struct + to be alive (and not added to freelist) until the end of this function */ + REF_BY(fd, 1, reason); + + 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); + UNREF_BY(exec_ctx, fd, 2, reason); /* Drop the reference */ + GRPC_LOG_IF_ERROR("fd_orphan", GRPC_ERROR_REF(error)); + GRPC_ERROR_UNREF(error); +} + +static grpc_pollset *fd_get_read_notifier_pollset(grpc_exec_ctx *exec_ctx, + grpc_fd *fd) { + gpr_atm notifier = gpr_atm_acq_load(&fd->read_notifier_pollset); + return (grpc_pollset *)notifier; +} + +static bool fd_is_shutdown(grpc_fd *fd) { + return grpc_lfev_is_shutdown(&fd->read_closure); +} + +/* Might be called multiple times */ +static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why) { + if (grpc_lfev_set_shutdown(exec_ctx, &fd->read_closure, + GRPC_ERROR_REF(why))) { + shutdown(fd->fd, SHUT_RDWR); + grpc_lfev_set_shutdown(exec_ctx, &fd->write_closure, GRPC_ERROR_REF(why)); + } + GRPC_ERROR_UNREF(why); +} + +static void fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd, + grpc_closure *closure) { + grpc_lfev_notify_on(exec_ctx, &fd->read_closure, closure); +} + +static void fd_notify_on_write(grpc_exec_ctx *exec_ctx, grpc_fd *fd, + grpc_closure *closure) { + grpc_lfev_notify_on(exec_ctx, &fd->write_closure, closure); +} + +static grpc_workqueue *fd_get_workqueue(grpc_fd *fd) { + REF_BY(fd, 2, "return_workqueue"); + return (grpc_workqueue *)fd; +} + +#ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG +static grpc_workqueue *workqueue_ref(grpc_workqueue *workqueue, + const char *file, int line, + const char *reason) { + if (workqueue != NULL) { + ref_by((grpc_fd *)workqueue, 2, file, line, reason); + } + return workqueue; +} + +static void workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue, + const char *file, int line, const char *reason) { + if (workqueue != NULL) { + unref_by(exec_ctx, (grpc_fd *)workqueue, 2, file, line, reason); + } +} +#else +static grpc_workqueue *workqueue_ref(grpc_workqueue *workqueue) { + if (workqueue != NULL) { + ref_by((grpc_fd *)workqueue, 2); + } + return workqueue; +} + +static void workqueue_unref(grpc_exec_ctx *exec_ctx, + grpc_workqueue *workqueue) { + if (workqueue != NULL) { + unref_by(exec_ctx, (grpc_fd *)workqueue, 2); + } +} +#endif + +static void workqueue_wakeup(grpc_fd *fd) { + GRPC_LOG_IF_ERROR("workqueue_enqueue", + grpc_wakeup_fd_wakeup(&fd->workqueue_wakeup_fd)); +} + +static void workqueue_enqueue(grpc_exec_ctx *exec_ctx, grpc_closure *closure, + grpc_error *error) { + GPR_TIMER_BEGIN("workqueue.enqueue", 0); + grpc_fd *fd = (grpc_fd *)(((char *)closure->scheduler) - + offsetof(grpc_fd, workqueue_scheduler)); + REF_BY(fd, 2, "workqueue_enqueue"); + gpr_atm last = gpr_atm_no_barrier_fetch_add(&fd->workqueue_item_count, 1); + closure->error_data.error = error; + gpr_mpscq_push(&fd->workqueue_items, &closure->next_data.atm_next); + if (last == 0) { + workqueue_wakeup(fd); + } + UNREF_BY(exec_ctx, fd, 2, "workqueue_enqueue"); +} + +static void fd_invoke_workqueue(grpc_exec_ctx *exec_ctx, grpc_fd *fd) { + /* handle spurious wakeups */ + if (!gpr_spinlock_trylock(&fd->workqueue_read_mu)) return; + gpr_mpscq_node *n = gpr_mpscq_pop(&fd->workqueue_items); + gpr_spinlock_unlock(&fd->workqueue_read_mu); + if (n != NULL) { + if (gpr_atm_full_fetch_add(&fd->workqueue_item_count, -1) > 1) { + workqueue_wakeup(fd); + } + grpc_closure *c = (grpc_closure *)n; + grpc_error *error = c->error_data.error; +#ifndef NDEBUG + c->scheduled = false; +#endif + c->cb(exec_ctx, c->cb_arg, error); + GRPC_ERROR_UNREF(error); + } else if (gpr_atm_no_barrier_load(&fd->workqueue_item_count) > 0) { + /* n == NULL might mean there's work but it's not available to be popped + * yet - try to ensure another workqueue wakes up to check shortly if so + */ + workqueue_wakeup(fd); + } +} + +static grpc_closure_scheduler *workqueue_scheduler(grpc_workqueue *workqueue) { + return &((grpc_fd *)workqueue)->workqueue_scheduler; +} + +/******************************************************************************* + * Pollable Definitions + */ + +static void pollable_init(pollable *p, polling_obj_type type) { + po_init(&p->po, type); + p->root_worker = NULL; + p->epfd = -1; +} + +static void pollable_destroy(pollable *p) { + 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 */ +static grpc_error *pollable_materialize(pollable *p) { + if (p->epfd == -1) { + int new_epfd = epoll_create1(EPOLL_CLOEXEC); + if (new_epfd < 0) { + return GRPC_OS_ERROR(errno, "epoll_create1"); + } else { + struct epoll_event ev = {.events = EPOLLIN | EPOLLET | EPOLLEXCLUSIVE, + .data.ptr = &global_wakeup_fd}; + if (epoll_ctl(new_epfd, EPOLL_CTL_ADD, global_wakeup_fd.read_fd, &ev) != + 0) { + grpc_error *err = GRPC_OS_ERROR(errno, "epoll_ctl"); + close(new_epfd); + return err; + } + } + grpc_error *err = grpc_wakeup_fd_init(&p->wakeup); + if (err != GRPC_ERROR_NONE) { + close(new_epfd); + return err; + } + struct epoll_event ev = {.events = EPOLLIN | EPOLLET, + .data.ptr = &p->wakeup}; + if (epoll_ctl(new_epfd, EPOLL_CTL_ADD, p->wakeup.read_fd, &ev) != 0) { + err = GRPC_OS_ERROR(errno, "epoll_ctl"); + close(new_epfd); + grpc_wakeup_fd_destroy(&p->wakeup); + return err; + } + + p->epfd = new_epfd; + } + return GRPC_ERROR_NONE; +} + +/* pollable must be materialized */ +static grpc_error *pollable_add_fd(pollable *p, grpc_fd *fd) { + grpc_error *error = GRPC_ERROR_NONE; + static const char *err_desc = "pollable_add_fd"; + 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 = { + .events = EPOLLET | EPOLLIN | EPOLLOUT | EPOLLEXCLUSIVE, .data.ptr = fd}; + if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd->fd, &ev_fd) != 0) { + switch (errno) { + case EEXIST: /* if this fd is already in the epoll set, the workqueue fd + must also be - just return */ + gpr_mu_unlock(&fd->orphaned_mu); + return GRPC_ERROR_NONE; + default: + append_error(&error, GRPC_OS_ERROR(errno, "epoll_ctl"), err_desc); + } + } + struct epoll_event ev_wq = {.events = EPOLLET | EPOLLIN | EPOLLEXCLUSIVE, + .data.ptr = (void *)(1 + (intptr_t)fd)}; + if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd->workqueue_wakeup_fd.read_fd, &ev_wq) != + 0) { + switch (errno) { + case EEXIST: /* if the workqueue fd is already in the epoll set we're ok + - no need to do anything special */ + break; + default: + append_error(&error, GRPC_OS_ERROR(errno, "epoll_ctl"), err_desc); + } + } + gpr_mu_unlock(&fd->orphaned_mu); + + return error; +} + +/******************************************************************************* + * Pollset Definitions + */ + +GPR_TLS_DECL(g_current_thread_pollset); +GPR_TLS_DECL(g_current_thread_worker); +static bool global_wakeup_fd_initialized = false; + +/* Global state management */ +static grpc_error *pollset_global_init(void) { + gpr_tls_init(&g_current_thread_pollset); + gpr_tls_init(&g_current_thread_worker); + grpc_error *error = GRPC_ERROR_NONE; + static const char *err_desc = "pollset_global_init"; + global_wakeup_fd_initialized = + append_error(&error, grpc_wakeup_fd_init(&global_wakeup_fd), err_desc); + pollable_init(&g_empty_pollable, PO_EMPTY_POLLABLE); + return error; +} + +static void pollset_global_shutdown(void) { + if (global_wakeup_fd_initialized) grpc_wakeup_fd_destroy(&global_wakeup_fd); + pollable_destroy(&g_empty_pollable); + gpr_tls_destroy(&g_current_thread_pollset); + gpr_tls_destroy(&g_current_thread_worker); +} + +static grpc_error *pollset_kick_all(grpc_pollset *pollset) { + grpc_error *error = GRPC_ERROR_NONE; + if (pollset->root_worker != NULL) { + grpc_pollset_worker *worker = pollset->root_worker; + do { + gpr_mu_lock(&worker->pollable->po.mu); + if (worker->initialized_cv) { + worker->kicked = true; + gpr_cv_signal(&worker->cv); + } else { + append_error(&error, grpc_wakeup_fd_wakeup(&worker->pollable->wakeup), + "pollset_shutdown"); + } + gpr_mu_unlock(&worker->pollable->po.mu); + + worker = worker->links[PWL_POLLSET].next; + } while (worker != pollset->root_worker); + } + return error; +} + +static grpc_error *pollset_kick_inner(grpc_pollset *pollset, pollable *p, + grpc_pollset_worker *specific_worker) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, + "PS:%p kick %p tls_pollset=%p tls_worker=%p " + "root_worker=(pollset:%p pollable:%p)", + p, specific_worker, (void *)gpr_tls_get(&g_current_thread_pollset), + (void *)gpr_tls_get(&g_current_thread_worker), pollset->root_worker, + p->root_worker); + } + if (specific_worker == NULL) { + if (gpr_tls_get(&g_current_thread_pollset) != (intptr_t)p) { + if (pollset->root_worker == NULL) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p kicked_any_without_poller", p); + } + pollset->kicked_without_poller = true; + return GRPC_ERROR_NONE; + } else { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p kicked_any_via_wakeup_fd", p); + } + grpc_error *err = pollable_materialize(p); + if (err != GRPC_ERROR_NONE) return err; + return grpc_wakeup_fd_wakeup(&p->wakeup); + } + } else { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p kicked_any_but_awake", p); + } + return GRPC_ERROR_NONE; + } + } else if (specific_worker->kicked) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p kicked_specific_but_already_kicked", p); + } + return GRPC_ERROR_NONE; + } else if (gpr_tls_get(&g_current_thread_worker) == + (intptr_t)specific_worker) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p kicked_specific_but_awake", p); + } + specific_worker->kicked = true; + return GRPC_ERROR_NONE; + } else if (specific_worker == p->root_worker) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p kicked_specific_via_wakeup_fd", p); + } + grpc_error *err = pollable_materialize(p); + if (err != GRPC_ERROR_NONE) return err; + specific_worker->kicked = true; + return grpc_wakeup_fd_wakeup(&p->wakeup); + } else { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p kicked_specific_via_cv", p); + } + specific_worker->kicked = true; + gpr_cv_signal(&specific_worker->cv); + return GRPC_ERROR_NONE; + } +} + +/* p->po.mu must be held before calling this function */ +static grpc_error *pollset_kick(grpc_pollset *pollset, + grpc_pollset_worker *specific_worker) { + pollable *p = pollset->current_pollable; + if (p != &pollset->pollable) { + gpr_mu_lock(&p->po.mu); + } + grpc_error *error = pollset_kick_inner(pollset, p, specific_worker); + if (p != &pollset->pollable) { + gpr_mu_unlock(&p->po.mu); + } + return error; +} + +static grpc_error *kick_poller(void) { + return grpc_wakeup_fd_wakeup(&global_wakeup_fd); +} + +static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) { + pollable_init(&pollset->pollable, PO_POLLSET); + pollset->current_pollable = &g_empty_pollable; + pollset->kicked_without_poller = false; + pollset->shutdown_closure = NULL; + pollset->root_worker = NULL; + *mu = &pollset->pollable.po.mu; +} + +/* Convert a timespec to milliseconds: + - Very small or negative poll times are clamped to zero to do a non-blocking + poll (which becomes spin polling) + - Other small values are rounded up to one millisecond + - Longer than a millisecond polls are rounded up to the next nearest + millisecond to avoid spinning + - Infinite timeouts are converted to -1 */ +static int poll_deadline_to_millis_timeout(gpr_timespec deadline, + gpr_timespec now) { + gpr_timespec timeout; + if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) == 0) { + return -1; + } + + if (gpr_time_cmp(deadline, now) <= 0) { + return 0; + } + + static const gpr_timespec round_up = { + .clock_type = GPR_TIMESPAN, .tv_sec = 0, .tv_nsec = GPR_NS_PER_MS - 1}; + timeout = gpr_time_sub(deadline, now); + int millis = gpr_time_to_millis(gpr_time_add(timeout, round_up)); + return millis >= 1 ? millis : 1; +} + +static void fd_become_readable(grpc_exec_ctx *exec_ctx, grpc_fd *fd, + grpc_pollset *notifier) { + grpc_lfev_set_ready(exec_ctx, &fd->read_closure); + + /* Note, it is possible that fd_become_readable might be called twice with + different 'notifier's when an fd becomes readable and it is in two epoll + sets (This can happen briefly during polling island merges). In such cases + it does not really matter which notifer is set as the read_notifier_pollset + (They would both point to the same polling island anyway) */ + /* Use release store to match with acquire load in fd_get_read_notifier */ + gpr_atm_rel_store(&fd->read_notifier_pollset, (gpr_atm)notifier); +} + +static void fd_become_writable(grpc_exec_ctx *exec_ctx, grpc_fd *fd) { + grpc_lfev_set_ready(exec_ctx, &fd->write_closure); +} + +static grpc_error *fd_become_pollable_locked(grpc_fd *fd) { + grpc_error *error = GRPC_ERROR_NONE; + static const char *err_desc = "fd_become_pollable"; + if (append_error(&error, pollable_materialize(&fd->pollable), err_desc)) { + append_error(&error, pollable_add_fd(&fd->pollable, fd), err_desc); + } + return error; +} + +static void pollset_maybe_finish_shutdown(grpc_exec_ctx *exec_ctx, + grpc_pollset *pollset) { + if (pollset->shutdown_closure != NULL && pollset->root_worker == NULL) { + grpc_closure_sched(exec_ctx, pollset->shutdown_closure, GRPC_ERROR_NONE); + pollset->shutdown_closure = NULL; + } +} + +/* pollset->po.mu lock must be held by the caller before calling this */ +static void pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, + grpc_closure *closure) { + GPR_ASSERT(pollset->shutdown_closure == NULL); + pollset->shutdown_closure = closure; + GRPC_LOG_IF_ERROR("pollset_shutdown", pollset_kick_all(pollset)); + pollset_maybe_finish_shutdown(exec_ctx, pollset); +} + +static bool pollset_is_pollable_fd(grpc_pollset *pollset, pollable *p) { + return p != &g_empty_pollable && p != &pollset->pollable; +} + +/* pollset_shutdown is guaranteed to be called before pollset_destroy. */ +static void pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) { + pollable_destroy(&pollset->pollable); + if (pollset_is_pollable_fd(pollset, pollset->current_pollable)) { + UNREF_BY(exec_ctx, (grpc_fd *)pollset->current_pollable, 2, + "pollset_pollable"); + } +} + +#define MAX_EPOLL_EVENTS 100 + +static grpc_error *pollset_epoll(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, + pollable *p, gpr_timespec now, + gpr_timespec deadline) { + struct epoll_event events[MAX_EPOLL_EVENTS]; + static const char *err_desc = "pollset_poll"; + + int timeout = poll_deadline_to_millis_timeout(deadline, now); + + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p poll %p for %dms", pollset, p, timeout); + } + + if (timeout != 0) { + GRPC_SCHEDULING_START_BLOCKING_REGION; + } + int r; + do { + r = epoll_wait(p->epfd, events, MAX_EPOLL_EVENTS, timeout); + } while (r < 0 && errno == EINTR); + if (timeout != 0) { + GRPC_SCHEDULING_END_BLOCKING_REGION; + } + + if (r < 0) return GRPC_OS_ERROR(errno, "epoll_wait"); + + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p poll %p got %d events", pollset, p, r); + } + + grpc_error *error = GRPC_ERROR_NONE; + for (int i = 0; i < r; i++) { + void *data_ptr = events[i].data.ptr; + if (data_ptr == &global_wakeup_fd) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p poll %p got global_wakeup_fd", pollset, p); + } + + grpc_timer_consume_kick(); + append_error(&error, grpc_wakeup_fd_consume_wakeup(&global_wakeup_fd), + err_desc); + } else if (data_ptr == &p->wakeup) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p poll %p got pollset_wakeup", pollset, p); + } + append_error(&error, grpc_wakeup_fd_consume_wakeup(&p->wakeup), err_desc); + } else { + grpc_fd *fd = (grpc_fd *)(((intptr_t)data_ptr) & ~(intptr_t)1); + bool is_workqueue = (((intptr_t)data_ptr) & 1) != 0; + bool cancel = (events[i].events & (EPOLLERR | EPOLLHUP)) != 0; + bool read_ev = (events[i].events & (EPOLLIN | EPOLLPRI)) != 0; + bool write_ev = (events[i].events & EPOLLOUT) != 0; + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, + "PS:%p poll %p got fd %p(%d/%d): is_wq=%d cancel=%d read=%d " + "write=%d", + pollset, p, fd, fd->fd, fd->workqueue_wakeup_fd.read_fd, + is_workqueue, cancel, read_ev, write_ev); + } + if (is_workqueue) { + append_error(&error, + grpc_wakeup_fd_consume_wakeup(&fd->workqueue_wakeup_fd), + err_desc); + fd_invoke_workqueue(exec_ctx, fd); + } else { + if (read_ev || cancel) { + fd_become_readable(exec_ctx, fd, pollset); + } + if (write_ev || cancel) { + fd_become_writable(exec_ctx, fd); + } + } + } + } + + return error; +} + +/* Return true if first in list */ +static bool worker_insert(grpc_pollset_worker **root, pollset_worker_links link, + grpc_pollset_worker *worker) { + if (*root == NULL) { + *root = worker; + worker->links[link].next = worker->links[link].prev = worker; + return true; + } else { + worker->links[link].next = *root; + worker->links[link].prev = worker->links[link].next->links[link].prev; + worker->links[link].next->links[link].prev = worker; + worker->links[link].prev->links[link].next = worker; + return false; + } +} + +/* Return true if last in list */ +typedef enum { EMPTIED, NEW_ROOT, REMOVED } worker_remove_result; + +static worker_remove_result worker_remove(grpc_pollset_worker **root, + pollset_worker_links link, + grpc_pollset_worker *worker) { + if (worker == *root) { + if (worker == worker->links[link].next) { + *root = NULL; + return EMPTIED; + } else { + *root = worker->links[link].next; + worker->links[link].prev->links[link].next = worker->links[link].next; + worker->links[link].next->links[link].prev = worker->links[link].prev; + return NEW_ROOT; + } + } else { + worker->links[link].prev->links[link].next = worker->links[link].next; + worker->links[link].next->links[link].prev = worker->links[link].prev; + return REMOVED; + } +} + +/* Return true if this thread should poll */ +static bool begin_worker(grpc_pollset *pollset, grpc_pollset_worker *worker, + grpc_pollset_worker **worker_hdl, gpr_timespec *now, + gpr_timespec deadline) { + bool do_poll = true; + if (worker_hdl != NULL) *worker_hdl = worker; + worker->initialized_cv = false; + worker->kicked = false; + worker->pollset = pollset; + worker->pollable = pollset->current_pollable; + + if (pollset_is_pollable_fd(pollset, worker->pollable)) { + REF_BY((grpc_fd *)worker->pollable, 2, "one_poll"); + } + + worker_insert(&pollset->root_worker, PWL_POLLSET, worker); + if (!worker_insert(&worker->pollable->root_worker, PWL_POLLABLE, worker)) { + worker->initialized_cv = true; + gpr_cv_init(&worker->cv); + if (worker->pollable != &pollset->pollable) { + gpr_mu_unlock(&pollset->pollable.po.mu); + } + if (grpc_polling_trace && worker->pollable->root_worker != worker) { + gpr_log(GPR_DEBUG, "PS:%p wait %p w=%p for %dms", pollset, + worker->pollable, worker, + poll_deadline_to_millis_timeout(deadline, *now)); + } + while (do_poll && worker->pollable->root_worker != worker) { + if (gpr_cv_wait(&worker->cv, &worker->pollable->po.mu, deadline)) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p timeout_wait %p w=%p", pollset, + worker->pollable, worker); + } + do_poll = false; + } else if (worker->kicked) { + if (grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p wakeup %p w=%p", pollset, worker->pollable, + worker); + } + do_poll = false; + } else if (grpc_polling_trace && + worker->pollable->root_worker != worker) { + gpr_log(GPR_DEBUG, "PS:%p spurious_wakeup %p w=%p", pollset, + worker->pollable, worker); + } + } + if (worker->pollable != &pollset->pollable) { + gpr_mu_unlock(&worker->pollable->po.mu); + gpr_mu_lock(&pollset->pollable.po.mu); + gpr_mu_lock(&worker->pollable->po.mu); + } + *now = gpr_now(now->clock_type); + } + + return do_poll && pollset->shutdown_closure == NULL && + pollset->current_pollable == worker->pollable; +} + +static void end_worker(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, + grpc_pollset_worker *worker, + grpc_pollset_worker **worker_hdl) { + if (NEW_ROOT == + worker_remove(&worker->pollable->root_worker, PWL_POLLABLE, worker)) { + gpr_cv_signal(&worker->pollable->root_worker->cv); + } + if (worker->initialized_cv) { + gpr_cv_destroy(&worker->cv); + } + if (pollset_is_pollable_fd(pollset, worker->pollable)) { + UNREF_BY(exec_ctx, (grpc_fd *)worker->pollable, 2, "one_poll"); + } + if (EMPTIED == worker_remove(&pollset->root_worker, PWL_POLLSET, worker)) { + pollset_maybe_finish_shutdown(exec_ctx, pollset); + } +} + +/* pollset->po.mu lock must be held by the caller before calling this. + The function pollset_work() may temporarily release the lock (pollset->po.mu) + during the course of its execution but it will always re-acquire the lock and + ensure that it is held by the time the function returns */ +static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, + grpc_pollset_worker **worker_hdl, + gpr_timespec now, gpr_timespec deadline) { + grpc_pollset_worker worker; + if (0 && grpc_polling_trace) { + gpr_log(GPR_DEBUG, "PS:%p work hdl=%p worker=%p now=%" PRId64 + ".%09d deadline=%" PRId64 ".%09d kwp=%d root_worker=%p", + pollset, worker_hdl, &worker, now.tv_sec, now.tv_nsec, + deadline.tv_sec, deadline.tv_nsec, pollset->kicked_without_poller, + pollset->root_worker); + } + grpc_error *error = GRPC_ERROR_NONE; + static const char *err_desc = "pollset_work"; + if (pollset->kicked_without_poller) { + pollset->kicked_without_poller = false; + return GRPC_ERROR_NONE; + } + if (pollset->current_pollable != &pollset->pollable) { + gpr_mu_lock(&pollset->current_pollable->po.mu); + } + if (begin_worker(pollset, &worker, worker_hdl, &now, deadline)) { + gpr_tls_set(&g_current_thread_pollset, (intptr_t)pollset); + gpr_tls_set(&g_current_thread_worker, (intptr_t)&worker); + GPR_ASSERT(!pollset->shutdown_closure); + append_error(&error, pollable_materialize(worker.pollable), err_desc); + if (worker.pollable != &pollset->pollable) { + gpr_mu_unlock(&worker.pollable->po.mu); + } + gpr_mu_unlock(&pollset->pollable.po.mu); + append_error(&error, pollset_epoll(exec_ctx, pollset, worker.pollable, now, + deadline), + err_desc); + grpc_exec_ctx_flush(exec_ctx); + gpr_mu_lock(&pollset->pollable.po.mu); + if (worker.pollable != &pollset->pollable) { + gpr_mu_lock(&worker.pollable->po.mu); + } + gpr_tls_set(&g_current_thread_pollset, 0); + gpr_tls_set(&g_current_thread_worker, 0); + pollset_maybe_finish_shutdown(exec_ctx, pollset); + } + end_worker(exec_ctx, pollset, &worker, worker_hdl); + if (worker.pollable != &pollset->pollable) { + gpr_mu_unlock(&worker.pollable->po.mu); + } + return error; +} + +static void unref_fd_no_longer_poller(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_fd *fd = arg; + UNREF_BY(exec_ctx, fd, 2, "pollset_pollable"); +} + +/* expects pollsets locked, flag whether fd is locked or not */ +static grpc_error *pollset_add_fd_locked(grpc_exec_ctx *exec_ctx, + grpc_pollset *pollset, grpc_fd *fd, + bool fd_locked) { + static const char *err_desc = "pollset_add_fd"; + grpc_error *error = GRPC_ERROR_NONE; + if (pollset->current_pollable == &g_empty_pollable) { + /* empty pollable --> single fd pollable */ + append_error(&error, pollset_kick_all(pollset), err_desc); + pollset->current_pollable = &fd->pollable; + if (!fd_locked) gpr_mu_lock(&fd->pollable.po.mu); + append_error(&error, fd_become_pollable_locked(fd), err_desc); + if (!fd_locked) gpr_mu_unlock(&fd->pollable.po.mu); + REF_BY(fd, 2, "pollset_pollable"); + } else if (pollset->current_pollable == &pollset->pollable) { + append_error(&error, pollable_add_fd(pollset->current_pollable, fd), + err_desc); + } else if (pollset->current_pollable != &fd->pollable) { + grpc_fd *had_fd = (grpc_fd *)pollset->current_pollable; + pollset->current_pollable = &pollset->pollable; + if (append_error(&error, pollable_materialize(&pollset->pollable), + err_desc)) { + pollable_add_fd(&pollset->pollable, had_fd); + pollable_add_fd(&pollset->pollable, fd); + } + grpc_closure_sched(exec_ctx, + grpc_closure_create(unref_fd_no_longer_poller, had_fd, + grpc_schedule_on_exec_ctx), + GRPC_ERROR_NONE); + } + return error; +} + +static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, + grpc_fd *fd) { + gpr_mu_lock(&pollset->pollable.po.mu); + grpc_error *error = pollset_add_fd_locked(exec_ctx, pollset, fd, false); + gpr_mu_unlock(&pollset->pollable.po.mu); + GRPC_LOG_IF_ERROR("pollset_add_fd", error); +} + +/******************************************************************************* + * Pollset-set Definitions + */ + +static grpc_pollset_set *pollset_set_create(void) { + grpc_pollset_set *pss = gpr_zalloc(sizeof(*pss)); + po_init(&pss->po, PO_POLLSET_SET); + return pss; +} + +static void pollset_set_destroy(grpc_exec_ctx *exec_ctx, + grpc_pollset_set *pss) { + po_destroy(&pss->po); + gpr_free(pss); +} + +static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss, + grpc_fd *fd) { + po_join(exec_ctx, &pss->po, &fd->pollable.po); +} + +static void pollset_set_del_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss, + grpc_fd *fd) {} + +static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx, + grpc_pollset_set *pss, grpc_pollset *ps) { + po_join(exec_ctx, &pss->po, &ps->pollable.po); +} + +static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx, + grpc_pollset_set *pss, grpc_pollset *ps) {} + +static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx, + grpc_pollset_set *bag, + grpc_pollset_set *item) { + po_join(exec_ctx, &bag->po, &item->po); +} + +static void pollset_set_del_pollset_set(grpc_exec_ctx *exec_ctx, + grpc_pollset_set *bag, + grpc_pollset_set *item) {} + +static void po_init(polling_obj *po, polling_obj_type type) { + gpr_mu_init(&po->mu); + po->type = type; + po->group = NULL; + po->next = po; + po->prev = po; +} + +static polling_group *pg_lock_latest(polling_group *pg) { + /* assumes pg unlocked; consumes ref, returns ref */ + gpr_mu_lock(&pg->po.mu); + while (pg->po.group != NULL) { + polling_group *new_pg = pg_ref(pg->po.group); + gpr_mu_unlock(&pg->po.mu); + pg_unref(pg); + pg = new_pg; + gpr_mu_lock(&pg->po.mu); + } + return pg; +} + +static void po_destroy(polling_obj *po) { + if (po->group != NULL) { + polling_group *pg = pg_lock_latest(po->group); + po->prev->next = po->next; + po->next->prev = po->prev; + gpr_mu_unlock(&pg->po.mu); + pg_unref(pg); + } + gpr_mu_destroy(&po->mu); +} + +static polling_group *pg_ref(polling_group *pg) { + gpr_ref(&pg->refs); + return pg; +} + +static void pg_unref(polling_group *pg) { + if (gpr_unref(&pg->refs)) { + po_destroy(&pg->po); + gpr_free(pg); + } +} + +static int po_cmp(polling_obj *a, polling_obj *b) { + if (a == b) return 0; + if (a->type < b->type) return -1; + if (a->type > b->type) return 1; + if (a < b) return -1; + assert(a > b); + return 1; +} + +static void po_join(grpc_exec_ctx *exec_ctx, polling_obj *a, polling_obj *b) { + switch (po_cmp(a, b)) { + case 0: + return; + case 1: + GPR_SWAP(polling_obj *, a, b); + /* fall through */ + case -1: + gpr_mu_lock(&a->mu); + gpr_mu_lock(&b->mu); + + if (a->group == NULL) { + if (b->group == NULL) { + polling_obj *initial_po[] = {a, b}; + pg_create(exec_ctx, initial_po, GPR_ARRAY_SIZE(initial_po)); + gpr_mu_unlock(&a->mu); + gpr_mu_unlock(&b->mu); + } else { + polling_group *b_group = pg_ref(b->group); + gpr_mu_unlock(&b->mu); + gpr_mu_unlock(&a->mu); + pg_join(exec_ctx, b_group, a); + } + } else if (b->group == NULL) { + polling_group *a_group = pg_ref(a->group); + gpr_mu_unlock(&a->mu); + gpr_mu_unlock(&b->mu); + pg_join(exec_ctx, a_group, b); + } else if (a->group == b->group) { + /* nothing to do */ + gpr_mu_unlock(&a->mu); + gpr_mu_unlock(&b->mu); + } else { + polling_group *a_group = pg_ref(a->group); + polling_group *b_group = pg_ref(b->group); + gpr_mu_unlock(&a->mu); + gpr_mu_unlock(&b->mu); + pg_merge(exec_ctx, a_group, b_group); + } + } +} + +static void pg_notify(grpc_exec_ctx *exec_ctx, polling_obj *a, polling_obj *b) { + if (a->type == PO_FD && b->type == PO_POLLSET) { + pollset_add_fd_locked(exec_ctx, (grpc_pollset *)b, (grpc_fd *)a, true); + } else if (a->type == PO_POLLSET && b->type == PO_FD) { + pollset_add_fd_locked(exec_ctx, (grpc_pollset *)a, (grpc_fd *)b, true); + } +} + +static void pg_broadcast(grpc_exec_ctx *exec_ctx, polling_group *from, + polling_group *to) { + for (polling_obj *a = from->po.next; a != &from->po; a = a->next) { + for (polling_obj *b = to->po.next; b != &to->po; b = b->next) { + if (po_cmp(a, b) < 0) { + gpr_mu_lock(&a->mu); + gpr_mu_lock(&b->mu); + } else { + GPR_ASSERT(po_cmp(a, b) != 0); + gpr_mu_lock(&b->mu); + gpr_mu_lock(&a->mu); + } + pg_notify(exec_ctx, a, b); + gpr_mu_unlock(&a->mu); + gpr_mu_unlock(&b->mu); + } + } +} + +static void pg_create(grpc_exec_ctx *exec_ctx, polling_obj **initial_po, + size_t initial_po_count) { + /* assumes all polling objects in initial_po are locked */ + polling_group *pg = gpr_malloc(sizeof(*pg)); + po_init(&pg->po, PO_POLLING_GROUP); + gpr_ref_init(&pg->refs, (int)initial_po_count); + for (size_t i = 0; i < initial_po_count; i++) { + GPR_ASSERT(initial_po[i]->group == NULL); + initial_po[i]->group = pg; + } + for (size_t i = 1; i < initial_po_count; i++) { + initial_po[i]->prev = initial_po[i - 1]; + } + for (size_t i = 0; i < initial_po_count - 1; i++) { + initial_po[i]->next = initial_po[i + 1]; + } + initial_po[0]->prev = &pg->po; + initial_po[initial_po_count - 1]->next = &pg->po; + pg->po.next = initial_po[0]; + pg->po.prev = initial_po[initial_po_count - 1]; + for (size_t i = 1; i < initial_po_count; i++) { + for (size_t j = 0; j < i; j++) { + pg_notify(exec_ctx, initial_po[i], initial_po[j]); + } + } +} + +static void pg_join(grpc_exec_ctx *exec_ctx, polling_group *pg, + polling_obj *po) { + /* assumes neither pg nor po are locked; consumes one ref to pg */ + pg = pg_lock_latest(pg); + /* pg locked */ + for (polling_obj *existing = pg->po.next /* skip pg - it's just a stub */; + existing != &pg->po; existing = existing->next) { + if (po_cmp(po, existing) < 0) { + gpr_mu_lock(&po->mu); + gpr_mu_lock(&existing->mu); + } else { + GPR_ASSERT(po_cmp(po, existing) != 0); + gpr_mu_lock(&existing->mu); + gpr_mu_lock(&po->mu); + } + /* pg, po, existing locked */ + if (po->group != NULL) { + gpr_mu_unlock(&pg->po.mu); + polling_group *po_group = pg_ref(po->group); + gpr_mu_unlock(&po->mu); + gpr_mu_unlock(&existing->mu); + pg_merge(exec_ctx, pg, po_group); + /* early exit: polling obj picked up a group during joining: we needed + to do a full merge */ + return; + } + pg_notify(exec_ctx, po, existing); + gpr_mu_unlock(&po->mu); + gpr_mu_unlock(&existing->mu); + } + gpr_mu_lock(&po->mu); + if (po->group != NULL) { + gpr_mu_unlock(&pg->po.mu); + polling_group *po_group = pg_ref(po->group); + gpr_mu_unlock(&po->mu); + pg_merge(exec_ctx, pg, po_group); + /* early exit: polling obj picked up a group during joining: we needed + to do a full merge */ + return; + } + po->group = pg; + po->next = &pg->po; + po->prev = pg->po.prev; + po->prev->next = po->next->prev = po; + gpr_mu_unlock(&pg->po.mu); + gpr_mu_unlock(&po->mu); +} + +static void pg_merge(grpc_exec_ctx *exec_ctx, polling_group *a, + polling_group *b) { + for (;;) { + if (a == b) { + pg_unref(a); + pg_unref(b); + return; + } + if (a > b) GPR_SWAP(polling_group *, a, b); + gpr_mu_lock(&a->po.mu); + gpr_mu_lock(&b->po.mu); + if (a->po.group != NULL) { + polling_group *m2 = pg_ref(a->po.group); + gpr_mu_unlock(&a->po.mu); + gpr_mu_unlock(&b->po.mu); + pg_unref(a); + a = m2; + } else if (b->po.group != NULL) { + polling_group *m2 = pg_ref(b->po.group); + gpr_mu_unlock(&a->po.mu); + gpr_mu_unlock(&b->po.mu); + pg_unref(b); + b = m2; + } else { + break; + } + } + polling_group **unref = NULL; + size_t unref_count = 0; + size_t unref_cap = 0; + b->po.group = a; + pg_broadcast(exec_ctx, a, b); + pg_broadcast(exec_ctx, b, a); + while (b->po.next != &b->po) { + polling_obj *po = b->po.next; + gpr_mu_lock(&po->mu); + if (unref_count == unref_cap) { + unref_cap = GPR_MAX(8, 3 * unref_cap / 2); + unref = gpr_realloc(unref, unref_cap * sizeof(*unref)); + } + unref[unref_count++] = po->group; + po->group = pg_ref(a); + // unlink from b + po->prev->next = po->next; + po->next->prev = po->prev; + // link to a + po->next = &a->po; + po->prev = a->po.prev; + po->next->prev = po->prev->next = po; + gpr_mu_unlock(&po->mu); + } + gpr_mu_unlock(&a->po.mu); + gpr_mu_unlock(&b->po.mu); + for (size_t i = 0; i < unref_count; i++) { + pg_unref(unref[i]); + } + gpr_free(unref); + pg_unref(b); +} + +/******************************************************************************* + * Event engine binding + */ + +static void shutdown_engine(void) { + fd_global_shutdown(); + pollset_global_shutdown(); +} + +static const grpc_event_engine_vtable vtable = { + .pollset_size = sizeof(grpc_pollset), + + .fd_create = fd_create, + .fd_wrapped_fd = fd_wrapped_fd, + .fd_orphan = fd_orphan, + .fd_shutdown = fd_shutdown, + .fd_is_shutdown = fd_is_shutdown, + .fd_notify_on_read = fd_notify_on_read, + .fd_notify_on_write = fd_notify_on_write, + .fd_get_read_notifier_pollset = fd_get_read_notifier_pollset, + .fd_get_workqueue = fd_get_workqueue, + + .pollset_init = pollset_init, + .pollset_shutdown = pollset_shutdown, + .pollset_destroy = pollset_destroy, + .pollset_work = pollset_work, + .pollset_kick = pollset_kick, + .pollset_add_fd = pollset_add_fd, + + .pollset_set_create = pollset_set_create, + .pollset_set_destroy = pollset_set_destroy, + .pollset_set_add_pollset = pollset_set_add_pollset, + .pollset_set_del_pollset = pollset_set_del_pollset, + .pollset_set_add_pollset_set = pollset_set_add_pollset_set, + .pollset_set_del_pollset_set = pollset_set_del_pollset_set, + .pollset_set_add_fd = pollset_set_add_fd, + .pollset_set_del_fd = pollset_set_del_fd, + + .kick_poller = kick_poller, + + .workqueue_ref = workqueue_ref, + .workqueue_unref = workqueue_unref, + .workqueue_scheduler = workqueue_scheduler, + + .shutdown_engine = shutdown_engine, +}; + +/* It is possible that GLIBC has epoll but the underlying kernel doesn't. + * Create a dummy epoll_fd to make sure epoll support is available */ +static bool is_epollex_available(void) { + static bool logged_why_not = false; + + int fd = epoll_create1(EPOLL_CLOEXEC); + if (fd < 0) { + if (!logged_why_not) { + gpr_log(GPR_ERROR, + "epoll_create1 failed with error: %d. Not using epollex polling " + "engine.", + fd); + logged_why_not = true; + } + return false; + } + grpc_wakeup_fd wakeup; + if (!GRPC_LOG_IF_ERROR("check_wakeupfd_for_epollex", + grpc_wakeup_fd_init(&wakeup))) { + return false; + } + struct epoll_event ev = { + /* choose events that should cause an error on + EPOLLEXCLUSIVE enabled kernels - specifically the combination of + EPOLLONESHOT and EPOLLEXCLUSIVE */ + .events = EPOLLET | EPOLLIN | EPOLLEXCLUSIVE | EPOLLONESHOT, + .data.ptr = NULL}; + if (epoll_ctl(fd, EPOLL_CTL_ADD, wakeup.read_fd, &ev) != 0) { + if (errno != EINVAL) { + if (!logged_why_not) { + gpr_log( + GPR_ERROR, + "epoll_ctl with EPOLLEXCLUSIVE | EPOLLONESHOT failed with error: " + "%d. Not using epollex polling engine.", + errno); + logged_why_not = true; + } + close(fd); + grpc_wakeup_fd_destroy(&wakeup); + return false; + } + } else { + if (!logged_why_not) { + gpr_log(GPR_ERROR, + "epoll_ctl with EPOLLEXCLUSIVE | EPOLLONESHOT succeeded. This is " + "evidence of no EPOLLEXCLUSIVE support. Not using " + "epollex polling engine."); + logged_why_not = true; + } + close(fd); + grpc_wakeup_fd_destroy(&wakeup); + return false; + } + grpc_wakeup_fd_destroy(&wakeup); + close(fd); + return true; +} + +const grpc_event_engine_vtable *grpc_init_epollex_linux(void) { + if (!grpc_has_wakeup_fd()) { + return NULL; + } + + if (!is_epollex_available()) { + return NULL; + } + + fd_global_init(); + + if (!GRPC_LOG_IF_ERROR("pollset_global_init", pollset_global_init())) { + pollset_global_shutdown(); + fd_global_shutdown(); + return NULL; + } + + return &vtable; +} + +#else /* defined(GRPC_LINUX_EPOLL) */ +#if defined(GRPC_POSIX_SOCKET) +#include "src/core/lib/iomgr/ev_posix.h" +/* If GRPC_LINUX_EPOLL is not defined, it means epoll is not available. Return + * NULL */ +const grpc_event_engine_vtable *grpc_init_epollex_linux(void) { return NULL; } +#endif /* defined(GRPC_POSIX_SOCKET) */ + +#endif /* !defined(GRPC_LINUX_EPOLL) */ diff --git a/src/core/lib/iomgr/ev_epollex_linux.h b/src/core/lib/iomgr/ev_epollex_linux.h new file mode 100644 index 0000000000..4d7cd713df --- /dev/null +++ b/src/core/lib/iomgr/ev_epollex_linux.h @@ -0,0 +1,42 @@ +/* + * + * 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_CORE_LIB_IOMGR_EV_EPOLLEX_LINUX_H +#define GRPC_CORE_LIB_IOMGR_EV_EPOLLEX_LINUX_H + +#include "src/core/lib/iomgr/ev_posix.h" +#include "src/core/lib/iomgr/port.h" + +const grpc_event_engine_vtable *grpc_init_epollex_linux(void); + +#endif /* GRPC_CORE_LIB_IOMGR_EV_EPOLLEX_LINUX_H */ diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c index 9834cdd197..a83bad319a 100644 --- a/src/core/lib/iomgr/ev_poll_posix.c +++ b/src/core/lib/iomgr/ev_poll_posix.c @@ -58,6 +58,8 @@ #include "src/core/lib/profiling/timers.h" #include "src/core/lib/support/block_annotate.h" +#define GRPC_POLLSET_KICK_BROADCAST ((grpc_pollset_worker *)1) + /******************************************************************************* * FD declarations */ @@ -815,7 +817,7 @@ static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) { pollset->pollset_set_count = 0; } -static void pollset_destroy(grpc_pollset *pollset) { +static void pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) { GPR_ASSERT(!pollset_has_workers(pollset)); GPR_ASSERT(pollset->idle_jobs.head == pollset->idle_jobs.tail); while (pollset->local_wakeup_cache) { diff --git a/src/core/lib/iomgr/ev_posix.c b/src/core/lib/iomgr/ev_posix.c index 13409a4de8..a01de93e9b 100644 --- a/src/core/lib/iomgr/ev_posix.c +++ b/src/core/lib/iomgr/ev_posix.c @@ -44,10 +44,14 @@ #include <grpc/support/string_util.h> #include <grpc/support/useful.h> +#include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/ev_epoll_linux.h" +#include "src/core/lib/iomgr/ev_epollex_linux.h" #include "src/core/lib/iomgr/ev_poll_posix.h" #include "src/core/lib/support/env.h" +int grpc_polling_trace = 0; /* Disabled by default */ + /** Default poll() function - a pointer so that it can be overridden by some * tests */ grpc_poll_function_type grpc_poll_function = poll; @@ -65,6 +69,7 @@ typedef struct { } event_engine_factory; static const event_engine_factory g_factories[] = { + {"epollex", grpc_init_epollex_linux}, {"epoll", grpc_init_epoll_linux}, {"poll", grpc_init_poll_posix}, {"poll-cv", grpc_init_poll_cv_posix}, @@ -121,6 +126,8 @@ void grpc_set_event_engine_test_only( const char *grpc_get_poll_strategy_name() { return g_poll_strategy_name; } void grpc_event_engine_init(void) { + grpc_register_tracer("polling", &grpc_polling_trace); + char *s = gpr_getenv("GRPC_POLL_STRATEGY"); if (s == NULL) { s = gpr_strdup("all"); @@ -197,8 +204,8 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, g_event_engine->pollset_shutdown(exec_ctx, pollset, closure); } -void grpc_pollset_destroy(grpc_pollset *pollset) { - g_event_engine->pollset_destroy(pollset); +void grpc_pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) { + g_event_engine->pollset_destroy(exec_ctx, pollset); } grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, diff --git a/src/core/lib/iomgr/ev_posix.h b/src/core/lib/iomgr/ev_posix.h index becc4d359e..59832aaf24 100644 --- a/src/core/lib/iomgr/ev_posix.h +++ b/src/core/lib/iomgr/ev_posix.h @@ -42,6 +42,8 @@ #include "src/core/lib/iomgr/wakeup_fd_posix.h" #include "src/core/lib/iomgr/workqueue.h" +extern int grpc_polling_trace; /* Disabled by default */ + typedef struct grpc_fd grpc_fd; typedef struct grpc_event_engine_vtable { @@ -64,7 +66,7 @@ typedef struct grpc_event_engine_vtable { void (*pollset_init)(grpc_pollset *pollset, gpr_mu **mu); void (*pollset_shutdown)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_closure *closure); - void (*pollset_destroy)(grpc_pollset *pollset); + void (*pollset_destroy)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset); grpc_error *(*pollset_work)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_pollset_worker **worker, gpr_timespec now, gpr_timespec deadline); diff --git a/src/core/lib/iomgr/pollset.h b/src/core/lib/iomgr/pollset.h index 9bf3cdac89..69e20098d7 100644 --- a/src/core/lib/iomgr/pollset.h +++ b/src/core/lib/iomgr/pollset.h @@ -40,8 +40,6 @@ #include "src/core/lib/iomgr/exec_ctx.h" -#define GRPC_POLLSET_KICK_BROADCAST ((grpc_pollset_worker *)1) - /* A grpc_pollset is a set of file descriptors that a higher level item is interested in. For example: - a server will typically keep a pollset containing all connected channels, @@ -59,7 +57,7 @@ void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu); * pollset's mutex must be held */ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_closure *closure); -void grpc_pollset_destroy(grpc_pollset *pollset); +void grpc_pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset); /* Do some work on a pollset. May involve invoking asynchronous callbacks, or actually polling file @@ -88,8 +86,7 @@ grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, gpr_timespec deadline) GRPC_MUST_USE_RESULT; /* Break one polling thread out of polling work for this pollset. - If specific_worker is GRPC_POLLSET_KICK_BROADCAST, kick ALL the workers. - Otherwise, if specific_worker is non-NULL, then kick that worker. */ + If specific_worker is non-NULL, then kick that worker. */ grpc_error *grpc_pollset_kick(grpc_pollset *pollset, grpc_pollset_worker *specific_worker) GRPC_MUST_USE_RESULT; diff --git a/src/core/lib/iomgr/pollset_windows.c b/src/core/lib/iomgr/pollset_windows.c index 04c6b71747..6dca37b481 100644 --- a/src/core/lib/iomgr/pollset_windows.c +++ b/src/core/lib/iomgr/pollset_windows.c @@ -43,6 +43,8 @@ #include "src/core/lib/iomgr/pollset.h" #include "src/core/lib/iomgr/pollset_windows.h" +#define GRPC_POLLSET_KICK_BROADCAST ((grpc_pollset_worker *)1) + gpr_mu grpc_polling_mu; static grpc_pollset_worker *g_active_poller; static grpc_pollset_worker g_global_root_worker; diff --git a/src/core/lib/security/credentials/google_default/google_default_credentials.c b/src/core/lib/security/credentials/google_default/google_default_credentials.c index 97501e6788..4d8c451ea8 100644 --- a/src/core/lib/security/credentials/google_default/google_default_credentials.c +++ b/src/core/lib/security/credentials/google_default/google_default_credentials.c @@ -99,7 +99,7 @@ static void on_compute_engine_detection_http_response(grpc_exec_ctx *exec_ctx, } static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *e) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } static int is_stack_running_on_compute_engine(grpc_exec_ctx *exec_ctx) { diff --git a/src/core/lib/surface/alarm.c b/src/core/lib/surface/alarm.c index e71c0ebfc5..b72d534b7e 100644 --- a/src/core/lib/surface/alarm.c +++ b/src/core/lib/surface/alarm.c @@ -81,7 +81,9 @@ void grpc_alarm_cancel(grpc_alarm *alarm) { } void grpc_alarm_destroy(grpc_alarm *alarm) { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_alarm_cancel(alarm); - GRPC_CQ_INTERNAL_UNREF(alarm->cq, "alarm"); + GRPC_CQ_INTERNAL_UNREF(&exec_ctx, alarm->cq, "alarm"); gpr_free(alarm); + grpc_exec_ctx_finish(&exec_ctx); } diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index 9aa457d792..2581c1c71f 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -520,7 +520,7 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, } } if (c->cq) { - GRPC_CQ_INTERNAL_UNREF(c->cq, "bind"); + GRPC_CQ_INTERNAL_UNREF(exec_ctx, c->cq, "bind"); } get_final_status(call, set_status_value_directly, &c->final_info.final_status, diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index 35e9f7eb30..366842d121 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -182,20 +182,21 @@ void grpc_cq_internal_ref(grpc_completion_queue *cc) { static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { grpc_completion_queue *cc = arg; - GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy"); + GRPC_CQ_INTERNAL_UNREF(exec_ctx, cc, "pollset_destroy"); } #ifdef GRPC_CQ_REF_COUNT_DEBUG -void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason, - const char *file, int line) { +void grpc_cq_internal_unref(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, + const char *reason, const char *file, int line) { gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p unref %d -> %d %s", cc, (int)cc->owning_refs.count, (int)cc->owning_refs.count - 1, reason); #else -void grpc_cq_internal_unref(grpc_completion_queue *cc) { +void grpc_cq_internal_unref(grpc_exec_ctx *exec_ctx, + grpc_completion_queue *cc) { #endif if (gpr_unref(&cc->owning_refs)) { GPR_ASSERT(cc->completed_head.next == (uintptr_t)&cc->completed_head); - grpc_pollset_destroy(POLLSET_FROM_CQ(cc)); + grpc_pollset_destroy(exec_ctx, POLLSET_FROM_CQ(cc)); #ifndef NDEBUG gpr_free(cc->outstanding_tags); #endif @@ -469,7 +470,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, is_finished_arg.first_loop = false; } GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); - GRPC_CQ_INTERNAL_UNREF(cc, "next"); + GRPC_CQ_INTERNAL_UNREF(&exec_ctx, cc, "next"); grpc_exec_ctx_finish(&exec_ctx); GPR_ASSERT(is_finished_arg.stolen_completion == NULL); @@ -664,7 +665,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, } done: GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); - GRPC_CQ_INTERNAL_UNREF(cc, "pluck"); + GRPC_CQ_INTERNAL_UNREF(&exec_ctx, cc, "pluck"); grpc_exec_ctx_finish(&exec_ctx); GPR_ASSERT(is_finished_arg.stolen_completion == NULL); @@ -701,7 +702,9 @@ void grpc_completion_queue_destroy(grpc_completion_queue *cc) { GRPC_API_TRACE("grpc_completion_queue_destroy(cc=%p)", 1, (cc)); GPR_TIMER_BEGIN("grpc_completion_queue_destroy", 0); grpc_completion_queue_shutdown(cc); - GRPC_CQ_INTERNAL_UNREF(cc, "destroy"); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + GRPC_CQ_INTERNAL_UNREF(&exec_ctx, cc, "destroy"); + grpc_exec_ctx_finish(&exec_ctx); GPR_TIMER_END("grpc_completion_queue_destroy", 0); } diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index 1ff3d64293..a2e7b2d6a4 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -65,17 +65,17 @@ typedef struct grpc_cq_completion { #ifdef GRPC_CQ_REF_COUNT_DEBUG void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason, const char *file, int line); -void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason, - const char *file, int line); +void grpc_cq_internal_unref(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, + const char *reason, const char *file, int line); #define GRPC_CQ_INTERNAL_REF(cc, reason) \ grpc_cq_internal_ref(cc, reason, __FILE__, __LINE__) -#define GRPC_CQ_INTERNAL_UNREF(cc, reason) \ - grpc_cq_internal_unref(cc, reason, __FILE__, __LINE__) +#define GRPC_CQ_INTERNAL_UNREF(ec, cc, reason) \ + grpc_cq_internal_unref(ec, cc, reason, __FILE__, __LINE__) #else void grpc_cq_internal_ref(grpc_completion_queue *cc); -void grpc_cq_internal_unref(grpc_completion_queue *cc); +void grpc_cq_internal_unref(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc); #define GRPC_CQ_INTERNAL_REF(cc, reason) grpc_cq_internal_ref(cc) -#define GRPC_CQ_INTERNAL_UNREF(cc, reason) grpc_cq_internal_unref(cc) +#define GRPC_CQ_INTERNAL_UNREF(ec, cc, reason) grpc_cq_internal_unref(ec, cc) #endif /* Flag that an operation is beginning: the completion channel will not finish diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index 26c81e9aca..ef14124fef 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -408,7 +408,7 @@ static void server_delete(grpc_exec_ctx *exec_ctx, grpc_server *server) { request_matcher_destroy(&server->unregistered_request_matcher); } for (i = 0; i < server->cq_count; i++) { - GRPC_CQ_INTERNAL_UNREF(server->cqs[i], "server"); + GRPC_CQ_INTERNAL_UNREF(exec_ctx, server->cqs[i], "server"); if (server->started) { gpr_stack_lockfree_destroy(server->request_freelist_per_cq[i]); gpr_free(server->requested_calls_per_cq[i]); diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index 16bb32bcc6..d3af7cace3 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -98,6 +98,7 @@ CORE_SOURCE_FILES = [ 'src/core/lib/iomgr/endpoint_pair_windows.c', 'src/core/lib/iomgr/error.c', 'src/core/lib/iomgr/ev_epoll_linux.c', + 'src/core/lib/iomgr/ev_epollex_linux.c', 'src/core/lib/iomgr/ev_poll_posix.c', 'src/core/lib/iomgr/ev_posix.c', 'src/core/lib/iomgr/exec_ctx.c', diff --git a/test/core/end2end/fixtures/http_proxy_fixture.c b/test/core/end2end/fixtures/http_proxy_fixture.c index 451ed268d3..259abfd151 100644 --- a/test/core/end2end/fixtures/http_proxy_fixture.c +++ b/test/core/end2end/fixtures/http_proxy_fixture.c @@ -469,7 +469,7 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create(void) { static void destroy_pollset(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { grpc_pollset* pollset = arg; - grpc_pollset_destroy(pollset); + grpc_pollset_destroy(exec_ctx, pollset); gpr_free(pollset); } diff --git a/test/core/http/httpcli_test.c b/test/core/http/httpcli_test.c index d3b45c4505..21135ddf6b 100644 --- a/test/core/http/httpcli_test.c +++ b/test/core/http/httpcli_test.c @@ -155,7 +155,7 @@ static void test_post(int port) { } static void destroy_pops(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(grpc_polling_entity_pollset(p)); + grpc_pollset_destroy(exec_ctx, grpc_polling_entity_pollset(p)); } int main(int argc, char **argv) { diff --git a/test/core/http/httpscli_test.c b/test/core/http/httpscli_test.c index acc94091f4..73eaae87d7 100644 --- a/test/core/http/httpscli_test.c +++ b/test/core/http/httpscli_test.c @@ -157,7 +157,7 @@ static void test_post(int port) { } static void destroy_pops(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(grpc_polling_entity_pollset(p)); + grpc_pollset_destroy(exec_ctx, grpc_polling_entity_pollset(p)); } int main(int argc, char **argv) { diff --git a/test/core/iomgr/endpoint_pair_test.c b/test/core/iomgr/endpoint_pair_test.c index c8a60776b9..4561c3846e 100644 --- a/test/core/iomgr/endpoint_pair_test.c +++ b/test/core/iomgr/endpoint_pair_test.c @@ -70,7 +70,7 @@ static grpc_endpoint_test_config configs[] = { static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } int main(int argc, char **argv) { diff --git a/test/core/iomgr/ev_epoll_linux_test.c b/test/core/iomgr/ev_epoll_linux_test.c index 0856023b14..7693b5a398 100644 --- a/test/core/iomgr/ev_epoll_linux_test.c +++ b/test/core/iomgr/ev_epoll_linux_test.c @@ -113,7 +113,7 @@ static void test_pollset_init(test_pollset *pollsets, int num_pollsets) { static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } static void test_pollset_cleanup(grpc_exec_ctx *exec_ctx, diff --git a/test/core/iomgr/fd_posix_test.c b/test/core/iomgr/fd_posix_test.c index 81d2692a08..d6cee8e43a 100644 --- a/test/core/iomgr/fd_posix_test.c +++ b/test/core/iomgr/fd_posix_test.c @@ -535,7 +535,7 @@ static void test_grpc_fd_change(void) { static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } int main(int argc, char **argv) { diff --git a/test/core/iomgr/pollset_set_test.c b/test/core/iomgr/pollset_set_test.c index 3a9d459579..469551c827 100644 --- a/test/core/iomgr/pollset_set_test.c +++ b/test/core/iomgr/pollset_set_test.c @@ -86,7 +86,7 @@ static void init_test_pollsets(test_pollset *pollsets, const int num_pollsets) { static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } static void cleanup_test_pollsets(grpc_exec_ctx *exec_ctx, diff --git a/test/core/iomgr/resolve_address_posix_test.c b/test/core/iomgr/resolve_address_posix_test.c index fa88aca431..7fba0b92be 100644 --- a/test/core/iomgr/resolve_address_posix_test.c +++ b/test/core/iomgr/resolve_address_posix_test.c @@ -81,7 +81,7 @@ void args_finish(grpc_exec_ctx *exec_ctx, args_struct *args) { grpc_pollset_shutdown(exec_ctx, args->pollset, &do_nothing_cb); // exec_ctx needs to be flushed before calling grpc_pollset_destroy() grpc_exec_ctx_flush(exec_ctx); - grpc_pollset_destroy(args->pollset); + grpc_pollset_destroy(exec_ctx, args->pollset); gpr_free(args->pollset); } diff --git a/test/core/iomgr/resolve_address_test.c b/test/core/iomgr/resolve_address_test.c index ea79adc090..0425c3b3f5 100644 --- a/test/core/iomgr/resolve_address_test.c +++ b/test/core/iomgr/resolve_address_test.c @@ -76,7 +76,7 @@ void args_finish(grpc_exec_ctx *exec_ctx, args_struct *args) { grpc_pollset_shutdown(exec_ctx, args->pollset, &do_nothing_cb); // exec_ctx needs to be flushed before calling grpc_pollset_destroy() grpc_exec_ctx_flush(exec_ctx); - grpc_pollset_destroy(args->pollset); + grpc_pollset_destroy(exec_ctx, args->pollset); gpr_free(args->pollset); } diff --git a/test/core/iomgr/tcp_client_posix_test.c b/test/core/iomgr/tcp_client_posix_test.c index 2fae6774e8..6e1bb43eb5 100644 --- a/test/core/iomgr/tcp_client_posix_test.c +++ b/test/core/iomgr/tcp_client_posix_test.c @@ -197,7 +197,7 @@ void test_fails(void) { static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } int main(int argc, char **argv) { diff --git a/test/core/iomgr/tcp_posix_test.c b/test/core/iomgr/tcp_posix_test.c index 2c53a003d2..a1c54e19c1 100644 --- a/test/core/iomgr/tcp_posix_test.c +++ b/test/core/iomgr/tcp_posix_test.c @@ -562,7 +562,7 @@ static grpc_endpoint_test_config configs[] = { static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } int main(int argc, char **argv) { diff --git a/test/core/iomgr/tcp_server_posix_test.c b/test/core/iomgr/tcp_server_posix_test.c index 112743b95b..88aead1dd0 100644 --- a/test/core/iomgr/tcp_server_posix_test.c +++ b/test/core/iomgr/tcp_server_posix_test.c @@ -444,7 +444,7 @@ static void test_connect(size_t num_connects, static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } int main(int argc, char **argv) { diff --git a/test/core/iomgr/udp_server_test.c b/test/core/iomgr/udp_server_test.c index 1f1696a7a7..ee78d6b4ad 100644 --- a/test/core/iomgr/udp_server_test.c +++ b/test/core/iomgr/udp_server_test.c @@ -307,7 +307,7 @@ static void test_receive(int number_of_clients) { static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } int main(int argc, char **argv) { diff --git a/test/core/security/secure_endpoint_test.c b/test/core/security/secure_endpoint_test.c index 71d8057ac3..cd6ff2ceac 100644 --- a/test/core/security/secure_endpoint_test.c +++ b/test/core/security/secure_endpoint_test.c @@ -185,7 +185,7 @@ static void test_leftover(grpc_endpoint_test_config config, size_t slice_size) { static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { - grpc_pollset_destroy(p); + grpc_pollset_destroy(exec_ctx, p); } int main(int argc, char **argv) { diff --git a/test/core/slice/slice_buffer_test.c b/test/core/slice/slice_buffer_test.c index bf9ae197d2..51cab2a87f 100644 --- a/test/core/slice/slice_buffer_test.c +++ b/test/core/slice/slice_buffer_test.c @@ -115,8 +115,8 @@ void test_slice_buffer_move_first() { grpc_slice_buffer_move_first(&src, 2, &dst); src_len -= 2; dst_len += 2; - GPR_ASSERT(src.length == src.length); - GPR_ASSERT(dst.length == dst.length); + GPR_ASSERT(src_len == src.length); + GPR_ASSERT(dst_len == dst.length); } int main(int argc, char **argv) { diff --git a/test/core/surface/concurrent_connectivity_test.c b/test/core/surface/concurrent_connectivity_test.c index d6841ea1f8..f0e3394b2e 100644 --- a/test/core/surface/concurrent_connectivity_test.c +++ b/test/core/surface/concurrent_connectivity_test.c @@ -162,7 +162,7 @@ void bad_server_thread(void *vargs) { static void done_pollset_shutdown(grpc_exec_ctx *exec_ctx, void *pollset, grpc_error *error) { - grpc_pollset_destroy(pollset); + grpc_pollset_destroy(exec_ctx, pollset); gpr_free(pollset); } diff --git a/test/core/util/port_server_client.c b/test/core/util/port_server_client.c index 38054dd1e7..a388d5f27b 100644 --- a/test/core/util/port_server_client.c +++ b/test/core/util/port_server_client.c @@ -58,9 +58,8 @@ typedef struct freereq { static void destroy_pops_and_shutdown(grpc_exec_ctx *exec_ctx, void *p, grpc_error *error) { grpc_pollset *pollset = grpc_polling_entity_pollset(p); - grpc_pollset_destroy(pollset); + grpc_pollset_destroy(exec_ctx, pollset); gpr_free(pollset); - grpc_shutdown(); } 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); grpc_httpcli_context_destroy(&exec_ctx, &context); - grpc_exec_ctx_finish(&exec_ctx); grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops), shutdown_closure); grpc_exec_ctx_finish(&exec_ctx); gpr_free(path); grpc_http_response_destroy(&rsp); + + grpc_shutdown(); } 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), &pr.response); grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); - grpc_exec_ctx_finish(&exec_ctx); gpr_mu_lock(pr.mu); while (pr.port == -1) { 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), shutdown_closure); grpc_exec_ctx_finish(&exec_ctx); + grpc_shutdown(); return pr.port; } diff --git a/test/core/util/test_tcp_server.c b/test/core/util/test_tcp_server.c index 496e579bc3..1908698009 100644 --- a/test/core/util/test_tcp_server.c +++ b/test/core/util/test_tcp_server.c @@ -106,6 +106,10 @@ void test_tcp_server_poll(test_tcp_server *server, int seconds) { } static void do_nothing(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {} +static void finish_pollset(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_pollset_destroy(exec_ctx, arg); +} void test_tcp_server_destroy(test_tcp_server *server) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; @@ -120,9 +124,10 @@ void test_tcp_server_destroy(test_tcp_server *server) { gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), shutdown_deadline) < 0) { test_tcp_server_poll(server, 1); } - grpc_pollset_shutdown(&exec_ctx, server->pollset, &do_nothing_cb); + grpc_pollset_shutdown(&exec_ctx, server->pollset, + grpc_closure_create(finish_pollset, server->pollset, + grpc_schedule_on_exec_ctx)); grpc_exec_ctx_finish(&exec_ctx); - grpc_pollset_destroy(server->pollset); gpr_free(server->pollset); grpc_shutdown(); } diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc index 9d7f65d292..0d267da723 100644 --- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc +++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc @@ -67,7 +67,9 @@ static void pollset_init(grpc_pollset* ps, gpr_mu** mu) { *mu = &ps->mu; } -static void pollset_destroy(grpc_pollset* ps) { gpr_mu_destroy(&ps->mu); } +static void pollset_destroy(grpc_exec_ctx* exec_ctx, grpc_pollset* ps) { + gpr_mu_destroy(&ps->mu); +} static grpc_error* pollset_kick(grpc_pollset* p, grpc_pollset_worker* worker) { return GRPC_ERROR_NONE; diff --git a/test/cpp/microbenchmarks/bm_pollset.cc b/test/cpp/microbenchmarks/bm_pollset.cc index 0f3d3cef66..19b3ba0a3c 100644 --- a/test/cpp/microbenchmarks/bm_pollset.cc +++ b/test/cpp/microbenchmarks/bm_pollset.cc @@ -59,7 +59,7 @@ extern "C" { auto& force_library_initialization = Library::get(); 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) { @@ -136,8 +136,7 @@ static void BM_PollEmptyPollset(benchmark::State& state) { gpr_timespec deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC); gpr_mu_lock(mu); while (state.KeepRunning()) { - grpc_pollset_worker* worker; - GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, &worker, now, deadline)); + GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, NULL, now, deadline)); } grpc_closure shutdown_ps_closure; grpc_closure_init(&shutdown_ps_closure, shutdown_ps, ps, @@ -150,6 +149,33 @@ static void BM_PollEmptyPollset(benchmark::State& state) { } BENCHMARK(BM_PollEmptyPollset); +static void BM_PollAddFd(benchmark::State& state) { + TrackCounters track_counters; + size_t ps_sz = grpc_pollset_size(); + grpc_pollset* ps = static_cast<grpc_pollset*>(gpr_zalloc(ps_sz)); + gpr_mu* mu; + grpc_pollset_init(ps, &mu); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_wakeup_fd wakeup_fd; + GPR_ASSERT(GRPC_LOG_IF_ERROR("wakeup_fd_init", grpc_wakeup_fd_init(&wakeup_fd))); + grpc_fd *fd = grpc_fd_create(wakeup_fd.read_fd, "xxx"); + while (state.KeepRunning()) { + grpc_pollset_add_fd(&exec_ctx, ps, fd); + grpc_exec_ctx_flush(&exec_ctx); + } + grpc_fd_orphan(&exec_ctx, fd, NULL, NULL, "xxx"); + grpc_closure shutdown_ps_closure; + grpc_closure_init(&shutdown_ps_closure, shutdown_ps, ps, + grpc_schedule_on_exec_ctx); + gpr_mu_lock(mu); + grpc_pollset_shutdown(&exec_ctx, ps, &shutdown_ps_closure); + gpr_mu_unlock(mu); + grpc_exec_ctx_finish(&exec_ctx); + gpr_free(ps); + track_counters.Finish(state); +} +BENCHMARK(BM_PollAddFd); + class Closure : public grpc_closure { public: virtual ~Closure() {} @@ -233,8 +259,7 @@ static void BM_SingleThreadPollOneFd(benchmark::State& state) { grpc_fd_notify_on_read(&exec_ctx, wakeup, continue_closure); gpr_mu_lock(mu); while (!done) { - grpc_pollset_worker* worker; - GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, &worker, now, deadline)); + GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, NULL, now, deadline)); } grpc_fd_orphan(&exec_ctx, wakeup, NULL, NULL, "done"); wakeup_fd.read_fd = 0; diff --git a/test/cpp/qps/gen_build_yaml.py b/test/cpp/qps/gen_build_yaml.py index 2f035abedd..805b0faeec 100755 --- a/test/cpp/qps/gen_build_yaml.py +++ b/test/cpp/qps/gen_build_yaml.py @@ -66,6 +66,7 @@ def _scenario_json_string(scenario_json, is_tsan): def threads_required(scenario_json, where, is_tsan): scenario_json = mutate_scenario(scenario_json, is_tsan) if scenario_json['%s_config' % where]['%s_type' % where] == 'ASYNC_%s' % where.upper(): + if scenario_json['client_config']['client_channels'] == 1: return 1 return scenario_json['%s_config' % where].get('async_%s_threads' % where, 0) return scenario_json['client_config']['outstanding_rpcs_per_channel'] * scenario_json['client_config']['client_channels'] diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index 88a9736d5b..f0768464d8 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -941,6 +941,8 @@ src/core/lib/iomgr/error.h \ src/core/lib/iomgr/error_internal.h \ src/core/lib/iomgr/ev_epoll_linux.c \ src/core/lib/iomgr/ev_epoll_linux.h \ +src/core/lib/iomgr/ev_epollex_linux.c \ +src/core/lib/iomgr/ev_epollex_linux.h \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_poll_posix.h \ src/core/lib/iomgr/ev_posix.c \ diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 924595eb95..02af1d69cb 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -1076,6 +1076,8 @@ src/core/lib/iomgr/error.h \ src/core/lib/iomgr/error_internal.h \ src/core/lib/iomgr/ev_epoll_linux.c \ src/core/lib/iomgr/ev_epoll_linux.h \ +src/core/lib/iomgr/ev_epollex_linux.c \ +src/core/lib/iomgr/ev_epollex_linux.h \ src/core/lib/iomgr/ev_poll_posix.c \ src/core/lib/iomgr/ev_poll_posix.h \ src/core/lib/iomgr/ev_posix.c \ diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json index 70e26194e7..ed005671ff 100644 --- a/tools/run_tests/generated/sources_and_headers.json +++ b/tools/run_tests/generated/sources_and_headers.json @@ -7633,6 +7633,7 @@ "src/core/lib/iomgr/error.h", "src/core/lib/iomgr/error_internal.h", "src/core/lib/iomgr/ev_epoll_linux.h", + "src/core/lib/iomgr/ev_epollex_linux.h", "src/core/lib/iomgr/ev_poll_posix.h", "src/core/lib/iomgr/ev_posix.h", "src/core/lib/iomgr/exec_ctx.h", @@ -7775,6 +7776,8 @@ "src/core/lib/iomgr/error_internal.h", "src/core/lib/iomgr/ev_epoll_linux.c", "src/core/lib/iomgr/ev_epoll_linux.h", + "src/core/lib/iomgr/ev_epollex_linux.c", + "src/core/lib/iomgr/ev_epollex_linux.h", "src/core/lib/iomgr/ev_poll_posix.c", "src/core/lib/iomgr/ev_poll_posix.h", "src/core/lib/iomgr/ev_posix.c", diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json index 8b7bc1f14d..5ef8fe5f96 100644 --- a/tools/run_tests/generated/tests.json +++ b/tools/run_tests/generated/tests.json @@ -41328,7 +41328,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 100, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41430,7 +41430,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41482,7 +41482,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41907,7 +41907,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 100, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42009,7 +42009,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42061,7 +42061,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42538,7 +42538,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 100, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -42692,7 +42692,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -42770,7 +42770,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43416,7 +43416,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 100, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43570,7 +43570,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43648,7 +43648,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", diff --git a/tools/run_tests/run_tests.py b/tools/run_tests/run_tests.py index a1ec1b2f45..d2e3506fe5 100755 --- a/tools/run_tests/run_tests.py +++ b/tools/run_tests/run_tests.py @@ -72,7 +72,7 @@ _FORCE_ENVIRON_FOR_WRAPPERS = { _POLLING_STRATEGIES = { - 'linux': ['epoll', 'poll', 'poll-cv'] + 'linux': ['epollex', 'epoll', 'poll', 'poll-cv'] } diff --git a/vsprojects/vcxproj/grpc++/grpc++.vcxproj b/vsprojects/vcxproj/grpc++/grpc++.vcxproj index 424af0d9e7..fa4610974a 100644 --- a/vsprojects/vcxproj/grpc++/grpc++.vcxproj +++ b/vsprojects/vcxproj/grpc++/grpc++.vcxproj @@ -397,6 +397,7 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error_internal.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h" /> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\exec_ctx.h" /> @@ -612,6 +613,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.c"> diff --git a/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters b/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters index a2f74655d0..e94aee357e 100644 --- a/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters +++ b/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters @@ -187,6 +187,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + <Filter>src\core\lib\iomgr</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> @@ -911,6 +914,9 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h"> + <Filter>src\core\lib\iomgr</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> diff --git a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj index 3f8a2f2757..5513816d77 100644 --- a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj @@ -391,6 +391,7 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error_internal.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h" /> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\exec_ctx.h" /> @@ -596,6 +597,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.c"> diff --git a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters index 929501a56c..646e35bbe8 100644 --- a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters @@ -172,6 +172,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + <Filter>src\core\lib\iomgr</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> @@ -878,6 +881,9 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h"> + <Filter>src\core\lib\iomgr</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj b/vsprojects/vcxproj/grpc/grpc.vcxproj index ca70dde793..d6881a98c0 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj @@ -321,6 +321,7 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error_internal.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h" /> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\exec_ctx.h" /> @@ -550,6 +551,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.c"> diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters index acadc0ad88..2a694cd4f3 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters @@ -67,6 +67,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + <Filter>src\core\lib\iomgr</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> @@ -869,6 +872,9 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h"> + <Filter>src\core\lib\iomgr</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> diff --git a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj index a5ec4ae171..1a3120c798 100644 --- a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj +++ b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj @@ -216,6 +216,7 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error_internal.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h" /> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\exec_ctx.h" /> @@ -385,6 +386,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.c"> diff --git a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters index 4078ebd5d7..d750cf78d9 100644 --- a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters @@ -124,6 +124,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + <Filter>src\core\lib\iomgr</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> @@ -626,6 +629,9 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h"> + <Filter>src\core\lib\iomgr</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj index c844e157e4..0716de4a9c 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj @@ -311,6 +311,7 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\error_internal.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h" /> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\exec_ctx.h" /> @@ -518,6 +519,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_posix.c"> diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters index 43d27b44a2..ebbd03b5fb 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters @@ -70,6 +70,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.c"> + <Filter>src\core\lib\iomgr</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.c"> <Filter>src\core\lib\iomgr</Filter> </ClCompile> @@ -782,6 +785,9 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epoll_linux.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_epollex_linux.h"> + <Filter>src\core\lib\iomgr</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\src\core\lib\iomgr\ev_poll_posix.h"> <Filter>src\core\lib\iomgr</Filter> </ClInclude> |