Leahnaya/sendou.ink
1
0
Fork 0
mirror of https://github.com/Sendouc/sendou.ink.git synced 2026-11-15 01:12:36 -06:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Make Scanner more resilient for web gpu worker failures

Browse Source
This commit is contained in:
Kalle
2026-10-04 17:04:12 +03:00
parent 1b86201888
commit 36d9bdaa26
5 changed files with 149 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
app/features/scanner/components/ScanWorkers.tsx
View File

@@ -4,7 +4,7 @@
* skimming past dead air, and how many games it has found.
*/
import clsx from "clsx";
import { Check } from "lucide-react";
import { Check, X } from "lucide-react";
import { formatPosition } from "../core/format";
import { buildScannerMatches } from "../core/match-builder";
import styles from "./ScanWorkers.module.css";
@@ -101,7 +101,11 @@ function LaneTile({
[styles.modeActive]: !lane.done && lane.mode === "active",
})}
>
{lane.done ? (
{lane.failed ? (
<>
<X size={12} /> Gave up
</>
) : lane.done ? (
<>
<Check size={12} /> Done
</>

12
app/features/scanner/components/vod-scan.ts
View File

@@ -24,6 +24,7 @@ import {
type ScanTelemetry,
} from "../core/detectors/telemetry";
import type { DetectedEvent } from "../core/detectors/types";
import { formatPosition } from "../core/format";
import {
buildScannerMatches,
invalidObjectiveEvents,
@@ -87,6 +88,8 @@ export interface VodScanLane {
t: number;
mode: VodScanMode;
done: boolean;
/** the lane gave up at `t`; the rest of its slice went unscanned */
failed: boolean;
}
export interface VodScanProgress {
@@ -320,6 +323,7 @@ export async function startVodScan(
t: i * span,
mode: "active" as VodScanMode,
done: false,
failed: false,
telemetry: null as ScanTelemetry | null,
}));
abortChunks = () => {
@@ -350,6 +354,7 @@ export async function startVodScan(
t: Math.min(c.t, c.tEnd),
mode: c.mode,
done: c.done,
failed: c.failed,
})),
},
telemetry: mergedTelemetry(),
@@ -380,7 +385,11 @@ export async function startVodScan(
},
(error) => {
chunk.done = true;
update({ error: describeError(error) });
chunk.failed = true;
pushUiUpdate({ force: true });
update({
error: `Worker ${chunkIndex + 1} gave up, ${formatPosition(Math.min(chunk.t, chunk.tEnd))} – ${formatPosition(chunk.tEnd)} was not scanned (${describeError(error)})`,
});
},
),
),
@@ -429,6 +438,7 @@ export async function startVodScan(
t,
mode: seek.doneInfo?.calm ? "skim" : "active",
done: false,
failed: false,
},
],
},

170
app/features/scanner/worker/analyzer.worker.ts
View File

