From 7f5fac90547d54f0f98d510995e697e6ea3ec9ed Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 7 Apr 2017 21:47:41 +0000 Subject: minor optimizations --- test/cpp/microbenchmarks/bm_pollset.cc | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) (limited to 'test/cpp') diff --git a/test/cpp/microbenchmarks/bm_pollset.cc b/test/cpp/microbenchmarks/bm_pollset.cc index 0f3d3cef66..3789deebde 100644 --- a/test/cpp/microbenchmarks/bm_pollset.cc +++ b/test/cpp/microbenchmarks/bm_pollset.cc @@ -136,8 +136,7 @@ static void BM_PollEmptyPollset(benchmark::State& state) { gpr_timespec deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC); gpr_mu_lock(mu); while (state.KeepRunning()) { - grpc_pollset_worker* worker; - GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, &worker, now, deadline)); + GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, NULL, now, deadline)); } grpc_closure shutdown_ps_closure; grpc_closure_init(&shutdown_ps_closure, shutdown_ps, ps, @@ -233,8 +232,7 @@ static void BM_SingleThreadPollOneFd(benchmark::State& state) { grpc_fd_notify_on_read(&exec_ctx, wakeup, continue_closure); gpr_mu_lock(mu); while (!done) { - grpc_pollset_worker* worker; - GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, &worker, now, deadline)); + GRPC_ERROR_UNREF(grpc_pollset_work(&exec_ctx, ps, NULL, now, deadline)); } grpc_fd_orphan(&exec_ctx, wakeup, NULL, NULL, "done"); wakeup_fd.read_fd = 0; -- cgit v1.2.3 From d6255951dc6ca662ddd9cd12c64f783a8218bef2 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Mon, 10 Apr 2017 23:51:42 +0000 Subject: Add a benchmark of repeatedly adding an fd to a pollset --- test/cpp/microbenchmarks/bm_pollset.cc | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) (limited to 'test/cpp') diff --git a/test/cpp/microbenchmarks/bm_pollset.cc b/test/cpp/microbenchmarks/bm_pollset.cc index 3789deebde..2a0e96ddf1 100644 --- a/test/cpp/microbenchmarks/bm_pollset.cc +++ b/test/cpp/microbenchmarks/bm_pollset.cc @@ -149,6 +149,33 @@ static void BM_PollEmptyPollset(benchmark::State& state) { } BENCHMARK(BM_PollEmptyPollset); +static void BM_PollAddFd(benchmark::State& state) { + TrackCounters track_counters; + size_t ps_sz = grpc_pollset_size(); + grpc_pollset* ps = static_cast(gpr_zalloc(ps_sz)); + gpr_mu* mu; + grpc_pollset_init(ps, &mu); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_wakeup_fd wakeup_fd; + GPR_ASSERT(GRPC_LOG_IF_ERROR("wakeup_fd_init", grpc_wakeup_fd_init(&wakeup_fd))); + grpc_fd *fd = grpc_fd_create(wakeup_fd.read_fd, "xxx"); + while (state.KeepRunning()) { + grpc_pollset_add_fd(&exec_ctx, ps, fd); + grpc_exec_ctx_flush(&exec_ctx); + } + grpc_fd_orphan(&exec_ctx, fd, NULL, NULL, "xxx"); + grpc_closure shutdown_ps_closure; + grpc_closure_init(&shutdown_ps_closure, shutdown_ps, ps, + grpc_schedule_on_exec_ctx); + gpr_mu_lock(mu); + grpc_pollset_shutdown(&exec_ctx, ps, &shutdown_ps_closure); + gpr_mu_unlock(mu); + grpc_exec_ctx_finish(&exec_ctx); + gpr_free(ps); + track_counters.Finish(state); +} +BENCHMARK(BM_PollAddFd); + class Closure : public grpc_closure { public: virtual ~Closure() {} -- cgit v1.2.3 From 4f98e25f8be7f77e026141adcb40bd079441be40 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 13 Apr 2017 18:55:46 +0000 Subject: Better cost estimation --- test/cpp/qps/gen_build_yaml.py | 1 + tools/run_tests/generated/tests.json | 24 ++++++++++++------------ 2 files changed, 13 insertions(+), 12 deletions(-) (limited to 'test/cpp') diff --git a/test/cpp/qps/gen_build_yaml.py b/test/cpp/qps/gen_build_yaml.py index 2f035abedd..805b0faeec 100755 --- a/test/cpp/qps/gen_build_yaml.py +++ b/test/cpp/qps/gen_build_yaml.py @@ -66,6 +66,7 @@ def _scenario_json_string(scenario_json, is_tsan): def threads_required(scenario_json, where, is_tsan): scenario_json = mutate_scenario(scenario_json, is_tsan) if scenario_json['%s_config' % where]['%s_type' % where] == 'ASYNC_%s' % where.upper(): + if scenario_json['client_config']['client_channels'] == 1: return 1 return scenario_json['%s_config' % where].get('async_%s_threads' % where, 0) return scenario_json['client_config']['outstanding_rpcs_per_channel'] * scenario_json['client_config']['client_channels'] diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json index a08caf30d3..e58ef3e7d6 100644 --- a/tools/run_tests/generated/tests.json +++ b/tools/run_tests/generated/tests.json @@ -41330,7 +41330,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 100, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41432,7 +41432,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41484,7 +41484,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41909,7 +41909,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 100, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42011,7 +42011,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42063,7 +42063,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42536,7 +42536,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 100, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -42686,7 +42686,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -42762,7 +42762,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43391,7 +43391,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 100, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43541,7 +43541,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43617,7 +43617,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": "capacity", + "cpu_cost": 2, "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", -- cgit v1.2.3 From 79d24fb8eb6a6dfe51aaea6de4bf388ba55bdde6 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Mon, 17 Apr 2017 19:35:19 +0000 Subject: Fixes --- src/core/lib/iomgr/ev_epollex_linux.c | 22 ++++++++++++++++++---- test/core/util/port_server_client.c | 6 +++--- test/cpp/microbenchmarks/bm_pollset.cc | 2 +- 3 files changed, 22 insertions(+), 8 deletions(-) (limited to 'test/cpp') diff --git a/src/core/lib/iomgr/ev_epollex_linux.c b/src/core/lib/iomgr/ev_epollex_linux.c index f0b9ee39f6..31316cfc4d 100644 --- a/src/core/lib/iomgr/ev_epollex_linux.c +++ b/src/core/lib/iomgr/ev_epollex_linux.c @@ -161,6 +161,7 @@ struct grpc_fd { /* The fd is either closed or we relinquished control of it. In either cases, this indicates that the 'fd' on this structure is no longer valid */ + gpr_mu orphaned_mu; bool orphaned; gpr_atm read_closure; @@ -285,6 +286,7 @@ static void fd_destroy(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { /* Add the fd to the freelist */ grpc_iomgr_unregister_object(&fd->iomgr_object); pollable_destroy(&fd->pollable); + gpr_mu_destroy(&fd->orphaned_mu); gpr_mu_lock(&fd_freelist_mu); fd->freelist_next = fd_freelist; fd_freelist = fd; @@ -347,6 +349,7 @@ static grpc_fd *fd_create(int fd, const char *name) { gpr_atm_rel_store(&new_fd->refst, (gpr_atm)1); new_fd->fd = fd; + gpr_mu_init(&new_fd->orphaned_mu); new_fd->orphaned = false; grpc_lfev_init(&new_fd->read_closure); grpc_lfev_init(&new_fd->write_closure); @@ -374,11 +377,11 @@ static grpc_fd *fd_create(int fd, const char *name) { static int fd_wrapped_fd(grpc_fd *fd) { int ret_fd = -1; - gpr_mu_lock(&fd->pollable.po.mu); + gpr_mu_lock(&fd->orphaned_mu); if (!fd->orphaned) { ret_fd = fd->fd; } - gpr_mu_unlock(&fd->pollable.po.mu); + gpr_mu_unlock(&fd->orphaned_mu); return ret_fd; } @@ -390,6 +393,7 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *error = GRPC_ERROR_NONE; gpr_mu_lock(&fd->pollable.po.mu); + gpr_mu_lock(&fd->orphaned_mu); fd->on_done_closure = on_done; /* If release_fd is not NULL, we should be relinquishing control of the file @@ -413,6 +417,7 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure_sched(exec_ctx, fd->on_done_closure, GRPC_ERROR_REF(error)); + gpr_mu_unlock(&fd->orphaned_mu); gpr_mu_unlock(&fd->pollable.po.mu); UNREF_BY(exec_ctx, fd, 2, reason); /* Drop the reference */ GRPC_LOG_IF_ERROR("fd_orphan", GRPC_ERROR_REF(error)); @@ -545,9 +550,11 @@ static void pollable_init(pollable *p, polling_obj_type type) { } static void pollable_destroy(pollable *p) { - close(p->epfd); - grpc_wakeup_fd_destroy(&p->wakeup); po_destroy(&p->po); + if (p->epfd != -1) { + close(p->epfd); + grpc_wakeup_fd_destroy(&p->wakeup); + } } /* ensure that p->epfd, p->wakeup are initialized; p->po.mu must be held */ @@ -590,7 +597,13 @@ static grpc_error *pollable_add_fd(pollable *p, grpc_fd *fd) { grpc_error *error = GRPC_ERROR_NONE; static const char *err_desc = "pollable_add_fd"; const int epfd = p->epfd; + GPR_ASSERT(epfd != -1); + gpr_mu_lock(&fd->orphaned_mu); + if (fd->orphaned) { + gpr_mu_unlock(&fd->orphaned_mu); + return GRPC_ERROR_NONE; + } struct epoll_event ev_fd = { .events = EPOLLET | EPOLLIN | EPOLLOUT | EPOLLEXCLUSIVE, .data.ptr = fd}; if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd->fd, &ev_fd) != 0) { @@ -614,6 +627,7 @@ static grpc_error *pollable_add_fd(pollable *p, grpc_fd *fd) { append_error(&error, GRPC_OS_ERROR(errno, "epoll_ctl"), err_desc); } } + gpr_mu_unlock(&fd->orphaned_mu); return error; } diff --git a/test/core/util/port_server_client.c b/test/core/util/port_server_client.c index 2cd50a4ff7..a388d5f27b 100644 --- a/test/core/util/port_server_client.c +++ b/test/core/util/port_server_client.c @@ -60,7 +60,6 @@ static void destroy_pops_and_shutdown(grpc_exec_ctx *exec_ctx, void *p, grpc_pollset *pollset = grpc_polling_entity_pollset(p); grpc_pollset_destroy(exec_ctx, pollset); gpr_free(pollset); - grpc_shutdown(); } static void freed_port_from_server(grpc_exec_ctx *exec_ctx, void *arg, @@ -122,12 +121,13 @@ void grpc_free_port_using_server(int port) { gpr_mu_unlock(pr.mu); grpc_httpcli_context_destroy(&exec_ctx, &context); - grpc_exec_ctx_finish(&exec_ctx); grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops), shutdown_closure); grpc_exec_ctx_finish(&exec_ctx); gpr_free(path); grpc_http_response_destroy(&rsp); + + grpc_shutdown(); } typedef struct portreq { @@ -239,7 +239,6 @@ int grpc_pick_port_using_server(void) { grpc_closure_create(got_port_from_server, &pr, grpc_schedule_on_exec_ctx), &pr.response); grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); - grpc_exec_ctx_finish(&exec_ctx); gpr_mu_lock(pr.mu); while (pr.port == -1) { grpc_pollset_worker *worker = NULL; @@ -258,6 +257,7 @@ int grpc_pick_port_using_server(void) { grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops), shutdown_closure); grpc_exec_ctx_finish(&exec_ctx); + grpc_shutdown(); return pr.port; } diff --git a/test/cpp/microbenchmarks/bm_pollset.cc b/test/cpp/microbenchmarks/bm_pollset.cc index 2a0e96ddf1..19b3ba0a3c 100644 --- a/test/cpp/microbenchmarks/bm_pollset.cc +++ b/test/cpp/microbenchmarks/bm_pollset.cc @@ -59,7 +59,7 @@ extern "C" { auto& force_library_initialization = Library::get(); static void shutdown_ps(grpc_exec_ctx* exec_ctx, void* ps, grpc_error* error) { - grpc_pollset_destroy(static_cast(ps)); + grpc_pollset_destroy(exec_ctx, static_cast(ps)); } static void BM_CreateDestroyPollset(benchmark::State& state) { -- cgit v1.2.3 From cc92eb42a4404b92f7dacace68090fd1f9111c4a Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 19 Apr 2017 06:54:25 -0700 Subject: Update to new API --- test/cpp/microbenchmarks/bm_cq_multiple_threads.cc | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) (limited to 'test/cpp') diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc index 9d7f65d292..0d267da723 100644 --- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc +++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc @@ -67,7 +67,9 @@ static void pollset_init(grpc_pollset* ps, gpr_mu** mu) { *mu = &ps->mu; } -static void pollset_destroy(grpc_pollset* ps) { gpr_mu_destroy(&ps->mu); } +static void pollset_destroy(grpc_exec_ctx* exec_ctx, grpc_pollset* ps) { + gpr_mu_destroy(&ps->mu); +} static grpc_error* pollset_kick(grpc_pollset* p, grpc_pollset_worker* worker) { return GRPC_ERROR_NONE; -- cgit v1.2.3 From f62f8d37b60ebb6b5f510210320e0aca79690222 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Mon, 24 Apr 2017 18:38:54 +0000 Subject: Revert "Better cost estimation" This reverts commit 4f98e25f8be7f77e026141adcb40bd079441be40. --- test/cpp/qps/gen_build_yaml.py | 1 - tools/run_tests/generated/tests.json | 24 ++++++++++++------------ 2 files changed, 12 insertions(+), 13 deletions(-) (limited to 'test/cpp') diff --git a/test/cpp/qps/gen_build_yaml.py b/test/cpp/qps/gen_build_yaml.py index 805b0faeec..2f035abedd 100755 --- a/test/cpp/qps/gen_build_yaml.py +++ b/test/cpp/qps/gen_build_yaml.py @@ -66,7 +66,6 @@ def _scenario_json_string(scenario_json, is_tsan): def threads_required(scenario_json, where, is_tsan): scenario_json = mutate_scenario(scenario_json, is_tsan) if scenario_json['%s_config' % where]['%s_type' % where] == 'ASYNC_%s' % where.upper(): - if scenario_json['client_config']['client_channels'] == 1: return 1 return scenario_json['%s_config' % where].get('async_%s_threads' % where, 0) return scenario_json['client_config']['outstanding_rpcs_per_channel'] * scenario_json['client_config']['client_channels'] diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json index d6b5cc561c..120a84e8a4 100644 --- a/tools/run_tests/generated/tests.json +++ b/tools/run_tests/generated/tests.json @@ -41350,7 +41350,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 100, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41452,7 +41452,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 2, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41504,7 +41504,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 2, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -41929,7 +41929,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 100, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42031,7 +42031,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 2, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42083,7 +42083,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 2, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "tsan", @@ -42560,7 +42560,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 100, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -42714,7 +42714,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 2, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -42792,7 +42792,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 2, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43438,7 +43438,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 100, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43592,7 +43592,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 2, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", @@ -43670,7 +43670,7 @@ "ci_platforms": [ "linux" ], - "cpu_cost": 2, + "cpu_cost": "capacity", "defaults": "boringssl", "exclude_configs": [ "asan-noleaks", -- cgit v1.2.3 From 956920d84ead617222fe073f2e877211dfb9ce81 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Mon, 24 Apr 2017 13:09:12 -0700 Subject: clang-format --- src/core/lib/iomgr/ev_epollex_linux.c | 5 +++-- src/core/lib/surface/completion_queue.c | 3 ++- test/cpp/microbenchmarks/bm_pollset.cc | 11 ++++++----- 3 files changed, 11 insertions(+), 8 deletions(-) (limited to 'test/cpp') diff --git a/src/core/lib/iomgr/ev_epollex_linux.c b/src/core/lib/iomgr/ev_epollex_linux.c index d12c7ffe90..ee944bcfad 100644 --- a/src/core/lib/iomgr/ev_epollex_linux.c +++ b/src/core/lib/iomgr/ev_epollex_linux.c @@ -564,8 +564,9 @@ static grpc_error *pollable_materialize(pollable *p) { if (new_epfd < 0) { return GRPC_OS_ERROR(errno, "epoll_create1"); } else { - struct epoll_event ev = {.events = (uint32_t)(EPOLLIN | EPOLLET | EPOLLEXCLUSIVE), - .data.ptr = &global_wakeup_fd}; + struct epoll_event ev = { + .events = (uint32_t)(EPOLLIN | EPOLLET | EPOLLEXCLUSIVE), + .data.ptr = &global_wakeup_fd}; if (epoll_ctl(new_epfd, EPOLL_CTL_ADD, global_wakeup_fd.read_fd, &ev) != 0) { grpc_error *err = GRPC_OS_ERROR(errno, "epoll_ctl"); diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index 7f950cdad5..a739af2fe9 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -98,7 +98,8 @@ static void non_polling_poller_init(grpc_pollset *pollset, gpr_mu **mu) { *mu = &npp->mu; } -static void non_polling_poller_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) { +static void non_polling_poller_destroy(grpc_exec_ctx *exec_ctx, + grpc_pollset *pollset) { non_polling_poller *npp = (non_polling_poller *)pollset; gpr_mu_destroy(&npp->mu); } diff --git a/test/cpp/microbenchmarks/bm_pollset.cc b/test/cpp/microbenchmarks/bm_pollset.cc index 19b3ba0a3c..f5e8d13881 100644 --- a/test/cpp/microbenchmarks/bm_pollset.cc +++ b/test/cpp/microbenchmarks/bm_pollset.cc @@ -157,17 +157,18 @@ static void BM_PollAddFd(benchmark::State& state) { grpc_pollset_init(ps, &mu); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_wakeup_fd wakeup_fd; - GPR_ASSERT(GRPC_LOG_IF_ERROR("wakeup_fd_init", grpc_wakeup_fd_init(&wakeup_fd))); - grpc_fd *fd = grpc_fd_create(wakeup_fd.read_fd, "xxx"); + GPR_ASSERT( + GRPC_LOG_IF_ERROR("wakeup_fd_init", grpc_wakeup_fd_init(&wakeup_fd))); + grpc_fd* fd = grpc_fd_create(wakeup_fd.read_fd, "xxx"); while (state.KeepRunning()) { - grpc_pollset_add_fd(&exec_ctx, ps, fd); - grpc_exec_ctx_flush(&exec_ctx); + grpc_pollset_add_fd(&exec_ctx, ps, fd); + grpc_exec_ctx_flush(&exec_ctx); } grpc_fd_orphan(&exec_ctx, fd, NULL, NULL, "xxx"); grpc_closure shutdown_ps_closure; grpc_closure_init(&shutdown_ps_closure, shutdown_ps, ps, grpc_schedule_on_exec_ctx); - gpr_mu_lock(mu); + gpr_mu_lock(mu); grpc_pollset_shutdown(&exec_ctx, ps, &shutdown_ps_closure); gpr_mu_unlock(mu); grpc_exec_ctx_finish(&exec_ctx); -- cgit v1.2.3 From c3571791a5e20ed82cc2efebd21d48f898f13eba Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 2 May 2017 12:33:38 -0700 Subject: Isolate timer checking in its own thread --- CMakeLists.txt | 7 + Makefile | 7 + binding.gyp | 1 + build.yaml | 2 + config.m4 | 1 + gRPC-Core.podspec | 3 + grpc.gemspec | 2 + package.xml | 2 + src/core/lib/iomgr/ev_epoll_linux.c | 25 +--- src/core/lib/iomgr/ev_poll_posix.c | 31 +--- src/core/lib/iomgr/ev_posix.c | 2 - src/core/lib/iomgr/ev_posix.h | 2 - src/core/lib/iomgr/iomgr.c | 3 + src/core/lib/iomgr/pollset_windows.c | 2 - src/core/lib/iomgr/timer_manager.c | 166 +++++++++++++++++++++ src/core/lib/iomgr/timer_manager.h | 43 ++++++ src/core/lib/surface/completion_queue.c | 67 +++------ src/python/grpcio/grpc_core_dependencies.py | 1 + test/cpp/qps/client_async.cc | 48 +++--- tools/doxygen/Doxyfile.c++.internal | 2 + tools/doxygen/Doxyfile.core.internal | 2 + tools/run_tests/generated/sources_and_headers.json | 3 + vsprojects/vcxproj/grpc++/grpc++.vcxproj | 3 + vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters | 6 + .../grpc++_unsecure/grpc++_unsecure.vcxproj | 3 + .../grpc++_unsecure.vcxproj.filters | 6 + vsprojects/vcxproj/grpc/grpc.vcxproj | 3 + vsprojects/vcxproj/grpc/grpc.vcxproj.filters | 6 + .../vcxproj/grpc_test_util/grpc_test_util.vcxproj | 3 + .../grpc_test_util/grpc_test_util.vcxproj.filters | 6 + .../vcxproj/grpc_unsecure/grpc_unsecure.vcxproj | 3 + .../grpc_unsecure/grpc_unsecure.vcxproj.filters | 6 + 32 files changed, 340 insertions(+), 127 deletions(-) create mode 100644 src/core/lib/iomgr/timer_manager.c create mode 100644 src/core/lib/iomgr/timer_manager.h (limited to 'test/cpp') diff --git a/CMakeLists.txt b/CMakeLists.txt index 325344a81c..b52a42f7e0 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -984,6 +984,7 @@ add_library(grpc src/core/lib/iomgr/time_averaged_stats.c src/core/lib/iomgr/timer_generic.c src/core/lib/iomgr/timer_heap.c + src/core/lib/iomgr/timer_manager.c src/core/lib/iomgr/timer_uv.c src/core/lib/iomgr/udp_server.c src/core/lib/iomgr/unix_sockets_posix.c @@ -1310,6 +1311,7 @@ add_library(grpc_cronet src/core/lib/iomgr/time_averaged_stats.c src/core/lib/iomgr/timer_generic.c src/core/lib/iomgr/timer_heap.c + src/core/lib/iomgr/timer_manager.c src/core/lib/iomgr/timer_uv.c src/core/lib/iomgr/udp_server.c src/core/lib/iomgr/unix_sockets_posix.c @@ -1621,6 +1623,7 @@ add_library(grpc_test_util src/core/lib/iomgr/time_averaged_stats.c src/core/lib/iomgr/timer_generic.c src/core/lib/iomgr/timer_heap.c + src/core/lib/iomgr/timer_manager.c src/core/lib/iomgr/timer_uv.c src/core/lib/iomgr/udp_server.c src/core/lib/iomgr/unix_sockets_posix.c @@ -1877,6 +1880,7 @@ add_library(grpc_unsecure src/core/lib/iomgr/time_averaged_stats.c src/core/lib/iomgr/timer_generic.c src/core/lib/iomgr/timer_heap.c + src/core/lib/iomgr/timer_manager.c src/core/lib/iomgr/timer_uv.c src/core/lib/iomgr/udp_server.c src/core/lib/iomgr/unix_sockets_posix.c @@ -2296,6 +2300,7 @@ add_library(grpc++ src/core/lib/iomgr/time_averaged_stats.c src/core/lib/iomgr/timer_generic.c src/core/lib/iomgr/timer_heap.c + src/core/lib/iomgr/timer_manager.c src/core/lib/iomgr/timer_uv.c src/core/lib/iomgr/udp_server.c src/core/lib/iomgr/unix_sockets_posix.c @@ -2621,6 +2626,7 @@ add_library(grpc++_cronet src/core/lib/iomgr/time_averaged_stats.c src/core/lib/iomgr/timer_generic.c src/core/lib/iomgr/timer_heap.c + src/core/lib/iomgr/timer_manager.c src/core/lib/iomgr/timer_uv.c src/core/lib/iomgr/udp_server.c src/core/lib/iomgr/unix_sockets_posix.c @@ -3390,6 +3396,7 @@ add_library(grpc++_unsecure src/core/lib/iomgr/time_averaged_stats.c src/core/lib/iomgr/timer_generic.c src/core/lib/iomgr/timer_heap.c + src/core/lib/iomgr/timer_manager.c src/core/lib/iomgr/timer_uv.c src/core/lib/iomgr/udp_server.c src/core/lib/iomgr/unix_sockets_posix.c diff --git a/Makefile b/Makefile index 4597b6fe97..d9de26f546 100644 --- a/Makefile +++ b/Makefile @@ -2967,6 +2967,7 @@ LIBGRPC_SRC = \ src/core/lib/iomgr/time_averaged_stats.c \ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_heap.c \ + src/core/lib/iomgr/timer_manager.c \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/udp_server.c \ src/core/lib/iomgr/unix_sockets_posix.c \ @@ -3291,6 +3292,7 @@ LIBGRPC_CRONET_SRC = \ src/core/lib/iomgr/time_averaged_stats.c \ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_heap.c \ + src/core/lib/iomgr/timer_manager.c \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/udp_server.c \ src/core/lib/iomgr/unix_sockets_posix.c \ @@ -3601,6 +3603,7 @@ LIBGRPC_TEST_UTIL_SRC = \ src/core/lib/iomgr/time_averaged_stats.c \ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_heap.c \ + src/core/lib/iomgr/timer_manager.c \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/udp_server.c \ src/core/lib/iomgr/unix_sockets_posix.c \ @@ -3829,6 +3832,7 @@ LIBGRPC_UNSECURE_SRC = \ src/core/lib/iomgr/time_averaged_stats.c \ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_heap.c \ + src/core/lib/iomgr/timer_manager.c \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/udp_server.c \ src/core/lib/iomgr/unix_sockets_posix.c \ @@ -4225,6 +4229,7 @@ LIBGRPC++_SRC = \ src/core/lib/iomgr/time_averaged_stats.c \ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_heap.c \ + src/core/lib/iomgr/timer_manager.c \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/udp_server.c \ src/core/lib/iomgr/unix_sockets_posix.c \ @@ -4558,6 +4563,7 @@ LIBGRPC++_CRONET_SRC = \ src/core/lib/iomgr/time_averaged_stats.c \ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_heap.c \ + src/core/lib/iomgr/timer_manager.c \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/udp_server.c \ src/core/lib/iomgr/unix_sockets_posix.c \ @@ -5317,6 +5323,7 @@ LIBGRPC++_UNSECURE_SRC = \ src/core/lib/iomgr/time_averaged_stats.c \ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_heap.c \ + src/core/lib/iomgr/timer_manager.c \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/udp_server.c \ src/core/lib/iomgr/unix_sockets_posix.c \ diff --git a/binding.gyp b/binding.gyp index 582c61282f..9434ab6d32 100644 --- a/binding.gyp +++ b/binding.gyp @@ -718,6 +718,7 @@ 'src/core/lib/iomgr/time_averaged_stats.c', 'src/core/lib/iomgr/timer_generic.c', 'src/core/lib/iomgr/timer_heap.c', + 'src/core/lib/iomgr/timer_manager.c', 'src/core/lib/iomgr/timer_uv.c', 'src/core/lib/iomgr/udp_server.c', 'src/core/lib/iomgr/unix_sockets_posix.c', diff --git a/build.yaml b/build.yaml index 4cfa75cae1..9453c1a2b3 100644 --- a/build.yaml +++ b/build.yaml @@ -238,6 +238,7 @@ filegroups: - src/core/lib/iomgr/timer.h - src/core/lib/iomgr/timer_generic.h - src/core/lib/iomgr/timer_heap.h + - src/core/lib/iomgr/timer_manager.h - src/core/lib/iomgr/timer_uv.h - src/core/lib/iomgr/udp_server.h - src/core/lib/iomgr/unix_sockets_posix.h @@ -350,6 +351,7 @@ filegroups: - src/core/lib/iomgr/time_averaged_stats.c - src/core/lib/iomgr/timer_generic.c - src/core/lib/iomgr/timer_heap.c + - src/core/lib/iomgr/timer_manager.c - src/core/lib/iomgr/timer_uv.c - src/core/lib/iomgr/udp_server.c - src/core/lib/iomgr/unix_sockets_posix.c diff --git a/config.m4 b/config.m4 index bbd667c9ec..36c04cf07b 100644 --- a/config.m4 +++ b/config.m4 @@ -154,6 +154,7 @@ if test "$PHP_GRPC" != "no"; then src/core/lib/iomgr/time_averaged_stats.c \ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_heap.c \ + src/core/lib/iomgr/timer_manager.c \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/udp_server.c \ src/core/lib/iomgr/unix_sockets_posix.c \ diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 07755ac727..120463f40c 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -320,6 +320,7 @@ Pod::Spec.new do |s| 'src/core/lib/iomgr/timer.h', 'src/core/lib/iomgr/timer_generic.h', 'src/core/lib/iomgr/timer_heap.h', + 'src/core/lib/iomgr/timer_manager.h', 'src/core/lib/iomgr/timer_uv.h', 'src/core/lib/iomgr/udp_server.h', 'src/core/lib/iomgr/unix_sockets_posix.h', @@ -531,6 +532,7 @@ Pod::Spec.new do |s| 'src/core/lib/iomgr/time_averaged_stats.c', 'src/core/lib/iomgr/timer_generic.c', 'src/core/lib/iomgr/timer_heap.c', + 'src/core/lib/iomgr/timer_manager.c', 'src/core/lib/iomgr/timer_uv.c', 'src/core/lib/iomgr/udp_server.c', 'src/core/lib/iomgr/unix_sockets_posix.c', @@ -781,6 +783,7 @@ Pod::Spec.new do |s| 'src/core/lib/iomgr/timer.h', 'src/core/lib/iomgr/timer_generic.h', 'src/core/lib/iomgr/timer_heap.h', + 'src/core/lib/iomgr/timer_manager.h', 'src/core/lib/iomgr/timer_uv.h', 'src/core/lib/iomgr/udp_server.h', 'src/core/lib/iomgr/unix_sockets_posix.h', diff --git a/grpc.gemspec b/grpc.gemspec index 1cd6d66335..7841f27102 100755 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -236,6 +236,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/iomgr/timer.h ) s.files += %w( src/core/lib/iomgr/timer_generic.h ) s.files += %w( src/core/lib/iomgr/timer_heap.h ) + s.files += %w( src/core/lib/iomgr/timer_manager.h ) s.files += %w( src/core/lib/iomgr/timer_uv.h ) s.files += %w( src/core/lib/iomgr/udp_server.h ) s.files += %w( src/core/lib/iomgr/unix_sockets_posix.h ) @@ -447,6 +448,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/iomgr/time_averaged_stats.c ) s.files += %w( src/core/lib/iomgr/timer_generic.c ) s.files += %w( src/core/lib/iomgr/timer_heap.c ) + s.files += %w( src/core/lib/iomgr/timer_manager.c ) s.files += %w( src/core/lib/iomgr/timer_uv.c ) s.files += %w( src/core/lib/iomgr/udp_server.c ) s.files += %w( src/core/lib/iomgr/unix_sockets_posix.c ) diff --git a/package.xml b/package.xml index e7d67eca18..3f7ddb9e49 100644 --- a/package.xml +++ b/package.xml @@ -245,6 +245,7 @@ + @@ -456,6 +457,7 @@ + diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c index e603a75593..735c37be95 100644 --- a/src/core/lib/iomgr/ev_epoll_linux.c +++ b/src/core/lib/iomgr/ev_epoll_linux.c @@ -76,11 +76,6 @@ static int grpc_polling_trace = 0; /* Disabled by default */ static int grpc_wakeup_signal = -1; static bool is_grpc_wakeup_signal_initialized = false; -/* TODO: sreek: Right now, this wakes up all pollers. In future we should make - * sure to wake up one polling thread (which can wake up other threads if - * needed) */ -static grpc_wakeup_fd global_wakeup_fd; - /* Implements the function defined in grpc_posix.h. This function might be * called before even calling grpc_init() to set either a different signal to * use. If signum == -1, then the use of signals is disabled */ @@ -454,8 +449,8 @@ static void polling_island_add_wakeup_fd_locked(polling_island *pi, gpr_asprintf(&err_msg, "epoll_ctl (epoll_fd: %d) add wakeup fd: %d failed with " "error: %d (%s)", - pi->epoll_fd, GRPC_WAKEUP_FD_GET_READ_FD(&global_wakeup_fd), - errno, strerror(errno)); + pi->epoll_fd, GRPC_WAKEUP_FD_GET_READ_FD(wakeup_fd), errno, + strerror(errno)); append_error(error, GRPC_OS_ERROR(errno, err_msg), err_desc); gpr_free(err_msg); } @@ -558,7 +553,6 @@ static polling_island *polling_island_create(grpc_exec_ctx *exec_ctx, goto done; } - polling_island_add_wakeup_fd_locked(pi, &global_wakeup_fd, error); polling_island_add_wakeup_fd_locked(pi, &pi->workqueue_wakeup_fd, error); if (initial_fd != NULL) { @@ -1116,11 +1110,10 @@ static grpc_error *pollset_global_init(void) { gpr_tls_init(&g_current_thread_pollset); gpr_tls_init(&g_current_thread_worker); poller_kick_init(); - return grpc_wakeup_fd_init(&global_wakeup_fd); + return GRPC_ERROR_NONE; } static void pollset_global_shutdown(void) { - grpc_wakeup_fd_destroy(&global_wakeup_fd); gpr_tls_destroy(&g_current_thread_pollset); gpr_tls_destroy(&g_current_thread_worker); } @@ -1226,10 +1219,6 @@ static grpc_error *pollset_kick(grpc_pollset *p, return error; } -static grpc_error *kick_poller(void) { - return grpc_wakeup_fd_wakeup(&global_wakeup_fd); -} - static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) { gpr_mu_init(&pollset->po.mu); *mu = &pollset->po.mu; @@ -1453,11 +1442,7 @@ static void pollset_work_and_unlock(grpc_exec_ctx *exec_ctx, for (int i = 0; i < ep_rv; ++i) { void *data_ptr = ep_ev[i].data.ptr; - if (data_ptr == &global_wakeup_fd) { - grpc_timer_consume_kick(); - append_error(error, grpc_wakeup_fd_consume_wakeup(&global_wakeup_fd), - err_desc); - } else if (data_ptr == &pi->workqueue_wakeup_fd) { + if (data_ptr == &pi->workqueue_wakeup_fd) { append_error(error, grpc_wakeup_fd_consume_wakeup(&pi->workqueue_wakeup_fd), err_desc); @@ -1897,8 +1882,6 @@ static const grpc_event_engine_vtable vtable = { .pollset_set_add_fd = pollset_set_add_fd, .pollset_set_del_fd = pollset_set_del_fd, - .kick_poller = kick_poller, - .workqueue_ref = workqueue_ref, .workqueue_unref = workqueue_unref, .workqueue_scheduler = workqueue_scheduler, diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c index 9834cdd197..3ef24be569 100644 --- a/src/core/lib/iomgr/ev_poll_posix.c +++ b/src/core/lib/iomgr/ev_poll_posix.c @@ -122,8 +122,6 @@ struct grpc_fd { grpc_pollset *read_notifier_pollset; }; -static grpc_wakeup_fd global_wakeup_fd; - /* 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 @@ -784,19 +782,14 @@ static grpc_error *pollset_kick(grpc_pollset *p, 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(&global_wakeup_fd); + return GRPC_ERROR_NONE; } static void pollset_global_shutdown(void) { - grpc_wakeup_fd_destroy(&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(&global_wakeup_fd); -} - /* main interface */ static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) { @@ -952,13 +945,10 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, } fd_count = 0; - pfd_count = 2; - pfds[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&global_wakeup_fd); + pfd_count = 1; + pfds[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker.wakeup_fd->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 < pollset->fd_count; i++) { if (fd_is_orphaned(pollset->fds[i])) { GRPC_FD_UNREF(pollset->fds[i], "multipoller"); @@ -974,7 +964,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, pollset->fd_count = fd_count; gpr_mu_unlock(&pollset->mu); - for (i = 2; i < pfd_count; i++) { + for (i = 1; i < pfd_count; i++) { grpc_fd *fd = watchers[i].fd; pfds[i].events = (short)fd_begin_poll(fd, pollset, &worker, POLLIN, POLLOUT, &watchers[i]); @@ -992,7 +982,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, work_combine_error(&error, GRPC_OS_ERROR(errno, "poll")); } - for (i = 2; i < pfd_count; i++) { + for (i = 1; i < pfd_count; i++) { if (watchers[i].fd == NULL) { fd_end_poll(exec_ctx, &watchers[i], 0, 0, NULL); } else { @@ -1002,20 +992,15 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, } } } else if (r == 0) { - for (i = 2; i < pfd_count; i++) { + for (i = 1; i < pfd_count; i++) { fd_end_poll(exec_ctx, &watchers[i], 0, 0, NULL); } } else { if (pfds[0].revents & POLLIN_CHECK) { - grpc_timer_consume_kick(); - work_combine_error(&error, - grpc_wakeup_fd_consume_wakeup(&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++) { + for (i = 1; i < pfd_count; i++) { if (watchers[i].fd == NULL) { fd_end_poll(exec_ctx, &watchers[i], 0, 0, NULL); } else { @@ -1560,8 +1545,6 @@ static const grpc_event_engine_vtable vtable = { .pollset_set_add_fd = pollset_set_add_fd, .pollset_set_del_fd = pollset_set_del_fd, - .kick_poller = kick_poller, - .workqueue_ref = workqueue_ref, .workqueue_unref = workqueue_unref, .workqueue_scheduler = workqueue_scheduler, diff --git a/src/core/lib/iomgr/ev_posix.c b/src/core/lib/iomgr/ev_posix.c index 13409a4de8..41464c137a 100644 --- a/src/core/lib/iomgr/ev_posix.c +++ b/src/core/lib/iomgr/ev_posix.c @@ -260,8 +260,6 @@ void grpc_pollset_set_del_fd(grpc_exec_ctx *exec_ctx, g_event_engine->pollset_set_del_fd(exec_ctx, pollset_set, fd); } -grpc_error *grpc_kick_poller(void) { return g_event_engine->kick_poller(); } - #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG grpc_workqueue *grpc_workqueue_ref(grpc_workqueue *workqueue, const char *file, int line, const char *reason) { diff --git a/src/core/lib/iomgr/ev_posix.h b/src/core/lib/iomgr/ev_posix.h index becc4d359e..a77720e61f 100644 --- a/src/core/lib/iomgr/ev_posix.h +++ b/src/core/lib/iomgr/ev_posix.h @@ -93,8 +93,6 @@ typedef struct grpc_event_engine_vtable { void (*pollset_set_del_fd)(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pollset_set, grpc_fd *fd); - grpc_error *(*kick_poller)(void); - void (*shutdown_engine)(void); #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG diff --git a/src/core/lib/iomgr/iomgr.c b/src/core/lib/iomgr/iomgr.c index 001e528409..445f7aa422 100644 --- a/src/core/lib/iomgr/iomgr.c +++ b/src/core/lib/iomgr/iomgr.c @@ -47,6 +47,7 @@ #include "src/core/lib/iomgr/iomgr_internal.h" #include "src/core/lib/iomgr/network_status_tracker.h" #include "src/core/lib/iomgr/timer.h" +#include "src/core/lib/iomgr/timer_manager.h" #include "src/core/lib/support/env.h" #include "src/core/lib/support/string.h" @@ -65,6 +66,7 @@ void grpc_iomgr_init(void) { g_root_object.name = "root"; grpc_network_status_init(); grpc_iomgr_platform_init(); + grpc_timer_manager_init(); } static size_t count_objects(void) { @@ -88,6 +90,7 @@ void grpc_iomgr_shutdown(grpc_exec_ctx *exec_ctx) { gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_seconds(10, GPR_TIMESPAN)); gpr_timespec last_warning_time = gpr_now(GPR_CLOCK_REALTIME); + grpc_timer_manager_shutdown(); grpc_iomgr_platform_flush(); gpr_mu_lock(&g_mu); diff --git a/src/core/lib/iomgr/pollset_windows.c b/src/core/lib/iomgr/pollset_windows.c index 04c6b71747..bea4232273 100644 --- a/src/core/lib/iomgr/pollset_windows.c +++ b/src/core/lib/iomgr/pollset_windows.c @@ -227,6 +227,4 @@ grpc_error *grpc_pollset_kick(grpc_pollset *p, return GRPC_ERROR_NONE; } -void grpc_kick_poller(void) { grpc_iocp_kick(); } - #endif /* GRPC_WINSOCK_SOCKET */ diff --git a/src/core/lib/iomgr/timer_manager.c b/src/core/lib/iomgr/timer_manager.c new file mode 100644 index 0000000000..00e868de01 --- /dev/null +++ b/src/core/lib/iomgr/timer_manager.c @@ -0,0 +1,166 @@ +/* + * + * Copyright 2017, 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/timer_manager.h" + +#include +#include +#include + +#include "src/core/lib/iomgr/timer.h" + +typedef struct completed_thread { + gpr_thd_id t; + struct completed_thread *next; +} completed_thread; + +static gpr_mu g_mu; +static gpr_cv g_cv_wait; +static gpr_cv g_cv_shutdown; +static int g_thread_count; +static int g_waiter_count; +static bool g_shutdown; +static completed_thread *g_completed_threads; +static bool g_kicked; + +#define MAX_WAITERS 3 + +static void timer_thread(void *unused); + +static void gc_completed_threads(void) { + if (g_completed_threads != NULL) { + completed_thread *to_gc = g_completed_threads; + g_completed_threads = NULL; + gpr_mu_unlock(&g_mu); + while (to_gc != NULL) { + gpr_thd_join(to_gc->t); + completed_thread *next = to_gc->next; + gpr_free(to_gc); + to_gc = next; + } + gpr_mu_lock(&g_mu); + } +} + +static void start_timer_thread_and_unlock(void) { + ++g_waiter_count; + ++g_thread_count; + gpr_mu_unlock(&g_mu); + gpr_log(GPR_DEBUG, "Spawn timer thread"); + gpr_thd_id thd; + gpr_thd_options opt = gpr_thd_options_default(); + gpr_thd_options_set_joinable(&opt); + gpr_thd_new(&thd, timer_thread, NULL, &opt); +} + +static void timer_thread(void *unused) { + grpc_exec_ctx exec_ctx = + GRPC_EXEC_CTX_INITIALIZER(0, grpc_never_ready_to_finish, NULL); + for (;;) { + gpr_timespec next = gpr_inf_future(GPR_CLOCK_MONOTONIC); + gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); + if (grpc_timer_check(&exec_ctx, now, &next)) { + gpr_mu_lock(&g_mu); + --g_waiter_count; + bool start_thread = g_waiter_count == 0; + if (start_thread && !g_shutdown) { + start_timer_thread_and_unlock(); + } else { + gpr_mu_unlock(&g_mu); + } + grpc_exec_ctx_flush(&exec_ctx); + gpr_mu_lock(&g_mu); + gc_completed_threads(); + ++g_waiter_count; + gpr_mu_unlock(&g_mu); + } else { + gpr_mu_lock(&g_mu); + if (g_shutdown) break; + if (gpr_cv_wait(&g_cv_wait, &g_mu, next)) { + if (g_kicked) { + grpc_timer_consume_kick(); + g_kicked = false; + } else if (g_waiter_count > MAX_WAITERS) { + break; + } + } + gpr_mu_unlock(&g_mu); + } + } + --g_waiter_count; + --g_thread_count; + if (0 == g_thread_count) { + gpr_cv_signal(&g_cv_shutdown); + } + completed_thread *ct = gpr_malloc(sizeof(*ct)); + ct->t = gpr_thd_currentid(); + ct->next = g_completed_threads; + g_completed_threads = ct; + gpr_mu_unlock(&g_mu); + gpr_log(GPR_DEBUG, "End timer thread"); +} + +void grpc_timer_manager_init(void) { + gpr_mu_init(&g_mu); + gpr_cv_init(&g_cv_wait); + gpr_cv_init(&g_cv_shutdown); + g_thread_count = 0; + g_waiter_count = 0; + g_shutdown = false; + g_completed_threads = NULL; + + gpr_mu_lock(&g_mu); + start_timer_thread_and_unlock(); +} + +void grpc_timer_manager_shutdown(void) { + gpr_mu_lock(&g_mu); + g_shutdown = true; + gpr_cv_broadcast(&g_cv_wait); + while (g_thread_count > 0) { + gpr_cv_wait(&g_cv_shutdown, &g_mu, gpr_inf_future(GPR_CLOCK_REALTIME)); + gc_completed_threads(); + } + gpr_mu_unlock(&g_mu); + + gpr_mu_destroy(&g_mu); + gpr_cv_destroy(&g_cv_wait); + gpr_cv_destroy(&g_cv_shutdown); +} + +void grpc_kick_poller(void) { + gpr_mu_lock(&g_mu); + g_kicked = true; + gpr_cv_signal(&g_cv_wait); + gpr_mu_unlock(&g_mu); +} diff --git a/src/core/lib/iomgr/timer_manager.h b/src/core/lib/iomgr/timer_manager.h new file mode 100644 index 0000000000..a24c9da328 --- /dev/null +++ b/src/core/lib/iomgr/timer_manager.h @@ -0,0 +1,43 @@ +/* + * + * Copyright 2017, 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_IOMGR_TIMER_MANAGER_H +#define GRPC_CORE_IOMGR_TIMER_MANAGER_H + +/* Timer Manager tries to keep one thread waiting for the next timeout at all + times */ + +void grpc_timer_manager_init(void); +void grpc_timer_manager_shutdown(void); + +#endif diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index eae3f103b1..048564e32f 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -580,31 +580,18 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, dump_pending_tags(cc); break; } - /* Check alarms - these are a global resource so we just ping - each time through on every pollset. - May update deadline to ensure timely wakeups. - TODO(ctiller): can this work be localized? */ - gpr_timespec iteration_deadline = deadline; - if (grpc_timer_check(&exec_ctx, now, &iteration_deadline)) { - GPR_TIMER_MARK("alarm_triggered", 0); + grpc_error *err = cc->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cc), + NULL, now, deadline); + if (err != GRPC_ERROR_NONE) { gpr_mu_unlock(cc->mu); - grpc_exec_ctx_flush(&exec_ctx); - gpr_mu_lock(cc->mu); - continue; - } else { - grpc_error *err = cc->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cc), - NULL, now, iteration_deadline); - if (err != GRPC_ERROR_NONE) { - gpr_mu_unlock(cc->mu); - const char *msg = grpc_error_string(err); - gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg); + const char *msg = grpc_error_string(err); + gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg); - GRPC_ERROR_UNREF(err); - memset(&ret, 0, sizeof(ret)); - ret.type = GRPC_QUEUE_TIMEOUT; - dump_pending_tags(cc); - break; - } + GRPC_ERROR_UNREF(err); + memset(&ret, 0, sizeof(ret)); + ret.type = GRPC_QUEUE_TIMEOUT; + dump_pending_tags(cc); + break; } is_finished_arg.first_loop = false; } @@ -773,31 +760,19 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, dump_pending_tags(cc); break; } - /* Check alarms - these are a global resource so we just ping - each time through on every pollset. - May update deadline to ensure timely wakeups. - TODO(ctiller): can this work be localized? */ - gpr_timespec iteration_deadline = deadline; - if (grpc_timer_check(&exec_ctx, now, &iteration_deadline)) { - GPR_TIMER_MARK("alarm_triggered", 0); + grpc_error *err = cc->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cc), + &worker, now, deadline); + if (err != GRPC_ERROR_NONE) { + del_plucker(cc, tag, &worker); gpr_mu_unlock(cc->mu); - grpc_exec_ctx_flush(&exec_ctx); - gpr_mu_lock(cc->mu); - } else { - grpc_error *err = cc->poller_vtable->work( - &exec_ctx, POLLSET_FROM_CQ(cc), &worker, now, iteration_deadline); - if (err != GRPC_ERROR_NONE) { - del_plucker(cc, tag, &worker); - gpr_mu_unlock(cc->mu); - const char *msg = grpc_error_string(err); - gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg); + const char *msg = grpc_error_string(err); + gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg); - GRPC_ERROR_UNREF(err); - memset(&ret, 0, sizeof(ret)); - ret.type = GRPC_QUEUE_TIMEOUT; - dump_pending_tags(cc); - break; - } + GRPC_ERROR_UNREF(err); + memset(&ret, 0, sizeof(ret)); + ret.type = GRPC_QUEUE_TIMEOUT; + dump_pending_tags(cc); + break; } is_finished_arg.first_loop = false; del_plucker(cc, tag, &worker); diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index d2a570cc87..02328867ae 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -143,6 +143,7 @@ CORE_SOURCE_FILES = [ 'src/core/lib/iomgr/time_averaged_stats.c', 'src/core/lib/iomgr/timer_generic.c', 'src/core/lib/iomgr/timer_heap.c', + 'src/core/lib/iomgr/timer_manager.c', 'src/core/lib/iomgr/timer_uv.c', 'src/core/lib/iomgr/udp_server.c', 'src/core/lib/iomgr/unix_sockets_posix.c', diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc index 29a79e7343..751986d7ac 100644 --- a/test/cpp/qps/client_async.cc +++ b/test/cpp/qps/client_async.cc @@ -238,39 +238,27 @@ class AsyncClient : public ClientImpl { void* got_tag; bool ok; - switch (cli_cqs_[thread_idx]->AsyncNext( - &got_tag, &ok, - std::chrono::system_clock::now() + std::chrono::milliseconds(10))) { - case CompletionQueue::GOT_EVENT: { - // Got a regular event, so process it - ClientRpcContext* ctx = ClientRpcContext::detag(got_tag); - // Proceed while holding a lock to make sure that - // this thread isn't supposed to shut down - std::lock_guard l(shutdown_state_[thread_idx]->mutex); - if (shutdown_state_[thread_idx]->shutdown) { - delete ctx; - return true; - } else if (!ctx->RunNextState(ok, entry)) { - // The RPC and callback are done, so clone the ctx - // and kickstart the new one - ctx->StartNewClone(cli_cqs_[thread_idx].get()); - // delete the old version - delete ctx; - } + if (cli_cqs_[thread_idx]->Next(&got_tag, &ok)) { + // Got a regular event, so process it + ClientRpcContext* ctx = ClientRpcContext::detag(got_tag); + // Proceed while holding a lock to make sure that + // this thread isn't supposed to shut down + std::lock_guard l(shutdown_state_[thread_idx]->mutex); + if (shutdown_state_[thread_idx]->shutdown) { + delete ctx; return true; + } else if (!ctx->RunNextState(ok, entry)) { + // The RPC and callback are done, so clone the ctx + // and kickstart the new one + ctx->StartNewClone(cli_cqs_[thread_idx].get()); + // delete the old version + delete ctx; } - case CompletionQueue::TIMEOUT: { - std::lock_guard l(shutdown_state_[thread_idx]->mutex); - if (shutdown_state_[thread_idx]->shutdown) { - return true; - } - return true; - } - case CompletionQueue::SHUTDOWN: // queue is shutting down, so we must be - // done - return true; + return true; + } else { + // queue is shutting down, so we must be done + return true; } - GPR_UNREACHABLE_CODE(return true); } std::vector> cli_cqs_; diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index 9664234f9f..e50d90648a 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -1026,6 +1026,8 @@ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_generic.h \ src/core/lib/iomgr/timer_heap.c \ src/core/lib/iomgr/timer_heap.h \ +src/core/lib/iomgr/timer_manager.c \ +src/core/lib/iomgr/timer_manager.h \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/timer_uv.h \ src/core/lib/iomgr/udp_server.c \ diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index f49c2de76c..24737de859 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -1161,6 +1161,8 @@ src/core/lib/iomgr/timer_generic.c \ src/core/lib/iomgr/timer_generic.h \ src/core/lib/iomgr/timer_heap.c \ src/core/lib/iomgr/timer_heap.h \ +src/core/lib/iomgr/timer_manager.c \ +src/core/lib/iomgr/timer_manager.h \ src/core/lib/iomgr/timer_uv.c \ src/core/lib/iomgr/timer_uv.h \ src/core/lib/iomgr/udp_server.c \ diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json index adaf5481b2..a9f6ac5cbd 100644 --- a/tools/run_tests/generated/sources_and_headers.json +++ b/tools/run_tests/generated/sources_and_headers.json @@ -7791,6 +7791,7 @@ "src/core/lib/iomgr/timer.h", "src/core/lib/iomgr/timer_generic.h", "src/core/lib/iomgr/timer_heap.h", + "src/core/lib/iomgr/timer_manager.h", "src/core/lib/iomgr/timer_uv.h", "src/core/lib/iomgr/udp_server.h", "src/core/lib/iomgr/unix_sockets_posix.h", @@ -7978,6 +7979,8 @@ "src/core/lib/iomgr/timer_generic.h", "src/core/lib/iomgr/timer_heap.c", "src/core/lib/iomgr/timer_heap.h", + "src/core/lib/iomgr/timer_manager.c", + "src/core/lib/iomgr/timer_manager.h", "src/core/lib/iomgr/timer_uv.c", "src/core/lib/iomgr/timer_uv.h", "src/core/lib/iomgr/udp_server.c", diff --git a/vsprojects/vcxproj/grpc++/grpc++.vcxproj b/vsprojects/vcxproj/grpc++/grpc++.vcxproj index 32d2e09a58..da9816f9a3 100644 --- a/vsprojects/vcxproj/grpc++/grpc++.vcxproj +++ b/vsprojects/vcxproj/grpc++/grpc++.vcxproj @@ -437,6 +437,7 @@ + @@ -702,6 +703,8 @@ + + diff --git a/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters b/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters index a3346bc297..924576d35b 100644 --- a/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters +++ b/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters @@ -322,6 +322,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr @@ -1031,6 +1034,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr diff --git a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj index 28ccefc651..3e4265bda2 100644 --- a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj @@ -431,6 +431,7 @@ + @@ -686,6 +687,8 @@ + + diff --git a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters index 83f869dab3..c17d4deca4 100644 --- a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters @@ -307,6 +307,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr @@ -998,6 +1001,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj b/vsprojects/vcxproj/grpc/grpc.vcxproj index 71520098a6..bb44a0772c 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj @@ -361,6 +361,7 @@ + @@ -641,6 +642,8 @@ + + diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters index de2dfe67e6..35002b1756 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters @@ -202,6 +202,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr @@ -992,6 +995,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr diff --git a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj index df89932a97..b5605a9381 100644 --- a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj +++ b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj @@ -256,6 +256,7 @@ + @@ -475,6 +476,8 @@ + + diff --git a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters index 22cfbe14d4..95e22922f4 100644 --- a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters @@ -259,6 +259,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr @@ -746,6 +749,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj index 0bfda72e81..8b59aa82dd 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj @@ -351,6 +351,7 @@ + @@ -608,6 +609,8 @@ + + diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters index 63c8d7f254..2cec671ebe 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters @@ -205,6 +205,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr @@ -902,6 +905,9 @@ src\core\lib\iomgr + + src\core\lib\iomgr + src\core\lib\iomgr -- cgit v1.2.3 From a1d00facbcfcd3a47d5617fd65f6ff5842c15649 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 10 May 2017 15:43:44 -0700 Subject: Finish pumped streams: should alleviate infinite hang --- test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc | 11 +++++++++++ 1 file changed, 11 insertions(+) (limited to 'test/cpp') diff --git a/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc b/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc index 47705d3031..01ff39121e 100644 --- a/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc +++ b/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc @@ -105,6 +105,17 @@ static void BM_PumpStreamClientToServer(benchmark::State& state) { GPR_ASSERT(need_tags & (1 << i)); need_tags &= ~(1 << i); } + response_rw.Finish(Status::OK, tag(0)); + Status final_status; + request_rw->Finish(&final_status, tag(1)); + need_tags = (1 << 0) | (1 << 1); + while (need_tags) { + GPR_ASSERT(fixture->cq()->Next(&t, &ok)); + int i = (int)(intptr_t)t; + GPR_ASSERT(need_tags & (1 << i)); + need_tags &= ~(1 << i); + } + GPR_ASSERT(final_status.ok()); } fixture->Finish(state); fixture.reset(); -- cgit v1.2.3