From 21e834b4c83b1fc562f2c65e315a0f309adfc86e Mon Sep 17 00:00:00 2001 From: watermeko <61347352+watermeko@users.noreply.github.com> Date: Fri, 31 Jul 2026 19:08:07 +0800 Subject: [PATCH] perf: reduce latency in 60 Hz mode (#844) - Add decode-driven flow control and bounded GOP-aware queues to prevent stale H.264 frames from accumulating in TCP and WebSocket buffers. - Isolate slow clients with dedicated writer goroutines, write deadlines, and safer connection lifecycle handling. - Move the Direct H.264 socket into the worker, simplify the low-latency decode path, and improve reconnect behavior. --- server/service/stream/direct/client.go | 348 ++++++++++++++++++ server/service/stream/direct/h264.go | 25 +- server/service/stream/direct/pool.go | 12 - server/service/stream/direct/streamer.go | 115 +++--- web/src/lib/websocket.ts | 2 +- web/src/pages/desktop/screen/direct.worker.ts | 285 ++++++++++---- web/src/pages/desktop/screen/h264-direct.tsx | 26 +- 7 files changed, 644 insertions(+), 169 deletions(-) create mode 100644 server/service/stream/direct/client.go delete mode 100644 server/service/stream/direct/pool.go diff --git a/server/service/stream/direct/client.go b/server/service/stream/direct/client.go new file mode 100644 index 0000000..231c4fb --- /dev/null +++ b/server/service/stream/direct/client.go @@ -0,0 +1,348 @@ +package direct + +import ( + "encoding/binary" + "sync" + "time" + + "github.com/gorilla/websocket" + log "github.com/sirupsen/logrus" +) + +const ( + frameAckMessage byte = 2 + streamResyncMessage byte = 3 + + defaultQueueFrames = 8 + defaultQueueBytes = 2 * 1024 * 1024 + maxFlowWindow = 8 + writeWait = 2 * time.Second + pingPeriod = 15 * time.Second +) + +type outboundFrame struct { + key bool + timestamp int64 + payload []byte +} + +type frameQueue struct { + mutex sync.Mutex + wake chan struct{} + + frames []*outboundFrame + queuedBytes int + inFlight []int64 + + maxFrames int + maxBytes int + window int + + flowControlled bool + waitingForKeyframe bool + closed bool +} + +func newFrameQueue(maxFrames int, maxBytes int) *frameQueue { + return &frameQueue{ + wake: make(chan struct{}, 1), + maxFrames: maxFrames, + maxBytes: maxBytes, + waitingForKeyframe: true, + } +} + +func (q *frameQueue) enableFlowControl(window int) { + if window < 1 { + window = 1 + } + if window > maxFlowWindow { + window = maxFlowWindow + } + + q.mutex.Lock() + q.clearFramesLocked() + q.inFlight = q.inFlight[:0] + q.window = window + q.flowControlled = true + q.waitingForKeyframe = true + q.mutex.Unlock() +} + +func (q *frameQueue) canAdvanceStream() bool { + q.mutex.Lock() + defer q.mutex.Unlock() + + if q.closed { + return false + } + if !q.flowControlled { + return true + } + + if q.waitingForKeyframe { + return len(q.inFlight) < q.window + } + + return len(q.frames)+len(q.inFlight) < q.window +} + +func (q *frameQueue) captureState() (active bool, flowControlled bool, canAdvance bool) { + q.mutex.Lock() + defer q.mutex.Unlock() + + if q.closed { + return false, q.flowControlled, false + } + if !q.flowControlled { + return true, false, true + } + if q.waitingForKeyframe { + return true, true, len(q.inFlight) < q.window + } + + return true, true, len(q.frames)+len(q.inFlight) < q.window +} + +func (q *frameQueue) offer(frame *outboundFrame) bool { + q.mutex.Lock() + defer q.mutex.Unlock() + + if q.closed { + return false + } + + if frame.key { + q.clearFramesLocked() + if q.flowControlled && len(q.inFlight) >= q.window { + q.waitingForKeyframe = true + return false + } + q.waitingForKeyframe = false + q.pushLocked(frame) + q.signalLocked() + return true + } + + if q.waitingForKeyframe { + return false + } + + if q.flowControlled { + if len(q.frames)+len(q.inFlight) >= q.window { + q.clearFramesLocked() + q.waitingForKeyframe = true + return false + } + } else if len(q.frames) >= q.maxFrames || q.queuedBytes+len(frame.payload) > q.maxBytes { + q.clearFramesLocked() + q.waitingForKeyframe = true + return false + } + + q.pushLocked(frame) + q.signalLocked() + return true +} + +func (q *frameQueue) popForWrite() *outboundFrame { + q.mutex.Lock() + defer q.mutex.Unlock() + + if q.closed || len(q.frames) == 0 { + return nil + } + + frame := q.frames[0] + q.frames[0] = nil + q.frames = q.frames[1:] + q.queuedBytes -= len(frame.payload) + if q.flowControlled { + q.inFlight = append(q.inFlight, frame.timestamp) + } + + return frame +} + +func (q *frameQueue) acknowledge(timestamp int64) { + q.mutex.Lock() + defer q.mutex.Unlock() + + acknowledged := 0 + for acknowledged < len(q.inFlight) && q.inFlight[acknowledged] <= timestamp { + acknowledged++ + } + if acknowledged == 0 { + return + } + + clear(q.inFlight[:acknowledged]) + q.inFlight = q.inFlight[acknowledged:] +} + +func (q *frameQueue) requestResync() { + q.mutex.Lock() + q.clearFramesLocked() + q.inFlight = q.inFlight[:0] + q.waitingForKeyframe = true + q.mutex.Unlock() +} + +func (q *frameQueue) close() { + q.mutex.Lock() + q.closed = true + q.clearFramesLocked() + q.inFlight = q.inFlight[:0] + q.mutex.Unlock() +} + +func (q *frameQueue) pushLocked(frame *outboundFrame) { + q.frames = append(q.frames, frame) + q.queuedBytes += len(frame.payload) +} + +func (q *frameQueue) clearFramesLocked() { + clear(q.frames) + q.frames = q.frames[:0] + q.queuedBytes = 0 +} + +func (q *frameQueue) signalLocked() { + select { + case q.wake <- struct{}{}: + default: + } +} + +type client struct { + conn *websocket.Conn + queue *frameQueue + + done chan struct{} + writerDone chan struct{} + closeOnce sync.Once +} + +func newClient(conn *websocket.Conn) *client { + return &client{ + conn: conn, + queue: newFrameQueue(defaultQueueFrames, defaultQueueBytes), + done: make(chan struct{}), + writerDone: make(chan struct{}), + } +} + +func (c *client) start() { + go c.writeLoop() +} + +func (c *client) close() { + c.closeOnce.Do(func() { + close(c.done) + c.queue.close() + _ = c.conn.Close() + }) +} + +func (c *client) wait() { + <-c.writerDone +} + +func (c *client) captureState() (active bool, flowControlled bool, canAdvance bool) { + return c.queue.captureState() +} + +func hasCaptureDemand(clients []*client) bool { + for _, client := range clients { + active, flowControlled, canAdvance := client.captureState() + if !active { + continue + } + if !flowControlled || canAdvance { + return true + } + } + + return false +} + +func (c *client) offer(frame *outboundFrame) { + c.queue.offer(frame) +} + +func (c *client) handleControl(messageType int, data []byte) { + if messageType != websocket.BinaryMessage || len(data) == 0 { + return + } + + switch data[0] { + case frameAckMessage: + if len(data) == 9 { + timestamp := int64(binary.LittleEndian.Uint64(data[1:])) + c.queue.acknowledge(timestamp) + } + case streamResyncMessage: + if len(data) == 1 { + c.queue.requestResync() + } + } +} + +func newOutboundFrame(isKeyFrame bool, timestamp int64, data []byte) *outboundFrame { + payload := make([]byte, 9+len(data)) + if isKeyFrame { + payload[0] = 1 + } + binary.LittleEndian.PutUint64(payload[1:9], uint64(timestamp)) + copy(payload[9:], data) + + return &outboundFrame{ + key: isKeyFrame, + timestamp: timestamp, + payload: payload, + } +} + +func (c *client) writeLoop() { + defer close(c.writerDone) + + pingTicker := time.NewTicker(pingPeriod) + defer pingTicker.Stop() + + for { + select { + case <-pingTicker.C: + deadline := time.Now().Add(writeWait) + if err := c.conn.WriteControl(websocket.PingMessage, nil, deadline); err != nil { + c.close() + return + } + default: + } + + if frame := c.queue.popForWrite(); frame != nil { + if err := c.conn.SetWriteDeadline(time.Now().Add(writeWait)); err != nil { + c.close() + return + } + if err := c.conn.WriteMessage(websocket.BinaryMessage, frame.payload); err != nil { + log.Debugf("failed to write h264 frame to %s: %s", c.conn.RemoteAddr(), err) + c.close() + return + } + continue + } + + select { + case <-c.done: + return + case <-c.queue.wake: + case <-pingTicker.C: + deadline := time.Now().Add(writeWait) + if err := c.conn.WriteControl(websocket.PingMessage, nil, deadline); err != nil { + c.close() + return + } + } + } +} diff --git a/server/service/stream/direct/h264.go b/server/service/stream/direct/h264.go index 600a52d..24b7b6a 100644 --- a/server/service/stream/direct/h264.go +++ b/server/service/stream/direct/h264.go @@ -3,6 +3,7 @@ package direct import ( "NanoKVM-Server/service/stream" "net/http" + "strconv" "time" "github.com/gin-gonic/gin" @@ -10,6 +11,8 @@ import ( log "github.com/sirupsen/logrus" ) +const pongWait = 30 * time.Second + var ( streamer = newStreamer() upgrader = websocket.Upgrader{ @@ -26,24 +29,36 @@ func Connect(c *gin.Context) { log.Errorf("failed to upgrade to websocket: %s", err) return } + client := newClient(ws) + if flowWindow, err := strconv.Atoi(c.Query("flow")); err == nil && flowWindow > 0 { + client.queue.enableFlowControl(flowWindow) + } defer func() { - _ = ws.Close() + streamer.removeClient(client) + client.close() + client.wait() log.Debugf("h264 websocket disconnected: %s", ws.RemoteAddr()) }() log.Debugf("h264 websocket connected: %s", ws.RemoteAddr()) - _ = ws.SetReadDeadline(time.Time{}) + ws.SetReadLimit(64) + _ = ws.SetReadDeadline(time.Now().Add(pongWait)) + ws.SetPongHandler(func(string) error { + return ws.SetReadDeadline(time.Now().Add(pongWait)) + }) - streamer.addClient(ws) - defer streamer.removeClient(ws) + streamer.addClient(client) unregisterMode := stream.RegisterH264Mode(stream.H264ModeDirect) defer unregisterMode() for { - if _, _, err := ws.NextReader(); err != nil { + messageType, data, err := ws.ReadMessage() + if err != nil { log.Debugf("failed to read message (client disconnected): %s", err) return } + _ = ws.SetReadDeadline(time.Now().Add(pongWait)) + client.handleControl(messageType, data) } } diff --git a/server/service/stream/direct/pool.go b/server/service/stream/direct/pool.go deleted file mode 100644 index 1f2fe06..0000000 --- a/server/service/stream/direct/pool.go +++ /dev/null @@ -1,12 +0,0 @@ -package direct - -import ( - "bytes" - "sync" -) - -var BufferPool = sync.Pool{ - New: func() interface{} { - return new(bytes.Buffer) - }, -} diff --git a/server/service/stream/direct/streamer.go b/server/service/stream/direct/streamer.go index 7c336ed..f9778d7 100644 --- a/server/service/stream/direct/streamer.go +++ b/server/service/stream/direct/streamer.go @@ -4,62 +4,71 @@ import ( "NanoKVM-Server/common" "NanoKVM-Server/service/stream" "NanoKVM-Server/service/vm" - "bytes" - "encoding/binary" "sync" "sync/atomic" "time" - "github.com/gorilla/websocket" log "github.com/sirupsen/logrus" ) type Streamer struct { mutex sync.Mutex - clients map[*websocket.Conn]bool - clientSnapshot atomic.Pointer[[]*websocket.Conn] - running int32 + clients map[*client]struct{} + clientSnapshot atomic.Pointer[[]*client] + running bool viewerVersion uint64 } func newStreamer() *Streamer { s := &Streamer{ - clients: make(map[*websocket.Conn]bool), + clients: make(map[*client]struct{}), } s.updateClientSnapshotLocked() return s } -func (s *Streamer) addClient(ws *websocket.Conn) { +func (s *Streamer) addClient(client *client) { + client.start() + s.mutex.Lock() - s.clients[ws] = true + s.clients[client] = struct{}{} count := s.updateClientSnapshotLocked() s.viewerVersion++ version := s.viewerVersion + start := !s.running + if start { + s.running = true + } s.mutex.Unlock() vm.UpdateHdmiViewerSnapshot("direct", count, version) - if atomic.CompareAndSwapInt32(&s.running, 0, 1) { + if start { go s.run() log.Debug("h264 stream started") } } -func (s *Streamer) removeClient(ws *websocket.Conn) { +func (s *Streamer) removeClient(client *client) { s.mutex.Lock() - delete(s.clients, ws) + if _, exists := s.clients[client]; !exists { + s.mutex.Unlock() + return + } + + delete(s.clients, client) count := s.updateClientSnapshotLocked() s.viewerVersion++ version := s.viewerVersion s.mutex.Unlock() + client.close() vm.UpdateHdmiViewerSnapshot("direct", count, version) log.Debugf("h264 websocket disconnected, remaining clients: %d", count) } func (s *Streamer) updateClientSnapshotLocked() int { - clients := make([]*websocket.Conn, 0, len(s.clients)) + clients := make([]*client, 0, len(s.clients)) for client := range s.clients { clients = append(clients, client) } @@ -68,7 +77,7 @@ func (s *Streamer) updateClientSnapshotLocked() int { return len(clients) } -func (s *Streamer) getClients() []*websocket.Conn { +func (s *Streamer) getClients() []*client { clients := s.clientSnapshot.Load() if clients == nil { return nil @@ -78,8 +87,6 @@ func (s *Streamer) getClients() []*websocket.Conn { } func (s *Streamer) run() { - defer atomic.StoreInt32(&s.running, 0) - screen := common.GetScreen() common.CheckScreen() fps := screen.FPS @@ -93,24 +100,10 @@ func (s *Streamer) run() { for range ticker.C { clients := s.getClients() if len(clients) == 0 { - log.Debug("h264 stream stopped due to no clients") - return - } - - data, result := vision.ReadH264(screen.Width, screen.Height, screen.BitRate) - stream.UpdateCaptureStatus(stream.CaptureModeDirect, result) - if result < 0 || len(data) == 0 { - continue - } - - isKeyFrame := byte(0) - if result == 3 { - isKeyFrame = byte(1) - } - - timestamp := time.Since(startTime).Microseconds() - - if err := s.send(clients, isKeyFrame, timestamp, data); err != nil { + if s.stopIfIdle() { + log.Debug("h264 stream stopped due to no clients") + return + } continue } @@ -119,42 +112,34 @@ func (s *Streamer) run() { ticker.Reset(time.Second / time.Duration(fps)) } + if !hasCaptureDemand(clients) { + continue + } + + data, result := vision.ReadH264(screen.Width, screen.Height, screen.BitRate) + stream.UpdateCaptureStatus(stream.CaptureModeDirect, result) + if result < 0 || len(data) == 0 { + continue + } + + timestamp := time.Since(startTime).Microseconds() + frame := newOutboundFrame(result == 3, timestamp, data) + for _, client := range clients { + client.offer(frame) + } + stream.GetFrameRateCounter().Update() } } -func (s *Streamer) send(clients []*websocket.Conn, isKeyFrame byte, timestamp int64, data []byte) error { - buf := BufferPool.Get().(*bytes.Buffer) - defer BufferPool.Put(buf) +func (s *Streamer) stopIfIdle() bool { + s.mutex.Lock() + defer s.mutex.Unlock() - buf.Reset() - buf.Grow(1 + 8 + len(data)) - - if err := buf.WriteByte(isKeyFrame); err != nil { - log.Errorf("failed to write keyframe flag: %s", err) - return err + if len(s.clients) > 0 { + return false } - var tsBytes [8]byte - binary.LittleEndian.PutUint64(tsBytes[:], uint64(timestamp)) - if _, err := buf.Write(tsBytes[:]); err != nil { - log.Errorf("failed to write timestamp: %s", err) - return err - } - - if _, err := buf.Write(data); err != nil { - log.Errorf("failed to write h264 data: %s", err) - return err - } - - for _, client := range clients { - if err := client.WriteMessage(websocket.BinaryMessage, buf.Bytes()); err != nil { - log.Errorf("failed to write message to client %s: %s.", client.RemoteAddr(), err) - - s.removeClient(client) - _ = client.Close() - } - } - - return nil + s.running = false + return true } diff --git a/web/src/lib/websocket.ts b/web/src/lib/websocket.ts index e67d181..f535fd7 100644 --- a/web/src/lib/websocket.ts +++ b/web/src/lib/websocket.ts @@ -22,7 +22,7 @@ const DEFAULT_OPTIONS: Required = { url: `${getBaseUrl('ws')}/api/ws`, heartbeatInterval: 10 * 1000, reconnectInterval: 3 * 1000, - maxReconnectAttempts: 1 + maxReconnectAttempts: Number.POSITIVE_INFINITY }; export class WsClient { diff --git a/web/src/pages/desktop/screen/direct.worker.ts b/web/src/pages/desktop/screen/direct.worker.ts index 81e9b1c..f39ff9a 100644 --- a/web/src/pages/desktop/screen/direct.worker.ts +++ b/web/src/pages/desktop/screen/direct.worker.ts @@ -5,38 +5,147 @@ let ctx: OffscreenCanvasRenderingContext2D | null = null; let rendering: boolean = false; let flushScheduled: boolean = false; let decoder: VideoDecoder | null = null; -let displayDecoder: VideoDecoder | null = null; -let replacementDecoder: VideoDecoder | null = null; -let resyncAtNextKeyFrame: boolean = false; +let streamUrl: string | null = null; +let socket: WebSocket | null = null; +let reconnectTimer: ReturnType | null = null; +let reconnectDelayMs = 250; +let stopped = false; +let resyncRequested = false; +let pendingAckTimestamp: number | null = null; +let decodeBackpressured = false; const maxQueuedFrames = 1; -const maxDecoderQueueSize = 6; +const maxReconnectDelayMs = 5_000; +const frameAckMessage = 2; +const streamResyncMessage = 3; +const flowControlWindow = 8; +const decoderHighWatermark = 6; +const decoderLowWatermark = 3; const frameQueue = new Queue(); const frameChannel = new MessageChannel(); +type WorkerMessage = { + type: 'h264' | 'stop'; + canvas?: OffscreenCanvas; + url?: string; +}; + frameChannel.port1.onmessage = () => { flushScheduled = false; processFrameQueue(); }; -self.onmessage = (event: MessageEvent) => { - const { type, data, canvas: offscreenCanvas } = event.data; +self.onmessage = (event: MessageEvent) => { + const { type, canvas: offscreenCanvas, url } = event.data; switch (type) { case 'h264': + if (!offscreenCanvas || !url) { + return; + } + canvas = offscreenCanvas; - ctx = canvas!.getContext('2d') as OffscreenCanvasRenderingContext2D; + ctx = canvas!.getContext('2d', { + alpha: false, + desynchronized: true + }) as OffscreenCanvasRenderingContext2D; + streamUrl = url; + stopped = false; + connect(); break; - case 'ws_message': - handleWsMessage(data); - break; - case 'error': - case 'close': + case 'stop': + stopped = true; + clearReconnectTimer(); + disconnect(); resetDecoder(); break; } }; +function connect() { + if (stopped || !streamUrl || socket) { + return; + } + + try { + const url = new URL(streamUrl); + url.searchParams.set('flow', String(flowControlWindow)); + const nextSocket = new WebSocket(url); + nextSocket.binaryType = 'arraybuffer'; + socket = nextSocket; + + nextSocket.onopen = () => { + if (socket !== nextSocket || stopped) { + return; + } + + reconnectDelayMs = 250; + resyncRequested = false; + pendingAckTimestamp = null; + decodeBackpressured = false; + }; + + nextSocket.onmessage = (event) => { + if (socket !== nextSocket || stopped || !(event.data instanceof ArrayBuffer)) { + return; + } + + handleWsMessage(event.data); + }; + + nextSocket.onerror = () => { + if (socket === nextSocket) { + nextSocket.close(); + } + }; + + nextSocket.onclose = () => { + if (socket !== nextSocket) { + return; + } + + socket = null; + resyncRequested = false; + pendingAckTimestamp = null; + decodeBackpressured = false; + resetDecoder(); + scheduleReconnect(); + }; + } catch (error) { + console.error('Failed to create Direct H264 WebSocket:', error); + scheduleReconnect(); + } +} + +function disconnect() { + const currentSocket = socket; + socket = null; + + if (currentSocket && currentSocket.readyState !== WebSocket.CLOSED) { + currentSocket.close(); + } +} + +function scheduleReconnect() { + if (stopped || reconnectTimer !== null || !streamUrl) { + return; + } + + const delay = reconnectDelayMs; + reconnectDelayMs = Math.min(reconnectDelayMs * 2, maxReconnectDelayMs); + reconnectTimer = setTimeout(() => { + reconnectTimer = null; + connect(); + }, delay); +} + +function clearReconnectTimer() { + if (reconnectTimer !== null) { + clearTimeout(reconnectTimer); + reconnectTimer = null; + } +} + function handleWsMessage(message: ArrayBuffer) { try { if (message.byteLength < 9) { @@ -48,33 +157,21 @@ function handleWsMessage(message: ArrayBuffer) { const timestamp = Number(view.getBigUint64(1, true)); const data = new Uint8Array(message, 9); - if ( - !replacementDecoder && - decoder?.state === 'configured' && - decoder.decodeQueueSize >= maxDecoderQueueSize - ) { - resyncAtNextKeyFrame = true; - } - - if (resyncAtNextKeyFrame && isKeyFrame && displayDecoder) { - const replacement = createDecoder(); - if (replacement) { - replacementDecoder = replacement; - decoder = replacement; - resyncAtNextKeyFrame = false; - } - } - if (!decoder) { if (!isKeyFrame) { + requestStreamResync(); return; } + + resyncRequested = false; const initial = createDecoder(); if (!initial) { + requestStreamResync(); return; } + decoder = initial; - displayDecoder = initial; + resyncRequested = false; } if (decoder?.state === 'configured') { @@ -97,12 +194,19 @@ function createDecoder(): VideoDecoder | null { handleDecodedFrame(instance, frame); }, error: () => { - resetDecoder(); + if (decoder === instance) { + requestStreamResync(); + resetDecoder(); + } } }; try { instance = new VideoDecoder(init); + const configuredDecoder = instance; + instance.ondequeue = () => { + releaseDecodeBackpressure(configuredDecoder); + }; instance.configure({ codec: 'avc1.42E02A', hardwareAcceleration: 'prefer-hardware', @@ -110,6 +214,9 @@ function createDecoder(): VideoDecoder | null { }); return instance; } catch (err) { + if (instance && instance.state !== 'closed') { + instance.close(); + } console.log(err); return null; } @@ -121,24 +228,17 @@ function handleDecodedFrame(source: VideoDecoder | null, frame: VideoFrame) { return; } - if (source === replacementDecoder) { - const previous = displayDecoder; - displayDecoder = source; - decoder = source; - replacementDecoder = null; - - Array.from(frameQueue.drain()).forEach((queuedFrame) => queuedFrame.close()); - if (previous && previous !== source && previous.state !== 'closed') { - previous.close(); - } - } else if (source !== displayDecoder) { + if (source !== decoder) { frame.close(); return; } frameQueue.enqueue(frame); while (frameQueue.size > maxQueuedFrames) { - frameQueue.dequeue()?.close(); + const droppedFrame = frameQueue.dequeue(); + if (droppedFrame) { + droppedFrame.close(); + } } if (!rendering) { @@ -156,8 +256,14 @@ function decode(target: VideoDecoder, isKeyFrame: boolean, timestamp: number, da try { target.decode(chunk); + pendingAckTimestamp = timestamp; + if (target.decodeQueueSize >= decoderHighWatermark) { + decodeBackpressured = true; + } + releaseDecodeBackpressure(target); } catch (err: any) { if (err.name === 'TypeError' || err.message.includes('configured')) { + requestStreamResync(); resetDecoder(); } } @@ -166,7 +272,14 @@ function decode(target: VideoDecoder, isKeyFrame: boolean, timestamp: number, da function processFrameQueue() { const frame = frameQueue.dequeue(); if (frame) { - renderFrame(frame); + try { + renderFrame(frame); + } catch (error) { + console.error('Failed to render Direct H264 frame:', error); + requestStreamResync(); + resetDecoder(); + return; + } } if (frameQueue.size > 0) { @@ -191,33 +304,81 @@ function renderFrame(frame: VideoFrame) { return; } - if (canvas.width !== frame.displayWidth || canvas.height !== frame.displayHeight) { - canvas.width = frame.displayWidth; - canvas.height = frame.displayHeight; - } + try { + if (canvas.width !== frame.displayWidth || canvas.height !== frame.displayHeight) { + canvas.width = frame.displayWidth; + canvas.height = frame.displayHeight; + } - ctx.drawImage(frame, 0, 0, canvas.width, canvas.height); - frame.close(); + ctx.drawImage(frame, 0, 0, canvas.width, canvas.height); + } finally { + frame.close(); + } } function resetDecoder() { - const decoders = new Set([decoder, displayDecoder, replacementDecoder]); - decoders.forEach((item) => { - if (item && item.state !== 'closed') { - try { - item.close(); - } catch (err) { - console.log(err); - } + if (decoder && decoder.state !== 'closed') { + try { + decoder.close(); + } catch (err) { + console.log(err); } - }); + } decoder = null; - displayDecoder = null; - replacementDecoder = null; - resyncAtNextKeyFrame = false; + pendingAckTimestamp = null; + decodeBackpressured = false; rendering = false; flushScheduled = false; Array.from(frameQueue.drain()).forEach((frame) => frame.close()); } + +function releaseDecodeBackpressure(source: VideoDecoder) { + if (source !== decoder || pendingAckTimestamp === null) { + return; + } + + if (decodeBackpressured && source.decodeQueueSize > decoderLowWatermark) { + return; + } + + decodeBackpressured = false; + acknowledgeFrame(pendingAckTimestamp); + pendingAckTimestamp = null; +} + +function acknowledgeFrame(timestamp: number) { + const currentSocket = socket; + if (!currentSocket || currentSocket.readyState !== WebSocket.OPEN) { + return; + } + + const message = new ArrayBuffer(9); + const view = new DataView(message); + view.setUint8(0, frameAckMessage); + view.setBigUint64(1, BigInt(timestamp), true); + try { + currentSocket.send(message); + } catch { + currentSocket.close(); + } +} + +function requestStreamResync() { + if (resyncRequested) { + return; + } + + const currentSocket = socket; + if (!currentSocket || currentSocket.readyState !== WebSocket.OPEN) { + return; + } + + try { + currentSocket.send(new Uint8Array([streamResyncMessage])); + resyncRequested = true; + } catch { + currentSocket.close(); + } +} diff --git a/web/src/pages/desktop/screen/h264-direct.tsx b/web/src/pages/desktop/screen/h264-direct.tsx index 5b36e8f..09f1c77 100644 --- a/web/src/pages/desktop/screen/h264-direct.tsx +++ b/web/src/pages/desktop/screen/h264-direct.tsx @@ -1,7 +1,6 @@ import { useEffect, useRef } from 'react'; import clsx from 'clsx'; import { useAtom, useAtomValue } from 'jotai'; -import { w3cwebsocket as W3cWebSocket } from 'websocket'; import * as storage from '@/lib/localstorage.ts'; import { getBaseUrl } from '@/lib/service.ts'; @@ -38,32 +37,11 @@ export const H264Direct = () => { workerRef.current = worker; const offscreen = canvasRef.current.transferControlToOffscreen(); - worker.postMessage({ type: 'h264', canvas: offscreen }, [offscreen]); - const url = `${getBaseUrl('ws')}/api/stream/h264/direct`; - const ws = new W3cWebSocket(url); - ws.binaryType = 'arraybuffer'; - - ws.onmessage = (event) => { - try { - worker.postMessage({ type: 'ws_message', data: event.data }, [event.data]); - } catch (error) { - console.error('Error processing WebSocket message:', error); - } - }; - - ws.onerror = () => { - worker.postMessage({ type: 'error' }); - }; - - ws.onclose = () => { - worker.postMessage({ type: 'close' }); - }; + worker.postMessage({ type: 'h264', canvas: offscreen, url }, [offscreen]); return () => { - if (ws.readyState === 1) { - ws.close(); - } + worker.postMessage({ type: 'stop' }); worker.terminate(); }; }, []);