Merge branch 'we-dont-need-no-backup' into oops-i-split-it-again

16 files changed, 129 insertions, 45 deletions

diff --git a/src/core/channel/channel_args.c b/src/core/channel/channel_args.c
index 1b0e33b123..166d559a45 100644
--- a/src/core/channel/channel_args.c
+++ b/src/core/channel/channel_args.c
@@ -115,3 +115,27 @@ int grpc_channel_args_is_census_enabled(const grpc_channel_args *a) {
   }
   return 0;
 }
+
+grpc_compression_level grpc_channel_args_get_compression_level(
+    const grpc_channel_args *a) {
+  size_t i;
+  if (a) {
+    for (i = 0; a && i < a->num_args; ++i) {
+      if (a->args[i].type == GRPC_ARG_INTEGER &&
+          !strcmp(GRPC_COMPRESSION_LEVEL_ARG, a->args[i].key)) {
+        return a->args[i].value.integer;
+        break;
+      }
+    }
+  }
+  return GRPC_COMPRESS_LEVEL_NONE;
+}
+
+void grpc_channel_args_set_compression_level(
+    grpc_channel_args **a, grpc_compression_level level) {
+  grpc_arg tmp;
+  tmp.type = GRPC_ARG_INTEGER;
+  tmp.key = GRPC_COMPRESSION_LEVEL_ARG;
+  tmp.value.integer = level;
+  *a = grpc_channel_args_copy_and_add(*a, &tmp);
+}
diff --git a/src/core/channel/channel_args.h b/src/core/channel/channel_args.h
index eb5bf63986..bf747b26e6 100644
--- a/src/core/channel/channel_args.h
+++ b/src/core/channel/channel_args.h
@@ -34,21 +34,31 @@
 #ifndef GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H
 #define GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H
 
+#include <grpc/compression.h>
 #include <grpc/grpc.h>
 
 /* Copy some arguments */
 grpc_channel_args *grpc_channel_args_copy(const grpc_channel_args *src);
 
-/* Copy some arguments and add the to_add parameter in the end.
+/** Copy some arguments and add the to_add parameter in the end.
    If to_add is NULL, it is equivalent to call grpc_channel_args_copy. */
 grpc_channel_args *grpc_channel_args_copy_and_add(const grpc_channel_args *src,
                                                   const grpc_arg *to_add);
 
-/* Destroy arguments created by grpc_channel_args_copy */
+/** Destroy arguments created by grpc_channel_args_copy */
 void grpc_channel_args_destroy(grpc_channel_args *a);
 
-/* Reads census_enabled settings from channel args. Returns 1 if census_enabled
-   is specified in channel args, otherwise returns 0. */
+/** Reads census_enabled settings from channel args. Returns 1 if census_enabled
+ * is specified in channel args, otherwise returns 0. */
 int grpc_channel_args_is_census_enabled(const grpc_channel_args *a);
 
+/** Returns the compression level set in \a a. */
+grpc_compression_level grpc_channel_args_get_compression_level(
+    const grpc_channel_args *a);
+
+/** Sets the compression level in \a a to \a level. Setting it to
+ * GRPC_COMPRESS_LEVEL_NONE disables compression for the channel. */
+void grpc_channel_args_set_compression_level(
+    grpc_channel_args **a, grpc_compression_level level);
+
 #endif  /* GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H */
