C++ize BDP estimator, introduce ManualConstructor

6 files changed, 201 insertions, 154 deletions

diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc
index e4b19a2c4a..0ef06ae6e0 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc
@@ -218,6 +218,8 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
     t->write_cb_pool = next;
   }
 
+  t->flow_control.bdp_estimator.Destroy();
+
   gpr_free(t->ping_acks);
   gpr_free(t->peer_string);
   gpr_free(t);
@@ -315,7 +317,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
                     keepalive_watchdog_fired_locked, t,
                     grpc_combiner_scheduler(t->combiner));
 
-  grpc_bdp_estimator_init(&t->flow_control.bdp_estimator, t->peer_string);
+  t->flow_control.bdp_estimator.Init(t->peer_string);
 
   grpc_chttp2_goaway_parser_init(&t->goaway_parser);
   grpc_chttp2_hpack_parser_init(exec_ctx, &t->hpack_parser);
@@ -2434,7 +2436,7 @@ void grpc_chttp2_act_on_flowctl_action(grpc_exec_ctx *exec_ctx,
   }
   if (action.need_ping) {
     GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping");
-    grpc_bdp_estimator_schedule_ping(&t->flow_control.bdp_estimator);
+    t->flow_control.bdp_estimator->SchedulePing();
     send_ping_locked(exec_ctx, t, &t->start_bdp_ping_locked,
                      &t->finish_bdp_ping_locked);
   }
@@ -2493,8 +2495,7 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
     grpc_error *errors[3] = {GRPC_ERROR_REF(error), GRPC_ERROR_NONE,
                              GRPC_ERROR_NONE};
     for (; i < t->read_buffer.count && errors[1] == GRPC_ERROR_NONE; i++) {
-      grpc_bdp_estimator_add_incoming_bytes(
-          &t->flow_control.bdp_estimator,
+      t->flow_control.bdp_estimator->AddIncomingBytes(
           (int64_t)GRPC_SLICE_LENGTH(t->read_buffer.slices[i]));
       errors[1] =
           grpc_chttp2_perform_read(exec_ctx, t, t->read_buffer.slices[i]);
@@ -2569,7 +2570,7 @@ static void start_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
   if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_WAITING) {
     grpc_timer_cancel(exec_ctx, &t->keepalive_ping_timer);
   }
-  grpc_bdp_estimator_start_ping(&t->flow_control.bdp_estimator);
+  t->flow_control.bdp_estimator->StartPing();
 }
 
 static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
@@ -2578,7 +2579,7 @@ static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
   if (GRPC_TRACER_ON(grpc_http_trace)) {
     gpr_log(GPR_DEBUG, "%s: Complete BDP ping", t->peer_string);
   }
-  grpc_bdp_estimator_complete_ping(exec_ctx, &t->flow_control.bdp_estimator);
+  t->flow_control.bdp_estimator->CompletePing(exec_ctx);
 
   GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping");
 }
