Merge pull request #12895 from ctiller/timer

Have BDP estimator use timers

src/core/ext/transport/chttp2/transport/chttp2_transport.cc

@@ -152,10 +152,14 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx,
 static void end_all_the_calls(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
                               grpc_error *error);
 
+static void schedule_bdp_ping_locked(grpc_exec_ctx *exec_ctx,
+                                     grpc_chttp2_transport *t);
 static void start_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
                                   grpc_error *error);
 static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
                                    grpc_error *error);
+static void next_bdp_ping_timer_expired_locked(grpc_exec_ctx *exec_ctx,
+                                               void *tp, grpc_error *error);
 
 static void cancel_pings(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
                          grpc_error *error);
@@ -220,6 +224,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
 
   t->flow_control.bdp_estimator.Destroy();
 
+  GRPC_ERROR_UNREF(t->closed_with_error);
   gpr_free(t->ping_acks);
   gpr_free(t->peer_string);
   gpr_free(t);
@@ -305,6 +310,9 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
                     grpc_combiner_scheduler(t->combiner));
   GRPC_CLOSURE_INIT(&t->finish_bdp_ping_locked, finish_bdp_ping_locked, t,
                     grpc_combiner_scheduler(t->combiner));
+  GRPC_CLOSURE_INIT(&t->next_bdp_ping_timer_expired_locked,
+                    next_bdp_ping_timer_expired_locked, t,
+                    grpc_combiner_scheduler(t->combiner));
   GRPC_CLOSURE_INIT(&t->init_keepalive_ping_locked, init_keepalive_ping_locked,
                     t, grpc_combiner_scheduler(t->combiner));
   GRPC_CLOSURE_INIT(&t->start_keepalive_ping_locked,
@@ -564,6 +572,11 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
     t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DISABLED;
   }
 
