diff options
Diffstat (limited to 'packages/meshbay-hub/src/meshbay_hub')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/file-utils.js | 25 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/transport.js | 98 |
2 files changed, 120 insertions, 3 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js index 294455a..585558f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js @@ -48,6 +48,11 @@ const CHUNK_SIZE = 1024 * 1024; // fragmented MP4. A stream that arrives in gulps has no margin for a network // that hesitates, and looks like a hang while it is quiet. const PIPELINE_WINDOW = 8; +// A leased download with at least this much left opens the transport's extra +// connections (`lanes` in transport.js). Below it the handshakes, half a second +// to a second each, cost more than they save; lanes already open are used +// whatever the size. +const LANE_MIN_CHUNKS = 64; // The most this code will ever collect in the page. // @@ -392,10 +397,24 @@ async function pipelinedDownload(transport, gekKey, fileId, totalChunks, onChunk let nextSend = fromChunk, nextRecv = fromChunk; const inflight = new Array(totalChunks); + // Chunk i goes over lane i modulo the lanes open at that moment, two chunks + // in flight per lane so none waits on the slowest. Order is restored below, + // where each chunk is awaited by its index. A lane that fails a request is + // dropped and the chunk asked for again on the main connection, which has + // the retries and the reconnect. const fire = () => { - while (nextSend < totalChunks && nextSend - nextRecv < PIPELINE_WINDOW) { - inflight[nextSend] = _fetchChunkResilient(transport, fileId, nextSend, tr, - signal); + const lanes = (tr && transport.lanes) + ? transport.lanes(totalChunks - nextRecv >= LANE_MIN_CHUNKS) : [transport]; + const depth = Math.max(PIPELINE_WINDOW, 2 * lanes.length); + while (nextSend < totalChunks && nextSend - nextRecv < depth) { + const index = nextSend; + const lane = lanes[index % lanes.length]; + inflight[index] = lane === transport + ? _fetchChunkResilient(transport, fileId, index, tr, signal) + : lane.fetchChunk(fileId, index, tr).catch(() => { + transport.dropLane(lane); + return _fetchChunkResilient(transport, fileId, index, tr, signal); + }); nextSend++; } }; diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index e995ffe..ee6645f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -436,6 +436,25 @@ const JOIN_REFUSALS = { */ const LEASE_WATCHDOG_MS = 60000; +// How many connections a large download is spread over, this one included. +// +// One WebRTC connection carries 5 to 6 MB/s at 22 ms whatever the line can do: +// the browser receives on a UDP socket with a 64 KB kernel buffer, a burst +// overflows it, and the node's SCTP sender halves its window at every loss. +// The ceiling is per connection, so the way past it is more connections, and +// they have to be separate peer connections: several channels on one share +// its socket, its association and its window. Measured against a browser, +// one sender: 1 connection 5 to 6 MB/s, 4 about 19, 8 between 27 and 34, and +// 16 no more than that (the node's sender is one Python process). +const DOWNLOAD_LANES = 8; +// A lane left unused this long is closed: each one is a peer connection the +// node holds open. +const LANE_IDLE_MS = 30000; +// After a lane failed to open, how long before another is tried. `lanes()` is +// asked at every chunk, and without this a node that refuses would be sent an +// offer per chunk. +const LANE_RETRY_MS = 30000; + class Lease { constructor(transport, tr, kind, bytes, chunks, onState) { this.transport = transport; @@ -566,6 +585,15 @@ class MeshBayTransport { this._inFlightUploads = new Set(); // tr → Lease. A transfer's slot on the node, from the client's side. this._leases = new Map(); + // The extra connections a large download is spread over (see `lanes`). + // `_lane` is true on those connections themselves, which open none of + // their own and are closed rather than reconnected. + this._lanes = []; + this._lanesOpening = 0; + this._lanesUsedAt = 0; + this._lanesTimer = 0; + this._laneRetryAt = 0; + this._lane = false; // Set once close() runs — stops the automatic reconnect from firing on a // connection the caller tore down on purpose (leaving the group, page // unload), which would otherwise race back in right as everything else @@ -1433,6 +1461,9 @@ class MeshBayTransport { */ _reconnect() { if (this._closed) return Promise.resolve(); + // A lane that failed is dropped: the download falls back to the main + // connection and `lanes` opens another. + if (this._lane) { this.close(); return Promise.resolve(); } if (!this._reconnectPromise) { this._reconnectPromise = this._reconnectLoop().finally(() => { this._reconnectPromise = null; @@ -1550,6 +1581,72 @@ class MeshBayTransport { return lease; } + /** + * The connections a leased download may read over: this one first, then + * every lane that is open right now. + * + * With `open`, the missing lanes are opened in the background, and they join + * the answer of a later call: a download starts on this connection at once + * and widens as they land. A lane is this connection again (same node, same + * group, same identity, same device), so the node serves a chunk request on + * it under a lease this connection opened (docs/MESHBAY_DESIGN.md §5.5). + * The lease itself, and everything else, stays here. + */ + lanes(open = false) { + if (this._lane || this._closed) return [this]; + for (const lane of this._lanes.filter((l) => !l.connected)) this.dropLane(lane); + if (open || this._lanes.length) { + this._lanesUsedAt = Date.now(); + if (!this._lanesTimer) { + this._lanesTimer = setInterval(() => { + if (Date.now() - this._lanesUsedAt > LANE_IDLE_MS) this._closeLanes(); + }, LANE_IDLE_MS / 2); + } + } + if (open && this._connected && Date.now() >= this._laneRetryAt) { + while (this._lanes.length + this._lanesOpening < DOWNLOAD_LANES - 1) { + this._openLane(); + } + } + return [this, ...this._lanes]; + } + + async _openLane() { + this._lanesOpening += 1; + const lane = new MeshBayTransport(this._hubUrl, this._lastToken); + lane._lane = true; + try { + const args = this._connectArgs; + const token = this._onNeedToken ? await this._onNeedToken() : this._lastToken; + await lane.connect(args.nodeId, token, args.groupId, this._gekRaw, + this._identity, args.bundleKey, args.username, + args.userId, undefined, undefined, + args.joinNodePk, args.nodePk); + // Closed, or gone idle, while this one was negotiating. + if (this._closed || !this._lanesTimer) lane.close(); + else this._lanes.push(lane); + } catch (e) { + console.warn('[MeshBay] could not open a download lane:', e && e.message); + this._laneRetryAt = Date.now() + LANE_RETRY_MS; + lane.close(); + } finally { + this._lanesOpening -= 1; + } + } + + /** Close one lane, e.g. one that failed a request. */ + dropLane(lane) { + const at = this._lanes.indexOf(lane); + if (at >= 0) this._lanes.splice(at, 1); + lane.close(); + } + + _closeLanes() { + clearInterval(this._lanesTimer); + this._lanesTimer = 0; + for (const lane of this._lanes.splice(0)) lane.close(); + } + /** Re-ask for every live lease. Called after a reconnect. */ _reopenTransfers() { for (const lease of this._leases.values()) { @@ -1638,6 +1735,7 @@ class MeshBayTransport { // only skips reconnecting because of this flag, not because "closed" is // absent from its own trigger condition. this._closed = true; + this._closeLanes(); document.removeEventListener('visibilitychange', this._onVisibilityWake); this._wakeReconnect(); if (this._diagCleanup) { this._diagCleanup(); this._diagCleanup = null; } |