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

up

Browse Source
This commit is contained in:
Dominik Werder
2021-05-14 21:27:39 +02:00
parent d6357954c1
commit f63624f8e2
2 changed files with 10 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
disk/src/dataopen.rs
View File

@@ -64,8 +64,8 @@ async fn open_files_inner(
}
timebins.sort_unstable();
let tgt_tb = (range.beg / MS) as f32 / (channel_config.time_bin_size.ns / MS) as f32;
trace!("tgt_tb: {:?}", tgt_tb);
trace!("timebins found: {:?}", timebins);
info!("tgt_tb: {:?}", tgt_tb);
info!("timebins found: {:?}", timebins);
for &tb in &timebins {
let ts_bin = Nanos {
ns: tb * channel_config.time_bin_size.ns,
@@ -76,11 +76,11 @@ async fn open_files_inner(
if ts_bin.ns + channel_config.time_bin_size.ns <= range.beg {
continue;
}
debug!("opening tb {:?}", &tb);
//debug!("opening tb {:?}", &tb);
let path = paths::datapath(tb, &channel_config, &node);
debug!("opening path {:?}", &path);
info!("opening path {:?}", &path);
let mut file = OpenOptions::new().read(true).open(&path).await?;
debug!("opened file {:?} {:?}", &path, &file);
//info!("opened file {:?} {:?}", &path, &file);
let ret = {
let index_path = paths::index_path(ts_bin, &channel_config, &node)?;
match OpenOptions::new().read(true).open(&index_path).await {

8
disk/src/eventblobs.rs
View File

@@ -33,7 +33,7 @@ impl EventBlobsComplete {
buffer_size: usize,
event_chunker_conf: EventChunkerConf,
) -> Self {
info!("EventBlobsComplete::new beg {}", range.beg / SEC);
//info!("EventBlobsComplete::new beg {}", range.beg / SEC);
Self {
file_chan: open_files(&range, &channel_config, node),
evs: None,
@@ -73,10 +73,10 @@ impl Stream for EventBlobsComplete {
Ready(Some(k)) => match k {
Ok(file) => {
let path = file.path;
//info!("handling {:?}", path);
let item = LogItem::quick(Level::INFO, format!("handle file {:?}", path));
match file.file {
Some(file) => {
info!("got file {:?}", path);
let inp = Box::pin(file_content_stream(file, self.buffer_size as usize));
let chunker = EventChunker::from_event_boundary(
inp,
@@ -88,7 +88,9 @@ impl Stream for EventBlobsComplete {
);
self.evs = Some(chunker);
}
None => {}
None => {
info!("skip {:?}", path);
}
}
Ready(Some(Ok(EventChunkerItem::Log(item))))
//continue 'outer;