Merge pull request #2529 from ctiller/prefer-gin-n-tonic

Move alarm subsystem to monotonic time

16 files changed, 66 insertions, 27 deletions

diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index 35f172683a..487f5afb35 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -300,7 +300,7 @@ static void continue_connect(grpc_subchannel *c) {
 }
 
 static void start_connect(grpc_subchannel *c) {
-  gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME);
+  gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
   c->next_attempt = now;
   c->backoff_delta = gpr_time_from_seconds(1, GPR_TIMESPAN);
 
@@ -585,7 +585,7 @@ static void subchannel_connected(void *arg, int iomgr_success) {
     c->have_alarm = 1;
     c->next_attempt = gpr_time_add(c->next_attempt, c->backoff_delta);
     c->backoff_delta = gpr_time_add(c->backoff_delta, c->backoff_delta);
-    grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_REALTIME));
+    grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_MONOTONIC));
     gpr_mu_unlock(&c->mu);
   }
 }
diff --git a/src/core/iomgr/alarm.c b/src/core/iomgr/alarm.c
index 5b9a37e5dd..931f746f75 100644
--- a/src/core/iomgr/alarm.c
+++ b/src/core/iomgr/alarm.c
@@ -36,6 +36,7 @@
 #include "src/core/iomgr/alarm_heap.h"
 #include "src/core/iomgr/alarm_internal.h"
 #include "src/core/iomgr/time_averaged_stats.h"
+#include <grpc/support/log.h>
 #include <grpc/support/sync.h>
 #include <grpc/support/useful.h>
 
