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

pagebench get-page: default to latest=true, make configurable via flag (#6252)

Browse Source 
fixes https://github.com/neondatabase/neon/issues/6209
This commit is contained in:
Christian Schwarz
2024-01-02 17:57:29 +01:00
committed by GitHub
parent 946c6a0006
commit aa9f1d4b69
2 changed files with 28 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
pageserver/client/src/page_service.rs
View File

@@ -115,15 +115,8 @@ impl PagestreamClient {
pub async fn getpage(
&mut self,
key: RelTagBlockNo,
lsn: Lsn,
req: PagestreamGetPageRequest,
) -> anyhow::Result<PagestreamGetPageResponse> {
let req = PagestreamGetPageRequest {
latest: false,
rel: key.rel_tag,
blkno: key.block_no,
lsn,
};
let req = PagestreamFeMessage::GetPage(req);
let req: bytes::Bytes = req.serialize();
// let mut req = tokio_util::io::ReaderStream::new(&req);

38
pageserver/pagebench/src/cmd/getpage_latest_lsn.rs
View File

@@ -3,7 +3,7 @@ use futures::future::join_all;
use pageserver::pgdatadir_mapping::key_to_rel_block;
use pageserver::repository;
use pageserver_api::key::is_rel_block_key;
use pageserver_client::page_service::RelTagBlockNo;
use pageserver_api::models::PagestreamGetPageRequest;
use utils::id::TenantTimelineId;
use utils::lsn::Lsn;
@@ -39,6 +39,9 @@ pub(crate) struct Args {
runtime: Option<humantime::Duration>,
#[clap(long)]
per_target_rate_limit: Option<usize>,
/// Probability for sending `latest=true` in the request (uniform distribution).
#[clap(long, default_value = "1")]
req_latest_probability: f64,
#[clap(long)]
limit_to_first_n_targets: Option<usize>,
targets: Option<Vec<TenantTimelineId>>,
@@ -200,18 +203,26 @@ async fn main_impl(
start_work_barrier.wait().await;
loop {
let (range, key) = {
let (timeline, req) = {
let mut rng = rand::thread_rng();
let r = &all_ranges[weights.sample(&mut rng)];
let key: i128 = rng.gen_range(r.start..r.end);
let key = repository::Key::from_i128(key);
let (rel_tag, block_no) =
key_to_rel_block(key).expect("we filter non-rel-block keys out above");
(r, RelTagBlockNo { rel_tag, block_no })
(
r.timeline,
PagestreamGetPageRequest {
latest: rng.gen_bool(args.req_latest_probability),
lsn: r.timeline_lsn,
rel: rel_tag,
blkno: block_no,
},
)
};
let sender = work_senders.get(&range.timeline).unwrap();
let sender = work_senders.get(&timeline).unwrap();
// TODO: what if this blocks?
sender.send((key, range.timeline_lsn)).await.ok().unwrap();
sender.send(req).await.ok().unwrap();
}
}),
Some(rps_limit) => Box::pin(async move {
@@ -240,16 +251,21 @@ async fn main_impl(
);
loop {
ticker.tick().await;
let (range, key) = {
let req = {
let mut rng = rand::thread_rng();
let r = &ranges[weights.sample(&mut rng)];
let key: i128 = rng.gen_range(r.start..r.end);
let key = repository::Key::from_i128(key);
let (rel_tag, block_no) = key_to_rel_block(key)
.expect("we filter non-rel-block keys out above");
(r, RelTagBlockNo { rel_tag, block_no })
PagestreamGetPageRequest {
latest: rng.gen_bool(args.req_latest_probability),
lsn: r.timeline_lsn,
rel: rel_tag,
blkno: block_no,
}
};
sender.send((key, range.timeline_lsn)).await.ok().unwrap();
sender.send(req).await.ok().unwrap();
}
})
};
@@ -303,7 +319,7 @@ async fn client(
args: &'static Args,
timeline: TenantTimelineId,
start_work_barrier: Arc<Barrier>,
mut work: tokio::sync::mpsc::Receiver<(RelTagBlockNo, Lsn)>,
mut work: tokio::sync::mpsc::Receiver<PagestreamGetPageRequest>,
all_work_done_barrier: Arc<Barrier>,
live_stats: Arc<LiveStats>,
) {
@@ -317,10 +333,10 @@ async fn client(
.await
.unwrap();
while let Some((key, lsn)) = work.recv().await {
while let Some(req) = work.recv().await {
let start = Instant::now();
client
.getpage(key, lsn)
.getpage(req)
.await
.with_context(|| format!("getpage for {timeline}"))
.unwrap();