rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-07-09 07:00:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

debug cruft, likely will revert but this proved useful, esp log_if_slow

Browse Source
This commit is contained in:
Christian Schwarz
2024-12-17 17:32:06 +01:00
parent d962b44c20
commit 0fd67b27e5
13 changed files with 186 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
Cargo.lock generated
View File

@@ -1316,6 +1316,45 @@ dependencies = [
"crossbeam-utils",
]
[[package]]
name = "console-api"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8030735ecb0d128428b64cd379809817e620a40e5001c54465b99ec5feec2857"
dependencies = [
"futures-core",
"prost",
"prost-types",
"tonic",
"tracing-core",
]
[[package]]
name = "console-subscriber"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6539aa9c6a4cd31f4b1c040f860a1eac9aa80e7df6b05d506a6e7179936d6a01"
dependencies = [
"console-api",
"crossbeam-channel",
"crossbeam-utils",
"futures-task",
"hdrhistogram",
"humantime",
"hyper-util",
"prost",
"prost-types",
"serde",
"serde_json",
"thread_local",
"tokio",
"tokio-stream",
"tonic",
"tracing",
"tracing-core",
"tracing-subscriber",
]
[[package]]
name = "const-oid"
version = "0.9.6"
@@ -6601,13 +6640,13 @@ dependencies = [
"signal-hook-registry",
"socket2",
"tokio-macros",
"tracing",
"windows-sys 0.48.0",
]
[[package]]
name = "tokio-epoll-uring"
version = "0.1.0"
source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#33e00106a268644d02ba0461bbd64476073b0ee1"
dependencies = [
"futures",
"nix 0.26.4",
@@ -7143,7 +7182,6 @@ dependencies = [
[[package]]
name = "uring-common"
version = "0.1.0"
source = "git+https://github.com/neondatabase/tokio-epoll-uring.git?branch=main#33e00106a268644d02ba0461bbd64476073b0ee1"
dependencies = [
"bytes",
"io-uring",
@@ -7206,6 +7244,7 @@ dependencies = [
"camino",
"camino-tempfile",
"chrono",
"console-subscriber",
"const_format",
"criterion",
"diatomic-waker",

3
libs/utils/Cargo.toml
View File

@@ -19,6 +19,7 @@ bincode.workspace = true
bytes.workspace = true
camino.workspace = true
chrono.workspace = true
console-subscriber = "*"
diatomic-waker.workspace = true
git-version.workspace = true
hex = { workspace = true, features = ["serde"] }
@@ -39,7 +40,7 @@ serde_with.workspace = true
serde_json.workspace = true
signal-hook.workspace = true
thiserror.workspace = true
tokio.workspace = true
tokio = {workspace = true, features = ["tracing"] }
tokio-tar.workspace = true
tokio-util.workspace = true
toml_edit = { workspace = true, features = ["serde"] }

18
libs/utils/src/logging.rs
View File

@@ -1,10 +1,12 @@
use std::str::FromStr;
use std::{num::NonZeroUsize, str::FromStr};
use anyhow::Context;
use metrics::{IntCounter, IntCounterVec};
use once_cell::sync::Lazy;
use strum_macros::{EnumString, VariantNames};
use crate::env;
#[derive(EnumString, strum_macros::Display, VariantNames, Eq, PartialEq, Debug, Clone, Copy)]
#[strum(serialize_all = "snake_case")]
pub enum LogFormat {
@@ -134,6 +136,20 @@ pub fn init(
let r = r.with(
TracingEventCountLayer(&TRACING_EVENT_COUNT_METRIC).with_filter(rust_log_env_filter()),
);
let r = r.with(
if let Some(n) = env::var("NEON_ENABLE_TOKIO_CONSOLE_SUBSCRIBER") {
let n: NonZeroUsize = n;
use console_subscriber::ConsoleLayer;
Some(
console_subscriber::Builder::default()
.event_buffer_capacity(n.get() * ConsoleLayer::DEFAULT_EVENT_BUFFER_CAPACITY)
.client_buffer_capacity(n.get() * ConsoleLayer::DEFAULT_CLIENT_BUFFER_CAPACITY)
.spawn(),
)
} else {
None
},
);
match tracing_error_layer_enablement {
TracingErrorLayerEnablement::EnableWithRustLogFilter => r
.with(tracing_error::ErrorLayer::default().with_filter(rust_log_env_filter()))

2
pageserver/Cargo.toml
View File

@@ -57,7 +57,7 @@ sysinfo.workspace = true
tokio-tar.workspace = true
thiserror.workspace = true
tikv-jemallocator.workspace = true
tokio = { workspace = true, features = ["process", "sync", "fs", "rt", "io-util", "time"] }
tokio = { workspace = true, features = ["process", "sync", "fs", "rt", "io-util", "time", "tracing"] }
tokio-epoll-uring.workspace = true
tokio-io-timeout.workspace = true
tokio-postgres.workspace = true

23
pageserver/src/lib.rs
View File

@@ -375,6 +375,29 @@ async fn timed_after_cancellation<Fut: std::future::Future>(
}
}
async fn log_if_slow<Fut: std::future::Future>(
name: &str,
warn_at: std::time::Duration,
fut: Fut,
) -> <Fut as std::future::Future>::Output {
let started = std::time::Instant::now();
let mut fut = std::pin::pin!(fut);
match tokio::time::timeout(warn_at, &mut fut).await {
Ok(ret) => ret,
Err(_) => {
tracing::trace!(
what = name,
elapsed_ms = started.elapsed().as_millis(),
"slow"
);
fut.await
}
}
}
#[cfg(test)]
mod timed_tests {
use super::timed;

6
pageserver/src/page_service.rs
View File

@@ -1208,6 +1208,7 @@ impl PageServerHandler {
{
let cancel = self.cancel.clone();
let err = loop {
trace!("waiting for message");
let msg = Self::pagestream_read_message(
&mut pgb_reader,
tenant_id,
@@ -1218,6 +1219,7 @@ impl PageServerHandler {
request_span.clone(),
)
.await;
trace!(is_err = msg.is_err(), "message received");
let msg = match msg {
Ok(msg) => msg,
Err(e) => break e,
@@ -1229,10 +1231,11 @@ impl PageServerHandler {
return ((pgb_reader, timeline_handles), Ok(()));
}
};
trace!("throttling message");
if let Err(cancelled) = msg.throttle(&self.cancel).await {
break cancelled;
}
trace!("handling message");
let err = self
.pagesteam_handle_batched_message(pgb_writer, msg, &cancel, ctx)
@@ -1241,6 +1244,7 @@ impl PageServerHandler {
Ok(()) => {}
Err(e) => break e,
}
trace!("message handled");
};
((pgb_reader, timeline_handles), Err(err))
}

30
pageserver/src/tenant/block_io.rs
View File

@@ -4,12 +4,14 @@
use super::storage_layer::delta_layer::{Adapter, DeltaLayerInner};
use crate::context::RequestContext;
use crate::log_if_slow;
use crate::page_cache::{self, FileId, PageReadGuard, PageWriteGuard, ReadBufResult, PAGE_SZ};
#[cfg(test)]
use crate::virtual_file::IoBufferMut;
use crate::virtual_file::VirtualFile;
use bytes::Bytes;
use std::ops::Deref;
use std::time::Duration;
/// This is implemented by anything that can read 8 kB (PAGE_SZ)
/// blocks, using the page cache
@@ -211,19 +213,27 @@ impl<'a> FileBlockReader<'a> {
ctx: &RequestContext,
) -> Result<BlockLease<'b>, std::io::Error> {
let cache = page_cache::get();
match cache
.read_immutable_buf(self.file_id, blknum, ctx)
.await
.map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Failed to read immutable buf: {e:#}"),
)
})? {
match log_if_slow(
"read_immutable_buf",
Duration::from_secs(1),
cache.read_immutable_buf(self.file_id, blknum, ctx),
)
.await
.map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::Other,
format!("Failed to read immutable buf: {e:#}"),
)
})? {
ReadBufResult::Found(guard) => Ok(guard.into()),
ReadBufResult::NotFound(write_guard) => {
// Read the page from disk into the buffer
let write_guard = self.fill_buffer(write_guard, blknum, ctx).await?;
let write_guard = log_if_slow(
"fill_buffer",
Duration::from_secs(1),
self.fill_buffer(write_guard, blknum, ctx),
)
.await?;
Ok(write_guard.mark_valid().into())
}
}

10
pageserver/src/tenant/disk_btree.rs
View File

@@ -29,15 +29,13 @@ use std::{
io,
iter::Rev,
ops::{Range, RangeInclusive},
result,
result, time::Duration,
};
use thiserror::Error;
use tracing::error;
use crate::{
context::{DownloadBehavior, RequestContext},
task_mgr::TaskKind,
tenant::block_io::{BlockReader, BlockWriter},
context::{DownloadBehavior, RequestContext}, log_if_slow, task_mgr::TaskKind, tenant::block_io::{BlockReader, BlockWriter}
};
// The maximum size of a value stored in the B-tree. 5 bytes is enough currently.
@@ -302,8 +300,8 @@ where
// We could keep the page cache read guard alive, but, at the time of writing,
// we run quite small PS PageCache s => can't risk running out of
// PageCache space because this stream isn't consumed fast enough.
let page_read_guard = block_cursor
.read_blk(self.start_blk + node_blknum, ctx)
let page_read_guard = log_if_slow("read_blk", Duration::from_secs(1), block_cursor
.read_blk(self.start_blk + node_blknum, ctx))
.await?;
node_buf.copy_from_slice(page_read_guard.as_ref());
drop(page_read_guard); // drop page cache read guard early

11
pageserver/src/tenant/storage_layer.rs
View File

@@ -22,9 +22,11 @@ use std::collections::{BinaryHeap, HashMap};
use std::future::Future;
use std::ops::Range;
use std::pin::Pin;
use std::sync::atomic::AtomicUsize;
use std::sync::Arc;
use std::task::Poll;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tracing::{trace, Instrument};
use utils::lsn::Lsn;
@@ -177,6 +179,15 @@ impl IoConcurrency {
where
F: std::future::Future<Output = ()> + Send + 'static,
{
static IO_NUM: AtomicUsize = AtomicUsize::new(0);
let io_num = IO_NUM.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let fut = async move {
trace!("start");
scopeguard::defer!({ trace!("end") });
fut.await
}
.instrument(tracing::trace_span!("spawned_io", %io_num));
tracing::trace!(%io_num, "spawning IO");
match self {
IoConcurrency::Serial => fut.await,
IoConcurrency::Parallel => {

26
pageserver/src/tenant/storage_layer/delta_layer.rs
View File

@@ -885,6 +885,7 @@ impl DeltaLayerInner {
Ok(())
}
#[instrument(level = "trace", skip_all)]
async fn plan_reads<Reader>(
keyspace: &KeySpace,
lsn_range: Range<Lsn>,
@@ -896,18 +897,30 @@ impl DeltaLayerInner {
where
Reader: BlockReader + Clone,
{
trace!("enter");
scopeguard::defer!({
trace!("exit");
});
let ctx = RequestContextBuilder::extend(ctx)
.page_content_kind(PageContentKind::DeltaLayerBtreeNode)
.build();
for range in keyspace.ranges.iter() {
let nranges = keyspace.ranges.len();
trace!("Planning reads for {nranges} ranges");
for (i, range) in keyspace.ranges.iter().enumerate() {
trace!("range {i}/{nranges}");
let mut range_end_handled = false;
let start_key = DeltaKey::from_key_lsn(&range.start, lsn_range.start);
let index_stream = index_reader.clone().into_stream(&start_key.0, &ctx);
let mut index_stream = std::pin::pin!(index_stream);
let mut n = 0;
while let Some(index_entry) = index_stream.next().await {
trace!("index entry {n}");
n += 1;
let (raw_key, value) = index_entry?;
let key = Key::from_slice(&raw_key[..KEY_SIZE]);
let lsn = DeltaKey::extract_lsn_from_buf(&raw_key);
@@ -986,12 +999,18 @@ impl DeltaLayerInner {
largest_read_size
}
#[instrument(level = "trace", skip_all)]
async fn do_reads_and_update_state(
&self,
reads: Vec<VectoredRead>,
reconstruct_state: &mut ValuesReconstructState,
ctx: &RequestContext,
) {
trace!("enter");
scopeguard::defer!({
trace!("exit");
});
let max_vectored_read_bytes = self
.max_vectored_read_bytes
.expect("Layer is loaded with max vectored bytes config")
@@ -1004,7 +1023,10 @@ impl DeltaLayerInner {
// Note that reads are processed in reverse order (from highest key+lsn).
// This is the order that `ReconstructState` requires such that it can
// track when a key is done.
for read in reads.into_iter().rev() {
let reads_len = reads.len();
trace!("Processing {reads_len} reads");
for (i, read) in reads.into_iter().rev().enumerate() {
trace!("read {i}/{reads_len}");
let mut senders: HashMap<
(Key, Lsn),
sync::oneshot::Sender<Result<OnDiskValue, std::io::Error>>,

18
pageserver/src/tenant/timeline.rs
View File

@@ -56,7 +56,7 @@ use utils::{
};
use wal_decoder::serialized_batch::{SerializedValueBatch, ValueMeta};
use std::sync::atomic::Ordering as AtomicOrdering;
use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex, RwLock, Weak};
use std::time::{Duration, Instant, SystemTime};
use std::{
@@ -1159,6 +1159,7 @@ impl Timeline {
vectored_res
}
#[instrument(level = "trace", skip_all, fields(request_num = tracing::field::Empty))]
pub(super) async fn get_vectored_impl(
&self,
keyspace: KeySpace,
@@ -1172,6 +1173,11 @@ impl Timeline {
GetKind::Vectored
};
static REQUEST_NUM: AtomicUsize = AtomicUsize::new(0);
let request_num = REQUEST_NUM.fetch_add(1, AtomicOrdering::Relaxed);
tracing::Span::current().record("request_num", &request_num);
trace!("getting reconstruct data");
let get_data_timer = crate::metrics::GET_RECONSTRUCT_DATA_TIME
.for_get_kind(get_kind)
.start_timer();
@@ -1184,6 +1190,7 @@ impl Timeline {
.start_timer();
let layers_visited = reconstruct_state.get_layers_visited();
trace!("waiting for reconstruct data and reconstructing values");
let futs = FuturesUnordered::new();
for (key, state) in std::mem::take(&mut reconstruct_state.keys) {
futs.push({
@@ -1191,6 +1198,7 @@ impl Timeline {
async move {
assert_eq!(state.situation, ValueReconstructSituation::Complete);
trace!("collecting pending IOs");
let converted = match state.collect_pending_ios().await {
Ok(ok) => ok,
Err(err) => {
@@ -1198,11 +1206,13 @@ impl Timeline {
}
};
trace!("reconstructing value");
(
key,
walredo_self.reconstruct_value(key, lsn, converted).await,
)
}
.instrument(tracing::trace_span!("key_loop", key = %key, lsn = lsn.0))
});
}
@@ -3423,7 +3433,11 @@ impl Timeline {
super::storage_layer::IoConcurrency::Serial => (),
super::storage_layer::IoConcurrency::Parallel => (),
super::storage_layer::IoConcurrency::FuturesUnordered { ref mut futures } => {
while let Some(()) = futures.next().await {}
trace!("waiting for futures to complete");
while let Some(()) = futures.next().await {
trace!("future completed");
}
trace!("futures completed");
}
}

11
pageserver/src/virtual_file.rs
View File

@@ -12,6 +12,7 @@
//! src/backend/storage/file/fd.c
//!
use crate::context::RequestContext;
use crate::log_if_slow;
use crate::metrics::{StorageIoOperation, STORAGE_IO_SIZE, STORAGE_IO_TIME_METRIC};
use crate::page_cache::{PageWriteGuard, PAGE_SZ};
@@ -28,6 +29,7 @@ use std::fs::File;
use std::io::{Error, ErrorKind, Seek, SeekFrom};
#[cfg(target_os = "linux")]
use std::os::unix::fs::OpenOptionsExt;
use std::time::Duration;
use tokio_epoll_uring::{BoundedBuf, IoBuf, IoBufMut, Slice};
use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd};
@@ -934,10 +936,11 @@ impl VirtualFileInner {
where
Buf: tokio_epoll_uring::IoBufMut + Send,
{
let file_guard = match self.lock_file().await {
Ok(file_guard) => file_guard,
Err(e) => return (buf, Err(e)),
};
let file_guard =
match log_if_slow("lock_file", Duration::from_secs(1), self.lock_file()).await {
Ok(file_guard) => file_guard,
Err(e) => return (buf, Err(e)),
};
observe_duration!(StorageIoOperation::Read, {
let ((_file_guard, buf), res) = io_engine::get().read_at(file_guard, offset, buf).await;

18
pageserver/src/virtual_file/io_engine.rs
View File

@@ -15,6 +15,8 @@ pub(super) mod tokio_epoll_uring_ext;
use tokio_epoll_uring::IoBuf;
use tracing::Instrument;
use crate::log_if_slow;
pub(crate) use super::api::IoEngineKind;
#[derive(Clone, Copy)]
#[repr(u8)]
@@ -109,7 +111,7 @@ pub(crate) fn get() -> IoEngine {
use std::{
os::unix::prelude::FileExt,
sync::atomic::{AtomicU8, Ordering},
sync::atomic::{AtomicU8, Ordering}, time::Duration,
};
use super::{
@@ -149,8 +151,18 @@ impl IoEngine {
}
#[cfg(target_os = "linux")]
IoEngine::TokioEpollUring => {
let system = tokio_epoll_uring_ext::thread_local_system().await;
let (resources, res) = system.read(file_guard, offset, slice).await;
let system = log_if_slow(
"thread_local_system",
Duration::from_secs(1),
tokio_epoll_uring_ext::thread_local_system(),
)
.await;
let (resources, res) = log_if_slow(
"system.read",
Duration::from_secs(1),
system.read(file_guard, offset, slice),
)
.await;
(resources, res.map_err(epoll_uring_error_to_std))
}
}