|
|
… |
16 unchanged lines |
| 17 |
17 |
|
pub const TICKET_LIFE_MS: u64 = 15 * 60 * 1000; |
| 18 |
18 |
|
/// Peers make no progress for this long and the hub is asked for anything. |
| 19 |
19 |
|
pub const STALL: Duration = Duration::from_secs(2); |
|
20 |
+ |
/// The most neighbouring chunks a fetcher takes in order where rarity does |
|
21 |
+ |
/// not decide (about 5.6 MB of an image's 88 KB chunks); a small image's |
|
22 |
+ |
/// runs are shorter, so it still has a few hundred to spread runners over. |
|
23 |
+ |
const RUN: usize = 64; |
| 20 |
24 |
|
/// Most sources asked for the same chunk in the endgame. |
| 21 |
25 |
|
const ENDGAME_COPIES: u8 = 2; |
| 22 |
26 |
|
|
|
|
… |
137 unchanged lines |
| 160 |
164 |
|
peers: HashMap<usize, (Bits, Bits)>, |
| 161 |
165 |
|
/// Missing chunks, rarest first; rebuilt when the peers' maps change. |
| 162 |
166 |
|
order: Vec<usize>, |
|
167 |
+ |
/// Where each chunk sits in `order`. |
|
168 |
+ |
pos: Vec<usize>, |
|
169 |
+ |
/// No chunk before this place in `order` can be picked fresh: each is |
|
170 |
+ |
/// here or asked for. Moved back when one fails. |
|
171 |
+ |
skip: usize, |
| 163 |
172 |
|
/// How many peers hold each chunk, as of the last rebuild. |
| 164 |
173 |
|
holders: Vec<u16>, |
| 165 |
174 |
|
/// Whether some peer is fetching each chunk now: the hub is not asked |
|
|
… |
12 unchanged lines |
| 178 |
187 |
|
first: vec![None; len], |
| 179 |
188 |
|
peers: HashMap::new(), |
| 180 |
189 |
|
order: Vec::new(), |
|
190 |
+ |
pos: vec![0; len], |
|
191 |
+ |
skip: 0, |
| 181 |
192 |
|
holders: vec![0; len], |
| 182 |
193 |
|
coming: vec![false; len], |
| 183 |
194 |
|
progress: Instant::now(), |
|
|
… |
45 unchanged lines |
| 229 |
240 |
|
self.seed |
| 230 |
241 |
|
} |
| 231 |
242 |
|
|
|
243 |
+ |
/// Rarest first; among chunks as rare as each other, runs of up to |
|
244 |
+ |
/// [`RUN`] neighbours in order, the runs shuffled. A peer reads them out of a |
|
245 |
+ |
/// whole image, and on a spinning disk scattered reads cost a seek each |
|
246 |
+ |
/// (Earth sent 2.3 GiB in 20 minutes, picked one chunk at a time at |
|
247 |
+ |
/// random); runners that start together still spread over the runs. |
| 232 |
248 |
|
fn reorder(&mut self) { |
|
249 |
+ |
let run = (self.len() / 256).clamp(1, RUN); |
|
250 |
+ |
let shuffle: Vec<u64> = (0..self.len().div_ceil(run)).map(|_| self.next_rand()).collect(); |
| 233 |
251 |
|
let mut keyed: Vec<(u16, u64, usize)> = Vec::new(); |
| 234 |
252 |
|
for i in 0..self.len() { |
| 235 |
253 |
|
self.holders[i] = self.peers.values().filter(|(b, _)| b.get(i)).count().min(u16::MAX as usize) as u16; |
| 236 |
254 |
|
self.coming[i] = self.peers.values().any(|(_, f)| f.get(i)); |
| 237 |
255 |
|
if !self.have.get(i) { |
| 238 |
|
- |
let r = self.next_rand(); |
| 239 |
|
- |
keyed.push((self.holders[i], r, i)); |
|
256 |
+ |
keyed.push((self.holders[i], shuffle[i / run], i)); |
| 240 |
257 |
|
} |
| 241 |
258 |
|
} |
| 242 |
259 |
|
keyed.sort_unstable(); |
| 243 |
260 |
|
self.order = keyed.into_iter().map(|(_, _, i)| i).collect(); |
|
261 |
+ |
for (at, &i) in self.order.iter().enumerate() { |
|
262 |
+ |
self.pos[i] = at; |
|
263 |
+ |
} |
|
264 |
+ |
self.skip = 0; |
| 244 |
265 |
|
} |
| 245 |
266 |
|
|
| 246 |
267 |
|
/// The next chunk for `src` to fetch. |
|
|
… |
8 unchanged lines |
| 255 |
276 |
|
let stalled = self.progress.elapsed() >= STALL; |
| 256 |
277 |
|
// Not yet asked for: a peer takes the rarest it holds; the hub |
| 257 |
278 |
|
// takes what no peer holds, or anything once the peers stall. |
| 258 |
|
- |
let fresh = self.order.iter().copied().find(|&i| { |
|
279 |
+ |
while self.skip < self.order.len() && (self.have.get(self.order[self.skip]) || self.asked[self.order[self.skip]] > 0) { |
|
280 |
+ |
self.skip += 1; |
|
281 |
+ |
} |
|
282 |
+ |
let fresh = self.order[self.skip..].iter().copied().find(|&i| { |
| 259 |
283 |
|
!self.have.get(i) |
| 260 |
284 |
|
&& self.asked[i] == 0 |
| 261 |
285 |
|
&& holds(i) |
|
|
… |
32 unchanged lines |
| 294 |
318 |
|
self.asked[i] = self.asked[i].saturating_sub(1); |
| 295 |
319 |
|
if self.asked[i] == 0 { |
| 296 |
320 |
|
self.first[i] = None; |
|
321 |
+ |
if !self.have.get(i) { |
|
322 |
+ |
self.skip = self.skip.min(self.pos[i]); |
|
323 |
+ |
} |
| 297 |
324 |
|
} |
| 298 |
325 |
|
} |
| 299 |
326 |
|
} |
| 300 |
327 |
|
|
| 301 |
328 |
|
#[cfg(test)] |
| 302 |
329 |
|
mod tests { |
|
330 |
+ |
#[test] |
|
331 |
+ |
fn from_one_peer_neighbouring_chunks_come_in_order() { |
|
332 |
+ |
let n = RUN * 256; |
|
333 |
+ |
let mut p = Picker::new(n, Bits::new(n), 3); |
|
334 |
+ |
p.peer(0, Bits::full(n)); |
|
335 |
+ |
let first: Vec<usize> = (0..RUN) |
|
336 |
+ |
.map(|_| match p.pick(Source::Peer(0)) { |
|
337 |
+ |
Pick::Chunk(i) => i, |
|
338 |
+ |
other => panic!("{other:?}"), |
|
339 |
+ |
}) |
|
340 |
+ |
.collect(); |
|
341 |
+ |
assert_eq!(first[0] % RUN, 0, "a run starts at its first chunk: {first:?}"); |
|
342 |
+ |
assert!(first.windows(2).all(|w| w[1] == w[0] + 1), "one run, in order: {first:?}"); |
|
343 |
+ |
// A failed chunk is asked for again before going further. |
|
344 |
+ |
p.failed(first[3]); |
|
345 |
+ |
assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(first[3])); |
|
346 |
+ |
} |
|
347 |
+ |
|
|
348 |
+ |
/// A fetch the size of the 14 GB Rust image (142,244 chunks) from one |
|
349 |
+ |
/// peer that holds them all: how long picking takes. |
|
350 |
+ |
#[test] |
|
351 |
+ |
#[ignore] |
|
352 |
+ |
fn picking_a_big_image_from_one_peer_is_quick() { |
|
353 |
+ |
let n = 142_244; |
|
354 |
+ |
let mut p = Picker::new(n, Bits::new(n), 7); |
|
355 |
+ |
p.peer(0, Bits::full(n)); |
|
356 |
+ |
let started = std::time::Instant::now(); |
|
357 |
+ |
let mut in_flight = std::collections::VecDeque::new(); |
|
358 |
+ |
loop { |
|
359 |
+ |
match p.pick(Source::Peer(0)) { |
|
360 |
+ |
Pick::Chunk(i) => { |
|
361 |
+ |
in_flight.push_back(i); |
|
362 |
+ |
if in_flight.len() >= 4 { |
|
363 |
+ |
p.got(in_flight.pop_front().unwrap()); |
|
364 |
+ |
} |
|
365 |
+ |
} |
|
366 |
+ |
Pick::Wait => p.got(in_flight.pop_front().unwrap()), |
|
367 |
+ |
Pick::Done => break, |
|
368 |
+ |
} |
|
369 |
+ |
} |
|
370 |
+ |
eprintln!("picked {n} chunks in {:?}", started.elapsed()); |
|
371 |
+ |
} |
|
372 |
+ |
|
| 303 |
373 |
|
use super::*; |
| 304 |
374 |
|
|
| 305 |
375 |
|
#[test] |
|
|
… |
125 unchanged lines |