diff --git a/src/core/ext/transport/chttp2/transport/flow_control.cc b/src/core/ext/transport/chttp2/transport/flow_control.cc
index 2428e2526d..60c43d840a 100644
--- a/src/core/ext/transport/chttp2/transport/flow_control.cc
+++ b/src/core/ext/transport/chttp2/transport/flow_control.cc
@@ -459,12 +459,11 @@ grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action(
     }
   }
   if (tfc->enable_bdp_probe) {
-    action.need_ping =
-        grpc_bdp_estimator_need_ping(exec_ctx, &tfc->bdp_estimator);
+    action.need_ping = tfc->bdp_estimator->NeedPing(exec_ctx);
 
     // get bdp estimate and update initial_window accordingly.
     int64_t estimate = -1;
-    if (grpc_bdp_estimator_get_estimate(&tfc->bdp_estimator, &estimate)) {
+    if (tfc->bdp_estimator->EstimateBdp(&estimate)) {
       double target = 1 + log2((double)estimate);
 
       // target might change based on how much memory pressure we are under
@@ -491,7 +490,7 @@ grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action(
 
     // get bandwidth estimate and update max_frame accordingly.
     double bw_dbl = -1;
-    if (grpc_bdp_estimator_get_bw(&tfc->bdp_estimator, &bw_dbl)) {
+    if (tfc->bdp_estimator->EstimateBandwidth(&bw_dbl)) {
       // we target the max of BDP or bandwidth in microseconds.
       int32_t frame_size = (int32_t)GPR_CLAMP(
           GPR_MAX((int32_t)GPR_CLAMP(bw_dbl, 0, INT_MAX) / 1000,
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index b51854fcf8..05b677dd4b 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -37,6 +37,7 @@
 #include "src/core/lib/iomgr/combiner.h"
 #include "src/core/lib/iomgr/endpoint.h"
 #include "src/core/lib/iomgr/timer.h"
+#include "src/core/lib/support/manual_constructor.h"
 #include "src/core/lib/transport/bdp_estimator.h"
 #include "src/core/lib/transport/connectivity_state.h"
 #include "src/core/lib/transport/pid_controller.h"
@@ -268,7 +269,7 @@ typedef struct {
   bool enable_bdp_probe;
 
   /* bdp estimation */
-  grpc_bdp_estimator bdp_estimator;
+  grpc_core::ManualConstructor<grpc_core::BdpEstimator> bdp_estimator;
 
   /* pid controller */
   bool pid_controller_initialized;
diff --git a/src/core/lib/support/manual_constructor.h b/src/core/lib/support/manual_constructor.h
new file mode 100644
index 0000000000..be3dd4c1e3
--- /dev/null
+++ b/src/core/lib/support/manual_constructor.h
@@ -0,0 +1,71 @@
+/*
+ *
+ * Copyright 2016 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+// manually construct a region of memory with some type
+
+#include <stddef.h>
+#include <new>
+#include <type_traits>
+#include <utility>
+
+namespace grpc_core {
+
+template <typename Type>
+class ManualConstructor {
+ public:
+  // No constructor or destructor because one of the most useful uses of
+  // this class is as part of a union, and members of a union could not have
+  // constructors or destructors till C++11.  And, anyway, the whole point of
+  // this class is to bypass constructor and destructor.
+
+  Type* get() { return reinterpret_cast<Type*>(&space_); }
+  const Type* get() const { return reinterpret_cast<const Type*>(&space_); }
+
+  Type* operator->() { return get(); }
+  const Type* operator->() const { return get(); }
+
+  Type& operator*() { return *get(); }
+  const Type& operator*() const { return *get(); }
+
+  void Init() { new (&space_) Type; }
+
+  // Init() constructs the Type instance using the given arguments
+  // (which are forwarded to Type's constructor).
+  //
+  // Note that Init() with no arguments performs default-initialization,
+  // not zero-initialization (i.e it behaves the same as "new Type;", not
+  // "new Type();"), so it will leave non-class types uninitialized.
+  template <typename... Ts>
+  void Init(Ts&&... args) {
+    new (&space_) Type(std::forward<Ts>(args)...);
+  }
+
+  // Init() that is equivalent to copy and move construction.
+  // Enables usage like this:
+  //   ManualConstructor<std::vector<int>> v;
+  //   v.Init({1, 2, 3});
+  void Init(const Type& x) { new (&space_) Type(x); }
+  void Init(Type&& x) { new (&space_) Type(std::move(x)); }
+
+  void Destroy() { get()->~Type(); }
+
+ private:
+  typename std::aligned_storage<sizeof(Type), alignof(Type)>::type space_;
+};
+
+}  // namespace grpc_core
diff --git a/src/core/lib/transport/bdp_estimator.cc b/src/core/lib/transport/bdp_estimator.cc
index 6ed427ce5c..2a1c97c84e 100644
--- a/src/core/lib/transport/bdp_estimator.cc
+++ b/src/core/lib/transport/bdp_estimator.cc
@@ -21,117 +21,65 @@
 #include <inttypes.h>
 #include <stdlib.h>
 
-#include <grpc/support/log.h>
 #include <grpc/support/useful.h>
 
 grpc_tracer_flag grpc_bdp_estimator_trace =
     GRPC_TRACER_INITIALIZER(false, "bdp_estimator");
 
-void grpc_bdp_estimator_init(grpc_bdp_estimator *estimator, const char *name) {
-  estimator->estimate = 65536;
-  estimator->ping_state = GRPC_BDP_PING_UNSCHEDULED;
-  estimator->ping_start_time = gpr_time_0(GPR_CLOCK_MONOTONIC);
-  estimator->next_ping_scheduled = 0;
-  estimator->name = name;
-  estimator->bw_est = 0;
-  estimator->inter_ping_delay = 100.0;  // start at 100ms
-  estimator->stable_estimate_count = 0;
-}
-
-bool grpc_bdp_estimator_get_estimate(const grpc_bdp_estimator *estimator,
-                                     int64_t *estimate) {
-  *estimate = estimator->estimate;
-  return true;
-}
-
-bool grpc_bdp_estimator_get_bw(const grpc_bdp_estimator *estimator,
-                               double *bw) {
-  *bw = estimator->bw_est;
-  return true;
-}
-
-void grpc_bdp_estimator_add_incoming_bytes(grpc_bdp_estimator *estimator,
-                                           int64_t num_bytes) {
-  estimator->accumulator += num_bytes;
-}
-
-bool grpc_bdp_estimator_need_ping(grpc_exec_ctx *exec_ctx,
-                                  const grpc_bdp_estimator *estimator) {
-  switch (estimator->ping_state) {
-    case GRPC_BDP_PING_UNSCHEDULED:
-      return grpc_exec_ctx_now(exec_ctx) >= estimator->next_ping_scheduled;
-    case GRPC_BDP_PING_SCHEDULED:
-      return false;
-    case GRPC_BDP_PING_STARTED:
-      return false;
-  }
-  GPR_UNREACHABLE_CODE(return false);
-}
+namespace grpc_core {
 
-void grpc_bdp_estimator_schedule_ping(grpc_bdp_estimator *estimator) {
-  if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) {
-    gpr_log(GPR_DEBUG, "bdp[%s]:sched acc=%" PRId64 " est=%" PRId64,
-            estimator->name, estimator->accumulator, estimator->estimate);
-  }
-  GPR_ASSERT(estimator->ping_state == GRPC_BDP_PING_UNSCHEDULED);
-  estimator->ping_state = GRPC_BDP_PING_SCHEDULED;
-  estimator->accumulator = 0;
-}
+BdpEstimator::BdpEstimator(const char *name)
+    : ping_state_(PingState::UNSCHEDULED),
+      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 grpc_bdp_estimator_start_ping(grpc_bdp_estimator *estimator) {
-  if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) {
-    gpr_log(GPR_DEBUG, "bdp[%s]:start acc=%" PRId64 " est=%" PRId64,
-            estimator->name, estimator->accumulator, estimator->estimate);
-  }
-  GPR_ASSERT(estimator->ping_state == GRPC_BDP_PING_SCHEDULED);
-  estimator->ping_state = GRPC_BDP_PING_STARTED;
-  estimator->accumulator = 0;
-  estimator->ping_start_time = gpr_now(GPR_CLOCK_MONOTONIC);
-}
-
-void grpc_bdp_estimator_complete_ping(grpc_exec_ctx *exec_ctx,
-                                      grpc_bdp_estimator *estimator) {
+void BdpEstimator::CompletePing(grpc_exec_ctx *exec_ctx) {
   gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
-  gpr_timespec dt_ts = gpr_time_sub(now, estimator->ping_start_time);
+  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;
-  double bw = dt > 0 ? ((double)estimator->accumulator / dt) : 0;
-  int start_inter_ping_delay = estimator->inter_ping_delay;
+  double bw = dt > 0 ? ((double)accumulator_ / dt) : 0;
+  int start_inter_ping_delay = inter_ping_delay_;
   if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) {
     gpr_log(GPR_DEBUG, "bdp[%s]:complete acc=%" PRId64 " est=%" PRId64
                        " dt=%lf bw=%lfMbs bw_est=%lfMbs",
-            estimator->name, estimator->accumulator, estimator->estimate, dt,
-            bw / 125000.0, estimator->bw_est / 125000.0);
+            name_, accumulator_, estimate_, dt, bw / 125000.0,
+            bw_est_ / 125000.0);
   }
-  GPR_ASSERT(estimator->ping_state == GRPC_BDP_PING_STARTED);
-  if (estimator->accumulator > 2 * estimator->estimate / 3 &&
-      bw > estimator->bw_est) {
-    estimator->estimate =
-        GPR_MAX(estimator->accumulator, estimator->estimate * 2);
-    estimator->bw_est = bw;
+  GPR_ASSERT(ping_state_ == PingState::STARTED);
+  if (accumulator_ > 2 * estimate_ / 3 && bw > bw_est_) {
+    estimate_ = GPR_MAX(accumulator_, estimate_ * 2);
+    bw_est_ = bw;
     if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) {
-      gpr_log(GPR_DEBUG, "bdp[%s]: estimate increased to %" PRId64,
-              estimator->name, estimator->estimate);
+      gpr_log(GPR_DEBUG, "bdp[%s]: estimate increased to %" PRId64, name_,
+              estimate_);
     }
-    estimator->inter_ping_delay /= 2;  // if the ping estimate changes,
-                                       // exponentially get faster at probing
-  } else if (estimator->inter_ping_delay < 10000) {
-    estimator->stable_estimate_count++;
-    if (estimator->stable_estimate_count >= 2) {
-      estimator->inter_ping_delay +=
+    inter_ping_delay_ /= 2;  // if the ping estimate changes,
+                             // exponentially get faster at probing
+  } else if (inter_ping_delay_ < 10000) {
+    stable_estimate_count_++;
+    if (stable_estimate_count_ >= 2) {
+      inter_ping_delay_ +=
           100 +
           (int)(rand() * 100.0 / RAND_MAX);  // if the ping estimate is steady,
                                              // slowly ramp down the probe time
     }
   }
-  if (start_inter_ping_delay != estimator->inter_ping_delay) {
-    estimator->stable_estimate_count = 0;
+  if (start_inter_ping_delay != inter_ping_delay_) {
+    stable_estimate_count_ = 0;
     if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) {
-      gpr_log(GPR_DEBUG, "bdp[%s]:update_inter_time to %dms", estimator->name,
-              estimator->inter_ping_delay);
+      gpr_log(GPR_DEBUG, "bdp[%s]:update_inter_time to %dms", name_,
+              inter_ping_delay_);
     }
   }
-  estimator->ping_state = GRPC_BDP_PING_UNSCHEDULED;
-  estimator->accumulator = 0;
-  estimator->next_ping_scheduled =
-      grpc_exec_ctx_now(exec_ctx) + estimator->inter_ping_delay;
+  ping_state_ = PingState::UNSCHEDULED;
+  accumulator_ = 0;
+  next_ping_scheduled_ = grpc_exec_ctx_now(exec_ctx) + inter_ping_delay_;
 }
+
+}  // namespace grpc_core
diff --git a/src/core/lib/transport/bdp_estimator.h b/src/core/lib/transport/bdp_estimator.h
index 480d5237b8..6b1ebb5f67 100644
--- a/src/core/lib/transport/bdp_estimator.h
+++ b/src/core/lib/transport/bdp_estimator.h
@@ -19,67 +19,94 @@
 #ifndef GRPC_CORE_LIB_TRANSPORT_BDP_ESTIMATOR_H
 #define GRPC_CORE_LIB_TRANSPORT_BDP_ESTIMATOR_H
 
-#include <grpc/support/time.h>
 #include <stdbool.h>
 #include <stdint.h>
+
+#include <grpc/support/log.h>
+#include <grpc/support/time.h>
+
 #include "src/core/lib/debug/trace.h"
 #include "src/core/lib/iomgr/exec_ctx.h"
 
-#define GRPC_BDP_SAMPLES 16
-#define GRPC_BDP_MIN_SAMPLES_FOR_ESTIMATE 3
+extern grpc_tracer_flag grpc_bdp_estimator_trace;
 
-#ifdef __cplusplus
-extern "C" {
-#endif
+namespace grpc_core {
 
-extern grpc_tracer_flag grpc_bdp_estimator_trace;
+class BdpEstimator {
+ public:
+  explicit BdpEstimator(const char *name);
+  ~BdpEstimator();
+
+  // Returns true if a reasonable estimate could be obtained
+  bool EstimateBdp(int64_t *estimate_out) {
+    *estimate_out = estimate_;
+    return true;
+  }
+  bool EstimateBandwidth(double *bw_out) {
+    *bw_out = bw_est_;
+    return true;
+  }
 
-typedef enum {
-  GRPC_BDP_PING_UNSCHEDULED,
-  GRPC_BDP_PING_SCHEDULED,
-  GRPC_BDP_PING_STARTED
-} grpc_bdp_estimator_ping_state;
+  void AddIncomingBytes(int64_t num_bytes) { accumulator_ += num_bytes; }
 
-typedef struct grpc_bdp_estimator {
-  grpc_bdp_estimator_ping_state ping_state;
-  int64_t accumulator;
-  int64_t estimate;
+  // Returns true if the user should schedule a ping
+  bool NeedPing(grpc_exec_ctx *exec_ctx) {
+    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)
+  void SchedulePing() {
+    if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) {
+      gpr_log(GPR_DEBUG, "bdp[%s]:sched acc=%" PRId64 " est=%" PRId64, name_,
+              accumulator_, estimate_);
+    }
+    GPR_ASSERT(ping_state_ == PingState::UNSCHEDULED);
+    ping_state_ = PingState::SCHEDULED;
+    accumulator_ = 0;
+  }
+
+  // Start a ping: call after calling grpc_bdp_estimator_schedule_ping and
+  // once
+  // the ping is on the wire
+  void StartPing() {
+    if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) {
+      gpr_log(GPR_DEBUG, "bdp[%s]:start acc=%" PRId64 " est=%" PRId64, name_,
+              accumulator_, estimate_);
+    }
+    GPR_ASSERT(ping_state_ == PingState::SCHEDULED);
+    ping_state_ = PingState::STARTED;
+    accumulator_ = 0;
+    ping_start_time_ = gpr_now(GPR_CLOCK_MONOTONIC);
+  }
+
+  // Completes a previously started ping
+  void CompletePing(grpc_exec_ctx *exec_ctx);
+
+ private:
+  enum class PingState { UNSCHEDULED, SCHEDULED, STARTED };
+
+  PingState ping_state_;
+  int64_t accumulator_;
+  int64_t estimate_;
   // when was the current ping started?
-  gpr_timespec ping_start_time;
+  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;
-  const char *name;
-} grpc_bdp_estimator;
-
-void grpc_bdp_estimator_init(grpc_bdp_estimator *estimator, const char *name);
-
-// Returns true if a reasonable estimate could be obtained
-bool grpc_bdp_estimator_get_estimate(const grpc_bdp_estimator *estimator,
-                                     int64_t *estimate);
-// Tracks new bytes read.
-bool grpc_bdp_estimator_get_bw(const grpc_bdp_estimator *estimator, double *bw);
-// Returns true if the user should schedule a ping
-void grpc_bdp_estimator_add_incoming_bytes(grpc_bdp_estimator *estimator,
-                                           int64_t num_bytes);
-// Returns true if the user should schedule a ping
-bool grpc_bdp_estimator_need_ping(grpc_exec_ctx *exec_ctx,
-                                  const grpc_bdp_estimator *estimator);
-// 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)
-void grpc_bdp_estimator_schedule_ping(grpc_bdp_estimator *estimator);
-// Start a ping: call after calling grpc_bdp_estimator_schedule_ping and once
-// the ping is on the wire
-void grpc_bdp_estimator_start_ping(grpc_bdp_estimator *estimator);
-// Completes a previously started ping
-void grpc_bdp_estimator_complete_ping(grpc_exec_ctx *exec_ctx,
-                                      grpc_bdp_estimator *estimator);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /* GRPC_CORE_LIB_TRANSPORT_BDP_ESTIMATOR_H */
\ No newline at end of file
+  grpc_millis next_ping_scheduled_;
+  int inter_ping_delay_;
+  int stable_estimate_count_;
+  double bw_est_;
+  const char *name_;
+};
+
+}  // namespace grpc_core
+
+#endif /* GRPC_CORE_LIB_TRANSPORT_BDP_ESTIMATOR_H */