daqbuf/daqbuffer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Show query range in CLI, improve cache read

Browse Source
This commit is contained in:
Dominik Werder
2021-05-27 15:02:59 +02:00
parent 7aecf59195
commit e55751ad67
2 changed files with 27 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
disk/src/binned.rs
View File

@@ -355,6 +355,7 @@ where
T: ReadableFromFile, T: ReadableFromFile,
{ {
buf: Vec<u8>, buf: Vec<u8>,
all: Vec<u8>,
file: Option<File>, file: Option<File>,
_marker: std::marker::PhantomData<T>, _marker: std::marker::PhantomData<T>,
} }
@@ -365,7 +366,9 @@ where
{ {
fn new(file: File) -> Self { fn new(file: File) -> Self {
Self { Self {
buf: vec![], // TODO make buffer size a parameter:
buf: vec![0; 4096],
all: vec![],
file: Some(file), file: Some(file),
_marker: std::marker::PhantomData::default(), _marker: std::marker::PhantomData::default(),
} }
@@ -380,20 +383,22 @@ where
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
use Poll::*; use Poll::*;
'outer: loop { let mut buf = std::mem::replace(&mut self.buf, Vec::new());
// TODO make buffer size a parameter: let ret = 'outer: loop {
let mut buf = vec![0; 4096];
let mut dst = ReadBuf::new(&mut buf); let mut dst = ReadBuf::new(&mut buf);
if dst.remaining() == 0 || dst.capacity() == 0 {
break Ready(Err(Error::with_msg("bad read buffer")));
}
let fp = self.file.as_mut().unwrap(); let fp = self.file.as_mut().unwrap();
let f = Pin::new(fp); let f = Pin::new(fp);
break match File::poll_read(f, cx, &mut dst) { break match File::poll_read(f, cx, &mut dst) {
Ready(res) => match res { Ready(res) => match res {
Ok(_) => { Ok(_) => {
if dst.filled().len() > 0 { if dst.filled().len() > 0 {
self.buf.extend_from_slice(&mut buf); self.all.extend_from_slice(dst.filled());
continue 'outer; continue 'outer;
} else { } else {
match T::from_buf(&mut self.buf) { match T::from_buf(&mut self.all) {
Ok(item) => Ready(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))), Ok(item) => Ready(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))),
Err(e) => Ready(Err(e)), Err(e) => Ready(Err(e)),
} }
@@ -403,7 +408,9 @@ where
}, },
Pending => Pending, Pending => Pending,
}; };
} };
self.buf = buf;
ret
} }
} }

15
retrieval/src/client.rs
View File

@@ -47,6 +47,10 @@ pub async fn get_binned(
disk_stats_every_kb: u32, disk_stats_every_kb: u32,
) -> Result<(), Error> { ) -> Result<(), Error> {
info!("------- get_binned client"); info!("------- get_binned client");
info!("channel {}", channel_name);
info!("beg {}", beg_date);
info!("end {}", end_date);
info!("-------");
let t1 = Utc::now(); let t1 = Utc::now();
let date_fmt = "%Y-%m-%dT%H:%M:%S.%3fZ"; let date_fmt = "%Y-%m-%dT%H:%M:%S.%3fZ";
let uri = format!( let uri = format!(
@@ -65,7 +69,6 @@ pub async fn get_binned(
cache_usage.query_param_value(), cache_usage.query_param_value(),
disk_stats_every_kb, disk_stats_every_kb,
); );
info!("get_binned uri {:?}", uri);
let req = hyper::Request::builder() let req = hyper::Request::builder()
.method(http::Method::GET) .method(http::Method::GET)
.uri(uri) .uri(uri)
@@ -78,7 +81,15 @@ pub async fn get_binned(
let (head, body) = res.into_parts(); let (head, body) = res.into_parts();
let buf = hyper::body::to_bytes(body).await?; let buf = hyper::body::to_bytes(body).await?;
let s = String::from_utf8_lossy(&buf); let s = String::from_utf8_lossy(&buf);
return Err(Error::with_msg(format!("Server error {:?}\n---------------------- message from http body:\n{}\n---------------------- end of http body", head, s))); return Err(Error::with_msg(format!(
concat!(
"Server error {:?}\n",
"---------------------- message from http body:\n",
"{}\n",
"---------------------- end of http body",
),
head, s
)));
} }
let perf_opts = PerfOpts { inmem_bufcap: 512 }; let perf_opts = PerfOpts { inmem_bufcap: 512 };
let s1 = disk::cache::HttpBodyAsAsyncRead::new(res); let s1 = disk::cache::HttpBodyAsAsyncRead::new(res);