(LedgerStore)(Refactor) Move metric reporting functions to a dedicate mod (#24060)
Previously, the metric reporting functions are implemented under LedgerColumnMetric. However, there're operations like write batch which is issued by the function inside Rocks. This PR moves reporting functions to its own dedicate mod so that both LedgerColumn and Rocks can report column perf metrics.
This commit is contained in:
parent
302142bb25
commit
24cc6c33de
|
@ -5,16 +5,14 @@ use {
|
|||
byteorder::{BigEndian, ByteOrder},
|
||||
log::*,
|
||||
prost::Message,
|
||||
rand::{thread_rng, Rng},
|
||||
rocksdb::{
|
||||
self,
|
||||
compaction_filter::CompactionFilter,
|
||||
compaction_filter_factory::{CompactionFilterContext, CompactionFilterFactory},
|
||||
perf::{set_perf_stats, PerfMetric, PerfStatsLevel},
|
||||
properties as RocksProperties, ColumnFamily, ColumnFamilyDescriptor, CompactionDecision,
|
||||
DBCompactionStyle, DBCompressionType as RocksCompressionType, DBIterator, DBRawIterator,
|
||||
DBRecoveryMode, FifoCompactOptions, IteratorMode as RocksIteratorMode, Options,
|
||||
PerfContext, WriteBatch as RWriteBatch, DB,
|
||||
WriteBatch as RWriteBatch, DB,
|
||||
},
|
||||
serde::{de::DeserializeOwned, Serialize},
|
||||
solana_metrics::datapoint_info,
|
||||
|
@ -26,7 +24,6 @@ use {
|
|||
},
|
||||
solana_storage_proto::convert::generated,
|
||||
std::{
|
||||
cell::RefCell,
|
||||
collections::{HashMap, HashSet},
|
||||
ffi::{CStr, CString},
|
||||
fs,
|
||||
|
@ -60,9 +57,6 @@ const DEFAULT_FIFO_COMPACTION_DATA_CF_SIZE: u64 = 125 * 1024 * 1024 * 1024;
|
|||
// used by coding shreds.
|
||||
const DEFAULT_FIFO_COMPACTION_CODING_CF_SIZE: u64 = 100 * 1024 * 1024 * 1024;
|
||||
|
||||
// Thread local instance of RocksDB's PerfContext.
|
||||
thread_local! {static PER_THREAD_ROCKS_PERF_CONTEXT: RefCell<PerfContext> = RefCell::new(PerfContext::default());}
|
||||
|
||||
// Column family for metadata about a leader slot
|
||||
const META_CF: &str = "meta";
|
||||
// Column family for slots that have been marked as dead
|
||||
|
@ -1934,6 +1928,15 @@ where
|
|||
}
|
||||
}
|
||||
|
||||
mod rocks_metrics_utils {
|
||||
use {
|
||||
rand::{thread_rng, Rng},
|
||||
rocksdb::{
|
||||
perf::{set_perf_stats, PerfMetric, PerfStatsLevel},
|
||||
PerfContext,
|
||||
},
|
||||
std::cell::RefCell,
|
||||
};
|
||||
const METRIC_SAMPLES_1K: i32 = 1000;
|
||||
// The default number of rocksdb perf samples in 1K
|
||||
const ROCKSDB_PERF_CONTEXT_SAMPLES_IN_1K_DEFAULT: i32 = 10;
|
||||
|
@ -1948,41 +1951,14 @@ std::env::var("SOLANA_METRICS_ROCKSDB_PERF_SAMPLES_IN_1K")
|
|||
|
||||
}
|
||||
|
||||
impl<C> LedgerColumn<C>
|
||||
where
|
||||
C: TypedColumn + ColumnName + ColumnMetrics,
|
||||
{
|
||||
pub fn get(&self, key: C::Index) -> Result<Option<C::Type>> {
|
||||
let mut result = Ok(None);
|
||||
let is_perf_context_enabled = Self::maybe_collect_perf_context();
|
||||
if let Some(serialized_value) = self.backend.get_cf(self.handle(), &C::key(key))? {
|
||||
let value = deserialize(&serialized_value)?;
|
||||
|
||||
result = Ok(Some(value))
|
||||
}
|
||||
|
||||
if is_perf_context_enabled {
|
||||
self.report_read_perf_context(C::rocksdb_get_perf_metric_header(&self.column_options));
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
pub fn put(&self, key: C::Index, value: &C::Type) -> Result<()> {
|
||||
let serialized_value = serialize(value)?;
|
||||
|
||||
self.backend
|
||||
.put_cf(self.handle(), &C::key(key), &serialized_value)
|
||||
}
|
||||
|
||||
pub fn delete(&self, key: C::Index) -> Result<()> {
|
||||
self.backend.delete_cf(self.handle(), &C::key(key))
|
||||
}
|
||||
// Thread local instance of RocksDB's PerfContext.
|
||||
thread_local! {static PER_THREAD_ROCKS_PERF_CONTEXT: RefCell<PerfContext> = RefCell::new(PerfContext::default());}
|
||||
|
||||
/// The function enables RocksDB's PerfContext in N out of 1000
|
||||
/// where N is ROCKSDB_PERF_CONTEXT_SAMPLES_IN_1K.
|
||||
///
|
||||
/// Returns true if the PerfContext is enabled.
|
||||
fn maybe_collect_perf_context() -> bool {
|
||||
pub fn maybe_collect_perf_context() -> bool {
|
||||
if thread_rng().gen_range(0, METRIC_SAMPLES_1K) > *ROCKSDB_PERF_CONTEXT_SAMPLES_IN_1K {
|
||||
return false;
|
||||
}
|
||||
|
@ -1995,7 +1971,7 @@ where
|
|||
|
||||
/// Reports the collected PerfContext and disables the PerfContext after
|
||||
/// reporting.
|
||||
fn report_read_perf_context(&self, metric_header: &'static str) {
|
||||
pub fn report_read_perf_context(metric_header: &'static str) {
|
||||
PER_THREAD_ROCKS_PERF_CONTEXT.with(|perf_context_cell| {
|
||||
set_perf_stats(PerfStatsLevel::Disable);
|
||||
let perf_context = perf_context_cell.borrow();
|
||||
|
@ -2174,6 +2150,40 @@ where
|
|||
});
|
||||
}
|
||||
}
|
||||
use crate::blockstore_db::rocks_metrics_utils::{
|
||||
maybe_collect_perf_context, report_read_perf_context,
|
||||
};
|
||||
|
||||
impl<C> LedgerColumn<C>
|
||||
where
|
||||
C: TypedColumn + ColumnName + ColumnMetrics,
|
||||
{
|
||||
pub fn get(&self, key: C::Index) -> Result<Option<C::Type>> {
|
||||
let mut result = Ok(None);
|
||||
let is_perf_context_enabled = maybe_collect_perf_context();
|
||||
if let Some(serialized_value) = self.backend.get_cf(self.handle(), &C::key(key))? {
|
||||
let value = deserialize(&serialized_value)?;
|
||||
|
||||
result = Ok(Some(value))
|
||||
}
|
||||
|
||||
if is_perf_context_enabled {
|
||||
report_read_perf_context(C::rocksdb_get_perf_metric_header(&self.column_options));
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
pub fn put(&self, key: C::Index, value: &C::Type) -> Result<()> {
|
||||
let serialized_value = serialize(value)?;
|
||||
|
||||
self.backend
|
||||
.put_cf(self.handle(), &C::key(key), &serialized_value)
|
||||
}
|
||||
|
||||
pub fn delete(&self, key: C::Index) -> Result<()> {
|
||||
self.backend.delete_cf(self.handle(), &C::key(key))
|
||||
}
|
||||
}
|
||||
|
||||
impl<C> LedgerColumn<C>
|
||||
where
|
||||
|
|
Loading…
Reference in New Issue