Merge branch 'master' of github.com:grpc/grpc into compression-accept-encoding

14 files changed, 103 insertions, 64 deletions

diff --git a/src/core/channel/channel_args.c b/src/core/channel/channel_args.c
index 55b4cb6170..54ee75af28 100644
--- a/src/core/channel/channel_args.c
+++ b/src/core/channel/channel_args.c
@@ -167,30 +167,36 @@ static int find_compression_algorithm_states_bitset(
 }
 
 grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
-    grpc_channel_args *a,
+    grpc_channel_args **a,
     grpc_compression_algorithm algorithm,
     int state) {
   int *states_arg;
-  grpc_channel_args *result = a;
+  grpc_channel_args *result = *a;
   const int states_arg_found =
-      find_compression_algorithm_states_bitset(a, &states_arg);
+      find_compression_algorithm_states_bitset(*a, &states_arg);
 
-  if (!states_arg_found) {
+  if (states_arg_found) {
+    if (state != 0) {
+      GPR_BITSET(states_arg, algorithm);
+    } else {
+      GPR_BITCLEAR(states_arg, algorithm);
+    }
+  } else {
     /* create a new arg */
     grpc_arg tmp;
     tmp.type = GRPC_ARG_INTEGER;
     tmp.key = GRPC_COMPRESSION_ALGORITHM_STATE_ARG;
-    states_arg = &tmp.value.integer;
-    result = grpc_channel_args_copy_and_add(a, &tmp, 1);
-  }
-
-  /* update either the new arg's value or the already present one */
-  if (state != 0) {
-    GPR_BITSET(states_arg, algorithm);
-  } else {
-    GPR_BITCLEAR(states_arg, algorithm);
+    /* all enabled by default */
+    tmp.value.integer = (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1;
+    if (state != 0) {
+      GPR_BITSET(&tmp.value.integer, algorithm);
+    } else {
+      GPR_BITCLEAR(&tmp.value.integer, algorithm);
+    }
+    result = grpc_channel_args_copy_and_add(*a, &tmp, 1);
+    grpc_channel_args_destroy(*a);
+    *a = result;
   }
-
   return result;
 }
 
@@ -200,6 +206,6 @@ int grpc_channel_args_compression_algorithm_get_states(
   if (find_compression_algorithm_states_bitset(a, &states_arg)) {
     return *states_arg;
   } else {
-    return  (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1; /* All algs. enabled */
+    return (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1; /* All algs. enabled */
   }
 }
diff --git a/src/core/channel/channel_args.h b/src/core/channel/channel_args.h
index e557f9a9d9..06a6012dee 100644
--- a/src/core/channel/channel_args.h
+++ b/src/core/channel/channel_args.h
@@ -70,9 +70,12 @@ grpc_channel_args *grpc_channel_args_set_compression_algorithm(
 /** Sets the support for the given compression algorithm. By default, all
  * compression algorithms are enabled. It's an error to disable an algorithm set
  * by grpc_channel_args_set_compression_algorithm.
- * */
+ *
+ * Returns an instance will the updated algorithm states. The \a a pointer is
+ * modified to point to the returned instance (which may be different from the
+ * input value of \a a). */
 grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
-    grpc_channel_args *a,
+    grpc_channel_args **a,
     grpc_compression_algorithm algorithm,
     int enabled);
 
diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c
index 39efc08ab5..eb27252d03 100644
--- a/src/core/channel/compress_filter.c
+++ b/src/core/channel/compress_filter.c
@@ -352,8 +352,8 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
   /* TODO(dgq): gpr_strjoin_sep could be made to work with statically allocated
    * arrays, as to avoid the heap allocs */
   accept_encoding_str =
-      gpr_strjoin_sep(supported_algorithms_names, supported_algorithms_idx,
-                      ", ", &accept_encoding_str_len);
+      gpr_strjoin_sep(supported_algorithms_names, supported_algorithms_idx, ",",
+                      &accept_encoding_str_len);
 
   channeld->mdelem_accept_encoding = grpc_mdelem_from_metadata_strings(
       mdctx, GRPC_MDSTR_REF(channeld->mdstr_compression_capabilities_key),
diff --git a/src/core/iomgr/pollset.h b/src/core/iomgr/pollset.h
index c474e4dbf1..337596cb74 100644
--- a/src/core/iomgr/pollset.h
+++ b/src/core/iomgr/pollset.h
@@ -74,10 +74,9 @@ void grpc_pollset_destroy(grpc_pollset *pollset);
    grpc_pollset_work, and it is guaranteed that GRPC_POLLSET_MU(pollset) will
    not be released by grpc_pollset_work AFTER worker has been destroyed.
 
-   Returns true if some work has been done, and false if the deadline
-   expired. */
-int grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
-                      gpr_timespec deadline);
+   Tries not to block past deadline. */
+void grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
+                       gpr_timespec now, gpr_timespec deadline);
 
 /* Break one polling thread out of polling work for this pollset.
    If specific_worker is GRPC_POLLSET_KICK_BROADCAST, kick ALL the workers.
diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c
index 5ea9dd2101..fe66ebed77 100644
--- a/src/core/iomgr/pollset_multipoller_with_epoll.c
+++ b/src/core/iomgr/pollset_multipoller_with_epoll.c
@@ -181,7 +181,7 @@ static void multipoll_with_epoll_pollset_maybe_work(
   pfds[1].events = POLLIN;
   pfds[1].revents = 0;
 
-  poll_rv = poll(pfds, 2, timeout_ms);
+  poll_rv = grpc_poll_function(pfds, 2, timeout_ms);
 
   if (poll_rv < 0) {
     if (errno != EINTR) {
diff --git a/src/core/iomgr/pollset_multipoller_with_poll_posix.c b/src/core/iomgr/pollset_multipoller_with_poll_posix.c
index 001fcecf76..30ee6e24db 100644
--- a/src/core/iomgr/pollset_multipoller_with_poll_posix.c
+++ b/src/core/iomgr/pollset_multipoller_with_poll_posix.c
@@ -144,7 +144,7 @@ static void multipoll_with_poll_pollset_maybe_work(
                                         POLLOUT, &watchers[i]);
   }
 
-  r = poll(pfds, pfd_count, timeout);
+  r = grpc_poll_function(pfds, pfd_count, timeout);
 
   for (i = 1; i < pfd_count; i++) {
     grpc_fd_end_poll(&watchers[i], pfds[i].revents & POLLIN,
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index a01f9ff727..6bd1b61f24 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -38,7 +38,6 @@
 #include "src/core/iomgr/pollset_posix.h"
 
 #include <errno.h>
-#include <poll.h>
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
@@ -57,6 +56,8 @@
 GPR_TLS_DECL(g_current_thread_poller);
 GPR_TLS_DECL(g_current_thread_worker);
 
+grpc_poll_function_type grpc_poll_function = poll;
+
 static void remove_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
   worker->prev->next = worker->next;
   worker->next->prev = worker->prev;
@@ -89,6 +90,7 @@ static void push_front_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
 }
 
 void grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) {
+  /* pollset->mu already held */
   if (specific_worker != NULL) {
     if (specific_worker == GRPC_POLLSET_KICK_BROADCAST) {
       for (specific_worker = p->root_worker.next;
@@ -168,14 +170,10 @@ static void finish_shutdown(grpc_pollset *pollset) {
   pollset->shutdown_done_cb(pollset->shutdown_done_arg);
 }
 
-int grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
-                      gpr_timespec deadline) {
+void grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
+                       gpr_timespec now, gpr_timespec deadline) {
   /* pollset->mu already held */
-  gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
   int added_worker = 0;
-  if (gpr_time_cmp(now, deadline) > 0) {
-    return 0;
-  }
   /* this must happen before we (potentially) drop pollset->mu */
   worker->next = worker->prev = NULL;
   /* TODO(ctiller): pool these */
@@ -217,7 +215,6 @@ done:
       gpr_mu_lock(&pollset->mu);
     }
   }
-  return 1;
 }
 
 void grpc_pollset_shutdown(grpc_pollset *pollset,
@@ -456,7 +453,7 @@ static void basic_pollset_maybe_work(grpc_pollset *pollset,
 
   /* poll fd count (argument 2) is shortened by one if we have no events
      to poll on - such that it only includes the kicker */
-  r = poll(pfd, nfds, timeout);
+  r = grpc_poll_function(pfd, nfds, timeout);
   GRPC_TIMER_MARK(GRPC_PTAG_POLL_FINISHED, r);
 
   if (fd) {
diff --git a/src/core/iomgr/pollset_posix.h b/src/core/iomgr/pollset_posix.h
index a3ea353de6..69bd9cca8c 100644
--- a/src/core/iomgr/pollset_posix.h
+++ b/src/core/iomgr/pollset_posix.h
@@ -34,6 +34,8 @@
 #ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_POSIX_H
 #define GRPC_INTERNAL_CORE_IOMGR_POLLSET_POSIX_H
 
+#include <poll.h>
+
 #include <grpc/support/sync.h>
 #include "src/core/iomgr/wakeup_fd_posix.h"
 
@@ -118,4 +120,8 @@ void grpc_poll_become_multipoller(grpc_pollset *pollset, struct grpc_fd **fds,
  * be locked) */
 int grpc_pollset_has_workers(grpc_pollset *pollset);
 
+/* override to allow tests to hook poll() usage */
+typedef int (*grpc_poll_function_type)(struct pollfd *, nfds_t, int);
+extern grpc_poll_function_type grpc_poll_function;
+
 #endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_POSIX_H */
diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c
index 8710395ab3..07522c8a0c 100644
--- a/src/core/iomgr/pollset_windows.c
+++ b/src/core/iomgr/pollset_windows.c
@@ -99,14 +99,9 @@ void grpc_pollset_destroy(grpc_pollset *pollset) {
   gpr_mu_destroy(&pollset->mu);
 }
 
-int grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker,
-                      gpr_timespec deadline) {
-  gpr_timespec now;
+void grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker, 
+                       gpr_timespec now, gpr_timespec deadline) {
   int added_worker = 0;
-  now = gpr_now(GPR_CLOCK_MONOTONIC);
-  if (gpr_time_cmp(now, deadline) > 0) {
-    return 0 /* GPR_FALSE */;
-  }
   worker->next = worker->prev = NULL;
   gpr_cv_init(&worker->cv);
   if (grpc_maybe_call_delayed_callbacks(&pollset->mu, 1 /* GPR_TRUE */)) {
@@ -127,7 +122,6 @@ done:
   if (added_worker) {
     remove_worker(pollset, worker);
   }
-  return 1 /* GPR_TRUE */;
 }
 
 void grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) {
diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c
index d6092ece32..3631de867a 100644
--- a/src/core/security/google_default_credentials.c
+++ b/src/core/security/google_default_credentials.c
@@ -115,7 +115,7 @@ static int is_stack_running_on_compute_engine(void) {
   gpr_mu_lock(GRPC_POLLSET_MU(&detector.pollset));
   while (!detector.is_done) {
     grpc_pollset_worker worker;
-    grpc_pollset_work(&detector.pollset, &worker,
+    grpc_pollset_work(&detector.pollset, &worker, gpr_now(GPR_CLOCK_MONOTONIC),
                       gpr_inf_future(GPR_CLOCK_MONOTONIC));
   }
   gpr_mu_unlock(GRPC_POLLSET_MU(&detector.pollset));
diff --git a/src/core/security/security_connector.c b/src/core/security/security_connector.c
index a354536dcd..ba9ac68c5f 100644
--- a/src/core/security/security_connector.c
+++ b/src/core/security/security_connector.c
@@ -575,6 +575,16 @@ grpc_security_status grpc_ssl_channel_security_connector_create(
   if (!check_request_metadata_creds(request_metadata_creds)) {
     goto error;
   }
+  if (config->pem_root_certs == NULL) {
+    pem_root_certs_size = grpc_get_default_ssl_roots(&pem_root_certs);
+    if (pem_root_certs == NULL || pem_root_certs_size == 0) {
+      gpr_log(GPR_ERROR, "Could not get default pem root certs.");
+      goto error;
+    }
+  } else {
+    pem_root_certs = config->pem_root_certs;
+    pem_root_certs_size = config->pem_root_certs_size;
+  }
 
   c = gpr_malloc(sizeof(grpc_ssl_channel_security_connector));
   memset(c, 0, sizeof(grpc_ssl_channel_security_connector));
@@ -590,16 +600,6 @@ grpc_security_status grpc_ssl_channel_security_connector_create(
   if (overridden_target_name != NULL) {
     c->overridden_target_name = gpr_strdup(overridden_target_name);
   }
-  if (config->pem_root_certs == NULL) {
-    pem_root_certs_size = grpc_get_default_ssl_roots(&pem_root_certs);
-    if (pem_root_certs == NULL || pem_root_certs_size == 0) {
-      gpr_log(GPR_ERROR, "Could not get default pem root certs.");
-      goto error;
-    }
-  } else {
-    pem_root_certs = config->pem_root_certs;
-    pem_root_certs_size = config->pem_root_certs_size;
-  }
   result = tsi_create_ssl_client_handshaker_factory(
       config->pem_private_key, config->pem_private_key_size,
       config->pem_cert_chain, config->pem_cert_chain_size, pem_root_certs,
diff --git a/src/core/security/server_auth_filter.c b/src/core/security/server_auth_filter.c
index 2f42f01f53..6e831431fa 100644
--- a/src/core/security/server_auth_filter.c
+++ b/src/core/security/server_auth_filter.c
@@ -104,24 +104,34 @@ static grpc_mdelem *remove_consumed_md(void *user_data, grpc_mdelem *md) {
   return md;
 }
 
-static void on_md_processing_done(void *user_data,
-                                  const grpc_metadata *consumed_md,
-                                  size_t num_consumed_md, int success) {
+static void on_md_processing_done(
+    void *user_data, const grpc_metadata *consumed_md, size_t num_consumed_md,
+    const grpc_metadata *response_md, size_t num_response_md,
+    grpc_status_code status, const char *error_details) {
   grpc_call_element *elem = user_data;
   call_data *calld = elem->call_data;
 
-  if (success) {
+  /* TODO(jboeuf): Implement support for response_md. */
+  if (response_md != NULL && num_response_md > 0) {
+    gpr_log(GPR_INFO,
+            "response_md in auth metadata processing not supported for now. "
+            "Ignoring...");
+  }
+
+  if (status == GRPC_STATUS_OK) {
     calld->consumed_md = consumed_md;
     calld->num_consumed_md = num_consumed_md;
     grpc_metadata_batch_filter(&calld->md_op->data.metadata, remove_consumed_md,
                                elem);
-    calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success);
+    calld->on_done_recv->cb(calld->on_done_recv->cb_arg, 1);
   } else {
-    gpr_slice message = gpr_slice_from_copied_string(
-        "Authentication metadata processing failed.");
+    gpr_slice message;
+    error_details = error_details != NULL
+                    ? error_details
+                    : "Authentication metadata processing failed.";
+    message = gpr_slice_from_copied_string(error_details);
     grpc_sopb_reset(calld->recv_ops);
-    grpc_transport_stream_op_add_close(&calld->transport_op,
-                                       GRPC_STATUS_UNAUTHENTICATED, &message);
+    grpc_transport_stream_op_add_close(&calld->transport_op, status, &message);
     grpc_call_next_op(elem, &calld->transport_op);
   }
 }
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 77443a7ae8..b58115a93f 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -170,6 +170,9 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
                                       gpr_timespec deadline, void *reserved) {
   grpc_event ret;
   grpc_pollset_worker worker;
+  int first_loop = 1;
+  gpr_timespec now;
+
   GPR_ASSERT(!reserved);
 
   deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
@@ -196,12 +199,15 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
       ret.type = GRPC_QUEUE_SHUTDOWN;
       break;
     }
-    if (!grpc_pollset_work(&cc->pollset, &worker, deadline)) {
+    now = gpr_now(GPR_CLOCK_MONOTONIC);
+    if (!first_loop && gpr_time_cmp(now, deadline) >= 0) {
       gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
       memset(&ret, 0, sizeof(ret));
       ret.type = GRPC_QUEUE_TIMEOUT;
       break;
     }
+    first_loop = 0;
+    grpc_pollset_work(&cc->pollset, &worker, now, deadline);
   }
   GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
   GRPC_CQ_INTERNAL_UNREF(cc, "next");
@@ -239,6 +245,9 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
   grpc_cq_completion *c;
   grpc_cq_completion *prev;
   grpc_pollset_worker worker;
+  gpr_timespec now;
+  int first_loop = 1;
+
   GPR_ASSERT(!reserved);
 
   deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
@@ -281,13 +290,16 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
       ret.type = GRPC_QUEUE_TIMEOUT;
       break;
     }
-    if (!grpc_pollset_work(&cc->pollset, &worker, deadline)) {
+    now = gpr_now(GPR_CLOCK_MONOTONIC);
+    if (!first_loop && gpr_time_cmp(now, deadline) >= 0) {
       del_plucker(cc, tag, &worker);
       gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
       memset(&ret, 0, sizeof(ret));
       ret.type = GRPC_QUEUE_TIMEOUT;
       break;
     }
+    first_loop = 0;
+    grpc_pollset_work(&cc->pollset, &worker, now, deadline);
     del_plucker(cc, tag, &worker);
   }
 done:
diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c
index 0f04169741..1ea697f71e 100644
--- a/src/core/transport/chttp2/stream_encoder.c
+++ b/src/core/transport/chttp2/stream_encoder.c
@@ -66,6 +66,8 @@ typedef struct {
   size_t header_idx;
   /* was the last frame emitted a header? (if yes, we'll need a CONTINUATION */
   gpr_uint8 last_was_header;
+  /* have we seen a regular (non-colon-prefixed) header yet? */
+  gpr_uint8 seen_regular_header;
   /* output stream id */
   gpr_uint32 stream_id;
   gpr_slice_buffer *output;
@@ -361,6 +363,15 @@ static grpc_mdelem *hpack_enc(grpc_chttp2_hpack_compressor *c,
   gpr_uint32 indices_key;
   int should_add_elem;
 
+  GPR_ASSERT (GPR_SLICE_LENGTH(elem->key->slice) > 0);
+  if (GPR_SLICE_START_PTR(elem->key->slice)[0] != ':') { /* regular header */
+    st->seen_regular_header = 1;
+  } else if (st->seen_regular_header != 0) { /* reserved header */
+    gpr_log(GPR_ERROR,
+            "Reserved header (colon-prefixed) happening after regular ones.");
+    abort();
+  }
+
   inc_filter(HASH_FRAGMENT_1(elem_hash), &c->filter_elems_sum, c->filter_elems);
 
   /* is this elem currently in the decoders table? */
@@ -566,6 +577,7 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
 
   st.cur_frame_type = NONE;
   st.last_was_header = 0;
+  st.seen_regular_header = 0;
   st.stream_id = stream_id;
   st.output = output;