diff --git a/pageserver/client/src/page_service.rs b/pageserver/client/src/page_service.rs index 3827e19aa4..f312b888b6 100644 --- a/pageserver/client/src/page_service.rs +++ b/pageserver/client/src/page_service.rs @@ -118,11 +118,13 @@ pub struct PagestreamClient { } pub struct PagestreamSender { + #[allow(dead_code)] shared: Arc>, sink: SplitSink, bytes::Bytes>, } pub struct PagestreamReceiver { + #[allow(dead_code)] shared: Arc>, stream: SplitStream>, } @@ -188,7 +190,7 @@ impl PagestreamClient { pub fn split(self) -> (PagestreamSender, PagestreamReceiver) { let Self { - shared, + shared: _, sink, stream, } = self; diff --git a/pageserver/src/bin/test_helper_slow_client_reads.rs b/pageserver/src/bin/test_helper_slow_client_reads.rs index 6fad937bf5..9a812e436d 100644 --- a/pageserver/src/bin/test_helper_slow_client_reads.rs +++ b/pageserver/src/bin/test_helper_slow_client_reads.rs @@ -4,11 +4,10 @@ use std::{ }; use clap::Parser; -use pageserver_api::models::{PagestreamGetPageRequest, PagestreamRequest, PagestreamTestRequest}; +use pageserver_api::models::{PagestreamRequest, PagestreamTestRequest}; use utils::{ id::{TenantId, TimelineId}, lsn::Lsn, - shard::TenantShardId, }; #[derive(clap::Parser)] @@ -27,7 +26,7 @@ async fn main() -> anyhow::Result<()> { } = Args::parse(); let client = pageserver_client::page_service::Client::new(connstr).await?; let client = client.pagestream(tenant_id, timeline_id).await?; - let (mut sender, mut receiver) = client.split(); + let (mut sender, _receiver) = client.split(); eprintln!("filling the pipe"); let mut msg = 0; @@ -44,7 +43,7 @@ async fn main() -> anyhow::Result<()> { message: format!("message {}", msg), }, )); - let Ok(res) = tokio::time::timeout(Duration::from_secs(1), fut).await else { + let Ok(res) = tokio::time::timeout(Duration::from_secs(10), fut).await else { eprintln!("pipe seems full"); break; }; @@ -54,6 +53,8 @@ async fn main() -> anyhow::Result<()> { stdout().write(b"R")?; stdout().flush()?; + eprintln!("waiting for signal to tell us to exit"); + let mut buf = [0u8; 1]; stdin().read_exact(&mut buf)?; diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index 301f248517..96d1a826a9 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -39,7 +39,6 @@ use tokio::io::{AsyncWriteExt, BufWriter}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::*; -use utils::failpoint_support; use utils::sync::spsc_fold; use utils::{ auth::{Claims, Scope, SwappableJwtAuth}, @@ -1221,10 +1220,7 @@ impl PageServerHandler { }); // what we want to do - let flush_fut = async { - failpoint_support::sleep_millis_async!("page_service:flush:pre"); - pgb_writer.flush().await - }; + let flush_fut = pgb_writer.flush(); // metric for how long flushing takes let flush_fut = match flushing_timer { Some(flushing_timer) => { @@ -1873,28 +1869,25 @@ impl PageServerHandler { )) } + // NB: this impl mimics what we do for batched getpage requests. #[cfg(feature = "testing")] #[instrument(skip_all, fields(shard_id))] async fn handle_test_request_batch( &mut self, timeline: &Timeline, requests: Vec, - ctx: &RequestContext, + _ctx: &RequestContext, ) -> Vec> { - use pageserver_api::models::PagestreamTestResponse; - - let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn(); - // real requests would do something with the timeline let mut results = Vec::with_capacity(requests.len()); - for req in requests.iter() { + for _req in requests.iter() { tokio::task::yield_now().await; results.push({ if timeline.cancel.is_cancelled() { Err(PageStreamError::Shutdown) } else { - Ok(PagestreamTestResponse { req: req.req.clone() }) + Ok(()) } }); } @@ -1905,7 +1898,7 @@ impl PageServerHandler { .into_iter() .zip(results.into_iter()) .map(|(req, res)| { - res.map(|page| { + res.map(|()| { ( PagestreamBeMessage::Test(models::PagestreamTestResponse { req: req.req.clone(), diff --git a/test_runner/regress/test_page_service_batching_regressions.py b/test_runner/regress/test_page_service_batching_regressions.py index e4312644b4..f781dd4850 100644 --- a/test_runner/regress/test_page_service_batching_regressions.py +++ b/test_runner/regress/test_page_service_batching_regressions.py @@ -19,8 +19,6 @@ def test_slow_flush(neon_env_builder: NeonEnvBuilder, neon_binpath: Path): env = neon_env_builder.init_start() log.info("make flush appear slow") - ps_http = env.pageserver.http_client() - ps_http.configure_failpoints(("page_service:flush:pre", "return(10000000)")) log.info("filling pipe") child = subprocess.Popen(