diff --git a/Build/index.ts b/Build/index.ts index 02d56fa8..7f782409 100644 --- a/Build/index.ts +++ b/Build/index.ts @@ -27,6 +27,7 @@ import { buildDeprecateFiles } from './build-deprecate-files'; import path from 'node:path'; import { ROOT_DIR } from './constants/dir'; import { isCI } from 'ci-info'; +import { printExternalDownloadStats } from './lib/download-stats'; process.on('uncaughtException', (error) => { console.error('Uncaught exception:', error); @@ -123,6 +124,7 @@ const buildFinishedLock = path.join(ROOT_DIR, '.BUILD_FINISHED'); printTraceResult(t); }); printStats(traces); + printExternalDownloadStats(); await Promise.all([ microsoftCdnWorker.end(), diff --git a/Build/lib/create-file.ts b/Build/lib/create-file.ts index 69a7fffb..839c1f68 100644 --- a/Build/lib/create-file.ts +++ b/Build/lib/create-file.ts @@ -10,6 +10,11 @@ import { promisify } from 'node:util'; const fileEqual = createCompareSource(fileEqualWithCommentComparator); +/** + * To keep metadata comment `last updated` not change if real content is the same, + * we compare real content lines and only write if actual content is different, + * and the new `last updated` will be written along with new content. + */ export async function compareAndWriteFile(span: Span, linesA: string[], filePath: string) { // readFileByLine will not include last empty line. So we always pop the linesA for comparison purpose if (linesA.length > 0 && linesA[linesA.length - 1] === '') { diff --git a/Build/lib/download-stats.ts b/Build/lib/download-stats.ts new file mode 100644 index 00000000..1df606ac --- /dev/null +++ b/Build/lib/download-stats.ts @@ -0,0 +1,143 @@ +import picocolors from 'picocolors'; +import { createPrettyBits, prettyBandwidth, prettyTraffic } from 'xbits'; + +export type ExternalDownloadOutcome = 'winner' | 'aborted' | 'failed'; + +export interface ExternalDownloadAttempt { + url: string, + kind: 'primary' | 'fallback', + outcome: ExternalDownloadOutcome, + startedAt: number, + headersAt: number | null, + bodyStartedAt: number | null, + endedAt: number, + bytes: number +} + +export interface ExternalDownloadStatsSnapshot { + startedAt: number | null, + endedAt: number | null, + totalBytes: number, + usefulBytes: number, + wastedBytes: number, + winners: number, + aborted: number, + failed: number +} + +function createEmptySnapshot(): ExternalDownloadStatsSnapshot { + return { + startedAt: null, + endedAt: null, + totalBytes: 0, + usefulBytes: 0, + wastedBytes: 0, + winners: 0, + aborted: 0, + failed: 0 + }; +} + +let stats = createEmptySnapshot(); +const prettyBinaryByteSpeed = createPrettyBits({ + bits: false, + binary: true, + speed: true, + largeK: true +}); + +export function downloadTimestamp() { + return performance.timeOrigin + performance.now(); +} + +export function formatDownloadSpeed(bytes: number, duration: number) { + if (duration <= 0) { + return 'n/a'; + } + const bytesPerSecond = bytes / duration * 1000; + return `${prettyBinaryByteSpeed(bytesPerSecond)} (${prettyBandwidth(bytesPerSecond * 8)})`; +} + +function formatDuration(duration: number | null) { + return duration == null ? 'n/a' : `${duration.toFixed(1)}ms`; +} + +export function recordExternalDownloadAttempt(attempt: ExternalDownloadAttempt) { + stats.startedAt = stats.startedAt == null ? attempt.startedAt : Math.min(stats.startedAt, attempt.startedAt); + stats.endedAt = stats.endedAt == null ? attempt.endedAt : Math.max(stats.endedAt, attempt.endedAt); + stats.totalBytes += attempt.bytes; + + if (attempt.outcome === 'winner') { + stats.winners++; + stats.usefulBytes += attempt.bytes; + } else { + stats.wastedBytes += attempt.bytes; + if (attempt.outcome === 'aborted') { + stats.aborted++; + } else { + stats.failed++; + } + } + + const ttfb = attempt.headersAt == null ? null : attempt.headersAt - attempt.startedAt; + const queueWait = attempt.headersAt == null || attempt.bodyStartedAt == null + ? null + : attempt.bodyStartedAt - attempt.headersAt; + const bodyDuration = attempt.bodyStartedAt == null ? null : attempt.endedAt - attempt.bodyStartedAt; + const speed = bodyDuration == null ? 'n/a' : formatDownloadSpeed(attempt.bytes, bodyDuration); + + console.log( + picocolors.gray('[external download]'), + attempt.kind, + attempt.outcome, + prettyTraffic(attempt.bytes), + speed, + `ttfb=${formatDuration(ttfb)}`, + `queue=${formatDuration(queueWait)}`, + attempt.url + ); +} + +export function mergeExternalDownloadStats(snapshot: ExternalDownloadStatsSnapshot | undefined) { + if (!snapshot) { + return; + } + + if (snapshot.startedAt != null) { + stats.startedAt = stats.startedAt == null ? snapshot.startedAt : Math.min(stats.startedAt, snapshot.startedAt); + } + if (snapshot.endedAt != null) { + stats.endedAt = stats.endedAt == null ? snapshot.endedAt : Math.max(stats.endedAt, snapshot.endedAt); + } + stats.totalBytes += snapshot.totalBytes; + stats.usefulBytes += snapshot.usefulBytes; + stats.wastedBytes += snapshot.wastedBytes; + stats.winners += snapshot.winners; + stats.aborted += snapshot.aborted; + stats.failed += snapshot.failed; +} + +export function takeExternalDownloadStats() { + const snapshot = stats; + stats = createEmptySnapshot(); + return snapshot; +} + +export function printExternalDownloadStats() { + if (stats.startedAt == null || stats.endedAt == null) { + return; + } + + const duration = stats.endedAt - stats.startedAt; + console.log( + picocolors.bold('[external downloads total]'), + `useful=${prettyTraffic(stats.usefulBytes)}`, + `transferred=${prettyTraffic(stats.totalBytes)}`, + `hedge-waste=${prettyTraffic(stats.wastedBytes)}`, + `avg=${formatDownloadSpeed(stats.totalBytes, duration)}`, + `wall=${formatDuration(duration)}`, + `winner=${stats.winners}`, + `aborted=${stats.aborted}`, + `failed=${stats.failed}` + ); +} diff --git a/Build/lib/fetch-assets.ts b/Build/lib/fetch-assets.ts index 4aa63540..7c10e8f3 100644 --- a/Build/lib/fetch-assets.ts +++ b/Build/lib/fetch-assets.ts @@ -8,23 +8,94 @@ import { AdGuardFilterIgnoreUnsupportedLinesStream } from './parse-filter/filter import { appendArrayInPlace } from 'foxts/append-array-in-place'; import { newQueue } from '@henrygd/queue'; -import { AbortError } from 'foxts/abort-error'; +import { AbortError, isAbortErrorLike } from 'foxts/abort-error'; +import { downloadTimestamp, recordExternalDownloadAttempt } from './download-stats'; +import type { ExternalDownloadOutcome } from './download-stats'; const reusedCustomAbortError = new AbortError(); const queue = newQueue(18); +const MIN_HEDGE_DELAY = 3000; +const HEDGE_DELAY_STEP = 1200; +const HEDGE_SPEED_SAMPLE_INTERVAL = 1000; +const HEDGE_QUEUE_POLL_INTERVAL = 250; +// 1 MiB/s is about 8.4 Mbps. A source throttled to 5 Mbps will be hedged, +// while larger responses making healthy progress will not be raced merely +// because their total download time exceeds three seconds. +const MIN_ACCEPTABLE_DOWNLOAD_BYTES_PER_SECOND = 1024 * 1024; + +export function isDownloadThroughputSlow(bytesReceived: number, elapsed: number) { + return bytesReceived / elapsed * 1000 < MIN_ACCEPTABLE_DOWNLOAD_BYTES_PER_SECOND; +} + +interface PrimaryDownloadProgress { + headersReceived: boolean, + bodyStartedAt: number | null, + bytesReceived: number, + failed: boolean +} + export async function fetchAssets( url: string, fallbackUrls: null | undefined | string[] | readonly string[], processLine = false, allowEmpty = false, filterAdGuardUnsupportedLines = false ) { const controller = new AbortController(); + const primaryProgress: PrimaryDownloadProgress = { + headersReceived: false, + bodyStartedAt: null, + bytesReceived: 0, + failed: false + }; + + const waitForSlowPrimary = async (fallbackIndex: number) => { + await waitWithAbort(MIN_HEDGE_DELAY + fallbackIndex * HEDGE_DELAY_STEP, controller.signal); + + let sampledAt: number | null = null; + let sampledBytes = 0; + + while (!controller.signal.aborted) { + // No response headers after the initial delay, or an explicit primary + // failure, is sufficient reason to begin the fallback immediately. + if (!primaryProgress.headersReceived || primaryProgress.failed) { + return; + } + + if (primaryProgress.bodyStartedAt == null) { + // The response is waiting for our local body-consumption queue. This is + // not an upstream slowdown and should not trigger a duplicate request. + // eslint-disable-next-line no-await-in-loop -- poll until local consumption starts + await waitWithAbort(HEDGE_QUEUE_POLL_INTERVAL, controller.signal); + continue; + } + + sampledAt ??= primaryProgress.bodyStartedAt; + const now = performance.now(); + const sampleDuration = now - sampledAt; + + if (sampleDuration < HEDGE_SPEED_SAMPLE_INTERVAL) { + // eslint-disable-next-line no-await-in-loop -- collect a complete throughput sample + await waitWithAbort(HEDGE_SPEED_SAMPLE_INTERVAL - sampleDuration, controller.signal); + continue; + } + + if (isDownloadThroughputSlow(primaryProgress.bytesReceived - sampledBytes, sampleDuration)) { + return; + } + + sampledAt = now; + sampledBytes = primaryProgress.bytesReceived; + // eslint-disable-next-line no-await-in-loop -- periodically re-sample a healthy transfer + await waitWithAbort(HEDGE_SPEED_SAMPLE_INTERVAL, controller.signal); + } + + throw reusedCustomAbortError; + }; const createFetchFallbackPromise = async (url: string, index: number) => { if (index >= 0) { - // To avoid wasting bandwidth, we will wait for a few time before downloading from the fallback URL. try { - await waitWithAbort(1800 + (index + 1) * 1200, controller.signal); + await waitForSlowPrimary(index); } catch { throw reusedCustomAbortError; } @@ -37,28 +108,82 @@ export async function fetchAssets( console.log(picocolors.yellowBright('[fetch fallback begin]'), picocolors.gray(url)); } - // we don't queue add here - const res = await $$fetch(url, { signal: controller.signal, ...defaultRequestInit }); + const isPrimary = index < 0; + const attemptStartedAt = downloadTimestamp(); + let headersAt: number | null = null; + let bodyStartedAt: number | null = null; + let attemptBytes = 0; + let finalized = false; - let stream = nullthrow(res.body, url + ' has an empty body') - .pipeThrough(new TextDecoderStream()) - .pipeThrough(new TextLineStream({ skipEmptyLines: processLine })); - if (processLine) { - stream = stream.pipeThrough(new ProcessLineStream()); + const finalizeAttempt = (outcome: ExternalDownloadOutcome) => { + if (finalized) { + return; + } + finalized = true; + recordExternalDownloadAttempt({ + url, + kind: isPrimary ? 'primary' : 'fallback', + outcome, + startedAt: attemptStartedAt, + headersAt, + bodyStartedAt, + endedAt: downloadTimestamp(), + bytes: attemptBytes + }); + }; + + try { + // We intentionally acquire the body-consumption queue after receiving + // headers. Request scheduling will be handled separately. + const res = await $$fetch(url, { signal: controller.signal, ...defaultRequestInit }); + headersAt = downloadTimestamp(); + + let body = nullthrow(res.body, url + ' has an empty body'); + if (isPrimary) { + primaryProgress.headersReceived = true; + } + body = body.pipeThrough(new TransformStream({ + transform(chunk, streamController) { + attemptBytes += chunk.byteLength; + if (isPrimary) { + primaryProgress.bytesReceived += chunk.byteLength; + } + streamController.enqueue(chunk); + } + })); + + let stream = body + .pipeThrough(new TextDecoderStream()) + .pipeThrough(new TextLineStream({ skipEmptyLines: processLine })); + if (processLine) { + stream = stream.pipeThrough(new ProcessLineStream()); + } + if (filterAdGuardUnsupportedLines) { + stream = stream.pipeThrough(new AdGuardFilterIgnoreUnsupportedLinesStream()); + } + + const arr = await queue.add(() => { + bodyStartedAt = downloadTimestamp(); + if (isPrimary) { + primaryProgress.bodyStartedAt = performance.now(); + } + return Array.fromAsync(stream); + }); + + if (arr.length < 1 && !allowEmpty) { + throw new ResponseError(res, url, 'empty response w/o 304'); + } + + finalizeAttempt('winner'); + controller.abort(); + return arr; + } catch (error) { + if (isPrimary) { + primaryProgress.failed = true; + } + finalizeAttempt(isAbortErrorLike(error) ? 'aborted' : 'failed'); + throw error; } - if (filterAdGuardUnsupportedLines) { - stream = stream.pipeThrough(new AdGuardFilterIgnoreUnsupportedLinesStream()); - } - - // we does queue during downloading - const arr = await queue.add(() => Array.fromAsync(stream)); - - if (arr.length < 1 && !allowEmpty) { - throw new ResponseError(res, url, 'empty response w/o 304'); - } - - controller.abort(); - return arr; }; const primaryPromise = createFetchFallbackPromise(url, -1); diff --git a/Build/trace/index.ts b/Build/trace/index.ts index 0525ecb4..45efe428 100644 --- a/Build/trace/index.ts +++ b/Build/trace/index.ts @@ -3,6 +3,8 @@ import { noop } from 'foxts/noop'; import { basename, extname } from 'node:path'; import process from 'node:process'; import picocolors from 'picocolors'; +import { mergeExternalDownloadStats, takeExternalDownloadStats } from '../lib/download-stats'; +import type { ExternalDownloadStatsSnapshot } from '../lib/download-stats'; export const SPAN_STATUS_START = 0; export const SPAN_STATUS_END = 1; @@ -82,8 +84,9 @@ export function makeSpan(rawSpan: RawSpan): Span { async traceWorkerChild(name: string, factory: (rawSpan: RawSpan) => Promise>): Promise { const childSpan = traceChild(name); - const { result, traceResult, workerTimeOrigin } = await factory(childSpan.rawSpan); + const { result, traceResult, workerTimeOrigin, externalDownloadStats } = await factory(childSpan.rawSpan); mergeWorkerTrace(childSpan, traceResult, workerTimeOrigin); + mergeExternalDownloadStats(externalDownloadStats); childSpan.stop(); return result; } @@ -207,7 +210,8 @@ function mergeWorkerTrace( export interface WorkerJobResult { result: T, traceResult: TraceResult, - workerTimeOrigin: number + workerTimeOrigin: number, + externalDownloadStats: ExternalDownloadStatsSnapshot } /** @@ -231,7 +235,8 @@ export async function workerJob( return { result, traceResult: span.traceResult, - workerTimeOrigin: performance.timeOrigin + workerTimeOrigin: performance.timeOrigin, + externalDownloadStats: takeExternalDownloadStats() }; } diff --git a/package.json b/package.json index 6e04430d..c76627df 100644 --- a/package.json +++ b/package.json @@ -13,6 +13,7 @@ "dexnode": "SWC_NODE_IGNORE_DYNAMIC=true dexnode -r @swc-node/register", "build": "pnpm run node ./Build/index.ts", "build-profile": "pnpm run dexnode -r @swc-node/register ./Build/index.ts", + "bench-download": "pnpm run node ./Build/benchmark-download-client.ts", "lint": "eslint --format=sukka .", "test": "SWC_NODE_IGNORE_DYNAMIC=true SWC_NODE_PROJECT=tsconfig.test.json mocha --require @swc-node/register --watch-extensions ts,tsx" },