concurrent reading

This commit is contained in:
Jana Dönszelmann 2026-02-25 15:32:13 +01:00
parent 3d9114dea9
commit 1f6679f57f
No known key found for this signature in database
8 changed files with 104 additions and 84 deletions

View file

@ -1,4 +1,4 @@
use std::rc::Rc;
use std::sync::Arc;
use serde::{Deserialize, Serialize};
@ -6,7 +6,7 @@ use crate::tui::filter::Filter;
#[derive(Serialize, Deserialize)]
pub struct Filters {
filters: Vec<Rc<Filter>>,
filters: Vec<Arc<Filter>>,
undo_pos: usize,
}
@ -18,11 +18,11 @@ impl Filters {
}
}
pub fn get(&self) -> &[Rc<Filter>] {
pub fn get(&self) -> &[Arc<Filter>] {
&self.filters[0..self.undo_pos]
}
pub fn push(&mut self, filter: Rc<Filter>) {
pub fn push(&mut self, filter: Arc<Filter>) {
self.filters.truncate(self.undo_pos);
self.filters.push(filter);
self.undo_pos = self.filters.len();

View file

@ -1,4 +1,4 @@
use std::{collections::HashMap, iter, mem, rc::Rc};
use std::{collections::HashMap, iter, mem, sync::Arc};
use crate::tui::{
filter::Filter,
@ -58,7 +58,7 @@ impl LogViewer {
pub fn filtered_root_stream(&self) -> Box<dyn LogStream> {
let mut curr = self.root_stream.clone();
for filter in self.filters.get() {
curr = Box::new(curr.filter(Rc::clone(filter)));
curr = Box::new(curr.filter(Arc::clone(filter)));
}
curr
@ -78,12 +78,12 @@ impl LogViewer {
fn find_elem_in_stream(
stream: &dyn LogStream,
elem: &Rc<LogEntry>,
elem: &Arc<LogEntry>,
) -> Option<Box<dyn LogStream>> {
let mut temp_stream = stream.clone();
let mut max = 100usize;
while let Some((curr, _)) = temp_stream.next() {
if Rc::ptr_eq(&curr, elem) {
if Arc::ptr_eq(&curr, elem) {
return Some(temp_stream);
}
@ -162,8 +162,8 @@ impl LogViewer {
self.stack = new_stack;
}
pub fn add_filter(&mut self, filter: Rc<Filter>) {
self.filters.push(Rc::clone(&filter));
pub fn add_filter(&mut self, filter: Arc<Filter>) {
self.filters.push(Arc::clone(&filter));
self.update_filters();
}
@ -200,7 +200,7 @@ impl LogViewer {
}
}
pub fn items(&mut self, max: usize) -> Option<(Vec<(Rc<LogEntry>, usize)>, usize)> {
pub fn items(&mut self, max: usize) -> Option<(Vec<(Arc<LogEntry>, usize)>, usize)> {
let mut temp_iter = self.curr.iter.clone();
let mut res = Vec::new();
for _ in 0..max {
@ -241,7 +241,7 @@ impl LogViewer {
}
}
pub fn selected(&self) -> Option<(Rc<LogEntry>, usize)> {
pub fn selected(&self) -> Option<(Arc<LogEntry>, usize)> {
self.curr.selected()
}

View file

@ -1,4 +1,4 @@
use std::rc::Rc;
use std::sync::Arc;
use crate::tui::{model::LogEntry, processing::LogStream};
@ -8,7 +8,7 @@ pub struct LogView {
}
impl LogView {
pub fn selected(&self) -> Option<(Rc<LogEntry>, usize)> {
pub fn selected(&self) -> Option<(Arc<LogEntry>, usize)> {
let mut temp_iter = self.iter.clone();
for _ in 0..self.selection_offset {
let _ = temp_iter.next()?;

View file

@ -12,7 +12,7 @@ use std::{
ops::ControlFlow,
path::{Path, PathBuf},
process::exit,
rc::Rc,
sync::Arc,
};
use tui_widget_list::{ListBuilder, ListView};
@ -31,7 +31,7 @@ use ratatui::{
crossterm::event::{self, Event, KeyCode, KeyEvent, KeyModifiers},
layout::{Constraint, HorizontalAlignment, Layout, Rect},
prelude::CrosstermBackend,
style::{Modifier, Style},
style::Style,
text::Line,
widgets::{
Block, Clear, List, ListItem, ListState, Padding, Paragraph, StatefulWidget, Widget, Wrap,
@ -250,7 +250,7 @@ impl App {
if let InputState::Target(t) = lv.input_state.clone()
&& let Some(m) = Matcher::from_input(t, lv)
{
lv.add_filter(Rc::new(Filter {
lv.add_filter(Arc::new(Filter {
matcher: m,
kind: FilterKind::Remove,
}));
@ -266,7 +266,7 @@ impl App {
if let InputState::Target(t) = lv.input_state.clone()
&& let Some(m) = Matcher::from_input(t, lv)
{
lv.add_filter(Rc::new(Filter {
lv.add_filter(Arc::new(Filter {
matcher: m,
kind: FilterKind::Inline,
}));

View file

@ -2,8 +2,7 @@ use std::{
collections::BTreeMap,
hash::{DefaultHasher, Hash, Hasher},
path::PathBuf,
rc::Rc,
sync::OnceLock,
sync::{Arc, OnceLock},
};
use jiff::Timestamp;
@ -45,7 +44,7 @@ pub enum LogEntry {
},
Sub {
enter: RawLogEntry,
sub_entries: Vec<Rc<LogEntry>>,
sub_entries: Vec<Arc<LogEntry>>,
exit: RawLogEntry,
count_sub: OnceLock<usize>,

View file

@ -1,4 +1,4 @@
use std::rc::Rc;
use std::sync::Arc;
use crate::tui::{
filter::{Filter, FilterKind},
@ -13,23 +13,23 @@ pub trait IntoLogStream {
}
pub struct LogEntryStream {
inner: Rc<LogEntry>,
inner: Arc<LogEntry>,
curr: usize,
inline_depth: usize,
}
impl LogStream for LogEntryStream {
fn next(&mut self) -> Option<(Rc<LogEntry>, usize)> {
fn next(&mut self) -> Option<(Arc<LogEntry>, usize)> {
let LogEntry::Sub { sub_entries, .. } = self.inner.as_ref() else {
return None;
};
let res = sub_entries.get(self.curr)?;
self.curr += 1;
Some((Rc::clone(res), self.inline_depth))
Some((Arc::clone(res), self.inline_depth))
}
fn prev(&mut self) -> Option<(Rc<LogEntry>, usize)> {
fn prev(&mut self) -> Option<(Arc<LogEntry>, usize)> {
let LogEntry::Sub { sub_entries, .. } = self.inner.as_ref() else {
return None;
};
@ -40,29 +40,29 @@ impl LogStream for LogEntryStream {
self.curr -= 1;
let res = sub_entries.get(self.curr)?;
Some((Rc::clone(res), self.inline_depth))
Some((Arc::clone(res), self.inline_depth))
}
fn clone(&self) -> Box<dyn LogStream> {
Box::new(Self {
inner: Rc::clone(&self.inner),
inner: Arc::clone(&self.inner),
curr: self.curr,
inline_depth: self.inline_depth,
})
}
fn enclosing_log_entry(&self) -> Option<(Rc<LogEntry>, usize)> {
Some((Rc::clone(&self.inner), self.inline_depth))
fn enclosing_log_entry(&self) -> Option<(Arc<LogEntry>, usize)> {
Some((Arc::clone(&self.inner), self.inline_depth))
}
}
impl IntoLogStream for &Rc<LogEntry> {
impl IntoLogStream for &Arc<LogEntry> {
type Stream = LogEntryStream;
fn from_end(self, inline_depth: usize) -> Option<Self::Stream> {
if let LogEntry::Sub { sub_entries, .. } = self.as_ref() {
Some(LogEntryStream {
inner: Rc::clone(&self),
inner: Arc::clone(&self),
curr: sub_entries.len(),
inline_depth,
})
@ -74,7 +74,7 @@ impl IntoLogStream for &Rc<LogEntry> {
fn from_start(self, inline_depth: usize) -> Option<Self::Stream> {
if let LogEntry::Sub { .. } = self.as_ref() {
Some(LogEntryStream {
inner: Rc::clone(&self),
inner: Arc::clone(&self),
curr: 0,
inline_depth,
})
@ -85,14 +85,14 @@ impl IntoLogStream for &Rc<LogEntry> {
}
pub trait LogStream {
fn next(&mut self) -> Option<(Rc<LogEntry>, usize)>;
fn prev(&mut self) -> Option<(Rc<LogEntry>, usize)>;
fn next(&mut self) -> Option<(Arc<LogEntry>, usize)>;
fn prev(&mut self) -> Option<(Arc<LogEntry>, usize)>;
fn enclosing_log_entry(&self) -> Option<(Rc<LogEntry>, usize)>;
fn enclosing_log_entry(&self) -> Option<(Arc<LogEntry>, usize)>;
fn clone(&self) -> Box<dyn LogStream>;
fn filter(&self, filter: Rc<Filter>) -> FilteredLogStream {
fn filter(&self, filter: Arc<Filter>) -> FilteredLogStream {
FilteredLogStream {
filter: filter,
stack: vec![(self.enclosing_log_entry(), self.clone())],
@ -101,8 +101,8 @@ pub trait LogStream {
}
pub struct FilteredLogStream {
filter: Rc<Filter>,
stack: Vec<(Option<(Rc<LogEntry>, usize)>, Box<dyn LogStream>)>,
filter: Arc<Filter>,
stack: Vec<(Option<(Arc<LogEntry>, usize)>, Box<dyn LogStream>)>,
}
macro_rules! generate_candidate {
@ -140,7 +140,7 @@ macro_rules! generate_filter {
if let Some(iter) = elem.$into_iter(inline_depth + 1) {
$_self
.stack
.push((Some((Rc::clone(&elem), inline_depth)), Box::new(iter)));
.push((Some((Arc::clone(&elem), inline_depth)), Box::new(iter)));
}
// Continue so we actually return a nested item.
if $forwards {
@ -162,26 +162,26 @@ macro_rules! generate_filter {
}
impl FilteredLogStream {
fn next_candidate(&mut self) -> Option<(Rc<LogEntry>, usize)> {
fn next_candidate(&mut self) -> Option<(Arc<LogEntry>, usize)> {
generate_candidate!(self, next, true)
}
fn prev_candidate(&mut self) -> Option<(Rc<LogEntry>, usize)> {
fn prev_candidate(&mut self) -> Option<(Arc<LogEntry>, usize)> {
generate_candidate!(self, prev, false)
}
}
impl LogStream for FilteredLogStream {
fn next(&mut self) -> Option<(Rc<LogEntry>, usize)> {
fn next(&mut self) -> Option<(Arc<LogEntry>, usize)> {
generate_filter!(self, next_candidate, from_start, true)
}
fn prev(&mut self) -> Option<(Rc<LogEntry>, usize)> {
fn prev(&mut self) -> Option<(Arc<LogEntry>, usize)> {
generate_filter!(self, prev_candidate, from_end, false)
}
fn clone(&self) -> Box<dyn LogStream> {
Box::new(Self {
filter: Rc::clone(&self.filter),
filter: Arc::clone(&self.filter),
stack: self
.stack
.iter()
@ -190,7 +190,7 @@ impl LogStream for FilteredLogStream {
})
}
fn enclosing_log_entry(&self) -> Option<(Rc<LogEntry>, usize)> {
fn enclosing_log_entry(&self) -> Option<(Arc<LogEntry>, usize)> {
self.stack[0].0.clone()
}
}

View file

@ -4,8 +4,12 @@ use std::{
io::{self, BufRead, BufReader},
mem,
path::{Path, PathBuf},
rc::Rc,
sync::OnceLock,
sync::Arc,
sync::{
OnceLock,
mpsc::{Receiver, sync_channel},
},
thread,
};
use crate::tui::{
@ -13,32 +17,14 @@ use crate::tui::{
processing::LogStream,
};
struct Inner {
pub path: PathBuf,
struct LogFileEntryGenerator {
file: BufReader<File>,
cached_entries: Vec<Rc<LogEntry>>,
}
#[derive(Clone)]
pub struct LogfileReader(Rc<RefCell<Inner>>);
impl LogfileReader {
pub fn new(p: &Path) -> io::Result<Self> {
Ok(Self(Rc::new(RefCell::new(Inner {
file: BufReader::new(File::open(p)?),
path: p.to_path_buf(),
cached_entries: Vec::new(),
}))))
}
pub fn path(&self) -> PathBuf {
self.0.borrow().path.clone()
}
impl LogFileEntryGenerator {
fn next_line(&mut self) -> Option<String> {
let mut res = String::new();
match self.0.borrow_mut().file.read_line(&mut res) {
match self.file.read_line(&mut res) {
Err(e) => {
eprintln!("error: {e:?}");
None
@ -59,14 +45,14 @@ impl LogfileReader {
}
}
fn next_entry(&mut self) -> Option<Rc<LogEntry>> {
let mut stack = Vec::<(RawLogEntry, Vec<Rc<LogEntry>>)>::new();
let mut curr = Vec::<Rc<LogEntry>>::new();
fn next_entry(&mut self) -> Option<Arc<LogEntry>> {
let mut stack = Vec::<(RawLogEntry, Vec<Arc<LogEntry>>)>::new();
let mut curr = Vec::<Arc<LogEntry>>::new();
loop {
let entry = self.next_raw_entry()?;
let new_entry = Rc::new(match entry.fields.message() {
let new_entry = Arc::new(match entry.fields.message() {
Some("enter") => {
stack.push((entry, mem::take(&mut curr)));
continue;
@ -94,14 +80,49 @@ impl LogfileReader {
}
}
}
}
fn fill_buf_to_access_index(&mut self, n: usize) -> Option<Rc<LogEntry>> {
struct Inner {
pub path: PathBuf,
entry_generator: Receiver<Arc<LogEntry>>,
cached_entries: Vec<Arc<LogEntry>>,
}
#[derive(Clone)]
pub struct LogfileReader(Arc<RefCell<Inner>>);
impl LogfileReader {
pub fn new(p: &Path) -> io::Result<Self> {
let (tx, rx) = sync_channel(1000);
let mut generator = LogFileEntryGenerator {
file: BufReader::new(File::open(p)?),
};
thread::spawn(move || {
while let Some(i) = generator.next_entry() {
tx.send(i).unwrap();
}
});
Ok(Self(Arc::new(RefCell::new(Inner {
path: p.to_path_buf(),
cached_entries: Vec::new(),
entry_generator: rx,
}))))
}
pub fn path(&self) -> PathBuf {
self.0.borrow().path.clone()
}
fn fill_buf_to_access_index(&mut self, n: usize) -> Option<Arc<LogEntry>> {
while self.0.borrow().cached_entries.len() <= n {
let entry = self.next_entry()?;
let entry = self.0.borrow().entry_generator.recv().ok()?;
self.0.borrow_mut().cached_entries.push(entry);
}
Some(Rc::clone(&self.0.borrow().cached_entries[n]))
Some(Arc::clone(&self.0.borrow().cached_entries[n]))
}
pub fn iter(&self) -> LogFileReaderStream {
@ -118,14 +139,14 @@ pub struct LogFileReaderStream {
}
impl LogStream for LogFileReaderStream {
fn next(&mut self) -> Option<(Rc<LogEntry>, usize)> {
fn next(&mut self) -> Option<(Arc<LogEntry>, usize)> {
let entry = self.reader.fill_buf_to_access_index(self.curr)?;
self.curr += 1;
Some((entry, 0))
}
fn prev(&mut self) -> Option<(Rc<LogEntry>, usize)> {
fn prev(&mut self) -> Option<(Arc<LogEntry>, usize)> {
if self.curr == 0 {
return None;
}
@ -143,7 +164,7 @@ impl LogStream for LogFileReaderStream {
})
}
fn enclosing_log_entry(&self) -> Option<(Rc<LogEntry>, usize)> {
fn enclosing_log_entry(&self) -> Option<(Arc<LogEntry>, usize)> {
None
}
}

View file

@ -1,4 +1,4 @@
use std::{cell::OnceCell, rc::Rc};
use std::{cell::OnceCell, sync::Arc};
use itertools::Itertools;
use ratatui::widgets::{List, ListItem, Widget};
@ -19,7 +19,7 @@ pub struct TreeState {
}
impl TreeState {
pub fn from_items(items: &[(Rc<LogEntry>, usize)]) -> Self {
pub fn from_items(items: &[(Arc<LogEntry>, usize)]) -> Self {
let mut res = Vec::new();
let mut curr = String::new();
let mut prev_depth = 0;
@ -88,7 +88,7 @@ impl TreeState {
}
pub struct Items<'a> {
items: Vec<(Rc<LogEntry>, usize)>,
items: Vec<(Arc<LogEntry>, usize)>,
selected_offset: usize,
input_state: &'a InputState,
@ -98,7 +98,7 @@ pub struct Items<'a> {
impl<'a> Items<'a> {
pub fn new(
items: Vec<(Rc<LogEntry>, usize)>,
items: Vec<(Arc<LogEntry>, usize)>,
selected_offset: usize,
input_state: &'a InputState,
selected_footer_field: Option<(String, String)>,
@ -113,7 +113,7 @@ impl<'a> Items<'a> {
}
}
pub fn selected(&self) -> Option<&Rc<LogEntry>> {
pub fn selected(&self) -> Option<&Arc<LogEntry>> {
self.items.get(self.selected_offset).map(|(s, _)| s)
}
}