aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib/iomgr
diff options
context:
space:
mode:
authorGravatar Craig Tiller <ctiller@google.com>2016-11-16 14:45:10 -0800
committerGravatar Craig Tiller <ctiller@google.com>2016-11-16 14:45:10 -0800
commitd57a1487b17a2fb7251cc575dd7161ae6c1bea13 (patch)
tree7635e13309c057d7449d3e1528429dca0070be69 /src/core/lib/iomgr
parent3e2048dd9cb2a5458105cbbbf4d4b4b67549bd5c (diff)
parent740665a6f65b3d827e0755de8bb1bcd57745b9f1 (diff)
Merge github.com:grpc/grpc into newlines
Diffstat (limited to 'src/core/lib/iomgr')
-rw-r--r--src/core/lib/iomgr/endpoint.c6
-rw-r--r--src/core/lib/iomgr/endpoint.h17
-rw-r--r--src/core/lib/iomgr/ev_epoll_linux.c2
-rw-r--r--src/core/lib/iomgr/ev_poll_and_epoll_posix.c2076
-rw-r--r--src/core/lib/iomgr/ev_poll_and_epoll_posix.h41
-rw-r--r--src/core/lib/iomgr/ev_posix.c2
-rw-r--r--src/core/lib/iomgr/load_file.c6
-rw-r--r--src/core/lib/iomgr/load_file.h4
-rw-r--r--src/core/lib/iomgr/resource_quota.c217
-rw-r--r--src/core/lib/iomgr/resource_quota.h98
-rw-r--r--src/core/lib/iomgr/socket_mutator.c98
-rw-r--r--src/core/lib/iomgr/socket_mutator.h80
-rw-r--r--src/core/lib/iomgr/socket_utils_common_posix.c9
-rw-r--r--src/core/lib/iomgr/socket_utils_posix.h5
-rw-r--r--src/core/lib/iomgr/tcp_client.h1
-rw-r--r--src/core/lib/iomgr/tcp_client_posix.c16
-rw-r--r--src/core/lib/iomgr/tcp_client_windows.c2
-rw-r--r--src/core/lib/iomgr/tcp_posix.c100
-rw-r--r--src/core/lib/iomgr/tcp_uv.c107
-rw-r--r--src/core/lib/iomgr/tcp_windows.c92
-rw-r--r--src/core/lib/iomgr/wakeup_fd_pipe.c2
21 files changed, 518 insertions, 2463 deletions
diff --git a/src/core/lib/iomgr/endpoint.c b/src/core/lib/iomgr/endpoint.c
index 74fa9c45df..2d300f4560 100644
--- a/src/core/lib/iomgr/endpoint.c
+++ b/src/core/lib/iomgr/endpoint.c
@@ -34,12 +34,12 @@
#include "src/core/lib/iomgr/endpoint.h"
void grpc_endpoint_read(grpc_exec_ctx* exec_ctx, grpc_endpoint* ep,
- gpr_slice_buffer* slices, grpc_closure* cb) {
+ grpc_slice_buffer* slices, grpc_closure* cb) {
ep->vtable->read(exec_ctx, ep, slices, cb);
}
void grpc_endpoint_write(grpc_exec_ctx* exec_ctx, grpc_endpoint* ep,
- gpr_slice_buffer* slices, grpc_closure* cb) {
+ grpc_slice_buffer* slices, grpc_closure* cb) {
ep->vtable->write(exec_ctx, ep, slices, cb);
}
@@ -66,6 +66,8 @@ char* grpc_endpoint_get_peer(grpc_endpoint* ep) {
return ep->vtable->get_peer(ep);
}
+int grpc_endpoint_get_fd(grpc_endpoint* ep) { return ep->vtable->get_fd(ep); }
+
grpc_workqueue* grpc_endpoint_get_workqueue(grpc_endpoint* ep) {
return ep->vtable->get_workqueue(ep);
}
diff --git a/src/core/lib/iomgr/endpoint.h b/src/core/lib/iomgr/endpoint.h
index 0ac5486ff5..1609b64f2b 100644
--- a/src/core/lib/iomgr/endpoint.h
+++ b/src/core/lib/iomgr/endpoint.h
@@ -34,8 +34,8 @@
#ifndef GRPC_CORE_LIB_IOMGR_ENDPOINT_H
#define GRPC_CORE_LIB_IOMGR_ENDPOINT_H
-#include <grpc/support/slice.h>
-#include <grpc/support/slice_buffer.h>
+#include <grpc/slice.h>
+#include <grpc/slice_buffer.h>
#include <grpc/support/time.h>
#include "src/core/lib/iomgr/pollset.h"
#include "src/core/lib/iomgr/pollset_set.h"
@@ -49,9 +49,9 @@ typedef struct grpc_endpoint_vtable grpc_endpoint_vtable;
struct grpc_endpoint_vtable {
void (*read)(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *slices, grpc_closure *cb);
+ grpc_slice_buffer *slices, grpc_closure *cb);
void (*write)(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *slices, grpc_closure *cb);
+ grpc_slice_buffer *slices, grpc_closure *cb);
grpc_workqueue *(*get_workqueue)(grpc_endpoint *ep);
void (*add_to_pollset)(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_pollset *pollset);
@@ -61,6 +61,7 @@ struct grpc_endpoint_vtable {
void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep);
grpc_resource_user *(*get_resource_user)(grpc_endpoint *ep);
char *(*get_peer)(grpc_endpoint *ep);
+ int (*get_fd)(grpc_endpoint *ep);
};
/* When data is available on the connection, calls the callback with slices.
@@ -69,10 +70,14 @@ struct grpc_endpoint_vtable {
Valid slices may be placed into \a slices even when the callback is
invoked with error != GRPC_ERROR_NONE. */
void grpc_endpoint_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *slices, grpc_closure *cb);
+ grpc_slice_buffer *slices, grpc_closure *cb);
char *grpc_endpoint_get_peer(grpc_endpoint *ep);
+/* Get the file descriptor used by \a ep. Return -1 if \a ep is not using an fd.
+ */
+int grpc_endpoint_get_fd(grpc_endpoint *ep);
+
/* Retrieve a reference to the workqueue associated with this endpoint */
grpc_workqueue *grpc_endpoint_get_workqueue(grpc_endpoint *ep);
@@ -87,7 +92,7 @@ grpc_workqueue *grpc_endpoint_get_workqueue(grpc_endpoint *ep);
it is a valid slice buffer.
*/
void grpc_endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *slices, grpc_closure *cb);
+ grpc_slice_buffer *slices, grpc_closure *cb);
/* Causes any pending and future read/write callbacks to run immediately with
success==0 */
diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c
index db51ec4939..91041a7c28 100644
--- a/src/core/lib/iomgr/ev_epoll_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_linux.c
@@ -163,7 +163,7 @@ static void fd_global_shutdown(void);
#define PI_ADD_REF(p, r) pi_add_ref((p))
#define PI_UNREF(exec_ctx, p, r) pi_unref((exec_ctx), (p))
-#endif /* !defined(GPRC_PI_REF_COUNT_DEBUG) */
+#endif /* !defined(GRPC_PI_REF_COUNT_DEBUG) */
/* This is also used as grpc_workqueue (by directly casing it) */
typedef struct polling_island {
diff --git a/src/core/lib/iomgr/ev_poll_and_epoll_posix.c b/src/core/lib/iomgr/ev_poll_and_epoll_posix.c
deleted file mode 100644
index bf51404203..0000000000
--- a/src/core/lib/iomgr/ev_poll_and_epoll_posix.c
+++ /dev/null
@@ -1,2076 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-/* This file will be removed shortly: it's here to keep refactoring
- * steps simple and auditable.
- * It's the combination of the old files:
- * - fd_posix.{h,c}
- * - pollset_posix.{h,c}
- * - pullset_multipoller_with_{poll,epoll}.{h,c}
- * The new version will be split into:
- * - ev_poll_posix.{h,c}
- * - ev_epoll_posix.{h,c}
- */
-
-#include "src/core/lib/iomgr/port.h"
-
-#ifdef GRPC_POSIX_SOCKET
-
-#include "src/core/lib/iomgr/ev_poll_and_epoll_posix.h"
-
-#include <assert.h>
-#include <errno.h>
-#include <poll.h>
-#include <string.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/iomgr_internal.h"
-#include "src/core/lib/iomgr/wakeup_fd_posix.h"
-#include "src/core/lib/profiling/timers.h"
-#include "src/core/lib/support/block_annotate.h"
-
-/*******************************************************************************
- * FD declarations
- */
-
-typedef struct grpc_fd_watcher {
- struct grpc_fd_watcher *next;
- struct grpc_fd_watcher *prev;
- grpc_pollset *pollset;
- grpc_pollset_worker *worker;
- grpc_fd *fd;
-} grpc_fd_watcher;
-
-struct grpc_fd {
- int fd;
- /* refst format:
- bit0: 1=active/0=orphaned
- bit1-n: refcount
- meaning that mostly we ref by two to avoid altering the orphaned bit,
- and just unref by 1 when we're ready to flag the object as orphaned */
- gpr_atm refst;
-
- gpr_mu mu;
- int shutdown;
- int closed;
- int released;
-
- /* The watcher list.
-
- The following watcher related fields are protected by watcher_mu.
-
- An fd_watcher is an ephemeral object created when an fd wants to
- begin polling, and destroyed after the poll.
-
- It denotes the fd's interest in whether to read poll or write poll
- or both or neither on this fd.
-
- If a watcher is asked to poll for reads or writes, the read_watcher
- or write_watcher fields are set respectively. A watcher may be asked
- to poll for both, in which case both fields will be set.
-
- read_watcher and write_watcher may be NULL if no watcher has been
- asked to poll for reads or writes.
-
- If an fd_watcher is not asked to poll for reads or writes, it's added
- to a linked list of inactive watchers, rooted at inactive_watcher_root.
- If at a later time there becomes need of a poller to poll, one of
- the inactive pollers may be kicked out of their poll loops to take
- that responsibility. */
- grpc_fd_watcher inactive_watcher_root;
- grpc_fd_watcher *read_watcher;
- grpc_fd_watcher *write_watcher;
-
- grpc_closure *read_closure;
- grpc_closure *write_closure;
-
- struct grpc_fd *freelist_next;
-
- grpc_closure *on_done_closure;
-
- grpc_iomgr_object iomgr_object;
-
- /* The pollset that last noticed and notified that the fd is readable */
- grpc_pollset *read_notifier_pollset;
-};
-
-/* Begin polling on an fd.
- Registers that the given pollset is interested in this fd - so that if read
- or writability interest changes, the pollset can be kicked to pick up that
- new interest.
- Return value is:
- (fd_needs_read? read_mask : 0) | (fd_needs_write? write_mask : 0)
- i.e. a combination of read_mask and write_mask determined by the fd's current
- interest in said events.
- Polling strategies that do not need to alter their behavior depending on the
- fd's current interest (such as epoll) do not need to call this function.
- MUST NOT be called with a pollset lock taken */
-static uint32_t fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset,
- grpc_pollset_worker *worker, uint32_t read_mask,
- uint32_t write_mask, grpc_fd_watcher *rec);
-/* Complete polling previously started with fd_begin_poll
- MUST NOT be called with a pollset lock taken
- if got_read or got_write are 1, also does the become_{readable,writable} as
- appropriate. */
-static void fd_end_poll(grpc_exec_ctx *exec_ctx, grpc_fd_watcher *rec,
- int got_read, int got_write,
- grpc_pollset *read_notifier_pollset);
-
-/* Return 1 if this fd is orphaned, 0 otherwise */
-static bool fd_is_orphaned(grpc_fd *fd);
-
-/* Reference counting for fds */
-/*#define GRPC_FD_REF_COUNT_DEBUG*/
-#ifdef GRPC_FD_REF_COUNT_DEBUG
-static void fd_ref(grpc_fd *fd, const char *reason, const char *file, int line);
-static void fd_unref(grpc_fd *fd, const char *reason, const char *file,
- int line);
-#define GRPC_FD_REF(fd, reason) fd_ref(fd, reason, __FILE__, __LINE__)
-#define GRPC_FD_UNREF(fd, reason) fd_unref(fd, reason, __FILE__, __LINE__)
-#else
-static void fd_ref(grpc_fd *fd);
-static void fd_unref(grpc_fd *fd);
-#define GRPC_FD_REF(fd, reason) fd_ref(fd)
-#define GRPC_FD_UNREF(fd, reason) fd_unref(fd)
-#endif
-
-static void fd_global_init(void);
-static void fd_global_shutdown(void);
-
-#define CLOSURE_NOT_READY ((grpc_closure *)0)
-#define CLOSURE_READY ((grpc_closure *)1)
-
-/*******************************************************************************
- * pollset declarations
- */
-
-typedef struct grpc_pollset_vtable grpc_pollset_vtable;
-
-typedef struct grpc_cached_wakeup_fd {
- grpc_wakeup_fd fd;
- struct grpc_cached_wakeup_fd *next;
-} grpc_cached_wakeup_fd;
-
-struct grpc_pollset_worker {
- grpc_cached_wakeup_fd *wakeup_fd;
- int reevaluate_polling_on_wakeup;
- int kicked_specifically;
- struct grpc_pollset_worker *next;
- struct grpc_pollset_worker *prev;
-};
-
-struct grpc_pollset {
- /* pollsets under posix can mutate representation as fds are added and
- removed.
- For example, we may choose a poll() based implementation on linux for
- few fds, and an epoll() based implementation for many fds */
- const grpc_pollset_vtable *vtable;
- gpr_mu mu;
- grpc_pollset_worker root_worker;
- int in_flight_cbs;
- int shutting_down;
- int called_shutdown;
- int kicked_without_pollers;
- grpc_closure *shutdown_done;
- grpc_closure_list idle_jobs;
- union {
- int fd;
- void *ptr;
- } data;
- /* Local cache of eventfds for workers */
- grpc_cached_wakeup_fd *local_wakeup_cache;
-};
-
-struct grpc_pollset_vtable {
- void (*add_fd)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
- struct grpc_fd *fd, int and_unlock_pollset);
- grpc_error *(*maybe_work_and_unlock)(grpc_exec_ctx *exec_ctx,
- grpc_pollset *pollset,
- grpc_pollset_worker *worker,
- gpr_timespec deadline, gpr_timespec now);
- void (*finish_shutdown)(grpc_pollset *pollset);
- void (*destroy)(grpc_pollset *pollset);
-};
-
-/* Add an fd to a pollset */
-static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
- struct grpc_fd *fd);
-
-static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx,
- grpc_pollset_set *pollset_set, grpc_fd *fd);
-
-/* 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);
-
-/* Allow kick to wakeup the currently polling worker */
-#define GRPC_POLLSET_CAN_KICK_SELF 1
-/* Force the wakee to repoll when awoken */
-#define GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP 2
-/* As per pollset_kick, with an extended set of flags (defined above)
- -- mostly for fd_posix's use. */
-static grpc_error *pollset_kick_ext(grpc_pollset *p,
- grpc_pollset_worker *specific_worker,
- uint32_t flags) GRPC_MUST_USE_RESULT;
-
-/* turn a pollset into a multipoller: platform specific */
-typedef void (*platform_become_multipoller_type)(grpc_exec_ctx *exec_ctx,
- grpc_pollset *pollset,
- struct grpc_fd **fds,
- size_t fd_count);
-static platform_become_multipoller_type platform_become_multipoller;
-
-/* Return 1 if the pollset has active threads in pollset_work (pollset must
- * be locked) */
-static int pollset_has_workers(grpc_pollset *pollset);
-
-static void remove_fd_from_all_epoll_sets(int fd);
-
-/*******************************************************************************
- * pollset_set definitions
- */
-
-struct grpc_pollset_set {
- gpr_mu mu;
-
- size_t pollset_count;
- size_t pollset_capacity;
- grpc_pollset **pollsets;
-
- size_t pollset_set_count;
- size_t pollset_set_capacity;
- struct grpc_pollset_set **pollset_sets;
-
- size_t fd_count;
- size_t fd_capacity;
- grpc_fd **fds;
-};
-
-/*******************************************************************************
- * fd_posix.c
- */
-
-/* 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.
- */
-/* TODO(klempner): We could use some form of polling generation count to know
- * when these are safe to free. */
-/* TODO(klempner): Consider disabling freelisting if we don't have multiple
- * threads in poll on the same fd */
-/* TODO(klempner): Batch these allocations to reduce fragmentation */
-static grpc_fd *fd_freelist = NULL;
-static gpr_mu fd_freelist_mu;
-
-static void freelist_fd(grpc_fd *fd) {
- gpr_mu_lock(&fd_freelist_mu);
- fd->freelist_next = fd_freelist;
- fd_freelist = fd;
- grpc_iomgr_unregister_object(&fd->iomgr_object);
- gpr_mu_unlock(&fd_freelist_mu);
-}
-
-static grpc_fd *alloc_fd(int fd) {
- grpc_fd *r = NULL;
- gpr_mu_lock(&fd_freelist_mu);
- if (fd_freelist != NULL) {
- r = fd_freelist;
- fd_freelist = fd_freelist->freelist_next;
- }
- gpr_mu_unlock(&fd_freelist_mu);
- if (r == NULL) {
- r = gpr_malloc(sizeof(grpc_fd));
- gpr_mu_init(&r->mu);
- }
-
- gpr_mu_lock(&r->mu);
- gpr_atm_rel_store(&r->refst, 1);
- r->shutdown = 0;
- r->read_closure = CLOSURE_NOT_READY;
- r->write_closure = CLOSURE_NOT_READY;
- r->fd = fd;
- r->inactive_watcher_root.next = r->inactive_watcher_root.prev =
- &r->inactive_watcher_root;
- r->freelist_next = NULL;
- r->read_watcher = r->write_watcher = NULL;
- r->on_done_closure = NULL;
- r->closed = 0;
- r->released = 0;
- r->read_notifier_pollset = NULL;
- gpr_mu_unlock(&r->mu);
- return r;
-}
-
-static void destroy(grpc_fd *fd) {
- gpr_mu_destroy(&fd->mu);
- gpr_free(fd);
-}
-
-#ifdef GRPC_FD_REF_COUNT_DEBUG
-#define REF_BY(fd, n, reason) ref_by(fd, n, reason, __FILE__, __LINE__)
-#define UNREF_BY(fd, n, reason) unref_by(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 %d -> %d [%s; %s:%d]", fd->fd, 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(fd, n, reason) unref_by(fd, n)
-static void ref_by(grpc_fd *fd, int n) {
-#endif
- GPR_ASSERT(gpr_atm_no_barrier_fetch_add(&fd->refst, n) > 0);
-}
-
-#ifdef GRPC_FD_REF_COUNT_DEBUG
-static void unref_by(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 %d -> %d [%s; %s:%d]", fd->fd, 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_fd *fd, int n) {
- gpr_atm old;
-#endif
- old = gpr_atm_full_fetch_add(&fd->refst, -n);
- if (old == n) {
- freelist_fd(fd);
- } 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;
- destroy(fd);
- }
- gpr_mu_destroy(&fd_freelist_mu);
-}
-
-static grpc_fd *fd_create(int fd, const char *name) {
- grpc_fd *r = alloc_fd(fd);
- char *name2;
- gpr_asprintf(&name2, "%s fd=%d", name, fd);
- grpc_iomgr_register_object(&r->iomgr_object, name2);
- gpr_free(name2);
-#ifdef GRPC_FD_REF_COUNT_DEBUG
- gpr_log(GPR_DEBUG, "FD %d %p create %s", fd, r, name);
-#endif
- return r;
-}
-
-static bool fd_is_orphaned(grpc_fd *fd) {
- return (gpr_atm_acq_load(&fd->refst) & 1) == 0;
-}
-
-static grpc_error *pollset_kick_locked(grpc_fd_watcher *watcher) {
- gpr_mu_lock(&watcher->pollset->mu);
- GPR_ASSERT(watcher->worker);
- grpc_error *err = pollset_kick_ext(watcher->pollset, watcher->worker,
- GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP);
- gpr_mu_unlock(&watcher->pollset->mu);
- return err;
-}
-
-static void maybe_wake_one_watcher_locked(grpc_fd *fd) {
- if (fd->inactive_watcher_root.next != &fd->inactive_watcher_root) {
- pollset_kick_locked(fd->inactive_watcher_root.next);
- } else if (fd->read_watcher) {
- pollset_kick_locked(fd->read_watcher);
- } else if (fd->write_watcher) {
- pollset_kick_locked(fd->write_watcher);
- }
-}
-
-static void wake_all_watchers_locked(grpc_fd *fd) {
- grpc_fd_watcher *watcher;
- for (watcher = fd->inactive_watcher_root.next;
- watcher != &fd->inactive_watcher_root; watcher = watcher->next) {
- pollset_kick_locked(watcher);
- }
- if (fd->read_watcher) {
- pollset_kick_locked(fd->read_watcher);
- }
- if (fd->write_watcher && fd->write_watcher != fd->read_watcher) {
- pollset_kick_locked(fd->write_watcher);
- }
-}
-
-static int has_watchers(grpc_fd *fd) {
- return fd->read_watcher != NULL || fd->write_watcher != NULL ||
- fd->inactive_watcher_root.next != &fd->inactive_watcher_root;
-}
-
-static void close_fd_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
- fd->closed = 1;
- if (!fd->released) {
- close(fd->fd);
- } else {
- remove_fd_from_all_epoll_sets(fd->fd);
- }
- grpc_exec_ctx_sched(exec_ctx, fd->on_done_closure, GRPC_ERROR_NONE, NULL);
-}
-
-static int fd_wrapped_fd(grpc_fd *fd) {
- if (fd->released || fd->closed) {
- return -1;
- } else {
- return fd->fd;
- }
-}
-
-static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
- grpc_closure *on_done, int *release_fd,
- const char *reason) {
- fd->on_done_closure = on_done;
- fd->released = release_fd != NULL;
- if (!fd->released) {
- shutdown(fd->fd, SHUT_RDWR);
- } else {
- *release_fd = fd->fd;
- }
- gpr_mu_lock(&fd->mu);
- REF_BY(fd, 1, reason); /* remove active status, but keep referenced */
- if (!has_watchers(fd)) {
- close_fd_locked(exec_ctx, fd);
- } else {
- wake_all_watchers_locked(fd);
- }
- gpr_mu_unlock(&fd->mu);
- UNREF_BY(fd, 2, reason); /* drop the reference */
-}
-
-/* increment refcount by two to avoid changing the orphan bit */
-#ifdef GRPC_FD_REF_COUNT_DEBUG
-static void fd_ref(grpc_fd *fd, const char *reason, const char *file,
- int line) {
- ref_by(fd, 2, reason, file, line);
-}
-
-static void fd_unref(grpc_fd *fd, const char *reason, const char *file,
- int line) {
- unref_by(fd, 2, reason, file, line);
-}
-#else
-static void fd_ref(grpc_fd *fd) { ref_by(fd, 2); }
-
-static void fd_unref(grpc_fd *fd) { unref_by(fd, 2); }
-#endif
-
-static grpc_error *fd_shutdown_error(bool shutdown) {
- if (!shutdown) {
- return GRPC_ERROR_NONE;
- } else {
- return GRPC_ERROR_CREATE("FD shutdown");
- }
-}
-
-static void notify_on_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
- grpc_closure **st, grpc_closure *closure) {
- if (fd->shutdown) {
- grpc_exec_ctx_sched(exec_ctx, closure, GRPC_ERROR_CREATE("FD shutdown"),
- NULL);
- } else if (*st == CLOSURE_NOT_READY) {
- /* not ready ==> switch to a waiting state by setting the closure */
- *st = closure;
- } else if (*st == CLOSURE_READY) {
- /* already ready ==> queue the closure to run immediately */
- *st = CLOSURE_NOT_READY;
- grpc_exec_ctx_sched(exec_ctx, closure, fd_shutdown_error(fd->shutdown),
- NULL);
- maybe_wake_one_watcher_locked(fd);
- } else {
- /* upcallptr was set to a different closure. This is an error! */
- gpr_log(GPR_ERROR,
- "User called a notify_on function with a previous callback still "
- "pending");
- abort();
- }
-}
-
-/* returns 1 if state becomes not ready */
-static int set_ready_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
- grpc_closure **st) {
- if (*st == CLOSURE_READY) {
- /* duplicate ready ==> ignore */
- return 0;
- } else if (*st == CLOSURE_NOT_READY) {
- /* not ready, and not waiting ==> flag ready */
- *st = CLOSURE_READY;
- return 0;
- } else {
- /* waiting ==> queue closure */
- grpc_exec_ctx_sched(exec_ctx, *st, fd_shutdown_error(fd->shutdown), NULL);
- *st = CLOSURE_NOT_READY;
- return 1;
- }
-}
-
-static void set_read_notifier_pollset_locked(
- grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_pollset *read_notifier_pollset) {
- fd->read_notifier_pollset = read_notifier_pollset;
-}
-
-static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
- gpr_mu_lock(&fd->mu);
- /* only shutdown once */
- if (!fd->shutdown) {
- fd->shutdown = 1;
- /* signal read/write closed to OS so that future operations fail */
- shutdown(fd->fd, SHUT_RDWR);
- set_ready_locked(exec_ctx, fd, &fd->read_closure);
- set_ready_locked(exec_ctx, fd, &fd->write_closure);
- }
- gpr_mu_unlock(&fd->mu);
-}
-
-static bool fd_is_shutdown(grpc_fd *fd) {
- gpr_mu_lock(&fd->mu);
- bool r = fd->shutdown;
- gpr_mu_unlock(&fd->mu);
- return r;
-}
-
-static void fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
- grpc_closure *closure) {
- gpr_mu_lock(&fd->mu);
- notify_on_locked(exec_ctx, fd, &fd->read_closure, closure);
- gpr_mu_unlock(&fd->mu);
-}
-
-static void fd_notify_on_write(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
- grpc_closure *closure) {
- gpr_mu_lock(&fd->mu);
- notify_on_locked(exec_ctx, fd, &fd->write_closure, closure);
- gpr_mu_unlock(&fd->mu);
-}
-
-/* Return the read-notifier pollset */
-static grpc_pollset *fd_get_read_notifier_pollset(grpc_exec_ctx *exec_ctx,
- grpc_fd *fd) {
- grpc_pollset *notifier = NULL;
-
- gpr_mu_lock(&fd->mu);
- notifier = fd->read_notifier_pollset;
- gpr_mu_unlock(&fd->mu);
-
- return notifier;
-}
-
-static uint32_t fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset,
- grpc_pollset_worker *worker, uint32_t read_mask,
- uint32_t write_mask, grpc_fd_watcher *watcher) {
- uint32_t mask = 0;
- grpc_closure *cur;
- int requested;
- /* keep track of pollers that have requested our events, in case they change
- */
- GRPC_FD_REF(fd, "poll");
-
- gpr_mu_lock(&fd->mu);
-
- /* if we are shutdown, then don't add to the watcher set */
- if (fd->shutdown) {
- watcher->fd = NULL;
- watcher->pollset = NULL;
- watcher->worker = NULL;
- gpr_mu_unlock(&fd->mu);
- GRPC_FD_UNREF(fd, "poll");
- return 0;
- }
-
- /* if there is nobody polling for read, but we need to, then start doing so */
- cur = fd->read_closure;
- requested = cur != CLOSURE_READY;
- if (read_mask && fd->read_watcher == NULL && requested) {
- fd->read_watcher = watcher;
- mask |= read_mask;
- }
- /* if there is nobody polling for write, but we need to, then start doing so
- */
- cur = fd->write_closure;
- requested = cur != CLOSURE_READY;
- if (write_mask && fd->write_watcher == NULL && requested) {
- fd->write_watcher = watcher;
- mask |= write_mask;
- }
- /* if not polling, remember this watcher in case we need someone to later */
- if (mask == 0 && worker != NULL) {
- watcher->next = &fd->inactive_watcher_root;
- watcher->prev = watcher->next->prev;
- watcher->next->prev = watcher->prev->next = watcher;
- }
- watcher->pollset = pollset;
- watcher->worker = worker;
- watcher->fd = fd;
- gpr_mu_unlock(&fd->mu);
-
- return mask;
-}
-
-static void fd_end_poll(grpc_exec_ctx *exec_ctx, grpc_fd_watcher *watcher,
- int got_read, int got_write,
- grpc_pollset *read_notifier_pollset) {
- int was_polling = 0;
- int kick = 0;
- grpc_fd *fd = watcher->fd;
-
- if (fd == NULL) {
- return;
- }
-
- gpr_mu_lock(&fd->mu);
-
- if (watcher == fd->read_watcher) {
- /* remove read watcher, kick if we still need a read */
- was_polling = 1;
- if (!got_read) {
- kick = 1;
- }
- fd->read_watcher = NULL;
- }
- if (watcher == fd->write_watcher) {
- /* remove write watcher, kick if we still need a write */
- was_polling = 1;
- if (!got_write) {
- kick = 1;
- }
- fd->write_watcher = NULL;
- }
- if (!was_polling && watcher->worker != NULL) {
- /* remove from inactive list */
- watcher->next->prev = watcher->prev;
- watcher->prev->next = watcher->next;
- }
- if (got_read) {
- if (set_ready_locked(exec_ctx, fd, &fd->read_closure)) {
- kick = 1;
- }
-
- if (read_notifier_pollset != NULL) {
- set_read_notifier_pollset_locked(exec_ctx, fd, read_notifier_pollset);
- }
- }
- if (got_write) {
- if (set_ready_locked(exec_ctx, fd, &fd->write_closure)) {
- kick = 1;
- }
- }
- if (kick) {
- maybe_wake_one_watcher_locked(fd);
- }
- if (fd_is_orphaned(fd) && !has_watchers(fd) && !fd->closed) {
- close_fd_locked(exec_ctx, fd);
- }
- gpr_mu_unlock(&fd->mu);
-
- GRPC_FD_UNREF(fd, "poll");
-}
-
-static grpc_workqueue *fd_get_workqueue(grpc_fd *fd) { return NULL; }
-
-/*******************************************************************************
- * pollset_posix.c
- */
-
-GPR_TLS_DECL(g_current_thread_poller);
-GPR_TLS_DECL(g_current_thread_worker);
-
-/** 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. */
-grpc_wakeup_fd grpc_global_wakeup_fd;
-
-static void remove_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
- worker->prev->next = worker->next;
- worker->next->prev = worker->prev;
-}
-
-static int pollset_has_workers(grpc_pollset *p) {
- return p->root_worker.next != &p->root_worker;
-}
-
-static grpc_pollset_worker *pop_front_worker(grpc_pollset *p) {
- if (pollset_has_workers(p)) {
- grpc_pollset_worker *w = p->root_worker.next;
- remove_worker(p, w);
- return w;
- } else {
- return NULL;
- }
-}
-
-static void push_back_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
- worker->next = &p->root_worker;
- worker->prev = worker->next->prev;
- worker->prev->next = worker->next->prev = worker;
-}
-
-static void push_front_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
- worker->prev = &p->root_worker;
- worker->next = worker->prev->next;
- worker->prev->next = worker->next->prev = worker;
-}
-
-static void kick_append_error(grpc_error **composite, grpc_error *error) {
- if (error == GRPC_ERROR_NONE) return;
- if (*composite == GRPC_ERROR_NONE) {
- *composite = GRPC_ERROR_CREATE("Kick Failure");
- }
- *composite = grpc_error_add_child(*composite, error);
-}
-
-static grpc_error *pollset_kick_ext(grpc_pollset *p,
- grpc_pollset_worker *specific_worker,
- uint32_t flags) {
- GPR_TIMER_BEGIN("pollset_kick_ext", 0);
- grpc_error *error = GRPC_ERROR_NONE;
-
- /* pollset->mu already held */
- if (specific_worker != NULL) {
- if (specific_worker == GRPC_POLLSET_KICK_BROADCAST) {
- GPR_TIMER_BEGIN("pollset_kick_ext.broadcast", 0);
- GPR_ASSERT((flags & GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP) == 0);
- for (specific_worker = p->root_worker.next;
- specific_worker != &p->root_worker;
- specific_worker = specific_worker->next) {
- kick_append_error(
- &error, grpc_wakeup_fd_wakeup(&specific_worker->wakeup_fd->fd));
- }
- p->kicked_without_pollers = true;
- GPR_TIMER_END("pollset_kick_ext.broadcast", 0);
- } else if (gpr_tls_get(&g_current_thread_worker) !=
- (intptr_t)specific_worker) {
- GPR_TIMER_MARK("different_thread_worker", 0);
- if ((flags & GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP) != 0) {
- specific_worker->reevaluate_polling_on_wakeup = true;
- }
- specific_worker->kicked_specifically = true;
- kick_append_error(&error,
- grpc_wakeup_fd_wakeup(&specific_worker->wakeup_fd->fd));
- } else if ((flags & GRPC_POLLSET_CAN_KICK_SELF) != 0) {
- GPR_TIMER_MARK("kick_yoself", 0);
- if ((flags & GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP) != 0) {
- specific_worker->reevaluate_polling_on_wakeup = true;
- }
- specific_worker->kicked_specifically = true;
- kick_append_error(&error,
- grpc_wakeup_fd_wakeup(&specific_worker->wakeup_fd->fd));
- }
- } else if (gpr_tls_get(&g_current_thread_poller) != (intptr_t)p) {
- GPR_ASSERT((flags & GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP) == 0);
- GPR_TIMER_MARK("kick_anonymous", 0);
- specific_worker = pop_front_worker(p);
- if (specific_worker != NULL) {
- if (gpr_tls_get(&g_current_thread_worker) == (intptr_t)specific_worker) {
- GPR_TIMER_MARK("kick_anonymous_not_self", 0);
- push_back_worker(p, specific_worker);
- specific_worker = pop_front_worker(p);
- if ((flags & GRPC_POLLSET_CAN_KICK_SELF) == 0 &&
- gpr_tls_get(&g_current_thread_worker) ==
- (intptr_t)specific_worker) {
- push_back_worker(p, specific_worker);
- specific_worker = NULL;
- }
- }
- if (specific_worker != NULL) {
- GPR_TIMER_MARK("finally_kick", 0);
- push_back_worker(p, specific_worker);
- kick_append_error(
- &error, grpc_wakeup_fd_wakeup(&specific_worker->wakeup_fd->fd));
- }
- } else {
- GPR_TIMER_MARK("kicked_no_pollers", 0);
- p->kicked_without_pollers = true;
- }
- }
-
- GPR_TIMER_END("pollset_kick_ext", 0);
- return error;
-}
-
-static grpc_error *pollset_kick(grpc_pollset *p,
- grpc_pollset_worker *specific_worker) {
- return pollset_kick_ext(p, specific_worker, 0);
-}
-
-/* global state management */
-
-static grpc_error *pollset_global_init(void) {
- gpr_tls_init(&g_current_thread_poller);
- gpr_tls_init(&g_current_thread_worker);
- return grpc_wakeup_fd_init(&grpc_global_wakeup_fd);
-}
-
-static void pollset_global_shutdown(void) {
- grpc_wakeup_fd_destroy(&grpc_global_wakeup_fd);
- gpr_tls_destroy(&g_current_thread_poller);
- gpr_tls_destroy(&g_current_thread_worker);
-}
-
-static grpc_error *kick_poller(void) {
- return grpc_wakeup_fd_wakeup(&grpc_global_wakeup_fd);
-}
-
-/* main interface */
-
-static void become_basic_pollset(grpc_pollset *pollset, grpc_fd *fd_or_null);
-
-static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) {
- gpr_mu_init(&pollset->mu);
- *mu = &pollset->mu;
- pollset->root_worker.next = pollset->root_worker.prev = &pollset->root_worker;
- pollset->in_flight_cbs = 0;
- pollset->shutting_down = 0;
- pollset->called_shutdown = 0;
- pollset->kicked_without_pollers = 0;
- pollset->idle_jobs.head = pollset->idle_jobs.tail = NULL;
- pollset->local_wakeup_cache = NULL;
- pollset->kicked_without_pollers = 0;
- become_basic_pollset(pollset, NULL);
-}
-
-static void pollset_destroy(grpc_pollset *pollset) {
- GPR_ASSERT(pollset->in_flight_cbs == 0);
- GPR_ASSERT(!pollset_has_workers(pollset));
- GPR_ASSERT(pollset->idle_jobs.head == pollset->idle_jobs.tail);
- pollset->vtable->destroy(pollset);
- while (pollset->local_wakeup_cache) {
- grpc_cached_wakeup_fd *next = pollset->local_wakeup_cache->next;
- grpc_wakeup_fd_destroy(&pollset->local_wakeup_cache->fd);
- gpr_free(pollset->local_wakeup_cache);
- pollset->local_wakeup_cache = next;
- }
- gpr_mu_destroy(&pollset->mu);
-}
-
-static void pollset_reset(grpc_pollset *pollset) {
- GPR_ASSERT(pollset->shutting_down);
- GPR_ASSERT(pollset->in_flight_cbs == 0);
- GPR_ASSERT(!pollset_has_workers(pollset));
- GPR_ASSERT(pollset->idle_jobs.head == pollset->idle_jobs.tail);
- pollset->vtable->destroy(pollset);
- pollset->shutting_down = 0;
- pollset->called_shutdown = 0;
- pollset->kicked_without_pollers = 0;
- become_basic_pollset(pollset, NULL);
-}
-
-static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
- grpc_fd *fd) {
- gpr_mu_lock(&pollset->mu);
- pollset->vtable->add_fd(exec_ctx, pollset, fd, 1);
-/* the following (enabled only in debug) will reacquire and then release
- our lock - meaning that if the unlocking flag passed to add_fd above is
- not respected, the code will deadlock (in a way that we have a chance of
- debugging) */
-#ifndef NDEBUG
- gpr_mu_lock(&pollset->mu);
- gpr_mu_unlock(&pollset->mu);
-#endif
-}
-
-static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
- GPR_ASSERT(grpc_closure_list_empty(pollset->idle_jobs));
- pollset->vtable->finish_shutdown(pollset);
- grpc_exec_ctx_sched(exec_ctx, pollset->shutdown_done, GRPC_ERROR_NONE, NULL);
-}
-
-static void work_combine_error(grpc_error **composite, grpc_error *error) {
- if (error == GRPC_ERROR_NONE) return;
- if (*composite == GRPC_ERROR_NONE) {
- *composite = GRPC_ERROR_CREATE("pollset_work");
- }
- *composite = grpc_error_add_child(*composite, error);
-}
-
-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;
- *worker_hdl = &worker;
- grpc_error *error = GRPC_ERROR_NONE;
-
- /* pollset->mu already held */
- int added_worker = 0;
- int locked = 1;
- int queued_work = 0;
- int keep_polling = 0;
- GPR_TIMER_BEGIN("pollset_work", 0);
- /* this must happen before we (potentially) drop pollset->mu */
- worker.next = worker.prev = NULL;
- worker.reevaluate_polling_on_wakeup = 0;
- if (pollset->local_wakeup_cache != NULL) {
- worker.wakeup_fd = pollset->local_wakeup_cache;
- pollset->local_wakeup_cache = worker.wakeup_fd->next;
- } else {
- worker.wakeup_fd = gpr_malloc(sizeof(*worker.wakeup_fd));
- error = grpc_wakeup_fd_init(&worker.wakeup_fd->fd);
- if (error != GRPC_ERROR_NONE) {
- return error;
- }
- }
- worker.kicked_specifically = 0;
- /* If there's work waiting for the pollset to be idle, and the
- pollset is idle, then do that work */
- if (!pollset_has_workers(pollset) &&
- !grpc_closure_list_empty(pollset->idle_jobs)) {
- GPR_TIMER_MARK("pollset_work.idle_jobs", 0);
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
- goto done;
- }
- /* If we're shutting down then we don't execute any extended work */
- if (pollset->shutting_down) {
- GPR_TIMER_MARK("pollset_work.shutting_down", 0);
- goto done;
- }
- /* Give do_promote priority so we don't starve it out */
- if (pollset->in_flight_cbs) {
- GPR_TIMER_MARK("pollset_work.in_flight_cbs", 0);
- gpr_mu_unlock(&pollset->mu);
- locked = 0;
- goto done;
- }
- /* Start polling, and keep doing so while we're being asked to
- re-evaluate our pollers (this allows poll() based pollers to
- ensure they don't miss wakeups) */
- keep_polling = 1;
- while (keep_polling) {
- keep_polling = 0;
- if (!pollset->kicked_without_pollers) {
- if (!added_worker) {
- push_front_worker(pollset, &worker);
- added_worker = 1;
- gpr_tls_set(&g_current_thread_worker, (intptr_t)&worker);
- }
- gpr_tls_set(&g_current_thread_poller, (intptr_t)pollset);
- GPR_TIMER_BEGIN("maybe_work_and_unlock", 0);
- work_combine_error(&error,
- pollset->vtable->maybe_work_and_unlock(
- exec_ctx, pollset, &worker, deadline, now));
- GPR_TIMER_END("maybe_work_and_unlock", 0);
- locked = 0;
- gpr_tls_set(&g_current_thread_poller, 0);
- } else {
- GPR_TIMER_MARK("pollset_work.kicked_without_pollers", 0);
- pollset->kicked_without_pollers = 0;
- }
- /* Finished execution - start cleaning up.
- Note that we may arrive here from outside the enclosing while() loop.
- In that case we won't loop though as we haven't added worker to the
- worker list, which means nobody could ask us to re-evaluate polling). */
- done:
- if (!locked) {
- queued_work |= grpc_exec_ctx_flush(exec_ctx);
- gpr_mu_lock(&pollset->mu);
- locked = 1;
- }
- /* If we're forced to re-evaluate polling (via pollset_kick with
- GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP) then we land here and force
- a loop */
- if (worker.reevaluate_polling_on_wakeup) {
- worker.reevaluate_polling_on_wakeup = 0;
- pollset->kicked_without_pollers = 0;
- if (queued_work || worker.kicked_specifically) {
- /* If there's queued work on the list, then set the deadline to be
- immediate so we get back out of the polling loop quickly */
- deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC);
- }
- keep_polling = 1;
- }
- }
- if (added_worker) {
- remove_worker(pollset, &worker);
- gpr_tls_set(&g_current_thread_worker, 0);
- }
- /* release wakeup fd to the local pool */
- worker.wakeup_fd->next = pollset->local_wakeup_cache;
- pollset->local_wakeup_cache = worker.wakeup_fd;
- /* check shutdown conditions */
- if (pollset->shutting_down) {
- if (pollset_has_workers(pollset)) {
- pollset_kick(pollset, NULL);
- } else if (!pollset->called_shutdown && pollset->in_flight_cbs == 0) {
- pollset->called_shutdown = 1;
- gpr_mu_unlock(&pollset->mu);
- finish_shutdown(exec_ctx, pollset);
- grpc_exec_ctx_flush(exec_ctx);
- /* Continuing to access pollset here is safe -- it is the caller's
- * responsibility to not destroy when it has outstanding calls to
- * pollset_work.
- * TODO(dklempner): Can we refactor the shutdown logic to avoid this? */
- gpr_mu_lock(&pollset->mu);
- } else if (!grpc_closure_list_empty(pollset->idle_jobs)) {
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
- gpr_mu_unlock(&pollset->mu);
- grpc_exec_ctx_flush(exec_ctx);
- gpr_mu_lock(&pollset->mu);
- }
- }
- *worker_hdl = NULL;
- GPR_TIMER_END("pollset_work", 0);
- return error;
-}
-
-static void pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
- grpc_closure *closure) {
- GPR_ASSERT(!pollset->shutting_down);
- pollset->shutting_down = 1;
- pollset->shutdown_done = closure;
- pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST);
- if (!pollset_has_workers(pollset)) {
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
- }
- if (!pollset->called_shutdown && pollset->in_flight_cbs == 0 &&
- !pollset_has_workers(pollset)) {
- pollset->called_shutdown = 1;
- finish_shutdown(exec_ctx, pollset);
- }
-}
-
-static int poll_deadline_to_millis_timeout(gpr_timespec deadline,
- gpr_timespec now) {
- gpr_timespec timeout;
- static const int64_t max_spin_polling_us = 10;
- if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) == 0) {
- return -1;
- }
- if (gpr_time_cmp(deadline, gpr_time_add(now, gpr_time_from_micros(
- max_spin_polling_us,
- GPR_TIMESPAN))) <= 0) {
- return 0;
- }
- timeout = gpr_time_sub(deadline, now);
- return gpr_time_to_millis(gpr_time_add(
- timeout, gpr_time_from_nanos(GPR_NS_PER_MS - 1, GPR_TIMESPAN)));
-}
-
-/*
- * basic_pollset - a vtable that provides polling for zero or one file
- * descriptor via poll()
- */
-
-typedef struct grpc_unary_promote_args {
- const grpc_pollset_vtable *original_vtable;
- grpc_pollset *pollset;
- grpc_fd *fd;
- grpc_closure promotion_closure;
-} grpc_unary_promote_args;
-
-static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args,
- grpc_error *error) {
- grpc_unary_promote_args *up_args = args;
- const grpc_pollset_vtable *original_vtable = up_args->original_vtable;
- grpc_pollset *pollset = up_args->pollset;
- grpc_fd *fd = up_args->fd;
-
- /*
- * This is quite tricky. There are a number of cases to keep in mind here:
- * 1. fd may have been orphaned
- * 2. The pollset may no longer be a unary poller (and we can't let case #1
- * leak to other pollset types!)
- * 3. pollset's fd (which may have changed) may have been orphaned
- * 4. The pollset may be shutting down.
- */
-
- gpr_mu_lock(&pollset->mu);
- /* First we need to ensure that nobody is polling concurrently */
- GPR_ASSERT(!pollset_has_workers(pollset));
-
- gpr_free(up_args);
- /* At this point the pollset may no longer be a unary poller. In that case
- * we should just call the right add function and be done. */
- /* TODO(klempner): If we're not careful this could cause infinite recursion.
- * That's not a problem for now because empty_pollset has a trivial poller
- * and we don't have any mechanism to unbecome multipoller. */
- pollset->in_flight_cbs--;
- if (pollset->shutting_down) {
- /* We don't care about this pollset anymore. */
- if (pollset->in_flight_cbs == 0 && !pollset->called_shutdown) {
- pollset->called_shutdown = 1;
- finish_shutdown(exec_ctx, pollset);
- }
- } else if (fd_is_orphaned(fd)) {
- /* Don't try to add it to anything, we'll drop our ref on it below */
- } else if (pollset->vtable != original_vtable) {
- pollset->vtable->add_fd(exec_ctx, pollset, fd, 0);
- } else if (fd != pollset->data.ptr) {
- grpc_fd *fds[2];
- fds[0] = pollset->data.ptr;
- fds[1] = fd;
-
- if (fds[0] && !fd_is_orphaned(fds[0])) {
- platform_become_multipoller(exec_ctx, pollset, fds, GPR_ARRAY_SIZE(fds));
- GRPC_FD_UNREF(fds[0], "basicpoll");
- } else {
- /* old fd is orphaned and we haven't cleaned it up until now, so remain a
- * unary poller */
- /* Note that it is possible that fds[1] is also orphaned at this point.
- * That's okay, we'll correct it at the next add or poll. */
- if (fds[0]) GRPC_FD_UNREF(fds[0], "basicpoll");
- pollset->data.ptr = fd;
- GRPC_FD_REF(fd, "basicpoll");
- }
- }
-
- gpr_mu_unlock(&pollset->mu);
-
- /* Matching ref in basic_pollset_add_fd */
- GRPC_FD_UNREF(fd, "basicpoll_add");
-}
-
-static void basic_pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
- grpc_fd *fd, int and_unlock_pollset) {
- grpc_unary_promote_args *up_args;
- GPR_ASSERT(fd);
- if (fd == pollset->data.ptr) goto exit;
-
- if (!pollset_has_workers(pollset)) {
- /* Fast path -- no in flight cbs */
- /* TODO(klempner): Comment this out and fix any test failures or establish
- * they are due to timing issues */
- grpc_fd *fds[2];
- fds[0] = pollset->data.ptr;
- fds[1] = fd;
-
- if (fds[0] == NULL) {
- pollset->data.ptr = fd;
- GRPC_FD_REF(fd, "basicpoll");
- } else if (!fd_is_orphaned(fds[0])) {
- platform_become_multipoller(exec_ctx, pollset, fds, GPR_ARRAY_SIZE(fds));
- GRPC_FD_UNREF(fds[0], "basicpoll");
- } else {
- /* old fd is orphaned and we haven't cleaned it up until now, so remain a
- * unary poller */
- GRPC_FD_UNREF(fds[0], "basicpoll");
- pollset->data.ptr = fd;
- GRPC_FD_REF(fd, "basicpoll");
- }
- goto exit;
- }
-
- /* Now we need to promote. This needs to happen when we're not polling. Since
- * this may be called from poll, the wait needs to happen asynchronously. */
- GRPC_FD_REF(fd, "basicpoll_add");
- pollset->in_flight_cbs++;
- up_args = gpr_malloc(sizeof(*up_args));
- up_args->fd = fd;
- up_args->original_vtable = pollset->vtable;
- up_args->pollset = pollset;
- up_args->promotion_closure.cb = basic_do_promote;
- up_args->promotion_closure.cb_arg = up_args;
-
- grpc_closure_list_append(&pollset->idle_jobs, &up_args->promotion_closure,
- GRPC_ERROR_NONE);
- pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST);
-
-exit:
- if (and_unlock_pollset) {
- gpr_mu_unlock(&pollset->mu);
- }
-}
-
-static grpc_error *basic_pollset_maybe_work_and_unlock(
- grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_pollset_worker *worker,
- gpr_timespec deadline, gpr_timespec now) {
-#define POLLOUT_CHECK (POLLOUT | POLLHUP | POLLERR)
-#define POLLIN_CHECK (POLLIN | POLLHUP | POLLERR)
-
- struct pollfd pfd[3];
- grpc_fd *fd;
- grpc_fd_watcher fd_watcher;
- int timeout;
- int r;
- nfds_t nfds;
- grpc_error *error = GRPC_ERROR_NONE;
-
- fd = pollset->data.ptr;
- if (fd && fd_is_orphaned(fd)) {
- GRPC_FD_UNREF(fd, "basicpoll");
- fd = pollset->data.ptr = NULL;
- }
- timeout = poll_deadline_to_millis_timeout(deadline, now);
- pfd[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd);
- pfd[0].events = POLLIN;
- pfd[0].revents = 0;
- pfd[1].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker->wakeup_fd->fd);
- pfd[1].events = POLLIN;
- pfd[1].revents = 0;
- nfds = 2;
- if (fd) {
- pfd[2].fd = fd->fd;
- pfd[2].revents = 0;
- GRPC_FD_REF(fd, "basicpoll_begin");
- gpr_mu_unlock(&pollset->mu);
- pfd[2].events =
- (short)fd_begin_poll(fd, pollset, worker, POLLIN, POLLOUT, &fd_watcher);
- if (pfd[2].events != 0) {
- nfds++;
- }
- } else {
- gpr_mu_unlock(&pollset->mu);
- }
-
- /* TODO(vpai): Consider first doing a 0 timeout poll here to avoid
- even going into the blocking annotation if possible */
- /* poll fd count (argument 2) is shortened by one if we have no events
- to poll on - such that it only includes the kicker */
- GPR_TIMER_BEGIN("poll", 0);
- GRPC_SCHEDULING_START_BLOCKING_REGION;
- r = grpc_poll_function(pfd, nfds, timeout);
- GRPC_SCHEDULING_END_BLOCKING_REGION;
- GPR_TIMER_END("poll", 0);
-
- if (r < 0) {
- if (errno != EINTR) {
- work_combine_error(&error, GRPC_OS_ERROR(errno, "poll"));
- }
- if (fd) {
- fd_end_poll(exec_ctx, &fd_watcher, 0, 0, NULL);
- }
- } else if (r == 0) {
- if (fd) {
- fd_end_poll(exec_ctx, &fd_watcher, 0, 0, NULL);
- }
- } else {
- if (pfd[0].revents & POLLIN_CHECK) {
- work_combine_error(&error,
- grpc_wakeup_fd_consume_wakeup(&grpc_global_wakeup_fd));
- }
- if (pfd[1].revents & POLLIN_CHECK) {
- work_combine_error(&error,
- grpc_wakeup_fd_consume_wakeup(&worker->wakeup_fd->fd));
- }
- if (nfds > 2) {
- fd_end_poll(exec_ctx, &fd_watcher, pfd[2].revents & POLLIN_CHECK,
- pfd[2].revents & POLLOUT_CHECK, pollset);
- } else if (fd) {
- fd_end_poll(exec_ctx, &fd_watcher, 0, 0, NULL);
- }
- }
-
- if (fd) {
- GRPC_FD_UNREF(fd, "basicpoll_begin");
- }
-
- return error;
-}
-
-static void basic_pollset_destroy(grpc_pollset *pollset) {
- if (pollset->data.ptr != NULL) {
- GRPC_FD_UNREF(pollset->data.ptr, "basicpoll");
- pollset->data.ptr = NULL;
- }
-}
-
-static const grpc_pollset_vtable basic_pollset = {
- basic_pollset_add_fd, basic_pollset_maybe_work_and_unlock,
- basic_pollset_destroy, basic_pollset_destroy};
-
-static void become_basic_pollset(grpc_pollset *pollset, grpc_fd *fd_or_null) {
- pollset->vtable = &basic_pollset;
- pollset->data.ptr = fd_or_null;
- if (fd_or_null != NULL) {
- GRPC_FD_REF(fd_or_null, "basicpoll");
- }
-}
-
-/*******************************************************************************
- * pollset_multipoller_with_poll_posix.c
- */
-
-#ifndef GRPC_LINUX_MULTIPOLL_WITH_EPOLL
-
-typedef struct {
- /* all polled fds */
- size_t fd_count;
- size_t fd_capacity;
- grpc_fd **fds;
- /* fds that have been removed from the pollset explicitly */
- size_t del_count;
- size_t del_capacity;
- grpc_fd **dels;
-} poll_hdr;
-
-static void multipoll_with_poll_pollset_add_fd(grpc_exec_ctx *exec_ctx,
- grpc_pollset *pollset,
- grpc_fd *fd,
- int and_unlock_pollset) {
- size_t i;
- poll_hdr *h = pollset->data.ptr;
- /* TODO(ctiller): this is O(num_fds^2); maybe switch to a hash set here */
- for (i = 0; i < h->fd_count; i++) {
- if (h->fds[i] == fd) goto exit;
- }
- if (h->fd_count == h->fd_capacity) {
- h->fd_capacity = GPR_MAX(h->fd_capacity + 8, h->fd_count * 3 / 2);
- h->fds = gpr_realloc(h->fds, sizeof(grpc_fd *) * h->fd_capacity);
- }
- h->fds[h->fd_count++] = fd;
- GRPC_FD_REF(fd, "multipoller");
-exit:
- if (and_unlock_pollset) {
- gpr_mu_unlock(&pollset->mu);
- }
-}
-
-static grpc_error *multipoll_with_poll_pollset_maybe_work_and_unlock(
- grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_pollset_worker *worker,
- gpr_timespec deadline, gpr_timespec now) {
-#define POLLOUT_CHECK (POLLOUT | POLLHUP | POLLERR)
-#define POLLIN_CHECK (POLLIN | POLLHUP | POLLERR)
-
- int timeout;
- int r;
- size_t i, j, fd_count;
- nfds_t pfd_count;
- poll_hdr *h;
- /* TODO(ctiller): inline some elements to avoid an allocation */
- grpc_fd_watcher *watchers;
- struct pollfd *pfds;
- grpc_error *error = GRPC_ERROR_NONE;
-
- h = pollset->data.ptr;
- timeout = poll_deadline_to_millis_timeout(deadline, now);
- /* TODO(ctiller): perform just one malloc here if we exceed the inline case */
- pfds = gpr_malloc(sizeof(*pfds) * (h->fd_count + 2));
- watchers = gpr_malloc(sizeof(*watchers) * (h->fd_count + 2));
- fd_count = 0;
- pfd_count = 2;
- pfds[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd);
- pfds[0].events = POLLIN;
- pfds[0].revents = 0;
- pfds[1].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker->wakeup_fd->fd);
- pfds[1].events = POLLIN;
- pfds[1].revents = 0;
- for (i = 0; i < h->fd_count; i++) {
- int remove = fd_is_orphaned(h->fds[i]);
- for (j = 0; !remove && j < h->del_count; j++) {
- if (h->fds[i] == h->dels[j]) remove = 1;
- }
- if (remove) {
- GRPC_FD_UNREF(h->fds[i], "multipoller");
- } else {
- h->fds[fd_count++] = h->fds[i];
- watchers[pfd_count].fd = h->fds[i];
- GRPC_FD_REF(watchers[pfd_count].fd, "multipoller_start");
- pfds[pfd_count].fd = h->fds[i]->fd;
- pfds[pfd_count].revents = 0;
- pfd_count++;
- }
- }
- for (j = 0; j < h->del_count; j++) {
- GRPC_FD_UNREF(h->dels[j], "multipoller_del");
- }
- h->del_count = 0;
- h->fd_count = fd_count;
- gpr_mu_unlock(&pollset->mu);
-
- for (i = 2; i < pfd_count; i++) {
- grpc_fd *fd = watchers[i].fd;
- pfds[i].events = (short)fd_begin_poll(fd, pollset, worker, POLLIN, POLLOUT,
- &watchers[i]);
- GRPC_FD_UNREF(fd, "multipoller_start");
- }
-
- /* TODO(vpai): Consider first doing a 0 timeout poll here to avoid
- even going into the blocking annotation if possible */
- GRPC_SCHEDULING_START_BLOCKING_REGION;
- r = grpc_poll_function(pfds, pfd_count, timeout);
- GRPC_SCHEDULING_END_BLOCKING_REGION;
-
- if (r < 0) {
- if (errno != EINTR) {
- work_combine_error(&error, GRPC_OS_ERROR(errno, "poll"));
- }
- for (i = 2; i < pfd_count; i++) {
- fd_end_poll(exec_ctx, &watchers[i], 0, 0, NULL);
- }
- } else if (r == 0) {
- for (i = 2; i < pfd_count; i++) {
- fd_end_poll(exec_ctx, &watchers[i], 0, 0, NULL);
- }
- } else {
- if (pfds[0].revents & POLLIN_CHECK) {
- work_combine_error(&error,
- grpc_wakeup_fd_consume_wakeup(&grpc_global_wakeup_fd));
- }
- if (pfds[1].revents & POLLIN_CHECK) {
- work_combine_error(&error,
- grpc_wakeup_fd_consume_wakeup(&worker->wakeup_fd->fd));
- }
- for (i = 2; i < pfd_count; i++) {
- if (watchers[i].fd == NULL) {
- fd_end_poll(exec_ctx, &watchers[i], 0, 0, NULL);
- continue;
- }
- fd_end_poll(exec_ctx, &watchers[i], pfds[i].revents & POLLIN_CHECK,
- pfds[i].revents & POLLOUT_CHECK, pollset);
- }
- }
-
- gpr_free(pfds);
- gpr_free(watchers);
-
- return error;
-}
-
-static void multipoll_with_poll_pollset_finish_shutdown(grpc_pollset *pollset) {
- size_t i;
- poll_hdr *h = pollset->data.ptr;
- for (i = 0; i < h->fd_count; i++) {
- GRPC_FD_UNREF(h->fds[i], "multipoller");
- }
- for (i = 0; i < h->del_count; i++) {
- GRPC_FD_UNREF(h->dels[i], "multipoller_del");
- }
- h->fd_count = 0;
- h->del_count = 0;
-}
-
-static void multipoll_with_poll_pollset_destroy(grpc_pollset *pollset) {
- poll_hdr *h = pollset->data.ptr;
- multipoll_with_poll_pollset_finish_shutdown(pollset);
- gpr_free(h->fds);
- gpr_free(h->dels);
- gpr_free(h);
-}
-
-static const grpc_pollset_vtable multipoll_with_poll_pollset = {
- multipoll_with_poll_pollset_add_fd,
- multipoll_with_poll_pollset_maybe_work_and_unlock,
- multipoll_with_poll_pollset_finish_shutdown,
- multipoll_with_poll_pollset_destroy};
-
-static void poll_become_multipoller(grpc_exec_ctx *exec_ctx,
- grpc_pollset *pollset, grpc_fd **fds,
- size_t nfds) {
- size_t i;
- poll_hdr *h = gpr_malloc(sizeof(poll_hdr));
- pollset->vtable = &multipoll_with_poll_pollset;
- pollset->data.ptr = h;
- h->fd_count = nfds;
- h->fd_capacity = nfds;
- h->fds = gpr_malloc(nfds * sizeof(grpc_fd *));
- h->del_count = 0;
- h->del_capacity = 0;
- h->dels = NULL;
- for (i = 0; i < nfds; i++) {
- h->fds[i] = fds[i];
- GRPC_FD_REF(fds[i], "multipoller");
- }
-}
-
-#endif /* !GRPC_LINUX_MULTIPOLL_WITH_EPOLL */
-
-/*******************************************************************************
- * pollset_multipoller_with_epoll_posix.c
- */
-
-#ifdef GRPC_LINUX_MULTIPOLL_WITH_EPOLL
-
-#include <errno.h>
-#include <poll.h>
-#include <string.h>
-#include <sys/epoll.h>
-#include <unistd.h>
-
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
-#include <grpc/support/useful.h>
-
-#include "src/core/lib/iomgr/ev_posix.h"
-#include "src/core/lib/profiling/timers.h"
-#include "src/core/lib/support/block_annotate.h"
-
-static void set_ready(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure **st,
- grpc_pollset *read_notifier_pollset) {
- /* only one set_ready can be active at once (but there may be a racing
- notify_on) */
- gpr_mu_lock(&fd->mu);
- set_ready_locked(exec_ctx, fd, st);
-
- /* A non-NULL read_notifier_pollset means that the fd is readable. */
- if (read_notifier_pollset != NULL) {
- /* Note: Since the fd might be a part of multiple pollsets, this might be
- * called multiple times (for each time the fd becomes readable) and it is
- * okay to set the fd's read-notifier pollset to anyone of these pollsets */
- set_read_notifier_pollset_locked(exec_ctx, fd, read_notifier_pollset);
- }
-
- gpr_mu_unlock(&fd->mu);
-}
-
-static void fd_become_readable(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
- grpc_pollset *notifier_pollset) {
- set_ready(exec_ctx, fd, &fd->read_closure, notifier_pollset);
-}
-
-static void fd_become_writable(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
- set_ready(exec_ctx, fd, &fd->write_closure, NULL);
-}
-
-struct epoll_fd_list {
- int *epoll_fds;
- size_t count;
- size_t capacity;
-};
-
-static struct epoll_fd_list epoll_fd_global_list;
-static gpr_once init_epoll_fd_list_mu = GPR_ONCE_INIT;
-static gpr_mu epoll_fd_list_mu;
-
-static void init_mu(void) { gpr_mu_init(&epoll_fd_list_mu); }
-
-static void add_epoll_fd_to_global_list(int epoll_fd) {
- gpr_once_init(&init_epoll_fd_list_mu, init_mu);
-
- gpr_mu_lock(&epoll_fd_list_mu);
- if (epoll_fd_global_list.count == epoll_fd_global_list.capacity) {
- epoll_fd_global_list.capacity =
- GPR_MAX((size_t)8, epoll_fd_global_list.capacity * 2);
- epoll_fd_global_list.epoll_fds =
- gpr_realloc(epoll_fd_global_list.epoll_fds,
- epoll_fd_global_list.capacity * sizeof(int));
- }
- epoll_fd_global_list.epoll_fds[epoll_fd_global_list.count++] = epoll_fd;
- gpr_mu_unlock(&epoll_fd_list_mu);
-}
-
-static void remove_epoll_fd_from_global_list(int epoll_fd) {
- gpr_mu_lock(&epoll_fd_list_mu);
- GPR_ASSERT(epoll_fd_global_list.count > 0);
- for (size_t i = 0; i < epoll_fd_global_list.count; i++) {
- if (epoll_fd == epoll_fd_global_list.epoll_fds[i]) {
- epoll_fd_global_list.epoll_fds[i] =
- epoll_fd_global_list.epoll_fds[--(epoll_fd_global_list.count)];
- break;
- }
- }
- gpr_mu_unlock(&epoll_fd_list_mu);
-}
-
-static void remove_fd_from_all_epoll_sets(int fd) {
- int err;
- gpr_once_init(&init_epoll_fd_list_mu, init_mu);
- gpr_mu_lock(&epoll_fd_list_mu);
- if (epoll_fd_global_list.count == 0) {
- gpr_mu_unlock(&epoll_fd_list_mu);
- return;
- }
- for (size_t i = 0; i < epoll_fd_global_list.count; i++) {
- err = epoll_ctl(epoll_fd_global_list.epoll_fds[i], EPOLL_CTL_DEL, fd, NULL);
- if (err < 0 && errno != ENOENT) {
- gpr_log(GPR_ERROR, "epoll_ctl del for %d failed: %s", fd,
- strerror(errno));
- }
- }
- gpr_mu_unlock(&epoll_fd_list_mu);
-}
-
-typedef struct {
- grpc_pollset *pollset;
- grpc_fd *fd;
- grpc_closure closure;
-} delayed_add;
-
-typedef struct { int epoll_fd; } epoll_hdr;
-
-static void finally_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
- grpc_fd *fd) {
- epoll_hdr *h = pollset->data.ptr;
- struct epoll_event ev;
- int err;
- grpc_fd_watcher watcher;
-
- /* We pretend to be polling whilst adding an fd to keep the fd from being
- closed during the add. This may result in a spurious wakeup being assigned
- to this pollset whilst adding, but that should be benign. */
- GPR_ASSERT(fd_begin_poll(fd, pollset, NULL, 0, 0, &watcher) == 0);
- if (watcher.fd != NULL) {
- ev.events = (uint32_t)(EPOLLIN | EPOLLOUT | EPOLLET);
- ev.data.ptr = fd;
- err = epoll_ctl(h->epoll_fd, EPOLL_CTL_ADD, fd->fd, &ev);
- if (err < 0) {
- /* FDs may be added to a pollset multiple times, so EEXIST is normal. */
- if (errno != EEXIST) {
- gpr_log(GPR_ERROR, "epoll_ctl add for %d failed: %s", fd->fd,
- strerror(errno));
- }
- }
- }
- fd_end_poll(exec_ctx, &watcher, 0, 0, NULL);
-}
-
-static void perform_delayed_add(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- delayed_add *da = arg;
-
- if (!fd_is_orphaned(da->fd)) {
- finally_add_fd(exec_ctx, da->pollset, da->fd);
- }
-
- gpr_mu_lock(&da->pollset->mu);
- da->pollset->in_flight_cbs--;
- if (da->pollset->shutting_down) {
- /* We don't care about this pollset anymore. */
- if (da->pollset->in_flight_cbs == 0 && !da->pollset->called_shutdown) {
- da->pollset->called_shutdown = 1;
- grpc_exec_ctx_sched(exec_ctx, da->pollset->shutdown_done, GRPC_ERROR_NONE,
- NULL);
- }
- }
- gpr_mu_unlock(&da->pollset->mu);
-
- GRPC_FD_UNREF(da->fd, "delayed_add");
-
- gpr_free(da);
-}
-
-static void multipoll_with_epoll_pollset_add_fd(grpc_exec_ctx *exec_ctx,
- grpc_pollset *pollset,
- grpc_fd *fd,
- int and_unlock_pollset) {
- if (and_unlock_pollset) {
- gpr_mu_unlock(&pollset->mu);
- finally_add_fd(exec_ctx, pollset, fd);
- } else {
- delayed_add *da = gpr_malloc(sizeof(*da));
- da->pollset = pollset;
- da->fd = fd;
- GRPC_FD_REF(fd, "delayed_add");
- grpc_closure_init(&da->closure, perform_delayed_add, da);
- pollset->in_flight_cbs++;
- grpc_exec_ctx_sched(exec_ctx, &da->closure, GRPC_ERROR_NONE, NULL);
- }
-}
-
-/* TODO(klempner): We probably want to turn this down a bit */
-#define GRPC_EPOLL_MAX_EVENTS 1000
-
-static grpc_error *multipoll_with_epoll_pollset_maybe_work_and_unlock(
- grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_pollset_worker *worker,
- gpr_timespec deadline, gpr_timespec now) {
- struct epoll_event ep_ev[GRPC_EPOLL_MAX_EVENTS];
- int ep_rv;
- int poll_rv;
- epoll_hdr *h = pollset->data.ptr;
- int timeout_ms;
- struct pollfd pfds[2];
- grpc_error *error = GRPC_ERROR_NONE;
-
- /* If you want to ignore epoll's ability to sanely handle parallel pollers,
- * for a more apples-to-apples performance comparison with poll, add a
- * if (pollset->counter != 0) { return 0; }
- * here.
- */
-
- gpr_mu_unlock(&pollset->mu);
-
- timeout_ms = poll_deadline_to_millis_timeout(deadline, now);
-
- pfds[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker->wakeup_fd->fd);
- pfds[0].events = POLLIN;
- pfds[0].revents = 0;
- pfds[1].fd = h->epoll_fd;
- pfds[1].events = POLLIN;
- pfds[1].revents = 0;
-
- /* TODO(vpai): Consider first doing a 0 timeout poll here to avoid
- even going into the blocking annotation if possible */
- GPR_TIMER_BEGIN("poll", 0);
- GRPC_SCHEDULING_START_BLOCKING_REGION;
- poll_rv = grpc_poll_function(pfds, 2, timeout_ms);
- GRPC_SCHEDULING_END_BLOCKING_REGION;
- GPR_TIMER_END("poll", 0);
-
- if (poll_rv < 0) {
- if (errno != EINTR) {
- work_combine_error(&error, GRPC_OS_ERROR(errno, "poll"));
- }
- } else if (poll_rv == 0) {
- /* do nothing */
- } else {
- if (pfds[0].revents) {
- work_combine_error(&error,
- grpc_wakeup_fd_consume_wakeup(&worker->wakeup_fd->fd));
- }
- if (pfds[1].revents) {
- do {
- /* The following epoll_wait never blocks; it has a timeout of 0 */
- ep_rv = epoll_wait(h->epoll_fd, ep_ev, GRPC_EPOLL_MAX_EVENTS, 0);
- if (ep_rv < 0) {
- if (errno != EINTR) {
- work_combine_error(&error, GRPC_OS_ERROR(errno, "epoll_wait"));
- }
- } else {
- int i;
- for (i = 0; i < ep_rv; ++i) {
- grpc_fd *fd = ep_ev[i].data.ptr;
- /* TODO(klempner): We might want to consider making err and pri
- * separate events */
- int cancel = ep_ev[i].events & (EPOLLERR | EPOLLHUP);
- int read_ev = ep_ev[i].events & (EPOLLIN | EPOLLPRI);
- int write_ev = ep_ev[i].events & EPOLLOUT;
- if (fd == NULL) {
- work_combine_error(&error, grpc_wakeup_fd_consume_wakeup(
- &grpc_global_wakeup_fd));
- } else {
- if (read_ev || cancel) {
- fd_become_readable(exec_ctx, fd, pollset);
- }
- if (write_ev || cancel) {
- fd_become_writable(exec_ctx, fd);
- }
- }
- }
- }
- } while (ep_rv == GRPC_EPOLL_MAX_EVENTS);
- }
- }
- return error;
-}
-
-static void multipoll_with_epoll_pollset_finish_shutdown(
- grpc_pollset *pollset) {}
-
-static void multipoll_with_epoll_pollset_destroy(grpc_pollset *pollset) {
- epoll_hdr *h = pollset->data.ptr;
- close(h->epoll_fd);
- remove_epoll_fd_from_global_list(h->epoll_fd);
- gpr_free(h);
-}
-
-static const grpc_pollset_vtable multipoll_with_epoll_pollset = {
- multipoll_with_epoll_pollset_add_fd,
- multipoll_with_epoll_pollset_maybe_work_and_unlock,
- multipoll_with_epoll_pollset_finish_shutdown,
- multipoll_with_epoll_pollset_destroy};
-
-static void epoll_become_multipoller(grpc_exec_ctx *exec_ctx,
- grpc_pollset *pollset, grpc_fd **fds,
- size_t nfds) {
- size_t i;
- epoll_hdr *h = gpr_malloc(sizeof(epoll_hdr));
- struct epoll_event ev;
- int err;
-
- pollset->vtable = &multipoll_with_epoll_pollset;
- pollset->data.ptr = h;
- h->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
- if (h->epoll_fd < 0) {
- /* TODO(klempner): Fall back to poll here, especially on ENOSYS */
- gpr_log(GPR_ERROR, "epoll_create1 failed: %s", strerror(errno));
- abort();
- }
- add_epoll_fd_to_global_list(h->epoll_fd);
-
- ev.events = (uint32_t)(EPOLLIN | EPOLLET);
- ev.data.ptr = NULL;
- err = epoll_ctl(h->epoll_fd, EPOLL_CTL_ADD,
- GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd), &ev);
- if (err < 0) {
- gpr_log(GPR_ERROR, "epoll_ctl add for %d failed: %s",
- GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd),
- strerror(errno));
- }
-
- for (i = 0; i < nfds; i++) {
- multipoll_with_epoll_pollset_add_fd(exec_ctx, pollset, fds[i], 0);
- }
-}
-
-#else /* GRPC_LINUX_MULTIPOLL_WITH_EPOLL */
-
-static void remove_fd_from_all_epoll_sets(int fd) {}
-
-#endif /* GRPC_LINUX_MULTIPOLL_WITH_EPOLL */
-
-/*******************************************************************************
- * pollset_set_posix.c
- */
-
-static grpc_pollset_set *pollset_set_create(void) {
- grpc_pollset_set *pollset_set = gpr_malloc(sizeof(*pollset_set));
- memset(pollset_set, 0, sizeof(*pollset_set));
- gpr_mu_init(&pollset_set->mu);
- return pollset_set;
-}
-
-static void pollset_set_destroy(grpc_pollset_set *pollset_set) {
- size_t i;
- gpr_mu_destroy(&pollset_set->mu);
- for (i = 0; i < pollset_set->fd_count; i++) {
- GRPC_FD_UNREF(pollset_set->fds[i], "pollset_set");
- }
- gpr_free(pollset_set->pollsets);
- gpr_free(pollset_set->pollset_sets);
- gpr_free(pollset_set->fds);
- gpr_free(pollset_set);
-}
-
-static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx,
- grpc_pollset_set *pollset_set,
- grpc_pollset *pollset) {
- size_t i, j;
- gpr_mu_lock(&pollset_set->mu);
- if (pollset_set->pollset_count == pollset_set->pollset_capacity) {
- pollset_set->pollset_capacity =
- GPR_MAX(8, 2 * pollset_set->pollset_capacity);
- pollset_set->pollsets =
- gpr_realloc(pollset_set->pollsets, pollset_set->pollset_capacity *
- sizeof(*pollset_set->pollsets));
- }
- pollset_set->pollsets[pollset_set->pollset_count++] = pollset;
- for (i = 0, j = 0; i < pollset_set->fd_count; i++) {
- if (fd_is_orphaned(pollset_set->fds[i])) {
- GRPC_FD_UNREF(pollset_set->fds[i], "pollset_set");
- } else {
- pollset_add_fd(exec_ctx, pollset, pollset_set->fds[i]);
- pollset_set->fds[j++] = pollset_set->fds[i];
- }
- }
- pollset_set->fd_count = j;
- gpr_mu_unlock(&pollset_set->mu);
-}
-
-static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx,
- grpc_pollset_set *pollset_set,
- grpc_pollset *pollset) {
- size_t i;
- gpr_mu_lock(&pollset_set->mu);
- for (i = 0; i < pollset_set->pollset_count; i++) {
- if (pollset_set->pollsets[i] == pollset) {
- pollset_set->pollset_count--;
- GPR_SWAP(grpc_pollset *, pollset_set->pollsets[i],
- pollset_set->pollsets[pollset_set->pollset_count]);
- break;
- }
- }
- gpr_mu_unlock(&pollset_set->mu);
-}
-
-static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx,
- grpc_pollset_set *bag,
- grpc_pollset_set *item) {
- size_t i, j;
- gpr_mu_lock(&bag->mu);
- if (bag->pollset_set_count == bag->pollset_set_capacity) {
- bag->pollset_set_capacity = GPR_MAX(8, 2 * bag->pollset_set_capacity);
- bag->pollset_sets =
- gpr_realloc(bag->pollset_sets,
- bag->pollset_set_capacity * sizeof(*bag->pollset_sets));
- }
- bag->pollset_sets[bag->pollset_set_count++] = item;
- for (i = 0, j = 0; i < bag->fd_count; i++) {
- if (fd_is_orphaned(bag->fds[i])) {
- GRPC_FD_UNREF(bag->fds[i], "pollset_set");
- } else {
- pollset_set_add_fd(exec_ctx, item, bag->fds[i]);
- bag->fds[j++] = bag->fds[i];
- }
- }
- bag->fd_count = j;
- gpr_mu_unlock(&bag->mu);
-}
-
-static void pollset_set_del_pollset_set(grpc_exec_ctx *exec_ctx,
- grpc_pollset_set *bag,
- grpc_pollset_set *item) {
- size_t i;
- gpr_mu_lock(&bag->mu);
- for (i = 0; i < bag->pollset_set_count; i++) {
- if (bag->pollset_sets[i] == item) {
- bag->pollset_set_count--;
- GPR_SWAP(grpc_pollset_set *, bag->pollset_sets[i],
- bag->pollset_sets[bag->pollset_set_count]);
- break;
- }
- }
- gpr_mu_unlock(&bag->mu);
-}
-
-static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx,
- grpc_pollset_set *pollset_set, grpc_fd *fd) {
- size_t i;
- gpr_mu_lock(&pollset_set->mu);
- if (pollset_set->fd_count == pollset_set->fd_capacity) {
- pollset_set->fd_capacity = GPR_MAX(8, 2 * pollset_set->fd_capacity);
- pollset_set->fds = gpr_realloc(
- pollset_set->fds, pollset_set->fd_capacity * sizeof(*pollset_set->fds));
- }
- GRPC_FD_REF(fd, "pollset_set");
- pollset_set->fds[pollset_set->fd_count++] = fd;
- for (i = 0; i < pollset_set->pollset_count; i++) {
- pollset_add_fd(exec_ctx, pollset_set->pollsets[i], fd);
- }
- for (i = 0; i < pollset_set->pollset_set_count; i++) {
- pollset_set_add_fd(exec_ctx, pollset_set->pollset_sets[i], fd);
- }
- gpr_mu_unlock(&pollset_set->mu);
-}
-
-static void pollset_set_del_fd(grpc_exec_ctx *exec_ctx,
- grpc_pollset_set *pollset_set, grpc_fd *fd) {
- size_t i;
- gpr_mu_lock(&pollset_set->mu);
- for (i = 0; i < pollset_set->fd_count; i++) {
- if (pollset_set->fds[i] == fd) {
- pollset_set->fd_count--;
- GPR_SWAP(grpc_fd *, pollset_set->fds[i],
- pollset_set->fds[pollset_set->fd_count]);
- GRPC_FD_UNREF(fd, "pollset_set");
- break;
- }
- }
- for (i = 0; i < pollset_set->pollset_set_count; i++) {
- pollset_set_del_fd(exec_ctx, pollset_set->pollset_sets[i], fd);
- }
- gpr_mu_unlock(&pollset_set->mu);
-}
-
-/*******************************************************************************
- * workqueue stubs
- */
-
-#ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG
-static grpc_workqueue *workqueue_ref(grpc_workqueue *workqueue,
- const char *file, int line,
- const char *reason) {
- return workqueue;
-}
-static void workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue,
- const char *file, int line, const char *reason) {}
-#else
-static grpc_workqueue *workqueue_ref(grpc_workqueue *workqueue) {
- return workqueue;
-}
-static void workqueue_unref(grpc_exec_ctx *exec_ctx,
- grpc_workqueue *workqueue) {}
-#endif
-
-static void workqueue_enqueue(grpc_exec_ctx *exec_ctx,
- grpc_workqueue *workqueue, grpc_closure *closure,
- grpc_error *error) {
- grpc_exec_ctx_sched(exec_ctx, closure, error, NULL);
-}
-
-/*******************************************************************************
- * 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_reset = pollset_reset,
- .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_enqueue = workqueue_enqueue,
-
- .shutdown_engine = shutdown_engine,
-};
-
-const grpc_event_engine_vtable *grpc_init_poll_and_epoll_posix(void) {
-#ifdef GRPC_LINUX_MULTIPOLL_WITH_EPOLL
- platform_become_multipoller = epoll_become_multipoller;
-#else
- platform_become_multipoller = poll_become_multipoller;
-#endif
- fd_global_init();
- pollset_global_init();
- return &vtable;
-}
-
-#endif
diff --git a/src/core/lib/iomgr/ev_poll_and_epoll_posix.h b/src/core/lib/iomgr/ev_poll_and_epoll_posix.h
deleted file mode 100644
index 06d6dbf29d..0000000000
--- a/src/core/lib/iomgr/ev_poll_and_epoll_posix.h
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#ifndef GRPC_CORE_LIB_IOMGR_EV_POLL_AND_EPOLL_POSIX_H
-#define GRPC_CORE_LIB_IOMGR_EV_POLL_AND_EPOLL_POSIX_H
-
-#include "src/core/lib/iomgr/ev_posix.h"
-
-const grpc_event_engine_vtable *grpc_init_poll_and_epoll_posix(void);
-
-#endif /* GRPC_CORE_LIB_IOMGR_EV_POLL_AND_EPOLL_POSIX_H */
diff --git a/src/core/lib/iomgr/ev_posix.c b/src/core/lib/iomgr/ev_posix.c
index ef36ba89b2..ab139895fd 100644
--- a/src/core/lib/iomgr/ev_posix.c
+++ b/src/core/lib/iomgr/ev_posix.c
@@ -45,7 +45,6 @@
#include <grpc/support/useful.h>
#include "src/core/lib/iomgr/ev_epoll_linux.h"
-#include "src/core/lib/iomgr/ev_poll_and_epoll_posix.h"
#include "src/core/lib/iomgr/ev_poll_posix.h"
#include "src/core/lib/support/env.h"
@@ -67,7 +66,6 @@ static const event_engine_factory g_factories[] = {
{"epoll", grpc_init_epoll_linux},
{"poll", grpc_init_poll_posix},
{"poll-cv", grpc_init_poll_cv_posix},
- {"legacy", grpc_init_poll_and_epoll_posix},
};
static void add(const char *beg, const char *end, char ***ss, size_t *ns) {
diff --git a/src/core/lib/iomgr/load_file.c b/src/core/lib/iomgr/load_file.c
index b62ecbc534..217bc5da59 100644
--- a/src/core/lib/iomgr/load_file.c
+++ b/src/core/lib/iomgr/load_file.c
@@ -44,10 +44,10 @@
#include "src/core/lib/support/string.h"
grpc_error *grpc_load_file(const char *filename, int add_null_terminator,
- gpr_slice *output) {
+ grpc_slice *output) {
unsigned char *contents = NULL;
size_t contents_size = 0;
- gpr_slice result = gpr_empty_slice();
+ grpc_slice result = gpr_empty_slice();
FILE *file;
size_t bytes_read = 0;
grpc_error *error = GRPC_ERROR_NONE;
@@ -72,7 +72,7 @@ grpc_error *grpc_load_file(const char *filename, int add_null_terminator,
if (add_null_terminator) {
contents[contents_size++] = 0;
}
- result = gpr_slice_new(contents, contents_size, gpr_free);
+ result = grpc_slice_new(contents, contents_size, gpr_free);
end:
*output = result;
diff --git a/src/core/lib/iomgr/load_file.h b/src/core/lib/iomgr/load_file.h
index 9aac2225d1..73ee8c3abf 100644
--- a/src/core/lib/iomgr/load_file.h
+++ b/src/core/lib/iomgr/load_file.h
@@ -36,7 +36,7 @@
#include <stdio.h>
-#include <grpc/support/slice.h>
+#include <grpc/slice.h>
#include "src/core/lib/iomgr/error.h"
@@ -47,7 +47,7 @@ extern "C" {
/* Loads the content of a file into a slice. add_null_terminator will add
a NULL terminator if non-zero. */
grpc_error *grpc_load_file(const char *filename, int add_null_terminator,
- gpr_slice *slice);
+ grpc_slice *slice);
#ifdef __cplusplus
}
diff --git a/src/core/lib/iomgr/resource_quota.c b/src/core/lib/iomgr/resource_quota.c
index 8a06443d58..051a30baa3 100644
--- a/src/core/lib/iomgr/resource_quota.c
+++ b/src/core/lib/iomgr/resource_quota.c
@@ -44,6 +44,81 @@
int grpc_resource_quota_trace = 0;
+/* Internal linked list pointers for a resource user */
+typedef struct {
+ grpc_resource_user *next;
+ grpc_resource_user *prev;
+} grpc_resource_user_link;
+
+/* Resource users are kept in (potentially) several intrusive linked lists
+ at once. These are the list names. */
+typedef enum {
+ /* Resource users that are waiting for an allocation */
+ GRPC_RULIST_AWAITING_ALLOCATION,
+ /* Resource users that have free memory available for internal reclamation */
+ GRPC_RULIST_NON_EMPTY_FREE_POOL,
+ /* Resource users that have published a benign reclamation is available */
+ GRPC_RULIST_RECLAIMER_BENIGN,
+ /* Resource users that have published a destructive reclamation is
+ available */
+ GRPC_RULIST_RECLAIMER_DESTRUCTIVE,
+ /* Number of lists: must be last */
+ GRPC_RULIST_COUNT
+} grpc_rulist;
+
+struct grpc_resource_user {
+ /* The quota this resource user consumes from */
+ grpc_resource_quota *resource_quota;
+
+ /* Closure to schedule an allocation under the resource quota combiner lock */
+ grpc_closure allocate_closure;
+ /* Closure to publish a non empty free pool under the resource quota combiner
+ lock */
+ grpc_closure add_to_free_pool_closure;
+
+ /* one ref for each ref call (released by grpc_resource_user_unref), and one
+ ref for each byte allocated (released by grpc_resource_user_free) */
+ gpr_atm refs;
+ /* is this resource user unlocked? starts at 0, increases for each shutdown
+ call */
+ gpr_atm shutdown;
+
+ gpr_mu mu;
+ /* The amount of memory (in bytes) this user has cached for its own use: to
+ avoid quota contention, each resource user can keep some memory in
+ addition to what it is immediately using (e.g., for caching), and the quota
+ can pull it back under memory pressure.
+ This value can become negative if more memory has been requested than
+ existed in the free pool, at which point the quota is consulted to bring
+ this value non-negative (asynchronously). */
+ int64_t free_pool;
+ /* A list of closures to call once free_pool becomes non-negative - ie when
+ all outstanding allocations have been granted. */
+ grpc_closure_list on_allocated;
+ /* True if we are currently trying to allocate from the quota, false if not */
+ bool allocating;
+ /* True if we are currently trying to add ourselves to the non-free quota
+ list, false otherwise */
+ bool added_to_free_pool;
+
+ /* Reclaimers: index 0 is the benign reclaimer, 1 is the destructive reclaimer
+ */
+ grpc_closure *reclaimers[2];
+ /* Trampoline closures to finish reclamation and re-enter the quota combiner
+ lock */
+ grpc_closure post_reclaimer_closure[2];
+
+ /* Closure to execute under the quota combiner to de-register and shutdown the
+ resource user */
+ grpc_closure destroy_closure;
+
+ /* Links in the various grpc_rulist lists */
+ grpc_resource_user_link links[GRPC_RULIST_COUNT];
+
+ /* The name of this resource user, for debugging/tracing */
+ char *name;
+};
+
struct grpc_resource_quota {
/* refcount */
gpr_refcount refs;
@@ -272,7 +347,7 @@ static bool rq_reclaim(grpc_exec_ctx *exec_ctx,
*/
typedef struct {
- gpr_slice_refcount base;
+ grpc_slice_refcount base;
gpr_refcount refs;
grpc_resource_user *resource_user;
size_t size;
@@ -289,7 +364,7 @@ static void ru_slice_unref(void *p) {
/* TODO(ctiller): this is dangerous, but I think safe for now:
we have no guarantee here that we're at a safe point for creating an
execution context, but we have no way of writing this code otherwise.
- In the future: consider lifting gpr_slice to grpc, and offering an
+ In the future: consider lifting grpc_slice to grpc, and offering an
internal_{ref,unref} pair that is execution context aware.
Alternatively,
make exec_ctx be thread local and 'do the right thing' (whatever that
@@ -302,15 +377,15 @@ static void ru_slice_unref(void *p) {
}
}
-static gpr_slice ru_slice_create(grpc_resource_user *resource_user,
- size_t size) {
+static grpc_slice ru_slice_create(grpc_resource_user *resource_user,
+ size_t size) {
ru_slice_refcount *rc = gpr_malloc(sizeof(ru_slice_refcount) + size);
rc->base.ref = ru_slice_ref;
rc->base.unref = ru_slice_unref;
gpr_ref_init(&rc->refs, 1);
rc->resource_user = resource_user;
rc->size = size;
- gpr_slice slice;
+ grpc_slice slice;
slice.refcount = &rc->base;
slice.data.refcounted.bytes = (uint8_t *)(rc + 1);
slice.data.refcounted.length = size;
@@ -373,9 +448,19 @@ static void ru_post_destructive_reclaimer(grpc_exec_ctx *exec_ctx, void *ru,
rulist_add_tail(resource_user, GRPC_RULIST_RECLAIMER_DESTRUCTIVE);
}
+static void ru_shutdown(grpc_exec_ctx *exec_ctx, void *ru, grpc_error *error) {
+ grpc_resource_user *resource_user = ru;
+ grpc_exec_ctx_sched(exec_ctx, resource_user->reclaimers[0],
+ GRPC_ERROR_CANCELLED, NULL);
+ grpc_exec_ctx_sched(exec_ctx, resource_user->reclaimers[1],
+ GRPC_ERROR_CANCELLED, NULL);
+ resource_user->reclaimers[0] = NULL;
+ resource_user->reclaimers[1] = NULL;
+}
+
static void ru_destroy(grpc_exec_ctx *exec_ctx, void *ru, grpc_error *error) {
grpc_resource_user *resource_user = ru;
- GPR_ASSERT(resource_user->allocated == 0);
+ GPR_ASSERT(gpr_atm_no_barrier_load(&resource_user->refs) == 0);
for (int i = 0; i < GRPC_RULIST_COUNT; i++) {
rulist_remove(resource_user, (grpc_rulist)i);
}
@@ -383,13 +468,14 @@ static void ru_destroy(grpc_exec_ctx *exec_ctx, void *ru, grpc_error *error) {
GRPC_ERROR_CANCELLED, NULL);
grpc_exec_ctx_sched(exec_ctx, resource_user->reclaimers[1],
GRPC_ERROR_CANCELLED, NULL);
- grpc_exec_ctx_sched(exec_ctx, (grpc_closure *)gpr_atm_no_barrier_load(
- &resource_user->on_done_destroy_closure),
- GRPC_ERROR_NONE, NULL);
if (resource_user->free_pool != 0) {
resource_user->resource_quota->free_pool += resource_user->free_pool;
rq_step_sched(exec_ctx, resource_user->resource_quota);
}
+ grpc_resource_quota_internal_unref(exec_ctx, resource_user->resource_quota);
+ gpr_mu_destroy(&resource_user->mu);
+ gpr_free(resource_user->name);
+ gpr_free(resource_user);
}
static void ru_allocated_slices(grpc_exec_ctx *exec_ctx, void *arg,
@@ -397,7 +483,7 @@ static void ru_allocated_slices(grpc_exec_ctx *exec_ctx, void *arg,
grpc_resource_user_slice_allocator *slice_allocator = arg;
if (error == GRPC_ERROR_NONE) {
for (size_t i = 0; i < slice_allocator->count; i++) {
- gpr_slice_buffer_add_indexed(
+ grpc_slice_buffer_add_indexed(
slice_allocator->dest, ru_slice_create(slice_allocator->resource_user,
slice_allocator->length));
}
@@ -539,9 +625,9 @@ const grpc_arg_pointer_vtable *grpc_resource_quota_arg_vtable(void) {
* grpc_resource_user api
*/
-void grpc_resource_user_init(grpc_resource_user *resource_user,
- grpc_resource_quota *resource_quota,
- const char *name) {
+grpc_resource_user *grpc_resource_user_create(
+ grpc_resource_quota *resource_quota, const char *name) {
+ grpc_resource_user *resource_user = gpr_malloc(sizeof(*resource_user));
resource_user->resource_quota =
grpc_resource_quota_internal_ref(resource_quota);
grpc_closure_init(&resource_user->allocate_closure, &ru_allocate,
@@ -555,12 +641,12 @@ void grpc_resource_user_init(grpc_resource_user *resource_user,
grpc_closure_init(&resource_user->destroy_closure, &ru_destroy,
resource_user);
gpr_mu_init(&resource_user->mu);
- resource_user->allocated = 0;
+ gpr_atm_rel_store(&resource_user->refs, 1);
+ gpr_atm_rel_store(&resource_user->shutdown, 0);
resource_user->free_pool = 0;
grpc_closure_list_init(&resource_user->on_allocated);
resource_user->allocating = false;
resource_user->added_to_free_pool = false;
- gpr_atm_no_barrier_store(&resource_user->on_done_destroy_closure, 0);
resource_user->reclaimers[0] = NULL;
resource_user->reclaimers[1] = NULL;
for (int i = 0; i < GRPC_RULIST_COUNT; i++) {
@@ -572,56 +658,54 @@ void grpc_resource_user_init(grpc_resource_user *resource_user,
gpr_asprintf(&resource_user->name, "anonymous_resource_user_%" PRIxPTR,
(intptr_t)resource_user);
}
+ return resource_user;
}
-void grpc_resource_user_shutdown(grpc_exec_ctx *exec_ctx,
- grpc_resource_user *resource_user,
- grpc_closure *on_done) {
- gpr_mu_lock(&resource_user->mu);
- GPR_ASSERT(gpr_atm_no_barrier_load(&resource_user->on_done_destroy_closure) ==
- 0);
- gpr_atm_no_barrier_store(&resource_user->on_done_destroy_closure,
- (gpr_atm)on_done);
- if (resource_user->allocated == 0) {
+static void ru_ref_by(grpc_resource_user *resource_user, gpr_atm amount) {
+ GPR_ASSERT(amount > 0);
+ GPR_ASSERT(gpr_atm_no_barrier_fetch_add(&resource_user->refs, amount) != 0);
+}
+
+static void ru_unref_by(grpc_exec_ctx *exec_ctx,
+ grpc_resource_user *resource_user, gpr_atm amount) {
+ GPR_ASSERT(amount > 0);
+ gpr_atm old = gpr_atm_full_fetch_add(&resource_user->refs, -amount);
+ GPR_ASSERT(old >= amount);
+ if (old == amount) {
grpc_combiner_execute(exec_ctx, resource_user->resource_quota->combiner,
&resource_user->destroy_closure, GRPC_ERROR_NONE,
false);
}
- gpr_mu_unlock(&resource_user->mu);
}
-void grpc_resource_user_destroy(grpc_exec_ctx *exec_ctx,
- grpc_resource_user *resource_user) {
- grpc_resource_quota_internal_unref(exec_ctx, resource_user->resource_quota);
- gpr_mu_destroy(&resource_user->mu);
- gpr_free(resource_user->name);
+void grpc_resource_user_ref(grpc_resource_user *resource_user) {
+ ru_ref_by(resource_user, 1);
+}
+
+void grpc_resource_user_unref(grpc_exec_ctx *exec_ctx,
+ grpc_resource_user *resource_user) {
+ ru_unref_by(exec_ctx, resource_user, 1);
+}
+
+void grpc_resource_user_shutdown(grpc_exec_ctx *exec_ctx,
+ grpc_resource_user *resource_user) {
+ if (gpr_atm_full_fetch_add(&resource_user->shutdown, 1) == 0) {
+ grpc_combiner_execute(exec_ctx, resource_user->resource_quota->combiner,
+ grpc_closure_create(ru_shutdown, resource_user),
+ GRPC_ERROR_NONE, false);
+ }
}
void grpc_resource_user_alloc(grpc_exec_ctx *exec_ctx,
grpc_resource_user *resource_user, size_t size,
grpc_closure *optional_on_done) {
gpr_mu_lock(&resource_user->mu);
- grpc_closure *on_done_destroy = (grpc_closure *)gpr_atm_no_barrier_load(
- &resource_user->on_done_destroy_closure);
- if (on_done_destroy != NULL) {
- /* already shutdown */
- if (grpc_resource_quota_trace) {
- gpr_log(GPR_DEBUG, "RQ %s %s: alloc %" PRIdPTR " after shutdown",
- resource_user->resource_quota->name, resource_user->name, size);
- }
- grpc_exec_ctx_sched(
- exec_ctx, optional_on_done,
- GRPC_ERROR_CREATE("Buffer pool user is already shutdown"), NULL);
- gpr_mu_unlock(&resource_user->mu);
- return;
- }
- resource_user->allocated += (int64_t)size;
+ ru_ref_by(resource_user, (gpr_atm)size);
resource_user->free_pool -= (int64_t)size;
if (grpc_resource_quota_trace) {
- gpr_log(GPR_DEBUG, "RQ %s %s: alloc %" PRIdPTR "; allocated -> %" PRId64
- ", free_pool -> %" PRId64,
+ gpr_log(GPR_DEBUG, "RQ %s %s: alloc %" PRIdPTR "; free_pool -> %" PRId64,
resource_user->resource_quota->name, resource_user->name, size,
- resource_user->allocated, resource_user->free_pool);
+ resource_user->free_pool);
}
if (resource_user->free_pool < 0) {
grpc_closure_list_append(&resource_user->on_allocated, optional_on_done,
@@ -641,15 +725,12 @@ void grpc_resource_user_alloc(grpc_exec_ctx *exec_ctx,
void grpc_resource_user_free(grpc_exec_ctx *exec_ctx,
grpc_resource_user *resource_user, size_t size) {
gpr_mu_lock(&resource_user->mu);
- GPR_ASSERT(resource_user->allocated >= (int64_t)size);
bool was_zero_or_negative = resource_user->free_pool <= 0;
resource_user->free_pool += (int64_t)size;
- resource_user->allocated -= (int64_t)size;
if (grpc_resource_quota_trace) {
- gpr_log(GPR_DEBUG, "RQ %s %s: free %" PRIdPTR "; allocated -> %" PRId64
- ", free_pool -> %" PRId64,
+ gpr_log(GPR_DEBUG, "RQ %s %s: free %" PRIdPTR "; free_pool -> %" PRId64,
resource_user->resource_quota->name, resource_user->name, size,
- resource_user->allocated, resource_user->free_pool);
+ resource_user->free_pool);
}
bool is_bigger_than_zero = resource_user->free_pool > 0;
if (is_bigger_than_zero && was_zero_or_negative &&
@@ -659,29 +740,23 @@ void grpc_resource_user_free(grpc_exec_ctx *exec_ctx,
&resource_user->add_to_free_pool_closure,
GRPC_ERROR_NONE, false);
}
- grpc_closure *on_done_destroy = (grpc_closure *)gpr_atm_no_barrier_load(
- &resource_user->on_done_destroy_closure);
- if (on_done_destroy != NULL && resource_user->allocated == 0) {
- grpc_combiner_execute(exec_ctx, resource_user->resource_quota->combiner,
- &resource_user->destroy_closure, GRPC_ERROR_NONE,
- false);
- }
gpr_mu_unlock(&resource_user->mu);
+ ru_unref_by(exec_ctx, resource_user, (gpr_atm)size);
}
void grpc_resource_user_post_reclaimer(grpc_exec_ctx *exec_ctx,
grpc_resource_user *resource_user,
bool destructive,
grpc_closure *closure) {
- if (gpr_atm_acq_load(&resource_user->on_done_destroy_closure) == 0) {
- GPR_ASSERT(resource_user->reclaimers[destructive] == NULL);
- resource_user->reclaimers[destructive] = closure;
- grpc_combiner_execute(exec_ctx, resource_user->resource_quota->combiner,
- &resource_user->post_reclaimer_closure[destructive],
- GRPC_ERROR_NONE, false);
- } else {
+ GPR_ASSERT(resource_user->reclaimers[destructive] == NULL);
+ if (gpr_atm_acq_load(&resource_user->shutdown) > 0) {
grpc_exec_ctx_sched(exec_ctx, closure, GRPC_ERROR_CANCELLED, NULL);
+ return;
}
+ resource_user->reclaimers[destructive] = closure;
+ grpc_combiner_execute(exec_ctx, resource_user->resource_quota->combiner,
+ &resource_user->post_reclaimer_closure[destructive],
+ GRPC_ERROR_NONE, false);
}
void grpc_resource_user_finish_reclamation(grpc_exec_ctx *exec_ctx,
@@ -708,7 +783,7 @@ void grpc_resource_user_slice_allocator_init(
void grpc_resource_user_alloc_slices(
grpc_exec_ctx *exec_ctx,
grpc_resource_user_slice_allocator *slice_allocator, size_t length,
- size_t count, gpr_slice_buffer *dest) {
+ size_t count, grpc_slice_buffer *dest) {
slice_allocator->length = length;
slice_allocator->count = count;
slice_allocator->dest = dest;
@@ -716,9 +791,9 @@ void grpc_resource_user_alloc_slices(
count * length, &slice_allocator->on_allocated);
}
-gpr_slice grpc_resource_user_slice_malloc(grpc_exec_ctx *exec_ctx,
- grpc_resource_user *resource_user,
- size_t size) {
+grpc_slice grpc_resource_user_slice_malloc(grpc_exec_ctx *exec_ctx,
+ grpc_resource_user *resource_user,
+ size_t size) {
grpc_resource_user_alloc(exec_ctx, resource_user, size, NULL);
return ru_slice_create(resource_user, size);
}
diff --git a/src/core/lib/iomgr/resource_quota.h b/src/core/lib/iomgr/resource_quota.h
index da68f21a2c..0181fd978b 100644
--- a/src/core/lib/iomgr/resource_quota.h
+++ b/src/core/lib/iomgr/resource_quota.h
@@ -84,91 +84,15 @@ void grpc_resource_quota_internal_unref(grpc_exec_ctx *exec_ctx,
grpc_resource_quota *grpc_resource_quota_from_channel_args(
const grpc_channel_args *channel_args);
-/* Resource users are kept in (potentially) several intrusive linked lists
- at once. These are the list names. */
-typedef enum {
- /* Resource users that are waiting for an allocation */
- GRPC_RULIST_AWAITING_ALLOCATION,
- /* Resource users that have free memory available for internal reclamation */
- GRPC_RULIST_NON_EMPTY_FREE_POOL,
- /* Resource users that have published a benign reclamation is available */
- GRPC_RULIST_RECLAIMER_BENIGN,
- /* Resource users that have published a destructive reclamation is
- available */
- GRPC_RULIST_RECLAIMER_DESTRUCTIVE,
- /* Number of lists: must be last */
- GRPC_RULIST_COUNT
-} grpc_rulist;
-
typedef struct grpc_resource_user grpc_resource_user;
-/* Internal linked list pointers for a resource user */
-typedef struct {
- grpc_resource_user *next;
- grpc_resource_user *prev;
-} grpc_resource_user_link;
-
-struct grpc_resource_user {
- /* The quota this resource user consumes from */
- grpc_resource_quota *resource_quota;
-
- /* Closure to schedule an allocation under the resource quota combiner lock */
- grpc_closure allocate_closure;
- /* Closure to publish a non empty free pool under the resource quota combiner
- lock */
- grpc_closure add_to_free_pool_closure;
-
- gpr_mu mu;
- /* Total allocated memory outstanding by this resource user in bytes;
- always positive */
- int64_t allocated;
- /* The amount of memory (in bytes) this user has cached for its own use: to
- avoid quota contention, each resource user can keep some memory in
- addition to what it is immediately using (e.g., for caching), and the quota
- can pull it back under memory pressure.
- This value can become negative if more memory has been requested than
- existed in the free pool, at which point the quota is consulted to bring
- this value non-negative (asynchronously). */
- int64_t free_pool;
- /* A list of closures to call once free_pool becomes non-negative - ie when
- all outstanding allocations have been granted. */
- grpc_closure_list on_allocated;
- /* True if we are currently trying to allocate from the quota, false if not */
- bool allocating;
- /* True if we are currently trying to add ourselves to the non-free quota
- list, false otherwise */
- bool added_to_free_pool;
-
- /* Reclaimers: index 0 is the benign reclaimer, 1 is the destructive reclaimer
- */
- grpc_closure *reclaimers[2];
- /* Trampoline closures to finish reclamation and re-enter the quota combiner
- lock */
- grpc_closure post_reclaimer_closure[2];
-
- /* Closure to execute under the quota combiner to de-register and shutdown the
- resource user */
- grpc_closure destroy_closure;
- /* User supplied closure to call once the user has finished shutting down AND
- all outstanding allocations have been freed. Real type is grpc_closure*,
- but it's stored as an atomic to avoid a mutex on some fast paths. */
- gpr_atm on_done_destroy_closure;
-
- /* Links in the various grpc_rulist lists */
- grpc_resource_user_link links[GRPC_RULIST_COUNT];
-
- /* The name of this resource user, for debugging/tracing */
- char *name;
-};
-
-void grpc_resource_user_init(grpc_resource_user *resource_user,
- grpc_resource_quota *resource_quota,
- const char *name);
+grpc_resource_user *grpc_resource_user_create(
+ grpc_resource_quota *resource_quota, const char *name);
+void grpc_resource_user_ref(grpc_resource_user *resource_user);
+void grpc_resource_user_unref(grpc_exec_ctx *exec_ctx,
+ grpc_resource_user *resource_user);
void grpc_resource_user_shutdown(grpc_exec_ctx *exec_ctx,
- grpc_resource_user *resource_user,
- grpc_closure *on_done);
-void grpc_resource_user_destroy(grpc_exec_ctx *exec_ctx,
- grpc_resource_user *resource_user);
+ grpc_resource_user *resource_user);
/* Allocate from the resource user (and its quota).
If optional_on_done is NULL, then allocate immediately. This may push the
@@ -203,7 +127,7 @@ typedef struct grpc_resource_user_slice_allocator {
/* Number of slices to allocate on the current request */
size_t count;
/* Destination for slices to allocate on the current request */
- gpr_slice_buffer *dest;
+ grpc_slice_buffer *dest;
/* Parent resource user */
grpc_resource_user *resource_user;
} grpc_resource_user_slice_allocator;
@@ -219,11 +143,11 @@ void grpc_resource_user_slice_allocator_init(
void grpc_resource_user_alloc_slices(
grpc_exec_ctx *exec_ctx,
grpc_resource_user_slice_allocator *slice_allocator, size_t length,
- size_t count, gpr_slice_buffer *dest);
+ size_t count, grpc_slice_buffer *dest);
/* Allocate one slice of length \a size synchronously. */
-gpr_slice grpc_resource_user_slice_malloc(grpc_exec_ctx *exec_ctx,
- grpc_resource_user *resource_user,
- size_t size);
+grpc_slice grpc_resource_user_slice_malloc(grpc_exec_ctx *exec_ctx,
+ grpc_resource_user *resource_user,
+ size_t size);
#endif /* GRPC_CORE_LIB_IOMGR_RESOURCE_QUOTA_H */
diff --git a/src/core/lib/iomgr/socket_mutator.c b/src/core/lib/iomgr/socket_mutator.c
new file mode 100644
index 0000000000..8b1efb6bab
--- /dev/null
+++ b/src/core/lib/iomgr/socket_mutator.c
@@ -0,0 +1,98 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "src/core/lib/iomgr/socket_mutator.h"
+
+#include <grpc/impl/codegen/grpc_types.h>
+#include <grpc/support/sync.h>
+#include <grpc/support/useful.h>
+
+void grpc_socket_mutator_init(grpc_socket_mutator *mutator,
+ const grpc_socket_mutator_vtable *vtable) {
+ mutator->vtable = vtable;
+ gpr_ref_init(&mutator->refcount, 1);
+}
+
+grpc_socket_mutator *grpc_socket_mutator_ref(grpc_socket_mutator *mutator) {
+ gpr_ref(&mutator->refcount);
+ return mutator;
+}
+
+bool grpc_socket_mutator_mutate_fd(grpc_socket_mutator *mutator, int fd) {
+ return mutator->vtable->mutate_fd(fd, mutator);
+}
+
+int grpc_socket_mutator_compare(grpc_socket_mutator *a,
+ grpc_socket_mutator *b) {
+ int c = GPR_ICMP(a, b);
+ if (c != 0) {
+ grpc_socket_mutator *sma = a;
+ grpc_socket_mutator *smb = b;
+ c = GPR_ICMP(sma->vtable, smb->vtable);
+ if (c == 0) {
+ c = sma->vtable->compare(sma, smb);
+ }
+ }
+ return c;
+}
+
+void grpc_socket_mutator_unref(grpc_socket_mutator *mutator) {
+ if (gpr_unref(&mutator->refcount)) {
+ mutator->vtable->destory(mutator);
+ }
+}
+
+static void *socket_mutator_arg_copy(void *p) {
+ return grpc_socket_mutator_ref(p);
+}
+
+static void socket_mutator_arg_destroy(void *p) {
+ grpc_socket_mutator_unref(p);
+}
+
+static int socket_mutator_cmp(void *a, void *b) {
+ return grpc_socket_mutator_compare((grpc_socket_mutator *)a,
+ (grpc_socket_mutator *)b);
+}
+
+static const grpc_arg_pointer_vtable socket_mutator_arg_vtable = {
+ socket_mutator_arg_copy, socket_mutator_arg_destroy, socket_mutator_cmp};
+
+grpc_arg grpc_socket_mutator_to_arg(grpc_socket_mutator *mutator) {
+ grpc_arg arg;
+ arg.type = GRPC_ARG_POINTER;
+ arg.key = GRPC_ARG_SOCKET_MUTATOR;
+ arg.value.pointer.vtable = &socket_mutator_arg_vtable;
+ arg.value.pointer.p = mutator;
+ return arg;
+}
diff --git a/src/core/lib/iomgr/socket_mutator.h b/src/core/lib/iomgr/socket_mutator.h
new file mode 100644
index 0000000000..2f5b6c248e
--- /dev/null
+++ b/src/core/lib/iomgr/socket_mutator.h
@@ -0,0 +1,80 @@
+/*
+ *
+ * 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_SOCKET_MUTATOR_H
+#define GRPC_CORE_LIB_IOMGR_SOCKET_MUTATOR_H
+
+#include <grpc/impl/codegen/grpc_types.h>
+#include <grpc/support/sync.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/** The virtual table of grpc_socket_mutator */
+typedef struct {
+ /** Mutates the socket opitons of \a fd */
+ bool (*mutate_fd)(int fd, grpc_socket_mutator *mutator);
+ /** Compare socket mutator \a a and \a b */
+ int (*compare)(grpc_socket_mutator *a, grpc_socket_mutator *b);
+ /** Destroys the socket mutator instance */
+ void (*destory)(grpc_socket_mutator *mutator);
+} grpc_socket_mutator_vtable;
+
+/** The Socket Mutator interface allows changes on socket options */
+struct grpc_socket_mutator {
+ const grpc_socket_mutator_vtable *vtable;
+ gpr_refcount refcount;
+};
+
+/** called by concrete implementations to initialize the base struct */
+void grpc_socket_mutator_init(grpc_socket_mutator *mutator,
+ const grpc_socket_mutator_vtable *vtable);
+
+/** Wrap \a mutator as a grpc_arg */
+grpc_arg grpc_socket_mutator_to_arg(grpc_socket_mutator *mutator);
+
+/** Perform the file descriptor mutation operation of \a mutator on \a fd */
+bool grpc_socket_mutator_mutate_fd(grpc_socket_mutator *mutator, int fd);
+
+/** Compare if \a a and \a b are the same mutator or have same settings */
+int grpc_socket_mutator_compare(grpc_socket_mutator *a, grpc_socket_mutator *b);
+
+grpc_socket_mutator *grpc_socket_mutator_ref(grpc_socket_mutator *mutator);
+void grpc_socket_mutator_unref(grpc_socket_mutator *mutator);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* GRPC_CORE_LIB_IOMGR_SOCKET_MUTATOR_H */
diff --git a/src/core/lib/iomgr/socket_utils_common_posix.c b/src/core/lib/iomgr/socket_utils_common_posix.c
index bc28bbe316..88e9ade253 100644
--- a/src/core/lib/iomgr/socket_utils_common_posix.c
+++ b/src/core/lib/iomgr/socket_utils_common_posix.c
@@ -209,6 +209,15 @@ grpc_error *grpc_set_socket_low_latency(int fd, int low_latency) {
return GRPC_ERROR_NONE;
}
+/* set a socket using a grpc_socket_mutator */
+grpc_error *grpc_set_socket_with_mutator(int fd, grpc_socket_mutator *mutator) {
+ GPR_ASSERT(mutator);
+ if (!grpc_socket_mutator_mutate_fd(mutator, fd)) {
+ return GRPC_ERROR_CREATE("grpc_socket_mutator failed.");
+ }
+ return GRPC_ERROR_NONE;
+}
+
static gpr_once g_probe_ipv6_once = GPR_ONCE_INIT;
static int g_ipv6_loopback_available;
diff --git a/src/core/lib/iomgr/socket_utils_posix.h b/src/core/lib/iomgr/socket_utils_posix.h
index 175fb2b717..e84d3781a1 100644
--- a/src/core/lib/iomgr/socket_utils_posix.h
+++ b/src/core/lib/iomgr/socket_utils_posix.h
@@ -39,7 +39,9 @@
#include <sys/socket.h>
#include <unistd.h>
+#include <grpc/impl/codegen/grpc_types.h>
#include "src/core/lib/iomgr/error.h"
+#include "src/core/lib/iomgr/socket_mutator.h"
/* a wrapper for accept or accept4 */
int grpc_accept4(int sockfd, grpc_resolved_address *resolved_addr, int nonblock,
@@ -88,6 +90,9 @@ grpc_error *grpc_set_socket_sndbuf(int fd, int buffer_size_bytes);
/* Tries to set the socket's receive buffer to given size. */
grpc_error *grpc_set_socket_rcvbuf(int fd, int buffer_size_bytes);
+/* Tries to set the socket using a grpc_socket_mutator */
+grpc_error *grpc_set_socket_with_mutator(int fd, grpc_socket_mutator *mutator);
+
/* An enum to keep track of IPv4/IPv6 socket modes.
Currently, this information is only used when a socket is first created, but
diff --git a/src/core/lib/iomgr/tcp_client.h b/src/core/lib/iomgr/tcp_client.h
index 18e6e60ebc..0485661316 100644
--- a/src/core/lib/iomgr/tcp_client.h
+++ b/src/core/lib/iomgr/tcp_client.h
@@ -34,6 +34,7 @@
#ifndef GRPC_CORE_LIB_IOMGR_TCP_CLIENT_H
#define GRPC_CORE_LIB_IOMGR_TCP_CLIENT_H
+#include <grpc/impl/codegen/grpc_types.h>
#include <grpc/support/time.h>
#include "src/core/lib/iomgr/endpoint.h"
#include "src/core/lib/iomgr/pollset_set.h"
diff --git a/src/core/lib/iomgr/tcp_client_posix.c b/src/core/lib/iomgr/tcp_client_posix.c
index bc08c94ee0..13347735df 100644
--- a/src/core/lib/iomgr/tcp_client_posix.c
+++ b/src/core/lib/iomgr/tcp_client_posix.c
@@ -51,6 +51,7 @@
#include "src/core/lib/iomgr/ev_posix.h"
#include "src/core/lib/iomgr/iomgr_posix.h"
#include "src/core/lib/iomgr/sockaddr_utils.h"
+#include "src/core/lib/iomgr/socket_mutator.h"
#include "src/core/lib/iomgr/socket_utils_posix.h"
#include "src/core/lib/iomgr/tcp_posix.h"
#include "src/core/lib/iomgr/timer.h"
@@ -73,7 +74,8 @@ typedef struct {
grpc_channel_args *channel_args;
} async_connect;
-static grpc_error *prepare_socket(const grpc_resolved_address *addr, int fd) {
+static grpc_error *prepare_socket(const grpc_resolved_address *addr, int fd,
+ const grpc_channel_args *channel_args) {
grpc_error *err = GRPC_ERROR_NONE;
GPR_ASSERT(fd >= 0);
@@ -88,6 +90,16 @@ static grpc_error *prepare_socket(const grpc_resolved_address *addr, int fd) {
}
err = grpc_set_socket_no_sigpipe_if_possible(fd);
if (err != GRPC_ERROR_NONE) goto error;
+ if (channel_args) {
+ for (size_t i = 0; i < channel_args->num_args; i++) {
+ if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_SOCKET_MUTATOR)) {
+ GPR_ASSERT(channel_args->args[i].type == GRPC_ARG_POINTER);
+ grpc_socket_mutator *mutator = channel_args->args[i].value.pointer.p;
+ err = grpc_set_socket_with_mutator(fd, mutator);
+ if (err != GRPC_ERROR_NONE) goto error;
+ }
+ }
+ }
goto done;
error:
@@ -287,7 +299,7 @@ static void tcp_client_connect_impl(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(grpc_sockaddr_is_v4mapped(addr, &addr4_copy));
addr = &addr4_copy;
}
- if ((error = prepare_socket(addr, fd)) != GRPC_ERROR_NONE) {
+ if ((error = prepare_socket(addr, fd, channel_args)) != GRPC_ERROR_NONE) {
grpc_exec_ctx_sched(exec_ctx, closure, error, NULL);
return;
}
diff --git a/src/core/lib/iomgr/tcp_client_windows.c b/src/core/lib/iomgr/tcp_client_windows.c
index 30f7c66f15..4d1e809872 100644
--- a/src/core/lib/iomgr/tcp_client_windows.c
+++ b/src/core/lib/iomgr/tcp_client_windows.c
@@ -37,10 +37,10 @@
#include "src/core/lib/iomgr/sockaddr_windows.h"
+#include <grpc/slice_buffer.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/log_windows.h>
-#include <grpc/support/slice_buffer.h>
#include <grpc/support/useful.h>
#include "src/core/lib/channel/channel_args.h"
diff --git a/src/core/lib/iomgr/tcp_posix.c b/src/core/lib/iomgr/tcp_posix.c
index 880af93ee1..12a4797e6f 100644
--- a/src/core/lib/iomgr/tcp_posix.c
+++ b/src/core/lib/iomgr/tcp_posix.c
@@ -46,9 +46,9 @@
#include <sys/types.h>
#include <unistd.h>
+#include <grpc/slice.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
-#include <grpc/support/slice.h>
#include <grpc/support/string_util.h>
#include <grpc/support/sync.h>
#include <grpc/support/time.h>
@@ -56,6 +56,7 @@
#include "src/core/lib/debug/trace.h"
#include "src/core/lib/iomgr/ev_posix.h"
#include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/slice/slice_string_helpers.h"
#include "src/core/lib/support/string.h"
#ifdef GRPC_HAVE_MSG_NOSIGNAL
@@ -83,10 +84,10 @@ typedef struct {
gpr_atm shutdown_count;
/* garbage after the last read */
- gpr_slice_buffer last_read_buffer;
+ grpc_slice_buffer last_read_buffer;
- gpr_slice_buffer *incoming_buffer;
- gpr_slice_buffer *outgoing_buffer;
+ grpc_slice_buffer *incoming_buffer;
+ grpc_slice_buffer *outgoing_buffer;
/** slice within outgoing_buffer to write next */
size_t outgoing_slice_idx;
/** byte within outgoing_buffer->slices[outgoing_slice_idx] to write next */
@@ -102,7 +103,7 @@ typedef struct {
char *peer_string;
- grpc_resource_user resource_user;
+ grpc_resource_user *resource_user;
grpc_resource_user_slice_allocator slice_allocator;
} grpc_tcp;
@@ -110,28 +111,18 @@ static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
grpc_error *error);
static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
grpc_error *error);
-static void tcp_unref_closure(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- grpc_error *error);
-
-static void tcp_maybe_shutdown_resource_user(grpc_exec_ctx *exec_ctx,
- grpc_tcp *tcp) {
- if (gpr_atm_full_fetch_add(&tcp->shutdown_count, 1) == 0) {
- grpc_resource_user_shutdown(exec_ctx, &tcp->resource_user,
- grpc_closure_create(tcp_unref_closure, tcp));
- }
-}
static void tcp_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *)ep;
- tcp_maybe_shutdown_resource_user(exec_ctx, tcp);
grpc_fd_shutdown(exec_ctx, tcp->em_fd);
+ grpc_resource_user_shutdown(exec_ctx, tcp->resource_user);
}
static void tcp_free(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
grpc_fd_orphan(exec_ctx, tcp->em_fd, tcp->release_fd_cb, tcp->release_fd,
"tcp_unref_orphan");
- gpr_slice_buffer_destroy(&tcp->last_read_buffer);
- grpc_resource_user_destroy(exec_ctx, &tcp->resource_user);
+ grpc_slice_buffer_destroy(&tcp->last_read_buffer);
+ grpc_resource_user_unref(exec_ctx, tcp->resource_user);
gpr_free(tcp->peer_string);
gpr_free(tcp);
}
@@ -168,16 +159,10 @@ static void tcp_unref(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); }
#endif
-static void tcp_unref_closure(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- TCP_UNREF(exec_ctx, arg, "resource_user");
-}
-
static void tcp_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
grpc_network_status_unregister_endpoint(ep);
grpc_tcp *tcp = (grpc_tcp *)ep;
- tcp_maybe_shutdown_resource_user(exec_ctx, tcp);
- gpr_slice_buffer_reset_and_unref(&tcp->last_read_buffer);
+ grpc_slice_buffer_reset_and_unref(&tcp->last_read_buffer);
TCP_UNREF(exec_ctx, tcp, "destroy");
}
@@ -191,8 +176,8 @@ static void call_read_cb(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp,
gpr_log(GPR_DEBUG, "read: error=%s", str);
grpc_error_free_string(str);
for (i = 0; i < tcp->incoming_buffer->count; i++) {
- char *dump = gpr_dump_slice(tcp->incoming_buffer->slices[i],
- GPR_DUMP_HEX | GPR_DUMP_ASCII);
+ char *dump = grpc_dump_slice(tcp->incoming_buffer->slices[i],
+ GPR_DUMP_HEX | GPR_DUMP_ASCII);
gpr_log(GPR_DEBUG, "READ %p (peer=%s): %s", tcp, tcp->peer_string, dump);
gpr_free(dump);
}
@@ -216,8 +201,8 @@ static void tcp_do_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
GPR_TIMER_BEGIN("tcp_continue_read", 0);
for (i = 0; i < tcp->incoming_buffer->count; i++) {
- iov[i].iov_base = GPR_SLICE_START_PTR(tcp->incoming_buffer->slices[i]);
- iov[i].iov_len = GPR_SLICE_LENGTH(tcp->incoming_buffer->slices[i]);
+ iov[i].iov_base = GRPC_SLICE_START_PTR(tcp->incoming_buffer->slices[i]);
+ iov[i].iov_len = GRPC_SLICE_LENGTH(tcp->incoming_buffer->slices[i]);
}
msg.msg_name = NULL;
@@ -244,19 +229,19 @@ static void tcp_do_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
/* We've consumed the edge, request a new one */
grpc_fd_notify_on_read(exec_ctx, tcp->em_fd, &tcp->read_closure);
} else {
- gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer);
+ grpc_slice_buffer_reset_and_unref(tcp->incoming_buffer);
call_read_cb(exec_ctx, tcp, GRPC_OS_ERROR(errno, "recvmsg"));
TCP_UNREF(exec_ctx, tcp, "read");
}
} else if (read_bytes == 0) {
/* 0 read size ==> end of stream */
- gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer);
+ grpc_slice_buffer_reset_and_unref(tcp->incoming_buffer);
call_read_cb(exec_ctx, tcp, GRPC_ERROR_CREATE("Socket closed"));
TCP_UNREF(exec_ctx, tcp, "read");
} else {
GPR_ASSERT((size_t)read_bytes <= tcp->incoming_buffer->length);
if ((size_t)read_bytes < tcp->incoming_buffer->length) {
- gpr_slice_buffer_trim_end(
+ grpc_slice_buffer_trim_end(
tcp->incoming_buffer,
tcp->incoming_buffer->length - (size_t)read_bytes,
&tcp->last_read_buffer);
@@ -275,8 +260,8 @@ static void tcp_read_allocation_done(grpc_exec_ctx *exec_ctx, void *tcpp,
grpc_error *error) {
grpc_tcp *tcp = tcpp;
if (error != GRPC_ERROR_NONE) {
- gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer);
- gpr_slice_buffer_reset_and_unref(&tcp->last_read_buffer);
+ grpc_slice_buffer_reset_and_unref(tcp->incoming_buffer);
+ grpc_slice_buffer_reset_and_unref(&tcp->last_read_buffer);
call_read_cb(exec_ctx, tcp, GRPC_ERROR_REF(error));
TCP_UNREF(exec_ctx, tcp, "read");
} else {
@@ -301,8 +286,8 @@ static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
GPR_ASSERT(!tcp->finished_edge);
if (error != GRPC_ERROR_NONE) {
- gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer);
- gpr_slice_buffer_reset_and_unref(&tcp->last_read_buffer);
+ grpc_slice_buffer_reset_and_unref(tcp->incoming_buffer);
+ grpc_slice_buffer_reset_and_unref(&tcp->last_read_buffer);
call_read_cb(exec_ctx, tcp, GRPC_ERROR_REF(error));
TCP_UNREF(exec_ctx, tcp, "read");
} else {
@@ -311,13 +296,13 @@ static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
}
static void tcp_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *incoming_buffer, grpc_closure *cb) {
+ grpc_slice_buffer *incoming_buffer, grpc_closure *cb) {
grpc_tcp *tcp = (grpc_tcp *)ep;
GPR_ASSERT(tcp->read_cb == NULL);
tcp->read_cb = cb;
tcp->incoming_buffer = incoming_buffer;
- gpr_slice_buffer_reset_and_unref(incoming_buffer);
- gpr_slice_buffer_swap(incoming_buffer, &tcp->last_read_buffer);
+ grpc_slice_buffer_reset_and_unref(incoming_buffer);
+ grpc_slice_buffer_swap(incoming_buffer, &tcp->last_read_buffer);
TCP_REF(tcp, "read");
if (tcp->finished_edge) {
tcp->finished_edge = false;
@@ -347,11 +332,11 @@ static bool tcp_flush(grpc_tcp *tcp, grpc_error **error) {
iov_size != MAX_WRITE_IOVEC;
iov_size++) {
iov[iov_size].iov_base =
- GPR_SLICE_START_PTR(
+ GRPC_SLICE_START_PTR(
tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]) +
tcp->outgoing_byte_idx;
iov[iov_size].iov_len =
- GPR_SLICE_LENGTH(
+ GRPC_SLICE_LENGTH(
tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]) -
tcp->outgoing_byte_idx;
sending_length += iov[iov_size].iov_len;
@@ -392,7 +377,7 @@ static bool tcp_flush(grpc_tcp *tcp, grpc_error **error) {
size_t slice_length;
tcp->outgoing_slice_idx--;
- slice_length = GPR_SLICE_LENGTH(
+ slice_length = GRPC_SLICE_LENGTH(
tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]);
if (slice_length > trailing) {
tcp->outgoing_byte_idx = slice_length - trailing;
@@ -442,7 +427,7 @@ static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
}
static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *buf, grpc_closure *cb) {
+ grpc_slice_buffer *buf, grpc_closure *cb) {
grpc_tcp *tcp = (grpc_tcp *)ep;
grpc_error *error = GRPC_ERROR_NONE;
@@ -451,7 +436,7 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
for (i = 0; i < buf->count; i++) {
char *data =
- gpr_dump_slice(buf->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII);
+ grpc_dump_slice(buf->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII);
gpr_log(GPR_DEBUG, "WRITE %p (peer=%s): %s", tcp, tcp->peer_string, data);
gpr_free(data);
}
@@ -508,6 +493,11 @@ static char *tcp_get_peer(grpc_endpoint *ep) {
return gpr_strdup(tcp->peer_string);
}
+static int tcp_get_fd(grpc_endpoint *ep) {
+ grpc_tcp *tcp = (grpc_tcp *)ep;
+ return tcp->fd;
+}
+
static grpc_workqueue *tcp_get_workqueue(grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *)ep;
return grpc_fd_get_workqueue(tcp->em_fd);
@@ -515,7 +505,7 @@ static grpc_workqueue *tcp_get_workqueue(grpc_endpoint *ep) {
static grpc_resource_user *tcp_get_resource_user(grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *)ep;
- return &tcp->resource_user;
+ return tcp->resource_user;
}
static const grpc_endpoint_vtable vtable = {tcp_read,
@@ -526,7 +516,8 @@ static const grpc_endpoint_vtable vtable = {tcp_read,
tcp_shutdown,
tcp_destroy,
tcp_get_resource_user,
- tcp_get_peer};
+ tcp_get_peer,
+ tcp_get_fd};
grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd,
grpc_resource_quota *resource_quota,
@@ -543,20 +534,18 @@ grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd,
tcp->slice_size = slice_size;
tcp->iov_size = 1;
tcp->finished_edge = true;
- /* paired with unref in grpc_tcp_destroy, and with the shutdown for our
- * resource_user */
- gpr_ref_init(&tcp->refcount, 2);
+ /* paired with unref in grpc_tcp_destroy */
+ gpr_ref_init(&tcp->refcount, 1);
gpr_atm_no_barrier_store(&tcp->shutdown_count, 0);
tcp->em_fd = em_fd;
tcp->read_closure.cb = tcp_handle_read;
tcp->read_closure.cb_arg = tcp;
tcp->write_closure.cb = tcp_handle_write;
tcp->write_closure.cb_arg = tcp;
- gpr_slice_buffer_init(&tcp->last_read_buffer);
- grpc_resource_user_init(&tcp->resource_user, resource_quota, peer_string);
- grpc_resource_user_slice_allocator_init(&tcp->slice_allocator,
- &tcp->resource_user,
- tcp_read_allocation_done, tcp);
+ grpc_slice_buffer_init(&tcp->last_read_buffer);
+ tcp->resource_user = grpc_resource_user_create(resource_quota, peer_string);
+ grpc_resource_user_slice_allocator_init(
+ &tcp->slice_allocator, tcp->resource_user, tcp_read_allocation_done, tcp);
/* Tell network status tracker about new endpoint */
grpc_network_status_register_endpoint(&tcp->base);
@@ -576,8 +565,7 @@ void grpc_tcp_destroy_and_release_fd(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
GPR_ASSERT(ep->vtable == &vtable);
tcp->release_fd = fd;
tcp->release_fd_cb = done;
- tcp_maybe_shutdown_resource_user(exec_ctx, tcp);
- gpr_slice_buffer_reset_and_unref(&tcp->last_read_buffer);
+ grpc_slice_buffer_reset_and_unref(&tcp->last_read_buffer);
TCP_UNREF(exec_ctx, tcp, "destroy");
}
diff --git a/src/core/lib/iomgr/tcp_uv.c b/src/core/lib/iomgr/tcp_uv.c
index 8e74c9e863..6e2ad1dbe9 100644
--- a/src/core/lib/iomgr/tcp_uv.c
+++ b/src/core/lib/iomgr/tcp_uv.c
@@ -38,14 +38,17 @@
#include <limits.h>
#include <string.h>
+#include <grpc/slice_buffer.h>
+
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
-#include <grpc/support/slice_buffer.h>
#include <grpc/support/string_util.h>
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/network_status_tracker.h"
+#include "src/core/lib/iomgr/resource_quota.h"
#include "src/core/lib/iomgr/tcp_uv.h"
+#include "src/core/lib/slice/slice_string_helpers.h"
#include "src/core/lib/support/string.h"
int grpc_tcp_trace = 0;
@@ -62,15 +65,14 @@ typedef struct {
grpc_closure *read_cb;
grpc_closure *write_cb;
- gpr_slice read_slice;
- gpr_slice_buffer *read_slices;
- gpr_slice_buffer *write_slices;
+ grpc_slice read_slice;
+ grpc_slice_buffer *read_slices;
+ grpc_slice_buffer *write_slices;
uv_buf_t *write_buffers;
- grpc_resource_user resource_user;
+ grpc_resource_user *resource_user;
bool shutting_down;
- bool resource_user_shutting_down;
char *peer_string;
grpc_pollset *pollset;
@@ -78,23 +80,23 @@ typedef struct {
static void uv_close_callback(uv_handle_t *handle) { gpr_free(handle); }
-static void tcp_free(grpc_tcp *tcp) {
- grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- grpc_resource_user_destroy(&exec_ctx, &tcp->resource_user);
+static void tcp_free(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
+ grpc_resource_user_unref(exec_ctx, tcp->resource_user);
gpr_free(tcp);
- grpc_exec_ctx_finish(&exec_ctx);
}
/*#define GRPC_TCP_REFCOUNT_DEBUG*/
#ifdef GRPC_TCP_REFCOUNT_DEBUG
-#define TCP_UNREF(tcp, reason) tcp_unref((tcp), (reason), __FILE__, __LINE__)
-#define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__)
-static void tcp_unref(grpc_tcp *tcp, const char *reason, const char *file,
- int line) {
+#define TCP_UNREF(exec_ctx, tcp, reason) \
+ tcp_unref((exec_ctx), (tcp), (reason), __FILE__, __LINE__)
+#define TCP_REF(tcp, reason) \
+ tcp_ref((exec_ctx), (tcp), (reason), __FILE__, __LINE__)
+static void tcp_unref(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp,
+ const char *reason, const char *file, int line) {
gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %d -> %d", tcp,
reason, tcp->refcount.count, tcp->refcount.count - 1);
if (gpr_unref(&tcp->refcount)) {
- tcp_free(tcp);
+ tcp_free(exec_ctx, tcp);
}
}
@@ -105,11 +107,11 @@ static void tcp_ref(grpc_tcp *tcp, const char *reason, const char *file,
gpr_ref(&tcp->refcount);
}
#else
-#define TCP_UNREF(tcp, reason) tcp_unref((tcp))
+#define TCP_UNREF(exec_ctx, tcp, reason) tcp_unref((exec_ctx), (tcp))
#define TCP_REF(tcp, reason) tcp_ref((tcp))
-static void tcp_unref(grpc_tcp *tcp) {
+static void tcp_unref(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
if (gpr_unref(&tcp->refcount)) {
- tcp_free(tcp);
+ tcp_free(exec_ctx, tcp);
}
}
@@ -122,15 +124,15 @@ static void alloc_uv_buf(uv_handle_t *handle, size_t suggested_size,
grpc_tcp *tcp = handle->data;
(void)suggested_size;
tcp->read_slice = grpc_resource_user_slice_malloc(
- &exec_ctx, &tcp->resource_user, GRPC_TCP_DEFAULT_READ_SLICE_SIZE);
- buf->base = (char *)GPR_SLICE_START_PTR(tcp->read_slice);
- buf->len = GPR_SLICE_LENGTH(tcp->read_slice);
+ &exec_ctx, tcp->resource_user, GRPC_TCP_DEFAULT_READ_SLICE_SIZE);
+ buf->base = (char *)GRPC_SLICE_START_PTR(tcp->read_slice);
+ buf->len = GRPC_SLICE_LENGTH(tcp->read_slice);
grpc_exec_ctx_finish(&exec_ctx);
}
static void read_callback(uv_stream_t *stream, ssize_t nread,
const uv_buf_t *buf) {
- gpr_slice sub;
+ grpc_slice sub;
grpc_error *error;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_tcp *tcp = stream->data;
@@ -139,7 +141,7 @@ static void read_callback(uv_stream_t *stream, ssize_t nread,
// Nothing happened. Wait for the next callback
return;
}
- TCP_UNREF(tcp, "read");
+ TCP_UNREF(&exec_ctx, tcp, "read");
tcp->read_cb = NULL;
// TODO(murgatroid99): figure out what the return value here means
uv_read_stop(stream);
@@ -147,8 +149,8 @@ static void read_callback(uv_stream_t *stream, ssize_t nread,
error = GRPC_ERROR_CREATE("EOF");
} else if (nread > 0) {
// Successful read
- sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, (size_t)nread);
- gpr_slice_buffer_add(tcp->read_slices, sub);
+ sub = grpc_slice_sub_no_ref(tcp->read_slice, 0, (size_t)nread);
+ grpc_slice_buffer_add(tcp->read_slices, sub);
error = GRPC_ERROR_NONE;
if (grpc_tcp_trace) {
size_t i;
@@ -156,8 +158,8 @@ static void read_callback(uv_stream_t *stream, ssize_t nread,
gpr_log(GPR_DEBUG, "read: error=%s", str);
grpc_error_free_string(str);
for (i = 0; i < tcp->read_slices->count; i++) {
- char *dump = gpr_dump_slice(tcp->read_slices->slices[i],
- GPR_DUMP_HEX | GPR_DUMP_ASCII);
+ char *dump = grpc_dump_slice(tcp->read_slices->slices[i],
+ GPR_DUMP_HEX | GPR_DUMP_ASCII);
gpr_log(GPR_DEBUG, "READ %p (peer=%s): %s", tcp, tcp->peer_string,
dump);
gpr_free(dump);
@@ -172,14 +174,14 @@ static void read_callback(uv_stream_t *stream, ssize_t nread,
}
static void uv_endpoint_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *read_slices, grpc_closure *cb) {
+ grpc_slice_buffer *read_slices, grpc_closure *cb) {
grpc_tcp *tcp = (grpc_tcp *)ep;
int status;
grpc_error *error = GRPC_ERROR_NONE;
GPR_ASSERT(tcp->read_cb == NULL);
tcp->read_cb = cb;
tcp->read_slices = read_slices;
- gpr_slice_buffer_reset_and_unref(read_slices);
+ grpc_slice_buffer_reset_and_unref(read_slices);
TCP_REF(tcp, "read");
// TODO(murgatroid99): figure out what the return value here means
status =
@@ -202,7 +204,7 @@ static void write_callback(uv_write_t *req, int status) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_closure *cb = tcp->write_cb;
tcp->write_cb = NULL;
- TCP_UNREF(tcp, "write");
+ TCP_UNREF(&exec_ctx, tcp, "write");
if (status == 0) {
error = GRPC_ERROR_NONE;
} else {
@@ -213,28 +215,28 @@ static void write_callback(uv_write_t *req, int status) {
gpr_log(GPR_DEBUG, "write complete on %p: error=%s", tcp, str);
}
gpr_free(tcp->write_buffers);
- grpc_resource_user_free(&exec_ctx, &tcp->resource_user,
+ grpc_resource_user_free(&exec_ctx, tcp->resource_user,
sizeof(uv_buf_t) * tcp->write_slices->count);
grpc_exec_ctx_sched(&exec_ctx, cb, error, NULL);
grpc_exec_ctx_finish(&exec_ctx);
}
static void uv_endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *write_slices,
+ grpc_slice_buffer *write_slices,
grpc_closure *cb) {
grpc_tcp *tcp = (grpc_tcp *)ep;
uv_buf_t *buffers;
unsigned int buffer_count;
unsigned int i;
- gpr_slice *slice;
+ grpc_slice *slice;
uv_write_t *write_req;
if (grpc_tcp_trace) {
size_t j;
for (j = 0; j < write_slices->count; j++) {
- char *data = gpr_dump_slice(write_slices->slices[j],
- GPR_DUMP_HEX | GPR_DUMP_ASCII);
+ char *data = grpc_dump_slice(write_slices->slices[j],
+ GPR_DUMP_HEX | GPR_DUMP_ASCII);
gpr_log(GPR_DEBUG, "WRITE %p (peer=%s): %s", tcp, tcp->peer_string, data);
gpr_free(data);
}
@@ -259,12 +261,12 @@ static void uv_endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
tcp->write_cb = cb;
buffer_count = (unsigned int)tcp->write_slices->count;
buffers = gpr_malloc(sizeof(uv_buf_t) * buffer_count);
- grpc_resource_user_alloc(exec_ctx, &tcp->resource_user,
+ grpc_resource_user_alloc(exec_ctx, tcp->resource_user,
sizeof(uv_buf_t) * buffer_count, NULL);
for (i = 0; i < buffer_count; i++) {
slice = &tcp->write_slices->slices[i];
- buffers[i].base = (char *)GPR_SLICE_START_PTR(*slice);
- buffers[i].len = GPR_SLICE_LENGTH(*slice);
+ buffers[i].base = (char *)GRPC_SLICE_START_PTR(*slice);
+ buffers[i].len = GRPC_SLICE_LENGTH(*slice);
}
tcp->write_buffers = buffers;
write_req = &tcp->write_req;
@@ -295,22 +297,6 @@ static void uv_add_to_pollset_set(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
static void shutdown_callback(uv_shutdown_t *req, int status) {}
-static void resource_user_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- TCP_UNREF(arg, "resource_user");
-}
-
-static void uv_resource_user_maybe_shutdown(grpc_exec_ctx *exec_ctx,
- grpc_tcp *tcp) {
- if (!tcp->resource_user_shutting_down) {
- tcp->resource_user_shutting_down = true;
- TCP_REF(tcp, "resource_user");
- grpc_resource_user_shutdown(
- exec_ctx, &tcp->resource_user,
- grpc_closure_create(resource_user_shutdown_done, tcp));
- }
-}
-
static void uv_endpoint_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *)ep;
if (!tcp->shutting_down) {
@@ -324,8 +310,7 @@ static void uv_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
grpc_network_status_unregister_endpoint(ep);
grpc_tcp *tcp = (grpc_tcp *)ep;
uv_close((uv_handle_t *)tcp->handle, uv_close_callback);
- uv_resource_user_maybe_shutdown(exec_ctx, tcp);
- TCP_UNREF(tcp, "destroy");
+ TCP_UNREF(exec_ctx, tcp, "destroy");
}
static char *uv_get_peer(grpc_endpoint *ep) {
@@ -335,15 +320,18 @@ static char *uv_get_peer(grpc_endpoint *ep) {
static grpc_resource_user *uv_get_resource_user(grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *)ep;
- return &tcp->resource_user;
+ return tcp->resource_user;
}
static grpc_workqueue *uv_get_workqueue(grpc_endpoint *ep) { return NULL; }
+static int uv_get_fd(grpc_endpoint *ep) { return -1; }
+
static grpc_endpoint_vtable vtable = {
uv_endpoint_read, uv_endpoint_write, uv_get_workqueue,
uv_add_to_pollset, uv_add_to_pollset_set, uv_endpoint_shutdown,
- uv_destroy, uv_get_resource_user, uv_get_peer};
+ uv_destroy, uv_get_resource_user, uv_get_peer,
+ uv_get_fd};
grpc_endpoint *grpc_tcp_create(uv_tcp_t *handle,
grpc_resource_quota *resource_quota,
@@ -364,8 +352,7 @@ grpc_endpoint *grpc_tcp_create(uv_tcp_t *handle,
gpr_ref_init(&tcp->refcount, 1);
tcp->peer_string = gpr_strdup(peer_string);
tcp->shutting_down = false;
- tcp->resource_user_shutting_down = false;
- grpc_resource_user_init(&tcp->resource_user, resource_quota, peer_string);
+ tcp->resource_user = grpc_resource_user_create(resource_quota, peer_string);
/* Tell network status tracking code about the new endpoint */
grpc_network_status_register_endpoint(&tcp->base);
diff --git a/src/core/lib/iomgr/tcp_windows.c b/src/core/lib/iomgr/tcp_windows.c
index 46f0491d10..62afbcef51 100644
--- a/src/core/lib/iomgr/tcp_windows.c
+++ b/src/core/lib/iomgr/tcp_windows.c
@@ -40,10 +40,10 @@
#include "src/core/lib/iomgr/network_status_tracker.h"
#include "src/core/lib/iomgr/sockaddr_windows.h"
+#include <grpc/slice_buffer.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/log_windows.h>
-#include <grpc/support/slice_buffer.h>
#include <grpc/support/string_util.h>
#include <grpc/support/useful.h>
@@ -105,50 +105,39 @@ typedef struct grpc_tcp {
grpc_closure *read_cb;
grpc_closure *write_cb;
- gpr_slice read_slice;
- gpr_slice_buffer *write_slices;
- gpr_slice_buffer *read_slices;
+ grpc_slice read_slice;
+ grpc_slice_buffer *write_slices;
+ grpc_slice_buffer *read_slices;
- grpc_resource_user resource_user;
+ grpc_resource_user *resource_user;
/* The IO Completion Port runs from another thread. We need some mechanism
to protect ourselves when requesting a shutdown. */
gpr_mu mu;
int shutting_down;
- gpr_atm resource_user_shutdown_count;
-
char *peer_string;
} grpc_tcp;
-static void win_unref_closure(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- grpc_error *error);
-
-static void win_maybe_shutdown_resource_user(grpc_exec_ctx *exec_ctx,
- grpc_tcp *tcp) {
- if (gpr_atm_full_fetch_add(&tcp->resource_user_shutdown_count, 1) == 0) {
- grpc_resource_user_shutdown(exec_ctx, &tcp->resource_user,
- grpc_closure_create(win_unref_closure, tcp));
- }
-}
-
-static void tcp_free(grpc_tcp *tcp) {
+static void tcp_free(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
grpc_winsocket_destroy(tcp->socket);
gpr_mu_destroy(&tcp->mu);
gpr_free(tcp->peer_string);
+ grpc_resource_user_unref(exec_ctx, tcp->resource_user);
gpr_free(tcp);
}
/*#define GRPC_TCP_REFCOUNT_DEBUG*/
#ifdef GRPC_TCP_REFCOUNT_DEBUG
-#define TCP_UNREF(tcp, reason) tcp_unref((tcp), (reason), __FILE__, __LINE__)
+#define TCP_UNREF(exec_ctx, tcp, reason) \
+ tcp_unref((exec_ctx), (tcp), (reason), __FILE__, __LINE__)
#define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__)
-static void tcp_unref(grpc_tcp *tcp, const char *reason, const char *file,
- int line) {
+static void tcp_unref(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp,
+ const char *reason, const char *file, int line) {
gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %d -> %d", tcp,
reason, tcp->refcount.count, tcp->refcount.count - 1);
if (gpr_unref(&tcp->refcount)) {
- tcp_free(tcp);
+ tcp_free(exec_ctx, tcp);
}
}
@@ -159,28 +148,23 @@ static void tcp_ref(grpc_tcp *tcp, const char *reason, const char *file,
gpr_ref(&tcp->refcount);
}
#else
-#define TCP_UNREF(tcp, reason) tcp_unref((tcp))
+#define TCP_UNREF(exec_ctx, tcp, reason) tcp_unref((exec_ctx), (tcp))
#define TCP_REF(tcp, reason) tcp_ref((tcp))
-static void tcp_unref(grpc_tcp *tcp) {
+static void tcp_unref(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
if (gpr_unref(&tcp->refcount)) {
- tcp_free(tcp);
+ tcp_free(exec_ctx, tcp);
}
}
static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); }
#endif
-static void win_unref_closure(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- TCP_UNREF(arg, "resource_user");
-}
-
/* Asynchronous callback from the IOCP, or the background thread. */
static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) {
grpc_tcp *tcp = tcpp;
grpc_closure *cb = tcp->read_cb;
grpc_winsocket *socket = tcp->socket;
- gpr_slice sub;
+ grpc_slice sub;
grpc_winsocket_callback_info *info = &socket->read_info;
GRPC_ERROR_REF(error);
@@ -190,25 +174,25 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) {
char *utf8_message = gpr_format_message(info->wsa_error);
error = GRPC_ERROR_CREATE(utf8_message);
gpr_free(utf8_message);
- gpr_slice_unref(tcp->read_slice);
+ grpc_slice_unref(tcp->read_slice);
} else {
if (info->bytes_transfered != 0 && !tcp->shutting_down) {
- sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered);
- gpr_slice_buffer_add(tcp->read_slices, sub);
+ sub = grpc_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered);
+ grpc_slice_buffer_add(tcp->read_slices, sub);
} else {
- gpr_slice_unref(tcp->read_slice);
+ grpc_slice_unref(tcp->read_slice);
error = GRPC_ERROR_CREATE("End of TCP stream");
}
}
}
tcp->read_cb = NULL;
- TCP_UNREF(tcp, "read");
+ TCP_UNREF(exec_ctx, tcp, "read");
grpc_exec_ctx_sched(exec_ctx, cb, error, NULL);
}
static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *read_slices, grpc_closure *cb) {
+ grpc_slice_buffer *read_slices, grpc_closure *cb) {
grpc_tcp *tcp = (grpc_tcp *)ep;
grpc_winsocket *handle = tcp->socket;
grpc_winsocket_callback_info *info = &handle->read_info;
@@ -225,13 +209,13 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
tcp->read_cb = cb;
tcp->read_slices = read_slices;
- gpr_slice_buffer_reset_and_unref(read_slices);
+ grpc_slice_buffer_reset_and_unref(read_slices);
- tcp->read_slice = gpr_slice_malloc(8192);
+ tcp->read_slice = grpc_slice_malloc(8192);
- buffer.len = (ULONG)GPR_SLICE_LENGTH(
+ buffer.len = (ULONG)GRPC_SLICE_LENGTH(
tcp->read_slice); // we know slice size fits in 32bit.
- buffer.buf = (char *)GPR_SLICE_START_PTR(tcp->read_slice);
+ buffer.buf = (char *)GRPC_SLICE_START_PTR(tcp->read_slice);
TCP_REF(tcp, "read");
@@ -287,13 +271,13 @@ static void on_write(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) {
}
}
- TCP_UNREF(tcp, "write");
+ TCP_UNREF(exec_ctx, tcp, "write");
grpc_exec_ctx_sched(exec_ctx, cb, error, NULL);
}
/* Initiates a write. */
static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
- gpr_slice_buffer *slices, grpc_closure *cb) {
+ grpc_slice_buffer *slices, grpc_closure *cb) {
grpc_tcp *tcp = (grpc_tcp *)ep;
grpc_winsocket *socket = tcp->socket;
grpc_winsocket_callback_info *info = &socket->write_info;
@@ -320,10 +304,10 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
}
for (i = 0; i < tcp->write_slices->count; i++) {
- len = GPR_SLICE_LENGTH(tcp->write_slices->slices[i]);
+ len = GRPC_SLICE_LENGTH(tcp->write_slices->slices[i]);
GPR_ASSERT(len <= ULONG_MAX);
buffers[i].len = (ULONG)len;
- buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices->slices[i]);
+ buffers[i].buf = (char *)GRPC_SLICE_START_PTR(tcp->write_slices->slices[i]);
}
/* First, let's try a synchronous, non-blocking write. */
@@ -355,7 +339,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
if (status != 0) {
int wsa_error = WSAGetLastError();
if (wsa_error != WSA_IO_PENDING) {
- TCP_UNREF(tcp, "write");
+ TCP_UNREF(exec_ctx, tcp, "write");
grpc_exec_ctx_sched(exec_ctx, cb, GRPC_WSA_ERROR(wsa_error, "WSASend"),
NULL);
return;
@@ -396,15 +380,14 @@ static void win_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
callback. See the comments in on_read and on_write. */
tcp->shutting_down = 1;
grpc_winsocket_shutdown(tcp->socket);
- win_maybe_shutdown_resource_user(exec_ctx, tcp);
gpr_mu_unlock(&tcp->mu);
+ grpc_resource_user_shutdown(exec_ctx, tcp->resource_user);
}
static void win_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
grpc_network_status_unregister_endpoint(ep);
grpc_tcp *tcp = (grpc_tcp *)ep;
- win_maybe_shutdown_resource_user(exec_ctx, tcp);
- TCP_UNREF(tcp, "destroy");
+ TCP_UNREF(exec_ctx, tcp, "destroy");
}
static char *win_get_peer(grpc_endpoint *ep) {
@@ -416,9 +399,11 @@ static grpc_workqueue *win_get_workqueue(grpc_endpoint *ep) { return NULL; }
static grpc_resource_user *win_get_resource_user(grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *)ep;
- return &tcp->resource_user;
+ return tcp->resource_user;
}
+static int win_get_fd(grpc_endpoint *ep) { return -1; }
+
static grpc_endpoint_vtable vtable = {win_read,
win_write,
win_get_workqueue,
@@ -427,7 +412,8 @@ static grpc_endpoint_vtable vtable = {win_read,
win_shutdown,
win_destroy,
win_get_resource_user,
- win_get_peer};
+ win_get_peer,
+ win_get_fd};
grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket,
grpc_resource_quota *resource_quota,
@@ -441,7 +427,7 @@ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket,
grpc_closure_init(&tcp->on_read, on_read, tcp);
grpc_closure_init(&tcp->on_write, on_write, tcp);
tcp->peer_string = gpr_strdup(peer_string);
- grpc_resource_user_init(&tcp->resource_user, resource_quota, peer_string);
+ tcp->resource_user = grpc_resource_user_create(resource_quota, peer_string);
/* Tell network status tracking code about the new endpoint */
grpc_network_status_register_endpoint(&tcp->base);
diff --git a/src/core/lib/iomgr/wakeup_fd_pipe.c b/src/core/lib/iomgr/wakeup_fd_pipe.c
index 183f0eb930..e186d63683 100644
--- a/src/core/lib/iomgr/wakeup_fd_pipe.c
+++ b/src/core/lib/iomgr/wakeup_fd_pipe.c
@@ -95,6 +95,8 @@ static void pipe_destroy(grpc_wakeup_fd* fd_info) {
static int pipe_check_availability(void) {
grpc_wakeup_fd fd;
+ fd.read_fd = fd.write_fd = -1;
+
if (pipe_init(&fd) == GRPC_ERROR_NONE) {
pipe_destroy(&fd);
return 1;