diff --git a/src/core/channel/child_channel.c b/src/core/channel/child_channel.c
index d3e6af7287..6690265d75 100644
--- a/src/core/channel/child_channel.c
+++ b/src/core/channel/child_channel.c
@@ -158,11 +158,8 @@ static void lb_destroy_channel_elem(grpc_channel_element *elem) {
 
 const grpc_channel_filter grpc_child_channel_top_filter = {
     lb_start_transport_op,   lb_channel_op,
-
     sizeof(lb_call_data),    lb_init_call_elem,    lb_destroy_call_elem,
-
     sizeof(lb_channel_data), lb_init_channel_elem, lb_destroy_channel_elem,
-
     "child-channel",
 };
 
diff --git a/src/core/channel/client_setup.c b/src/core/channel/client_setup.c
index a804a4712c..5be8fa66e9 100644
--- a/src/core/channel/client_setup.c
+++ b/src/core/channel/client_setup.c
@@ -56,6 +56,8 @@ struct grpc_client_setup {
   gpr_cv cv;
   grpc_client_setup_request *active_request;
   int refs;
+  /** The set of pollsets that are currently interested in this
+      connection being established */
   grpc_pollset_set interested_parties;
 };
 
@@ -92,7 +94,6 @@ static void setup_initiate(grpc_transport_setup *sp) {
   int in_alarm = 0;
 
   r->setup = s;
-  /* TODO(klempner): Actually set a deadline */
   r->deadline = gpr_time_add(gpr_now(), gpr_time_from_seconds(60));
 
   gpr_mu_lock(&s->mu);
@@ -117,25 +118,23 @@ static void setup_initiate(grpc_transport_setup *sp) {
   }
 }
 
+/** implementation of add_interested_party for setup vtable */
 static void setup_add_interested_party(grpc_transport_setup *sp,
                                        grpc_pollset *pollset) {
   grpc_client_setup *s = (grpc_client_setup *)sp;
 
   gpr_mu_lock(&s->mu);
-
   grpc_pollset_set_add_pollset(&s->interested_parties, pollset);
-
   gpr_mu_unlock(&s->mu);
 }
 
+/** implementation of del_interested_party for setup vtable */
 static void setup_del_interested_party(grpc_transport_setup *sp,
                                        grpc_pollset *pollset) {
   grpc_client_setup *s = (grpc_client_setup *)sp;
 
   gpr_mu_lock(&s->mu);
-
   grpc_pollset_set_del_pollset(&s->interested_parties, pollset);
-
   gpr_mu_unlock(&s->mu);
 }
 
@@ -232,7 +231,8 @@ int grpc_client_setup_request_should_continue(grpc_client_setup_request *r,
   return result;
 }
 
-static void backoff_alarm_done(void *arg /* grpc_client_setup */, int success) {
+static void backoff_alarm_done(void *arg /* grpc_client_setup_request */, 
+                               int success) {
   grpc_client_setup_request *r = arg;
   grpc_client_setup *s = r->setup;
   /* Handle status cancelled? */
diff --git a/src/core/compression/algorithm.c b/src/core/compression/algorithm.c
index 36ead843d2..4db48df6cb 100644
--- a/src/core/compression/algorithm.c
+++ b/src/core/compression/algorithm.c
@@ -31,6 +31,7 @@
  *
  */
 
+#include <stdlib.h>
 #include <grpc/compression.h>
 
 const char *grpc_compression_algorithm_name(
@@ -47,3 +48,20 @@ const char *grpc_compression_algorithm_name(
   }
   return "error";
 }
+
+/* TODO(dgq): Add the ability to specify parameters to the individual
+ * compression algorithms */
+grpc_compression_algorithm grpc_compression_algorithm_for_level(
+    grpc_compression_level level) {
+  switch (level) {
+    case GRPC_COMPRESS_LEVEL_NONE:
+      return GRPC_COMPRESS_NONE;
+    case GRPC_COMPRESS_LEVEL_LOW:
+    case GRPC_COMPRESS_LEVEL_MED:
+    case GRPC_COMPRESS_LEVEL_HIGH:
+      return GRPC_COMPRESS_DEFLATE;
+    default:
+      /* we shouldn't be making it here */
+      abort();
+  }
+}
diff --git a/src/core/httpcli/httpcli.h b/src/core/httpcli/httpcli.h
index cee374fe8e..06699e88c2 100644
--- a/src/core/httpcli/httpcli.h
+++ b/src/core/httpcli/httpcli.h
@@ -93,6 +93,10 @@ void grpc_httpcli_context_init(grpc_httpcli_context *context);
 void grpc_httpcli_context_destroy(grpc_httpcli_context *context);
 
 /* Asynchronously perform a HTTP GET.
+   'context' specifies the http context under which to do the get
+   'pollset' indicates a grpc_pollset that is interested in the result
+     of the get - work on this pollset may be used to progress the get
+     operation
    'request' contains request parameters - these are caller owned and can be
      destroyed once the call returns
    'deadline' contains a deadline for the request (or gpr_inf_future)
@@ -106,7 +110,19 @@ void grpc_httpcli_get(grpc_httpcli_context *context, grpc_pollset *pollset,
                       grpc_httpcli_response_cb on_response, void *user_data);
 
 /* Asynchronously perform a HTTP POST.
-   When there is no body, pass in NULL as body_bytes.
+   'context' specifies the http context under which to do the post
+   'pollset' indicates a grpc_pollset that is interested in the result
+     of the post - work on this pollset may be used to progress the post
+     operation
+   'request' contains request parameters - these are caller owned and can be
+     destroyed once the call returns
+   'body_bytes' and 'body_size' specify the payload for the post.
+     When there is no body, pass in NULL as body_bytes.
+   'deadline' contains a deadline for the request (or gpr_inf_future)
+   'em' points to a caller owned event manager that must be alive for the
+     lifetime of the request
+   'on_response' is a callback to report results to (and 'user_data' is a user
+     supplied pointer to pass to said call)
    Does not support ?var1=val1&var2=val2 in the path. */
 void grpc_httpcli_post(grpc_httpcli_context *context, grpc_pollset *pollset,
                        const grpc_httpcli_request *request,
diff --git a/src/core/iomgr/pollset.h b/src/core/iomgr/pollset.h
index da3c94ec11..7472b6144f 100644
--- a/src/core/iomgr/pollset.h
+++ b/src/core/iomgr/pollset.h
@@ -65,7 +65,7 @@ void grpc_pollset_destroy(grpc_pollset *pollset);
    May unlock GRPC_POLLSET_MU(pollset) during its execution. */
 int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline);
 
-/* Break a pollset out of polling work
+/* Break one polling thread out of polling work for this pollset.
    Requires GRPC_POLLSET_MU(pollset) locked. */
 void grpc_pollset_kick(grpc_pollset *pollset);
 
diff --git a/src/core/iomgr/pollset_kick_posix.h b/src/core/iomgr/pollset_kick_posix.h
index 0a404b601b..77e32a8d51 100644
--- a/src/core/iomgr/pollset_kick_posix.h
+++ b/src/core/iomgr/pollset_kick_posix.h
@@ -37,6 +37,11 @@
 #include "src/core/iomgr/wakeup_fd_posix.h"
 #include <grpc/support/sync.h>
 
+/* pollset kicking allows breaking a thread out of polling work for
+   a given pollset.
+   writing a byte to a pipe is used as a posix-ly portable base
+   mechanism, and eventfds are utilized on Linux for better performance. */
+
 typedef struct grpc_kick_fd_info {
   grpc_wakeup_fd_info wakeup_fd;
   /* used for polling list and free list */
@@ -67,7 +72,7 @@ void grpc_pollset_kick_destroy(grpc_pollset_kick_state *kick_state);
  * applicable. Intended for testing. */
 void grpc_pollset_kick_global_init_fallback_fd(void);
 
-/* Must be called before entering poll(). If return value is -1, this consumed
+/* Must be called before entering poll(). If return value is NULL, this consumed
    an existing kick. Otherwise the return value is an FD to add to the poll set.
  */
 grpc_kick_fd_info *grpc_pollset_kick_pre_poll(
@@ -82,6 +87,7 @@ void grpc_pollset_kick_consume(grpc_pollset_kick_state *kick_state,
 void grpc_pollset_kick_post_poll(grpc_pollset_kick_state *kick_state,
                                  grpc_kick_fd_info *fd_info);
 
+/* Actually kick */
 void grpc_pollset_kick_kick(grpc_pollset_kick_state *kick_state);
 
 #endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_POSIX_H */
diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c
index e5e3435feb..b4a526b9e7 100644
--- a/src/core/iomgr/pollset_multipoller_with_epoll.c
+++ b/src/core/iomgr/pollset_multipoller_with_epoll.c
@@ -97,15 +97,7 @@ static int multipoll_with_epoll_pollset_maybe_work(
    * here.
    */
 
-  if (gpr_time_cmp(deadline, gpr_inf_future) == 0) {
-    timeout_ms = -1;
-  } else {
-    timeout_ms = gpr_time_to_millis(
-        gpr_time_add(gpr_time_sub(deadline, now), gpr_time_from_micros(500)));
-    if (timeout_ms < 0) {
-      return 1;
-    }
-  }
+  timeout_ms = grpc_poll_deadline_to_millis_timeout(deadline, now);
   pollset->counter += 1;
   gpr_mu_unlock(&pollset->mu);
 
diff --git a/src/core/iomgr/pollset_multipoller_with_poll_posix.c b/src/core/iomgr/pollset_multipoller_with_poll_posix.c
index d21c52c0f0..2f108da66a 100644
--- a/src/core/iomgr/pollset_multipoller_with_poll_posix.c
+++ b/src/core/iomgr/pollset_multipoller_with_poll_posix.c
@@ -113,15 +113,7 @@ static int multipoll_with_poll_pollset_maybe_work(
   grpc_kick_fd_info *kfd;
 
   h = pollset->data.ptr;
-  if (gpr_time_cmp(deadline, gpr_inf_future) == 0) {
-    timeout = -1;
-  } else {
-    timeout = gpr_time_to_millis(
-        gpr_time_add(gpr_time_sub(deadline, now), gpr_time_from_micros(500)));
-    if (timeout < 0) {
-      return 1;
-    }
-  }
+  timeout = grpc_poll_deadline_to_millis_timeout(deadline, now);
   if (h->pfd_capacity < h->fd_count + 1) {
     h->pfd_capacity = GPR_MAX(h->pfd_capacity * 3 / 2, h->fd_count + 1);
     gpr_free(h->pfds);
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index 83ff1f26a5..46d3d132ce 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -188,6 +188,22 @@ void grpc_pollset_destroy(grpc_pollset *pollset) {
   gpr_mu_destroy(&pollset->mu);
 }
 
+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) == 0) {
+    return -1;
+  }
+  if (gpr_time_cmp(
+        deadline, 
+        gpr_time_add(now, gpr_time_from_micros(max_spin_polling_us))) <= 0) {
+    return 0;
+  }
+  timeout = gpr_time_sub(deadline, now);
+  return gpr_time_to_millis(
+      gpr_time_add(timeout, gpr_time_from_nanos(GPR_NS_PER_SEC - 1)));
+}
+
 /*
  * basic_pollset - a vtable that provides polling for zero or one file
  *                 descriptor via poll()
@@ -344,15 +360,7 @@ static int basic_pollset_maybe_work(grpc_pollset *pollset,
     GRPC_FD_UNREF(fd, "basicpoll");
     fd = pollset->data.ptr = NULL;
   }
-  if (gpr_time_cmp(deadline, gpr_inf_future) == 0) {
-    timeout = -1;
-  } else {
-    timeout = gpr_time_to_millis(
-        gpr_time_add(gpr_time_sub(deadline, now), gpr_time_from_micros(500)));
-    if (timeout < 0) {
-      return 1;
-    }
-  }
+  timeout = grpc_poll_deadline_to_millis_timeout(deadline, now);
   kfd = grpc_pollset_kick_pre_poll(&pollset->kick_state);
   if (kfd == NULL) {
     /* Already kicked */
diff --git a/src/core/iomgr/pollset_posix.h b/src/core/iomgr/pollset_posix.h
index 92c258e0cd..ba3d638d41 100644
--- a/src/core/iomgr/pollset_posix.h
+++ b/src/core/iomgr/pollset_posix.h
@@ -94,6 +94,15 @@ int grpc_kick_read_fd(grpc_pollset *p);
 /* Call after polling has been kicked to leave the kicked state */
 void grpc_kick_drain(grpc_pollset *p);
 
+/* Convert a timespec to milliseconds:
+   - very small or negative poll times are clamped to zero to do a 
+     non-blocking poll (which becomes spin polling)
+   - other small values are rounded up to one millisecond
+   - longer than a millisecond polls are rounded up to the next nearest 
+     millisecond to avoid spinning
+   - infinite timeouts are converted to -1 */
+int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline, gpr_timespec now);
+
 /* turn a pollset into a multipoller: platform specific */
 typedef void (*grpc_platform_become_multipoller_type)(grpc_pollset *pollset,
                                                       struct grpc_fd **fds,
diff --git a/src/core/iomgr/pollset_set.h b/src/core/iomgr/pollset_set.h
index 335ffb21b9..98e3b552a7 100644
--- a/src/core/iomgr/pollset_set.h
+++ b/src/core/iomgr/pollset_set.h
@@ -39,7 +39,7 @@
 /* A grpc_pollset_set is a set of pollsets that are interested in an
    action. Adding a pollset to a pollset_set automatically adds any
    fd's (etc) that have been registered with the set_set with that pollset.
-   Registering fd's automatically iterates all current pollsets. */
+   Registering fd's automatically adds them to all current pollsets. */
 
 #ifdef GPR_POSIX_SOCKET
 #include "src/core/iomgr/pollset_set_posix.h"
diff --git a/src/core/iomgr/tcp_client.h b/src/core/iomgr/tcp_client.h
index f40a5043c8..0fa08b52b0 100644
--- a/src/core/iomgr/tcp_client.h
+++ b/src/core/iomgr/tcp_client.h
@@ -41,7 +41,9 @@
 
 /* Asynchronously connect to an address (specified as (addr, len)), and call
    cb with arg and the completed connection when done (or call cb with arg and
-   NULL on failure) */
+   NULL on failure). 
+   interested_parties points to a set of pollsets that would be interested
+   in this connection being established (in order to continue their work) */
 void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp),
                              void *arg, grpc_pollset_set *interested_parties,
                              const struct sockaddr *addr, int addr_len,
diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c
index 2ac35f863a..5854031c9b 100644
--- a/src/core/iomgr/tcp_server_posix.c
+++ b/src/core/iomgr/tcp_server_posix.c
@@ -108,6 +108,7 @@ struct grpc_tcp_server {
   /* destroyed port count: how many ports are completely destroyed */
   size_t destroyed_ports;
 
+  /* is this server shutting down? (boolean) */
   int shutdown;
 
   /* all listening ports */
@@ -119,7 +120,9 @@ struct grpc_tcp_server {
   void (*shutdown_complete)(void *);
   void *shutdown_complete_arg;
 
+  /* all pollsets interested in new connections */
   grpc_pollset **pollsets;
+  /* number of pollsets in the pollsets array */
   size_t pollset_count;
 };
 
@@ -160,6 +163,9 @@ static void destroyed_port(void *server, int success) {
 
 static void dont_care_about_shutdown_completion(void *ignored) {}
 
+/* called when all listening endpoints have been shutdown, so no further
+   events will be received on them - at this point it's safe to destroy
+   things */
 static void deactivated_all_ports(grpc_tcp_server *s) {
   size_t i;
 
diff --git a/src/core/security/client_auth_filter.c b/src/core/security/client_auth_filter.c
index d11706ece0..e9bd45db68 100644
--- a/src/core/security/client_auth_filter.c
+++ b/src/core/security/client_auth_filter.c
@@ -53,6 +53,10 @@ typedef struct {
   grpc_credentials *creds;
   grpc_mdstr *host;
   grpc_mdstr *method;
+  /* pollset bound to this call; if we need to make external
+     network requests, they should be done under this pollset
+     so that work can progress when this call wants work to
+     progress */
   grpc_pollset *pollset;
   grpc_transport_op op;
   size_t op_md_idx;