+  if (t->flow_control.enable_bdp_probe) {
+    GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping");
+    schedule_bdp_ping_locked(exec_ctx, t);
+  }
+
   grpc_chttp2_act_on_flowctl_action(
       exec_ctx,
       grpc_chttp2_flowctl_get_action(exec_ctx, &t->flow_control, NULL), t,
@@ -597,7 +610,9 @@ static void destroy_transport(grpc_exec_ctx *exec_ctx, grpc_transport *gt) {
 static void close_transport_locked(grpc_exec_ctx *exec_ctx,
                                    grpc_chttp2_transport *t,
                                    grpc_error *error) {
-  if (!t->closed) {
+  end_all_the_calls(exec_ctx, t, GRPC_ERROR_REF(error));
+  cancel_pings(exec_ctx, t, GRPC_ERROR_REF(error));
+  if (t->closed_with_error == GRPC_ERROR_NONE) {
     if (!grpc_error_has_clear_grpc_status(error)) {
       error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS,
                                  GRPC_STATUS_UNAVAILABLE);
@@ -612,13 +627,16 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx,
           grpc_error_add_child(t->close_transport_on_writes_finished, error);
       return;
     }
-    t->closed = 1;
+    GPR_ASSERT(error != GRPC_ERROR_NONE);
+    t->closed_with_error = GRPC_ERROR_REF(error);
     connectivity_state_set(exec_ctx, t, GRPC_CHANNEL_SHUTDOWN,
                            GRPC_ERROR_REF(error), "close_transport");
-    grpc_endpoint_shutdown(exec_ctx, t->ep, GRPC_ERROR_REF(error));
     if (t->ping_state.is_delayed_ping_timer_set) {
       grpc_timer_cancel(exec_ctx, &t->ping_state.delayed_ping_timer);
     }
+    if (t->have_next_bdp_ping_timer) {
+      grpc_timer_cancel(exec_ctx, &t->next_bdp_ping_timer);
+    }
     switch (t->keepalive_state) {
       case GRPC_CHTTP2_KEEPALIVE_STATE_WAITING:
         grpc_timer_cancel(exec_ctx, &t->keepalive_ping_timer);
@@ -638,8 +656,8 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx,
     while (grpc_chttp2_list_pop_writable_stream(t, &s)) {
       GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "chttp2_writing:close");
     }
-    end_all_the_calls(exec_ctx, t, GRPC_ERROR_REF(error));
-    cancel_pings(exec_ctx, t, GRPC_ERROR_REF(error));
+    GPR_ASSERT(t->write_state == GRPC_CHTTP2_WRITE_STATE_IDLE);
+    grpc_endpoint_shutdown(exec_ctx, t->ep, GRPC_ERROR_REF(error));
   }
   GRPC_ERROR_UNREF(error);
 }
@@ -942,7 +960,8 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
 void grpc_chttp2_mark_stream_writable(grpc_exec_ctx *exec_ctx,
                                       grpc_chttp2_transport *t,
                                       grpc_chttp2_stream *s) {
-  if (!t->closed && grpc_chttp2_list_add_writable_stream(t, s)) {
+  if (t->closed_with_error == GRPC_ERROR_NONE &&
+      grpc_chttp2_list_add_writable_stream(t, s)) {
     GRPC_CHTTP2_STREAM_REF(s, "chttp2_writing:become");
   }
 }
@@ -995,7 +1014,7 @@ static void write_action_begin_locked(grpc_exec_ctx *exec_ctx, void *gt,
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
   GPR_ASSERT(t->write_state != GRPC_CHTTP2_WRITE_STATE_IDLE);
   grpc_chttp2_begin_write_result r;
-  if (t->closed) {
+  if (t->closed_with_error != GRPC_ERROR_NONE) {
     r.writing = false;
   } else {
     r = grpc_chttp2_begin_write(exec_ctx, t);
@@ -1458,7 +1477,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
       }
       if (!s->write_closed) {
         if (t->is_client) {
-          if (!t->closed) {
+          if (t->closed_with_error == GRPC_ERROR_NONE) {
             GPR_ASSERT(s->id == 0);
             grpc_chttp2_list_add_waiting_for_concurrency(t, s);
             maybe_start_some_streams(exec_ctx, t);
@@ -1466,7 +1485,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
             grpc_chttp2_cancel_stream(
                 exec_ctx, t, s,
                 grpc_error_set_int(
-                    GRPC_ERROR_CREATE_FROM_STATIC_STRING("Transport closed"),
+                    GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
+                        "Transport closed", &t->closed_with_error, 1),
                     GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE));
           }
         } else {
@@ -1688,6 +1708,7 @@ static void cancel_pings(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   /* callback remaining pings: they're not allowed to call into the transpot,
      and maybe they hold resources that need to be freed */
   grpc_chttp2_ping_queue *pq = &t->ping_queue;
+  GPR_ASSERT(error != GRPC_ERROR_NONE);
   for (size_t j = 0; j < GRPC_CHTTP2_PCL_COUNT; j++) {
     grpc_closure_list_fail_all(&pq->lists[j], GRPC_ERROR_REF(error));
     GRPC_CLOSURE_LIST_SCHED(exec_ctx, &pq->lists[j]);
@@ -1697,6 +1718,12 @@ static void cancel_pings(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
 
 static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
                              grpc_closure *on_initiate, grpc_closure *on_ack) {
+  if (t->closed_with_error != GRPC_ERROR_NONE) {
+    GRPC_CLOSURE_SCHED(exec_ctx, on_initiate,
+                       GRPC_ERROR_REF(t->closed_with_error));
+    GRPC_CLOSURE_SCHED(exec_ctx, on_ack, GRPC_ERROR_REF(t->closed_with_error));
+    return;
+  }
   grpc_chttp2_ping_queue *pq = &t->ping_queue;
   grpc_closure_list_append(&pq->lists[GRPC_CHTTP2_PCL_INITIATE], on_initiate,
                            GRPC_ERROR_NONE);
@@ -1755,7 +1782,9 @@ void grpc_chttp2_add_ping_strike(grpc_exec_ctx *exec_ctx,
                     GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_ENHANCE_YOUR_CALM));
     /*The transport will be closed after the write is done */
     close_transport_locked(
-        exec_ctx, t, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many pings"));
+        exec_ctx, t, grpc_error_set_int(
+                         GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many pings"),
+                         GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE));
   }
 }
 
@@ -2434,12 +2463,6 @@ void grpc_chttp2_act_on_flowctl_action(grpc_exec_ctx *exec_ctx,
                                  GRPC_CHTTP2_INITIATE_WRITE_SEND_SETTINGS);
     }
   }
-  if (action.need_ping) {
-    GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping");
-    t->flow_control.bdp_estimator->SchedulePing();
-    send_ping_locked(exec_ctx, t, &t->start_bdp_ping_locked,
-                     &t->finish_bdp_ping_locked);
-  }
 }
 
 static grpc_error *try_http_parsing(grpc_exec_ctx *exec_ctx,
@@ -2489,7 +2512,7 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
   }
   GPR_SWAP(grpc_error *, err, error);
   GRPC_ERROR_UNREF(err);
-  if (!t->closed) {
+  if (t->closed_with_error == GRPC_ERROR_NONE) {
     GPR_TIMER_BEGIN("reading_action.parse", 0);
     size_t i = 0;
     grpc_error *errors[3] = {GRPC_ERROR_REF(error), GRPC_ERROR_NONE,
@@ -2529,13 +2552,14 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
 
   GPR_TIMER_BEGIN("post_reading_action_locked", 0);
   bool keep_reading = false;
-  if (error == GRPC_ERROR_NONE && t->closed) {
-    error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Transport closed");
+  if (error == GRPC_ERROR_NONE && t->closed_with_error != GRPC_ERROR_NONE) {
+    error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
+        "Transport closed", &t->closed_with_error, 1);
   }
   if (error != GRPC_ERROR_NONE) {
     close_transport_locked(exec_ctx, t, GRPC_ERROR_REF(error));
     t->endpoint_reading = 0;
-  } else if (!t->closed) {
+  } else if (t->closed_with_error == GRPC_ERROR_NONE) {
     keep_reading = true;
     GRPC_CHTTP2_REF_TRANSPORT(t, "keep_reading");
   }
@@ -2560,11 +2584,21 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
   GPR_TIMER_END("reading_action_locked", 0);
 }
 
+// t is reffed prior to calling the first time, and once the callback chain
+// that kicks off finishes, it's unreffed
+static void schedule_bdp_ping_locked(grpc_exec_ctx *exec_ctx,
+                                     grpc_chttp2_transport *t) {
+  t->flow_control.bdp_estimator->SchedulePing();
+  send_ping_locked(exec_ctx, t, &t->start_bdp_ping_locked,
+                   &t->finish_bdp_ping_locked);
+}
+
 static void start_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
                                   grpc_error *error) {
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp;
   if (GRPC_TRACER_ON(grpc_http_trace)) {
-    gpr_log(GPR_DEBUG, "%s: Start BDP ping", t->peer_string);
+    gpr_log(GPR_DEBUG, "%s: Start BDP ping err=%s", t->peer_string,
+            grpc_error_string(error));
   }
   /* Reset the keepalive ping timer */
   if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_WAITING) {
@@ -2577,11 +2611,30 @@ static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
                                    grpc_error *error) {
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp;
   if (GRPC_TRACER_ON(grpc_http_trace)) {
-    gpr_log(GPR_DEBUG, "%s: Complete BDP ping", t->peer_string);
+    gpr_log(GPR_DEBUG, "%s: Complete BDP ping err=%s", t->peer_string,
+            grpc_error_string(error));
   }
-  t->flow_control.bdp_estimator->CompletePing(exec_ctx);
+  if (error != GRPC_ERROR_NONE) {
+    GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping");
+    return;
+  }
+  grpc_millis next_ping = t->flow_control.bdp_estimator->CompletePing(exec_ctx);
+  GPR_ASSERT(!t->have_next_bdp_ping_timer);
+  t->have_next_bdp_ping_timer = true;
+  grpc_timer_init(exec_ctx, &t->next_bdp_ping_timer, next_ping,
+                  &t->next_bdp_ping_timer_expired_locked);
+}
 
-  GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping");
+static void next_bdp_ping_timer_expired_locked(grpc_exec_ctx *exec_ctx,
+                                               void *tp, grpc_error *error) {
+  grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp;
+  GPR_ASSERT(t->have_next_bdp_ping_timer);
+  t->have_next_bdp_ping_timer = false;
+  if (error != GRPC_ERROR_NONE) {
+    GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping");
+    return;
+  }
+  schedule_bdp_ping_locked(exec_ctx, t);
 }
 
 void grpc_chttp2_config_default_keepalive_args(grpc_channel_args *args,
@@ -2646,7 +2699,7 @@ static void init_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg,
                                        grpc_error *error) {
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)arg;
   GPR_ASSERT(t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_WAITING);
-  if (t->destroying || t->closed) {
+  if (t->destroying || t->closed_with_error != GRPC_ERROR_NONE) {
     t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DYING;
   } else if (error == GRPC_ERROR_NONE) {
     if (t->keepalive_permit_without_calls ||
@@ -2704,8 +2757,11 @@ static void keepalive_watchdog_fired_locked(grpc_exec_ctx *exec_ctx, void *arg,
   if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_PINGING) {
     if (error == GRPC_ERROR_NONE) {
       t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DYING;
-      close_transport_locked(exec_ctx, t, GRPC_ERROR_CREATE_FROM_STATIC_STRING(
-                                              "keepalive watchdog timeout"));
+      close_transport_locked(
+          exec_ctx, t,
+          grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+                                 "keepalive watchdog timeout"),
+                             GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_INTERNAL));
     }
   } else {
     /* The watchdog timer should have been cancelled by

src/core/ext/transport/chttp2/transport/flow_control.cc

@@ -459,8 +459,6 @@ grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action(
     }
   }
   if (tfc->enable_bdp_probe) {
-    action.need_ping = tfc->bdp_estimator->NeedPing(exec_ctx);
-
     // get bdp estimate and update initial_window accordingly.
     int64_t estimate = -1;
     if (tfc->bdp_estimator->EstimateBdp(&estimate)) {

src/core/ext/transport/chttp2/transport/internal.h

@@ -298,7 +298,7 @@ struct grpc_chttp2_transport {
   /** is the transport destroying itself? */
   uint8_t destroying;
   /** has the upper layer closed the transport? */
-  uint8_t closed;
+  grpc_error *closed_with_error;
 
   /** is there a read request to the endpoint outstanding? */
   uint8_t endpoint_reading;
@@ -340,7 +340,7 @@ struct grpc_chttp2_transport {
   /** hpack encoding */
   grpc_chttp2_hpack_compressor hpack_compressor;
   /** is this a client? */
-  uint8_t is_client;
+  bool is_client;
 
   /** data to write next write */
   grpc_slice_buffer qbuf;
@@ -350,14 +350,14 @@ struct grpc_chttp2_transport {
   uint32_t write_buffer_size;
 
   /** have we seen a goaway */
-  uint8_t seen_goaway;
+  bool seen_goaway;
   /** have we sent a goaway */
   grpc_chttp2_sent_goaway_state sent_goaway_state;
 
   /** are the local settings dirty and need to be sent? */
-  uint8_t dirtied_local_settings;
+  bool dirtied_local_settings;
   /** have local settings been sent? */
-  uint8_t sent_local_settings;
+  bool sent_local_settings;
   /** bitmask of setting indexes to send out */
   uint32_t force_send_settings;
   /** settings values */
@@ -422,6 +422,7 @@ struct grpc_chttp2_transport {
   grpc_chttp2_write_cb *write_cb_pool;
 
   /* bdp estimator */
+  grpc_closure next_bdp_ping_timer_expired_locked;
   grpc_closure start_bdp_ping_locked;
   grpc_closure finish_bdp_ping_locked;
 
@@ -442,6 +443,10 @@ struct grpc_chttp2_transport {
   /** destructive cleanup closure */
   grpc_closure destructive_reclaimer_locked;
 
+  /* next bdp ping timer */
+  bool have_next_bdp_ping_timer;
+  grpc_timer next_bdp_ping_timer;
+
   /* keep-alive ping support */
   /** Closure to initialize a keepalive ping */
   grpc_closure init_keepalive_ping_locked;
@@ -749,7 +754,6 @@ typedef struct {
   grpc_chttp2_flowctl_urgency send_setting_update;
   uint32_t initial_window_size;
   uint32_t max_frame_size;
-  bool need_ping;
 } grpc_chttp2_flowctl_action;
 
 // Reads the flow control data and returns and actionable struct that will tell

src/core/ext/transport/chttp2/transport/writing.cc

@@ -245,7 +245,8 @@ class WriteContext {
   void UpdateStreamsNoLongerStalled() {
     grpc_chttp2_stream *s;
     while (grpc_chttp2_list_pop_stalled_by_transport(t_, &s)) {
-      if (!t_->closed && grpc_chttp2_list_add_writable_stream(t_, s)) {
+      if (t_->closed_with_error == GRPC_ERROR_NONE &&
+          grpc_chttp2_list_add_writable_stream(t_, s)) {
         if (!stream_ref_if_not_destroyed(&s->refcount->refs)) {
           grpc_chttp2_list_remove_writable_stream(t_, s);
         }

src/core/lib/transport/bdp_estimator.cc

@@ -33,13 +33,12 @@ BdpEstimator::BdpEstimator(const char *name)
       accumulator_(0),
       estimate_(65536),
       ping_start_time_(gpr_time_0(GPR_CLOCK_MONOTONIC)),
-      next_ping_scheduled_(0),
       inter_ping_delay_(100.0),  // start at 100ms
       stable_estimate_count_(0),
       bw_est_(0),
       name_(name) {}
 
-void BdpEstimator::CompletePing(grpc_exec_ctx *exec_ctx) {
+grpc_millis BdpEstimator::CompletePing(grpc_exec_ctx *exec_ctx) {
   gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
   gpr_timespec dt_ts = gpr_time_sub(now, ping_start_time_);
   double dt = (double)dt_ts.tv_sec + 1e-9 * (double)dt_ts.tv_nsec;
@@ -79,7 +78,7 @@ void BdpEstimator::CompletePing(grpc_exec_ctx *exec_ctx) {
   }
   ping_state_ = PingState::UNSCHEDULED;
   accumulator_ = 0;
-  next_ping_scheduled_ = grpc_exec_ctx_now(exec_ctx) + inter_ping_delay_;
+  return grpc_exec_ctx_now(exec_ctx) + inter_ping_delay_;
 }
 
 }  // namespace grpc_core

src/core/lib/transport/bdp_estimator.h

@@ -52,18 +52,6 @@ class BdpEstimator {
 
   void AddIncomingBytes(int64_t num_bytes) { accumulator_ += num_bytes; }
 
-  // Returns true if the user should schedule a ping
-  bool NeedPing(grpc_exec_ctx *exec_ctx) const {
-    switch (ping_state_) {
-      case PingState::UNSCHEDULED:
-        return grpc_exec_ctx_now(exec_ctx) >= next_ping_scheduled_;
-      case PingState::SCHEDULED:
-      case PingState::STARTED:
-        return false;
-    }
-    GPR_UNREACHABLE_CODE(return false);
-  }
-
   // Schedule a ping: call in response to receiving a true from
   // grpc_bdp_estimator_add_incoming_bytes once a ping has been scheduled by a
   // transport (but not necessarily started)
@@ -91,8 +79,8 @@ class BdpEstimator {
     ping_start_time_ = gpr_now(GPR_CLOCK_MONOTONIC);
   }
 
-  // Completes a previously started ping
-  void CompletePing(grpc_exec_ctx *exec_ctx);
+  // Completes a previously started ping, returns when to schedule the next one
+  grpc_millis CompletePing(grpc_exec_ctx *exec_ctx);
 
  private:
   enum class PingState { UNSCHEDULED, SCHEDULED, STARTED };
@@ -102,8 +90,6 @@ class BdpEstimator {
   int64_t estimate_;
   // when was the current ping started?
   gpr_timespec ping_start_time_;
-  // when should the next ping start?
-  grpc_millis next_ping_scheduled_;
   int inter_ping_delay_;
   int stable_estimate_count_;
   double bw_est_;

test/core/end2end/bad_server_response_test.c

@@ -62,8 +62,6 @@
 #define HTTP2_DETAIL_MSG(STATUS_CODE) \
   "Received http2 header with status: " #STATUS_CODE
 
-#define UNPARSEABLE_DETAIL_MSG "Failed parsing HTTP/2"
-
 #define HTTP1_DETAIL_MSG "Trying to connect an http1.x server"
 
 /* TODO(zyc) Check the content of incomming data instead of using this length */
@@ -208,8 +206,10 @@ static void start_rpc(int target_port, grpc_status_code expected_status,
   cq_verify(cqv);
 
   GPR_ASSERT(status == expected_status);
-  GPR_ASSERT(-1 != grpc_slice_slice(details, grpc_slice_from_static_string(
-                                                 expected_detail)));
+  if (expected_detail != NULL) {
+    GPR_ASSERT(-1 != grpc_slice_slice(details, grpc_slice_from_static_string(
+                                                   expected_detail)));
+  }
 
   grpc_metadata_array_destroy(&initial_metadata_recv);
   grpc_metadata_array_destroy(&trailing_metadata_recv);
@@ -330,8 +330,8 @@ int main(int argc, char **argv) {
            HTTP2_DETAIL_MSG(502));
 
   /* unparseable response */
-  run_test(UNPARSEABLE_RESP, sizeof(UNPARSEABLE_RESP) - 1,
-           GRPC_STATUS_UNAVAILABLE, UNPARSEABLE_DETAIL_MSG);
+  run_test(UNPARSEABLE_RESP, sizeof(UNPARSEABLE_RESP) - 1, GRPC_STATUS_UNKNOWN,
+           NULL);
 
   /* http1 response */
   run_test(HTTP1_RESP, sizeof(HTTP1_RESP) - 1, GRPC_STATUS_UNAVAILABLE,

test/core/end2end/tests/bad_ping.c

@@ -203,7 +203,6 @@ static void test_bad_ping(grpc_end2end_test_config config) {
   // The connection should be closed immediately after the misbehaved pings,
   // the in-progress RPC should fail.
   GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
-  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "Endpoint read failed"));
   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
   validate_host_override_string("foo.test.google.fr:1234", call_details.host,
                                 config);

test/core/end2end/tests/keepalive_timeout.c

@@ -193,7 +193,7 @@ static void test_keepalive_timeout(grpc_end2end_test_config config) {
 
   char *details_str = grpc_slice_to_c_string(details);
   char *method_str = grpc_slice_to_c_string(call_details.method);
-  GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
+  GPR_ASSERT(status == GRPC_STATUS_INTERNAL);
   GPR_ASSERT(0 == grpc_slice_str_cmp(details, "keepalive watchdog timeout"));
   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
   validate_host_override_string("foo.test.google.fr:1234", call_details.host,

test/core/end2end/tests/max_connection_age.c

@@ -203,8 +203,7 @@ static void test_max_age_forcibly_close(grpc_end2end_test_config config) {
 
   /* The connection should be closed immediately after the max age grace period,
      the in-progress RPC should fail. */
-  GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
-  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "Endpoint read failed"));
+  GPR_ASSERT(status == GRPC_STATUS_INTERNAL);
   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
   validate_host_override_string("foo.test.google.fr:1234", call_details.host,
                                 config);

test/core/end2end/tests/shutdown_finishes_calls.c

@@ -159,7 +159,9 @@ static void test_early_server_shutdown_finishes_inflight_calls(
 
   grpc_server_destroy(f.server);
 
-  GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
+  // new code should give INTERNAL, some older code will give UNAVAILABLE
+  GPR_ASSERT(status == GRPC_STATUS_INTERNAL ||
+             status == GRPC_STATUS_UNAVAILABLE);
   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
   validate_host_override_string("foo.test.google.fr:1234", call_details.host,
                                 config);

test/core/transport/bdp_estimator_test.cc

@@ -60,12 +60,10 @@ void AddSamples(BdpEstimator *estimator, int64_t *samples, size_t n) {
   estimator->AddIncomingBytes(1234567);
   inc_time();
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  EXPECT_TRUE(estimator->NeedPing(&exec_ctx));
   estimator->SchedulePing();
   estimator->StartPing();
   for (size_t i = 0; i < n; i++) {
     estimator->AddIncomingBytes(samples[i]);
-    EXPECT_FALSE(estimator->NeedPing(&exec_ctx));
   }
   gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
                                gpr_time_from_millis(1, GPR_TIMESPAN)));

tools/run_tests/run_tests.py

@@ -122,7 +122,7 @@ def max_parallel_tests_for_current_platform():
   # so far on windows.
   if jobset.platform_string() == 'windows':
     return 64
-  return 128
+  return 1024
 
 # SimpleConfig: just compile with CONFIG=config, and run the binary to test
 class Config(object):