diff --git a/src/bun/transfers/engine.ts b/src/bun/transfers/engine.ts index 135e126..3f52825 100644 --- a/src/bun/transfers/engine.ts +++ b/src/bun/transfers/engine.ts @@ -6,12 +6,48 @@ import type { SessionRegistry } from "../fs/registry"; import { walkTree } from "./walk"; const PROGRESS_INTERVAL_MS = 100; +// workers per job; the shared slot budget is the real limit +const MAX_WORKERS_PER_JOB = 8; interface Job { snapshot: TransferSnapshot; type: EntryType; cancelled: boolean; - currentStream: Readable | null; + activeStreams: Set; +} + +/** Counting semaphore with live resize (concurrency setting changes mid-flight). */ +class Semaphore { + private inUse = 0; + private waiters: (() => void)[] = []; + + constructor(private capacity: number) {} + + async acquire(): Promise { + if (this.inUse < this.capacity) { + this.inUse++; + return; + } + // wake() reserves the slot on our behalf before resolving + await new Promise((resolve) => this.waiters.push(resolve)); + } + + release(): void { + this.inUse--; + this.wake(); + } + + resize(capacity: number): void { + this.capacity = capacity; + this.wake(); + } + + private wake(): void { + while (this.waiters.length && this.inUse < this.capacity) { + this.inUse++; + this.waiters.shift()!(); + } + } } const TERMINAL: TransferState[] = ["completed", "failed", "cancelled"]; @@ -39,13 +75,17 @@ export class TransferEngine { private running = 0; private counter = 0; private lastEmit = new Map(); - // serial by default; the transfers setting raises it (jobs are independent) + // serial by default; the transfers setting raises it. The same number limits + // active jobs AND total concurrent file streams (folder jobs parallelize + // internally out of this shared budget). private concurrency = 1; + private slots = new Semaphore(1); constructor(private registry: SessionRegistry) {} setConcurrency(n: number): void { this.concurrency = Math.max(1, Math.min(8, Math.round(n) || 1)); + this.slots.resize(this.concurrency); this.pump(); // a raised limit starts queued jobs immediately } @@ -62,7 +102,7 @@ export class TransferEngine { const job: Job = { type: item.type, cancelled: false, - currentStream: null, + activeStreams: new Set(), snapshot: { id, name: src.basename(item.path), @@ -97,7 +137,7 @@ export class TransferEngine { this.queue = this.queue.filter((j) => j !== job); this.finish(job, "cancelled"); } else { - job.currentStream?.destroy(new Error("transfer cancelled")); + this.destroyStreams(job, new Error("transfer cancelled")); } } @@ -118,7 +158,7 @@ export class TransferEngine { this.queue = this.queue.filter((j) => j !== job); this.finish(job, "failed"); } else { - job.currentStream?.destroy(new Error(reason)); + this.destroyStreams(job, new Error(reason)); } } } @@ -173,30 +213,59 @@ export class TransferEngine { snap.state = "running"; this.emit(job, true); - for (const file of files) { - if (job.cancelled) throw new Error("transfer cancelled"); + // copy with a small worker pool drawing from the SHARED stream budget: + // one folder job can use several slots, several jobs share them fairly + let nextIndex = 0; + let firstError: unknown = null; + + const copyOne = async (file: (typeof files)[number]): Promise => { snap.currentFile = file.name; snap.currentFileBytes = 0; snap.currentFileTotal = file.size; this.emit(job, true); const readStream = src.createReadStream(file.srcPath); - job.currentStream = readStream; + job.activeStreams.add(readStream); readStream.on("data", (chunk: Buffer | string) => { const n = typeof chunk === "string" ? Buffer.byteLength(chunk) : chunk.length; - snap.currentFileBytes += n; snap.doneBytes += n; this.emit(job); }); - await pipeline(readStream, dst.createWriteStream(file.dstPath)); - job.currentStream = null; + try { + await pipeline(readStream, dst.createWriteStream(file.dstPath)); + } finally { + job.activeStreams.delete(readStream); + } snap.doneFiles++; this.emit(job, true); - } + }; + + const worker = async (): Promise => { + while (!job.cancelled && !firstError && nextIndex < files.length) { + const file = files[nextIndex++]; + await this.slots.acquire(); + try { + if (job.cancelled || firstError) return; + await copyOne(file); + } catch (error) { + firstError ??= error; + this.destroyStreams( + job, + error instanceof Error ? error : new Error(String(error)), + ); + } finally { + this.slots.release(); + } + } + }; + + const workerCount = Math.max(1, Math.min(files.length, MAX_WORKERS_PER_JOB)); + await Promise.all(Array.from({ length: workerCount }, worker)); + if (job.cancelled) throw new Error("transfer cancelled"); + if (firstError) throw firstError; this.finish(job, "completed"); } catch (error) { - job.currentStream = null; if (!job.cancelled && !snap.error) { snap.error = error instanceof Error ? error.message : String(error); } @@ -204,6 +273,14 @@ export class TransferEngine { } } + /** Tear down every in-flight stream of a job (cancel, session death, first error). */ + private destroyStreams(job: Job, reason: Error): void { + for (const stream of job.activeStreams) { + stream.destroy(reason); + } + job.activeStreams.clear(); + } + /** Copying onto itself truncates the source; into a descendant never ends. */ private guardAgainstSelfCopy(snap: TransferSnapshot, dstPath: string): void { if (snap.srcSessionId !== snap.dstSessionId) return;