Merge pull request #1372 from ctiller/latency

Stop kicking ourselves

include/grpc/support/tls.h

@@ -44,7 +44,7 @@
    Thread locals have type gpr_intptr.
 
    Declaring a thread local variable 'foo':
-     GPR_TLS_DECL(foo, initial_value);
+     GPR_TLS_DECL(foo);
    Thread locals always have static scope.
 
    Initializing a thread local (must be done at library initialization 

src/core/iomgr/pollset_multipoller_with_poll_posix.c

@@ -203,7 +203,7 @@ static int multipoll_with_poll_pollset_maybe_work(
 }
 
 static void multipoll_with_poll_pollset_kick(grpc_pollset *p) {
-  grpc_pollset_kick_kick(&p->kick_state);
+  grpc_pollset_force_kick(p);
 }
 
 static void multipoll_with_poll_pollset_destroy(grpc_pollset *pollset) {

src/core/iomgr/pollset_posix.c

@@ -47,9 +47,11 @@
 #include "src/core/iomgr/fd_posix.h"
 #include "src/core/iomgr/iomgr_internal.h"
 #include "src/core/iomgr/socket_utils_posix.h"
+#include "src/core/profiling/timers.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/thd.h>
+#include <grpc/support/tls.h>
 #include <grpc/support/useful.h>
 
 static grpc_pollset g_backup_pollset;
@@ -57,6 +59,8 @@ static int g_shutdown_backup_poller;
 static gpr_event g_backup_poller_done;
 static gpr_event g_backup_pollset_shutdown_done;
 
+GPR_TLS_DECL(g_current_thread_poller);
+
 static void backup_poller(void *p) {
   gpr_timespec delta = gpr_time_from_millis(100);
   gpr_timespec last_poll = gpr_now();
@@ -76,17 +80,21 @@ static void backup_poller(void *p) {
 }
 
 void grpc_pollset_kick(grpc_pollset *p) {
-  if (p->counter) {
+  if (gpr_tls_get(&g_current_thread_poller) != (gpr_intptr)p && p->counter) {
     p->vtable->kick(p);
   }
 }
 
 void grpc_pollset_force_kick(grpc_pollset *p) {
-  grpc_pollset_kick_kick(&p->kick_state);
+  if (gpr_tls_get(&g_current_thread_poller) != (gpr_intptr)p) {
+    grpc_pollset_kick_kick(&p->kick_state);
+  }
 }
 
 static void kick_using_pollset_kick(grpc_pollset *p) {
-  grpc_pollset_kick_kick(&p->kick_state);
+  if (gpr_tls_get(&g_current_thread_poller) != (gpr_intptr)p) {
+    grpc_pollset_kick_kick(&p->kick_state);
+  }
 }
 
 /* global state management */
@@ -96,6 +104,8 @@ grpc_pollset *grpc_backup_pollset(void) { return &g_backup_pollset; }
 void grpc_pollset_global_init(void) {
   gpr_thd_id id;
 
+  gpr_tls_init(&g_current_thread_poller);
+
   /* Initialize kick fd state */
   grpc_pollset_kick_global_init();
 
@@ -129,6 +139,8 @@ void grpc_pollset_global_shutdown(void) {
 
   /* destroy the kick pipes */
   grpc_pollset_kick_global_destroy();
+
+  gpr_tls_destroy(&g_current_thread_poller);
 }
 
 /* main interface */
@@ -161,8 +173,8 @@ void grpc_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) {
 
 int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
   /* pollset->mu already held */
-  gpr_timespec now;
-  now = gpr_now();
+  gpr_timespec now = gpr_now();
+  int r;
   if (gpr_time_cmp(now, deadline) > 0) {
     return 0;
   }
@@ -172,7 +184,10 @@ int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
   if (grpc_alarm_check(&pollset->mu, now, &deadline)) {
     return 1;
   }
-  return pollset->vtable->maybe_work(pollset, deadline, now, 1);
+  gpr_tls_set(&g_current_thread_poller, (gpr_intptr)pollset);
+  r = pollset->vtable->maybe_work(pollset, deadline, now, 1);
+  gpr_tls_set(&g_current_thread_poller, 0);
+  return r;
 }
 
 void grpc_pollset_shutdown(grpc_pollset *pollset,
@@ -396,6 +411,7 @@ static int unary_poll_pollset_maybe_work(grpc_pollset *pollset,
   pfd[1].events = grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT, &fd_watcher);
 
   r = poll(pfd, GPR_ARRAY_SIZE(pfd), timeout);
+  GRPC_TIMER_MARK(POLL_FINISHED, r);
 
   grpc_fd_end_poll(&fd_watcher);