@@ -67,6 +68,7 @@ typedef struct {
 static gpr_mu g_mu;
 /* Allow only one run_some_expired_alarms at once */
 static gpr_mu g_checker_mu;
+static gpr_clock_type g_clock_type;
 static shard_type g_shards[NUM_SHARDS];
 /* Protected by g_mu */
 static shard_type *g_shard_queue[NUM_SHARDS];
@@ -85,6 +87,7 @@ void grpc_alarm_list_init(gpr_timespec now) {
 
   gpr_mu_init(&g_mu);
   gpr_mu_init(&g_checker_mu);
+  g_clock_type = now.clock_type;
 
   for (i = 0; i < NUM_SHARDS; i++) {
     shard_type *shard = &g_shards[i];
@@ -102,7 +105,7 @@ void grpc_alarm_list_init(gpr_timespec now) {
 
 void grpc_alarm_list_shutdown(void) {
   int i;
-  while (run_some_expired_alarms(NULL, gpr_inf_future(GPR_CLOCK_REALTIME), NULL,
+  while (run_some_expired_alarms(NULL, gpr_inf_future(g_clock_type), NULL,
                                  0))
     ;
   for (i = 0; i < NUM_SHARDS; i++) {
@@ -175,6 +178,8 @@ void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline,
                      gpr_timespec now) {
   int is_first_alarm = 0;
   shard_type *shard = &g_shards[shard_idx(alarm)];
+  GPR_ASSERT(deadline.clock_type == g_clock_type);
+  GPR_ASSERT(now.clock_type == g_clock_type);
   alarm->cb = alarm_cb;
   alarm->cb_arg = alarm_cb_arg;
   alarm->deadline = deadline;
@@ -355,6 +360,7 @@ static int run_some_expired_alarms(gpr_mu *drop_mu, gpr_timespec now,
 }
 
 int grpc_alarm_check(gpr_mu *drop_mu, gpr_timespec now, gpr_timespec *next) {
+  GPR_ASSERT(now.clock_type == g_clock_type);
   return run_some_expired_alarms(drop_mu, now, next, 1);
 }
 
diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c
index 0244f689b1..a18c176b30 100644
--- a/src/core/iomgr/iomgr.c
+++ b/src/core/iomgr/iomgr.c
@@ -57,9 +57,9 @@ static grpc_iomgr_object g_root_object;
 static void background_callback_executor(void *ignored) {
   gpr_mu_lock(&g_mu);
   while (!g_shutdown) {
-    gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
+    gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
     gpr_timespec short_deadline = gpr_time_add(
-        gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(100, GPR_TIMESPAN));
+        gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_millis(100, GPR_TIMESPAN));
     if (g_cbs_head) {
       grpc_iomgr_closure *closure = g_cbs_head;
       g_cbs_head = closure->next;
@@ -67,7 +67,7 @@ static void background_callback_executor(void *ignored) {
       gpr_mu_unlock(&g_mu);
       closure->cb(closure->cb_arg, closure->success);
       gpr_mu_lock(&g_mu);
-    } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_REALTIME),
+    } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_MONOTONIC),
                                 &deadline)) {
     } else {
       gpr_mu_unlock(&g_mu);
@@ -90,7 +90,7 @@ void grpc_iomgr_init(void) {
   gpr_thd_id id;
   gpr_mu_init(&g_mu);
   gpr_cv_init(&g_rcv);
-  grpc_alarm_list_init(gpr_now(GPR_CLOCK_REALTIME));
+  grpc_alarm_list_init(gpr_now(GPR_CLOCK_MONOTONIC));
   g_root_object.next = g_root_object.prev = &g_root_object;
   g_root_object.name = "root";
   grpc_iomgr_platform_init();
@@ -145,7 +145,7 @@ void grpc_iomgr_shutdown(void) {
       } while (g_cbs_head);
       continue;
     }
-    if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_REALTIME), NULL)) {
+    if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL)) {
       gpr_log(GPR_DEBUG, "got late alarm");
       continue;
     }
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index efb301d81c..c8646af615 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -136,7 +136,7 @@ static void finish_shutdown(grpc_pollset *pollset) {
 
 int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
   /* pollset->mu already held */
-  gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME);
+  gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
   if (gpr_time_cmp(now, deadline) > 0) {
     return 0;
   }
@@ -205,7 +205,7 @@ int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline,
                                          gpr_timespec now) {
   gpr_timespec timeout;
   static const int max_spin_polling_us = 10;
-  if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) {
+  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(
diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c
index 24226cc980..a9c4739c7c 100644
--- a/src/core/iomgr/pollset_windows.c
+++ b/src/core/iomgr/pollset_windows.c
@@ -70,7 +70,7 @@ void grpc_pollset_destroy(grpc_pollset *pollset) {
 
 int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
   gpr_timespec now;
-  now = gpr_now(GPR_CLOCK_REALTIME);
+  now = gpr_now(GPR_CLOCK_MONOTONIC);
   if (gpr_time_cmp(now, deadline) > 0) {
     return 0 /* GPR_FALSE */;
   }
diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c
index 2ae0251abb..41d8b169e0 100644
--- a/src/core/iomgr/tcp_client_posix.c
+++ b/src/core/iomgr/tcp_client_posix.c
@@ -253,8 +253,8 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep),
   ac->write_closure.cb_arg = ac;
 
   gpr_mu_lock(&ac->mu);
-  grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac,
-                  gpr_now(GPR_CLOCK_REALTIME));
+  grpc_alarm_init(&ac->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), 
+                  on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC));
   grpc_fd_notify_on_write(ac->fd, &ac->write_closure);
   gpr_mu_unlock(&ac->mu);
 
diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c
index 16741452b9..39fd43130b 100644
--- a/src/core/iomgr/tcp_client_windows.c
+++ b/src/core/iomgr/tcp_client_windows.c
@@ -216,7 +216,7 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp),
   ac->aborted = 0;
 
   grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac,
-                  gpr_now(GPR_CLOCK_REALTIME));
+                  gpr_now(GPR_CLOCK_MONOTONIC));
   socket->write_info.outstanding = 1;
   grpc_socket_notify_on_write(socket, on_connect, ac);
   return;
