-
Notifications
You must be signed in to change notification settings - Fork 45
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Revisit serialization dependencies, add progress bar for playback (ex…
…perimental) (#244) * Add progress bar for playback, migrate msgpackr => @msgpack/msgpack for (in-browser) stream support * Migrate msgpack => msgspec on Python side msgspec claims to be much faster, ~3x for serialization (our main bottleneck): https://jcristharif.com/msgspec/benchmarks.html#messagepack-serialization * Revert gtag
- Loading branch information
Showing
10 changed files
with
160 additions
and
142 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,81 +1,141 @@ | ||
| import { unpack } from "msgpackr"; | ||
| import { decodeAsync, decode } from "@msgpack/msgpack"; | ||
| import { Message } from "./WebsocketMessages"; | ||
| import React, { useContext, useEffect } from "react"; | ||
| import { decompress } from "fflate"; | ||
|
|
||
| import { useContext, useEffect, useState } from "react"; | ||
| import { ViewerContext } from "./App"; | ||
| import { Progress, useMantineTheme } from "@mantine/core"; | ||
|
|
||
| interface SerializedMessages { | ||
| loopStartIndex: number | null; | ||
| durationSeconds: number; | ||
| messages: [number, Message][]; | ||
| } | ||
|
|
||
| async function deserializeMsgpackFile<T>(fileUrl: string): Promise<T> { | ||
| // Fetch the file using fetch() | ||
| /** Download, decompress, and deserialize a file, which should be serialized | ||
| * via msgpack and compressed via gzip. Also takes a hook for status updates. */ | ||
| async function deserializeGzippedMsgpackFile<T>( | ||
| fileUrl: string, | ||
| setStatus: (status: { downloaded: number; total: number }) => void, | ||
| ): Promise<T> { | ||
| const response = await fetch(fileUrl); | ||
| if (!response.ok) { | ||
| throw new Error(`Failed to fetch the file: ${response.statusText}`); | ||
| } | ||
| return new Promise<T>((resolve) => { | ||
| let length = 0; | ||
| const buffer: Uint8Array[] = []; | ||
| response.body!.pipeThrough(new DecompressionStream("gzip")).pipeTo( | ||
| new WritableStream<Uint8Array>({ | ||
| write(chunk) { | ||
| buffer.push(chunk); | ||
| length += chunk.length; | ||
| }, | ||
| close() { | ||
| const output = new Uint8Array(length); | ||
| let offset = 0; | ||
| for (const chunk of buffer) { | ||
| output.set(chunk, offset); | ||
| offset += chunk.length; | ||
| } | ||
| console.log(output.length); | ||
| resolve(unpack(output)); | ||
| }, | ||
| abort(err) { | ||
| console.error("Stream aborted:", err); | ||
| }, | ||
| }), | ||
| ); | ||
| const gzipTotalLength = parseInt(response.headers.get("Content-Length")!); | ||
|
|
||
| if (!DecompressionStream) { | ||
| // Implementation without streaming. | ||
| console.log( | ||
| "DecompressionStream is unavailable. Falling back to approach without streams.", | ||
| ); | ||
| setStatus({ downloaded: gzipTotalLength * 0.0, total: gzipTotalLength }); | ||
| response.arrayBuffer().then((buffer) => { | ||
| // Down downloading. | ||
| setStatus({ | ||
| downloaded: gzipTotalLength * 0.8, | ||
| total: gzipTotalLength, | ||
| }); | ||
| decompress(new Uint8Array(buffer), (error, result) => { | ||
| // Done decompressing, time to unpack. | ||
| setStatus({ | ||
| downloaded: gzipTotalLength * 0.9, | ||
| total: gzipTotalLength, | ||
| }); | ||
| resolve(decode(result) as T); | ||
| setStatus({ | ||
| downloaded: gzipTotalLength, | ||
| total: gzipTotalLength, | ||
| }); | ||
| }); | ||
| }); | ||
| } else { | ||
| // Counters for processed bytes, both before and after compression. | ||
| let gzipReceived = 0; | ||
|
|
||
| // Stream: fetch -> gzip -> msgpack. | ||
| decodeAsync( | ||
| response | ||
| .body!.pipeThrough( | ||
| // Count number of (compressed) bytes. | ||
| new TransformStream({ | ||
| transform(chunk, controller) { | ||
| gzipReceived += chunk.length; | ||
| setStatus({ downloaded: gzipReceived, total: gzipTotalLength }); | ||
| controller.enqueue(chunk); | ||
| // return new Promise((resolve) => setTimeout(resolve, 100)); | ||
| }, | ||
| }), | ||
| ) | ||
| .pipeThrough(new DecompressionStream("gzip")), | ||
| ).then((val) => resolve(val as T)); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| export function PlaybackFromFile({ fileUrl }: { fileUrl: string }) { | ||
| const viewer = useContext(ViewerContext)!; | ||
| const messageQueueRef = viewer.messageQueueRef; | ||
|
|
||
| const darkMode = viewer.useGui((state) => state.theme.dark_mode); | ||
| const [status, setStatus] = useState({ downloaded: 0.0, total: 0.0 }); | ||
| const [loaded, setLoaded] = useState(false); | ||
| const theme = useMantineTheme(); | ||
|
|
||
| useEffect(() => { | ||
| deserializeMsgpackFile<SerializedMessages>(fileUrl).then((recording) => { | ||
| let messageIndex = 0; | ||
| deserializeGzippedMsgpackFile<SerializedMessages>(fileUrl, setStatus).then( | ||
| (recording) => { | ||
| let messageIndex = 0; | ||
|
|
||
| function continuePlayback() { | ||
| const currentTimeSeconds = recording.messages[messageIndex][0]; | ||
| while (currentTimeSeconds >= recording.messages[messageIndex][0]) { | ||
| messageQueueRef.current.push(recording.messages[messageIndex][1]); | ||
| messageIndex += 1; | ||
| function continuePlayback() { | ||
| setLoaded(true); | ||
| const currentTimeSeconds = recording.messages[messageIndex][0]; | ||
| while (currentTimeSeconds >= recording.messages[messageIndex][0]) { | ||
| messageQueueRef.current.push(recording.messages[messageIndex][1]); | ||
| messageIndex += 1; | ||
|
|
||
| // Either finish playback or loop. | ||
| if (messageIndex === recording.messages.length) { | ||
| if (recording.loopStartIndex === null) return; | ||
| messageIndex = recording.loopStartIndex; | ||
| setTimeout( | ||
| continuePlayback, | ||
| (recording.durationSeconds - currentTimeSeconds) * 1000.0, | ||
| ); | ||
| return; | ||
| // Either finish playback or loop. | ||
| if (messageIndex === recording.messages.length) { | ||
| if (recording.loopStartIndex === null) return; | ||
| messageIndex = recording.loopStartIndex; | ||
| setTimeout( | ||
| continuePlayback, | ||
| (recording.durationSeconds - currentTimeSeconds) * 1000.0, | ||
| ); | ||
| return; | ||
| } | ||
| } | ||
|
|
||
| // Handle next set of frames. | ||
| setTimeout( | ||
| continuePlayback, | ||
| (recording.messages[messageIndex][0] - currentTimeSeconds) * 1000.0, | ||
| ); | ||
| } | ||
| setTimeout(continuePlayback, recording.messages[0][0] * 1000.0); | ||
| }, | ||
| ); | ||
| }, []); | ||
|
|
||
| // Handle next set of frames. | ||
| setTimeout( | ||
| continuePlayback, | ||
| (recording.messages[messageIndex][0] - currentTimeSeconds) * 1000.0, | ||
| ); | ||
| } | ||
| setTimeout(continuePlayback, recording.messages[0][0] * 1000.0); | ||
| }); | ||
| }); | ||
| return <></>; | ||
| return ( | ||
| <div | ||
| style={{ | ||
| position: "fixed", | ||
| zIndex: 1000, | ||
| top: 0, | ||
| bottom: 0, | ||
| left: 0, | ||
| right: 0, | ||
| display: loaded ? "none" : "block", | ||
| backgroundColor: darkMode ? theme.colors.dark[9] : "#fff", | ||
| }} | ||
| > | ||
| <Progress | ||
| value={(status.downloaded / status.total) * 100.0} | ||
| radius={0} | ||
| transitionDuration={100} | ||
| /> | ||
| </div> | ||
| ); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.