use std::{
path::{Path, PathBuf},
sync::atomic::{AtomicUsize, Ordering},
time::{Duration, Instant},
};
use bytesize::ByteSize;
use gix::{
Count, NestedProgress, Progress, Result,
error::{ErrorExt, OptionExt, ResultExt, bail, message, validation},
};
use rusqlite::params;
use super::db;
use crate::{
corpus,
corpus::{Engine, Task},
organize::find_git_repository_workdirs,
};
pub type ProgressItem = gix::progress::DoOrDiscard<gix::progress::prodash::tree::Item>;
pub struct State {
pub progress: ProgressItem,
pub gitoxide_version: String,
pub trace: u8,
pub trace_output: crate::trace::Output,
}
impl Engine {
pub fn open_or_create(db: PathBuf, state: State) -> Result<Engine> {
let con = crate::corpus::db::create(db).or_raise(|| message("Could not open or create database"))?;
Ok(Engine { con, state })
}
pub fn run(
&mut self,
corpus_path: PathBuf,
threads: Option<usize>,
dry_run: bool,
repo_sql_suffix: Option<String>,
allowed_task_names: Vec<String>,
) -> Result<()> {
let tasks = self.tasks_or_insert(&allowed_task_names)?;
if tasks.is_empty() {
bail!(validation("Cannot run without any task to perform on the repositories"));
}
let (corpus_path, corpus_id) = self.prepare_corpus_path(corpus_path)?;
let gitoxide_id = self.gitoxide_version_id_or_insert()?;
let runner_id = self.runner_id_or_insert()?;
let repos = self.find_repos_or_insert(&corpus_path, corpus_id, repo_sql_suffix)?;
self.perform_run(&corpus_path, gitoxide_id, runner_id, &tasks, repos, threads, dry_run)
}
pub fn refresh(&mut self, corpus_path: PathBuf) -> Result<()> {
let (corpus_path, corpus_id) = self.prepare_corpus_path(corpus_path)?;
let repos = self.refresh_repos(&corpus_path, corpus_id)?;
self.state.progress.set_name("refresh repos".into());
self.state.progress.info(format!(
"Added or updated {} repositories under \"{corpus_path}\"",
repos.len(),
corpus_path = corpus_path.display(),
));
Ok(())
}
}
impl Engine {
#[expect(clippy::too_many_arguments)]
fn perform_run(
&mut self,
corpus_path: &Path,
gitoxide_id: db::Id,
runner_id: db::Id,
tasks: &[(db::Id, &'static Task)],
mut repos: Vec<db::Repo>,
threads: Option<usize>,
dry_run: bool,
) -> Result<()> {
let start = Instant::now();
let threads = gix::parallel::num_threads(threads);
let db_path = self.con.path().expect("opened from path on disk").to_owned();
let subscriber = (!dry_run)
.then(|| corpus::trace::subscriber(db_path.as_str(), self.state.trace, self.state.trace_output.clone()))
.transpose()?;
let repo_progress = &mut self.state.progress;
for (task_id, task) in tasks {
let task_start = Instant::now();
let task_info = format!("run '{}'", task.short_name);
repo_progress.set_name(task_info.clone());
repo_progress.init(Some(repos.len()), gix::progress::count("repos"));
if dry_run {
repo_progress.set_name("WOULD run".into());
for repo in &repos {
repo_progress.info(format!(
"{}",
repo.path
.strip_prefix(corpus_path)
.expect("corpus contains repo")
.display()
));
repo_progress.inc();
}
repo_progress.info(format!("with {} tasks", tasks.len()));
for (_, task) in tasks {
repo_progress.info(format!("task '{}' ({})", task.description, task.short_name));
}
break;
}
let subscriber = subscriber.as_ref().expect("dry runs do not execute tasks");
if task.execute_exclusive || threads == 1 {
let mut run_progress = repo_progress.add_child("set later");
let _guard = tracing::dispatcher::set_default(subscriber);
let mut num_errors = 0;
for repo in &repos {
if gix::interrupt::is_triggered() {
bail!(gix::error::cancelled("interrupted by user"));
}
run_progress.set_name(format!(
"{}",
repo.path
.strip_prefix(corpus_path)
.expect("corpus contains repo")
.display()
));
let mut run = Self::insert_run(&self.con, gitoxide_id, runner_id, *task_id, repo.id)?;
tracing::info_span!(parent: None, "run", run_id = run.id).in_scope(|| {
task.perform(
&mut run,
&repo.path,
&mut run_progress,
Some(threads),
&gix::interrupt::IS_INTERRUPTED,
);
});
if let Some(err) = run.error.as_deref() {
num_errors += 1;
repo_progress.fail(err.to_owned());
}
Self::update_run(&self.con, run)?;
repo_progress.inc();
}
repo_progress.show_throughput(task_start);
if num_errors != 0 {
repo_progress.fail(format!(
"{} repositories failed to run task {}",
num_errors, task.short_name
));
}
} else {
let counter = repo_progress.counter();
let num_errors = AtomicUsize::default();
let repo_progress =
gix::parallel::OwnShared::new(gix::parallel::Mutable::new(repo_progress.add_child("in parallel")));
gix::parallel::in_parallel_with_slice(
&mut repos,
Some(threads),
{
let shared_repo_progress = repo_progress.clone();
let db_path = db_path.clone();
move |tid| {
let mut progress = gix::parallel::lock(&shared_repo_progress);
let lane_progress = progress.add_child(format!("{tid}"));
let guard = tracing::dispatcher::set_default(subscriber);
(guard, lane_progress, rusqlite::Connection::open(&db_path))
}
},
|repo, (_guard, progress, con), _threads_left, should_interrupt| -> Result<()> {
progress.set_name(format!(
"{}",
repo.path
.strip_prefix(corpus_path)
.expect("corpus contains repo")
.display()
));
let con = match con {
Ok(con) => con,
Err(err) => {
progress.fail(format!("{err:#?}"));
should_interrupt.store(true, Ordering::SeqCst);
return Ok(());
}
};
let mut run = Self::insert_run(con, gitoxide_id, runner_id, *task_id, repo.id)?;
tracing::info_span!(parent: None, "run", run_id = run.id).in_scope(|| {
task.perform(&mut run, &repo.path, progress, Some(1), should_interrupt);
});
if let Some(err) = run.error.as_deref() {
num_errors.fetch_add(1, Ordering::Relaxed);
progress.fail(err.to_owned());
}
Self::update_run(con, run)?;
counter.fetch_add(1, Ordering::Relaxed);
Ok(())
},
|| (!gix::interrupt::is_triggered()).then(|| Duration::from_millis(100)),
drop,
)?;
let repo_progress = gix::parallel::lock(&repo_progress);
repo_progress.show_throughput(task_start);
let num_errors = num_errors.load(Ordering::Relaxed);
if num_errors != 0 {
repo_progress.fail(format!(
"{} repositories failed to run task {}",
num_errors, task.short_name
));
}
}
repo_progress.inc();
}
repo_progress.show_throughput(start);
Ok(())
}
fn prepare_corpus_path(&self, corpus_path: PathBuf) -> Result<(PathBuf, db::Id)> {
let corpus_path = gix::path::realpath(corpus_path)?;
let corpus_id = self.corpus_id_or_insert(&corpus_path)?;
Ok((corpus_path, corpus_id))
}
fn find_repos(&mut self, corpus_path: &Path, corpus_id: db::Id, sql_suffix: Option<&str>) -> Result<Vec<db::Repo>> {
self.state.progress.set_name("query db-repos".into());
self.state.progress.init(None, gix::progress::count("repos"));
self.con
.prepare(&format!(
"SELECT id, rela_path, odb_size, num_objects, num_references FROM repository WHERE corpus = ?1 {}",
sql_suffix.unwrap_or_default()
))
.or_error()?
.query_map([corpus_id], |r| {
Ok(db::Repo {
id: r.get(0)?,
path: corpus_path.join(r.get::<_, String>(1)?),
odb_size: ByteSize(r.get(2)?),
num_objects: r.get(3)?,
num_references: r.get(4)?,
})
})
.or_error()?
.inspect(|_| self.state.progress.inc())
.collect::<std::result::Result<_, _>>()
.or_error()
}
fn refresh_repos(&mut self, corpus_path: &Path, corpus_id: db::Id) -> Result<Vec<db::Repo>> {
let start = Instant::now();
self.state.progress.set_name("refresh".into());
self.state.progress.init(None, gix::progress::count("repos"));
let repos = std::thread::scope({
let progress = &mut self.state.progress;
let con = &mut self.con;
|scope| -> Result<_> {
let threads = std::thread::available_parallelism()
.map(std::num::NonZeroUsize::get)
.ok()
.unwrap_or(1);
let (path_tx, repo_rx) = {
let (path_tx, path_rx) = crossbeam_channel::bounded(threads * 2);
let (repo_tx, repo_rx) = std::sync::mpsc::channel::<(PathBuf, Result<db::Repo>)>();
(0..threads).for_each(|_| {
scope.spawn(gix::trace::in_thread({
let path_rx = path_rx.clone();
let repo_tx = repo_tx.clone();
move || -> Result<_> {
for repo_path in path_rx {
let res = (|| {
let repo = gix::open_opts(&repo_path, gix::open::Options::isolated())?;
db::Repo::try_from(&repo)
})();
repo_tx.send((repo_path, res)).or_error()?;
}
Ok(())
}
}));
});
(path_tx, repo_rx)
};
let find_progress = progress.add_child("find");
let write_db = scope.spawn(gix::trace::in_thread(move || -> Result<Vec<db::Repo>> {
progress.set_name("write to DB".into());
progress.init(None, gix::progress::count("repos"));
let mut out = Vec::new();
let transaction = con.transaction().or_error()?;
let mut statement = transaction.prepare("INSERT INTO repository (rela_path, corpus, odb_size, num_objects, num_references) VALUES (?1, ?2, ?3, ?4, ?5)\
ON CONFLICT DO UPDATE SET rela_path = rela_path, corpus = corpus, odb_size = ?3, num_objects = ?4, num_references = ?5\
RETURNING id").or_error()?;
for (repo_path, repo_res) in repo_rx {
match repo_res {
Ok(mut repo) => {
let rela_path = repo.path.strip_prefix(corpus_path).or_error()?;
repo.id = statement.query_row(params![rela_path.to_str().ok_or_raise(|| gix::error::validation("only valid UTF8 is allowed for repository paths"))?, corpus_id, repo.odb_size.as_u64(), repo.num_objects, repo.num_references], |r| r.get(0)).or_error()?;
out.push(repo);
progress.inc();
}
Err(err) => progress.fail(format!("{repo_path}: {err:#?}", repo_path = repo_path.display())),
}
}
statement.finalize().or_error()?;
transaction.commit().or_error()?;
progress.show_throughput(start);
Ok(out)
}));
let repos = gix::interrupt::Iter::new(
find_git_repository_workdirs(corpus_path, find_progress, false, Some(threads)),
|| gix::error::cancelled("interrupted by user").raise(),
);
for res in repos {
let (repo_path, _kind) = res?;
path_tx.send(repo_path).or_error()?;
}
drop(path_tx);
write_db.join().expect("no panic")
}
})?;
Ok(repos)
}
fn find_repos_or_insert(
&mut self,
corpus_path: &Path,
corpus_id: db::Id,
sql_suffix: Option<String>,
) -> Result<Vec<db::Repo>> {
let start = Instant::now();
let repos = self.find_repos(corpus_path, corpus_id, sql_suffix.as_deref())?;
if repos.is_empty() {
let res = self.refresh_repos(corpus_path, corpus_id);
if sql_suffix.is_some() {
self.find_repos(corpus_path, corpus_id, sql_suffix.as_deref())
} else {
res
}
} else {
self.state.progress.show_throughput(start);
Ok(repos)
}
}
}