diff --git a/jsonrpc.go b/jsonrpc.go index 7a60656e1..b9281bb7d 100644 --- a/jsonrpc.go +++ b/jsonrpc.go @@ -99,6 +99,12 @@ func writeJSONRPCEvent(event string, params any, session *Session) { } } +// onRPCMessage parses a single RPC message off the per-session pump and +// dispatches it. Runs synchronously on the pump goroutine; handlers +// flagged Synchronous keep running there (so dequeue order is preserved +// for ordering-sensitive RPCs like pauseVideo / resumeVideo), everything +// else is dispatched in a fresh goroutine so a slow handler can't +// head-of-line block the queue. func onRPCMessage(message webrtc.DataChannelMessage, session *Session) { var request JSONRPCRequest err := json.Unmarshal(message.Data, &request) @@ -120,14 +126,6 @@ func onRPCMessage(message webrtc.DataChannelMessage, session *Session) { return } - scopedLogger := jsonRpcLogger.With(). - Str("method", request.Method). - Interface("params", request.Params). - Interface("id", request.ID).Logger() - - scopedLogger.Trace().Msg("Received RPC request") - t := time.Now() - handler, ok := rpcHandlers[request.Method] if !ok { errorResponse := JSONRPCResponse{ @@ -142,7 +140,26 @@ func onRPCMessage(message webrtc.DataChannelMessage, session *Session) { return } - result, err := callRPCHandler(scopedLogger, handler, request.Params) + if handler.Synchronous { + invokeRPCHandler(request, handler, session) + } else { + go invokeRPCHandler(request, handler, session) + } +} + +// invokeRPCHandler runs a single RPC handler and writes its response back +// to the session. Called either inline on the pump (Synchronous handlers) +// or from a per-message goroutine (the default). +func invokeRPCHandler(request JSONRPCRequest, handler RPCHandler, session *Session) { + scopedLogger := jsonRpcLogger.With(). + Str("method", request.Method). + Interface("params", request.Params). + Interface("id", request.ID).Logger() + + scopedLogger.Trace().Msg("Received RPC request") + t := time.Now() + + result, err := callRPCHandler(scopedLogger, handler, session, request.Params) if err != nil { scopedLogger.Error().Err(err).Msg("Error calling RPC handler") errorResponse := JSONRPCResponse{ @@ -524,10 +541,25 @@ type RPCHandler struct { Func any Params []string OptionalParams []string + + // TakesSession: the handler's first parameter is *Session and the + // dispatcher injects the receiving session into it. Used by + // session-bound RPCs (e.g. pauseVideo / resumeVideo) that must act + // on the session whose data channel delivered the message, not on + // the global currentSession. + TakesSession bool + + // Synchronous: the handler runs inline on the per-session rpcQueue + // pump goroutine rather than in a fresh per-message goroutine. Use + // for ordering-sensitive handlers — pause/resume toggle a shared + // global refcount and the Go scheduler can otherwise reorder a + // tight pause→resume pair. Slow or blocking handlers MUST stay + // async (the default). + Synchronous bool } // call the handler but recover from a panic to ensure our RPC thread doesn't collapse on malformed calls -func callRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[string]any) (result any, err error) { +func callRPCHandler(logger zerolog.Logger, handler RPCHandler, session *Session, params map[string]any) (result any, err error) { // Use defer to recover from a panic defer func() { if r := recover(); r != nil { @@ -541,11 +573,11 @@ func callRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[string }() // Call the handler - result, err = riskyCallRPCHandler(logger, handler, params) + result, err = riskyCallRPCHandler(logger, handler, session, params) return result, err // do not combine these two lines into one, as it breaks the above defer function's setting of err } -func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[string]any) (any, error) { +func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, session *Session, params map[string]any) (any, error) { handlerValue := reflect.ValueOf(handler.Func) handlerType := handlerValue.Type() @@ -556,8 +588,12 @@ func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[s numParams := handlerType.NumIn() allParamNames := append(handler.Params, handler.OptionalParams...) //nolint:gocritic - if len(allParamNames) != numParams { - err := fmt.Errorf("mismatch between handler parameters (%d) and defined parameter names (%d)", numParams, len(allParamNames)) + expectedSlots := len(allParamNames) + if handler.TakesSession { + expectedSlots++ + } + if expectedSlots != numParams { + err := fmt.Errorf("mismatch between handler parameters (%d) and defined parameter names (%d)", numParams, expectedSlots) logger.Error().Strs("paramNames", allParamNames).Err(err).Msg("Cannot call RPC handler") return nil, err } @@ -568,14 +604,21 @@ func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[s } args := make([]reflect.Value, numParams) + // Reflective params start after the injected *Session slot, if any. + argOffset := 0 + if handler.TakesSession { + args[0] = reflect.ValueOf(session) + argOffset = 1 + } - for i := range numParams { - paramType := handlerType.In(i) + for i := range len(allParamNames) { + paramType := handlerType.In(i + argOffset) paramName := allParamNames[i] paramValue, ok := params[paramName] + argIdx := i + argOffset if !ok { if optionalSet[paramName] { - args[i] = reflect.Zero(paramType) + args[argIdx] = reflect.Zero(paramType) continue } err := fmt.Errorf("missing parameter: %s", paramName) @@ -609,7 +652,7 @@ func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[s newSlice.Index(j).Set(elemValue.Convert(paramType.Elem())) } } - args[i] = newSlice + args[argIdx] = newSlice } else if paramType.Kind() == reflect.Struct && convertedValue.Kind() == reflect.Map { jsonData, err := json.Marshal(convertedValue.Interface()) if err != nil { @@ -620,12 +663,12 @@ func riskyCallRPCHandler(logger zerolog.Logger, handler RPCHandler, params map[s if err := json.Unmarshal(jsonData, newStruct); err != nil { return nil, fmt.Errorf("failed to unmarshal JSON into struct: %v for parameter %s", err, paramName) } - args[i] = reflect.ValueOf(newStruct).Elem() + args[argIdx] = reflect.ValueOf(newStruct).Elem() } else { return nil, fmt.Errorf("invalid parameter type for: %s, type: %s", paramName, paramType.Kind()) } } else { - args[i] = convertedValue.Convert(paramType) + args[argIdx] = convertedValue.Convert(paramType) } } @@ -1347,6 +1390,8 @@ var rpcHandlers = map[string]RPCHandler{ "wheelReport": {Func: rpcWheelReport, Params: []string{"wheelY", "wheelX"}}, "wakeHost": {Func: rpcWakeHost}, "getVideoState": {Func: rpcGetVideoState}, + "pauseVideo": {Func: rpcPauseVideo, TakesSession: true, Synchronous: true}, + "resumeVideo": {Func: rpcResumeVideo, TakesSession: true, Synchronous: true}, "getUSBState": {Func: rpcGetUSBState}, "unmountImage": {Func: rpcUnmountImage}, "rpcMountBuiltInImage": {Func: rpcMountBuiltInImage, Params: []string{"filename"}}, diff --git a/ui/src/components/WebRTCVideo.tsx b/ui/src/components/WebRTCVideo.tsx index 36a4be61f..5aedd6c5b 100644 --- a/ui/src/components/WebRTCVideo.tsx +++ b/ui/src/components/WebRTCVideo.tsx @@ -546,6 +546,30 @@ export default function WebRTCVideo({ [keyDownHandler, keyUpHandler, resetKeyboardState], ); + // Pause/resume the server-side video feed when the tab is hidden so we + // don't burn WAN bandwidth decoding-then-discarding frames the user + // can't see. The encoder is restarted on resume so the first frame is + // an IDR and decode is artifact-free. + useEffect( + function pauseVideoOnTabHidden() { + const sync = () => { + sendRpc(document.hidden ? "pauseVideo" : "resumeVideo", {}); + }; + + // Sync once on mount in case the tab is already hidden when we + // (re)connect, then track every visibility change. + sync(); + + const abortController = new AbortController(); + document.addEventListener("visibilitychange", sync, { + signal: abortController.signal, + }); + + return () => abortController.abort(); + }, + [sendRpc], + ); + // Setup Video Event Listeners useEffect( function setupVideoEventListeners() { diff --git a/video.go b/video.go index e981979d6..083b5d91c 100644 --- a/video.go +++ b/video.go @@ -13,8 +13,56 @@ var ( lastVideoState native.VideoState videoSleepModeCtx context.Context videoSleepModeCancel context.CancelFunc + + videoConsumersMu sync.Mutex + videoConsumers = map[string]struct{}{} ) +// acquireVideoStream registers a named consumer of the capture pipeline. +// The first acquirer starts the native video stream and pauses the HDMI +// sleep ticker; subsequent acquirers from different consumers are recorded +// without touching the underlying stream. Idempotent per consumer key — +// re-acquiring an already-held key is a no-op. +func acquireVideoStream(consumer string) { + videoConsumersMu.Lock() + defer videoConsumersMu.Unlock() + + if _, exists := videoConsumers[consumer]; exists { + return + } + videoConsumers[consumer] = struct{}{} + if len(videoConsumers) == 1 { + _ = nativeInstance.VideoStart() + stopVideoSleepModeTicker() + } +} + +// releaseVideoStream unregisters a consumer. When the last consumer is +// released, the native video stream is stopped and the HDMI sleep ticker +// is restarted. Idempotent — releasing an unknown key is a no-op. +func releaseVideoStream(consumer string) { + videoConsumersMu.Lock() + defer videoConsumersMu.Unlock() + + if _, exists := videoConsumers[consumer]; !exists { + return + } + delete(videoConsumers, consumer) + if len(videoConsumers) == 0 { + _ = nativeInstance.VideoStop() + startVideoSleepModeTicker() + } +} + +// videoStreamHasConsumers reports whether any consumer currently holds the +// capture pipeline open. The HDMI sleep ticker uses this to decide whether +// it is safe to put the capture chip to sleep. +func videoStreamHasConsumers() bool { + videoConsumersMu.Lock() + defer videoConsumersMu.Unlock() + return len(videoConsumers) > 0 +} + const ( defaultVideoSleepModeDuration = 1 * time.Minute ) @@ -109,6 +157,22 @@ func updateHostDisplayAdvertisement(reason string, force bool) error { return setHostDisplayAdvertisedLocked(shouldAdvertiseHostDisplayLocked(), reason, force) } +// rpcPauseVideo releases this session's slot in the video stream +// refcount. Registered with TakesSession + Synchronous so it acts on +// the receiving session and can't be reordered relative to a +// resumeVideo from the same source. +func rpcPauseVideo(s *Session) error { + releaseVideoStream(s.videoConsumerKey()) + return nil +} + +// rpcResumeVideo re-acquires this session's slot. See rpcPauseVideo for +// the dispatcher flags this relies on. +func rpcResumeVideo(s *Session) error { + acquireVideoStream(s.videoConsumerKey()) + return nil +} + type rpcVideoSleepModeResponse struct { Supported bool `json:"supported"` Enabled bool `json:"enabled"` @@ -184,8 +248,8 @@ func doVideoSleepModeTicker(ctx context.Context, duration time.Duration) { for { select { case <-timer.C: - if getActiveSessions() > 0 { - nativeLogger.Warn().Msg("not going to enter HDMI sleep mode because there are active sessions") + if videoStreamHasConsumers() { + nativeLogger.Warn().Msg("not going to enter HDMI sleep mode because the capture pipeline has consumers") continue } diff --git a/webrtc.go b/webrtc.go index 7acd5f1d9..72228c36f 100644 --- a/webrtc.go +++ b/webrtc.go @@ -18,6 +18,7 @@ import ( "github.com/coder/websocket" "github.com/coder/websocket/wsjson" "github.com/gin-gonic/gin" + "github.com/google/uuid" "github.com/pion/ice/v4" "github.com/pion/interceptor" "github.com/pion/webrtc/v4" @@ -25,6 +26,11 @@ import ( ) type Session struct { + // id is a stable per-session identifier used as the suffix of the + // "webrtc:" consumer key for the video stream refcount in video.go. + // Generated in newSession; never reused. + id string + peerConnection *webrtc.PeerConnection VideoTrack *webrtc.TrackLocalStaticSample AudioTrack *webrtc.TrackLocalStaticSample @@ -48,6 +54,13 @@ type Session struct { codecMimeType string } +// videoConsumerKey is the key this session holds in the video stream +// refcount (see video.go). Acquired when ICE reaches Connected, released +// on Closed; pauseVideo / resumeVideo toggle it for this session only. +func (s *Session) videoConsumerKey() string { + return "webrtc:" + s.id +} + var ( actionSessions int = 0 activeSessionsMutex = &sync.Mutex{} @@ -509,6 +522,7 @@ func newSession(config SessionConfig) (*Session, error) { } session := &Session{ + id: uuid.New().String(), peerConnection: peerConnection, done: make(chan struct{}), rpcQueue: make(chan webrtc.DataChannelMessage, 256), @@ -518,6 +532,10 @@ func newSession(config SessionConfig) (*Session, error) { rpcQueue := session.rpcQueue go func() { + // onRPCMessage runs synchronously on this pump goroutine so + // handlers flagged Synchronous (pause/resume) keep their + // dequeue order. The dispatcher spawns its own goroutine for + // every async handler internally. for { select { case <-session.done: @@ -529,8 +547,7 @@ func newSession(config SessionConfig) (*Session, error) { case <-session.done: return case msg := <-rpcQueue: - // TODO: only use goroutine if the task is asynchronous - go onRPCMessage(msg, session) + onRPCMessage(msg, session) } } }() @@ -604,7 +621,16 @@ func newSession(config SessionConfig) (*Session, error) { if incrActiveSessions() == 1 { onFirstSessionConnected() } + // Per-session setup (codec, audio start) must run + // before acquireVideoStream — the 0→1 refcount + // transition starts the encoder, which has to use + // this session's codec. onSessionConnected(session) + // Per-session slot in the video stream refcount; the + // 0→1 transition (first consumer overall) starts the + // encoder, so the first frame after a fresh acquire + // is an IDR. + acquireVideoStream(session.videoConsumerKey()) if mqttManager != nil { mqttManager.publishSessionsState() } @@ -642,6 +668,10 @@ func newSession(config SessionConfig) (*Session, error) { } if isConnected { isConnected = false + // Drop our slot in the video stream refcount. Idempotent — + // if pauseVideo already released it the call is a no-op. + // On the N→0 transition the encoder is stopped. + releaseVideoStream(session.videoConsumerKey()) onActiveSessionsChanged() if decrActiveSessions() == 0 { scopedLogger.Info().Msg("last session disconnected, stopping video stream") @@ -665,16 +695,22 @@ func onActiveSessionsChanged() { // capture is a shared pipeline; starting it again on a handoff connect (count // 1→2) would issue redundant native start calls and re-run the sleep-mode // re-lock wait while video is already streaming. +// +// VideoStart / sleep-ticker stop are owned by the video stream refcount +// (acquireVideoStream in video.go); we only handle the global lifecycle +// concerns that aren't a refcount transition here. func onFirstSessionConnected() { - stopVideoSleepModeTicker() _ = setHostDisplayAdvertised(true, "first_session_connected", false) - _ = nativeInstance.VideoStart() } // onSessionConnected runs per session when ICE reaches Connected. Uses the // session parameter directly rather than the currentSession global — that // global is assigned by the caller AFTER ExchangeOffer returns, and ICE // connected can fire before then, racing the assignment. +// +// Codec selection must happen here (before the per-session +// acquireVideoStream that follows) so the encoder restarts with this +// session's codec on the refcount 0→1 transition. func onSessionConnected(session *Session) { notifyFailsafeMode(session) if session.codecMimeType == webrtc.MimeTypeH265 { @@ -688,10 +724,10 @@ func onSessionConnected(session *Session) { } func onLastSessionDisconnected() { - // Safety net: ensure all keys are released when the last session disconnects + // Safety net: ensure all keys are released when the last session disconnects. + // VideoStop / sleep ticker are owned by releaseVideoStream (called when each + // session's ICE state reaches Closed, just above this). _ = rpcKeyboardReport(0, keyboardClearStateKeys) stopAudio() - _ = nativeInstance.VideoStop() _ = applyHostDisplayAdvertisement("last_session_disconnected") - startVideoSleepModeTicker() }