Swarm: neighbouring chunks in runs, and a picker that does not rescan (closes #97) 069e16fba3bd

- **The swarm reads images in order (#97).** A fetcher took equally rare chunks at random, so a peer read its whole image in random order, a seek per chunk on a spinning disk: plex-1 got 2.3 GiB from earth-1 at about 2 MB/s. Neighbouring chunks now come in runs, shuffled so runners starting together still spread, and the picker no longer rescans what it has: picking a 142,244-chunk image went from 19 s to 0.35 s.

/// The most neighbouring chunks a fetcher takes in order where rarity does

/// not decide (about 5.6 MB of an image's 88 KB chunks); a small image's

/// runs are shorter, so it still has a few hundred to spread runners over.

const RUN: usize = 64;

    /// Where each chunk sits in `order`.

    pos: Vec<usize>,

    /// No chunk before this place in `order` can be picked fresh: each is

    /// here or asked for. Moved back when one fails.

    skip: usize,

            pos: vec![0; len],

            skip: 0,

    /// Rarest first; among chunks as rare as each other, runs of up to

    /// [`RUN`] neighbours in order, the runs shuffled. A peer reads them out of a

    /// whole image, and on a spinning disk scattered reads cost a seek each

    /// (Earth sent 2.3 GiB in 20 minutes, picked one chunk at a time at

    /// random); runners that start together still spread over the runs.

        let run = (self.len() / 256).clamp(1, RUN);

        let shuffle: Vec<u64> = (0..self.len().div_ceil(run)).map(|_| self.next_rand()).collect();

                keyed.push((self.holders[i], shuffle[i / run], i));

        for (at, &i) in self.order.iter().enumerate() {

            self.pos[i] = at;

        }

        self.skip = 0;

        while self.skip < self.order.len() && (self.have.get(self.order[self.skip]) || self.asked[self.order[self.skip]] > 0) {

            self.skip += 1;

        }

        let fresh = self.order[self.skip..].iter().copied().find(|&i| {

            if !self.have.get(i) {

                self.skip = self.skip.min(self.pos[i]);

            }

    #[test]

    fn from_one_peer_neighbouring_chunks_come_in_order() {

        let n = RUN * 256;

        let mut p = Picker::new(n, Bits::new(n), 3);

        p.peer(0, Bits::full(n));

        let first: Vec<usize> = (0..RUN)

            .map(|_| match p.pick(Source::Peer(0)) {

                Pick::Chunk(i) => i,

                other => panic!("{other:?}"),

            })

            .collect();

        assert_eq!(first[0] % RUN, 0, "a run starts at its first chunk: {first:?}");

        assert!(first.windows(2).all(|w| w[1] == w[0] + 1), "one run, in order: {first:?}");

        // A failed chunk is asked for again before going further.

        p.failed(first[3]);

        assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(first[3]));

    }

    /// A fetch the size of the 14 GB Rust image (142,244 chunks) from one

    /// peer that holds them all: how long picking takes.

    #[test]

    #[ignore]

    fn picking_a_big_image_from_one_peer_is_quick() {

        let n = 142_244;

        let mut p = Picker::new(n, Bits::new(n), 7);

        p.peer(0, Bits::full(n));

        let started = std::time::Instant::now();

        let mut in_flight = std::collections::VecDeque::new();

        loop {

            match p.pick(Source::Peer(0)) {

                Pick::Chunk(i) => {

                    in_flight.push_back(i);

                    if in_flight.len() >= 4 {

                        p.got(in_flight.pop_front().unwrap());

                    }

                }

                Pick::Wait => p.got(in_flight.pop_front().unwrap()),

                Pick::Done => break,

            }

        }

        eprintln!("picked {n} chunks in {:?}", started.elapsed());

    }