a first example of a task that can run in parallel

Author: Byron

gitoxide-core/src/corpus/db.rs

@@ -1,11 +1,12 @@
-use crate::corpus::engine::Id;
-use crate::corpus::Engine;
+use crate::corpus::{Engine, Run};
 use anyhow::{bail, Context};
 use bytesize::ByteSize;
 use rusqlite::{params, OptionalExtension};
 use std::path::{Path, PathBuf};
 use sysinfo::{CpuExt, CpuRefreshKind, RefreshKind, SystemExt};
 
+pub(crate) type Id = u32;
+
 /// a husk of a repository
 pub(crate) struct Repo {
     pub(crate) id: Id,
@@ -110,16 +111,27 @@ pub fn create(path: impl AsRef<std::path::Path>) -> anyhow::Result<rusqlite::Con
     )
     "#,
     )?;
+    con.execute_batch(
+        r#"
+    CREATE TABLE if not exists task(
+        id integer PRIMARY KEY,
+        name text UNIQUE -- the unique name of the task, which is considered its id and which is immutable once run once
+    )
+    "#,
+    )?;
     con.execute_batch(
         r#"
     CREATE TABLE if not exists run(
+        id integer PRIMARY KEY,
         repository integer,
         runner integer,
+        task integer,
         gitoxide_version integer,
-        start_time integer,
-        end_time integer, -- or NULL if not yet finished (either successfull or with failure)
-        error text, -- or NULL if there was on error
+        insertion_time integer NOT NULL, -- in seconds since UNIX epoch
+        duration real, -- in seconds or NULL if not yet finished (either successfull or with failure)
+        error text, -- or NULL if there was no error
         FOREIGN KEY (repository) REFERENCES repository (id),
+        FOREIGN KEY (task) REFERENCES task (id),
         FOREIGN KEY (runner) REFERENCES runner (id),
         FOREIGN KEY (gitoxide_version) REFERENCES gitoxide_version (id)
     )
@@ -163,4 +175,37 @@ impl<P> Engine<P> {
                     |r| r.get(0),
                 )?)
     }
+    pub(crate) fn tasks_or_insert(&self) -> anyhow::Result<Vec<(Id, &'static super::Task)>> {
+        let mut out: Vec<_> = super::run::ALL.iter().map(|task| (0, task)).collect();
+        for (id, task) in out.iter_mut() {
+            *id = self.con.query_row(
+                "INSERT INTO task (name) VALUES (?1) ON CONFLICT DO UPDATE SET name = name RETURNING id",
+                [task.name],
+                |r| r.get(0),
+            )?;
+        }
+        Ok(out)
+    }
+    pub(crate) fn insert_run(
+        con: &rusqlite::Connection,
+        gitoxide_version: Id,
+        runner: Id,
+        task: Id,
+        repository: Id,
+    ) -> anyhow::Result<Run> {
+        let insertion_time = std::time::UNIX_EPOCH.elapsed()?.as_secs();
+        let id = con.query_row("INSERT INTO run (gitoxide_version, runner, task, repository, insertion_time) VALUES (?1, ?2, ?3, ?4, ?5) RETURNING id", params![gitoxide_version, runner, task, repository, insertion_time], |r| r.get(0))?;
+        Ok(Run {
+            id,
+            duration: Default::default(),
+            error: None,
+        })
+    }
+    pub(crate) fn update_run(con: &rusqlite::Connection, run: Run) -> anyhow::Result<()> {
+        con.execute(
+            "UPDATE run SET duration = ?2, error = ?3 WHERE id = ?1",
+            params![run.id, run.duration.as_secs_f64(), run.error.as_deref()],
+        )?;
+        Ok(())
+    }
 }

gitoxide-core/src/corpus/engine.rs

@@ -1,14 +1,13 @@
 use super::db;
-use crate::corpus::Engine;
+use crate::corpus::{Engine, Task};
 use crate::organize::find_git_repository_workdirs;
-use anyhow::Context;
+use anyhow::{bail, Context};
 use bytesize::ByteSize;
+use gix::Progress;
 use rusqlite::params;
 use std::path::{Path, PathBuf};
 use std::time::Instant;
 
-pub(crate) type Id = u32;
-
 impl<P> Engine<P>
 where
     P: gix::Progress,
@@ -29,7 +28,8 @@ where
         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)?;
