Better implementation, flip timer logic to make 0-init pre-triggered

src/core/lib/channel/deadline_filter.c

@@ -73,45 +73,48 @@ static void start_timer_if_needed(grpc_exec_ctx* exec_ctx,
     return;
   }
   grpc_deadline_state* deadline_state = elem->call_data;
-  for (size_t i = 0; i < GPR_ARRAY_SIZE(deadline_state->timers); i++) {
-    if (gpr_atm_acq_load(&deadline_state->timers[i]) == 0) {
-      grpc_deadline_timer* timer = (i == 0 ? &deadline_state->inlined_timer
-                                           : gpr_malloc(sizeof(*timer)));
-      if (gpr_atm_rel_cas(&deadline_state->timers[i], 0, (gpr_atm)timer)) {
-        grpc_timer_init(
-            exec_ctx, &timer->timer, deadline,
-            grpc_closure_init(&timer->timer_callback, timer_callback, elem,
-                              grpc_schedule_on_exec_ctx),
-            gpr_now(GPR_CLOCK_MONOTONIC));
-      } else if (i != 0) {
-        gpr_free(timer);
+  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(grpc_exec_ctx* exec_ctx,
                                    grpc_deadline_state* deadline_state) {
-  for (size_t i = 0; i < GPR_ARRAY_SIZE(deadline_state->timers); i++) {
-    gpr_atm timer_val;
-    timer_val = gpr_atm_acq_load(&deadline_state->timers[i]);
-    switch (timer_val) {
-      case 0:
-        return;
-      case TOMBSTONE_TIMER:
-        break;
-      default:
-        if (!gpr_atm_rel_cas(&deadline_state->timers[i], timer_val,
-                             TOMBSTONE_TIMER)) {
-          break;  // must have become a tombstone
-        }
-        grpc_deadline_timer* timer = (grpc_deadline_timer*)timer_val;
-        grpc_timer_cancel(exec_ctx, &timer->timer);
-        if (i != 0) gpr_free(timer);
-        break;
-    }
+  if (gpr_atm_acq_load(&deadline_state->timer_state) !=
+      GRPC_DEADLINE_STATE_INITIAL) {
+    grpc_timer_cancel(exec_ctx, &deadline_state->timer);
   }
 }
 
@@ -120,8 +123,8 @@ static void on_complete(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
   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.

src/core/lib/channel/deadline_filter.h

@@ -35,21 +35,20 @@
 #include "src/core/lib/channel/channel_stack.h"
 #include "src/core/lib/iomgr/timer.h"
 
-typedef struct grpc_deadline_timer {
-  grpc_timer timer;
-  grpc_closure timer_callback;
-} grpc_deadline_timer;
+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;
-  // We allow an initial timer and one reset.. these atomics point to the
-  // grpc_deadline_timer instances
-  gpr_atm timers[2];
-  // Pre-allocated initial timer.
-  grpc_deadline_timer inlined_timer;
+  gpr_atm timer_state;
+  grpc_timer timer;
+  grpc_closure timer_callback;
   // Closure to invoke when the call is complete.
   // We use this to cancel the timer.
   grpc_closure on_complete;

src/core/lib/iomgr/timer_generic.c

@@ -186,25 +186,25 @@ void grpc_timer_init(grpc_exec_ctx *exec_ctx, grpc_timer *timer,
   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) {
@@ -249,9 +249,9 @@ void grpc_timer_cancel(grpc_exec_ctx *exec_ctx, grpc_timer *timer) {
 
   shard_type *shard = &g_shards[shard_idx(timer)];
   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 {
@@ -302,7 +302,7 @@ static grpc_timer *pop_one(shard_type *shard, gpr_timespec now) {
     }
     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;
   }

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;

src/core/lib/iomgr/timer_uv.c

@@ -53,8 +53,8 @@ static void stop_uv_timer(uv_timer_t *handle) {
 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 @@ void grpc_timer_init(grpc_exec_ctx *exec_ctx, grpc_timer *timer,
   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_init(grpc_exec_ctx *exec_ctx, grpc_timer *timer,
 }
 
 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);
   }

src/core/lib/iomgr/timer_uv.h

@@ -41,7 +41,7 @@ struct grpc_timer {
   /* 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 */