aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib/iomgr/ev_epollex_linux.cc
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib/iomgr/ev_epollex_linux.cc')
-rw-r--r--src/core/lib/iomgr/ev_epollex_linux.cc80
1 files changed, 32 insertions, 48 deletions
diff --git a/src/core/lib/iomgr/ev_epollex_linux.cc b/src/core/lib/iomgr/ev_epollex_linux.cc
index bed47e4388..59dd8fd2fe 100644
--- a/src/core/lib/iomgr/ev_epollex_linux.cc
+++ b/src/core/lib/iomgr/ev_epollex_linux.cc
@@ -25,6 +25,7 @@
#include <assert.h>
#include <errno.h>
+#include <limits.h>
#include <poll.h>
#include <pthread.h>
#include <string.h>
@@ -38,7 +39,7 @@
#include <grpc/support/useful.h>
#include "src/core/lib/debug/stats.h"
-#include "src/core/lib/iomgr/ev_posix.h"
+#include "src/core/lib/iomgr/block_annotate.h"
#include "src/core/lib/iomgr/iomgr_internal.h"
#include "src/core/lib/iomgr/is_epollexclusive_available.h"
#include "src/core/lib/iomgr/lockfree_event.h"
@@ -46,19 +47,18 @@
#include "src/core/lib/iomgr/timer.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"
#include "src/core/lib/support/spinlock.h"
/*******************************************************************************
* Polling object
*/
-
typedef enum {
PO_POLLING_GROUP,
PO_POLLSET_SET,
PO_POLLSET,
- PO_FD, /* ordering is important: we always want to lock pollsets before fds:
- this guarantees that using an fd as a pollable is safe */
+ PO_FD,
+ /* ordering is important: we always want to lock pollsets before fds:
+ this guarantees that using an fd as a pollable is safe */
PO_EMPTY_POLLABLE,
PO_COUNT
} polling_obj_type;
@@ -690,32 +690,16 @@ static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) {
*mu = &pollset->pollable_obj.po.mu;
}
-/* Convert a timespec to milliseconds:
- - Very small or negative poll times are clamped to zero to do a non-blocking
- poll (which becomes spin polling)
- - Other small values are rounded up to one millisecond
- - Longer than a millisecond polls are rounded up to the next nearest
- millisecond to avoid spinning
- - Infinite timeouts are converted to -1 */
-static int poll_deadline_to_millis_timeout(gpr_timespec deadline,
- gpr_timespec now) {
- gpr_timespec timeout;
- if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) == 0) {
- return -1;
- }
-
- if (gpr_time_cmp(deadline, now) <= 0) {
+static int poll_deadline_to_millis_timeout(grpc_exec_ctx *exec_ctx,
+ grpc_millis millis) {
+ if (millis == GRPC_MILLIS_INF_FUTURE) return -1;
+ grpc_millis delta = millis - grpc_exec_ctx_now(exec_ctx);
+ if (delta > INT_MAX)
+ return INT_MAX;
+ else if (delta < 0)
return 0;
- }
-
- static const gpr_timespec round_up = {
- 0, /* tv_sec */
- GPR_NS_PER_MS - 1, /* tv_nsec */
- GPR_TIMESPAN /* clock_type */
- };
- timeout = gpr_time_sub(deadline, now);
- int millis = gpr_time_to_millis(gpr_time_add(timeout, round_up));
- return millis >= 1 ? millis : 1;
+ else
+ return (int)delta;
}
static void fd_become_readable(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
@@ -810,9 +794,8 @@ static void pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
}
static grpc_error *pollset_epoll(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
- pollable *p, gpr_timespec now,
- gpr_timespec deadline) {
- int timeout = poll_deadline_to_millis_timeout(deadline, now);
+ pollable *p, grpc_millis deadline) {
+ int timeout = poll_deadline_to_millis_timeout(exec_ctx, deadline);
if (GRPC_TRACER_ON(grpc_polling_trace)) {
char *desc = pollable_desc(p);
@@ -829,7 +812,7 @@ static grpc_error *pollset_epoll(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
r = epoll_wait(p->epfd, pollset->events, MAX_EPOLL_EVENTS, timeout);
} while (r < 0 && errno == EINTR);
if (timeout != 0) {
- GRPC_SCHEDULING_END_BLOCKING_REGION;
+ GRPC_SCHEDULING_END_BLOCKING_REGION_WITH_EXEC_CTX(exec_ctx);
}
if (r < 0) return GRPC_OS_ERROR(errno, "epoll_wait");
@@ -884,9 +867,10 @@ static worker_remove_result worker_remove(grpc_pollset_worker **root,
}
/* Return true if this thread should poll */
-static bool begin_worker(grpc_pollset *pollset, grpc_pollset_worker *worker,
- grpc_pollset_worker **worker_hdl, gpr_timespec *now,
- gpr_timespec deadline) {
+static bool begin_worker(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
+ grpc_pollset_worker *worker,
+ grpc_pollset_worker **worker_hdl,
+ grpc_millis deadline) {
bool do_poll = true;
if (worker_hdl != NULL) *worker_hdl = worker;
worker->initialized_cv = false;
@@ -910,10 +894,11 @@ static bool begin_worker(grpc_pollset *pollset, grpc_pollset_worker *worker,
worker->pollable_obj->root_worker != worker) {
gpr_log(GPR_DEBUG, "PS:%p wait %p w=%p for %dms", pollset,
worker->pollable_obj, worker,
- poll_deadline_to_millis_timeout(deadline, *now));
+ poll_deadline_to_millis_timeout(exec_ctx, deadline));
}
while (do_poll && worker->pollable_obj->root_worker != worker) {
- if (gpr_cv_wait(&worker->cv, &worker->pollable_obj->po.mu, deadline)) {
+ if (gpr_cv_wait(&worker->cv, &worker->pollable_obj->po.mu,
+ grpc_millis_to_timespec(deadline, GPR_CLOCK_REALTIME))) {
if (GRPC_TRACER_ON(grpc_polling_trace)) {
gpr_log(GPR_DEBUG, "PS:%p timeout_wait %p w=%p", pollset,
worker->pollable_obj, worker);
@@ -936,7 +921,7 @@ static bool begin_worker(grpc_pollset *pollset, grpc_pollset_worker *worker,
gpr_mu_lock(&pollset->pollable_obj.po.mu);
gpr_mu_lock(&worker->pollable_obj->po.mu);
}
- *now = gpr_now(now->clock_type);
+ grpc_exec_ctx_invalidate_now(exec_ctx);
}
return do_poll && pollset->shutdown_closure == NULL &&
@@ -967,14 +952,13 @@ static void end_worker(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
ensure that it is held by the time the function returns */
static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_pollset_worker **worker_hdl,
- gpr_timespec now, gpr_timespec deadline) {
+ grpc_millis deadline) {
grpc_pollset_worker worker;
if (0 && GRPC_TRACER_ON(grpc_polling_trace)) {
- gpr_log(GPR_DEBUG, "PS:%p work hdl=%p worker=%p now=%" PRId64
- ".%09d deadline=%" PRId64 ".%09d kwp=%d root_worker=%p",
- pollset, worker_hdl, &worker, now.tv_sec, now.tv_nsec,
- deadline.tv_sec, deadline.tv_nsec, pollset->kicked_without_poller,
- pollset->root_worker);
+ gpr_log(GPR_DEBUG, "PS:%p work hdl=%p worker=%p now=%" PRIdPTR
+ " deadline=%" PRIdPTR " kwp=%d root_worker=%p",
+ pollset, worker_hdl, &worker, grpc_exec_ctx_now(exec_ctx), deadline,
+ pollset->kicked_without_poller, pollset->root_worker);
}
grpc_error *error = GRPC_ERROR_NONE;
static const char *err_desc = "pollset_work";
@@ -985,7 +969,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
if (pollset->current_pollable_obj != &pollset->pollable_obj) {
gpr_mu_lock(&pollset->current_pollable_obj->po.mu);
}
- if (begin_worker(pollset, &worker, worker_hdl, &now, deadline)) {
+ if (begin_worker(exec_ctx, pollset, &worker, worker_hdl, deadline)) {
gpr_tls_set(&g_current_thread_pollset, (intptr_t)pollset);
gpr_tls_set(&g_current_thread_worker, (intptr_t)&worker);
GPR_ASSERT(!pollset->shutdown_closure);
@@ -996,7 +980,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
gpr_mu_unlock(&pollset->pollable_obj.po.mu);
if (pollset->event_cursor == pollset->event_count) {
append_error(&error, pollset_epoll(exec_ctx, pollset, worker.pollable_obj,
- now, deadline),
+ deadline),
err_desc);
}
append_error(&error, pollset_process_events(exec_ctx, pollset, false),
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-16 06:08:36 +0000