mirror of
https://github.com/facebook/rocksdb.git
synced 2026-10-23 08:09:02 +03:00
0561153b73
Summary: Motivation A WriteBufferManager can be shared by many RocksDB instances and column families. Under the existing policy, the DB whose writer first observes memory pressure flushes its own oldest mutable memtable. That DB might own only a small fraction of the reclaimable memory, so workloads with many DBs or CFs can generate many small flushes while the largest memory consumer remains untouched. Scanning every registered DB while holding process-wide or DB mutexes would improve selection accuracy but would serialize unrelated write paths and is not viable for deployments with thousands of DBs. Previous behavior kFlushOldest was the only behavior: the write-group leader that noticed WBM pressure synchronously selected the oldest eligible mutable memtable in its own DB. The WBM had no per-DB view of reclaimable memory and could not direct reclamation to another DB sharing the same budget. New behavior This change adds configurable WriteBufferFlushPolicy modes: - kFlushOldest preserves the existing behavior and remains the default. - kFlushLargest selects the largest eligible flushable column family in the writing DB. - kFlushLargestAcrossDBs selects the DB advertising the most reclaimable memory across all DBs sharing the WBM, then flushes its largest eligible CF. Atomic-flush DBs advertise and flush the aggregate eligible memory across their CFs. High-level implementation Each writable DB registers a lightweight FlushInitiator with the shared WBM. Per-DB accounting publishes two approximate values using atomics: total mutable memory and the largest flushable CF contribution, including eligible immutable memtables waiting for flush. Allocator reporting is batched, so the allocation hot path avoids registry scans and DB mutexes. A dedicated WBM coordinator thread maintains an approximate ranked snapshot of registered DBs. Refreshes are driven by accounting changes, threshold crossings, and periodic pressure checks. Ranking and O(number of DBs) work happen on that coordinator thread; writers only update counters and signal the coordinator. Registration lifetime is protected independently so DB close can safely drain an in-flight handoff. Threading and flush handoff The coordinator reserves and pins one ranked candidate at a time, without holding a registry lock while invoking the DB. The target DB uses TryLock on its DB mutex and rejects the handoff if it is closing, paused, stopped, stalled, already handling a WBM job, or no longer has flushable memory. Rejected candidates do not block consideration of the next DB. An accepted handoff queues one LOW-priority WBM job. The LOW worker joins the target DB write thread and revalidates eligibility before switching the memtable. The resulting ordinary flush is then scheduled through RocksDB's existing flush machinery on the HIGH-priority flush pool. Keeping the write-thread join off the requesting writer avoids making one DB's write path wait on another DB's mutex or write queue. Only one manager-wide remote handoff is active at a time. Completion or cancellation clears the handoff and requests a rerank. If memory reaches the hard limit while remote coordination is delayed, a rate-limited local fallback lets a writer seal a local memtable to preserve forward progress. Write stalls, shutdown, WaitForCompact, policy changes, and DB deregistration explicitly drain or reject outstanding handoffs. Compatibility The legacy constructor and kFlushOldest behavior remain unchanged unless a new policy is explicitly selected. There are no on-disk format changes. Pull Request resolved: https://github.com/facebook/rocksdb/pull/15047 Reviewed By: xingbowang Differential Revision: D112396106 fbshipit-source-id: 3049adc7034dd4e00d74d7d86e1d19004292d299
129 lines
4.7 KiB
C++
129 lines
4.7 KiB
C++
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
|
|
// This source code is licensed under both the GPLv2 (found in the
|
|
// COPYING file in the root directory) and Apache 2.0 License
|
|
// (found in the LICENSE.Apache file in the root directory).
|
|
//
|
|
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
|
|
// Use of this source code is governed by a BSD-style license that can be
|
|
// found in the LICENSE file. See the AUTHORS file for names of contributors.
|
|
|
|
#include <cassert>
|
|
|
|
#include "memory/allocator.h"
|
|
#include "memory/arena.h"
|
|
#include "memtable/flush_initiator.h"
|
|
#include "rocksdb/write_buffer_manager.h"
|
|
|
|
namespace ROCKSDB_NAMESPACE {
|
|
|
|
AllocTracker::AllocTracker(WriteBufferManager* write_buffer_manager,
|
|
FlushInitiator* flush_initiator)
|
|
: write_buffer_manager_(write_buffer_manager),
|
|
flush_initiator_(flush_initiator),
|
|
bytes_allocated_(0),
|
|
bytes_reported_(0),
|
|
flush_initiator_active_(false),
|
|
done_allocating_(false),
|
|
freed_(false) {}
|
|
|
|
AllocTracker::~AllocTracker() { FreeMem(); }
|
|
|
|
void AllocTracker::Allocate(size_t bytes) {
|
|
assert(write_buffer_manager_ != nullptr);
|
|
if (write_buffer_manager_->enabled() ||
|
|
write_buffer_manager_->cost_to_cache()) {
|
|
const size_t memtable_mem =
|
|
bytes_allocated_.fetch_add(bytes, std::memory_order_relaxed) + bytes;
|
|
if (flush_initiator_ != nullptr &&
|
|
flush_initiator_active_.load(std::memory_order_acquire) &&
|
|
write_buffer_manager_->ShouldTrackFlushInitiator()) {
|
|
const size_t bytes_reported =
|
|
bytes_reported_.load(std::memory_order_relaxed);
|
|
if (memtable_mem > bytes_reported &&
|
|
memtable_mem - bytes_reported >=
|
|
write_buffer_manager_->GetFlushInitiatorReportBytes()) {
|
|
ReportAllocations(memtable_mem);
|
|
}
|
|
}
|
|
// When a batch is due, publish it before ReserveMem() wakes the WBM's
|
|
// candidate-refresh thread on a threshold crossing.
|
|
write_buffer_manager_->ReserveMem(bytes);
|
|
}
|
|
}
|
|
|
|
void AllocTracker::ActivateFlushInitiator() {
|
|
if (write_buffer_manager_ == nullptr || flush_initiator_ == nullptr ||
|
|
done_allocating_ || !write_buffer_manager_->ShouldTrackFlushInitiator()) {
|
|
return;
|
|
}
|
|
flush_initiator_active_.store(true, std::memory_order_release);
|
|
if (write_buffer_manager_->ShouldTrackFlushInitiator()) {
|
|
ReportAllocations(bytes_allocated_.load(std::memory_order_relaxed));
|
|
flush_initiator_->InvalidateLargestFlushableCFMem();
|
|
write_buffer_manager_->NotifyFlushInitiatorChanged();
|
|
}
|
|
}
|
|
|
|
void AllocTracker::DeactivateFlushInitiator() {
|
|
if (write_buffer_manager_ != nullptr && flush_initiator_ != nullptr &&
|
|
flush_initiator_active_.exchange(false, std::memory_order_acq_rel)) {
|
|
if (write_buffer_manager_->ShouldTrackFlushInitiator()) {
|
|
ReportAllocations(bytes_allocated_.load(std::memory_order_relaxed));
|
|
}
|
|
const size_t bytes_reported =
|
|
bytes_reported_.exchange(0, std::memory_order_relaxed);
|
|
flush_initiator_->ScheduleFreeMem(bytes_reported);
|
|
}
|
|
}
|
|
|
|
void AllocTracker::RefreshFlushInitiator() {
|
|
if (write_buffer_manager_ != nullptr && flush_initiator_ != nullptr &&
|
|
!done_allocating_ && write_buffer_manager_->ShouldTrackFlushInitiator()) {
|
|
flush_initiator_active_.store(true, std::memory_order_release);
|
|
ReportAllocations(bytes_allocated_.load(std::memory_order_relaxed));
|
|
}
|
|
}
|
|
|
|
void AllocTracker::ReportAllocations(size_t bytes_allocated) {
|
|
size_t bytes_reported = bytes_reported_.load(std::memory_order_relaxed);
|
|
while (bytes_reported < bytes_allocated) {
|
|
if (bytes_reported_.compare_exchange_weak(bytes_reported, bytes_allocated,
|
|
std::memory_order_relaxed)) {
|
|
flush_initiator_->ReserveMem(bytes_allocated - bytes_reported,
|
|
bytes_allocated);
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
|
|
void AllocTracker::DoneAllocating() {
|
|
if (write_buffer_manager_ != nullptr && !done_allocating_) {
|
|
if (write_buffer_manager_->enabled() ||
|
|
write_buffer_manager_->cost_to_cache()) {
|
|
const size_t mem = bytes_allocated_.load(std::memory_order_relaxed);
|
|
write_buffer_manager_->ScheduleFreeMem(mem);
|
|
} else {
|
|
assert(bytes_allocated_.load(std::memory_order_relaxed) == 0);
|
|
}
|
|
DeactivateFlushInitiator();
|
|
done_allocating_ = true;
|
|
}
|
|
}
|
|
|
|
void AllocTracker::FreeMem() {
|
|
if (!done_allocating_) {
|
|
DoneAllocating();
|
|
}
|
|
if (write_buffer_manager_ != nullptr && !freed_) {
|
|
if (write_buffer_manager_->enabled() ||
|
|
write_buffer_manager_->cost_to_cache()) {
|
|
write_buffer_manager_->FreeMem(
|
|
bytes_allocated_.load(std::memory_order_relaxed));
|
|
} else {
|
|
assert(bytes_allocated_.load(std::memory_order_relaxed) == 0);
|
|
}
|
|
freed_ = true;
|
|
}
|
|
}
|
|
} // namespace ROCKSDB_NAMESPACE
|