indicatrix_dispatch/pool/
feed.rs

1//! Lanes that join a run already in progress: [`LaneFeed`] and the thread that polls it.
2//!
3//! A late lane claims from the same [`crate::SampleCursor`] as the lanes that started
4//! with the run (a requeued range first, fresh samples after) and merges into the same
5//! [`crate::Merger`], so it takes part in the epoch exactly like any other lane. The
6//! merger folds chunks in ascending `first_sample` order whichever lane traced them, so a
7//! lane's start time cannot change which samples end up in the image, only which lane
8//! traced which chunk.
9
10use super::{LaneSlot, epoch::Epoch, lane_loop};
11use crate::{RateModel, WorkerLane};
12use std::{
13    sync::{Arc, Mutex},
14    thread,
15    time::Duration,
16};
17
18/// How long the feeder thread waits inside one [`LaneFeed::next`] call: also the longest a
19/// finished run waits for the feeder to notice it is over.
20const FEED_WAIT: Duration = Duration::from_millis(50);
21
22/// A source of lanes that join a run in progress (see
23/// [`LanePool::run_into_fed`](super::LanePool::run_into_fed)).
24pub trait LaneFeed: Sync {
25    /// Waits up to `wait` for lanes to add and returns them with their starting rate
26    /// models, or an empty list. The first returned lane gets the event index
27    /// `first_index`, the next one `first_index + 1`, and so on, so an implementation
28    /// can set up whatever it keys by lane index before the lane starts.
29    ///
30    /// Called repeatedly from one thread until the run ends; lanes returned after the
31    /// last lane has left are dropped unused.
32    fn next(&self, first_index: usize, wait: Duration) -> Vec<(Arc<dyn WorkerLane>, RateModel)>;
33}
34
35/// Starts lane `index` on its own scoped thread; the roster counts it out when it ends.
36pub(super) fn spawn_lane<'scope, 'env>(
37    scope: &'scope thread::Scope<'scope, 'env>,
38    epoch: &'env Epoch<'env>,
39    index: usize,
40    slot: Arc<LaneSlot>,
41) {
42    scope.spawn(move || {
43        let left = lane_loop::run_lane(epoch, index, &*slot.lane, &slot.rate);
44        if !left {
45            epoch.roster.leave();
46        }
47    });
48}
49
50/// The feeder thread's body: asks `feed` for lanes until the run ends or is cancelled and
51/// starts each one that arrives while the roster is still open.
52pub(super) fn feed_lanes<'scope, 'env>(
53    scope: &'scope thread::Scope<'scope, 'env>,
54    epoch: &'env Epoch<'env>,
55    feed: &'env dyn LaneFeed,
56) {
57    while !epoch.roster.is_closed() && !epoch.cancel.is_cancelled() {
58        let late = feed.next(epoch.lane_count(), FEED_WAIT);
59        for (lane, rate) in late {
60            if !epoch.roster.try_join() {
61                return;
62            }
63            let slot = Arc::new(LaneSlot {
64                lane,
65                rate: Mutex::new(rate),
66            });
67            let index = epoch.push_lane(Arc::clone(&slot));
68            spawn_lane(scope, epoch, index, slot);
69        }
70    }
71}
indicatrix_dispatch/pool/feed.rs

indicatrix_dispatch/pool/
feed.rs