@@ -65,6 +65,16 @@ const PROGRESS_POST_INTERVAL_MS = 400;
const PREVIEW_POST_INTERVAL_MS = 600;
const PREVIEW_WIDTH = 480;
const PREVIEW_HEIGHT = 270;
/**
* Failures (decode errors, a crashed GPU process taking the hardware decoder
* down) a chunk scan recovers from by resuming at its cursor before it gives
* up on the rest of its slice.
*/
const MAX_CHUNK_FAILURES = 10;
/** A crashed GPU process takes a moment to come back. */
const CHUNK_RETRY_DELAY_MS = 1000;
/** Footage hopped over when a spot fails to decode even in software. */
const UNDECODABLE_SKIP_S = 5;
/** A read-back frame normalized to the canonical size, its capture pixels kept for the stored frame. */
interface PreparedFrame {
@@ -288,6 +298,10 @@ async function scanChunk({
let lastPreviewAt = 0;
let cursor = tStart;
let mode: "active" | "skim" = "active";
let hardwareAcceleration: HardwareAcceleration = "no-preference";
let failures = 0;
let failuresHere = 0;
let failedAt = Number.NEGATIVE_INFINITY;
const input = new Input({
formats: ALL_FORMATS,
@@ -298,7 +312,7 @@ async function scanChunk({
if (!track || !(await track.canDecode())) {
throw new Error("worker cannot decode this file");
}
const samples = new VideoSampleSink(track);
let samples = new VideoSampleSink(track);
const packets = new EncodedPacketSink(track);
frameReader ??= createFrameReader(readFrame);
@@ -372,11 +386,18 @@ async function scanChunk({
* likely due read back in the helper worker.
*/
const scanActive = async (): Promise<"calm" | "end"> => {
const source = openFrameSource(await decodeWorker(), file, cursor, tEnd, {
intervalMs: PREVIEW_POST_INTERVAL_MS,
width: PREVIEW_WIDTH,
height: PREVIEW_HEIGHT,
});
const source = openFrameSource(
await decodeWorker(),
file,
cursor,
tEnd,
{
intervalMs: PREVIEW_POST_INTERVAL_MS,
width: PREVIEW_WIDTH,
height: PREVIEW_HEIGHT,
},
hardwareAcceleration,
);
let upcoming: Promise<SourceItem | null> | null = null;
let exhausted = false;
/** decoded, stream order, steps not taken yet */
@@ -439,30 +460,38 @@ async function scanChunk({
wake();
},
);
while (!finished) {
for (const item of queue) if (item.t < bound) release(item);
const held = queue.filter((item) => item.frame || item.read);
if (!held.some((item) => item.read)) {
const likely = held.find((item) => item.t >= guess);
if (likely) readAhead(likely);
try {
while (!finished) {
for (const item of queue) if (item.t < bound) release(item);
const held = queue.filter((item) => item.frame || item.read);
if (!held.some((item) => item.read)) {
const likely = held.find((item) => item.t >= guess);
if (likely) readAhead(likely);
}
const pulling =
!exhausted &&
held.length < MAX_OPEN_FRAMES &&
!held.some((item) => item.read) &&
queue.every((item) => item.t < tEnd);
const woken = new Promise<void>((resolve) => {
wake = resolve;
});
if (!pulling) {
await woken;
continue;
}
upcoming ??= source.next();
if (
(await Promise.race([upcoming.then(() => true), woken])) !== true
)
continue;
const pulled = await next();
if (pulled) queue.push(pulled);
}
const pulling =
!exhausted &&
held.length < MAX_OPEN_FRAMES &&
!held.some((item) => item.read) &&
queue.every((item) => item.t < tEnd);
const woken = new Promise<void>((resolve) => {
wake = resolve;
});
if (!pulling) {
await woken;
continue;
}
upcoming ??= source.next();
if ((await Promise.race([upcoming.then(() => true), woken])) !== true)
continue;
const pulled = await next();
if (pulled) queue.push(pulled);
} catch (error) {
// a decode failure must not leave the pass running into the retry
await done.catch(() => {});
throw error;
}
await done;
};
@@ -504,37 +533,60 @@ async function scanChunk({
};
while (!chunkAborted && cursor < tEnd) {
if (mode === "active") {
// dense sequential decode: every frame is seen, the scheduler
// decides which are worth analyzing
if ((await scanActive()) === "end") break;
mode = "skim";
} else {
// skim: hop keyframe to keyframe (single-frame decodes) while
// calm, capped so long GOPs cannot hide a short screen
const key = await packets.getKeyPacket(cursor + MAX_SKIM_STRIDE_S, {
verifyKeyPackets: true,
});
const target =
key && key.timestamp > cursor
? key.timestamp
: cursor + MAX_SKIM_STRIDE_S;
if (target >= tEnd) {
cursor = tEnd;
break;
try {
if (mode === "active") {
// dense sequential decode: every frame is seen, the scheduler
// decides which are worth analyzing
if ((await scanActive()) === "end") break;
mode = "skim";
} else {
// skim: hop keyframe to keyframe (single-frame decodes) while
// calm, capped so long GOPs cannot hide a short screen
const key = await packets.getKeyPacket(cursor + MAX_SKIM_STRIDE_S, {
verifyKeyPackets: true,
});
const target =
key && key.timestamp > cursor
? key.timestamp
: cursor + MAX_SKIM_STRIDE_S;
if (target >= tEnd) {
cursor = tEnd;
break;
}
const sample = await samples.getSample(target);
if (!sample) {
cursor = target;
continue;
}
const { t, frame, preview } = await pull(sample);
account(t);
if (t >= scheduler!.nextDueT()) await readAndAnalyze(frame, t);
else frame.close();
report(preview);
cursor = Math.max(cursor, target);
if (!scheduler!.calm(cursor)) mode = "active";
}
const sample = await samples.getSample(target);
if (!sample) {
cursor = target;
continue;
} catch (error) {
// resume at the cursor: a first failure there retries as is
// (a crashed GPU process is back by then), a second decodes in
// software from then on, a third hops over the spot
if (chunkAborted || ++failures > MAX_CHUNK_FAILURES) throw error;
failuresHere = cursor > failedAt ? 1 : failuresHere + 1;
failedAt = cursor;
// biome-ignore lint/suspicious/noConsole: the scan carries on, so say what it recovered from
console.warn(
`scanner: chunk ${chunkIndex} failed at ${cursor.toFixed(1)}s, resuming`,
error,
);
if (failuresHere >= 2 && hardwareAcceleration !== "prefer-software") {
hardwareAcceleration = "prefer-software";
samples = new VideoSampleSink(track, { hardwareAcceleration });
} else if (failuresHere >= 2) {
cursor = Math.min(tEnd, cursor + UNDECODABLE_SKIP_S);
}
const { t, frame, preview } = await pull(sample);
account(t);
if (t >= scheduler!.nextDueT()) await readAndAnalyze(frame, t);
else frame.close();
report(preview);
cursor = Math.max(cursor, target);
if (!scheduler!.calm(cursor)) mode = "active";
await new Promise((resolve) =>
setTimeout(resolve, CHUNK_RETRY_DELAY_MS),
);
}
}
@@ -547,7 +599,7 @@ async function scanChunk({
} catch (error) {
post({
kind: "error",
message: `chunk ${chunkIndex} scan failed: ${String(error)}`,
message: String(error),
});
} finally {
input.dispose();

4
app/features/scanner/worker/decode.worker.ts
View File

@@ -79,7 +79,9 @@ async function decode(
const track = await input.getPrimaryVideoTrack();
if (!track) throw new Error("no video track");
await pumpSamples({
samples: new VideoSampleSink(track).samples(request.start),
samples: new VideoSampleSink(track, {
hardwareAcceleration: request.hardwareAcceleration,
}).samples(request.start),
end: request.end,
preview: request.preview,
floor: () => own.floor,

22
app/features/scanner/worker/frame-source.ts
View File

@@ -56,6 +56,7 @@ export type DecodeRequest =
/** the stream ends at the first sample at or past this time */
end: number;
preview: PreviewOptions;
hardwareAcceleration: HardwareAcceleration;
}
| { kind: "floor"; session: number; t: number }
| { kind: "release"; session: number }
@@ -153,10 +154,11 @@ export function openFrameSource(
start: number,
end: number,
preview: PreviewOptions,
hardwareAcceleration: HardwareAcceleration,
): FrameSource {
return worker
? workerSource(worker, file, start, end, preview)
: localSource(file, start, end, preview);
? workerSource(worker, file, start, end, preview, hardwareAcceleration)
: localSource(file, start, end, preview, hardwareAcceleration);
}
let nextSession = 0;
@@ -168,6 +170,7 @@ function workerSource(
start: number,
end: number,
preview: PreviewOptions,
hardwareAcceleration: HardwareAcceleration,
): FrameSource {
const session = nextSession++;
const queue: SourceItem[] = [];
@@ -201,7 +204,15 @@ function workerSource(
worker.addEventListener("message", onMessage);
worker.addEventListener("error", onError);
const send = (request: DecodeRequest) => worker.postMessage(request);
send({ kind: "open", session, file, start, end, preview });
send({
kind: "open",
session,
file,
start,
end,
preview,
hardwareAcceleration,
});
return {
async next() {
while (queue.length === 0 && !ended && !failure) {
@@ -244,6 +255,7 @@ function localSource(
start: number,
end: number,
preview: PreviewOptions,
hardwareAcceleration: HardwareAcceleration,
): FrameSource {
const input = new Input({
formats: ALL_FORMATS,
@@ -265,7 +277,9 @@ function localSource(
const track = await input.getPrimaryVideoTrack();
if (!track) throw new Error("no video track");
await pumpSamples({
samples: new VideoSampleSink(track).samples(start),
samples: new VideoSampleSink(track, { hardwareAcceleration }).samples(
start,
),
end,
preview,
floor: () => floor,