| … | … | @@ -21,7 +21,14 @@ |
| 21 | 21 | |
| 22 | 22 | import { hashChunk } from './chunker.js' |
| 23 | 23 | |
| 24 | | const DEFAULT_CAP = 8 |
| 24 | // P2 — per-peer inflight request window, modelled on hypercore's replicator |
| 25 | // Peer.getMaxInflight (holepunchto/hypercore lib/replicator.js:590). Our content |
| 26 | // chunks are ~64 KB (tier 0 avg 16 KB … tier 3 max 4 MB) — much larger than |
| 27 | // hypercore's blocks — so the range is tuned DOWN from its [16, 512]. A no-udx |
| 28 | // stream (in-memory pair / relay, no RTT signal) gets a modest fixed min*3; a udx |
| 29 | // stream scales the window with RTT and shrinks it as more peers join (fair share). |
| 30 | const INFLIGHT_RANGE = [4, 64] |
| 31 | const SCALE_LATENCY = 50 // ms — at/under this RTT the window stays at min |
| 25 | 32 | // Head-of-line fairness — cede the event loop every YIELD_EVERY accepted chunks |
| 26 | 33 | // so a burst of buffered chunkData frames (protomux dispatches every frame from |
| 27 | 34 | // one socket read synchronously) can't starve other listeners multiplexed on the |
| … | … | @@ -59,7 +66,7 @@ export class ChunkScheduler { |
| 59 | 66 | this.destPath = opts.destPath |
| 60 | 67 | this._transfer = opts.transfer |
| 61 | 68 | this._sendNeed = opts.sendNeed |
| 62 | | this._cap = opts.cap || DEFAULT_CAP |
| 69 | this._cap = opts.cap || 0 // >0 = fixed override; 0 = dynamic RTT-scaled window (P2) |
| 63 | 70 | this._onProgress = opts.onProgress || null |
| 64 | 71 | this._onVerify = opts.onVerify || null |
| 65 | 72 | this._onBaseline = opts.onBaseline || null |
| … | … | @@ -313,13 +320,28 @@ export class ChunkScheduler { |
| 313 | 320 | this._finish() |
| 314 | 321 | } |
| 315 | 322 | |
| 323 | // P2: per-peer inflight window. A fixed opts.cap overrides it (back-compat / tests); |
| 324 | // otherwise scale with the peer's udx RTT and the peer count exactly as hypercore's |
| 325 | // Peer.getMaxInflight — window stays at min for low-RTT links, grows for high-RTT |
| 326 | // ones, and shrinks toward a fair share (×min(1, 2/peers)) as more peers join. |
| 327 | _maxInflight (peer) { |
| 328 | if (this._cap) return this._cap |
| 329 | const [min, max] = INFLIGHT_RANGE |
| 330 | const raw = peer && peer.mux && peer.mux.stream && peer.mux.stream.rawStream |
| 331 | if (!raw || !raw.udx || typeof raw.rtt !== 'number') return Math.min(max, min * 3) |
| 332 | const scale = raw.rtt <= SCALE_LATENCY |
| 333 | ? 1 |
| 334 | : (raw.rtt / SCALE_LATENCY) * Math.min(1, 2 / Math.max(1, this._peers.size)) |
| 335 | return Math.max(min, Math.round(Math.min(max, min * scale))) |
| 336 | } |
| 337 | |
| 316 | 338 | /** Assign needed-but-not-inflight chunks to peers with spare capacity. */ |
| 317 | 339 | _assign () { |
| 318 | 340 | if (this._done || !this._chunks) return |
| 319 | 341 | // Per peer, batch the indices we hand it this round. |
| 320 | 342 | const batches = new Map() |
| 321 | 343 | for (const peer of this._peers) { |
| 322 | | let slots = this._cap - (this._peerInflight.get(peer) || 0) |
| 344 | let slots = this._maxInflight(peer) - (this._peerInflight.get(peer) || 0) |
| 323 | 345 | if (slots <= 0) continue |
| 324 | 346 | for (const index of this._needed) { |
| 325 | 347 | if (slots <= 0) break |