Runner: lock fetches per repository, not the whole runner ed4aef8b70ca

/// Held while a job takes its cache disks and images its workspace (see

/// `run_claim`): one job at a time on the runner's disk and cache pool.

/// Held while the first clone of any repository passes its token through

/// the process environment.

static CLONE: std::sync::Mutex<()> = std::sync::Mutex::new(());

/// One lock per cached repository: fetches into it, and exports from it,

/// never overlap, while jobs of other repositories go on. A slow repository

/// (one kept in a bucket, say) once held every job on the runner waiting

/// behind a single global lock.

fn repo_lock(name: &str) -> std::sync::Arc<std::sync::Mutex<()>> {

    static LOCKS: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<String, std::sync::Arc<std::sync::Mutex<()>>>>> =

        std::sync::OnceLock::new();

    let mut locks = LOCKS.get_or_init(Default::default).lock().unwrap_or_else(|p| p.into_inner());

    locks.entry(name.to_string()).or_default().clone()

}

/// Seconds, to one decimal place, for the job's log.

fn secs(d: std::time::Duration) -> String {

    format!("{:.1} s", d.as_secs_f64())

}

    // Preparing a workspace: fetches into a cached repository, and exports

    // from it, are one job at a time per repository (`repo_lock`); taking

    // cache disks and imaging the workspace are one job at a time on the

    // runner (`PREPARE`); a first clone, whose token goes through the process

    // environment, one at a time (`CLONE`).

    let main_lock = repo_lock(&claim.repo);

    let asked = Instant::now();

    let mut main_guard = Some(main_lock.lock().unwrap_or_else(|p| p.into_inner()));

    if asked.elapsed().as_secs() >= 2 {

        note(&mut sink, format!("waited {} for another job fetching {}", secs(asked.elapsed()), claim.repo));

    }

    let fetching = Instant::now();

    let fetched = fetching.elapsed();

    let mut preparing = None;

        let exporting = Instant::now();

        main_guard.take();

        note(&mut sink, format!("fetched in {}, {} files exported in {}", secs(fetched), files.len(), secs(exporting.elapsed())));

            let sub_lock = repo_lock(&sub.repo);

            let asked = Instant::now();

            let sub_guard = sub_lock.lock().unwrap_or_else(|p| p.into_inner());

            if asked.elapsed().as_secs() >= 2 {

                note(&mut sink, format!("waited {} for another job fetching {}", secs(asked.elapsed()), sub.repo));

            }

            drop(sub_guard);

        let asked = Instant::now();

        preparing = Some(PREPARE.lock().unwrap_or_else(|p| p.into_inner()));

        if asked.elapsed().as_secs() >= 2 {

            note(&mut sink, format!("waited {} for another job imaging its workspace", secs(asked.elapsed())));

        }

    // First job for this repository: clone with the job's token, which goes

    // through the process environment, so one clone at a time.

    let _clone = CLONE.lock().unwrap_or_else(|p| p.into_inner());