diff --git a/src/core/support/sync_posix.c b/src/core/support/sync_posix.c
index 41af8ceb0a..61572b9a8e 100644
--- a/src/core/support/sync_posix.c
+++ b/src/core/support/sync_posix.c
@@ -63,10 +63,11 @@ void gpr_cv_destroy(gpr_cv *cv) { GPR_ASSERT(pthread_cond_destroy(cv) == 0); }
 
 int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) {
   int err = 0;
-  if (gpr_time_cmp(abs_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) {
+  if (gpr_time_cmp(abs_deadline, gpr_inf_future(abs_deadline.clock_type)) == 0) {
     err = pthread_cond_wait(cv, mu);
   } else {
     struct timespec abs_deadline_ts;
+    abs_deadline = gpr_convert_clock_type(abs_deadline, GPR_CLOCK_REALTIME);
     abs_deadline_ts.tv_sec = abs_deadline.tv_sec;
     abs_deadline_ts.tv_nsec = abs_deadline.tv_nsec;
     err = pthread_cond_timedwait(cv, mu, &abs_deadline_ts);
diff --git a/src/core/support/sync_win32.c b/src/core/support/sync_win32.c
index 63196d10d3..54f84a46ac 100644
--- a/src/core/support/sync_win32.c
+++ b/src/core/support/sync_win32.c
@@ -83,10 +83,10 @@ int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) {
   int timeout = 0;
   DWORD timeout_max_ms;
   mu->locked = 0;
-  if (gpr_time_cmp(abs_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) {
+  if (gpr_time_cmp(abs_deadline, gpr_inf_future(abs_deadline.clock_type)) == 0) {
     SleepConditionVariableCS(cv, &mu->cs, INFINITE);
   } else {
-    gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME);
+    gpr_timespec now = gpr_now(abs_deadline.clock_type);
     gpr_int64 now_ms = now.tv_sec * 1000 + now.tv_nsec / 1000000;
     gpr_int64 deadline_ms =
         abs_deadline.tv_sec * 1000 + abs_deadline.tv_nsec / 1000000;
diff --git a/src/core/support/time.c b/src/core/support/time.c
index 570f195bd1..b523ae01cc 100644
--- a/src/core/support/time.c
+++ b/src/core/support/time.c
@@ -290,3 +290,30 @@ gpr_int32 gpr_time_to_millis(gpr_timespec t) {
 double gpr_timespec_to_micros(gpr_timespec t) {
   return (double)t.tv_sec * GPR_US_PER_SEC + t.tv_nsec * 1e-3;
 }
+
+gpr_timespec gpr_convert_clock_type(gpr_timespec t, gpr_clock_type clock_type) {
+  if (t.clock_type == clock_type) {
+    return t;
+  }
+
+  if (t.tv_nsec == 0) {
+    if (t.tv_sec == TYPE_MAX(time_t)) {
+      t.clock_type = clock_type;
+      return t;
+    }
+    if (t.tv_sec == TYPE_MIN(time_t)) {
+      t.clock_type = clock_type;
+      return t;
+    }
+  }
+
+  if (clock_type == GPR_TIMESPAN) {
+    return gpr_time_sub(t, gpr_now(t.clock_type));
+  }
+
+  if (t.clock_type == GPR_TIMESPAN) {
+    return gpr_time_add(gpr_now(clock_type), t);
+  }
+
+  return gpr_time_add(gpr_now(clock_type), gpr_time_sub(t, gpr_now(t.clock_type)));
+}
diff --git a/src/core/support/time_posix.c b/src/core/support/time_posix.c
index 258b2e640e..841485c4b4 100644
--- a/src/core/support/time_posix.c
+++ b/src/core/support/time_posix.c
@@ -120,7 +120,7 @@ void gpr_sleep_until(gpr_timespec until) {
   for (;;) {
     /* We could simplify by using clock_nanosleep instead, but it might be
      * slightly less portable. */
-    now = gpr_now(GPR_CLOCK_REALTIME);
+    now = gpr_now(until.clock_type);
     if (gpr_time_cmp(until, now) <= 0) {
       return;
     }
diff --git a/src/core/support/time_win32.c b/src/core/support/time_win32.c
index 238cd07ebc..7f64c80e27 100644
--- a/src/core/support/time_win32.c
+++ b/src/core/support/time_win32.c
@@ -80,7 +80,7 @@ void gpr_sleep_until(gpr_timespec until) {
   for (;;) {
     /* We could simplify by using clock_nanosleep instead, but it might be
      * slightly less portable. */
-    now = gpr_now(GPR_CLOCK_REALTIME);
+    now = gpr_now(until.clock_type);
     if (gpr_time_cmp(until, now) <= 0) {
       return;
     }
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 6e643b591c..e08273e451 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -348,7 +348,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
   }
   grpc_call_stack_init(channel_stack, server_transport_data, initial_op_ptr,
                        CALL_STACK_FROM_CALL(call));
-  if (gpr_time_cmp(send_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) {
+  if (gpr_time_cmp(send_deadline, gpr_inf_future(send_deadline.clock_type)) != 0) {
     set_deadline_alarm(call, send_deadline);
   }
   return call;
@@ -1278,8 +1278,8 @@ static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline) {
   }
   GRPC_CALL_INTERNAL_REF(call, "alarm");
   call->have_alarm = 1;
-  grpc_alarm_init(&call->alarm, deadline, call_alarm, call,
-                  gpr_now(GPR_CLOCK_REALTIME));
+  grpc_alarm_init(&call->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), call_alarm, call,
+                  gpr_now(GPR_CLOCK_MONOTONIC));
 }
 
 /* we offset status by a small amount when storing it into transport metadata
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 8484418247..3f60b0b0ba 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -148,6 +148,8 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
                                       gpr_timespec deadline) {
   grpc_event ret;
 
+  deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
+
   GRPC_CQ_INTERNAL_REF(cc, "next");
   gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
   for (;;) {
@@ -188,6 +190,8 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
   grpc_cq_completion *c;
   grpc_cq_completion *prev;
 
+  deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
+
   GRPC_CQ_INTERNAL_REF(cc, "pluck");
   gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
   for (;;) {
diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c
index d30059abf8..65b31a5afd 100644
--- a/src/core/transport/chttp2/stream_encoder.c
+++ b/src/core/transport/chttp2/stream_encoder.c
@@ -438,7 +438,7 @@ static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline,
   char timeout_str[GRPC_CHTTP2_TIMEOUT_ENCODE_MIN_BUFSIZE];
   grpc_mdelem *mdelem;
   grpc_chttp2_encode_timeout(
-      gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME)), timeout_str);
+      gpr_time_sub(deadline, gpr_now(deadline.clock_type)), timeout_str);
   mdelem = grpc_mdelem_from_metadata_strings(
       c->mdctx, GRPC_MDSTR_REF(c->timeout_key_str),
       grpc_mdstr_from_string(c->mdctx, timeout_str));
@@ -560,6 +560,7 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
   grpc_mdctx *mdctx = compressor->mdctx;
   grpc_linked_mdelem *l;
   int need_unref = 0;
+  gpr_timespec deadline;
 
   GPR_ASSERT(stream_id != 0);
 
@@ -589,9 +590,9 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
           l->md = hpack_enc(compressor, l->md, &st);
           need_unref |= l->md != NULL;
         }
-        if (gpr_time_cmp(op->data.metadata.deadline,
-                         gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) {
-          deadline_enc(compressor, op->data.metadata.deadline, &st);
+        deadline = op->data.metadata.deadline;
+        if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) != 0) {
+          deadline_enc(compressor, deadline, &st);
         }
         curop++;
         break;
diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c
index 9d127c5472..10d796fc15 100644
--- a/src/core/transport/transport_op_string.c
+++ b/src/core/transport/transport_op_string.c
@@ -61,7 +61,7 @@ static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) {
     if (m != md.list.head) gpr_strvec_add(b, gpr_strdup(", "));
     put_metadata(b, m->md);
   }
-  if (gpr_time_cmp(md.deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) {
+  if (gpr_time_cmp(md.deadline, gpr_inf_future(md.deadline.clock_type)) != 0) {
     char *tmp;
     gpr_asprintf(&tmp, " deadline=%d.%09d", md.deadline.tv_sec,
                  md.deadline.tv_nsec);