diff options
Diffstat (limited to 'src/core/lib')
-rw-r--r-- | src/core/lib/support/manual_constructor.h | 71 | ||||
-rw-r--r-- | src/core/lib/transport/bdp_estimator.cc | 128 | ||||
-rw-r--r-- | src/core/lib/transport/bdp_estimator.h | 133 |
3 files changed, 189 insertions, 143 deletions
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 */ |