diff --git a/Cargo.lock b/Cargo.lock index e2d5e03613..cca3a54a06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/libs/utils/Cargo.toml b/libs/utils/Cargo.toml index 66500fb141..a7ee53a52d 100644 --- a/libs/utils/Cargo.toml +++ b/libs/utils/Cargo.toml @@ -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"] } diff --git a/libs/utils/src/logging.rs b/libs/utils/src/logging.rs index e205d60d74..0f20718eee 100644 --- a/libs/utils/src/logging.rs +++ b/libs/utils/src/logging.rs @@ -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())) diff --git a/pageserver/Cargo.toml b/pageserver/Cargo.toml index 140b287ccc..e33cbd98a1 100644 --- a/pageserver/Cargo.toml +++ b/pageserver/Cargo.toml @@ -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 diff --git a/pageserver/src/lib.rs b/pageserver/src/lib.rs index ff6af3566c..1ac7586769 100644 --- a/pageserver/src/lib.rs +++ b/pageserver/src/lib.rs @@ -375,6 +375,29 @@ async fn timed_after_cancellation( } } +async fn log_if_slow( + name: &str, + warn_at: std::time::Duration, + fut: Fut, +) -> ::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; diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index 97d94bbe7f..928da3852d 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -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)) } diff --git a/pageserver/src/tenant/block_io.rs b/pageserver/src/tenant/block_io.rs index 2bd7f2d619..cf9cfcd827 100644 --- a/pageserver/src/tenant/block_io.rs +++ b/pageserver/src/tenant/block_io.rs @@ -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, 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()) } } diff --git a/pageserver/src/tenant/disk_btree.rs b/pageserver/src/tenant/disk_btree.rs index b302cbc975..ac86216a41 100644 --- a/pageserver/src/tenant/disk_btree.rs +++ b/pageserver/src/tenant/disk_btree.rs @@ -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 diff --git a/pageserver/src/tenant/storage_layer.rs b/pageserver/src/tenant/storage_layer.rs index f56f9bf6e4..0e9e27e135 100644 --- a/pageserver/src/tenant/storage_layer.rs +++ b/pageserver/src/tenant/storage_layer.rs @@ -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 + 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 => { diff --git a/pageserver/src/tenant/storage_layer/delta_layer.rs b/pageserver/src/tenant/storage_layer/delta_layer.rs index a27fc39c73..516aa9b291 100644 --- a/pageserver/src/tenant/storage_layer/delta_layer.rs +++ b/pageserver/src/tenant/storage_layer/delta_layer.rs @@ -885,6 +885,7 @@ impl DeltaLayerInner { Ok(()) } + #[instrument(level = "trace", skip_all)] async fn plan_reads( keyspace: &KeySpace, lsn_range: Range, @@ -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, 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>, diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index cd7e4d250d..cd52843ecf 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -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"); } } diff --git a/pageserver/src/virtual_file.rs b/pageserver/src/virtual_file.rs index 8a7f4a4bf5..3f93404ece 100644 --- a/pageserver/src/virtual_file.rs +++ b/pageserver/src/virtual_file.rs @@ -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; diff --git a/pageserver/src/virtual_file/io_engine.rs b/pageserver/src/virtual_file/io_engine.rs index ccde90ee1a..41abff382c 100644 --- a/pageserver/src/virtual_file/io_engine.rs +++ b/pageserver/src/virtual_file/io_engine.rs @@ -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)) } }