mirror of https://github.com/grpc/grpc.git
[resource_quota] Periodic update tracker (#29853)
* [resource_quota] Periodic update tracker Periodic update type that tries to do updates at some measurable timescale without polling the current time every tick, instead trying to approximate how many ticks will be required to fill the desired duration. The intent here is that we use this in RQ to donate memory back periodically whilst keeping timer checks off the fast path. * fix * Automated change: Fix sanity tests * review feedback * fix * fix * review feedback * Automated change: Fix sanity tests Co-authored-by: ctiller <ctiller@users.noreply.github.com>pull/29950/head
Craig Tiller
3 years ago
committed by
GitHub
parent
dd07751c03
commit
720f9d65c9
8 changed files with 407 additions and 0 deletions
@ -0,0 +1,72 @@ |
||||
// Copyright 2022 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.
|
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/resource_quota/periodic_update.h" |
||||
|
||||
#include <atomic> |
||||
|
||||
#include "src/core/lib/gpr/useful.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
bool PeriodicUpdate::MaybeEndPeriod() { |
||||
// updates_remaining_ just reached 0 and the thread calling this function was
|
||||
// the decrementer that got us there.
|
||||
// We can now safely mutate any non-atomic mutable variables (we've got a
|
||||
// guarantee that no other thread will), and by the time this function returns
|
||||
// we must store a postive number into updates_remaining_.
|
||||
auto now = ExecCtx::Get()->Now(); |
||||
Duration time_so_far = now - period_start_; |
||||
if (time_so_far < period_) { |
||||
// At most double the number of updates remaining until the next period.
|
||||
// At least try to estimate when we'll reach it.
|
||||
int64_t better_guess; |
||||
if (time_so_far.millis() == 0) { |
||||
better_guess = expected_updates_per_period_ * 2; |
||||
} else { |
||||
// Determine a scaling factor that would have gotten us to the next
|
||||
// period, but clamp between 1.01 (at least 1% increase in guesses)
|
||||
// and 2.0 (at most doubling) - to avoid running completely out of
|
||||
// control.
|
||||
const double scale = |
||||
Clamp(period_.seconds() / time_so_far.seconds(), 1.01, 2.0); |
||||
better_guess = expected_updates_per_period_ * scale; |
||||
if (better_guess <= expected_updates_per_period_) { |
||||
better_guess = expected_updates_per_period_ + 1; |
||||
} |
||||
} |
||||
// Store the remainder left. Note that updates_remaining_ may have been
|
||||
// decremented by another thread whilst we performed the above calculations:
|
||||
// we simply discard those decrements.
|
||||
updates_remaining_.store(better_guess - expected_updates_per_period_, |
||||
std::memory_order_release); |
||||
// Not quite done, return false, try for longer.
|
||||
return false; |
||||
} |
||||
// Finished period, start a new one and return true.
|
||||
// We try to predict how many update periods we'd need to cover the full time
|
||||
// span, and we increase that by 1% to attempt to tend to not enter the above
|
||||
// stanza.
|
||||
expected_updates_per_period_ = |
||||
period_.seconds() * expected_updates_per_period_ / time_so_far.seconds(); |
||||
if (expected_updates_per_period_ < 1) expected_updates_per_period_ = 1; |
||||
period_start_ = now; |
||||
updates_remaining_.store(expected_updates_per_period_, |
||||
std::memory_order_release); |
||||
return true; |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
@ -0,0 +1,70 @@ |
||||
// Copyright 2022 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.
|
||||
|
||||
#ifndef GRPC_CORE_LIB_RESOURCE_QUOTA_PERIODIC_UPDATE_H |
||||
#define GRPC_CORE_LIB_RESOURCE_QUOTA_PERIODIC_UPDATE_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <inttypes.h> |
||||
|
||||
#include <atomic> |
||||
|
||||
#include "src/core/lib/gprpp/time.h" |
||||
#include "src/core/lib/iomgr/exec_ctx.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
// Lightweight timer-like mechanism for periodic updates.
|
||||
// Fast path only decrements an atomic int64.
|
||||
// Slow path runs corrections and estimates how many ticks are required to hit
|
||||
// the target period.
|
||||
// This is super inaccurate of course, but for places where we can't run timers,
|
||||
// or places where continuous registration/unregistration would cause problems
|
||||
// it can be quite useful.
|
||||
class PeriodicUpdate { |
||||
public: |
||||
explicit PeriodicUpdate(Duration period) : period_(period) {} |
||||
|
||||
// Tick the update, return true if we think the period expired.
|
||||
GRPC_MUST_USE_RESULT bool Tick() { |
||||
// Atomically decrement the remaining ticks counter.
|
||||
// If we hit 0 our estimate of period length has expired.
|
||||
// See the comment next to the data members for a description of thread
|
||||
// safety.
|
||||
if (updates_remaining_.fetch_sub(1, std::memory_order_acquire) == 1) { |
||||
return MaybeEndPeriod(); |
||||
} |
||||
return false; |
||||
} |
||||
|
||||
private: |
||||
GRPC_MUST_USE_RESULT bool MaybeEndPeriod(); |
||||
|
||||
// Thread safety:
|
||||
// When updates_remaining_ reaches 0 the thread that decremented becomes
|
||||
// responsible for updating any mutable variables and then setting
|
||||
// updates_remaining_ to a value greater than zero.
|
||||
// Whilst in this state other threads *may* decrement updates_remaining_, but
|
||||
// this is fine because they'll observe an ignorable negative value.
|
||||
|
||||
const Duration period_; |
||||
Timestamp period_start_ = ExecCtx::Get()->Now(); |
||||
int64_t expected_updates_per_period_ = 1; |
||||
std::atomic<int64_t> updates_remaining_{1}; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_CORE_LIB_RESOURCE_QUOTA_PERIODIC_UPDATE_H
|
||||
@ -0,0 +1,111 @@ |
||||
// Copyright 2021 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.
|
||||
|
||||
#include "src/core/lib/resource_quota/periodic_update.h" |
||||
|
||||
#include <thread> |
||||
|
||||
#include <gtest/gtest.h> |
||||
|
||||
#include "absl/synchronization/notification.h" |
||||
|
||||
#include "src/core/lib/iomgr/exec_ctx.h" |
||||
|
||||
namespace grpc_core { |
||||
namespace testing { |
||||
|
||||
TEST(PeriodicUpdateTest, SimpleTest) { |
||||
std::unique_ptr<PeriodicUpdate> upd; |
||||
Timestamp start; |
||||
// Create a periodic update that updates every second.
|
||||
{ |
||||
ExecCtx exec_ctx; |
||||
upd = absl::make_unique<PeriodicUpdate>(Duration::Seconds(1)); |
||||
start = exec_ctx.Now(); |
||||
} |
||||
// Wait until the first period has elapsed.
|
||||
while (true) { |
||||
ExecCtx exec_ctx; |
||||
if (upd->Tick()) break; |
||||
} |
||||
// Ensure that took at least 1 second.
|
||||
{ |
||||
ExecCtx exec_ctx; |
||||
EXPECT_GE(exec_ctx.Now() - start, Duration::Seconds(1)); |
||||
start = exec_ctx.Now(); |
||||
} |
||||
// Do ten more update cycles
|
||||
for (int i = 0; i < 10; i++) { |
||||
while (true) { |
||||
ExecCtx exec_ctx; |
||||
if (upd->Tick()) break; |
||||
} |
||||
// Ensure the time taken was between 1 and 1.5 seconds - we make a little
|
||||
// allowance for the presumed inaccuracy of this type.
|
||||
{ |
||||
ExecCtx exec_ctx; |
||||
EXPECT_GE(exec_ctx.Now() - start, Duration::Seconds(1)); |
||||
EXPECT_LE(exec_ctx.Now() - start, Duration::Milliseconds(1500)); |
||||
start = exec_ctx.Now(); |
||||
} |
||||
} |
||||
} |
||||
|
||||
TEST(PeriodicUpdate, ThreadTest) { |
||||
std::unique_ptr<PeriodicUpdate> upd; |
||||
std::atomic<int> count(0); |
||||
Timestamp start; |
||||
// Create a periodic update that updates every second.
|
||||
{ |
||||
ExecCtx exec_ctx; |
||||
upd = absl::make_unique<PeriodicUpdate>(Duration::Seconds(1)); |
||||
start = exec_ctx.Now(); |
||||
} |
||||
// Run ten threads all updating the counter continuously, for a total of ten
|
||||
// update cycles.
|
||||
// This allows TSAN to catch threading issues.
|
||||
std::vector<std::thread> threads; |
||||
for (size_t i = 0; i < 10; i++) { |
||||
threads.push_back(std::thread([&]() { |
||||
while (count.load() < 10) { |
||||
ExecCtx exec_ctx; |
||||
if (upd->Tick()) count.fetch_add(1); |
||||
} |
||||
})); |
||||
} |
||||
|
||||
// Finish all threads.
|
||||
for (auto& th : threads) { |
||||
th.join(); |
||||
} |
||||
// Ensure our ten cycles took at least 10 seconds, and no more than 15.
|
||||
{ |
||||
ExecCtx exec_ctx; |
||||
EXPECT_GE(exec_ctx.Now() - start, Duration::Seconds(10)); |
||||
EXPECT_LE(exec_ctx.Now() - start, Duration::Seconds(15)); |
||||
} |
||||
} |
||||
|
||||
} // namespace testing
|
||||
} // namespace grpc_core
|
||||
|
||||
// Hook needed to run ExecCtx outside of iomgr.
|
||||
void grpc_set_default_iomgr_platform() {} |
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
gpr_log_verbosity_init(); |
||||
gpr_time_init(); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue