Merge branch 'bm_call_create' into deadline_mu
diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c
index 720cfb4..8a8c3cb 100644
--- a/src/core/lib/channel/deadline_filter.c
+++ b/src/core/lib/channel/deadline_filter.c
@@ -43,6 +43,8 @@
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/slice/slice_internal.h"
+#define TOMBSTONE_TIMER 1
+
//
// grpc_deadline_state
//
@@ -52,9 +54,6 @@
grpc_error* error) {
grpc_call_element* elem = arg;
grpc_deadline_state* deadline_state = elem->call_data;
- gpr_mu_lock(&deadline_state->timer_mu);
- deadline_state->timer_pending = false;
- gpr_mu_unlock(&deadline_state->timer_mu);
if (error != GRPC_ERROR_CANCELLED) {
grpc_call_element_signal_error(
exec_ctx, elem,
@@ -66,53 +65,57 @@
}
// Starts the deadline timer.
-static void start_timer_if_needed_locked(grpc_exec_ctx* exec_ctx,
- grpc_call_element* elem,
- gpr_timespec deadline) {
- grpc_deadline_state* deadline_state = elem->call_data;
- deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
- // Note: We do not start the timer if there is already a timer
- // pending. This should be okay, because this is only called from two
- // functions exported by this module: grpc_deadline_state_start(), which
- // starts the initial timer, and grpc_deadline_state_reset(), which
- // cancels any pre-existing timer before starting a new one. In
- // particular, we want to ensure that if grpc_deadline_state_start()
- // winds up trying to start the timer after grpc_deadline_state_reset()
- // has already done so, we ignore the value from the former.
- if (!deadline_state->timer_pending &&
- gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_MONOTONIC)) != 0) {
- // Take a reference to the call stack, to be owned by the timer.
- GRPC_CALL_STACK_REF(deadline_state->call_stack, "deadline_timer");
- deadline_state->timer_pending = true;
- grpc_closure_init(&deadline_state->timer_callback, timer_callback, elem,
- grpc_schedule_on_exec_ctx);
- grpc_timer_init(exec_ctx, &deadline_state->timer, deadline,
- &deadline_state->timer_callback,
- gpr_now(GPR_CLOCK_MONOTONIC));
- }
-}
static void start_timer_if_needed(grpc_exec_ctx* exec_ctx,
grpc_call_element* elem,
gpr_timespec deadline) {
+ deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
+ if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_MONOTONIC)) == 0) {
+ return;
+ }
grpc_deadline_state* deadline_state = elem->call_data;
- gpr_mu_lock(&deadline_state->timer_mu);
- start_timer_if_needed_locked(exec_ctx, elem, deadline);
- gpr_mu_unlock(&deadline_state->timer_mu);
+ grpc_deadline_timer_state cur_state;
+ grpc_closure* closure = NULL;
+retry:
+ cur_state =
+ (grpc_deadline_timer_state)gpr_atm_acq_load(&deadline_state->timer_state);
+ switch (cur_state) {
+ case GRPC_DEADLINE_STATE_PENDING:
+ return;
+ case GRPC_DEADLINE_STATE_FINISHED:
+ if (gpr_atm_rel_cas(&deadline_state->timer_state,
+ GRPC_DEADLINE_STATE_FINISHED,
+ GRPC_DEADLINE_STATE_PENDING)) {
+ closure = grpc_closure_create(timer_callback, elem,
+ grpc_schedule_on_exec_ctx);
+ } else {
+ goto retry;
+ }
+ break;
+ case GRPC_DEADLINE_STATE_INITIAL:
+ if (gpr_atm_rel_cas(&deadline_state->timer_state,
+ GRPC_DEADLINE_STATE_INITIAL,
+ GRPC_DEADLINE_STATE_PENDING)) {
+ closure =
+ grpc_closure_init(&deadline_state->timer_callback, timer_callback,
+ elem, grpc_schedule_on_exec_ctx);
+ } else {
+ goto retry;
+ }
+ break;
+ }
+ GPR_ASSERT(closure);
+ grpc_timer_init(exec_ctx, &deadline_state->timer, deadline, closure,
+ gpr_now(GPR_CLOCK_MONOTONIC));
+ GPR_UNREACHABLE_CODE(return;);
}
// Cancels the deadline timer.
-static void cancel_timer_if_needed_locked(grpc_exec_ctx* exec_ctx,
- grpc_deadline_state* deadline_state) {
- if (deadline_state->timer_pending) {
- grpc_timer_cancel(exec_ctx, &deadline_state->timer);
- deadline_state->timer_pending = false;
- }
-}
static void cancel_timer_if_needed(grpc_exec_ctx* exec_ctx,
grpc_deadline_state* deadline_state) {
- gpr_mu_lock(&deadline_state->timer_mu);
- cancel_timer_if_needed_locked(exec_ctx, deadline_state);
- gpr_mu_unlock(&deadline_state->timer_mu);
+ if (gpr_atm_acq_load(&deadline_state->timer_state) !=
+ GRPC_DEADLINE_STATE_INITIAL) {
+ grpc_timer_cancel(exec_ctx, &deadline_state->timer);
+ }
}
// Callback run when the call is complete.
@@ -120,8 +123,8 @@
grpc_deadline_state* deadline_state = arg;
cancel_timer_if_needed(exec_ctx, deadline_state);
// Invoke the next callback.
- deadline_state->next_on_complete->cb(
- exec_ctx, deadline_state->next_on_complete->cb_arg, error);
+ grpc_closure_run(exec_ctx, deadline_state->next_on_complete,
+ GRPC_ERROR_REF(error));
}
// Inject our own on_complete callback into op.
@@ -138,14 +141,12 @@
grpc_deadline_state* deadline_state = elem->call_data;
memset(deadline_state, 0, sizeof(*deadline_state));
deadline_state->call_stack = call_stack;
- gpr_mu_init(&deadline_state->timer_mu);
}
void grpc_deadline_state_destroy(grpc_exec_ctx* exec_ctx,
grpc_call_element* elem) {
grpc_deadline_state* deadline_state = elem->call_data;
cancel_timer_if_needed(exec_ctx, deadline_state);
- gpr_mu_destroy(&deadline_state->timer_mu);
}
// Callback and associated state for starting the timer after call stack
@@ -187,10 +188,8 @@
void grpc_deadline_state_reset(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
gpr_timespec new_deadline) {
grpc_deadline_state* deadline_state = elem->call_data;
- gpr_mu_lock(&deadline_state->timer_mu);
- cancel_timer_if_needed_locked(exec_ctx, deadline_state);
- start_timer_if_needed_locked(exec_ctx, elem, new_deadline);
- gpr_mu_unlock(&deadline_state->timer_mu);
+ cancel_timer_if_needed(exec_ctx, deadline_state);
+ start_timer_if_needed(exec_ctx, elem, new_deadline);
}
void grpc_deadline_state_client_start_transport_stream_op(
diff --git a/src/core/lib/channel/deadline_filter.h b/src/core/lib/channel/deadline_filter.h
index bd2b84f..94717f6 100644
--- a/src/core/lib/channel/deadline_filter.h
+++ b/src/core/lib/channel/deadline_filter.h
@@ -35,16 +35,18 @@
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/iomgr/timer.h"
+typedef enum grpc_deadline_timer_state {
+ GRPC_DEADLINE_STATE_INITIAL,
+ GRPC_DEADLINE_STATE_PENDING,
+ GRPC_DEADLINE_STATE_FINISHED
+} grpc_deadline_timer_state;
+
// State used for filters that enforce call deadlines.
// Must be the first field in the filter's call_data.
typedef struct grpc_deadline_state {
// We take a reference to the call stack for the timer callback.
grpc_call_stack* call_stack;
- // Guards access to timer_pending and timer.
- gpr_mu timer_mu;
- // True if the timer callback is currently pending.
- bool timer_pending;
- // The deadline timer.
+ gpr_atm timer_state;
grpc_timer timer;
grpc_closure timer_callback;
// Closure to invoke when the call is complete.
diff --git a/src/core/lib/iomgr/timer_generic.c b/src/core/lib/iomgr/timer_generic.c
index 8a5617e..6d638bc 100644
--- a/src/core/lib/iomgr/timer_generic.c
+++ b/src/core/lib/iomgr/timer_generic.c
@@ -180,25 +180,25 @@
GPR_ASSERT(now.clock_type == g_clock_type);
timer->closure = closure;
timer->deadline = deadline;
- timer->triggered = 0;
if (!g_initialized) {
- timer->triggered = 1;
+ timer->pending = false;
grpc_closure_sched(
exec_ctx, timer->closure,
GRPC_ERROR_CREATE("Attempt to create timer before initialization"));
return;
}
+ gpr_mu_lock(&shard->mu);
+ timer->pending = true;
if (gpr_time_cmp(deadline, now) <= 0) {
- timer->triggered = 1;
+ timer->pending = false;
grpc_closure_sched(exec_ctx, timer->closure, GRPC_ERROR_NONE);
+ gpr_mu_unlock(&shard->mu);
+ /* early out */
return;
}
- /* TODO(ctiller): check deadline expired */
-
- gpr_mu_lock(&shard->mu);
grpc_time_averaged_stats_add_sample(&shard->stats,
ts_to_dbl(gpr_time_sub(deadline, now)));
if (gpr_time_cmp(deadline, shard->queue_deadline_cap) < 0) {
@@ -243,9 +243,9 @@
shard_type *shard = &g_shards[GPR_HASH_POINTER(timer, NUM_SHARDS)];
gpr_mu_lock(&shard->mu);
- if (!timer->triggered) {
+ if (timer->pending) {
grpc_closure_sched(exec_ctx, timer->closure, GRPC_ERROR_CANCELLED);
- timer->triggered = 1;
+ timer->pending = false;
if (timer->heap_index == INVALID_HEAP_INDEX) {
list_remove(timer);
} else {
@@ -296,7 +296,7 @@
}
timer = grpc_timer_heap_top(&shard->heap);
if (gpr_time_cmp(timer->deadline, now) > 0) return NULL;
- timer->triggered = 1;
+ timer->pending = false;
grpc_timer_heap_pop(&shard->heap);
return timer;
}
diff --git a/src/core/lib/iomgr/timer_generic.h b/src/core/lib/iomgr/timer_generic.h
index 9d901c7..1608dce 100644
--- a/src/core/lib/iomgr/timer_generic.h
+++ b/src/core/lib/iomgr/timer_generic.h
@@ -40,7 +40,7 @@
struct grpc_timer {
gpr_timespec deadline;
uint32_t heap_index; /* INVALID_HEAP_INDEX if not in heap */
- int triggered;
+ bool pending;
struct grpc_timer *next;
struct grpc_timer *prev;
grpc_closure *closure;
diff --git a/src/core/lib/iomgr/timer_uv.c b/src/core/lib/iomgr/timer_uv.c
index fa2cdee..f28a144 100644
--- a/src/core/lib/iomgr/timer_uv.c
+++ b/src/core/lib/iomgr/timer_uv.c
@@ -53,8 +53,8 @@
void run_expired_timer(uv_timer_t *handle) {
grpc_timer *timer = (grpc_timer *)handle->data;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- GPR_ASSERT(!timer->triggered);
- timer->triggered = 1;
+ GPR_ASSERT(timer->pending);
+ timer->pending = 0;
grpc_closure_sched(&exec_ctx, timer->closure, GRPC_ERROR_NONE);
stop_uv_timer(handle);
grpc_exec_ctx_finish(&exec_ctx);
@@ -67,11 +67,11 @@
uv_timer_t *uv_timer;
timer->closure = closure;
if (gpr_time_cmp(deadline, now) <= 0) {
- timer->triggered = 1;
+ timer->pending = 0;
grpc_closure_sched(exec_ctx, timer->closure, GRPC_ERROR_NONE);
return;
}
- timer->triggered = 0;
+ timer->pending = 1;
timeout = (uint64_t)gpr_time_to_millis(gpr_time_sub(deadline, now));
uv_timer = gpr_malloc(sizeof(uv_timer_t));
uv_timer_init(uv_default_loop(), uv_timer);
@@ -81,8 +81,8 @@
}
void grpc_timer_cancel(grpc_exec_ctx *exec_ctx, grpc_timer *timer) {
- if (!timer->triggered) {
- timer->triggered = 1;
+ if (timer->pending) {
+ timer->pending = 0;
grpc_closure_sched(exec_ctx, timer->closure, GRPC_ERROR_CANCELLED);
stop_uv_timer((uv_timer_t *)timer->uv_timer);
}
diff --git a/src/core/lib/iomgr/timer_uv.h b/src/core/lib/iomgr/timer_uv.h
index 13cf8bd..9870cd4 100644
--- a/src/core/lib/iomgr/timer_uv.h
+++ b/src/core/lib/iomgr/timer_uv.h
@@ -41,7 +41,7 @@
/* This is actually a uv_timer_t*, but we want to keep platform-specific
types out of headers */
void *uv_timer;
- int triggered;
+ int pending;
};
#endif /* GRPC_CORE_LIB_IOMGR_TIMER_UV_H */
diff --git a/tools/internal_ci/linux/grpc_interop.cfg b/tools/internal_ci/linux/grpc_interop_badserver_java.cfg
similarity index 94%
copy from tools/internal_ci/linux/grpc_interop.cfg
copy to tools/internal_ci/linux/grpc_interop_badserver_java.cfg
index 9259faf..e521b08 100644
--- a/tools/internal_ci/linux/grpc_interop.cfg
+++ b/tools/internal_ci/linux/grpc_interop_badserver_java.cfg
@@ -1,4 +1,3 @@
-#!/bin/bash
# Copyright 2017, Google Inc.
# All rights reserved.
#
@@ -31,11 +30,11 @@
# Config file for the internal CI (in protobuf text format)
# Location of the continuous shell script in repository.
-build_file: "grpc/tools/internal_ci/linux/grpc_interop.sh"
+build_file: "grpc/tools/internal_ci/linux/grpc_interop_badserver_java.sh"
# grpc_interop tests can take 6+ hours to complete.
timeout_mins: 480
action {
define_artifacts {
- regex: "**/sponge_log.xml"
+ regex: "**/report.xml"
}
}
diff --git a/tools/internal_ci/linux/grpc_interop.sh b/tools/internal_ci/linux/grpc_interop_badserver_java.sh
similarity index 85%
rename from tools/internal_ci/linux/grpc_interop.sh
rename to tools/internal_ci/linux/grpc_interop_badserver_java.sh
index 68bb419..0985e65 100755
--- a/tools/internal_ci/linux/grpc_interop.sh
+++ b/tools/internal_ci/linux/grpc_interop_badserver_java.sh
@@ -37,6 +37,5 @@
git submodule update --init
-tools/run_tests/run_interop_tests.py -l all -s all --cloud_to_prod --cloud_to_prod_auth --use_docker --http2_interop -t -j 12 $@ || FAILED="true"
-tools/run_tests/run_interop_tests.py -l java --use_docker --http2_badserver_interop $@ || FAILED="true"
-tools/run_tests/run_interop_tests.py -l python --use_docker --http2_badserver_interop $@ || FAILED="true"
+tools/run_tests/run_interop_tests.py -l java --use_docker --http2_badserver_interop $@
+
diff --git a/tools/internal_ci/linux/grpc_interop.cfg b/tools/internal_ci/linux/grpc_interop_badserver_python.cfg
similarity index 94%
rename from tools/internal_ci/linux/grpc_interop.cfg
rename to tools/internal_ci/linux/grpc_interop_badserver_python.cfg
index 9259faf..940f760 100644
--- a/tools/internal_ci/linux/grpc_interop.cfg
+++ b/tools/internal_ci/linux/grpc_interop_badserver_python.cfg
@@ -1,4 +1,3 @@
-#!/bin/bash
# Copyright 2017, Google Inc.
# All rights reserved.
#
@@ -31,11 +30,11 @@
# Config file for the internal CI (in protobuf text format)
# Location of the continuous shell script in repository.
-build_file: "grpc/tools/internal_ci/linux/grpc_interop.sh"
+build_file: "grpc/tools/internal_ci/linux/grpc_interop_badserver_python.sh"
# grpc_interop tests can take 6+ hours to complete.
timeout_mins: 480
action {
define_artifacts {
- regex: "**/sponge_log.xml"
+ regex: "**/report.xml"
}
}
diff --git a/tools/internal_ci/linux/grpc_interop.sh b/tools/internal_ci/linux/grpc_interop_badserver_python.sh
similarity index 85%
copy from tools/internal_ci/linux/grpc_interop.sh
copy to tools/internal_ci/linux/grpc_interop_badserver_python.sh
index 68bb419..3fff537 100755
--- a/tools/internal_ci/linux/grpc_interop.sh
+++ b/tools/internal_ci/linux/grpc_interop_badserver_python.sh
@@ -37,6 +37,5 @@
git submodule update --init
-tools/run_tests/run_interop_tests.py -l all -s all --cloud_to_prod --cloud_to_prod_auth --use_docker --http2_interop -t -j 12 $@ || FAILED="true"
-tools/run_tests/run_interop_tests.py -l java --use_docker --http2_badserver_interop $@ || FAILED="true"
-tools/run_tests/run_interop_tests.py -l python --use_docker --http2_badserver_interop $@ || FAILED="true"
+tools/run_tests/run_interop_tests.py -l python --use_docker --http2_badserver_interop $@
+
diff --git a/tools/internal_ci/linux/grpc_interop.cfg b/tools/internal_ci/linux/grpc_interop_tocloud.cfg
similarity index 94%
copy from tools/internal_ci/linux/grpc_interop.cfg
copy to tools/internal_ci/linux/grpc_interop_tocloud.cfg
index 9259faf..2b53644 100644
--- a/tools/internal_ci/linux/grpc_interop.cfg
+++ b/tools/internal_ci/linux/grpc_interop_tocloud.cfg
@@ -1,4 +1,3 @@
-#!/bin/bash
# Copyright 2017, Google Inc.
# All rights reserved.
#
@@ -31,11 +30,11 @@
# Config file for the internal CI (in protobuf text format)
# Location of the continuous shell script in repository.
-build_file: "grpc/tools/internal_ci/linux/grpc_interop.sh"
+build_file: "grpc/tools/internal_ci/linux/grpc_interop_tocloud.sh"
# grpc_interop tests can take 6+ hours to complete.
timeout_mins: 480
action {
define_artifacts {
- regex: "**/sponge_log.xml"
+ regex: "**/report.xml"
}
}
diff --git a/tools/internal_ci/linux/grpc_interop.sh b/tools/internal_ci/linux/grpc_interop_tocloud.sh
similarity index 82%
copy from tools/internal_ci/linux/grpc_interop.sh
copy to tools/internal_ci/linux/grpc_interop_tocloud.sh
index 68bb419..572001d 100755
--- a/tools/internal_ci/linux/grpc_interop.sh
+++ b/tools/internal_ci/linux/grpc_interop_tocloud.sh
@@ -37,6 +37,4 @@
git submodule update --init
-tools/run_tests/run_interop_tests.py -l all -s all --cloud_to_prod --cloud_to_prod_auth --use_docker --http2_interop -t -j 12 $@ || FAILED="true"
-tools/run_tests/run_interop_tests.py -l java --use_docker --http2_badserver_interop $@ || FAILED="true"
-tools/run_tests/run_interop_tests.py -l python --use_docker --http2_badserver_interop $@ || FAILED="true"
+tools/run_tests/run_interop_tests.py -l all -s all --use_docker --http2_interop -t -j 12 $@
diff --git a/tools/internal_ci/linux/grpc_master.cfg b/tools/internal_ci/linux/grpc_master.cfg
index 8ce2ef1..d5901bb 100644
--- a/tools/internal_ci/linux/grpc_master.cfg
+++ b/tools/internal_ci/linux/grpc_master.cfg
@@ -32,7 +32,7 @@
# Location of the continuous shell script in repository.
build_file: "grpc/tools/internal_ci/linux/grpc_master.sh"
-timeout_mins: 60
+timeout_mins: 240
action {
define_artifacts {
regex: "**/sponge_log.xml"