-        self.perform_run(gitoxide_id, runner_id, repos)
+        let tasks = self.tasks_or_insert()?;
+        self.perform_run(gitoxide_id, runner_id, &tasks, &repos)
     }
 
     pub fn refresh(&mut self, corpus_path: PathBuf) -> anyhow::Result<()> {
@@ -48,17 +48,51 @@ impl<P> Engine<P>
 where
     P: gix::Progress,
 {
-    fn perform_run(&self, _gitoxide_id: Id, _runner_id: Id, _repos: Vec<db::Repo>) -> anyhow::Result<()> {
-        todo!()
+    fn perform_run(
+        &mut self,
+        gitoxide_id: db::Id,
+        runner_id: db::Id,
+        tasks: &[(db::Id, &'static Task)],
+        repos: &[db::Repo],
+    ) -> anyhow::Result<()> {
+        let start = Instant::now();
+        let task_progress = &mut self.progress;
+        task_progress.set_name("run");
+        task_progress.init(Some(tasks.len()), gix::progress::count("tasks"));
+        for (task_id, task) in tasks {
+            let task_start = Instant::now();
+            let mut run_progress = task_progress.add_child(format!("run '{}'", task.name));
+            run_progress.init(Some(repos.len()), gix::progress::count("repos"));
+
+            if task.execute_exclusive {
+                for repo in repos {
+                    if gix::interrupt::is_triggered() {
+                        bail!("interrupted by user");
+                    }
+                    let mut run = Self::insert_run(&self.con, gitoxide_id, runner_id, *task_id, repo.id)?;
+                    task.perform(&mut run, &repo.path);
+                    Self::update_run(&self.con, run)?;
+                    run_progress.inc();
+                }
+            } else {
+                // gix::parallel::in_parallel_with_slice()
+                todo!("shared")
+            }
+
+            run_progress.show_throughput(task_start);
+            task_progress.inc();
+        }
+        task_progress.show_throughput(start);
+        Ok(())
     }
 
-    fn prepare_corpus_path(&self, corpus_path: PathBuf) -> anyhow::Result<(PathBuf, Id)> {
+    fn prepare_corpus_path(&self, corpus_path: PathBuf) -> anyhow::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_id: Id) -> anyhow::Result<Vec<db::Repo>> {
+    fn find_repos(&mut self, corpus_path: &Path, corpus_id: db::Id) -> anyhow::Result<Vec<db::Repo>> {
         self.progress.set_name("query db-repos");
         self.progress.init(None, gix::progress::count("repos"));
 
@@ -68,7 +102,7 @@ where
             .query_map([corpus_id], |r| {
                 Ok(db::Repo {
                     id: r.get(0)?,
-                    path: r.get::<_, String>(1)?.into(),
+                    path: corpus_path.join(r.get::<_, String>(1)?),
                     odb_size: ByteSize(r.get(2)?),
                     num_objects: r.get(3)?,
                     num_references: r.get(4)?,
@@ -78,7 +112,7 @@ where
             .collect::<Result<_, _>>()?)
     }
 
-    fn refresh_repos(&mut self, corpus_path: &Path, corpus_id: Id) -> anyhow::Result<Vec<db::Repo>> {
+    fn refresh_repos(&mut self, corpus_path: &Path, corpus_id: db::Id) -> anyhow::Result<Vec<db::Repo>> {
         let start = Instant::now();
         self.progress.set_name("refresh");
         self.progress.init(None, gix::progress::count("repos"));
@@ -117,7 +151,6 @@ where
                 let write_db = scope.spawn(move || -> anyhow::Result<Vec<db::Repo>> {
                     progress.set_name("write to DB");
                     progress.init(None, gix::progress::count("repos"));
-                    let start = Instant::now();
 
                     let mut out = Vec::new();
                     let transaction = con.transaction()?;
@@ -154,13 +187,12 @@ where
             }
         })?;
 
-        self.progress.show_throughput(start);
         Ok(repos)
     }
 
-    fn find_repos_or_insert(&mut self, corpus_path: &Path, corpus_id: Id) -> anyhow::Result<Vec<db::Repo>> {
+    fn find_repos_or_insert(&mut self, corpus_path: &Path, corpus_id: db::Id) -> anyhow::Result<Vec<db::Repo>> {
         let start = Instant::now();
-        let repos = self.find_repos(corpus_id)?;
+        let repos = self.find_repos(corpus_path, corpus_id)?;
         if repos.is_empty() {
             self.refresh_repos(corpus_path, corpus_id)
         } else {

gitoxide-core/src/corpus/mod.rs

@@ -13,3 +13,60 @@ pub struct RunOutcome {
 
 pub(crate) mod db;
 pub(crate) mod engine;
+
+/// Contains all information necessary to run a task.
+pub(crate) struct Task {
+    /// The unique name of the task, which must not be changed after creating it.
+    ///
+    /// However, if it is changed it will be treated as new kind of task entirely and won't compare
+    /// to previous runs of the task.
+    name: &'static str,
+    /// `true` if the task cannot be run in parallel as it needs all resources by itself.
+    execute_exclusive: bool,
+    /// The actual implementation
+    execute: &'static (dyn run::Execute + Send + Sync),
+}
+
+pub(crate) struct Run {
+    /// Our own ID for finding the respective database row.
+    id: db::Id,
+    /// The time at which the run was inserted.
+    duration: std::time::Duration,
+    error: Option<String>,
+}
+
+pub(crate) mod run {
+    use crate::corpus::{Run, Task};
+    use std::path::Path;
+
+    impl Task {
+        pub fn perform(&self, run: &mut Run, repo: &Path) {
+            let start = std::time::Instant::now();
+            if let Err(err) = self.execute.execute(repo) {
+                run.error = Some(format!("{err:#?}"))
+            }
+            run.duration = start.elapsed();
+        }
+    }
+
+    /// Note that once runs have been recorded, the implementation must not change anymore to keep it comparable.
+    /// If changes have be done, rather change the name of the owning task to start a new kind of task.
+    pub(crate) trait Execute {
+        fn execute(&self, repo: &Path) -> anyhow::Result<()>;
+    }
+
+    pub(crate) static ALL: &'static [Task] = &[Task {
+        name: "open repository (isolated)",
+        execute_exclusive: true, // TODO: false
+        execute: &OpenRepo,
+    }];
+
+    struct OpenRepo;
+
+    impl Execute for OpenRepo {
+        fn execute(&self, repo: &Path) -> anyhow::Result<()> {
+            gix::open_opts(&repo, gix::open::Options::isolated())?;
+            Ok(())
+        }
+    }
+}