| … | … | @@ -672,21 +672,37 @@ export class OverlayProtocol { |
| 672 | 672 | // _onChunkNeed, so exactly one job streams at a time per peer. |
| 673 | 673 | async _serveJob (peer, job) { |
| 674 | 674 | const { path, diskPath, chunkMap, indices } = job |
| 675 | | for (let i = 0; i < indices.length; i++) { |
| 676 | | const index = indices[i] |
| 677 | | if (index >= chunkMap.length) continue |
| 678 | | const c = chunkMap[index] |
| 679 | | const data = this._transferManager.readChunk(diskPath, c.offset, c.length) |
| 680 | | if (!data) continue |
| 681 | | const flushed = peer.msgs.chunkData.send({ path, index, data }) |
| 682 | | if (this._chunkServeCb) { |
| 683 | | try { this._chunkServeCb({ path, index, bytes: data.length, peer, from: peer.authorizedServe.get(path) || null }) } catch {} |
| 684 | | } |
| 685 | | if (flushed === false && i < indices.length - 1) { |
| 686 | | const alive = await this._waitForDrain(peer) |
| 687 | | if (!alive) return |
| 688 | | if (this._serveAuthorizer && !peer.authorizedServe.has(path)) return |
| 675 | // P3: cork the mux so a run of chunkData frames coalesces into fewer stream |
| 676 | // writes (hypercore's replicator corks around each request/response batch, |
| 677 | // replicator.js:722/738). Uncork before parking on drain (flush the batch) and |
| 678 | // in the finally. The `corked` guard keeps cork/uncork balanced across the early |
| 679 | // returns, and no-ops when the mux stub has no cork (unit tests). |
| 680 | const mux = peer.mux |
| 681 | let corked = false |
| 682 | const cork = () => { if (mux && mux.cork && !corked) { mux.cork(); corked = true } } |
| 683 | const uncork = () => { if (mux && mux.uncork && corked) { mux.uncork(); corked = false } } |
| 684 | cork() |
| 685 | try { |
| 686 | for (let i = 0; i < indices.length; i++) { |
| 687 | const index = indices[i] |
| 688 | if (index >= chunkMap.length) continue |
| 689 | const c = chunkMap[index] |
| 690 | const data = this._transferManager.readChunk(diskPath, c.offset, c.length) |
| 691 | if (!data) continue |
| 692 | const flushed = peer.msgs.chunkData.send({ path, index, data }) |
| 693 | if (this._chunkServeCb) { |
| 694 | try { this._chunkServeCb({ path, index, bytes: data.length, peer, from: peer.authorizedServe.get(path) || null }) } catch {} |
| 695 | } |
| 696 | if (flushed === false && i < indices.length - 1) { |
| 697 | uncork() // flush the corked batch before we park on drain |
| 698 | const alive = await this._waitForDrain(peer) |
| 699 | if (!alive) return |
| 700 | if (this._serveAuthorizer && !peer.authorizedServe.has(path)) return |
| 701 | cork() // re-cork for the next run |
| 702 | } |
| 689 | 703 | } |
| 704 | } finally { |
| 705 | uncork() |
| 690 | 706 | } |
| 691 | 707 | } |
| 692 | 708 | |